1import * as AWS from "@/AWS";
2import * as Context from "effect/Context";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Stream from "effect/Stream";
7import { HttpServerRequest } from "effect/http/HttpServerRequest";
8import * as HttpServerResponse from "effect/http/HttpServerResponse";
9
10/**
11 * Shared infrastructure for the BusSink fixture:
12 * - a custom event bus the sink publishes to
13 * - an SQS queue routed from the bus (via `toQueue`) so the test can observe
14 * delivered events out-of-band via `sqs.receiveMessage`.
15 */
16export class BusSinkInfra extends Context.Service<
17 BusSinkInfra,
18 {
19 bus: AWS.EventBridge.EventBus;
20 queue: AWS.SQS.Queue;
21 }
22>()("EventBridgeBusSinkInfra") {}
23
24export const BusSinkInfraLive = Layer.effect(
25 BusSinkInfra,
26 Effect.gen(function* () {
27 const bus = yield* AWS.EventBridge.EventBus("BusSinkBus", {
28 name: "alchemy-test-eb-bus-sink",
29 });
30 // The delivery policy is a plain prop (not a `toQueue` binding) so the
31 // Queue never depends on the Rule — Queue<->Rule would otherwise form a
32 // cycle neither resource can precreate out of. A queue policy only ever
33 // governs the queue it is attached to, so `Resource: "*"` is safely
34 // scoped for this fixture.
35 const queue = yield* AWS.SQS.Queue("BusSinkQueue", {
36 policy: {
37 Version: "2012-10-17",
38 Statement: [
39 {
40 Effect: "Allow",
41 Principal: { Service: "events.amazonaws.com" },
42 Action: "sqs:SendMessage",
43 Resource: "*",
44 },
45 ],
46 },
47 });
48 // Route every sink-published event to the queue for out-of-band
49 // observation.
50 yield* AWS.EventBridge.Rule("BusSinkRoute", {
51 eventBusName: bus.eventBusName,
52 eventPattern: { source: ["alchemy.test.bussink"] },
53 targets: [{ Id: "BusSinkQueueTarget", Arn: queue.queueArn as any }],
54 });
55 return { bus, queue };
56 }),
57);
58
59export class BusSinkFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
60 "BusSinkFunction",
61) {}
62
63export const BusSinkFunctionLive = BusSinkFunction.make(
64 {
65 main: import.meta.url,
66 functionUrl: true,
67 // The sink's bounded partial-failure retry can sleep up to ~6s, which
68 // exceeds Lambda's 3s default timeout (see PATTERNS §7).
69 timeout: Duration.seconds(30),
70 },
71 Effect.gen(function* () {
72 const { bus, queue } = yield* BusSinkInfra;
73 const sink = yield* AWS.EventBridge.BusSink(bus);
74 const queueUrl = yield* queue.queueUrl;
75
76 return {
77 fetch: Effect.gen(function* () {
78 const request = yield* HttpServerRequest;
79 const pathname = new URL(request.originalUrl).pathname;
80
81 if (request.method === "GET" && pathname === "/ready") {
82 return yield* HttpServerResponse.json({
83 ok: true,
84 queueUrl: yield* queueUrl,
85 });
86 }
87
88 if (request.method === "POST" && pathname === "/sink") {
89 const body = (yield* request.json) as {
90 markers: string[];
91 includeMalformed?: boolean;
92 };
93
94 const entries: AWS.EventBridge.BusSinkEntry[] = body.markers.map(
95 (marker) => ({
96 Source: "alchemy.test.bussink",
97 DetailType: "BusSinkEvent",
98 Detail: JSON.stringify({ marker }),
99 }),
100 );
101 if (body.includeMalformed) {
102 // Detail must be valid JSON. EventBridge rejects this entry
103 // per-entry (ErrorCode: MalformedDetail) without failing the
104 // batch — the sink must drop it and deliver the rest.
105 entries.splice(1, 0, {
106 Source: "alchemy.test.bussink",
107 DetailType: "BusSinkEvent",
108 Detail: "{ this is not json",
109 });
110 }
111
112 yield* Stream.fromIterable(entries).pipe(Stream.run(sink));
113
114 return yield* HttpServerResponse.json({
115 ok: true,
116 count: entries.length,
117 });
118 }
119
120 return yield* HttpServerResponse.json(
121 { error: "Not found", method: request.method, pathname },
122 { status: 404 },
123 );
124 }).pipe(
125 Effect.catch((error) =>
126 Effect.succeed(
127 HttpServerResponse.text(`Internal server error: ${String(error)}`, {
128 status: 500,
129 }),
130 ),
131 ),
132 ),
133 };
134 }).pipe(
135 Effect.provide(
136 Layer.provideMerge(
137 Layer.mergeAll(BusSinkInfraLive, AWS.EventBridge.BusSinkHttp),
138 Layer.mergeAll(AWS.EventBridge.PutEventsHttp),
139 ),
140 ),
141 ),
142);
143
144export default BusSinkFunctionLive;