1import * as pipes from "@distilled.cloud/aws/pipes";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Redacted from "effect/Redacted";
5import * as Schedule from "effect/Schedule";
6import * as Stream from "effect/Stream";
7import { Unowned } from "../../AdoptPolicy.ts";
8import { isResolved } from "../../Diff.ts";
9import { createPhysicalName } from "../../PhysicalName.ts";
10import * as Provider from "../../Provider.ts";
11import { Resource } from "../../Resource.ts";
12import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
13import type { Providers } from "../Providers.ts";
14
15/**
16 * The state you want the pipe to be in after reconciliation.
17 *
18 * - `RUNNING` — the pipe polls the source and delivers to the target.
19 * - `STOPPED` — the pipe exists but does not poll the source.
20 */
21export type PipeDesiredState = "RUNNING" | "STOPPED";
22
23export interface PipeProps {
24 /**
25 * Name of the pipe (1-64 characters). If omitted, a deterministic
26 * physical name is generated from the app, stage, and logical ID.
27 */
28 pipeName?: string;
29 /**
30 * A description of the pipe.
31 */
32 description?: string;
33 /**
34 * The state the pipe should be in.
35 * @default "RUNNING"
36 */
37 desiredState?: PipeDesiredState;
38 /**
39 * The ARN of the source resource (SQS queue, Kinesis stream, DynamoDB
40 * stream, etc.). Changing the source triggers a **replacement** — the
41 * EventBridge Pipes API does not allow updating a pipe's source.
42 */
43 source: string;
44 /**
45 * Source-specific parameters (batching, starting position, filter
46 * criteria). `FilterCriteria` holds EventBridge event-pattern strings
47 * that select which source events reach the enrichment/target.
48 * Changing a stream source's `StartingPosition` triggers a replacement.
49 */
50 sourceParameters?: pipes.PipeSourceParameters;
51 /**
52 * The ARN of the enrichment resource (Lambda function, Step Functions
53 * state machine, or API destination) invoked between source and target.
54 */
55 enrichment?: string;
56 /**
57 * Parameters for the enrichment step (input template, HTTP parameters).
58 */
59 enrichmentParameters?: pipes.PipeEnrichmentParameters;
60 /**
61 * The ARN of the target resource (Lambda function, SQS queue, Kinesis
62 * stream, EventBridge bus, CloudWatch Logs group, ECS task, etc.).
63 */
64 target: string;
65 /**
66 * Target-specific invocation parameters (input template, invocation
67 * type, partition key, message group, etc.).
68 */
69 targetParameters?: pipes.PipeTargetParameters;
70 /**
71 * The ARN of the IAM role that EventBridge Pipes assumes to read from
72 * the source and deliver to the target (and invoke the enrichment).
73 * Must trust the `pipes.amazonaws.com` service principal. Use
74 * `Pipes.from(source).toLambda(fn)` to have the role synthesized
75 * automatically.
76 */
77 roleArn: string;
78 /**
79 * Pipe execution log configuration (CloudWatch Logs, Firehose, or S3).
80 */
81 logConfiguration?: pipes.PipeLogConfigurationParameters;
82 /**
83 * The identifier of the KMS customer managed key used to encrypt pipe
84 * data at rest.
85 */
86 kmsKeyIdentifier?: string;
87 /**
88 * Tags to apply to the pipe. Merged with internal Alchemy tags.
89 */
90 tags?: Record<string, string>;
91}
92
93export interface Pipe extends Resource<
94 "AWS.Pipes.Pipe",
95 PipeProps,
96 {
97 /**
98 * Name of the pipe.
99 */
100 pipeName: string;
101 /**
102 * ARN of the pipe.
103 */
104 pipeArn: string;
105 /**
106 * Observed state of the pipe (e.g. `RUNNING`, `STOPPED`).
107 */
108 currentState: string | undefined;
109 },
110 never,
111 Providers
112> {}
113
114/**
115 * An Amazon EventBridge Pipe — point-to-point source→(filter)→(enrich)→target
116 * plumbing between AWS services without glue code.
117 *
118 * `Pipe` owns the lifecycle of an EventBridge Pipe. Reconcile waits (bounded)
119 * for the pipe to leave its `CREATING`/`UPDATING` transitional states, and a
120 * pipe that lands in a `*_FAILED` state surfaces as a typed {@link PipeFailed}
121 * error rather than hanging. Prefer the {@link from} builder for the common
122 * pairs — it synthesizes the `pipes.amazonaws.com` execution role with
123 * source-read and target-invoke policies for you.
124 * ### Creating Pipes
125 * **Example:** SQS to Lambda (builder — role synthesized automatically)
126 * ```typescript
127 * import * as AWS from "alchemy/AWS";
128 *
129 * const queue = yield* AWS.SQS.Queue("OrdersQueue");
130 * const pipe = yield* AWS.Pipes.from(queue, { batchSize: 1 }).toLambda(fn);
131 * ```
132 *
133 * **Example:** SQS to SQS (canonical resource with an explicit role)
134 * ```typescript
135 * const pipe = yield* AWS.Pipes.Pipe("OrdersPipe", {
136 * source: source.queueArn,
137 * target: target.queueArn,
138 * roleArn: role.roleArn,
139 * sourceParameters: {
140 * SqsQueueParameters: { BatchSize: 1 },
141 * },
142 * });
143 * ```
144 *
145 * ### Filtering
146 * **Example:** Only deliver matching events
147 * ```typescript
148 * const pipe = yield* AWS.Pipes.from(queue)
149 * .filter(JSON.stringify({ body: { type: ["order.created"] } }))
150 * .toLambda(fn);
151 * ```
152 *
153 * ### Enrichment
154 * **Example:** Enrich events with a Lambda function before delivery
155 * ```typescript
156 * const pipe = yield* AWS.Pipes.from(queue)
157 * .enrich(enricherFn)
158 * .toQueue(target);
159 * ```
160 *
161 * ### Stream Sources
162 * **Example:** Kinesis stream source
163 * ```typescript
164 * const pipe = yield* AWS.Pipes.from(stream, {
165 * startingPosition: "TRIM_HORIZON",
166 * batchSize: 10,
167 * }).toLambda(fn);
168 * ```
169 *
170 * **Example:** Stop a pipe without deleting it
171 * ```typescript
172 * const pipe = yield* AWS.Pipes.Pipe("OrdersPipe", {
173 * source: source.queueArn,
174 * target: target.queueArn,
175 * roleArn: role.roleArn,
176 * desiredState: "STOPPED",
177 * });
178 * ```
179 *
180 * @resource
181 */
182export const Pipe = Resource<Pipe>("AWS.Pipes.Pipe");
183
184/**
185 * Raised when a pipe lands in a terminal failed state
186 * (`CREATE_FAILED`, `UPDATE_FAILED`, `START_FAILED`, `STOP_FAILED`,
187 * `DELETE_FAILED`, ...). Carries the pipe's `StateReason` so the
188 * misconfiguration (usually IAM) is visible in the failure.
189 */
190export class PipeFailed extends Data.TaggedError("PipeFailed")<{
191 pipeName: string;
192 state: string;
193 stateReason: string | undefined;
194}> {}
195
196/**
197 * Raised when a pipe does not settle out of a transitional state
198 * (`CREATING`, `UPDATING`, `STARTING`, `STOPPING`, `DELETING`) within the
199 * bounded wait (~60s). Surfaces the
200 * last observed state instead of hanging the deploy.
201 */
202export class PipeStateTimeout extends Data.TaggedError("PipeStateTimeout")<{
203 pipeName: string;
204 state: string | undefined;
205 message: string;
206}> {}
207
208/** Internal poll signal: the pipe is still in a transitional state. */
209class PipeStillTransitioning extends Data.TaggedError(
210 "PipeStillTransitioning",
211)<{ pipeName: string; state: string }> {}
212
213/** Internal poll signal: the pipe still exists after a delete. */
214class PipeStillPresent extends Data.TaggedError("PipeStillPresent")<{
215 pipeName: string;
216 state: string | undefined;
217}> {}
218
219const TRANSITIONAL_STATES: ReadonlySet<string> = new Set([
220 "CREATING",
221 "UPDATING",
222 "STARTING",
223 "STOPPING",
224 "DELETING",
225]);
226
227const isFailedState = (state: string) => state.includes("FAILED");
228
229const unwrap = (
230 value: string | Redacted.Redacted<string> | undefined,
231): string | undefined =>
232 value === undefined
233 ? undefined
234 : Redacted.isRedacted(value)
235 ? Redacted.value(value)
236 : value;
237
238const unwrapTags = (
239 tags:
240 | { [key: string]: string | Redacted.Redacted<string> | undefined }
241 | undefined,
242): Record<string, string> => {
243 const out: Record<string, string> = {};
244 for (const [key, value] of Object.entries(tags ?? {})) {
245 const plain = unwrap(value);
246 if (plain !== undefined) out[key] = plain;
247 }
248 return out;
249};
250
251/**
252 * Bounded retry helpers with EXPLICIT `Effect.Effect<A, E, R>` annotations.
253 * Inlining `Effect.retry` with options in provider lifecycle code leaves
254 * `Retry.Return`'s conditional type unresolved in the provider's inferred
255 * layer type, which declaration emit widens to an `unknown` R — poisoning
256 * `AWS.providers()` for every downstream consumer (see
257 * `retryThroughDeletionWindow` in AWS/SecretsManager/Secret.ts).
258 */
259const retryWhileTransitioning = <A, E extends { _tag: string }, R>(
260 self: Effect.Effect<A, E, R>,
261): Effect.Effect<A, E, R> =>
262 Effect.retry(self, {
263 while: (e) => e._tag === "PipeStillTransitioning",
264 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(30)]),
265 });
266
267const retryWhilePresent = <A, E extends { _tag: string }, R>(
268 self: Effect.Effect<A, E, R>,
269): Effect.Effect<A, E, R> =>
270 Effect.retry(self, {
271 while: (e) => e._tag === "PipeStillPresent",
272 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(30)]),
273 });
274
275/**
276 * EventBridge Pipes validates that the execution role is assumable by
277 * `pipes.amazonaws.com` at create/update time. A freshly-created IAM role
278 * can take a minute to propagate, surfacing as a `ValidationException`
279 * mentioning role assumption. Retry (bounded) until propagation completes.
280 */
281const retryUntilRoleAssumable = <A, E extends { _tag: string }, R>(
282 self: Effect.Effect<A, E, R>,
283): Effect.Effect<A, E, R> =>
284 Effect.retry(self, {
285 while: (e) => {
286 if (e._tag !== "ValidationException") return false;
287 const message = (e as { message?: unknown }).message;
288 return (
289 typeof message === "string" &&
290 (message.includes("assume") || message.includes("trust"))
291 );
292 },
293 schedule: Schedule.max([Schedule.spaced("3 seconds"), Schedule.recurs(20)]),
294 });
295
296/**
297 * Pipe mutations racing an in-flight transition surface as
298 * `ConflictException`. Retry on a bounded schedule so a delete/update
299 * issued while the pipe is still `CREATING`/`UPDATING` converges.
300 */
301const retryWhileConflicting = <A, E extends { _tag: string }, R>(
302 self: Effect.Effect<A, E, R>,
303): Effect.Effect<A, E, R> =>
304 Effect.retry(self, {
305 while: (e) => e._tag === "ConflictException",
306 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(20)]),
307 });
308
309/**
310 * `true` when every DEFINED field of `desired` deep-equals the observed
311 * value (Redacted values are unwrapped on both sides). Undefined desired
312 * fields are skipped — AWS fills unspecified parameter fields with
313 * defaults, so only user-specified values participate in drift detection.
314 */
315const subsetMatches = (desired: unknown, observed: unknown): boolean => {
316 if (desired === undefined) return true;
317 const want = Redacted.isRedacted(desired) ? Redacted.value(desired) : desired;
318 const have = Redacted.isRedacted(observed)
319 ? Redacted.value(observed)
320 : observed;
321 if (Array.isArray(want)) {
322 if (!Array.isArray(have) || have.length !== want.length) return false;
323 return want.every((item, i) => subsetMatches(item, have[i]));
324 }
325 if (typeof want === "object" && want !== null) {
326 if (typeof have !== "object" || have === null) return false;
327 return Object.entries(want).every(([key, value]) =>
328 subsetMatches(value, (have as Record<string, unknown>)[key]),
329 );
330 }
331 return want === have;
332};
333
334const filterPatternsOf = (
335 criteria: pipes.FilterCriteria | undefined,
336): string[] =>
337 (criteria?.Filters ?? [])
338 .map((filter) => unwrap(filter.Pattern))
339 .filter((pattern): pattern is string => pattern !== undefined);
340
341const sameFilters = (a: string[], b: string[]) =>
342 a.length === b.length && a.every((pattern, i) => pattern === b[i]);
343
344const startingPositionOf = (params: pipes.PipeSourceParameters | undefined) =>
345 params?.KinesisStreamParameters?.StartingPosition ??
346 params?.DynamoDBStreamParameters?.StartingPosition;
347
348/**
349 * Convert create-shape source parameters into the update shape. Stream
350 * starting positions are immutable (replacement — see `diff`), so they are
351 * stripped. `FilterCriteria` is always sent (defaulting to no filters) so
352 * removing a filter from props converges: EventBridge resets unspecified
353 * fields inside `SourceParameters` to their defaults on update.
354 */
355const toUpdateSourceParameters = (
356 params: pipes.PipeSourceParameters | undefined,
357): pipes.UpdatePipeSourceParameters => ({
358 FilterCriteria: params?.FilterCriteria ?? { Filters: [] },
359 SqsQueueParameters: params?.SqsQueueParameters,
360 KinesisStreamParameters: params?.KinesisStreamParameters
361 ? {
362 BatchSize: params.KinesisStreamParameters.BatchSize,
363 DeadLetterConfig: params.KinesisStreamParameters.DeadLetterConfig,
364 OnPartialBatchItemFailure:
365 params.KinesisStreamParameters.OnPartialBatchItemFailure,
366 MaximumBatchingWindowInSeconds:
367 params.KinesisStreamParameters.MaximumBatchingWindowInSeconds,
368 MaximumRecordAgeInSeconds:
369 params.KinesisStreamParameters.MaximumRecordAgeInSeconds,
370 MaximumRetryAttempts:
371 params.KinesisStreamParameters.MaximumRetryAttempts,
372 ParallelizationFactor:
373 params.KinesisStreamParameters.ParallelizationFactor,
374 }
375 : undefined,
376 DynamoDBStreamParameters: params?.DynamoDBStreamParameters
377 ? {
378 BatchSize: params.DynamoDBStreamParameters.BatchSize,
379 DeadLetterConfig: params.DynamoDBStreamParameters.DeadLetterConfig,
380 OnPartialBatchItemFailure:
381 params.DynamoDBStreamParameters.OnPartialBatchItemFailure,
382 MaximumBatchingWindowInSeconds:
383 params.DynamoDBStreamParameters.MaximumBatchingWindowInSeconds,
384 MaximumRecordAgeInSeconds:
385 params.DynamoDBStreamParameters.MaximumRecordAgeInSeconds,
386 MaximumRetryAttempts:
387 params.DynamoDBStreamParameters.MaximumRetryAttempts,
388 ParallelizationFactor:
389 params.DynamoDBStreamParameters.ParallelizationFactor,
390 }
391 : undefined,
392});
393
394/**
395 * Source parameters with `FilterCriteria` removed — filters get dedicated
396 * removal-sensitive comparison, everything else uses `subsetMatches`.
397 */
398const sourceParametersSansFilters = (
399 params: pipes.PipeSourceParameters | undefined,
400): Record<string, unknown> | undefined => {
401 if (params === undefined) return undefined;
402 const { FilterCriteria: _filters, ...rest } = params;
403 return rest;
404};
405
406export const PipeProvider = () =>
407 Provider.effect(
408 Pipe,
409 Effect.gen(function* () {
410 const createName = Effect.fn(function* (
411 id: string,
412 props: { pipeName?: string | undefined },
413 ) {
414 return (
415 props.pipeName ?? (yield* createPhysicalName({ id, maxLength: 64 }))
416 );
417 });
418
419 const describeOrUndefined = (pipeName: string) =>
420 pipes
421 .describePipe({ Name: pipeName })
422 .pipe(
423 Effect.catchTag("NotFoundException", () =>
424 Effect.succeed(undefined),
425 ),
426 );
427
428 /**
429 * Poll (bounded, ~60s) until the pipe leaves its transitional state.
430 * A `*_FAILED` terminal state surfaces as a typed `PipeFailed`; a pipe
431 * still transitioning after the bounded wait surfaces as a typed
432 * `PipeStateTimeout` — never a hang.
433 */
434 const awaitSettled = Effect.fn(function* (pipeName: string) {
435 const described = yield* describeOrUndefined(pipeName).pipe(
436 Effect.flatMap((d) =>
437 d?.CurrentState !== undefined &&
438 TRANSITIONAL_STATES.has(d.CurrentState)
439 ? Effect.fail(
440 new PipeStillTransitioning({
441 pipeName,
442 state: d.CurrentState,
443 }),
444 )
445 : Effect.succeed(d),
446 ),
447 retryWhileTransitioning,
448 Effect.catchTag("PipeStillTransitioning", (e) =>
449 Effect.fail(
450 new PipeStateTimeout({
451 pipeName,
452 state: e.state,
453 message: `Pipe '${pipeName}' did not settle within the bounded wait (last state: ${e.state})`,
454 }),
455 ),
456 ),
457 );
458 if (
459 described?.CurrentState !== undefined &&
460 isFailedState(described.CurrentState)
461 ) {
462 return yield* Effect.fail(
463 new PipeFailed({
464 pipeName,
465 state: described.CurrentState,
466 stateReason: described.StateReason,
467 }),
468 );
469 }
470 return described;
471 });
472
473 /** Poll (bounded, ~60s) until `describePipe` returns NotFound. */
474 const awaitGone = Effect.fn(function* (pipeName: string) {
475 yield* pipes.describePipe({ Name: pipeName }).pipe(
476 Effect.flatMap((d) =>
477 Effect.fail(
478 new PipeStillPresent({ pipeName, state: d.CurrentState }),
479 ),
480 ),
481 Effect.catchTag("NotFoundException", () => Effect.void),
482 retryWhilePresent,
483 Effect.catchTag("PipeStillPresent", (e) =>
484 Effect.fail(
485 new PipeStateTimeout({
486 pipeName,
487 state: e.state,
488 message: `Pipe '${pipeName}' did not finish deleting within the bounded wait (last state: ${e.state})`,
489 }),
490 ),
491 ),
492 );
493 });
494
495 return Pipe.Provider.of({
496 stables: ["pipeName", "pipeArn"],
497
498 list: () =>
499 Effect.gen(function* () {
500 const summaries = yield* pipes.listPipes.pages({}).pipe(
501 Stream.runCollect,
502 Effect.map((chunk) =>
503 Array.from(chunk).flatMap((page) => page.Pipes ?? []),
504 ),
505 );
506 return summaries
507 .filter(
508 (p): p is pipes.Pipe & { Name: string; Arn: string } =>
509 p.Name !== undefined && p.Arn !== undefined,
510 )
511 .map((p) => ({
512 pipeName: p.Name,
513 pipeArn: p.Arn,
514 currentState: p.CurrentState,
515 }));
516 }),
517
518 read: Effect.fn(function* ({ id, olds, output }) {
519 const pipeName =
520 output?.pipeName ?? (yield* createName(id, olds ?? {}));
521 const described = yield* describeOrUndefined(pipeName);
522 if (!described?.Arn) return undefined;
523 const attrs = {
524 pipeName,
525 pipeArn: described.Arn,
526 currentState: described.CurrentState,
527 };
528 const tags = unwrapTags(described.Tags);
529 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
530 }),
531
532 diff: Effect.fn(function* ({ id, news, olds }) {
533 if (!isResolved(news)) return undefined;
534 const oldName = yield* createName(id, olds);
535 const newName = yield* createName(id, news);
536 if (oldName !== newName) {
537 return { action: "replace" } as const;
538 }
539 // The Pipes API has no way to change a pipe's source.
540 if (olds.source !== undefined && olds.source !== news.source) {
541 return { action: "replace" } as const;
542 }
543 // Stream starting positions are create-only.
544 if (
545 startingPositionOf(olds.sourceParameters) !==
546 startingPositionOf(news.sourceParameters)
547 ) {
548 return { action: "replace" } as const;
549 }
550 // fall through: undefined → default update path
551 }),
552
553 reconcile: Effect.fn(function* ({ id, news, output, session }) {
554 const pipeName = output?.pipeName ?? (yield* createName(id, news));
555 const internalTags = yield* createInternalTags(id);
556 const desiredTags: Record<string, string> = {
557 ...news.tags,
558 ...internalTags,
559 };
560 const desiredState = news.desiredState ?? "RUNNING";
561
562 // 1. OBSERVE — settle any in-flight transition first; cloud state
563 // is authoritative, `output` is only a name cache. A pipe stuck
564 // in a `*_FAILED` state can't be repaired in place (the failure
565 // is usually IAM propagation from a crashed prior run), so it
566 // is deleted and recreated below.
567 let observed = yield* awaitSettled(pipeName).pipe(
568 Effect.catchTag("PipeFailed", (e) =>
569 Effect.gen(function* () {
570 yield* session.note(
571 `Pipe '${pipeName}' is in failed state ${e.state} (${e.stateReason ?? "no reason"}); deleting and recreating`,
572 );
573 yield* pipes.deletePipe({ Name: pipeName }).pipe(
574 retryWhileConflicting,
575 Effect.catchTag("NotFoundException", () => Effect.void),
576 );
577 yield* awaitGone(pipeName);
578 return undefined;
579 }),
580 ),
581 );
582
583 // 2. ENSURE — create if missing; tolerate the concurrent-create
584 // race and retry while a freshly-created IAM role propagates.
585 if (observed?.Arn === undefined) {
586 yield* pipes
587 .createPipe({
588 Name: pipeName,
589 Description: news.description,
590 DesiredState: desiredState,
591 Source: news.source,
592 SourceParameters: news.sourceParameters,
593 Enrichment: news.enrichment,
594 EnrichmentParameters: news.enrichmentParameters,
595 Target: news.target,
596 TargetParameters: news.targetParameters,
597 RoleArn: news.roleArn,
598 Tags: desiredTags,
599 LogConfiguration: news.logConfiguration,
600 KmsKeyIdentifier: news.kmsKeyIdentifier,
601 })
602 .pipe(
603 retryUntilRoleAssumable,
604 Effect.catchTag("ConflictException", () => Effect.void),
605 );
606 observed = yield* awaitSettled(pipeName);
607 }
608
609 if (observed?.Arn === undefined) {
610 return yield* Effect.fail(
611 new PipeFailed({
612 pipeName,
613 state: "MISSING",
614 stateReason: `describePipe returned no pipe after createPipe('${pipeName}') succeeded`,
615 }),
616 );
617 }
618
619 // 3. SYNC — diff OBSERVED cloud state against desired and update
620 // only on drift. Filters/desired-state/target/enrichment are
621 // removal-sensitive; parameter objects compare user-specified
622 // fields only (AWS fills the rest with defaults).
623 const needsUpdate =
624 (news.description !== undefined &&
625 unwrap(observed.Description) !== news.description) ||
626 (observed.DesiredState ?? "RUNNING") !== desiredState ||
627 observed.Target !== news.target ||
628 (observed.Enrichment !== undefined && observed.Enrichment !== ""
629 ? observed.Enrichment
630 : undefined) !== news.enrichment ||
631 observed.RoleArn !== news.roleArn ||
632 (news.kmsKeyIdentifier !== undefined &&
633 observed.KmsKeyIdentifier !== news.kmsKeyIdentifier) ||
634 !sameFilters(
635 filterPatternsOf(observed.SourceParameters?.FilterCriteria),
636 filterPatternsOf(news.sourceParameters?.FilterCriteria),
637 ) ||
638 !subsetMatches(
639 sourceParametersSansFilters(news.sourceParameters),
640 observed.SourceParameters,
641 ) ||
642 !subsetMatches(news.targetParameters, observed.TargetParameters) ||
643 !subsetMatches(
644 news.enrichmentParameters,
645 observed.EnrichmentParameters,
646 );
647
648 if (needsUpdate) {
649 yield* pipes
650 .updatePipe({
651 Name: pipeName,
652 Description: news.description,
653 DesiredState: desiredState,
654 SourceParameters: toUpdateSourceParameters(
655 news.sourceParameters,
656 ),
657 // "" clears a previously-configured enrichment.
658 Enrichment: news.enrichment ?? "",
659 EnrichmentParameters: news.enrichmentParameters,
660 Target: news.target,
661 TargetParameters: news.targetParameters,
662 RoleArn: news.roleArn,
663 LogConfiguration: news.logConfiguration,
664 KmsKeyIdentifier: news.kmsKeyIdentifier,
665 })
666 .pipe(retryUntilRoleAssumable, retryWhileConflicting);
667 observed = yield* awaitSettled(pipeName);
668 }
669
670 if (observed?.Arn === undefined) {
671 return yield* Effect.fail(
672 new PipeFailed({
673 pipeName,
674 state: "MISSING",
675 stateReason: `Pipe '${pipeName}' disappeared during reconciliation`,
676 }),
677 );
678 }
679
680 // 3b. SYNC TAGS — diff against OBSERVED cloud tags so adoption
681 // converges (never against olds/output).
682 const observedTags = unwrapTags(observed.Tags);
683 const { upsert, removed } = diffTags(observedTags, desiredTags);
684 if (upsert.length > 0) {
685 yield* pipes.tagResource({
686 resourceArn: observed.Arn,
687 tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
688 });
689 }
690 if (removed.length > 0) {
691 yield* pipes.untagResource({
692 resourceArn: observed.Arn,
693 tagKeys: removed,
694 });
695 }
696
697 // 4. RETURN fresh Attributes.
698 yield* session.note(observed.Arn);
699 return {
700 pipeName,
701 pipeArn: observed.Arn,
702 currentState: observed.CurrentState,
703 };
704 }),
705
706 delete: Effect.fn(function* ({ output }) {
707 yield* pipes.deletePipe({ Name: output.pipeName }).pipe(
708 // A delete racing an in-flight CREATE/UPDATE transition
709 // returns ConflictException — retry until the pipe settles.
710 retryWhileConflicting,
711 Effect.catchTag("NotFoundException", () => Effect.void),
712 );
713 yield* awaitGone(output.pipeName);
714 }),
715 });
716 }),
717 );