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