1import * as aoss from "@distilled.cloud/aws/opensearchserverless";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
10import type { Providers } from "../Providers.ts";
11import {
12 awaitCollectionActive,
13 recordToTagList,
14 retryWhileConflict,
15 tagsToRecord,
16} from "./internal.ts";
17
18/**
19 * The collection type, which determines the OpenSearch features available:
20 * - `SEARCH` — full-text search and analytics.
21 * - `TIMESERIES` — log analytics over time-ordered data.
22 * - `VECTORSEARCH` — vector (k-NN) similarity search — the type required to
23 * back an Amazon Bedrock Knowledge Base.
24 */
25export type CollectionType = "SEARCH" | "TIMESERIES" | "VECTORSEARCH";
26
27export interface CollectionProps {
28 /**
29 * Name of the collection (3-32 characters, lowercase; must start with a
30 * lowercase letter). Must be matched by an encryption
31 * {@link SecurityPolicy | security policy} before it can be created. Changing
32 * the name replaces the collection.
33 * @default a generated physical name
34 */
35 collectionName?: string;
36 /**
37 * The collection type. Changing the type replaces the collection.
38 * @default "SEARCH"
39 */
40 type?: CollectionType;
41 /**
42 * A human-readable description of the collection.
43 */
44 description?: string;
45 /**
46 * Whether to deploy redundant standby replicas (`ENABLED` doubles the OCU
47 * floor but provides higher availability). Set at creation time.
48 * @default "ENABLED"
49 */
50 standbyReplicas?: "ENABLED" | "DISABLED";
51 /**
52 * Whether deletion protection is enabled. When `ENABLED`, the collection
53 * cannot be deleted until protection is turned off.
54 * @default "DISABLED"
55 */
56 deletionProtection?: "ENABLED" | "DISABLED";
57 /**
58 * Tags to apply to the collection. Merged with the internal Alchemy tags.
59 */
60 tags?: Record<string, string>;
61}
62
63export interface Collection extends Resource<
64 "AWS.OpenSearchServerless.Collection",
65 CollectionProps,
66 {
67 /**
68 * Unique identifier of the collection.
69 */
70 collectionId: string;
71 /**
72 * Name of the collection.
73 */
74 collectionName: string;
75 /**
76 * ARN of the collection.
77 */
78 collectionArn: string;
79 /**
80 * Collection type (`SEARCH`, `TIMESERIES`, or `VECTORSEARCH`).
81 */
82 type?: string;
83 /**
84 * Collection status (e.g. `ACTIVE`, `CREATING`, `DELETING`).
85 */
86 status?: string;
87 /**
88 * ARN of the KMS key encrypting the collection.
89 */
90 kmsKeyArn?: string;
91 /**
92 * OpenSearch endpoint for data-plane requests.
93 */
94 collectionEndpoint?: string;
95 /**
96 * OpenSearch Dashboards endpoint for the collection.
97 */
98 dashboardEndpoint?: string;
99 },
100 {},
101 Providers
102> {}
103
104/**
105 * An Amazon OpenSearch Serverless collection — a group of OpenSearch indexes
106 * that scales OpenSearch Compute Units (OCUs) automatically. A collection of
107 * type `VECTORSEARCH` is the vector store required to back an Amazon Bedrock
108 * Knowledge Base.
109 *
110 * A collection requires a matching encryption {@link SecurityPolicy} to exist
111 * before creation, plus a network security policy and a data
112 * {@link AccessPolicy} to be reachable and usable. Creation is asynchronous —
113 * the provider polls (bounded, ~5 minutes) until the collection reaches
114 * `ACTIVE`.
115 *
116 * ### Creating Collections
117 * **Example:** Vector Search Collection for a Bedrock Knowledge Base
118 * ```typescript
119 * import * as AWS from "alchemy/AWS";
120 *
121 * const encryption = yield* AWS.OpenSearchServerless.SecurityPolicy("Enc", {
122 * policyName: "kb-enc",
123 * type: "encryption",
124 * policy: {
125 * Rules: [{ ResourceType: "collection", Resource: ["collection/kb"] }],
126 * AWSOwnedKey: true,
127 * },
128 * });
129 * const network = yield* AWS.OpenSearchServerless.SecurityPolicy("Net", {
130 * policyName: "kb-net",
131 * type: "network",
132 * policy: [
133 * {
134 * Rules: [
135 * { ResourceType: "collection", Resource: ["collection/kb"] },
136 * { ResourceType: "dashboard", Resource: ["collection/kb"] },
137 * ],
138 * AllowFromPublic: true,
139 * },
140 * ],
141 * });
142 * const collection = yield* AWS.OpenSearchServerless.Collection("KB", {
143 * collectionName: "kb",
144 * type: "VECTORSEARCH",
145 * });
146 * // collection.collectionEndpoint is the aoss data-plane endpoint
147 * ```
148 *
149 * ### Search Collections
150 * **Example:** Simple Search Collection
151 * ```typescript
152 * const collection = yield* AWS.OpenSearchServerless.Collection("Search", {
153 * collectionName: "logs",
154 * type: "SEARCH",
155 * description: "application logs",
156 * });
157 * ```
158 *
159 * @resource
160 */
161export const Collection = Resource<Collection>(
162 "AWS.OpenSearchServerless.Collection",
163);
164
165export const CollectionProvider = () =>
166 Provider.effect(
167 Collection,
168 Effect.gen(function* () {
169 const createName = Effect.fn(function* (
170 id: string,
171 props: { collectionName?: string | undefined },
172 ) {
173 return (
174 props.collectionName ??
175 (yield* createPhysicalName({ id, maxLength: 32, lowercase: true }))
176 );
177 });
178
179 const toAttributes = (detail: aoss.CollectionDetail) => ({
180 collectionId: detail.id!,
181 collectionName: detail.name!,
182 collectionArn: detail.arn!,
183 type: detail.type,
184 status: detail.status,
185 kmsKeyArn: detail.kmsKeyArn,
186 collectionEndpoint: detail.collectionEndpoint,
187 dashboardEndpoint: detail.dashboardEndpoint,
188 });
189
190 const observeByName = Effect.fn(function* (name: string) {
191 const response = yield* aoss.batchGetCollection({ names: [name] });
192 return response.collectionDetails?.[0];
193 });
194
195 const syncTags = Effect.fn(function* (
196 arn: string,
197 desired: Record<string, string>,
198 ) {
199 const observed = yield* aoss
200 .listTagsForResource({ resourceArn: arn })
201 .pipe(Effect.map((r) => tagsToRecord(r.tags)));
202 const { upsert, removed } = diffTags(observed, desired);
203 if (upsert.length > 0) {
204 yield* aoss.tagResource({
205 resourceArn: arn,
206 tags: recordToTagList(
207 Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
208 ),
209 });
210 }
211 if (removed.length > 0) {
212 yield* aoss.untagResource({ resourceArn: arn, tagKeys: removed });
213 }
214 });
215
216 return Collection.Provider.of({
217 stables: ["collectionId", "collectionName", "collectionArn"],
218
219 list: () =>
220 Effect.gen(function* () {
221 const pages = yield* aoss.listCollections
222 .pages({})
223 .pipe(Stream.runCollect);
224 return Array.from(pages)
225 .flatMap((page) => page.collectionSummaries ?? [])
226 .filter(
227 (s) =>
228 s.id !== undefined &&
229 s.name !== undefined &&
230 s.arn !== undefined,
231 )
232 .map((s) => ({
233 collectionId: s.id!,
234 collectionName: s.name!,
235 collectionArn: s.arn!,
236 status: s.status,
237 kmsKeyArn: s.kmsKeyArn,
238 }));
239 }),
240
241 read: Effect.fn(function* ({ id, olds, output }) {
242 const name =
243 output?.collectionName ?? (yield* createName(id, olds ?? {}));
244 const detail = yield* observeByName(name);
245 if (detail?.id === undefined || detail.arn === undefined) {
246 return undefined;
247 }
248 const attrs = toAttributes(detail);
249 const tags = yield* aoss
250 .listTagsForResource({ resourceArn: detail.arn })
251 .pipe(
252 Effect.map((r) => r.tags),
253 Effect.catch(() => Effect.succeed(undefined)),
254 );
255 return (yield* hasAlchemyTags(
256 id,
257 (tags ?? []).map((t) => ({ Key: t.key, Value: t.value })),
258 ))
259 ? attrs
260 : Unowned(attrs);
261 }),
262
263 diff: Effect.fn(function* ({ id, news, olds }) {
264 if (!isResolved(news)) return undefined;
265 const oldName = yield* createName(id, olds);
266 const newName = yield* createName(id, news);
267 if (oldName !== newName) {
268 return { action: "replace" } as const;
269 }
270 if ((olds.type ?? "SEARCH") !== (news.type ?? "SEARCH")) {
271 return { action: "replace" } as const;
272 }
273 // description/deletionProtection/tags fall through to update
274 }),
275
276 reconcile: Effect.fn(function* ({ id, news, output, session }) {
277 const name = output?.collectionName ?? (yield* createName(id, news));
278 const type = news.type ?? "SEARCH";
279 const internalTags = yield* createInternalTags(id);
280 const desiredTags = { ...news.tags, ...internalTags };
281
282 // 1. OBSERVE — cloud state is authoritative
283 let detail = yield* observeByName(name);
284
285 // 2. ENSURE — create if missing; wait for ACTIVE (async provisioning)
286 if (detail === undefined) {
287 const created = yield* aoss
288 .createCollection({
289 name,
290 type,
291 description: news.description,
292 standbyReplicas: news.standbyReplicas,
293 deletionProtection: news.deletionProtection,
294 tags: recordToTagList(desiredTags),
295 })
296 .pipe(
297 Effect.map((r) => r.createCollectionDetail),
298 // a concurrent reconciler already created it — fall through
299 Effect.catchTag("ConflictException", () =>
300 Effect.succeed(undefined),
301 ),
302 );
303 const collectionId =
304 created?.id ?? (yield* observeByName(name))?.id;
305 if (collectionId === undefined) {
306 return yield* Effect.fail(
307 new aoss.ResourceNotFoundException({
308 message: `collection ${name} not visible after create`,
309 }),
310 );
311 }
312 yield* session.note(`creating collection ${name} (async)...`);
313 detail = yield* awaitCollectionActive(collectionId);
314 } else if (detail.status !== "ACTIVE") {
315 detail = yield* awaitCollectionActive(detail.id!);
316 }
317
318 // 3. SYNC — description (deletionProtection updates alongside)
319 const descriptionDrift =
320 news.description !== undefined &&
321 news.description !== detail.description;
322 const protectionDrift =
323 news.deletionProtection !== undefined &&
324 news.deletionProtection !== detail.deletionProtection;
325 if (descriptionDrift || protectionDrift) {
326 yield* aoss.updateCollection({
327 id: detail.id!,
328 description: descriptionDrift ? news.description : undefined,
329 deletionProtection: protectionDrift
330 ? news.deletionProtection
331 : undefined,
332 });
333 }
334
335 // 3b. SYNC TAGS — diff against observed cloud tags
336 yield* syncTags(detail.arn!, desiredTags);
337
338 yield* session.note(detail.id!);
339 return toAttributes(detail);
340 }),
341
342 delete: Effect.fn(function* ({ output }) {
343 yield* retryWhileConflict(
344 aoss.deleteCollection({ id: output.collectionId }),
345 ).pipe(
346 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
347 );
348 }),
349 });
350 }),
351 );