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";
5import * as Stream from "effect/Stream";
6import { isResolved } from "../../Diff.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import type { Providers } from "../Providers.ts";
10import { jsonEquals } from "./internal.ts";
11
12export interface PipelineEndpointVpcOptions {
13 /**
14 * Subnet IDs the endpoint's network interfaces are placed into. Changing
15 * subnets replaces the endpoint.
16 */
17 subnetIds: string[];
18 /**
19 * Security group IDs applied to the endpoint's network interfaces.
20 * Changing security groups replaces the endpoint.
21 */
22 securityGroupIds?: string[];
23}
24
25export interface PipelineEndpointProps {
26 /**
27 * ARN of the OSIS pipeline the endpoint ingests into. Changing the
28 * pipeline replaces the endpoint.
29 */
30 pipelineArn: string;
31 /**
32 * VPC placement for the endpoint. All fields are create-only — any change
33 * replaces the endpoint.
34 */
35 vpcOptions: PipelineEndpointVpcOptions;
36}
37
38export interface PipelineEndpoint extends Resource<
39 "AWS.OSIS.PipelineEndpoint",
40 PipelineEndpointProps,
41 {
42 /**
43 * Id of the pipeline endpoint (`pe-…`), assigned by OSIS on create.
44 */
45 endpointId: string;
46 /**
47 * ARN of the pipeline the endpoint ingests into.
48 */
49 pipelineArn: string;
50 /**
51 * Endpoint status (e.g. `ACTIVE`, `CREATING`, `REVOKED`).
52 */
53 status: string;
54 /**
55 * Id of the VPC the endpoint lives in.
56 */
57 vpcId: string | undefined;
58 /**
59 * The VPC-private ingest URL for the endpoint.
60 */
61 ingestEndpointUrl: string | undefined;
62 },
63 never,
64 Providers
65> {}
66
67/**
68 * A VPC endpoint for an Amazon OpenSearch Ingestion (OSIS) pipeline — lets
69 * clients inside a VPC ingest data into a pipeline privately, without
70 * traversing the public ingest endpoint.
71 *
72 * All properties are create-only; any change replaces the endpoint. The
73 * endpoint id is assigned by OSIS on create.
74 * ### Creating a Pipeline Endpoint
75 * **Example:** Private Ingest From a VPC
76 * ```typescript
77 * const endpoint = yield* OSIS.PipelineEndpoint("Private", {
78 * pipelineArn: pipeline.pipelineArn,
79 * vpcOptions: {
80 * subnetIds: [subnet.subnetId],
81 * securityGroupIds: [securityGroup.securityGroupId],
82 * },
83 * });
84 * // endpoint.ingestEndpointUrl — the VPC-private ingest URL
85 * ```
86 *
87 * @resource
88 */
89export const PipelineEndpoint = Resource<PipelineEndpoint>(
90 "AWS.OSIS.PipelineEndpoint",
91);
92
93/**
94 * A pipeline endpoint whose asynchronous create converged to a failed
95 * status.
96 */
97export class PipelineEndpointCreateFailed extends Data.TaggedError(
98 "OsisPipelineEndpointCreateFailed",
99)<{
100 readonly endpointId: string;
101 readonly status: string;
102}> {}
103
104export const PipelineEndpointProvider = () =>
105 Provider.effect(
106 PipelineEndpoint,
107 Effect.gen(function* () {
108 /**
109 * There is no `getPipelineEndpoint` — observe by enumerating the
110 * account's endpoints and matching the id.
111 */
112 const findEndpoint = Effect.fn(function* (endpointId: string) {
113 return yield* osis.listPipelineEndpoints.pages({}).pipe(
114 Stream.runCollect,
115 Effect.map((chunk) =>
116 Array.from(chunk)
117 .flatMap((page) => page.PipelineEndpoints ?? [])
118 .find((endpoint) => endpoint.EndpointId === endpointId),
119 ),
120 );
121 });
122
123 const toAttrs = Effect.fn(function* (endpoint: osis.PipelineEndpoint) {
124 if (!endpoint.EndpointId || !endpoint.PipelineArn) {
125 return yield* Effect.fail(
126 new Error(
127 `OSIS pipeline endpoint '${endpoint.EndpointId}' is missing its id or pipeline ARN`,
128 ),
129 );
130 }
131 return {
132 endpointId: endpoint.EndpointId,
133 pipelineArn: endpoint.PipelineArn,
134 status: endpoint.Status ?? "ACTIVE",
135 vpcId: endpoint.VpcId,
136 ingestEndpointUrl: endpoint.IngestEndpointUrl,
137 };
138 });
139
140 /**
141 * Endpoint creation provisions VPC network interfaces and typically
142 * takes a few minutes. Bounded: 15s x 40 (~10 minutes). The explicit
143 * return annotation is load-bearing — see `waitForPipelineSettled` in
144 * `internal.ts`.
145 */
146 const waitForSettled = <E, R>(
147 read: Effect.Effect<osis.PipelineEndpoint | undefined, E, R>,
148 ): Effect.Effect<osis.PipelineEndpoint | undefined, E, R> =>
149 Effect.repeat(read, {
150 schedule: Schedule.max([
151 Schedule.fixed("15 seconds"),
152 Schedule.recurs(40),
153 ]),
154 until: (endpoint): boolean =>
155 endpoint === undefined || endpoint.Status !== "CREATING",
156 });
157
158 return {
159 stables: ["endpointId", "pipelineArn"],
160
161 diff: Effect.fn(function* ({ olds, news }) {
162 if (!isResolved(news)) return undefined;
163 // Every property is create-only.
164 if (
165 olds?.pipelineArn !== news.pipelineArn ||
166 !jsonEquals(olds?.vpcOptions, news.vpcOptions)
167 ) {
168 return { action: "replace" } as const;
169 }
170 }),
171
172 read: Effect.fn(function* ({ output }) {
173 // The endpoint id is assigned by OSIS on create; without a
174 // persisted id there is nothing deterministic to look up.
175 if (output?.endpointId === undefined) return undefined;
176 const endpoint = yield* findEndpoint(output.endpointId);
177 if (endpoint === undefined) return undefined;
178 // Endpoints are not taggable — ownership is implied by the
179 // persisted create-assigned id.
180 return yield* toAttrs(endpoint);
181 }),
182
183 reconcile: Effect.fn(function* ({ news, output, session }) {
184 const props = news!;
185
186 // 1. Observe — the persisted id is a cache, not a guarantee.
187 let observed =
188 output?.endpointId !== undefined
189 ? yield* findEndpoint(output.endpointId)
190 : undefined;
191
192 // 2. Ensure — create if missing (ids are server-assigned, so
193 // there is no AlreadyExists race to tolerate).
194 if (observed === undefined) {
195 const created = yield* osis.createPipelineEndpoint({
196 PipelineArn: props.pipelineArn,
197 VpcOptions: {
198 SubnetIds: props.vpcOptions.subnetIds,
199 SecurityGroupIds: props.vpcOptions.securityGroupIds,
200 },
201 });
202 if (created.EndpointId === undefined) {
203 return yield* Effect.fail(
204 new Error("OSIS createPipelineEndpoint returned no EndpointId"),
205 );
206 }
207 observed = yield* findEndpoint(created.EndpointId);
208 }
209
210 // Wait (bounded) for the asynchronous create to settle; there is
211 // no mutable aspect to sync — every property is create-only.
212 if (observed?.Status === "CREATING" && observed.EndpointId) {
213 observed = yield* waitForSettled(findEndpoint(observed.EndpointId));
214 }
215 if (observed === undefined) {
216 return yield* Effect.fail(
217 new Error(
218 "OSIS pipeline endpoint disappeared while waiting for create",
219 ),
220 );
221 }
222 if (observed.Status === "CREATE_FAILED") {
223 return yield* Effect.fail(
224 new PipelineEndpointCreateFailed({
225 endpointId: observed.EndpointId ?? "",
226 status: observed.Status,
227 }),
228 );
229 }
230
231 yield* session.note(observed.EndpointId ?? "");
232 return yield* toAttrs(observed);
233 }),
234
235 delete: Effect.fn(function* ({ output }) {
236 // Observe first for idempotency — the delete error union carries
237 // no typed not-found tag.
238 const observed = yield* findEndpoint(output.endpointId);
239 if (observed === undefined || observed.Status === "DELETING") {
240 return;
241 }
242 yield* osis.deletePipelineEndpoint({
243 EndpointId: output.endpointId,
244 });
245 }),
246
247 list: () =>
248 osis.listPipelineEndpoints.pages({}).pipe(
249 Stream.runCollect,
250 Effect.map((chunk) =>
251 Array.from(chunk).flatMap((page) => page.PipelineEndpoints ?? []),
252 ),
253 Effect.flatMap(
254 Effect.forEach((endpoint) => toAttrs(endpoint), {
255 concurrency: 4,
256 }),
257 ),
258 ),
259 };
260 }),
261 );