1import * as appintegrations from "@distilled.cloud/aws/appintegrations";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
11import type { Providers } from "../Providers.ts";
12import { definedTags } from "./internal.ts";
13
14export interface DataIntegrationScheduleConfig {
15 /**
16 * The start date for objects to import, in ISO 8601 format,
17 * e.g. `2024-01-01T00:00:00Z`.
18 */
19 firstExecutionFrom?: string;
20 /**
21 * The name of the object to pull from the data source.
22 */
23 object?: string;
24 /**
25 * How often the data should be pulled from the data source, e.g.
26 * `rate(1 hours)`.
27 */
28 scheduleExpression: string;
29}
30
31export interface DataIntegrationProps {
32 /**
33 * Name of the data integration. If omitted, a unique name is generated
34 * from the app, stage, and logical ID. The name can be updated in place.
35 */
36 name?: string;
37 /**
38 * Description of the data integration (1-1000 characters).
39 */
40 description?: string;
41 /**
42 * The ARN of the KMS key used to encrypt the data integration. Changing
43 * the key replaces the data integration.
44 */
45 kmsKey: string;
46 /**
47 * The URI of the data source, e.g. `s3://my-bucket` or
48 * `Salesforce://AppFlow/my-connector-profile`. Changing the source URI
49 * replaces the data integration.
50 */
51 sourceURI: string;
52 /**
53 * The scheduling configuration for pulling data from the source. Required
54 * for SaaS (AppFlow) sources; must be omitted for S3 sources. Changing it
55 * replaces the data integration.
56 */
57 scheduleConfig?: DataIntegrationScheduleConfig;
58 /**
59 * The configuration for which files to pull from the source (folders and
60 * filters). Changing it replaces the data integration.
61 */
62 fileConfiguration?: {
63 /**
64 * Identifiers for the source folders to pull all files from recursively.
65 */
66 folders: string[];
67 /**
68 * Restrictions for what files should be pulled from the source.
69 */
70 filters?: Record<string, string[]>;
71 };
72 /**
73 * The configuration for which objects and fields to pull from the source.
74 * Changing it replaces the data integration.
75 */
76 objectConfiguration?: Record<string, Record<string, string[]>>;
77 /**
78 * Tags to apply to the data integration. Merged with internal Alchemy tags.
79 */
80 tags?: Record<string, string>;
81}
82
83export interface DataIntegration extends Resource<
84 "AWS.AppIntegrations.DataIntegration",
85 DataIntegrationProps,
86 {
87 dataIntegrationId: string;
88 dataIntegrationArn: string;
89 dataIntegrationName: string;
90 kmsKey: string;
91 sourceURI: string;
92 },
93 never,
94 Providers
95> {}
96
97/**
98 * An Amazon AppIntegrations data integration. Data integrations reference an
99 * external data source (an S3 bucket, or a SaaS application through Amazon
100 * AppFlow) so services like Amazon Q in Connect can ingest its content.
101 *
102 * The KMS key, source URI, schedule, file configuration, and object
103 * configuration are immutable; changing any of them replaces the data
104 * integration. Only the name and description can be updated in place.
105 * ### Creating a Data Integration
106 * **Example:** S3 Data Integration
107 * ```typescript
108 * import * as AppIntegrations from "alchemy/AWS/AppIntegrations";
109 * import * as KMS from "alchemy/AWS/KMS";
110 * import * as S3 from "alchemy/AWS/S3";
111 *
112 * const bucket = yield* S3.Bucket("Content");
113 * const key = yield* KMS.Key("ContentKey");
114 *
115 * const integration = yield* AppIntegrations.DataIntegration("Content", {
116 * kmsKey: key.keyArn,
117 * sourceURI: Output.interpolate`s3://${bucket.bucketName}`,
118 * });
119 * ```
120 *
121 * **Example:** Scheduled SaaS Data Integration
122 * ```typescript
123 * const integration = yield* AppIntegrations.DataIntegration("Salesforce", {
124 * kmsKey: key.keyArn,
125 * sourceURI: "Salesforce://AppFlow/my-connector-profile",
126 * scheduleConfig: {
127 * firstExecutionFrom: "2024-01-01T00:00:00Z",
128 * object: "Account",
129 * scheduleExpression: "rate(1 hours)",
130 * },
131 * });
132 * ```
133 *
134 * @resource
135 */
136export const DataIntegration = Resource<DataIntegration>(
137 "AWS.AppIntegrations.DataIntegration",
138);
139
140/**
141 * Raised when the AppIntegrations API returns a data integration without
142 * the fields required to build the resource attributes.
143 */
144export class DataIntegrationIncomplete extends Data.TaggedError(
145 "DataIntegrationIncomplete",
146)<{ message: string }> {}
147
148export const DataIntegrationProvider = () =>
149 Provider.effect(
150 DataIntegration,
151 Effect.gen(function* () {
152 const createName = Effect.fn(function* (
153 id: string,
154 props: { name?: string },
155 ) {
156 return (
157 props.name ?? (yield* createPhysicalName({ id, maxLength: 255 }))
158 );
159 });
160
161 /** Get a single data integration by ARN or ID; undefined if absent. */
162 const observe = (arnOrId: string) =>
163 appintegrations
164 .getDataIntegration({ Identifier: arnOrId })
165 .pipe(
166 Effect.catchTag("ResourceNotFoundException", () =>
167 Effect.succeed(undefined),
168 ),
169 );
170
171 /** Find a data integration ARN by name via list enumeration. */
172 const findByName = (name: string) =>
173 appintegrations.listDataIntegrations.items({}).pipe(
174 Stream.filter((item) => item.Name === name),
175 Stream.take(1),
176 Stream.runCollect,
177 Effect.map((chunk) => Array.from(chunk)[0]?.Arn),
178 );
179
180 const toAttrs = Effect.fn(function* (
181 live: appintegrations.GetDataIntegrationResponse,
182 ) {
183 if (
184 live.Arn === undefined ||
185 live.Id === undefined ||
186 live.Name === undefined ||
187 live.KmsKey === undefined ||
188 live.SourceURI === undefined
189 ) {
190 return yield* new DataIntegrationIncomplete({
191 message: `data integration '${live.Name}' is missing Arn, Id, Name, KmsKey, or SourceURI`,
192 });
193 }
194 return {
195 dataIntegrationId: live.Id,
196 dataIntegrationArn: live.Arn,
197 dataIntegrationName: live.Name,
198 kmsKey: live.KmsKey,
199 sourceURI: live.SourceURI,
200 };
201 });
202
203 const toScheduleConfiguration = (
204 config: DataIntegrationScheduleConfig,
205 ): appintegrations.ScheduleConfiguration => ({
206 ScheduleExpression: config.scheduleExpression,
207 ...(config.firstExecutionFrom !== undefined
208 ? { FirstExecutionFrom: config.firstExecutionFrom }
209 : {}),
210 ...(config.object !== undefined ? { Object: config.object } : {}),
211 });
212
213 return DataIntegration.Provider.of({
214 stables: [
215 "dataIntegrationId",
216 "dataIntegrationArn",
217 "kmsKey",
218 "sourceURI",
219 ],
220
221 // The list API only returns Arn/Name/SourceURI summaries — hydrate
222 // each into the full attributes shape, tolerating per-item NotFound
223 // races.
224 list: () =>
225 Effect.gen(function* () {
226 const arns = yield* appintegrations.listDataIntegrations
227 .items({})
228 .pipe(
229 Stream.runCollect,
230 Effect.map((chunk) =>
231 Array.from(chunk).flatMap((item) =>
232 item.Arn !== undefined ? [item.Arn] : [],
233 ),
234 ),
235 );
236 const items = yield* Effect.forEach(
237 arns,
238 (arn) =>
239 observe(arn).pipe(
240 Effect.flatMap((live) =>
241 live === undefined
242 ? Effect.succeed(undefined)
243 : toAttrs(live),
244 ),
245 ),
246 { concurrency: 10 },
247 );
248 return items.filter(
249 (item): item is DataIntegration["Attributes"] =>
250 item !== undefined,
251 );
252 }),
253
254 read: Effect.fn(function* ({ id, olds, output }) {
255 let identifier: string | undefined = output?.dataIntegrationId;
256 if (identifier === undefined) {
257 const name = yield* createName(id, olds ?? {});
258 identifier = yield* findByName(name);
259 }
260 if (identifier === undefined) return undefined;
261 const live = yield* observe(identifier);
262 if (live === undefined) return undefined;
263 const attrs = yield* toAttrs(live);
264 const tags = definedTags(live.Tags);
265 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
266 }),
267
268 diff: Effect.fn(function* ({ news, olds }) {
269 if (!isResolved(news)) return undefined;
270 if (olds === undefined) return undefined;
271 if (
272 olds.kmsKey !== news.kmsKey ||
273 olds.sourceURI !== news.sourceURI ||
274 JSON.stringify(olds.scheduleConfig) !==
275 JSON.stringify(news.scheduleConfig) ||
276 JSON.stringify(olds.fileConfiguration) !==
277 JSON.stringify(news.fileConfiguration) ||
278 JSON.stringify(olds.objectConfiguration) !==
279 JSON.stringify(news.objectConfiguration)
280 ) {
281 return { action: "replace" } as const;
282 }
283 // fall through: default update path (name, description, tags)
284 }),
285
286 reconcile: Effect.fn(function* ({ id, news, output, session }) {
287 const name = yield* createName(id, news);
288 const internalTags = yield* createInternalTags(id);
289 const desiredTags = { ...news.tags, ...internalTags };
290
291 // 1. Observe — prefer the cached ID; fall back to enumerating by
292 // name so a lost-state re-run converges.
293 let identifier: string | undefined = output?.dataIntegrationId;
294 let live =
295 identifier === undefined ? undefined : yield* observe(identifier);
296 if (live === undefined) {
297 identifier = yield* findByName(name);
298 live =
299 identifier === undefined ? undefined : yield* observe(identifier);
300 }
301
302 // 2. Ensure — create if missing.
303 if (live === undefined) {
304 const created = yield* appintegrations.createDataIntegration({
305 Name: name,
306 Description: news.description,
307 KmsKey: news.kmsKey,
308 SourceURI: news.sourceURI,
309 ScheduleConfig: news.scheduleConfig
310 ? toScheduleConfiguration(news.scheduleConfig)
311 : undefined,
312 FileConfiguration: news.fileConfiguration
313 ? {
314 Folders: news.fileConfiguration.folders,
315 ...(news.fileConfiguration.filters
316 ? { Filters: news.fileConfiguration.filters }
317 : {}),
318 }
319 : undefined,
320 ObjectConfiguration: news.objectConfiguration,
321 Tags: desiredTags,
322 });
323 if (created.Arn === undefined) {
324 return yield* new DataIntegrationIncomplete({
325 message: `createDataIntegration for '${name}' returned no Arn`,
326 });
327 }
328 live = yield* appintegrations.getDataIntegration({
329 Identifier: created.Arn,
330 });
331 }
332 const attrs = yield* toAttrs(live);
333
334 // 3. Sync mutable aspects — only the name and description can be
335 // updated in place. The API cannot clear a description (min
336 // length 1), so only push a defined value that differs.
337 const update: Omit<
338 appintegrations.UpdateDataIntegrationRequest,
339 "Identifier"
340 > = {};
341 if (live.Name !== name) {
342 update.Name = name;
343 }
344 if (
345 news.description !== undefined &&
346 news.description !== live.Description
347 ) {
348 update.Description = news.description;
349 }
350 if (Object.keys(update).length > 0) {
351 yield* appintegrations.updateDataIntegration({
352 Identifier: attrs.dataIntegrationId,
353 ...update,
354 });
355 }
356
357 // 4. Sync tags — diff against OBSERVED cloud tags so adoption
358 // converges.
359 const observedTags = definedTags(live.Tags);
360 const { upsert, removed } = diffTags(observedTags, desiredTags);
361 if (upsert.length > 0) {
362 yield* appintegrations.tagResource({
363 resourceArn: attrs.dataIntegrationArn,
364 tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
365 });
366 }
367 if (removed.length > 0) {
368 yield* appintegrations.untagResource({
369 resourceArn: attrs.dataIntegrationArn,
370 tagKeys: removed,
371 });
372 }
373
374 yield* session.note(name);
375 return { ...attrs, dataIntegrationName: name };
376 }),
377
378 delete: Effect.fn(function* ({ output }) {
379 yield* appintegrations
380 .deleteDataIntegration({
381 DataIntegrationIdentifier: output.dataIntegrationId,
382 })
383 .pipe(
384 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
385 );
386 }),
387 });
388 }),
389 );