1import * as AWS from "@/AWS";
2import type { PolicyStatement } from "@/AWS/IAM";
3import * as Context from "effect/Context";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Stream from "effect/Stream";
8import { HttpServerRequest } from "effect/http/HttpServerRequest";
9import * as HttpServerResponse from "effect/http/HttpServerResponse";
10
11// Deterministic names shared by the fixture and the test (the test seeds the
12// source prefix out-of-band and filters flow events by name).
13export const FLOW_BUCKET = "alchemy-test-appflow-bindings";
14export const FLOW_NAME = "alchemy-test-appflow-bindings-flow";
15const BUCKET_ARN = `arn:aws:s3:::${FLOW_BUCKET}`;
16
17// AppFlow's S3 connector requires the bucket policy to authorize the service
18// principal for reads (source) and writes (destination).
19const appflowBucketPolicy: PolicyStatement[] = [
20 {
21 Effect: "Allow",
22 Principal: { Service: "appflow.amazonaws.com" },
23 Action: ["s3:GetObject", "s3:ListBucket"],
24 Resource: [BUCKET_ARN, `${BUCKET_ARN}/*`],
25 },
26 {
27 Effect: "Allow",
28 Principal: { Service: "appflow.amazonaws.com" },
29 Action: [
30 "s3:PutObject",
31 "s3:AbortMultipartUpload",
32 "s3:ListMultipartUploadParts",
33 "s3:ListBucketMultipartUploads",
34 "s3:GetBucketAcl",
35 "s3:PutObjectAcl",
36 ],
37 Resource: [BUCKET_ARN, `${BUCKET_ARN}/*`],
38 },
39];
40
41/**
42 * The flow's source/destination bucket, exported on its own so the test can
43 * deploy it FIRST and seed `input/` before the flow is created — AppFlow
44 * validates the source prefix is non-empty at `CreateFlow` time.
45 */
46export const FlowBucket = Effect.gen(function* () {
47 return yield* AWS.S3.Bucket("FlowBucket", {
48 bucketName: FLOW_BUCKET,
49 forceDestroy: true,
50 policy: appflowBucketPolicy,
51 });
52});
53
54export class AppFlowApiFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
55 "AppFlowApiFunction",
56) {}
57
58/**
59 * Shared infrastructure for the AppFlow bindings fixture:
60 * - the S3-to-S3 on-demand flow the bindings are exercised against
61 * - an SQS sink queue where the `flowEvents` consume loop forwards AppFlow
62 * run reports so the test can observe them out-of-band.
63 */
64export class FlowFixture extends Context.Service<
65 FlowFixture,
66 {
67 bucket: AWS.S3.Bucket;
68 flow: AWS.AppFlow.Flow;
69 eventsQueue: AWS.SQS.Queue;
70 }
71>()("AppFlowFlowFixture") {}
72
73export const FlowFixtureLive = Layer.effect(
74 FlowFixture,
75 Effect.gen(function* () {
76 const bucket = yield* FlowBucket;
77 const flow = yield* AWS.AppFlow.Flow("BindingsFlow", {
78 flowName: FLOW_NAME,
79 triggerConfig: { triggerType: "OnDemand" },
80 sourceFlowConfig: {
81 connectorType: "S3",
82 sourceConnectorProperties: {
83 S3: { bucketName: bucket.bucketName, bucketPrefix: "input" },
84 },
85 },
86 destinationFlowConfigList: [
87 {
88 connectorType: "S3",
89 destinationConnectorProperties: {
90 S3: { bucketName: bucket.bucketName, bucketPrefix: "output" },
91 },
92 },
93 ],
94 tasks: [
95 {
96 taskType: "Map_all",
97 sourceFields: [],
98 connectorOperator: { S3: "NO_OP" },
99 taskProperties: {},
100 },
101 ],
102 });
103 const eventsQueue = yield* AWS.SQS.Queue("FlowEventsSink");
104 return { bucket, flow, eventsQueue };
105 }),
106);
107
108export const AppFlowApiFunctionLive = AppFlowApiFunction.make(
109 {
110 main: import.meta.url,
111 functionUrl: true,
112 timeout: Duration.seconds(30),
113 },
114 Effect.gen(function* () {
115 const { flow, eventsQueue } = yield* FlowFixture;
116
117 const startFlow = yield* AWS.AppFlow.StartFlow(flow);
118 const stopFlow = yield* AWS.AppFlow.StopFlow(flow);
119 const cancelFlowExecutions = yield* AWS.AppFlow.CancelFlowExecutions(flow);
120 const describeFlowExecutionRecords =
121 yield* AWS.AppFlow.DescribeFlowExecutionRecords(flow);
122 const eventsSink = yield* AWS.SQS.QueueSink(eventsQueue);
123 const flowEventPattern = AWS.AppFlow.flowEvents({
124 flowNames: [FLOW_NAME],
125 });
126
127 // Forward this flow's AppFlow run reports into the sink queue so the test
128 // can observe them out-of-band via `sqs.receiveMessage`. Native AppFlow
129 // reports use the reserved `aws.appflow` source and are best-effort. Keep
130 // that production pattern and add a custom source solely so the test can
131 // publish a deterministic readiness/report probe through PutEvents.
132 yield* AWS.EventBridge.consumeBusEvents(
133 {
134 ...flowEventPattern,
135 source: [...flowEventPattern.source, "alchemy.test.appflow"],
136 },
137 (events: Stream.Stream<AWS.EventBridge.EventRecord>) =>
138 events.pipe(
139 Stream.map((event) => ({
140 MessageBody: JSON.stringify({
141 detailType: event["detail-type"],
142 detail: event.detail,
143 }),
144 })),
145 Stream.run(eventsSink),
146 Effect.orDie,
147 ),
148 );
149
150 return {
151 fetch: Effect.gen(function* () {
152 const request = yield* HttpServerRequest;
153 const url = new URL(request.originalUrl);
154 const pathname = url.pathname;
155
156 if (request.method === "GET" && pathname === "/ready") {
157 return yield* HttpServerResponse.json({ ok: true });
158 }
159
160 if (request.method === "POST" && pathname === "/start") {
161 const result = yield* startFlow();
162 return yield* HttpServerResponse.json({
163 executionId: result.executionId,
164 flowStatus: result.flowStatus,
165 flowArn: result.flowArn,
166 });
167 }
168
169 // OnDemand flows cannot be stopped; AppFlow answers with a typed
170 // UnsupportedOperationException — surfaced as `{ ok: false, error }`
171 // so the test can assert the exact tag.
172 if (request.method === "POST" && pathname === "/stop") {
173 const response = yield* stopFlow().pipe(
174 Effect.match({
175 onFailure: (error) => ({
176 ok: false as const,
177 error: error._tag,
178 }),
179 onSuccess: (value) => ({ ok: true as const, value }),
180 }),
181 );
182 return yield* HttpServerResponse.json(response);
183 }
184
185 if (request.method === "POST" && pathname === "/cancel") {
186 const body = (yield* request.json) as {
187 executionIds?: string[];
188 };
189 const response = yield* cancelFlowExecutions({
190 executionIds: body.executionIds,
191 }).pipe(
192 Effect.match({
193 onFailure: (error) => ({
194 ok: false as const,
195 error: error._tag,
196 }),
197 onSuccess: (value) => ({
198 ok: true as const,
199 invalidExecutions: value.invalidExecutions ?? [],
200 }),
201 }),
202 );
203 return yield* HttpServerResponse.json(response);
204 }
205
206 if (request.method === "GET" && pathname === "/executions") {
207 const result = yield* describeFlowExecutionRecords();
208 return yield* HttpServerResponse.json({
209 flowExecutions: (result.flowExecutions ?? []).map((record) => ({
210 executionId: record.executionId,
211 executionStatus: record.executionStatus,
212 })),
213 });
214 }
215
216 return yield* HttpServerResponse.json(
217 { error: "Not found", method: request.method, pathname },
218 { status: 404 },
219 );
220 }).pipe(Effect.orDie),
221 };
222 }).pipe(
223 Effect.provide(
224 Layer.provideMerge(
225 Layer.mergeAll(
226 AWS.Lambda.EventSource,
227 AWS.AppFlow.StartFlowHttp,
228 AWS.AppFlow.StopFlowHttp,
229 AWS.AppFlow.CancelFlowExecutionsHttp,
230 AWS.AppFlow.DescribeFlowExecutionRecordsHttp,
231 AWS.SQS.QueueSinkHttp,
232 FlowFixtureLive,
233 ),
234 Layer.mergeAll(AWS.SQS.SendMessageBatchHttp),
235 ),
236 ),
237 ),
238 // Re-merge so the deploying test can `yield* FlowFixture` and expose the
239 // flow name / queue url as deploy outputs. Reusing the same reference keeps
240 // it a single shared flow/queue.
241).pipe(Layer.provideMerge(FlowFixtureLive));
242
243export default AppFlowApiFunctionLive;