1import type * as TSW from "@distilled.cloud/aws/timestream-write";
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 Timestream `Record`. Callers stay in control of
10 * `Dimensions`, `MeasureName`, `Time`, `Version`, etc. — the sink performs no
11 * auto-marshalling.
12 */
13export type RecordsSinkRecord = TSW.Record;
14
15export interface RecordsSinkProps {
16 /**
17 * Attributes shared by every record written through this sink (dimensions,
18 * measure name, time defaults, version). Sent as `CommonAttributes` on each
19 * `WriteRecords` call; Timestream merges them into every record server-side.
20 */
21 readonly commonAttributes?: TSW.Record;
22}
23
24export type RecordsSinkError =
25 | Exclude<
26 TSW.WriteRecordsError | TSW.DescribeEndpointsError,
27 TSW.RejectedRecordsException
28 >
29 | BatchRetryExhaustedError<RecordsSinkRecord>;
30
31/**
32 * A batching sink over Timestream `WriteRecords` (100 records per call).
33 *
34 * Timestream ingests the valid subset of each batch and reports invalid
35 * records positionally via `RejectedRecordsException` (schema conflicts,
36 * timestamps outside the retention window, version conflicts). Rejections are
37 * **permanent** — the sink drops them (surfacing a warning with the rejected
38 * count) and keeps draining; there is no transient per-record failure mode to
39 * retry.
40 *
41 * Provide `Timestream.RecordsSinkHttp` on the Function to implement the
42 * binding.
43 *
44 * ### Streaming Records
45 * **Example:** Stream records into a table
46 * ```typescript
47 * // init — bind the sink to the table; shared attributes are sent once per
48 * // batch as CommonAttributes and merged into every record server-side
49 * const sink = yield* Timestream.RecordsSink(table, {
50 * commonAttributes: {
51 * MeasureName: "cpu",
52 * MeasureValueType: "DOUBLE",
53 * TimeUnit: "MILLISECONDS",
54 * },
55 * });
56 *
57 * // runtime — drain a stream through the sink (batched 100 records/call)
58 * yield* Stream.fromIterable(
59 * samples.map((s) => ({
60 * Dimensions: [{ Name: "host", Value: s.host }],
61 * MeasureValue: `${s.value}`,
62 * Time: `${s.time}`,
63 * })),
64 * ).pipe(Stream.run(sink));
65 * ```
66 *
67 * @binding
68 */
69export interface RecordsSink extends Binding.Service<
70 RecordsSink,
71 "AWS.Timestream.RecordsSink",
72 (
73 table: Table,
74 props?: RecordsSinkProps,
75 ) => Effect.Effect<
76 Sink.Sink<
77 void,
78 RecordsSinkRecord,
79 readonly RecordsSinkRecord[],
80 RecordsSinkError
81 >
82 >
83> {}
84
85export const RecordsSink = Binding.Service<RecordsSink>(
86 "AWS.Timestream.RecordsSink",
87);