1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as sd from "@distilled.cloud/aws/servicediscovery";
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 CloudMapTestFunctionLive, { CloudMapTestFunction } from "./handler";
12
13const testOptions = { providers: AWS.providers() };
14const { test, beforeAll, afterAll } = Test.make(testOptions);
15const sharedStack = Core.scratchStack(testOptions, "CloudMapBindings");
16
17const readinessPolicy = Schedule.max([
18 Schedule.fixed("2 seconds"),
19 Schedule.recurs(75),
20]);
21
22let baseUrl: string;
23let serviceId: string;
24let customServiceId: string;
25
26class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
27 readonly status: number;
28 readonly body: string;
29}> {}
30
31// Retry transient 5xx from the shared fixture under parallel-suite load;
32// genuine 4xx failures surface immediately.
33const send = (request: HttpClientRequest.HttpClientRequest) =>
34 HttpClient.execute(request).pipe(
35 Effect.flatMap((response) =>
36 response.status >= 500
37 ? response.text.pipe(
38 Effect.flatMap((body) =>
39 Effect.fail(
40 new TransientUpstream({ status: response.status, body }),
41 ),
42 ),
43 )
44 : Effect.succeed(response),
45 ),
46 Effect.retry({
47 while: (e) => e._tag === "TransientUpstream",
48 schedule: Schedule.max([
49 Schedule.exponential("500 millis"),
50 Schedule.recurs(6),
51 ]),
52 }),
53 );
54
55class CloudMapOperationPending extends Data.TaggedError(
56 "CloudMapOperationPending",
57)<{ readonly operationId: string }> {}
58
59/** Await a Cloud Map async operation out-of-band (bounded). */
60const waitOperation = (operationId: string) =>
61 sd.getOperation({ OperationId: operationId }).pipe(
62 Effect.map((r) => r.Operation),
63 Effect.repeat({
64 schedule: Schedule.spaced("2 seconds"),
65 until: (op) => op?.Status === "SUCCESS" || op?.Status === "FAIL",
66 times: 45,
67 }),
68 Effect.flatMap((op) =>
69 op?.Status === "SUCCESS"
70 ? Effect.void
71 : Effect.fail(new CloudMapOperationPending({ operationId })),
72 ),
73 );
74
75const registerOutOfBand = (instanceId: string, ipv4: string) =>
76 sd
77 .registerInstance({
78 ServiceId: serviceId,
79 InstanceId: instanceId,
80 Attributes: { AWS_INSTANCE_IPV4: ipv4 },
81 })
82 .pipe(
83 Effect.flatMap((r) =>
84 r.OperationId !== undefined
85 ? waitOperation(r.OperationId)
86 : Effect.void,
87 ),
88 );
89
90const deregisterOutOfBand = (instanceId: string, inServiceId?: string) =>
91 sd
92 .deregisterInstance({
93 ServiceId: inServiceId ?? serviceId,
94 InstanceId: instanceId,
95 })
96 .pipe(
97 Effect.flatMap((r) =>
98 r.OperationId !== undefined
99 ? waitOperation(r.OperationId)
100 : Effect.void,
101 ),
102 Effect.catchTag("InstanceNotFound", () => Effect.void),
103 );
104
105interface DiscoveredInstance {
106 instanceId: string;
107 healthStatus: string | undefined;
108 attributes: Record<string, string>;
109}
110
111/** Call the deployed Lambda's /discover route (through the binding). */
112const discoverViaLambda = Effect.gen(function* () {
113 const response = yield* send(
114 HttpClientRequest.get(`${baseUrl}/discover?health=ALL`),
115 );
116 const body = (yield* response.json) as unknown as {
117 instances: DiscoveredInstance[];
118 };
119 return body.instances;
120});
121
122class UnexpectedInstances extends Data.TaggedError("UnexpectedInstances")<{
123 readonly expected: string[];
124 readonly actual: string[];
125}> {}
126
127/**
128 * Poll the Lambda's /discover route until — restricted to the given id
129 * prefix — exactly `expected` instance ids are visible (bounded; the
130 * data-plane view is eventually consistent after register/deregister).
131 */
132const expectDiscovered = (prefix: string, expected: string[]) =>
133 discoverViaLambda.pipe(
134 Effect.flatMap((instances) => {
135 const actual = instances
136 .map((i) => i.instanceId)
137 .filter((id): id is string => id !== undefined)
138 .filter((id) => id.startsWith(prefix))
139 .sort();
140 const want = [...expected].sort();
141 return actual.length === want.length &&
142 actual.every((id, i) => id === want[i])
143 ? Effect.succeed(instances)
144 : Effect.fail(new UnexpectedInstances({ expected: want, actual }));
145 }),
146 Effect.retry({
147 while: (e) => e._tag === "UnexpectedInstances",
148 schedule: Schedule.max([
149 Schedule.spaced("3 seconds"),
150 Schedule.recurs(20),
151 ]),
152 }),
153 );
154
155class InstanceNotVisible extends Data.TaggedError("InstanceNotVisible")<{
156 readonly instanceId: string;
157}> {}
158
159const expectInstanceState = (instanceId: string, present: boolean) =>
160 sd.getInstance({ ServiceId: serviceId, InstanceId: instanceId }).pipe(
161 Effect.map(() => true),
162 Effect.catchTag("InstanceNotFound", () => Effect.succeed(false)),
163 Effect.flatMap((exists) =>
164 exists === present
165 ? Effect.void
166 : Effect.fail(new InstanceNotVisible({ instanceId })),
167 ),
168 Effect.retry({
169 while: (e) => e._tag === "InstanceNotVisible",
170 schedule: Schedule.max([
171 Schedule.spaced("3 seconds"),
172 Schedule.recurs(20),
173 ]),
174 }),
175 );
176
177describe(
178 "CloudMap Bindings",
179 {
180 tags: [
181 "provider:aws",
182 "provider:aws:cloudmap",
183 "provider:aws:ec2",
184 "provider:aws:lambda",
185 "live",
186 ],
187 },
188 () => {
189 beforeAll(
190 Effect.gen(function* () {
191 yield* Effect.logInfo(
192 "CloudMap test setup: destroying previous resources",
193 );
194 yield* sharedStack.destroy();
195
196 yield* Effect.logInfo("CloudMap test setup: deploying fixture");
197 const { functionUrl } = yield* sharedStack.deploy(
198 Effect.gen(function* () {
199 return yield* CloudMapTestFunction;
200 }).pipe(Effect.provide(CloudMapTestFunctionLive)),
201 );
202
203 expect(functionUrl).toBeTruthy();
204 baseUrl = functionUrl!.replace(/\/+$/, "");
205
206 yield* Effect.logInfo(
207 `CloudMap test setup: probing readiness at ${baseUrl}/info`,
208 );
209 // readiness = 200 AND the Output-backed env vars are visible. Lambda
210 // applies the code update and the env-var configuration update
211 // separately, so there is a brief window where the real handler serves
212 // 200 while `FixtureService_*` env vars are still missing.
213 const info = yield* HttpClient.get(`${baseUrl}/info`).pipe(
214 Effect.flatMap((response) =>
215 response.status === 200
216 ? response.json
217 : Effect.fail(
218 new Error(`Function not ready: ${response.status}`),
219 ),
220 ),
221 Effect.flatMap((body) =>
222 (body as { serviceId?: string }).serviceId
223 ? Effect.succeed(body)
224 : Effect.fail(new Error("env vars not yet propagated")),
225 ),
226 Effect.retry({ schedule: readinessPolicy }),
227 );
228 yield* Effect.logInfo(
229 `CloudMap test setup: fixture info = ${JSON.stringify(info)}`,
230 );
231 serviceId = (info as { serviceId: string }).serviceId;
232 customServiceId = (info as { customServiceId: string }).customServiceId;
233 expect(serviceId).toBeTruthy();
234 expect(customServiceId).toBeTruthy();
235 }),
236 { timeout: 300_000 },
237 );
238
239 afterAll.skipIf(!!process.env.NO_DESTROY)(
240 // The out-of-band listInstances/deregisterInstance calls need the AWS
241 // providers (Credentials/HttpClient/Region); afterAll effects run with
242 // R=never, so provide them explicitly (Stack/Stage come from destroy()).
243 Core.withProviders(
244 Effect.gen(function* () {
245 // deregister any instances the tests left behind so the services (and
246 // then the namespace) can delete without ResourceInUse churn
247 for (const cleanupServiceId of [serviceId, customServiceId]) {
248 if (cleanupServiceId !== undefined) {
249 const instances = yield* sd
250 .listInstances({ ServiceId: cleanupServiceId })
251 .pipe(
252 Effect.map((r) => r.Instances ?? []),
253 Effect.catchTag("ServiceNotFound", () => Effect.succeed([])),
254 );
255 yield* Effect.forEach(
256 instances,
257 (instance) =>
258 instance.Id !== undefined
259 ? deregisterOutOfBand(instance.Id, cleanupServiceId)
260 : Effect.void,
261 { concurrency: 4 },
262 );
263 }
264 }
265 yield* sharedStack.destroy();
266 }),
267 testOptions,
268 "CloudMapBindings",
269 ),
270 { timeout: 240_000 },
271 );
272
273 describe("DiscoverInstances", () => {
274 test.provider(
275 "lambda discovers instances registered out-of-band; deregistration converges",
276 (_stack) =>
277 Effect.gen(function* () {
278 // register two instances out-of-band via distilled
279 yield* registerOutOfBand("disc-a", "10.0.0.1");
280 yield* registerOutOfBand("disc-b", "10.0.0.2");
281
282 // the deployed Lambda sees both through the DiscoverInstances binding
283 const instances = yield* expectDiscovered("disc-", [
284 "disc-a",
285 "disc-b",
286 ]);
287 const discA = instances.find((i) => i.instanceId === "disc-a");
288 expect(discA?.attributes.AWS_INSTANCE_IPV4).toBe("10.0.0.1");
289
290 // deregister one; a bounded re-poll sees exactly the survivor
291 yield* deregisterOutOfBand("disc-a");
292 yield* expectDiscovered("disc-", ["disc-b"]);
293
294 // cleanup
295 yield* deregisterOutOfBand("disc-b");
296 yield* expectInstanceState("disc-b", false);
297 }),
298 { timeout: 180_000 },
299 );
300 });
301
302 describe("RegisterInstance", () => {
303 test.provider(
304 "lambda self-registers an instance through the binding",
305 (_stack) =>
306 Effect.gen(function* () {
307 const response = yield* send(
308 HttpClientRequest.bodyJsonUnsafe(
309 HttpClientRequest.post(`${baseUrl}/register`),
310 {
311 instanceId: "bind-reg",
312 attributes: { AWS_INSTANCE_IPV4: "10.0.0.9" },
313 },
314 ),
315 );
316 expect(response.status).toBe(200);
317 const body = (yield* response.json) as { operationId?: string };
318 expect(body.operationId).toBeTruthy();
319
320 // visible out-of-band once the async registration completes
321 yield* expectInstanceState("bind-reg", true);
322 const instance = yield* sd.getInstance({
323 ServiceId: serviceId,
324 InstanceId: "bind-reg",
325 });
326 expect(instance.Instance?.Attributes?.AWS_INSTANCE_IPV4).toBe(
327 "10.0.0.9",
328 );
329
330 // cleanup
331 yield* deregisterOutOfBand("bind-reg");
332 }),
333 { timeout: 180_000 },
334 );
335 });
336
337 describe("DeregisterInstance", () => {
338 test.provider(
339 "lambda deregisters an instance through the binding",
340 (_stack) =>
341 Effect.gen(function* () {
342 yield* registerOutOfBand("bind-dereg", "10.0.0.8");
343 yield* expectInstanceState("bind-dereg", true);
344
345 const response = yield* send(
346 HttpClientRequest.bodyJsonUnsafe(
347 HttpClientRequest.post(`${baseUrl}/deregister`),
348 { instanceId: "bind-dereg" },
349 ),
350 );
351 expect(response.status).toBe(200);
352
353 yield* expectInstanceState("bind-dereg", false);
354 }),
355 { timeout: 180_000 },
356 );
357 });
358
359 describe("GetInstance + ListInstances", () => {
360 test.provider(
361 "lambda reads a registered instance through the control-plane bindings",
362 (_stack) =>
363 Effect.gen(function* () {
364 yield* registerOutOfBand("read-a", "10.0.0.31");
365
366 // GetInstance — full attribute map by id
367 const single = yield* send(
368 HttpClientRequest.get(`${baseUrl}/instance?id=read-a`),
369 );
370 expect(single.status).toBe(200);
371 const singleBody = (yield* single.json) as {
372 instanceId: string;
373 attributes: Record<string, string>;
374 };
375 expect(singleBody.instanceId).toBe("read-a");
376 expect(singleBody.attributes.AWS_INSTANCE_IPV4).toBe("10.0.0.31");
377
378 // ListInstances — control-plane enumeration includes it
379 const listed = yield* send(
380 HttpClientRequest.get(`${baseUrl}/instances`),
381 );
382 expect(listed.status).toBe(200);
383 const listedBody = (yield* listed.json) as {
384 instances: { instanceId: string }[];
385 };
386 expect(listedBody.instances.map((i) => i.instanceId)).toContain(
387 "read-a",
388 );
389
390 // cleanup
391 yield* deregisterOutOfBand("read-a");
392 }),
393 { timeout: 180_000 },
394 );
395 });
396
397 describe("DiscoverInstancesRevision", () => {
398 test.provider(
399 "lambda reads the instance-set revision through the binding",
400 (_stack) =>
401 Effect.gen(function* () {
402 const response = yield* send(
403 HttpClientRequest.get(`${baseUrl}/revision`),
404 );
405 expect(response.status).toBe(200);
406 const body = (yield* response.json) as { revision: number };
407 expect(typeof body.revision).toBe("number");
408 }),
409 { timeout: 120_000 },
410 );
411 });
412
413 describe("GetServiceAttributes", () => {
414 test.provider(
415 "lambda reads the service attributes declared on the Service resource",
416 (_stack) =>
417 Effect.gen(function* () {
418 const response = yield* send(
419 HttpClientRequest.get(`${baseUrl}/service-attributes`),
420 );
421 expect(response.status).toBe(200);
422 const body = (yield* response.json) as {
423 attributes: Record<string, string>;
424 };
425 expect(body.attributes).toEqual({ tier: "backend", version: "1" });
426 }),
427 { timeout: 120_000 },
428 );
429 });
430
431 describe("UpdateInstanceCustomHealthStatus + GetInstancesHealthStatus + GetOperation", () => {
432 test.provider(
433 "lambda registers, awaits the operation, pushes custom health, and reads it back",
434 (_stack) =>
435 Effect.gen(function* () {
436 // register an instance on the custom-health service via the binding
437 const registered = yield* send(
438 HttpClientRequest.bodyJsonUnsafe(
439 HttpClientRequest.post(`${baseUrl}/custom/register`),
440 {
441 instanceId: "cust-1",
442 attributes: { AWS_INSTANCE_IPV4: "10.0.0.41" },
443 },
444 ),
445 );
446 expect(registered.status).toBe(200);
447 const { operationId } = (yield* registered.json) as {
448 operationId: string;
449 };
450 expect(operationId).toBeTruthy();
451
452 // await the async registration THROUGH the GetOperation binding
453 const operation = yield* send(
454 HttpClientRequest.get(`${baseUrl}/operation?id=${operationId}`),
455 ).pipe(
456 Effect.flatMap((r) => r.json),
457 Effect.map((body) => body as { status?: string }),
458 Effect.repeat({
459 schedule: Schedule.spaced("2 seconds"),
460 until: (op): boolean =>
461 op.status === "SUCCESS" || op.status === "FAIL",
462 times: 45,
463 }),
464 );
465 expect(operation.status).toBe("SUCCESS");
466
467 const pushHealth = (status: "HEALTHY" | "UNHEALTHY") =>
468 send(
469 HttpClientRequest.bodyJsonUnsafe(
470 HttpClientRequest.post(`${baseUrl}/custom/health`),
471 { instanceId: "cust-1", status },
472 ),
473 );
474
475 const readHealth = send(
476 HttpClientRequest.get(`${baseUrl}/custom/health-status`),
477 ).pipe(
478 Effect.flatMap((r) => r.json),
479 Effect.map(
480 (body) =>
481 (body as { status: Record<string, string | undefined> })
482 .status["cust-1"],
483 ),
484 );
485
486 const expectHealth = (expected: "HEALTHY" | "UNHEALTHY") =>
487 readHealth.pipe(
488 Effect.repeat({
489 schedule: Schedule.spaced("3 seconds"),
490 until: (status): boolean => status === expected,
491 times: 20,
492 }),
493 Effect.map((status) => expect(status).toBe(expected)),
494 );
495
496 // push UNHEALTHY then HEALTHY and observe both through the bindings
497 yield* pushHealth("UNHEALTHY");
498 yield* expectHealth("UNHEALTHY");
499 yield* pushHealth("HEALTHY");
500 yield* expectHealth("HEALTHY");
501
502 // cleanup
503 yield* deregisterOutOfBand("cust-1", customServiceId);
504 }),
505 { timeout: 300_000 },
506 );
507 });
508 },
509);