1import type * as Kinesis from "@distilled.cloud/aws/kinesis";
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 { Stream } from "./Stream.ts";
7
8export type StreamSinkRecord = Kinesis.PutRecordsRequestEntry;
9
10export type StreamSinkError =
11 | Kinesis.PutRecordsError
12 | BatchRetryExhaustedError<StreamSinkRecord>;
13
14/**
15 * A partition-aware sink for batching `PutRecords` requests into a stream
16 * (500 records / 5 MiB per call).
17 *
18 * Each input element is a raw `PutRecordsRequestEntry`, so callers stay in
19 * control of `PartitionKey` and optional `ExplicitHashKey`.
20 *
21 * Records the API reports as failed (`FailedRecordCount > 0`, per-record
22 * `ErrorCode` — throughput exceeded or internal failure) are re-submitted on
23 * a bounded schedule; exhausting retries fails the sink with a typed
24 * `BatchRetryExhaustedError` carrying the stranded records.
25 *
26 * Provide the implementation with `Effect.provide(AWS.Kinesis.StreamSinkHttp)`.
27 * ### Writing Streams of Records
28 * **Example:** Run an Effect Stream into a Kinesis Stream
29 * ```typescript
30 * // init — bind the sink to the stream
31 * const sink = yield* AWS.Kinesis.StreamSink(stream);
32 *
33 * // runtime — batches into PutRecords calls of up to 500 records / 5 MiB
34 * yield* Stream.fromIterable(
35 * orders.map((order) => ({
36 * PartitionKey: order.id,
37 * Data: new TextEncoder().encode(JSON.stringify(order)),
38 * })),
39 * ).pipe(Stream.run(sink));
40 * ```
41 *
42 * @binding
43 */
44export interface StreamSink extends Binding.Service<
45 StreamSink,
46 "AWS.Kinesis.StreamSink",
47 (
48 stream: Stream,
49 ) => Effect.Effect<
50 Sink.Sink<
51 void,
52 StreamSinkRecord,
53 readonly StreamSinkRecord[],
54 StreamSinkError
55 >
56 >
57> {}
58
59export const StreamSink = Binding.Service<StreamSink>("AWS.Kinesis.StreamSink");