1import type * as sqs from "@distilled.cloud/aws/sqs";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Binding from "../../Binding.ts";
5import * as Output from "../../Output.ts";
6import { makeBatchedSink } from "../internal/BatchedSink.ts";
7import { isBindingHost } from "../Lambda/Function.ts";
8import type { Queue } from "./Queue.ts";
9import { QueueSink, type QueueSinkEntry } from "./QueueSink.ts";
10import { SendMessageBatch } from "./SendMessageBatch.ts";
11
12const encoder = new TextEncoder();
13
14/**
15 * Select the original entries referenced by `Failed` results with the given
16 * `SenderFault` classification, preserving input order. The sink assigns
17 * `Id = String(index)` per call, so failed ids map back positionally.
18 */
19const selectFailed = (
20 failed: readonly sqs.BatchResultErrorEntry[] | undefined,
21 batch: readonly QueueSinkEntry[],
22 senderFault: boolean,
23): readonly QueueSinkEntry[] => {
24 if (failed === undefined || failed.length === 0) {
25 return [];
26 }
27 const indices = new Set(
28 failed
29 .filter((entry) => entry.SenderFault === senderFault)
30 .map((entry) => Number(entry.Id)),
31 );
32 return batch.filter((_, index) => indices.has(index));
33};
34
35export const QueueSinkHttp = Layer.effect(
36 QueueSink,
37 Effect.gen(function* () {
38 const sendMessageBatch = yield* SendMessageBatch;
39
40 return Effect.fn(function* (queue: Queue) {
41 if (!globalThis.__ALCHEMY_RUNTIME__) {
42 const host = yield* Binding.Host;
43 if (isBindingHost(host)) {
44 yield* host.bind`Allow(${host}, AWS.SQS.QueueSink(${queue}))`({
45 policyStatements: [
46 {
47 Effect: "Allow",
48 Action: ["sqs:SendMessage", "sqs:SendMessageBatch"],
49 Resource: [Output.interpolate`${queue.queueArn}`],
50 },
51 ],
52 });
53 }
54 }
55 const sendBatch = yield* sendMessageBatch(queue);
56 return makeBatchedSink<
57 QueueSinkEntry,
58 sqs.SendMessageBatchResult,
59 sqs.SendMessageBatchError
60 >({
61 maxRecords: 10,
62 maxBytes: 262_144,
63 sizeOf: (entry) => encoder.encode(entry.MessageBody).length,
64 send: (batch) =>
65 sendBatch({
66 Entries: batch.map((entry, index) => ({
67 ...entry,
68 Id: `${index}`,
69 })),
70 }),
71 // SenderFault=false failures (throttling, internal errors) are
72 // transient — re-submit them on the bounded schedule.
73 unprocessed: (out, batch) => selectFailed(out.Failed, batch, false),
74 // SenderFault=true failures are permanent — drop and surface.
75 rejected: (out, batch) => selectFailed(out.Failed, batch, true),
76 });
77 });
78 }),
79);