1import type * as DynamoDB from "@distilled.cloud/aws/dynamodb";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Binding from "../../Binding.ts";
5import { makeBatchedSink } from "../internal/BatchedSink.ts";
6import { isBindingHost } from "../Lambda/Function.ts";
7import { BatchWriteItem } from "./BatchWriteItem.ts";
8import type { Table } from "./Table.ts";
9import { TableSink, type TableSinkEntry } from "./TableSink.ts";
10
11const encoder = new TextEncoder();
12
13/**
14 * Map the `UnprocessedItems` echoed by `BatchWriteItem` back onto the
15 * original batch entries, preserving input order. The response is
16 * deserialized (identity is lost), so entries are matched by canonical JSON
17 * as a multiset — DynamoDB rejects batches containing duplicate keys, so a
18 * multiset over serialized entries is unambiguous in practice. If any echoed
19 * entry fails to match (e.g. key-order drift in the echo), fall back to the
20 * echoed entries verbatim so nothing is ever silently dropped from the retry.
21 */
22const selectUnprocessed = (
23 unprocessed:
24 | { [key: string]: DynamoDB.WriteRequest[] | undefined }
25 | undefined,
26 batch: readonly TableSinkEntry[],
27): readonly TableSinkEntry[] => {
28 // The sink writes a single table, so at most one key is present.
29 const echoed = Object.values(unprocessed ?? {}).flatMap(
30 (requests) => requests ?? [],
31 );
32 if (echoed.length === 0) {
33 return [];
34 }
35 const remaining = new Map<string, number>();
36 for (const request of echoed) {
37 const key = JSON.stringify(request);
38 remaining.set(key, (remaining.get(key) ?? 0) + 1);
39 }
40 const matched = batch.filter((entry) => {
41 const key = JSON.stringify(entry);
42 const count = remaining.get(key) ?? 0;
43 if (count === 0) {
44 return false;
45 }
46 remaining.set(key, count - 1);
47 return true;
48 });
49 return matched.length === echoed.length ? matched : echoed;
50};
51
52export const TableSinkHttp = Layer.effect(
53 TableSink,
54 Effect.gen(function* () {
55 const batchWriteItem = yield* BatchWriteItem;
56
57 return Effect.fn(function* (table: Table) {
58 if (!globalThis.__ALCHEMY_RUNTIME__) {
59 const host = yield* Binding.Host;
60 if (isBindingHost(host)) {
61 yield* host.bind`Allow(${host}, AWS.DynamoDB.TableSink(${table}))`({
62 policyStatements: [
63 {
64 Effect: "Allow",
65 Action: ["dynamodb:BatchWriteItem"],
66 Resource: [table.tableArn],
67 },
68 ],
69 });
70 }
71 }
72 const write = yield* batchWriteItem(table);
73 return makeBatchedSink<
74 TableSinkEntry,
75 DynamoDB.BatchWriteItemOutput,
76 DynamoDB.BatchWriteItemError
77 >({
78 maxRecords: 25,
79 maxBytes: 16_777_216,
80 sizeOf: (request) => encoder.encode(JSON.stringify(request)).length,
81 send: (batch) =>
82 write({ RequestItems: { [table.LogicalId]: [...batch] } }),
83 // UnprocessedItems are transient (throttling, internal errors) —
84 // re-submit them on the bounded schedule.
85 unprocessed: (out, batch) =>
86 selectUnprocessed(out.UnprocessedItems, batch),
87 });
88 });
89 }),
90);