1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import { describe, expect } from "alchemy-test";
4import * as Effect from "effect/Effect";
5import * as Schedule from "effect/Schedule";
6import * as HttpClient from "effect/http/HttpClient";
7import * as HttpClientRequest from "effect/http/HttpClientRequest";
8
9import TimestreamSinkFunctionLive, {
10 TimestreamSinkFunction,
11} from "./sink-handler";
12
13const { test } = Test.make({ providers: AWS.providers() });
14
15const postJson = (baseUrl: string, pathname: string, body: unknown) =>
16 HttpClient.execute(
17 HttpClientRequest.post(`${baseUrl}${pathname}`).pipe(
18 HttpClientRequest.bodyJsonUnsafe(body),
19 ),
20 ).pipe(
21 Effect.flatMap((response) =>
22 response.status === 200
23 ? Effect.succeed(response)
24 : Effect.fail(new Error(`${pathname} not ready: ${response.status}`)),
25 ),
26 // Retry through function-URL cold start / IAM propagation.
27 Effect.retry({
28 schedule: Schedule.max([
29 Schedule.exponential("1 second"),
30 Schedule.recurs(8),
31 ]),
32 }),
33 );
34
35const countForHost = (baseUrl: string, host: string) =>
36 HttpClient.get(`${baseUrl}/count?host=${host}`).pipe(
37 Effect.flatMap((response) =>
38 response.status === 200
39 ? response.json
40 : Effect.fail(new Error(`count failed: ${response.status}`)),
41 ),
42 Effect.map((body) =>
43 Number(
44 (body as { rows: Array<{ Data: Array<{ ScalarValue?: string }> }> })
45 .rows[0]?.Data[0]?.ScalarValue ?? "0",
46 ),
47 ),
48 );
49
50// Amazon Timestream for LiveAnalytics is closed to new AWS customers; the
51// testing account receives the typed `TimestreamNotOnboarded` gate on every
52// operation (asserted ungated in Database.test.ts). The sink E2E — stream
53// records through RecordsSink, read them back out-of-band via the Query
54// binding — is therefore gated behind AWS_TEST_TIMESTREAM=1 so an onboarded
55// account can run it unchanged.
56describe(
57 "AWS.Timestream.RecordsSink",
58 {
59 tags: [
60 "provider:aws",
61 "provider:aws:lambda",
62 "provider:aws:timestream",
63 "live",
64 ],
65 },
66 () => {
67 test.provider.skipIf(!process.env.AWS_TEST_TIMESTREAM)(
68 "Lambda streams records through the sink; rejected records are dropped",
69 (stack) =>
70 Effect.gen(function* () {
71 const { functionUrl } = yield* stack.deploy(
72 Effect.gen(function* () {
73 return yield* TimestreamSinkFunction;
74 }).pipe(Effect.provide(TimestreamSinkFunctionLive)),
75 );
76 expect(functionUrl).toBeTruthy();
77 const baseUrl = functionUrl!.replace(/\/+$/, "");
78
79 // 150 records > the 100-record WriteRecords limit, so the sink must
80 // split the stream into two sequential batches.
81 yield* postJson(baseUrl, "/sink", { host: "sink-bulk", count: 150 });
82
83 // Memory-store writes are queryable near-immediately; retry a few
84 // times regardless.
85 const bulkCount = yield* countForHost(baseUrl, "sink-bulk").pipe(
86 Effect.flatMap((count) =>
87 count >= 150
88 ? Effect.succeed(count)
89 : Effect.fail(new Error(`only ${count} rows counted yet`)),
90 ),
91 Effect.retry({
92 schedule: Schedule.max([
93 Schedule.spaced("2 seconds"),
94 Schedule.recurs(10),
95 ]),
96 }),
97 );
98 expect(bulkCount).toBe(150);
99
100 // Partial failure: the out-of-retention record is permanently
101 // rejected (RejectedRecordsException); the sink drops it, lands the
102 // two valid records, and the handler still returns 200.
103 yield* postJson(baseUrl, "/sink-rejects", { host: "sink-rejects" });
104
105 const rejectsCount = yield* countForHost(
106 baseUrl,
107 "sink-rejects",
108 ).pipe(
109 Effect.flatMap((count) =>
110 count >= 2
111 ? Effect.succeed(count)
112 : Effect.fail(new Error(`only ${count} rows counted yet`)),
113 ),
114 Effect.retry({
115 schedule: Schedule.max([
116 Schedule.spaced("2 seconds"),
117 Schedule.recurs(10),
118 ]),
119 }),
120 );
121 expect(rejectsCount).toBe(2);
122
123 yield* stack.destroy();
124 }),
125 { timeout: 600_000 },
126 );
127 },
128);