1import * as kendra from "@distilled.cloud/aws/kendra";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as EffectStream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import { createPhysicalName } from "../../PhysicalName.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import {
12 createInternalTags,
13 diffTags,
14 hasAlchemyTags,
15 type Tags,
16} from "../../Tags.ts";
17import { AWSEnvironment } from "../Environment.ts";
18import type { Providers } from "../Providers.ts";
19
20export type DataSourceStatus = kendra.DataSourceStatus;
21export type DataSourceType = kendra.DataSourceType;
22
23export interface DataSourceProps {
24 /**
25 * The identifier of the Kendra index the data source connects to.
26 * Changing it replaces the data source.
27 */
28 indexId: string;
29 /**
30 * Name of the data source.
31 * @default ${app}-${stage}-${id}
32 */
33 name?: string;
34 /**
35 * The connector type (e.g. `S3`, `TEMPLATE`, `WEBCRAWLER`, `CUSTOM`).
36 * Changing it replaces the data source.
37 */
38 type: DataSourceType;
39 /**
40 * Connector-specific configuration (e.g. `S3Configuration` or
41 * `TemplateConfiguration`). Required for all types except `CUSTOM`.
42 */
43 configuration?: kendra.DataSourceConfiguration;
44 /**
45 * VPC configuration for connectors that reach into a VPC.
46 */
47 vpcConfiguration?: kendra.DataSourceVpcConfiguration;
48 /**
49 * A description of the data source.
50 */
51 description?: string;
52 /**
53 * Sync schedule as a cron expression (e.g. `cron(0 12 * * ? *)`). When
54 * omitted, syncs run only on demand.
55 */
56 schedule?: string;
57 /**
58 * ARN of the IAM role Kendra assumes to access the source content.
59 * Required for all types except `CUSTOM`.
60 */
61 roleArn?: string;
62 /**
63 * The code for a language the source documents are in.
64 * @default "en"
65 */
66 languageCode?: string;
67 /**
68 * Alter document metadata/content during ingestion.
69 */
70 customDocumentEnrichmentConfiguration?: kendra.CustomDocumentEnrichmentConfiguration;
71 /**
72 * Tags to associate with the data source.
73 */
74 tags?: Record<string, string>;
75}
76
77export interface DataSource extends Resource<
78 "AWS.Kendra.DataSource",
79 DataSourceProps,
80 {
81 /**
82 * Service-assigned unique identifier of the data source (unique within
83 * its index).
84 */
85 id: string;
86 /**
87 * The identifier of the index the data source belongs to.
88 */
89 indexId: string;
90 /**
91 * ARN of the data source.
92 */
93 arn: string;
94 /**
95 * The data source's name.
96 */
97 name: string;
98 /**
99 * The connector type.
100 */
101 type: DataSourceType | undefined;
102 /**
103 * Current lifecycle status of the data source.
104 */
105 status: DataSourceStatus | undefined;
106 /**
107 * Current tags reported for the data source.
108 */
109 tags: Record<string, string>;
110 },
111 never,
112 Providers
113> {}
114
115/**
116 * An Amazon Kendra data source — a connector that syncs documents from a
117 * repository (S3 bucket, SharePoint, website, ...) into a Kendra index.
118 *
119 * ### Creating Data Sources
120 * **Example:** S3 Data Source
121 * ```typescript
122 * import * as AWS from "alchemy/AWS";
123 *
124 * const source = yield* AWS.Kendra.DataSource("Docs", {
125 * indexId: index.id,
126 * type: "S3",
127 * roleArn: dataSourceRole.roleArn,
128 * configuration: {
129 * S3Configuration: {
130 * BucketName: bucket.bucketName,
131 * },
132 * },
133 * });
134 * ```
135 *
136 * **Example:** Scheduled Sync
137 * ```typescript
138 * const source = yield* AWS.Kendra.DataSource("Docs", {
139 * indexId: index.id,
140 * type: "S3",
141 * roleArn: dataSourceRole.roleArn,
142 * schedule: "cron(0 12 * * ? *)",
143 * configuration: {
144 * S3Configuration: { BucketName: bucket.bucketName },
145 * },
146 * });
147 * ```
148 *
149 * @resource
150 */
151export const DataSource = Resource<DataSource>("AWS.Kendra.DataSource");
152
153const createDataSourceName = (
154 id: string,
155 props: { name?: string | undefined },
156) =>
157 props.name
158 ? Effect.succeed(props.name)
159 : createPhysicalName({ id, maxLength: 100 });
160
161const fetchTags = Effect.fn(function* (arn: string) {
162 const response = yield* kendra
163 .listTagsForResource({ ResourceARN: arn })
164 .pipe(
165 Effect.catchTag(
166 ["ResourceNotFoundException", "ResourceUnavailableException"],
167 () => Effect.succeed(undefined),
168 ),
169 );
170 return Object.fromEntries(
171 (response?.Tags ?? []).map((tag) => [tag.Key, tag.Value]),
172 );
173});
174
175const currentArnOf = Effect.gen(function* () {
176 const { accountId, region } = yield* AWSEnvironment.current;
177 return (indexId: string, dataSourceId: string) =>
178 `arn:aws:kendra:${region}:${accountId}:index/${indexId}/data-source/${dataSourceId}`;
179});
180
181interface DataSourceState {
182 attrs: DataSource["Attributes"];
183 described: kendra.DescribeDataSourceResponse;
184}
185
186const readDataSourceById = Effect.fn(function* (
187 indexId: string,
188 dataSourceId: string,
189 arnOf: (indexId: string, dataSourceId: string) => string,
190) {
191 const described = yield* kendra
192 .describeDataSource({ IndexId: indexId, Id: dataSourceId })
193 .pipe(
194 Effect.catchTag("ResourceNotFoundException", () =>
195 Effect.succeed(undefined),
196 ),
197 );
198 if (!described || described.Status === "DELETING") return undefined;
199 const arn = arnOf(described.IndexId ?? indexId, described.Id ?? dataSourceId);
200 const state: DataSourceState = {
201 described,
202 attrs: {
203 id: described.Id ?? dataSourceId,
204 indexId: described.IndexId ?? indexId,
205 arn,
206 name: described.Name ?? "",
207 type: described.Type,
208 status: described.Status,
209 tags: yield* fetchTags(arn),
210 },
211 };
212 return state;
213});
214
215const findDataSourceByName = Effect.fn(function* (
216 indexId: string,
217 name: string,
218 arnOf: (indexId: string, dataSourceId: string) => string,
219) {
220 const summaries = yield* kendra.listDataSources
221 .pages({ IndexId: indexId })
222 .pipe(
223 EffectStream.runCollect,
224 Effect.map((chunk) =>
225 Array.from(chunk).flatMap((page) => page.SummaryItems ?? []),
226 ),
227 // The parent index may itself be gone.
228 Effect.catchTag("ResourceNotFoundException", () =>
229 Effect.succeed([] as kendra.DataSourceSummary[]),
230 ),
231 );
232 const match = summaries.find(
233 (summary) => summary.Name === name && summary.Status !== "DELETING",
234 );
235 if (!match?.Id) return undefined;
236 return yield* readDataSourceById(indexId, match.Id, arnOf);
237});
238
239/**
240 * A data source still transitioning toward the awaited status — retried by
241 * {@link waitForDataSourceStatus}'s bounded schedule.
242 */
243class DataSourceNotReady extends Data.TaggedError("DataSourceNotReady")<{
244 readonly id: string;
245 readonly status: string | undefined;
246}> {}
247
248/**
249 * A data source whose asynchronous provisioning converged to the terminal
250 * `FAILED` status.
251 */
252export class DataSourceProvisioningFailed extends Data.TaggedError(
253 "DataSourceProvisioningFailed",
254)<{
255 readonly id: string;
256 readonly message: string | undefined;
257}> {}
258
259// Explicitly-typed retry wrapper — an inline `Effect.retry` in provider
260// lifecycle code leaks `Retry.Return`'s conditional type into declaration
261// emit and widens the provider layer to `unknown` for every consumer of
262// `AWS.providers()`.
263const retryWhileNotReady = <A, E extends { readonly _tag: string }, R>(
264 self: Effect.Effect<A, E, R>,
265): Effect.Effect<A, E, R> =>
266 Effect.retry(self, {
267 while: (e) => e._tag === "DataSourceNotReady",
268 // Data source provisioning is usually well under a minute; budget ~5 min.
269 schedule: Schedule.max([Schedule.spaced("5 seconds"), Schedule.recurs(60)]),
270 });
271
272// CreateDataSource validates the IAM role up front; a freshly-created role
273// may not have propagated yet and surfaces as ValidationException. Bounded
274// retry through the propagation window (~60s). Explicitly typed for the same
275// declaration-emit reason as retryWhileNotReady.
276const retryThroughIamPropagation = <A, E extends { readonly _tag: string }, R>(
277 self: Effect.Effect<A, E, R>,
278): Effect.Effect<A, E, R> =>
279 Effect.retry(self, {
280 while: (e) => e._tag === "ValidationException",
281 schedule: Schedule.max([Schedule.spaced("5 seconds"), Schedule.recurs(12)]),
282 });
283
284const waitForDataSourceStatus = (
285 indexId: string,
286 id: string,
287 target: "ACTIVE" | "DELETED",
288) =>
289 retryWhileNotReady(
290 Effect.gen(function* () {
291 const described = yield* kendra
292 .describeDataSource({ IndexId: indexId, Id: id })
293 .pipe(
294 Effect.catchTag("ResourceNotFoundException", () =>
295 Effect.succeed(undefined),
296 ),
297 );
298 if (target === "DELETED") {
299 if (described === undefined) return;
300 return yield* Effect.fail(
301 new DataSourceNotReady({ id, status: described.Status }),
302 );
303 }
304 if (described?.Status === "ACTIVE") return;
305 if (described?.Status === "FAILED") {
306 return yield* Effect.fail(
307 new DataSourceProvisioningFailed({
308 id,
309 message: described.ErrorMessage,
310 }),
311 );
312 }
313 return yield* Effect.fail(
314 new DataSourceNotReady({ id, status: described?.Status }),
315 );
316 }),
317 );
318
319export const DataSourceProvider = () =>
320 Provider.effect(
321 DataSource,
322 Effect.gen(function* () {
323 return {
324 stables: ["id", "indexId", "arn"],
325 // Keyed by a parent index; cannot be enumerated account-wide without
326 // iterating every index — treated as a sub-resource per the factory
327 // list() convention.
328 list: () => Effect.succeed([]),
329 read: Effect.fn(function* ({ id, olds, output }) {
330 const arnOf = yield* currentArnOf;
331 const indexId = output?.indexId ?? olds?.indexId;
332 if (indexId === undefined) return undefined;
333 const state = output?.id
334 ? yield* readDataSourceById(indexId, output.id, arnOf)
335 : yield* findDataSourceByName(
336 indexId,
337 yield* createDataSourceName(id, olds ?? {}),
338 arnOf,
339 );
340 if (!state) return undefined;
341 return (yield* hasAlchemyTags(id, state.attrs.tags as Tags))
342 ? state.attrs
343 : Unowned(state.attrs);
344 }),
345 diff: Effect.fn(function* ({ news, olds }) {
346 if (!isResolved(news)) return;
347 if (olds === undefined) return;
348 // The parent index and connector type are fixed at creation.
349 if (olds.indexId !== news.indexId || olds.type !== news.type) {
350 return { action: "replace" } as const;
351 }
352 }),
353 reconcile: Effect.fn(function* ({ id, news, output, session }) {
354 if (!news) {
355 return yield* Effect.fail(
356 new Error("Kendra DataSource requires props"),
357 );
358 }
359 const arnOf = yield* currentArnOf;
360 const indexId = news.indexId;
361 const name = yield* createDataSourceName(id, news);
362 const internalTags = yield* createInternalTags(id);
363 const desiredTags = { ...internalTags, ...news.tags };
364
365 // Observe — prefer the cached id; fall back to a name lookup so a
366 // create whose state failed to persist is adopted, not duplicated.
367 let state = output?.id
368 ? yield* readDataSourceById(indexId, output.id, arnOf)
369 : yield* findDataSourceByName(indexId, name, arnOf);
370
371 // Ensure — create if missing, then wait for ACTIVE.
372 if (state === undefined) {
373 const created = yield* retryThroughIamPropagation(
374 kendra.createDataSource({
375 IndexId: indexId,
376 Name: name,
377 Type: news.type,
378 Configuration: news.configuration,
379 VpcConfiguration: news.vpcConfiguration,
380 Description: news.description,
381 Schedule: news.schedule,
382 RoleArn: news.roleArn,
383 LanguageCode: news.languageCode,
384 CustomDocumentEnrichmentConfiguration:
385 news.customDocumentEnrichmentConfiguration,
386 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
387 Key,
388 Value,
389 })),
390 }),
391 );
392 yield* session.note(
393 `Creating data source ${name} (${created.Id})...`,
394 );
395 yield* waitForDataSourceStatus(indexId, created.Id, "ACTIVE");
396 state = yield* readDataSourceById(indexId, created.Id, arnOf);
397 if (state === undefined) {
398 return yield* Effect.fail(
399 new Error(`failed to read created data source ${name}`),
400 );
401 }
402 }
403
404 // Sync mutable settings via UpdateDataSource — only when drifted.
405 const described = state.described;
406 const needsUpdate =
407 name !== described.Name ||
408 (news.description ?? "") !== (described.Description ?? "") ||
409 (news.schedule ?? "") !== (described.Schedule ?? "") ||
410 (news.roleArn !== undefined &&
411 news.roleArn !== described.RoleArn) ||
412 (news.languageCode !== undefined &&
413 news.languageCode !== described.LanguageCode) ||
414 news.configuration !== undefined ||
415 news.vpcConfiguration !== undefined ||
416 news.customDocumentEnrichmentConfiguration !== undefined;
417 if (needsUpdate) {
418 yield* kendra.updateDataSource({
419 Id: state.attrs.id,
420 IndexId: indexId,
421 Name: name,
422 Configuration: news.configuration,
423 VpcConfiguration: news.vpcConfiguration,
424 Description: news.description,
425 Schedule: news.schedule,
426 RoleArn: news.roleArn,
427 LanguageCode: news.languageCode,
428 CustomDocumentEnrichmentConfiguration:
429 news.customDocumentEnrichmentConfiguration,
430 });
431 yield* waitForDataSourceStatus(indexId, state.attrs.id, "ACTIVE");
432 yield* session.note(`Updated data source ${name}`);
433 }
434
435 // Sync tags — diff against observed cloud tags.
436 const { removed, upsert } = diffTags(state.attrs.tags, desiredTags);
437 if (removed.length > 0) {
438 yield* kendra.untagResource({
439 ResourceARN: state.attrs.arn,
440 TagKeys: removed,
441 });
442 }
443 if (upsert.length > 0) {
444 yield* kendra.tagResource({
445 ResourceARN: state.attrs.arn,
446 Tags: upsert.map(({ Key, Value }) => ({ Key, Value })),
447 });
448 }
449
450 yield* session.note(state.attrs.arn);
451
452 const final = yield* readDataSourceById(
453 indexId,
454 state.attrs.id,
455 arnOf,
456 );
457 if (!final) {
458 return yield* Effect.fail(
459 new Error(`failed to read reconciled data source ${name}`),
460 );
461 }
462 return final.attrs;
463 }),
464 delete: Effect.fn(function* ({ output }) {
465 yield* kendra
466 .deleteDataSource({ IndexId: output.indexId, Id: output.id })
467 .pipe(
468 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
469 );
470 yield* waitForDataSourceStatus(output.indexId, output.id, "DELETED");
471 }),
472 };
473 }),
474 );