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 { HttpServerRequest } from "effect/http/HttpServerRequest";
7import * as HttpServerResponse from "effect/http/HttpServerResponse";
8
9// Bindings fixture: a standalone SQS→SQS pipe (no Lambda target, so no
10// circularity) plus a Lambda that exercises the four Pipes runtime bindings
11// against it — DescribePipe / ListPipes / StopPipe / StartPipe.
12export class PipesBindingsFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
13 "PipesBindingsFunction",
14) {}
15
16export class BoundPipe extends Context.Service<
17 BoundPipe,
18 { pipe: AWS.Pipes.Pipe }
19>()("BoundPipe") {}
20
21export const BoundPipeLive = Layer.effect(
22 BoundPipe,
23 Effect.gen(function* () {
24 const source = yield* AWS.SQS.Queue("BindingsSourceQueue");
25 const target = yield* AWS.SQS.Queue("BindingsTargetQueue");
26 const role = yield* AWS.IAM.Role("BindingsPipeRole", {
27 assumeRolePolicyDocument: {
28 Version: "2012-10-17",
29 Statement: [
30 {
31 Effect: "Allow",
32 Principal: { Service: "pipes.amazonaws.com" },
33 Action: ["sts:AssumeRole"],
34 },
35 ],
36 },
37 inlinePolicies: {
38 PipeAccess: {
39 Version: "2012-10-17",
40 Statement: [
41 {
42 Effect: "Allow",
43 Action: [
44 "sqs:ReceiveMessage",
45 "sqs:DeleteMessage",
46 "sqs:GetQueueAttributes",
47 ],
48 Resource: [source.queueArn],
49 },
50 {
51 Effect: "Allow",
52 Action: ["sqs:SendMessage"],
53 Resource: [target.queueArn],
54 },
55 ],
56 },
57 },
58 });
59 const pipe = yield* AWS.Pipes.Pipe("BindingsPipe", {
60 source: source.queueArn,
61 target: target.queueArn,
62 roleArn: role.roleArn,
63 sourceParameters: {
64 SqsQueueParameters: { BatchSize: 1 },
65 },
66 });
67 return { pipe };
68 }),
69);
70
71export default PipesBindingsFunction.make(
72 {
73 main: import.meta.url,
74 functionUrl: true,
75 timeout: Duration.seconds(30),
76 },
77 Effect.gen(function* () {
78 const { pipe } = yield* BoundPipe;
79
80 const describePipe = yield* AWS.Pipes.DescribePipe(pipe);
81 const listPipes = yield* AWS.Pipes.ListPipes();
82 const stopPipe = yield* AWS.Pipes.StopPipe(pipe);
83 const startPipe = yield* AWS.Pipes.StartPipe(pipe);
84
85 const bound = { describePipe, listPipes, stopPipe, startPipe };
86
87 const pipeName = yield* pipe.pipeName;
88
89 return {
90 fetch: Effect.gen(function* () {
91 const request = yield* HttpServerRequest;
92 const url = new URL(request.originalUrl);
93 const pathname = url.pathname;
94
95 if (request.method === "GET" && pathname === "/bindings") {
96 return yield* HttpServerResponse.json({
97 bound: Object.keys(bound),
98 });
99 }
100
101 if (request.method === "GET" && pathname === "/describe") {
102 const described = yield* describePipe();
103 return yield* HttpServerResponse.json({
104 name: described.Name,
105 currentState: described.CurrentState,
106 desiredState: described.DesiredState,
107 });
108 }
109
110 if (request.method === "GET" && pathname === "/list") {
111 const { Pipes } = yield* listPipes({
112 NamePrefix: yield* pipeName,
113 });
114 return yield* HttpServerResponse.json({
115 names: (Pipes ?? []).flatMap((p) =>
116 p.Name !== undefined ? [p.Name] : [],
117 ),
118 });
119 }
120
121 if (request.method === "GET" && pathname === "/stop") {
122 const response = yield* stopPipe();
123 return yield* HttpServerResponse.json({
124 desiredState: response.DesiredState,
125 currentState: response.CurrentState,
126 });
127 }
128
129 if (request.method === "GET" && pathname === "/start") {
130 const response = yield* startPipe();
131 return yield* HttpServerResponse.json({
132 desiredState: response.DesiredState,
133 currentState: response.CurrentState,
134 });
135 }
136
137 return yield* HttpServerResponse.json(
138 { error: "Not found", method: request.method, pathname },
139 { status: 404 },
140 );
141 }).pipe(Effect.orDie),
142 };
143 }).pipe(
144 Effect.provide(
145 Layer.mergeAll(
146 AWS.Pipes.DescribePipeHttp,
147 AWS.Pipes.ListPipesHttp,
148 AWS.Pipes.StopPipeHttp,
149 AWS.Pipes.StartPipeHttp,
150 BoundPipeLive,
151 ),
152 ),
153 ),
154);