1import type * as sns from "@distilled.cloud/aws/sns";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Binding from "../../Binding.ts";
5import { makeBatchedSink } from "../internal/BatchedSink.ts";
6import { isBindingHost } from "../Lambda/Function.ts";
7import { PublishBatch } from "./PublishBatch.ts";
8import type { Topic } from "./Topic.ts";
9import { TopicSink, type TopicSinkEntry } from "./TopicSink.ts";
10
11const encoder = new TextEncoder();
12
13/**
14 * Select the original entries referenced by `Failed` results with the given
15 * `SenderFault` classification, preserving input order. The sink assigns
16 * `Id = String(index)` per call, so failed ids map back positionally.
17 */
18const selectFailed = (
19 failed: readonly sns.BatchResultErrorEntry[] | undefined,
20 batch: readonly TopicSinkEntry[],
21 senderFault: boolean,
22): readonly TopicSinkEntry[] => {
23 if (failed === undefined || failed.length === 0) {
24 return [];
25 }
26 const indices = new Set(
27 failed
28 .filter((entry) => entry.SenderFault === senderFault)
29 .map((entry) => Number(entry.Id)),
30 );
31 return batch.filter((_, index) => indices.has(index));
32};
33
34export const TopicSinkHttp = Layer.effect(
35 TopicSink,
36 Effect.gen(function* () {
37 const publishBatch = yield* PublishBatch;
38
39 return Effect.fn(function* (topic: Topic) {
40 if (!globalThis.__ALCHEMY_RUNTIME__) {
41 const host = yield* Binding.Host;
42 if (isBindingHost(host)) {
43 yield* host.bind`Allow(${host}, AWS.SNS.TopicSink(${topic}))`({
44 policyStatements: [
45 {
46 Effect: "Allow",
47 Action: ["sns:Publish"],
48 Resource: [topic.topicArn],
49 },
50 ],
51 });
52 }
53 }
54 const publish = yield* publishBatch(topic);
55
56 return makeBatchedSink<
57 TopicSinkEntry,
58 sns.PublishBatchResponse,
59 sns.PublishBatchError
60 >({
61 maxRecords: 10,
62 maxBytes: 262_144,
63 sizeOf: (entry) => encoder.encode(entry.Message).length,
64 send: (batch) =>
65 publish({
66 PublishBatchRequestEntries: 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);