1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import * as pipes from "@distilled.cloud/aws/pipes";
4import { describe, expect } from "alchemy-test";
5import * as Effect from "effect/Effect";
6import * as Redacted from "effect/Redacted";
7
8const { test } = Test.make({ providers: AWS.providers() });
9
10const unwrap = (
11 value: string | Redacted.Redacted<string> | undefined,
12): string | undefined =>
13 value === undefined
14 ? undefined
15 : Redacted.isRedacted(value)
16 ? Redacted.value(value)
17 : value;
18
19const firstFilterPattern = (
20 described: pipes.DescribePipeResponse,
21): string | undefined =>
22 unwrap(described.SourceParameters?.FilterCriteria?.Filters?.[0]?.Pattern);
23
24// Read/receive statements the pipes.amazonaws.com role needs on the source
25// queue, plus send on the target queue.
26const pipeRole = (source: AWS.SQS.Queue, target: AWS.SQS.Queue) =>
27 AWS.IAM.Role("LifecyclePipeRole", {
28 assumeRolePolicyDocument: {
29 Version: "2012-10-17",
30 Statement: [
31 {
32 Effect: "Allow",
33 Principal: { Service: "pipes.amazonaws.com" },
34 Action: ["sts:AssumeRole"],
35 },
36 ],
37 },
38 inlinePolicies: {
39 PipeSource: {
40 Version: "2012-10-17",
41 Statement: [
42 {
43 Effect: "Allow",
44 Action: [
45 "sqs:ReceiveMessage",
46 "sqs:DeleteMessage",
47 "sqs:GetQueueAttributes",
48 ],
49 Resource: [source.queueArn],
50 },
51 ],
52 },
53 PipeTarget: {
54 Version: "2012-10-17",
55 Statement: [
56 {
57 Effect: "Allow",
58 Action: ["sqs:SendMessage"],
59 Resource: [target.queueArn],
60 },
61 ],
62 },
63 },
64 });
65
66describe(
67 "AWS.Pipes.Pipe",
68 {
69 tags: [
70 "provider:aws",
71 "provider:aws:iam",
72 "provider:aws:pipes",
73 "provider:aws:sqs",
74 "live",
75 ],
76 },
77 () => {
78 test.provider(
79 "lifecycle: create SQS→SQS pipe with filter, update filter, destroy",
80 (stack) =>
81 Effect.gen(function* () {
82 yield* stack.destroy();
83
84 const filterA = JSON.stringify({ body: { type: ["order.created"] } });
85 const filterB = JSON.stringify({ body: { type: ["order.updated"] } });
86
87 const deployPipe = (pattern: string) =>
88 stack.deploy(
89 Effect.gen(function* () {
90 const source = yield* AWS.SQS.Queue("LifecycleSourceQueue");
91 const target = yield* AWS.SQS.Queue("LifecycleTargetQueue");
92 const role = yield* pipeRole(source, target);
93 return yield* AWS.Pipes.Pipe("LifecyclePipe", {
94 source: source.queueArn,
95 target: target.queueArn,
96 roleArn: role.roleArn,
97 sourceParameters: {
98 FilterCriteria: { Filters: [{ Pattern: pattern }] },
99 SqsQueueParameters: { BatchSize: 1 },
100 },
101 });
102 }),
103 );
104
105 // Create — the provider waits (bounded) for RUNNING before
106 // returning, so the deployed attributes already reflect it.
107 const created = yield* deployPipe(filterA);
108 expect(created.currentState).toEqual("RUNNING");
109
110 // Out-of-band verification via distilled.
111 const describedA = yield* pipes.describePipe({
112 Name: created.pipeName,
113 });
114 expect(describedA.CurrentState).toEqual("RUNNING");
115 expect(firstFilterPattern(describedA)).toEqual(filterA);
116 expect(
117 describedA.SourceParameters?.SqsQueueParameters?.BatchSize,
118 ).toEqual(1);
119
120 // Update the filter criteria in place (same source → no replace).
121 const updated = yield* deployPipe(filterB);
122 expect(updated.pipeArn).toEqual(created.pipeArn);
123 const describedB = yield* pipes.describePipe({
124 Name: updated.pipeName,
125 });
126 expect(firstFilterPattern(describedB)).toEqual(filterB);
127 expect(describedB.CurrentState).toEqual("RUNNING");
128
129 // Destroy — the provider's delete waits until describePipe reports
130 // NotFound, so a single typed check suffices here.
131 yield* stack.destroy();
132 const gone = yield* pipes
133 .describePipe({ Name: created.pipeName })
134 .pipe(
135 Effect.map(() => false),
136 Effect.catchTag("NotFoundException", () => Effect.succeed(true)),
137 );
138 expect(gone).toBe(true);
139 }),
140 { timeout: 240_000 },
141 );
142
143 test.provider(
144 "replacement: changing the source replaces the pipe",
145 (stack) =>
146 Effect.gen(function* () {
147 yield* stack.destroy();
148
149 // Keep BOTH candidate source queues deployed across every step —
150 // replacing the pipe while simultaneously removing its old source
151 // dependency deadlocks the engine.
152 const deployPipe = (useSecondSource: boolean) =>
153 stack.deploy(
154 Effect.gen(function* () {
155 const sourceA = yield* AWS.SQS.Queue("ReplaceSourceQueueA");
156 const sourceB = yield* AWS.SQS.Queue("ReplaceSourceQueueB");
157 const target = yield* AWS.SQS.Queue("ReplaceTargetQueue");
158 const role = yield* AWS.IAM.Role("ReplacePipeRole", {
159 assumeRolePolicyDocument: {
160 Version: "2012-10-17",
161 Statement: [
162 {
163 Effect: "Allow",
164 Principal: { Service: "pipes.amazonaws.com" },
165 Action: ["sts:AssumeRole"],
166 },
167 ],
168 },
169 inlinePolicies: {
170 PipeAccess: {
171 Version: "2012-10-17",
172 Statement: [
173 {
174 Effect: "Allow",
175 Action: [
176 "sqs:ReceiveMessage",
177 "sqs:DeleteMessage",
178 "sqs:GetQueueAttributes",
179 ],
180 Resource: [sourceA.queueArn, sourceB.queueArn],
181 },
182 {
183 Effect: "Allow",
184 Action: ["sqs:SendMessage"],
185 Resource: [target.queueArn],
186 },
187 ],
188 },
189 },
190 });
191 const active = useSecondSource ? sourceB : sourceA;
192 const pipe = yield* AWS.Pipes.Pipe("ReplacePipe", {
193 source: active.queueArn,
194 target: target.queueArn,
195 roleArn: role.roleArn,
196 });
197 return { pipe, sourceA, sourceB };
198 }),
199 );
200
201 const first = yield* deployPipe(false);
202 expect(first.pipe.currentState).toEqual("RUNNING");
203
204 const second = yield* deployPipe(true);
205 // Replacement mints a new instance: new physical name + ARN.
206 expect(second.pipe.pipeArn).not.toEqual(first.pipe.pipeArn);
207
208 const described = yield* pipes.describePipe({
209 Name: second.pipe.pipeName,
210 });
211 expect(described.Source).toEqual(second.sourceB.queueArn);
212
213 // The replaced pipe is deleted.
214 const oldGone = yield* pipes
215 .describePipe({ Name: first.pipe.pipeName })
216 .pipe(
217 Effect.map(() => false),
218 Effect.catchTag("NotFoundException", () => Effect.succeed(true)),
219 );
220 expect(oldGone).toBe(true);
221
222 yield* stack.destroy();
223 }),
224 { timeout: 240_000 },
225 );
226 },
227);