1import * as glue from "@distilled.cloud/aws/glue";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
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, hasAlchemyTags } from "../../Tags.ts";
11import { AWSEnvironment } from "../Environment.ts";
12import type { Providers } from "../Providers.ts";
13import {
14 crawlerArn,
15 fetchObservedTags,
16 retryCrawlerDelete,
17 retryWhileCrawlerTargetNotReady,
18 retryWhileCrawlerRunning,
19 retryWhileRoleNotAssumable,
20 syncTags,
21} from "./internal.ts";
22
23export interface CrawlerS3Target {
24 /** S3 path to crawl, e.g. `s3://my-bucket/data/`. */
25 path: string;
26 /** Glob patterns to exclude. */
27 exclusions?: string[];
28 /** Name of a Glue connection to use (for VPC-scoped S3). */
29 connectionName?: string;
30 /** Sample size (files per leaf) to speed up crawling. */
31 sampleSize?: number;
32 /** SQS queue ARN for event-based (incremental) crawling. */
33 eventQueueArn?: string;
34 /** Dead-letter SQS queue ARN for event-based crawling. */
35 dlqEventQueueArn?: string;
36}
37
38export interface CrawlerJdbcTarget {
39 /** Name of the Glue connection to the JDBC source. */
40 connectionName?: string;
41 /** The path of the JDBC target, e.g. `database/schema/%`. */
42 path?: string;
43 /** Glob patterns to exclude. */
44 exclusions?: string[];
45}
46
47export interface CrawlerDynamoDbTarget {
48 /** The DynamoDB table name to crawl. */
49 path?: string;
50 /** Whether to scan all records (vs a sample). */
51 scanAll?: boolean;
52 /** Fraction of the table's read capacity to consume (0.1–1.5). */
53 scanRate?: number;
54}
55
56export interface CrawlerCatalogTarget {
57 /** The database of the catalog tables to recrawl. */
58 databaseName: string;
59 /** The tables in the database to recrawl. */
60 tables: string[];
61 /** Name of a Glue connection. */
62 connectionName?: string;
63}
64
65export interface CrawlerTargets {
66 /** S3 data-store targets. */
67 s3Targets?: CrawlerS3Target[];
68 /** JDBC data-store targets. */
69 jdbcTargets?: CrawlerJdbcTarget[];
70 /** DynamoDB targets. */
71 dynamoDbTargets?: CrawlerDynamoDbTarget[];
72 /** Glue Data Catalog targets (recrawl existing tables). */
73 catalogTargets?: CrawlerCatalogTarget[];
74}
75
76export interface CrawlerProps {
77 /**
78 * Name of the crawler. If omitted, a unique name is generated. Changing the
79 * name replaces the crawler.
80 * @default a generated physical name
81 */
82 crawlerName?: string;
83 /**
84 * The IAM role (ARN or name) the crawler assumes to access data stores and
85 * write to the Data Catalog.
86 */
87 role: string;
88 /**
89 * The Glue database where the crawler writes discovered tables.
90 */
91 databaseName?: string;
92 /**
93 * A description of the crawler.
94 */
95 description?: string;
96 /**
97 * The data stores to crawl.
98 */
99 targets: CrawlerTargets;
100 /**
101 * A `cron(...)` schedule expression. Omit for on-demand crawling.
102 * @example "cron(0 12 * * ? *)"
103 */
104 schedule?: string;
105 /**
106 * A prefix prepended to the names of tables the crawler creates.
107 */
108 tablePrefix?: string;
109 /**
110 * Custom classifier names, in priority order.
111 */
112 classifiers?: string[];
113 /**
114 * How the crawler handles schema changes and deleted objects.
115 */
116 schemaChangePolicy?: {
117 /** What to do when a schema changes: `LOG` or `UPDATE_IN_DATABASE`. */
118 updateBehavior?: "LOG" | "UPDATE_IN_DATABASE";
119 /**
120 * What to do with deleted objects: `LOG`, `DELETE_FROM_DATABASE`, or
121 * `DEPRECATE_IN_DATABASE`.
122 */
123 deleteBehavior?: "LOG" | "DELETE_FROM_DATABASE" | "DEPRECATE_IN_DATABASE";
124 };
125 /**
126 * Incremental-crawl behavior.
127 */
128 recrawlPolicy?: {
129 /**
130 * `CRAWL_EVERYTHING`, `CRAWL_NEW_FOLDERS_ONLY`, or `CRAWL_EVENT_MODE`.
131 */
132 recrawlBehavior?:
133 | "CRAWL_EVERYTHING"
134 | "CRAWL_NEW_FOLDERS_ONLY"
135 | "CRAWL_EVENT_MODE";
136 };
137 /**
138 * Crawler configuration JSON string (grouping/partitions behavior).
139 */
140 configuration?: string;
141 /**
142 * Tags to apply to the crawler. Merged with internal Alchemy tags.
143 */
144 tags?: Record<string, string>;
145}
146
147export interface Crawler extends Resource<
148 "AWS.Glue.Crawler",
149 CrawlerProps,
150 {
151 /** The name of the crawler. */
152 crawlerName: string;
153 /** The ARN of the crawler. */
154 crawlerArn: string;
155 /** The IAM role the crawler assumes. */
156 role: string;
157 /** The Glue database the crawler writes discovered tables to. */
158 databaseName: string | undefined;
159 /** The crawler state: `READY`, `RUNNING`, or `STOPPING`. */
160 state: string | undefined;
161 },
162 {},
163 Providers
164> {}
165
166/**
167 * An AWS Glue crawler — connects to an S3 (or JDBC/DynamoDB/catalog) data
168 * store, infers schemas, and populates the Glue Data Catalog with tables.
169 * Runs are asynchronous: create the crawler, then invoke `startCrawler` (or
170 * attach a schedule).
171 * ### Creating Crawlers
172 * **Example:** S3 Crawler
173 * ```typescript
174 * import * as AWS from "alchemy/AWS";
175 *
176 * const database = yield* AWS.Glue.Database("Analytics", {
177 * databaseName: "analytics",
178 * });
179 *
180 * const crawler = yield* AWS.Glue.Crawler("EventsCrawler", {
181 * role: crawlerRole.roleArn,
182 * databaseName: database.databaseName,
183 * targets: {
184 * s3Targets: [{ path: "s3://my-data-lake/events/" }],
185 * },
186 * });
187 * ```
188 *
189 * **Example:** Scheduled Crawler with Schema Policy
190 * ```typescript
191 * const crawler = yield* AWS.Glue.Crawler("EventsCrawler", {
192 * role: crawlerRole.roleArn,
193 * databaseName: database.databaseName,
194 * targets: { s3Targets: [{ path: "s3://my-data-lake/events/" }] },
195 * schedule: "cron(0 12 * * ? *)",
196 * tablePrefix: "raw_",
197 * schemaChangePolicy: {
198 * updateBehavior: "UPDATE_IN_DATABASE",
199 * deleteBehavior: "DEPRECATE_IN_DATABASE",
200 * },
201 * });
202 * ```
203 *
204 * @resource
205 */
206export const Crawler = Resource<Crawler>("AWS.Glue.Crawler");
207
208const toTargets = (targets: CrawlerTargets) => ({
209 S3Targets: targets.s3Targets?.map((t) => ({
210 Path: t.path,
211 Exclusions: t.exclusions,
212 ConnectionName: t.connectionName,
213 SampleSize: t.sampleSize,
214 EventQueueArn: t.eventQueueArn,
215 DlqEventQueueArn: t.dlqEventQueueArn,
216 })),
217 JdbcTargets: targets.jdbcTargets?.map((t) => ({
218 ConnectionName: t.connectionName,
219 Path: t.path,
220 Exclusions: t.exclusions,
221 })),
222 DynamoDBTargets: targets.dynamoDbTargets?.map((t) => ({
223 Path: t.path,
224 scanAll: t.scanAll,
225 scanRate: t.scanRate,
226 })),
227 CatalogTargets: targets.catalogTargets?.map((t) => ({
228 DatabaseName: t.databaseName,
229 Tables: t.tables,
230 ConnectionName: t.connectionName,
231 })),
232});
233
234export const CrawlerProvider = () =>
235 Provider.effect(
236 Crawler,
237 Effect.gen(function* () {
238 const createName = Effect.fn(function* (
239 id: string,
240 props: { crawlerName?: string | undefined },
241 ) {
242 return (
243 props.crawlerName ??
244 (yield* createPhysicalName({ id, maxLength: 255 }))
245 );
246 });
247
248 const observe = Effect.fn(function* (name: string) {
249 return yield* glue.getCrawler({ Name: name }).pipe(
250 Effect.map((r) => r.Crawler),
251 Effect.catchTag("EntityNotFoundException", () =>
252 Effect.succeed(undefined),
253 ),
254 );
255 });
256
257 return Crawler.Provider.of({
258 stables: ["crawlerName", "crawlerArn"],
259
260 list: () =>
261 Effect.gen(function* () {
262 const { accountId, region } = yield* AWSEnvironment.current;
263 const pages = yield* glue.getCrawlers
264 .pages({})
265 .pipe(Stream.runCollect);
266 return Array.from(pages)
267 .flatMap((page) => page.Crawlers ?? [])
268 .filter((c) => c.Name !== undefined)
269 .map((c) => ({
270 crawlerName: c.Name!,
271 crawlerArn: crawlerArn(region, accountId, c.Name!),
272 role: c.Role ?? "",
273 databaseName: c.DatabaseName,
274 state: c.State,
275 }));
276 }),
277
278 read: Effect.fn(function* ({ id, olds, output }) {
279 const { accountId, region } = yield* AWSEnvironment.current;
280 const name =
281 output?.crawlerName ?? (yield* createName(id, olds ?? {}));
282 const crawler = yield* observe(name);
283 if (crawler?.Name === undefined) return undefined;
284 const arn = crawlerArn(region, accountId, crawler.Name);
285 const attrs = {
286 crawlerName: crawler.Name,
287 crawlerArn: arn,
288 role: crawler.Role ?? "",
289 databaseName: crawler.DatabaseName,
290 state: crawler.State,
291 };
292 const tags = yield* fetchObservedTags(arn);
293 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
294 }),
295
296 diff: Effect.fn(function* ({ id, news, olds }) {
297 if (!isResolved(news)) return undefined;
298 const oldName = yield* createName(id, olds);
299 const newName = yield* createName(id, news);
300 if (oldName !== newName) return { action: "replace" } as const;
301 // role / targets / schedule / policies / description → update
302 }),
303
304 reconcile: Effect.fn(function* ({ id, news, output, session }) {
305 const { accountId, region } = yield* AWSEnvironment.current;
306 const name = output?.crawlerName ?? (yield* createName(id, news));
307 const internalTags = yield* createInternalTags(id);
308 const desiredTags = { ...news.tags, ...internalTags };
309 const arn = crawlerArn(region, accountId, name);
310
311 const common = {
312 Role: news.role,
313 DatabaseName: news.databaseName,
314 Description: news.description,
315 Targets: toTargets(news.targets),
316 Schedule: news.schedule,
317 Classifiers: news.classifiers,
318 TablePrefix: news.tablePrefix,
319 SchemaChangePolicy: news.schemaChangePolicy
320 ? {
321 UpdateBehavior: news.schemaChangePolicy.updateBehavior,
322 DeleteBehavior: news.schemaChangePolicy.deleteBehavior,
323 }
324 : undefined,
325 RecrawlPolicy: news.recrawlPolicy
326 ? { RecrawlBehavior: news.recrawlPolicy.recrawlBehavior }
327 : undefined,
328 Configuration: news.configuration,
329 };
330
331 // 1. OBSERVE
332 let crawler = yield* observe(name);
333
334 // 2. ENSURE / 3. SYNC
335 if (crawler === undefined) {
336 yield* retryWhileCrawlerTargetNotReady(
337 retryWhileRoleNotAssumable(
338 glue.createCrawler({
339 Name: name,
340 ...common,
341 Tags: desiredTags,
342 }),
343 ),
344 ).pipe(
345 Effect.catchTag("AlreadyExistsException", () => Effect.void),
346 );
347 } else {
348 // updateCrawler fails with CrawlerRunningException mid-crawl and
349 // with GlueRoleNotAssumable during IAM propagation.
350 yield* retryWhileCrawlerRunning(
351 retryWhileCrawlerTargetNotReady(
352 retryWhileRoleNotAssumable(
353 glue.updateCrawler({ Name: name, ...common }),
354 ),
355 ),
356 );
357 }
358
359 // 3b. SYNC TAGS
360 const observedTags = yield* fetchObservedTags(arn);
361 yield* syncTags(arn, observedTags, desiredTags);
362
363 crawler = yield* observe(name);
364
365 yield* session.note(name);
366 return {
367 crawlerName: name,
368 crawlerArn: arn,
369 role: crawler?.Role ?? news.role,
370 databaseName: crawler?.DatabaseName ?? news.databaseName,
371 state: crawler?.State,
372 };
373 }),
374
375 delete: Effect.fn(function* ({ output }) {
376 const name = output.crawlerName;
377 let crawler = yield* observe(name);
378 if (crawler === undefined) return;
379
380 if (crawler.State === "RUNNING") {
381 yield* glue
382 .stopCrawler({ Name: name })
383 .pipe(
384 Effect.catchTag(
385 [
386 "CrawlerNotRunningException",
387 "CrawlerStoppingException",
388 "EntityNotFoundException",
389 ],
390 () => Effect.void,
391 ),
392 );
393 }
394
395 // Stop is asynchronous. Wait for READY (deletable) or absence;
396 // interrupted deletes can resume from RUNNING/STOPPING safely.
397 crawler = yield* Effect.repeat(observe(name), {
398 schedule: Schedule.fixed("2 seconds"),
399 until: (current) =>
400 current === undefined || current.State === "READY",
401 times: 15,
402 });
403 if (crawler === undefined) return;
404
405 yield* retryCrawlerDelete(glue.deleteCrawler({ Name: name })).pipe(
406 Effect.catchTag("EntityNotFoundException", () => Effect.void),
407 );
408
409 const remaining = yield* Effect.repeat(observe(name), {
410 schedule: Schedule.fixed("2 seconds"),
411 until: (current) => current === undefined,
412 times: 15,
413 });
414 if (remaining !== undefined) {
415 return yield* Effect.fail(
416 new glue.OperationTimeoutException({
417 message: `crawler ${name} remained visible after delete for 30 seconds`,
418 }),
419 );
420 }
421 }),
422 });
423 }),
424 );