1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import * as logs from "@distilled.cloud/aws/cloudwatch-logs";
4import * as SQS from "@distilled.cloud/aws/sqs";
5import { describe, expect } from "alchemy-test";
6import * as Clock from "effect/Clock";
7import * as Data from "effect/Data";
8import * as Effect from "effect/Effect";
9import * as Schedule from "effect/Schedule";
10import * as HttpClient from "effect/http/HttpClient";
11import LogGroupEventSourceFunctionLive, {
12 LogGroupEventSourceFunction,
13} from "./event-source-handler.ts";
14
15const { test } = Test.make({ providers: AWS.providers() });
16
17class FunctionNotReady extends Data.TaggedError("FunctionNotReady")<{
18 message: string;
19}> {
20 constructor(message: string) {
21 super({ message });
22 }
23}
24
25class MessageNotDelivered extends Data.TaggedError("MessageNotDelivered") {}
26
27describe.sequential(
28 "AWS.Logs.LogGroupEventSource",
29 {
30 tags: [
31 "provider:aws",
32 "provider:aws:lambda",
33 "provider:aws:logs",
34 "provider:aws:sqs",
35 "live",
36 ],
37 },
38 () => {
39 test.provider(
40 "delivers decoded log events through the Lambda event source",
41 (stack) =>
42 Effect.gen(function* () {
43 yield* stack.destroy();
44
45 const fn = yield* stack.deploy(
46 LogGroupEventSourceFunction.pipe(
47 Effect.provide(LogGroupEventSourceFunctionLive),
48 ),
49 );
50
51 const functionUrl = fn.functionUrl!;
52
53 // First request rides out cold-start / URL propagation; keep polling
54 // until the fixture reports its identifiers.
55 const { sourceLogGroupName, resultQueueUrl } = yield* HttpClient.get(
56 functionUrl,
57 ).pipe(
58 Effect.flatMap((response) =>
59 response.status === 200
60 ? (response.json as Effect.Effect<{
61 sourceLogGroupName?: string;
62 resultQueueUrl?: string;
63 }>)
64 : Effect.fail(
65 new FunctionNotReady(
66 `Function not ready: ${response.status}`,
67 ),
68 ),
69 ),
70 Effect.flatMap((body) =>
71 body.sourceLogGroupName && body.resultQueueUrl
72 ? Effect.succeed(
73 body as {
74 sourceLogGroupName: string;
75 resultQueueUrl: string;
76 },
77 )
78 : Effect.fail(
79 new FunctionNotReady("Function returned empty identifiers"),
80 ),
81 ),
82 Effect.retry({
83 schedule: Schedule.max([
84 Schedule.fixed("1 seconds"),
85 Schedule.recurs(60),
86 ]),
87 }),
88 );
89
90 // Deploy-shape assertion (ungated): the subscription filter targeting
91 // the Lambda exists on the source log group.
92 const described = yield* logs.describeSubscriptionFilters({
93 logGroupName: sourceLogGroupName,
94 });
95 const filter = (described.subscriptionFilters ?? []).find((f) =>
96 f.destinationArn?.includes(fn.functionName),
97 );
98 expect(filter).toBeDefined();
99 expect(filter?.destinationArn).toContain(":function:");
100
101 // Write a marker log event into the source group out-of-band; the
102 // subscription delivers it to the Lambda, which decodes the gzipped
103 // payload and forwards the message to the result queue.
104 const marker = `log-event-source-${crypto.randomUUID()}`;
105 const logStreamName = "alchemy-test-event-source-stream";
106 yield* logs
107 .createLogStream({
108 logGroupName: sourceLogGroupName,
109 logStreamName,
110 })
111 .pipe(
112 Effect.catchTag(
113 "ResourceAlreadyExistsException",
114 () => Effect.void,
115 ),
116 );
117 const timestamp = yield* Clock.currentTimeMillis;
118 yield* logs.putLogEvents({
119 logGroupName: sourceLogGroupName,
120 logStreamName,
121 logEvents: [{ timestamp, message: marker }],
122 });
123
124 // Poll the result queue until the forwarded marker shows up.
125 // Subscription delivery latency is typically a few seconds; bounded at
126 // ~30 polls × (2s long-poll + 1s spacing) ≈ 90s.
127 const received = yield* Effect.gen(function* () {
128 const result = yield* SQS.receiveMessage({
129 QueueUrl: resultQueueUrl,
130 MaxNumberOfMessages: 10,
131 WaitTimeSeconds: 2,
132 });
133 const match = (result.Messages ?? []).find(
134 (message) => message.Body === marker,
135 );
136 if (!match?.ReceiptHandle) {
137 return yield* Effect.fail(new MessageNotDelivered());
138 }
139 yield* SQS.deleteMessage({
140 QueueUrl: resultQueueUrl,
141 ReceiptHandle: match.ReceiptHandle,
142 });
143 return match.Body!;
144 }).pipe(
145 Effect.retry({
146 while: (error) => error._tag === "MessageNotDelivered",
147 schedule: Schedule.max([
148 Schedule.fixed("1 seconds"),
149 Schedule.recurs(30),
150 ]),
151 }),
152 );
153
154 expect(received).toEqual(marker);
155
156 yield* stack.destroy();
157
158 // The subscription filter is deleted with the log group.
159 const afterDestroy = yield* logs
160 .describeSubscriptionFilters({ logGroupName: sourceLogGroupName })
161 .pipe(
162 Effect.catchTag("ResourceNotFoundException", () =>
163 Effect.succeed({ subscriptionFilters: [] }),
164 ),
165 );
166 expect(afterDestroy.subscriptionFilters ?? []).toHaveLength(0);
167 }),
168 { timeout: 240_000 },
169 );
170 },
171);