1import type * as DynamoDB from "@distilled.cloud/aws/dynamodb";
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 { Table } from "./Table.ts";
7
8/**
9 * A raw distilled `WriteRequest` — exactly one of `PutRequest` or
10 * `DeleteRequest`. Entries are NOT auto-marshalled: callers construct the
11 * `AttributeValue` maps themselves, keeping full control over item shape,
12 * ordering, and partitioning.
13 */
14export interface TableSinkEntry extends DynamoDB.WriteRequest {}
15
16export type TableSinkError =
17 | DynamoDB.BatchWriteItemError
18 | BatchRetryExhaustedError<TableSinkEntry>;
19
20/**
21 * A batching sink over DynamoDB `BatchWriteItem` (25 write requests / 16 MB
22 * per call). Entries the API echoes back in `UnprocessedItems` (throttling,
23 * internal errors) are re-submitted on a bounded schedule; exhausting retries
24 * fails the sink with a typed `BatchRetryExhaustedError` carrying the
25 * stranded entries.
26 *
27 * Sinks are request-scoped consumers: acquire the sink during the Function's
28 * init, drive it with `Stream.run` inside a handler, and let it drain fully
29 * before the handler returns.
30 *
31 * ### Streaming Writes
32 * **Example:** Stream Put Requests into a Table
33 * ```typescript
34 * const sink = yield* AWS.DynamoDB.TableSink(table);
35 *
36 * yield* Stream.fromIterable(records).pipe(
37 * Stream.map((record): AWS.DynamoDB.TableSinkEntry => ({
38 * PutRequest: {
39 * Item: {
40 * pk: { S: record.pk },
41 * sk: { S: record.sk },
42 * },
43 * },
44 * })),
45 * Stream.run(sink),
46 * );
47 * ```
48 *
49 * **Example:** Stream Delete Requests into a Table
50 * ```typescript
51 * yield* Stream.fromIterable(keys).pipe(
52 * Stream.map((key): AWS.DynamoDB.TableSinkEntry => ({
53 * DeleteRequest: {
54 * Key: {
55 * pk: { S: key.pk },
56 * sk: { S: key.sk },
57 * },
58 * },
59 * })),
60 * Stream.run(sink),
61 * );
62 * ```
63 *
64 * @binding
65 */
66export interface TableSink extends Binding.Service<
67 TableSink,
68 "AWS.DynamoDB.TableSink",
69 (
70 table: Table,
71 ) => Effect.Effect<
72 Sink.Sink<void, TableSinkEntry, readonly TableSinkEntry[], TableSinkError>
73 >
74> {}
75
76export const TableSink = Binding.Service<TableSink>("AWS.DynamoDB.TableSink");