1import * as AWS from "@/AWS";
2import { Flow } from "@/AWS/AppFlow";
3import type { PolicyStatement } from "@/AWS/IAM";
4import { Bucket } from "@/AWS/S3";
5import * as Test from "@/Test/Alchemy";
6import * as appflow from "@distilled.cloud/aws/appflow";
7import * as s3 from "@distilled.cloud/aws/s3";
8import { expect } from "alchemy-test";
9import * as Effect from "effect/Effect";
10import * as Schedule from "effect/Schedule";
11
12const { test } = Test.make({ providers: AWS.providers() });
13
14// Deterministic bucket used as both flow source and destination (distinct
15// prefixes). AppFlow requires the bucket policy to authorize the service
16// principal for reads (source) and writes (destination).
17const BUCKET = "alchemy-test-appflow-flow";
18const BUCKET_ARN = `arn:aws:s3:::${BUCKET}`;
19
20const appflowBucketPolicy: PolicyStatement[] = [
21 {
22 Effect: "Allow",
23 Principal: { Service: "appflow.amazonaws.com" },
24 Action: ["s3:GetObject", "s3:ListBucket"],
25 Resource: [BUCKET_ARN, `${BUCKET_ARN}/*`],
26 },
27 {
28 Effect: "Allow",
29 Principal: { Service: "appflow.amazonaws.com" },
30 Action: [
31 "s3:PutObject",
32 "s3:AbortMultipartUpload",
33 "s3:ListMultipartUploadParts",
34 "s3:ListBucketMultipartUploads",
35 "s3:GetBucketAcl",
36 "s3:PutObjectAcl",
37 ],
38 Resource: [BUCKET_ARN, `${BUCKET_ARN}/*`],
39 },
40];
41
42const bucketProgram = Effect.gen(function* () {
43 return yield* Bucket("FlowBucket", {
44 bucketName: BUCKET,
45 forceDestroy: true,
46 policy: appflowBucketPolicy,
47 });
48});
49
50const flowProgram = (description: string) =>
51 Effect.gen(function* () {
52 const bucket = yield* bucketProgram;
53 return yield* Flow("CopyFlow", {
54 description,
55 triggerConfig: { triggerType: "OnDemand" },
56 sourceFlowConfig: {
57 connectorType: "S3",
58 sourceConnectorProperties: {
59 S3: { bucketName: bucket.bucketName, bucketPrefix: "input" },
60 },
61 },
62 destinationFlowConfigList: [
63 {
64 connectorType: "S3",
65 destinationConnectorProperties: {
66 S3: { bucketName: bucket.bucketName, bucketPrefix: "output" },
67 },
68 },
69 ],
70 tasks: [
71 {
72 taskType: "Map_all",
73 sourceFields: [],
74 connectorOperator: { S3: "NO_OP" },
75 taskProperties: {},
76 },
77 ],
78 });
79 });
80
81const assertFlowGone = (flowName: string) =>
82 Effect.gen(function* () {
83 const result = yield* appflow.describeFlow({ flowName }).pipe(
84 Effect.map(() => "present" as const),
85 Effect.catchTag("ResourceNotFoundException", () =>
86 Effect.succeed("gone" as const),
87 ),
88 );
89 if (result === "present") {
90 return yield* Effect.fail(new Error(`Flow '${flowName}' still exists`));
91 }
92 }).pipe(
93 Effect.retry({
94 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(8)]),
95 }),
96 );
97
98// The S3-to-S3 path is credential-free, so the full lifecycle runs ungated.
99test.provider(
100 "create, update, and destroy an S3-to-S3 flow",
101 (stack) =>
102 Effect.gen(function* () {
103 yield* stack.destroy();
104
105 // Phase 1: the bucket alone. AppFlow's S3 connector validates the flow
106 // by listing the source prefix at createFlow time — the prefix must
107 // contain at least one object or createFlow fails with
108 // ConnectorServerException.
109 yield* stack.deploy(bucketProgram);
110 yield* s3.putObject({
111 Bucket: BUCKET,
112 Key: "input/data.csv",
113 Body: new TextEncoder().encode("id,name\n1,alpha\n2,beta\n"),
114 ContentType: "text/csv",
115 });
116
117 // Phase 2: create the flow.
118 const created = yield* stack.deploy(flowProgram("initial description"));
119 expect(created.flowName).toBeDefined();
120 expect(created.flowArn).toContain(":appflow:");
121
122 // Out-of-band verification.
123 const described = yield* appflow.describeFlow({
124 flowName: created.flowName,
125 });
126 expect(described.description).toBe("initial description");
127 expect(described.sourceFlowConfig?.connectorType).toBe("S3");
128 expect(described.tags?.["alchemy::id"]).toBe("CopyFlow");
129
130 // No-op redeploy keeps the same ARN.
131 const noop = yield* stack.deploy(flowProgram("initial description"));
132 expect(noop.flowArn).toBe(created.flowArn);
133
134 // Update the description in place.
135 const updated = yield* stack.deploy(flowProgram("updated description"));
136 expect(updated.flowArn).toBe(created.flowArn);
137 const reDescribed = yield* appflow.describeFlow({
138 flowName: created.flowName,
139 });
140 expect(reDescribed.description).toBe("updated description");
141
142 // Destroy and verify.
143 yield* stack.destroy();
144 yield* assertFlowGone(created.flowName);
145 }),
146 {
147 tags: ["provider:aws", "provider:aws:appflow", "provider:aws:s3", "live"],
148 timeout: 150_000,
149 },
150);