1import type * as sqs from "@distilled.cloud/aws/sqs";
2import type * as Effect from "effect/Effect";
3import type * as Sink from "effect/Sink";
4import * as Binding from "../../Binding.ts";
5import type { BatchRetryExhaustedError } from "../internal/BatchedSink.ts";
6import type { Queue } from "./Queue.ts";
7
8/**
9 * A raw `SendMessageBatchRequestEntry` minus the batch-correlation `Id`,
10 * which the sink assigns per API call. Callers stay in control of
11 * `MessageBody`, `MessageGroupId`, `MessageDeduplicationId`, attributes, etc.
12 */
13export interface QueueSinkEntry extends Omit<
14 sqs.SendMessageBatchRequestEntry,
15 "Id"
16> {}
17
18export type QueueSinkError =
19 | sqs.SendMessageBatchError
20 | BatchRetryExhaustedError<QueueSinkEntry>;
21
22/**
23 * A batching sink over SQS `SendMessageBatch` (10 entries / 256 KiB per
24 * call). Per-entry failures with `SenderFault: false` (throttling, internal
25 * errors) 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 `sqs:SendMessage` and
30 * `sqs:SendMessageBatch` on the queue. Provide the `QueueSinkHttp` layer
31 * (which itself needs `SendMessageBatchHttp`) on the Function to implement
32 * the binding.
33 * ### Streaming Messages into a Queue
34 * **Example:** Run a Stream into a Queue
35 * ```typescript
36 * // init (provide SQS.QueueSinkHttp + SQS.SendMessageBatchHttp on the Function)
37 * const sink = yield* SQS.QueueSink(queue);
38 *
39 * // runtime: batching, size limits, and transient-failure retry are handled
40 * // by the sink — each element is a SendMessageBatchRequestEntry minus `Id`.
41 * yield* Stream.fromIterable(messages).pipe(
42 * Stream.map((message) => ({ MessageBody: message })),
43 * Stream.run(sink),
44 * );
45 * ```
46 *
47 * **Example:** Forward Event-Source Records into a Result Queue
48 * ```typescript
49 * const sink = yield* SQS.QueueSink(resultQueue);
50 *
51 * yield* SQS.consumeQueueMessages(sourceQueue, (records) =>
52 * records.pipe(
53 * Stream.map((record) => ({ MessageBody: record.body })),
54 * Stream.run(sink),
55 * Effect.orDie,
56 * ),
57 * );
58 * ```
59 *
60 * @binding
61 */
62export interface QueueSink extends Binding.Service<
63 QueueSink,
64 "AWS.SQS.QueueSink",
65 (
66 queue: Queue,
67 ) => Effect.Effect<
68 Sink.Sink<void, QueueSinkEntry, readonly QueueSinkEntry[], QueueSinkError>
69 >
70> {}
71
72export const QueueSink = Binding.Service<QueueSink>("AWS.SQS.QueueSink");