| 1 | import type * as sns from "@distilled.cloud/aws/sns"; |
| 2 | import type * as Effect from "effect/Effect"; |
| 3 | import type * as Sink from "effect/Sink"; |
| 4 | import * as Binding from "../../Binding.ts"; |
| 5 | import type { BatchRetryExhaustedError } from "../internal/BatchedSink.ts"; |
| 6 | import type { Topic } from "./Topic.ts"; |
| 7 | |
| 8 | /** |
| 9 | * A raw `PublishBatchRequestEntry` minus the batch-correlation `Id`, which |
| 10 | * the sink assigns per API call. Callers stay in control of `Message`, |
| 11 | * `Subject`, `MessageGroupId`, `MessageDeduplicationId`, attributes, etc. |
| 12 | */ |
| 13 | export interface TopicSinkEntry extends Omit< |
| 14 | sns.PublishBatchRequestEntry, |
| 15 | "Id" |
| 16 | > {} |
| 17 | |
| 18 | export type TopicSinkError = |
| 19 | | sns.PublishBatchError |
| 20 | | BatchRetryExhaustedError<TopicSinkEntry>; |
| 21 | |
| 22 | /** |
| 23 | * A batching sink over SNS `PublishBatch` (10 entries / 256 KiB per call). |
| 24 | * Per-entry failures with `SenderFault: false` (throttling, internal errors) |
| 25 | * are retried on a bounded schedule; `SenderFault: true` failures are |
| 26 | * permanent and dropped. Exhausting retries fails the sink with a typed |
| 27 | * `BatchRetryExhaustedError` carrying the stranded entries. |
| 28 | * |
| 29 | * The binding grants the host function `sns:Publish` on the topic. Provide |
| 30 | * the `TopicSinkHttp` layer (which itself needs `PublishBatchHttp`) on the |
| 31 | * Function to implement the binding. |
| 32 | * ### Streaming Messages into a Topic |
| 33 | * **Example:** Run a Stream into a Topic |
| 34 | * ```typescript |
| 35 | * // init (provide SNS.TopicSinkHttp + SNS.PublishBatchHttp on the Function) |
| 36 | * const sink = yield* SNS.TopicSink(topic); |
| 37 | * |
| 38 | * // runtime: batching, size limits, and transient-failure retry are handled |
| 39 | * // by the sink — each element is a PublishBatchRequestEntry minus `Id`. |
| 40 | * yield* Stream.fromIterable(messages).pipe( |
| 41 | * Stream.map((message) => ({ Message: message })), |
| 42 | * Stream.run(sink), |
| 43 | * ); |
| 44 | * ``` |
| 45 | * |
| 46 | * @binding |
| 47 | */ |
| 48 | export interface TopicSink extends Binding.Service< |
| 49 | TopicSink, |
| 50 | "AWS.SNS.TopicSink", |
| 51 | ( |
| 52 | topic: Topic, |
| 53 | ) => Effect.Effect< |
| 54 | Sink.Sink<void, TopicSinkEntry, readonly TopicSinkEntry[], TopicSinkError> |
| 55 | > |
| 56 | > {} |
| 57 | |
| 58 | export const TopicSink = Binding.Service<TopicSink>("AWS.SNS.TopicSink"); |