1import * as osis from "@distilled.cloud/aws/osis";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5
6/**
7 * Convert OSIS's `[{ Key, Value }]` tag list into a plain record, dropping
8 * any entry missing a key or value.
9 */
10export const toTagRecord = (
11 tags: ReadonlyArray<{ Key?: string; Value?: string }> | undefined,
12): Record<string, string> =>
13 Object.fromEntries(
14 (tags ?? [])
15 .filter(
16 (tag): tag is { Key: string; Value: string } =>
17 typeof tag.Key === "string" && typeof tag.Value === "string",
18 )
19 .map((tag) => [tag.Key, tag.Value]),
20 );
21
22/**
23 * Read the observed tags for an OSIS pipeline by ARN. A pipeline that has
24 * just been created (or is mid-transition) can transiently reject the call;
25 * treat any failure as "no observed tags" so tag reconciliation still runs.
26 */
27export const readPipelineTags = Effect.fn(function* (arn: string) {
28 const response = yield* osis
29 .listTagsForResource({ Arn: arn })
30 .pipe(Effect.catch(() => Effect.succeed(undefined)));
31 return toTagRecord(response?.Tags);
32});
33
34/**
35 * An OSIS pipeline whose asynchronous create/update converged to a
36 * `*_FAILED` status.
37 */
38export class PipelineOperationFailed extends Data.TaggedError(
39 "OsisPipelineOperationFailed",
40)<{
41 readonly pipelineName: string;
42 readonly status: string;
43 readonly reason: string | undefined;
44}> {}
45
46const isTransitional = (status: string | undefined): boolean =>
47 status === "CREATING" ||
48 status === "UPDATING" ||
49 status === "STARTING" ||
50 status === "STOPPING" ||
51 status === "DELETING";
52
53const isFailedStatus = (status: string | undefined): boolean =>
54 status === "CREATE_FAILED" ||
55 status === "UPDATE_FAILED" ||
56 status === "START_FAILED";
57
58/**
59 * Poll an OSIS pipeline until its `Status` reaches a terminal value
60 * (bounded: 15s x 60 = ~15 minutes; pipeline creation typically takes 5-10
61 * minutes). `*_FAILED` fails immediately with the status reason. If the
62 * bounded schedule exhausts while still transitional, the last observation
63 * is returned as-is so the caller sees the real (still-converging) pipeline
64 * rather than a spurious failure.
65 *
66 * The explicit `Effect.Effect<A, E, R>` return annotation is load-bearing:
67 * inlining repeat/retry combinators in provider lifecycle code lets their
68 * conditional return types survive into declaration emit and widen the
69 * provider layer to `unknown` R, poisoning `AWS.providers()` downstream.
70 */
71export const waitForPipelineSettled = <E extends { readonly _tag: string }, R>(
72 pipelineName: string,
73 read: Effect.Effect<osis.Pipeline | undefined, E, R>,
74): Effect.Effect<osis.Pipeline | undefined, E | PipelineOperationFailed, R> =>
75 Effect.flatMap(
76 Effect.repeat(read, {
77 schedule: Schedule.max([
78 Schedule.fixed("15 seconds"),
79 Schedule.recurs(60),
80 ]),
81 until: (pipeline) =>
82 pipeline === undefined || !isTransitional(pipeline.Status),
83 }),
84 (
85 pipeline,
86 ): Effect.Effect<osis.Pipeline | undefined, PipelineOperationFailed> =>
87 pipeline !== undefined && isFailedStatus(pipeline.Status)
88 ? Effect.fail(
89 new PipelineOperationFailed({
90 pipelineName,
91 status: pipeline.Status!,
92 reason: pipeline.StatusReason?.Description,
93 }),
94 )
95 : Effect.succeed(pipeline),
96 );
97
98/**
99 * Retry an effect while OSIS reports `ConflictException` — raised when a
100 * delete/update races an in-flight state transition. Bounded: 15s x 20.
101 */
102export const retryWhilePipelineConflict = <
103 A,
104 E extends { readonly _tag: string },
105 R,
106>(
107 self: Effect.Effect<A, E, R>,
108): Effect.Effect<A, E, R> =>
109 Effect.retry(self, {
110 while: (e) => e._tag === "ConflictException",
111 schedule: Schedule.max([Schedule.fixed("15 seconds"), Schedule.recurs(20)]),
112 });
113
114/** Structural deep equality over JSON-shaped values (order-insensitive keys). */
115export const jsonEquals = (a: unknown, b: unknown): boolean => {
116 if (a === b) return true;
117 if (Array.isArray(a) && Array.isArray(b)) {
118 return a.length === b.length && a.every((v, i) => jsonEquals(v, b[i]));
119 }
120 if (
121 typeof a === "object" &&
122 a !== null &&
123 typeof b === "object" &&
124 b !== null &&
125 !Array.isArray(a) &&
126 !Array.isArray(b)
127 ) {
128 const ka = Object.keys(a as Record<string, unknown>);
129 const kb = Object.keys(b as Record<string, unknown>);
130 return (
131 ka.length === kb.length &&
132 ka.every((k) =>
133 jsonEquals(
134 (a as Record<string, unknown>)[k],
135 (b as Record<string, unknown>)[k],
136 ),
137 )
138 );
139 }
140 return false;
141};