| 1 | import * as AWS from "@/AWS"; |
| 2 | import * as Context from "effect/Context"; |
| 3 | import * as Duration from "effect/Duration"; |
| 4 | import * as Effect from "effect/Effect"; |
| 5 | import * as Layer from "effect/Layer"; |
| 6 | import * as HttpServerResponse from "effect/http/HttpServerResponse"; |
| 7 | |
| 8 | // End-to-end Pipe fixture: an EventBridge Pipe (created at deploy time via |
| 9 | // the `AWS.Pipes.from(...).toLambda(...)` builder, which synthesizes the |
| 10 | // pipes.amazonaws.com execution role) delivers batches from a source SQS |
| 11 | // queue to this Lambda target. The runtime listener forwards each record |
| 12 | // body into a sink queue so the test can observe delivery out-of-band |
| 13 | // (mirrors test/AWS/SQS/event-source-handler.ts). |
| 14 | export class PipeTargetFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()( |
| 15 | "PipeTargetFunction", |
| 16 | ) {} |
| 17 | |
| 18 | export class PipeQueues extends Context.Service< |
| 19 | PipeQueues, |
| 20 | { |
| 21 | source: AWS.SQS.Queue; |
| 22 | sink: AWS.SQS.Queue; |
| 23 | } |
| 24 | >()("PipeQueues") {} |
| 25 | |
| 26 | export const PipeQueuesLive = Layer.effect( |
| 27 | PipeQueues, |
| 28 | Effect.gen(function* () { |
| 29 | const source = yield* AWS.SQS.Queue("PipeSourceQueue"); |
| 30 | const sink = yield* AWS.SQS.Queue("PipeSinkQueue"); |
| 31 | return { source, sink }; |
| 32 | }), |
| 33 | ); |
| 34 | |
| 35 | // EventBridge Pipes invokes a Lambda target with the raw batch: a JSON |
| 36 | // array of source records (NOT the `{ Records: [...] }` Lambda event-source |
| 37 | // envelope). SQS records carry `eventSource: "aws:sqs"`. |
| 38 | const isPipeSqsBatch = ( |
| 39 | event: unknown, |
| 40 | ): event is Array<{ body: string; messageId: string }> => |
| 41 | Array.isArray(event) && |
| 42 | event.length > 0 && |
| 43 | (event[0] as { eventSource?: string })?.eventSource === "aws:sqs"; |
| 44 | |
| 45 | export default PipeTargetFunction.make( |
| 46 | { |
| 47 | main: import.meta.url, |
| 48 | functionUrl: true, |
| 49 | timeout: Duration.seconds(30), |
| 50 | }, |
| 51 | Effect.gen(function* () { |
| 52 | const { source, sink } = yield* PipeQueues; |
| 53 | const sendMessage = yield* AWS.SQS.SendMessage(sink); |
| 54 | const host = yield* AWS.Lambda.Function; |
| 55 | |
| 56 | // Deploy-time: wire the pipe. Skipped at runtime — the deployed Lambda |
| 57 | // re-executes this init effect but must not touch the deploy plane. |
| 58 | if (!globalThis.__ALCHEMY_RUNTIME__) { |
| 59 | yield* AWS.Pipes.from(source, { batchSize: 1 }).toLambda(host); |
| 60 | } |
| 61 | |
| 62 | // Runtime: forward every pipe-delivered record body into the sink queue. |
| 63 | yield* host.listen( |
| 64 | Effect.gen(function* () { |
| 65 | return (event: unknown) => { |
| 66 | if (isPipeSqsBatch(event)) { |
| 67 | return Effect.forEach( |
| 68 | event, |
| 69 | (record) => sendMessage({ MessageBody: record.body }), |
| 70 | { discard: true }, |
| 71 | ).pipe(Effect.orDie); |
| 72 | } |
| 73 | }; |
| 74 | }), |
| 75 | ); |
| 76 | |
| 77 | const sourceQueueUrl = yield* source.queueUrl; |
| 78 | const sinkQueueUrl = yield* sink.queueUrl; |
| 79 | |
| 80 | return { |
| 81 | fetch: Effect.gen(function* () { |
| 82 | return yield* HttpServerResponse.json({ |
| 83 | ok: true, |
| 84 | sourceQueueUrl: yield* sourceQueueUrl, |
| 85 | sinkQueueUrl: yield* sinkQueueUrl, |
| 86 | }); |
| 87 | }).pipe(Effect.orDie), |
| 88 | }; |
| 89 | }).pipe( |
| 90 | Effect.provide(Layer.mergeAll(AWS.SQS.SendMessageHttp, PipeQueuesLive)), |
| 91 | ), |
| 92 | ); |