1import * as aoss from "@distilled.cloud/aws/opensearchserverless";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5
6/**
7 * Raised when a collection (or VPC endpoint) provisioning terminates in a
8 * `FAILED` state or never reaches `ACTIVE` within the bounded polling budget.
9 */
10export class OpenSearchServerlessProvisioningFailed extends Data.TaggedError(
11 "OpenSearchServerlessProvisioningFailed",
12)<{
13 readonly resource: string;
14 readonly id: string;
15 readonly status: string | undefined;
16 readonly failureCode: string | undefined;
17 readonly failureMessage: string | undefined;
18}> {}
19
20/**
21 * Serialize a security/access policy document to the compact JSON string the
22 * OpenSearch Serverless API expects. Passing through a string is a no-op so
23 * callers may interpolate `Output`-derived collection names into the policy
24 * themselves.
25 */
26export const stringifyPolicy = (
27 policy: string | Record<string, unknown> | readonly unknown[],
28): string => (typeof policy === "string" ? policy : JSON.stringify(policy));
29
30/**
31 * Normalize a policy document (string or object) to a stable JSON string for
32 * drift comparison — parse-then-stringify so key ordering differences between
33 * the desired document and the API's echoed document do not force a spurious
34 * update.
35 */
36export const canonicalizePolicy = (
37 policy: string | Record<string, unknown> | readonly unknown[] | undefined,
38): string => {
39 if (policy === undefined) {
40 return "";
41 }
42 try {
43 const parsed = typeof policy === "string" ? JSON.parse(policy) : policy;
44 return JSON.stringify(parsed);
45 } catch {
46 return typeof policy === "string" ? policy : JSON.stringify(policy);
47 }
48};
49
50/** Convert the API's `[{ key, value }]` tag list into a plain record. */
51export const tagsToRecord = (
52 tags: aoss.Tag[] | undefined,
53): Record<string, string> =>
54 Object.fromEntries((tags ?? []).map((t) => [t.key, t.value]));
55
56/** Convert a plain record into the API's `[{ key, value }]` tag list. */
57export const recordToTagList = (tags: Record<string, string>): aoss.Tag[] =>
58 Object.entries(tags).map(([key, value]) => ({ key, value }));
59
60/**
61 * Poll `batchGetCollection` (bounded, spaced 10s, ~5 min budget) until the
62 * collection reaches a terminal status. Explicitly typed so `Effect.repeat`'s
63 * conditional return type does not widen the provider layer in declaration
64 * emit (see PATTERNS §7).
65 */
66const untilCollectionTerminal = <E, R>(
67 self: Effect.Effect<aoss.BatchGetCollectionResponse, E, R>,
68): Effect.Effect<aoss.CollectionDetail | undefined, E, R> =>
69 Effect.repeat(
70 Effect.map(self, (response) => response.collectionDetails?.[0]),
71 {
72 schedule: Schedule.spaced("10 seconds"),
73 until: (detail) =>
74 detail?.status === "ACTIVE" || detail?.status === "FAILED",
75 times: 30,
76 },
77 );
78
79/**
80 * Await a collection reaching `ACTIVE`, failing with
81 * `OpenSearchServerlessProvisioningFailed` on a `FAILED` terminal status or a
82 * budget timeout.
83 */
84export const awaitCollectionActive = Effect.fn(
85 "AWS.OpenSearchServerless.awaitCollectionActive",
86)(function* (id: string) {
87 const detail = yield* untilCollectionTerminal(
88 aoss.batchGetCollection({ ids: [id] }),
89 );
90 if (detail?.status !== "ACTIVE") {
91 return yield* Effect.fail(
92 new OpenSearchServerlessProvisioningFailed({
93 resource: "Collection",
94 id,
95 status: detail?.status,
96 failureCode: detail?.failureCode,
97 failureMessage: detail?.failureMessage,
98 }),
99 );
100 }
101 return detail;
102});
103
104const untilVpcEndpointTerminal = <E, R>(
105 self: Effect.Effect<aoss.BatchGetVpcEndpointResponse, E, R>,
106): Effect.Effect<aoss.VpcEndpointDetail | undefined, E, R> =>
107 Effect.repeat(
108 Effect.map(self, (response) => response.vpcEndpointDetails?.[0]),
109 {
110 schedule: Schedule.spaced("10 seconds"),
111 until: (detail) =>
112 detail?.status === "ACTIVE" || detail?.status === "FAILED",
113 times: 30,
114 },
115 );
116
117/**
118 * Await a VPC endpoint reaching `ACTIVE`, failing with
119 * `OpenSearchServerlessProvisioningFailed` on a `FAILED` terminal status or a
120 * budget timeout.
121 */
122export const awaitVpcEndpointActive = Effect.fn(
123 "AWS.OpenSearchServerless.awaitVpcEndpointActive",
124)(function* (id: string) {
125 const detail = yield* untilVpcEndpointTerminal(
126 aoss.batchGetVpcEndpoint({ ids: [id] }),
127 );
128 if (detail?.status !== "ACTIVE") {
129 return yield* Effect.fail(
130 new OpenSearchServerlessProvisioningFailed({
131 resource: "VpcEndpoint",
132 id,
133 status: detail?.status,
134 failureCode: detail?.failureCode,
135 failureMessage: detail?.failureMessage,
136 }),
137 );
138 }
139 return detail;
140});
141
142/**
143 * Bounded retry through the `ConflictException` window that AOSS raises while a
144 * dependent resource is still tearing down — e.g. deleting an encryption
145 * `SecurityPolicy` while its `Collection` deletion is still propagating.
146 * Explicitly typed for the same declaration-emit reason as above.
147 */
148export const retryWhileConflict = <A, E extends { _tag: string }, R>(
149 self: Effect.Effect<A, E, R>,
150): Effect.Effect<A, E, R> =>
151 Effect.retry(self, {
152 while: (e) => e._tag === "ConflictException",
153 // ~3 min budget: a security policy cannot be deleted until every
154 // collection it covers is FULLY deleted (not merely DELETING), which can
155 // take a couple of minutes after deleteCollection returns.
156 schedule: Schedule.max([Schedule.fixed("5 seconds"), Schedule.recurs(36)]),
157 });