1import type * as TSW from "@distilled.cloud/aws/timestream-write";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Binding from "../../Binding.ts";
5import * as Output from "../../Output.ts";
6import { makeBatchedSink } from "../internal/BatchedSink.ts";
7import { isBindingHost } from "../Lambda/Function.ts";
8import {
9 RecordsSink,
10 type RecordsSinkProps,
11 type RecordsSinkRecord,
12} from "./RecordsSink.ts";
13import type { Table } from "./Table.ts";
14import { WriteRecords } from "./WriteRecords.ts";
15
16/**
17 * Normalized `WriteRecords` outcome: which positions of the submitted batch
18 * Timestream permanently rejected. A fully-ingested call has no rejections.
19 */
20interface WriteOutcome {
21 readonly rejectedIndices: ReadonlySet<number>;
22}
23
24const noRejections: WriteOutcome = { rejectedIndices: new Set() };
25
26type WriteSendError = Exclude<
27 TSW.WriteRecordsError | TSW.DescribeEndpointsError,
28 TSW.RejectedRecordsException
29>;
30
31export const RecordsSinkHttp = Layer.effect(
32 RecordsSink,
33 Effect.gen(function* () {
34 const writeRecords = yield* WriteRecords;
35
36 return Effect.fn(function* (table: Table, props?: RecordsSinkProps) {
37 if (!globalThis.__ALCHEMY_RUNTIME__) {
38 const host = yield* Binding.Host;
39 if (isBindingHost(host)) {
40 yield* host.bind`Allow(${host}, AWS.Timestream.RecordsSink(${table}))`(
41 {
42 policyStatements: [
43 {
44 Effect: "Allow",
45 Action: ["timestream:WriteRecords"],
46 Resource: [Output.interpolate`${table.tableArn}`],
47 },
48 // Timestream requires endpoint discovery; the ingest endpoint
49 // is resolved at runtime via DescribeEndpoints, which is not
50 // scoped to a resource.
51 {
52 Effect: "Allow",
53 Action: ["timestream:DescribeEndpoints"],
54 Resource: ["*"],
55 },
56 ],
57 },
58 );
59 }
60 }
61 const write = yield* writeRecords(table);
62 const commonAttributes = props?.commonAttributes;
63 return makeBatchedSink<RecordsSinkRecord, WriteOutcome, WriteSendError>({
64 maxRecords: 100,
65 send: (batch) =>
66 write({
67 Records: [...batch],
68 ...(commonAttributes === undefined
69 ? {}
70 : { CommonAttributes: commonAttributes }),
71 }).pipe(
72 Effect.map(() => noRejections),
73 // Timestream ingests the valid subset and reports invalid records
74 // positionally via RejectedRecordsException. Rejections are
75 // permanent (schema conflicts, out-of-retention timestamps,
76 // version conflicts) — drop them, never retry.
77 Effect.catchTag("RejectedRecordsException", (error) =>
78 Effect.succeed<WriteOutcome>({
79 rejectedIndices: new Set(
80 (error.RejectedRecords ?? []).flatMap(
81 (rejected: TSW.RejectedRecord) =>
82 rejected.RecordIndex === undefined
83 ? []
84 : [rejected.RecordIndex],
85 ),
86 ),
87 }),
88 ),
89 ),
90 rejected: (out, batch) =>
91 out.rejectedIndices.size === 0
92 ? []
93 : batch.filter((_, index) => out.rejectedIndices.has(index)),
94 });
95 });
96 }),
97);