1import * as databrew from "@distilled.cloud/aws/databrew";
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, hasAlchemyTags } from "../../Tags.ts";
10import { AWSEnvironment } from "../Environment.ts";
11import type { Providers } from "../Providers.ts";
12import {
13 databrewArn,
14 fetchObservedTags,
15 retryWhileConflict,
16 syncTags,
17} from "./internal.ts";
18
19/** An Amazon S3 location (bucket + optional key prefix). */
20export interface S3Location {
21 /** The S3 bucket name. */
22 bucket: string;
23 /** The object key (or key prefix) within the bucket. */
24 key?: string;
25 /** The account ID of the bucket owner (for cross-account buckets). */
26 bucketOwner?: string;
27}
28
29/** Where DataBrew reads the dataset's data from. Exactly one definition. */
30export interface DatasetInput {
31 /** Read directly from Amazon S3. */
32 s3InputDefinition?: S3Location;
33 /** Read from an AWS Glue Data Catalog table. */
34 dataCatalogInputDefinition?: {
35 /** The Data Catalog ID (defaults to the caller's account). */
36 catalogId?: string;
37 /** The Glue database name. */
38 databaseName: string;
39 /** The Glue table name. */
40 tableName: string;
41 /** S3 temp directory for intermediate results. */
42 tempDirectory?: S3Location;
43 };
44 /** Read from a JDBC database through a Glue connection. */
45 databaseInputDefinition?: {
46 /** The Glue connection name. */
47 glueConnectionName: string;
48 /** The database table to read. Mutually exclusive with `queryString`. */
49 databaseTableName?: string;
50 /** S3 temp directory for intermediate results. */
51 tempDirectory?: S3Location;
52 /** A custom SQL query to select the data. */
53 queryString?: string;
54 };
55}
56
57/** Format-specific parsing options. */
58export interface DatasetFormatOptions {
59 /** JSON parsing options. */
60 json?: {
61 /** Whether the file contains multi-line JSON records. */
62 multiLine?: boolean;
63 };
64 /** Excel parsing options. */
65 excel?: {
66 /** Sheet names to include (max 1). */
67 sheetNames?: string[];
68 /** Zero-based sheet indexes to include (max 1). */
69 sheetIndexes?: number[];
70 /** Whether the first row is a header. @default true */
71 headerRow?: boolean;
72 };
73 /** CSV parsing options. */
74 csv?: {
75 /** The single-character field delimiter. @default "," */
76 delimiter?: string;
77 /** Whether the first row is a header. @default true */
78 headerRow?: boolean;
79 };
80}
81
82/** A filter expression with substitution variables, e.g. `relative_before :dateParam`. */
83export interface FilterExpression {
84 /** The expression, e.g. `ends_with :suffix`. */
85 expression: string;
86 /** Variable substitutions keyed by `:name`. */
87 valuesMap: Record<string, string>;
88}
89
90/** Options that shape how S3 path parameters select matching files. */
91export interface DatasetPathOptions {
92 /** Only include files last modified within this condition. */
93 lastModifiedDateCondition?: FilterExpression;
94 /** Cap the number of matched files. */
95 filesLimit?: {
96 /** Maximum number of files to include. */
97 maxFiles: number;
98 /** Ordering attribute. @default "LAST_MODIFIED_DATE" */
99 orderedBy?: "LAST_MODIFIED_DATE" | (string & {});
100 /** Sort direction. @default "DESCENDING" */
101 order?: "DESCENDING" | "ASCENDING" | (string & {});
102 };
103 /** Path parameter definitions keyed by parameter name. */
104 parameters?: Record<
105 string,
106 {
107 /** The parameter name (must match the key). */
108 name: string;
109 /** The parameter type. */
110 type: "Datetime" | "Number" | "String" | (string & {});
111 /** Datetime parsing options (required for `Datetime`). */
112 datetimeOptions?: {
113 /** The datetime format, e.g. `yyyy-MM-dd`. */
114 format: string;
115 /** Timezone offset, e.g. `Z` or `+02:00`. */
116 timezoneOffset?: string;
117 /** Locale code for month/day names. */
118 localeCode?: string;
119 };
120 /** Whether to add a column holding the parameter value. */
121 createColumn?: boolean;
122 /** Filter which parameter values match. */
123 filter?: FilterExpression;
124 }
125 >;
126}
127
128export interface DatasetProps {
129 /**
130 * Name of the dataset. If omitted, a unique name is generated. Changing
131 * the name replaces the dataset.
132 * @default a generated physical name
133 */
134 datasetName?: string;
135 /**
136 * The file format of the source data: `CSV`, `JSON`, `PARQUET`, `EXCEL`,
137 * or `ORC`.
138 */
139 format?: "CSV" | "JSON" | "PARQUET" | "EXCEL" | "ORC" | (string & {});
140 /**
141 * Format-specific parsing options (CSV delimiter, JSON multi-line, Excel
142 * sheets).
143 */
144 formatOptions?: DatasetFormatOptions;
145 /**
146 * Where the data lives — S3, a Glue Data Catalog table, or a JDBC
147 * database.
148 */
149 input: DatasetInput;
150 /**
151 * Path options for dynamic S3 datasets (parameterized paths, file limits).
152 */
153 pathOptions?: DatasetPathOptions;
154 /**
155 * Tags to apply to the dataset. Merged with internal Alchemy tags.
156 */
157 tags?: Record<string, string>;
158}
159
160export interface Dataset extends Resource<
161 "AWS.DataBrew.Dataset",
162 DatasetProps,
163 {
164 /** Name of the dataset. */
165 datasetName: string;
166 /** ARN of the dataset. */
167 datasetArn: string;
168 },
169 {},
170 Providers
171> {}
172
173/**
174 * An AWS Glue DataBrew dataset — a pointer to source data (S3 file/prefix,
175 * Glue Data Catalog table, or JDBC query) plus parsing options. The dataset
176 * definition itself stores no data and is free; it is consumed by DataBrew
177 * projects and jobs.
178 * ### Creating Datasets
179 * **Example:** CSV Dataset from S3
180 * ```typescript
181 * import * as AWS from "alchemy/AWS";
182 *
183 * const dataset = yield* AWS.DataBrew.Dataset("Sales", {
184 * format: "CSV",
185 * formatOptions: { csv: { delimiter: ",", headerRow: true } },
186 * input: {
187 * s3InputDefinition: {
188 * bucket: bucket.bucketName,
189 * key: "raw/sales.csv",
190 * },
191 * },
192 * });
193 * ```
194 *
195 * **Example:** JSON Dataset
196 * ```typescript
197 * const dataset = yield* AWS.DataBrew.Dataset("Events", {
198 * format: "JSON",
199 * formatOptions: { json: { multiLine: false } },
200 * input: {
201 * s3InputDefinition: { bucket: bucket.bucketName, key: "events/" },
202 * },
203 * });
204 * ```
205 *
206 * ### Glue Data Catalog
207 * **Example:** Dataset from a Catalog Table
208 * ```typescript
209 * const dataset = yield* AWS.DataBrew.Dataset("Curated", {
210 * input: {
211 * dataCatalogInputDefinition: {
212 * databaseName: glueDatabase.databaseName,
213 * tableName: "curated_sales",
214 * },
215 * },
216 * });
217 * ```
218 *
219 * @resource
220 */
221export const Dataset = Resource<Dataset>("AWS.DataBrew.Dataset");
222
223export const buildS3Location = (location: S3Location) => ({
224 Bucket: location.bucket,
225 Key: location.key,
226 BucketOwner: location.bucketOwner,
227});
228
229const buildFilterExpression = (filter: FilterExpression) => ({
230 Expression: filter.expression,
231 ValuesMap: filter.valuesMap,
232});
233
234const buildInput = (input: DatasetInput) => ({
235 S3InputDefinition: input.s3InputDefinition
236 ? buildS3Location(input.s3InputDefinition)
237 : undefined,
238 DataCatalogInputDefinition: input.dataCatalogInputDefinition
239 ? {
240 CatalogId: input.dataCatalogInputDefinition.catalogId,
241 DatabaseName: input.dataCatalogInputDefinition.databaseName,
242 TableName: input.dataCatalogInputDefinition.tableName,
243 TempDirectory: input.dataCatalogInputDefinition.tempDirectory
244 ? buildS3Location(input.dataCatalogInputDefinition.tempDirectory)
245 : undefined,
246 }
247 : undefined,
248 DatabaseInputDefinition: input.databaseInputDefinition
249 ? {
250 GlueConnectionName: input.databaseInputDefinition.glueConnectionName,
251 DatabaseTableName: input.databaseInputDefinition.databaseTableName,
252 TempDirectory: input.databaseInputDefinition.tempDirectory
253 ? buildS3Location(input.databaseInputDefinition.tempDirectory)
254 : undefined,
255 QueryString: input.databaseInputDefinition.queryString,
256 }
257 : undefined,
258});
259
260const buildFormatOptions = (options: DatasetFormatOptions | undefined) =>
261 options
262 ? {
263 Json: options.json ? { MultiLine: options.json.multiLine } : undefined,
264 Excel: options.excel
265 ? {
266 SheetNames: options.excel.sheetNames,
267 SheetIndexes: options.excel.sheetIndexes,
268 HeaderRow: options.excel.headerRow,
269 }
270 : undefined,
271 Csv: options.csv
272 ? {
273 Delimiter: options.csv.delimiter,
274 HeaderRow: options.csv.headerRow,
275 }
276 : undefined,
277 }
278 : undefined;
279
280const buildPathOptions = (options: DatasetPathOptions | undefined) =>
281 options
282 ? {
283 LastModifiedDateCondition: options.lastModifiedDateCondition
284 ? buildFilterExpression(options.lastModifiedDateCondition)
285 : undefined,
286 FilesLimit: options.filesLimit
287 ? {
288 MaxFiles: options.filesLimit.maxFiles,
289 OrderedBy: options.filesLimit.orderedBy,
290 Order: options.filesLimit.order,
291 }
292 : undefined,
293 Parameters: options.parameters
294 ? Object.fromEntries(
295 Object.entries(options.parameters).map(([key, param]) => [
296 key,
297 {
298 Name: param.name,
299 Type: param.type,
300 DatetimeOptions: param.datetimeOptions
301 ? {
302 Format: param.datetimeOptions.format,
303 TimezoneOffset: param.datetimeOptions.timezoneOffset,
304 LocaleCode: param.datetimeOptions.localeCode,
305 }
306 : undefined,
307 CreateColumn: param.createColumn,
308 Filter: param.filter
309 ? buildFilterExpression(param.filter)
310 : undefined,
311 },
312 ]),
313 )
314 : undefined,
315 }
316 : undefined;
317
318export const DatasetProvider = () =>
319 Provider.effect(
320 Dataset,
321 Effect.gen(function* () {
322 const createName = Effect.fn(function* (
323 id: string,
324 props: { datasetName?: string | undefined },
325 ) {
326 return (
327 props.datasetName ??
328 (yield* createPhysicalName({ id, maxLength: 255 }))
329 );
330 });
331
332 const observe = Effect.fn(function* (name: string) {
333 return yield* databrew
334 .describeDataset({ Name: name })
335 .pipe(
336 Effect.catchTag("ResourceNotFoundException", () =>
337 Effect.succeed(undefined),
338 ),
339 );
340 });
341
342 const buildDefinition = (props: DatasetProps) => ({
343 Format: props.format,
344 FormatOptions: buildFormatOptions(props.formatOptions),
345 Input: buildInput(props.input),
346 PathOptions: buildPathOptions(props.pathOptions),
347 });
348
349 return Dataset.Provider.of({
350 stables: ["datasetName", "datasetArn"],
351
352 list: () =>
353 Effect.gen(function* () {
354 const { accountId, region } = yield* AWSEnvironment.current;
355 const pages = yield* databrew.listDatasets
356 .pages({})
357 .pipe(Stream.runCollect);
358 return Array.from(pages)
359 .flatMap((page) => page.Datasets ?? [])
360 .map((d) => ({
361 datasetName: d.Name,
362 datasetArn:
363 d.ResourceArn ??
364 databrewArn(region, accountId, "dataset", d.Name),
365 }));
366 }),
367
368 read: Effect.fn(function* ({ id, olds, output }) {
369 const { accountId, region } = yield* AWSEnvironment.current;
370 const name =
371 output?.datasetName ?? (yield* createName(id, olds ?? {}));
372 const dataset = yield* observe(name);
373 if (dataset === undefined) return undefined;
374 const arn =
375 dataset.ResourceArn ??
376 databrewArn(region, accountId, "dataset", name);
377 const attrs = { datasetName: name, datasetArn: arn };
378 const tags = yield* fetchObservedTags(arn);
379 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
380 }),
381
382 diff: Effect.fn(function* ({ id, news, olds }) {
383 if (!isResolved(news)) return undefined;
384 const oldName = yield* createName(id, olds);
385 const newName = yield* createName(id, news);
386 if (oldName !== newName) return { action: "replace" } as const;
387 // everything else is UpdateDataset-able
388 }),
389
390 reconcile: Effect.fn(function* ({ id, news, output, session }) {
391 const { accountId, region } = yield* AWSEnvironment.current;
392 const name = output?.datasetName ?? (yield* createName(id, news));
393 const internalTags = yield* createInternalTags(id);
394 const desiredTags = { ...news.tags, ...internalTags };
395
396 // 1. OBSERVE
397 const dataset = yield* observe(name);
398
399 // 2. ENSURE / 3. SYNC — UpdateDataset is a full PUT of the definition
400 if (dataset === undefined) {
401 yield* databrew
402 .createDataset({
403 Name: name,
404 ...buildDefinition(news),
405 Tags: desiredTags,
406 })
407 .pipe(
408 // creation race — another reconcile won; fall through to sync
409 Effect.catchTag("ConflictException", () => Effect.void),
410 );
411 } else {
412 yield* databrew.updateDataset({
413 Name: name,
414 ...buildDefinition(news),
415 });
416 }
417
418 const arn =
419 dataset?.ResourceArn ??
420 databrewArn(region, accountId, "dataset", name);
421
422 // 3b. SYNC TAGS against observed cloud tags
423 const observedTags = yield* fetchObservedTags(arn);
424 yield* syncTags(arn, observedTags, desiredTags);
425
426 yield* session.note(name);
427 return { datasetName: name, datasetArn: arn };
428 }),
429
430 delete: Effect.fn(function* ({ output }) {
431 // A ConflictException surfaces briefly after an associated project
432 // or job is deleted (eventual consistency) — retry bounded.
433 yield* retryWhileConflict(
434 databrew.deleteDataset({ Name: output.datasetName }),
435 ).pipe(
436 Effect.catchTag("ResourceNotFoundException", () =>
437 Effect.succeed(undefined),
438 ),
439 );
440 }),
441 });
442 }),
443 );