1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as eventbridge from "@distilled.cloud/aws/eventbridge";
5import { describe, expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import * as HttpClient from "effect/http/HttpClient";
10import * as HttpClientRequest from "effect/http/HttpClientRequest";
11import SchemasTestFunctionLive, { SchemasTestFunction } from "./handler";
12
13const testOptions = { providers: AWS.providers() };
14const { test, beforeAll, afterAll } = Test.make(testOptions);
15const sharedStack = Core.scratchStack(testOptions, "SchemasBindings");
16
17// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
18// over 60s on a fresh deploy.
19const readinessPolicy = Schedule.max([
20 Schedule.fixed("2 seconds"),
21 Schedule.recurs(75),
22]);
23
24let baseUrl: string;
25let functionArn: string;
26
27class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
28 readonly status: number;
29 readonly body: string;
30}> {}
31
32// The shared Lambda fixture occasionally answers a transient 5xx under load
33// (cold re-init, IAM propagation on the freshly attached policy that the
34// handler's `Effect.orDie` surfaces as a 500). Retry only 5xx; a genuine
35// 4xx/assertion failure surfaces immediately.
36const send = (request: HttpClientRequest.HttpClientRequest) =>
37 HttpClient.execute(request).pipe(
38 Effect.flatMap((response) =>
39 response.status >= 500
40 ? response.text.pipe(
41 Effect.flatMap((body) =>
42 Effect.fail(
43 new TransientUpstream({ status: response.status, body }),
44 ),
45 ),
46 )
47 : Effect.succeed(response),
48 ),
49 Effect.retry({
50 while: (e) => e._tag === "TransientUpstream",
51 schedule: Schedule.max([
52 Schedule.exponential("500 millis"),
53 Schedule.recurs(6),
54 ]),
55 }),
56 );
57
58const getJson = (path: string) =>
59 send(HttpClientRequest.get(`${baseUrl}${path}`)).pipe(
60 Effect.flatMap((r) => r.json),
61 );
62
63const postJson = (path: string) =>
64 send(HttpClientRequest.post(`${baseUrl}${path}`)).pipe(
65 Effect.flatMap((r) => r.json),
66 );
67
68describe.sequential(
69 "Schemas Bindings",
70 {
71 tags: [
72 "provider:aws",
73 "provider:aws:eventbridge",
74 "provider:aws:lambda",
75 "provider:aws:schemas",
76 "live",
77 ],
78 },
79 () => {
80 beforeAll(
81 Effect.gen(function* () {
82 yield* Effect.logInfo(
83 "Schemas test setup: destroying previous resources",
84 );
85 yield* sharedStack.destroy();
86
87 yield* Effect.logInfo("Schemas test setup: deploying fixture");
88 const attrs = yield* sharedStack.deploy(
89 Effect.gen(function* () {
90 return yield* SchemasTestFunction;
91 }).pipe(Effect.provide(SchemasTestFunctionLive)),
92 );
93
94 expect(attrs.functionUrl).toBeTruthy();
95 baseUrl = attrs.functionUrl!.replace(/\/+$/, "");
96 functionArn = attrs.functionArn;
97
98 const readinessUrl = `${baseUrl}/bindings`;
99 yield* Effect.logInfo(
100 `Schemas test setup: probing readiness at ${readinessUrl}`,
101 );
102 yield* HttpClient.get(readinessUrl).pipe(
103 Effect.flatMap((response) =>
104 response.status === 200
105 ? Effect.succeed(response)
106 : Effect.fail(
107 new Error(`Function not ready: ${response.status}`),
108 ),
109 ),
110 Effect.tapError((error) =>
111 Effect.logWarning(
112 `Schemas test setup: fixture not ready yet (${String(error)})`,
113 ),
114 ),
115 Effect.retry({ schedule: readinessPolicy }),
116 );
117 }),
118 { timeout: 240_000 },
119 );
120
121 afterAll.skipIf(!!process.env.NO_DESTROY)(sharedStack.destroy(), {
122 timeout: 120_000,
123 });
124
125 describe("binding registration", () => {
126 test.provider(
127 "all ten capabilities initialize in the runtime",
128 (_stack) =>
129 Effect.gen(function* () {
130 const response = (yield* getJson("/bindings")) as {
131 bound: string[];
132 };
133 expect(response.bound).toHaveLength(10);
134 }),
135 );
136 });
137
138 describe("DescribeSchema", () => {
139 test.provider(
140 "reads the bound schema's document (injected registry + schema names)",
141 (_stack) =>
142 Effect.gen(function* () {
143 const response = (yield* getJson("/schema")) as {
144 name: string;
145 type: string;
146 version: string;
147 hasContent: boolean;
148 };
149 expect(response.name).toBeTruthy();
150 expect(response.type).toBe("OpenApi3");
151 expect(response.version).toBe("1");
152 expect(response.hasContent).toBe(true);
153 }),
154 );
155 });
156
157 describe("ExportSchema", () => {
158 test.provider(
159 "surfaces the typed ForbiddenException for custom-registry schemas",
160 (_stack) =>
161 Effect.gen(function* () {
162 // AWS only allows exporting discovered / AWS-managed schemas. The
163 // call reaching the API and failing with the typed tag proves the
164 // binding + IAM wiring end-to-end.
165 const response = (yield* getJson("/export")) as {
166 outcome: string;
167 message?: string;
168 };
169 expect(response.outcome).toBe("forbidden");
170 expect(response.message).toContain("export");
171 }),
172 );
173 });
174
175 describe("ListSchemaVersions", () => {
176 test.provider("lists the schema's single published version", (_stack) =>
177 Effect.gen(function* () {
178 const response = (yield* getJson("/versions")) as {
179 versions: string[];
180 };
181 expect(response.versions).toContain("1");
182 }),
183 );
184 });
185
186 describe("SearchSchemas", () => {
187 test.provider(
188 "finds the schema by keyword in the bound registry",
189 (_stack) =>
190 Effect.gen(function* () {
191 // Search indexing lags schema creation by a few seconds.
192 const response = yield* getJson("/search").pipe(
193 Effect.map((r) => r as { names: string[] }),
194 Effect.repeat({
195 schedule: Schedule.spaced("3 seconds"),
196 until: (r): boolean => (r.names ?? []).length > 0,
197 times: 15,
198 }),
199 );
200 expect(response.names.length).toBeGreaterThanOrEqual(1);
201 }),
202 { timeout: 90_000 },
203 );
204 });
205
206 describe("GetDiscoveredSchema", () => {
207 test.provider("infers a schema from a sample event envelope", (_stack) =>
208 Effect.gen(function* () {
209 const response = (yield* postJson("/discover")) as {
210 hasContent: boolean;
211 };
212 expect(response.hasContent).toBe(true);
213 }),
214 );
215 });
216
217 describe("PutCodeBinding / DescribeCodeBinding / GetCodeBindingSource", () => {
218 test.provider(
219 "generates a code-binding package and downloads it",
220 (_stack) =>
221 Effect.gen(function* () {
222 // Kick off async generation.
223 const put = (yield* postJson("/codebinding")) as { status: string };
224 expect(["CREATE_IN_PROGRESS", "CREATE_COMPLETE"]).toContain(
225 put.status,
226 );
227
228 // Poll until generation completes (usually well under 30s).
229 const status = yield* getJson("/codebinding").pipe(
230 Effect.map((r) => (r as { status: string }).status),
231 Effect.repeat({
232 schedule: Schedule.spaced("3 seconds"),
233 until: (s): boolean => s === "CREATE_COMPLETE",
234 times: 20,
235 }),
236 );
237 expect(status).toBe("CREATE_COMPLETE");
238
239 // The generated zip streams back as Body.
240 const source = (yield* getJson("/codebinding/source")) as {
241 bytes: number;
242 };
243 expect(source.bytes).toBeGreaterThan(0);
244 }),
245 { timeout: 120_000 },
246 );
247 });
248
249 describe("StopDiscoverer / StartDiscoverer", () => {
250 test.provider(
251 "pauses and resumes the bound discoverer (injected discoverer id)",
252 (_stack) =>
253 Effect.gen(function* () {
254 const stopped = (yield* postJson("/discoverer/stop")) as {
255 state: string;
256 };
257 expect(stopped.state).toBe("STOPPED");
258
259 const started = (yield* postJson("/discoverer/start")) as {
260 state: string;
261 };
262 expect(started.state).toBe("STARTED");
263 }),
264 );
265 });
266
267 describe("consumeSchemaEvents", () => {
268 test.provider(
269 "the deploy created an EventBridge rule targeting the function",
270 (_stack) =>
271 Effect.gen(function* () {
272 // Out-of-band via distilled: the fixture's consumeSchemaEvents
273 // must have materialized as a rule on the default bus with the
274 // Lambda as target.
275 const { RuleNames } = yield* eventbridge.listRuleNamesByTarget({
276 TargetArn: functionArn,
277 });
278 expect((RuleNames ?? []).length).toBeGreaterThanOrEqual(1);
279 }),
280 );
281 });
282 },
283);