1import type * as Firehose from "@distilled.cloud/aws/firehose";
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 { DeliveryStream } from "./DeliveryStream.ts";
7
8/**
9 * The raw distilled `PutRecordBatch` entry. Callers stay in control of the
10 * record payload (including any delimiters — Firehose does not add newlines
11 * between records).
12 */
13export type DeliveryStreamSinkRecord = Firehose.Record;
14
15export type DeliveryStreamSinkError =
16 | Firehose.PutRecordBatchError
17 | BatchRetryExhaustedError<DeliveryStreamSinkRecord>;
18
19/**
20 * A batching sink over Firehose `PutRecordBatch` (500 records / 4 MiB per
21 * call).
22 *
23 * Each input element is a raw `Firehose.Record` (`{ Data: Uint8Array }`), so
24 * callers stay in control of encoding and record framing.
25 *
26 * Even a 200 response can carry per-record failures (`FailedPutCount > 0`,
27 * per-record `ErrorCode` — `ServiceUnavailableException` / internal failure).
28 * All of them are transient, so the failed subset is re-submitted in input
29 * order on a bounded schedule; exhausting retries fails the sink with a typed
30 * `BatchRetryExhaustedError` carrying the stranded records.
31 *
32 * ### Streaming Records
33 * **Example:** Run a Stream of Records into the Delivery Stream
34 * ```typescript
35 * // init — bind the sink (provide AWS.Firehose.DeliveryStreamSinkHttp on the Function)
36 * const sink = yield* AWS.Firehose.DeliveryStreamSink(deliveryStream);
37 *
38 * return {
39 * fetch: Effect.gen(function* () {
40 * // runtime — stream newline-framed records into Firehose
41 * yield* Stream.fromIterable(lines).pipe(
42 * Stream.map((line) => ({
43 * Data: new TextEncoder().encode(`${line}\n`),
44 * })),
45 * Stream.run(sink),
46 * );
47 * return HttpServerResponse.json({ ok: true });
48 * }),
49 * };
50 * ```
51 *
52 * @binding
53 */
54export interface DeliveryStreamSink extends Binding.Service<
55 DeliveryStreamSink,
56 "AWS.Firehose.DeliveryStreamSink",
57 (
58 deliveryStream: DeliveryStream,
59 ) => Effect.Effect<
60 Sink.Sink<
61 void,
62 DeliveryStreamSinkRecord,
63 readonly DeliveryStreamSinkRecord[],
64 DeliveryStreamSinkError
65 >
66 >
67> {}
68
69export const DeliveryStreamSink = Binding.Service<DeliveryStreamSink>(
70 "AWS.Firehose.DeliveryStreamSink",
71);