1import * as bcm from "@distilled.cloud/aws/bcm-data-exports";
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, diffTags, hasAlchemyTags } from "../../Tags.ts";
11import type { Providers } from "../Providers.ts";
12
13/**
14 * The SQL query and table configurations for a data export.
15 */
16export interface ExportDataQuery {
17 /**
18 * The SQL statement selecting the columns to export, e.g.
19 * `SELECT identity_line_item_id, line_item_unblended_cost FROM COST_AND_USAGE_REPORT`.
20 * Data Exports supports a limited subset of SQL — see the AWS Data Exports
21 * table dictionary for available tables and columns.
22 */
23 queryStatement: string;
24 /**
25 * Per-table property overrides, e.g.
26 * `{ COST_AND_USAGE_REPORT: { TIME_GRANULARITY: "HOURLY" } }`.
27 * Every table property has a default it assumes when omitted.
28 */
29 tableConfigurations?: Record<string, Record<string, string>>;
30}
31
32/**
33 * Formatting options for the objects written to S3.
34 */
35export interface ExportS3OutputConfigurations {
36 /**
37 * The output type of the export.
38 * @default "CUSTOM"
39 */
40 outputType?: "CUSTOM" | (string & {});
41 /**
42 * The file format of the exported objects.
43 * @default "TEXT_OR_CSV"
44 */
45 format?: "TEXT_OR_CSV" | "PARQUET" | (string & {});
46 /**
47 * The compression of the exported objects. Use `PARQUET` compression with
48 * the `PARQUET` format.
49 * @default "GZIP"
50 */
51 compression?: "GZIP" | "PARQUET" | (string & {});
52 /**
53 * Whether each delivery overwrites the previous report or writes a new one.
54 * @default "OVERWRITE_REPORT"
55 */
56 overwrite?: "CREATE_NEW_REPORT" | "OVERWRITE_REPORT" | (string & {});
57}
58
59/**
60 * The S3 bucket the export is delivered to. The bucket policy must allow the
61 * `bcm-data-exports.amazonaws.com` and `billingreports.amazonaws.com` service
62 * principals to `s3:PutObject` and `s3:GetBucketPolicy`.
63 */
64export interface ExportS3Destination {
65 /**
66 * Name of the destination S3 bucket.
67 */
68 s3Bucket: string;
69 /**
70 * Key prefix the export is written under.
71 */
72 s3Prefix: string;
73 /**
74 * Region of the destination S3 bucket.
75 */
76 s3Region: string;
77 /**
78 * Account ID that owns the destination bucket, for cross-account delivery.
79 */
80 s3BucketOwner?: string;
81 /**
82 * Output formatting options.
83 * @default `{ outputType: "CUSTOM", format: "TEXT_OR_CSV", compression: "GZIP", overwrite: "OVERWRITE_REPORT" }`
84 */
85 s3OutputConfigurations?: ExportS3OutputConfigurations;
86}
87
88export interface ExportProps {
89 /**
90 * Name of the export. Must be unique within the account. If omitted, a
91 * unique name is generated from the app, stage, and logical ID.
92 *
93 * Changing the name replaces the export.
94 */
95 exportName?: string;
96 /**
97 * Description of the export.
98 */
99 description?: string;
100 /**
101 * The SQL data query — the statement plus optional table configurations.
102 */
103 dataQuery: ExportDataQuery;
104 /**
105 * The S3 destination the export is delivered to.
106 */
107 s3Destination: ExportS3Destination;
108 /**
109 * How often the export refreshes. `SYNCHRONOUS` refreshes whenever the
110 * source billing data updates.
111 * @default `{ frequency: "SYNCHRONOUS" }`
112 */
113 refreshCadence?: {
114 /**
115 * The refresh frequency.
116 * @default "SYNCHRONOUS"
117 */
118 frequency: "SYNCHRONOUS" | (string & {});
119 };
120 /**
121 * Tags applied to the export.
122 */
123 tags?: Record<string, string>;
124}
125
126export interface Export extends Resource<
127 "AWS.BCMDataExports.Export",
128 ExportProps,
129 {
130 /**
131 * Name of the export.
132 */
133 exportName: string;
134 /**
135 * The ARN of the export.
136 */
137 exportArn: string;
138 },
139 never,
140 Providers
141> {}
142
143/**
144 * An AWS Billing and Cost Management data export (Data Exports / CUR 2.0) —
145 * delivers billing and cost management data selected by an SQL query to an
146 * S3 bucket on a refresh cadence.
147 *
148 * Data Exports is a global service pinned to `us-east-1`; exports are free
149 * (standard S3 storage rates apply to the delivered objects).
150 *
151 * The destination bucket must grant the Data Exports service principals
152 * write access via its bucket policy (see the example below).
153 *
154 * ### Creating an Export
155 * **Example:** CUR 2.0 export to an S3 bucket
156 * ```typescript
157 * import * as AWS from "alchemy/AWS";
158 *
159 * const bucket = yield* AWS.S3.Bucket("BillingData", {
160 * forceDestroy: true,
161 * policy: [
162 * {
163 * Effect: "Allow",
164 * Principal: {
165 * Service: [
166 * "billingreports.amazonaws.com",
167 * "bcm-data-exports.amazonaws.com",
168 * ],
169 * },
170 * Action: ["s3:PutObject", "s3:GetBucketPolicy"],
171 * Resource: [bucket.bucketArn, AWS.interpolate`${bucket.bucketArn}/*`],
172 * },
173 * ],
174 * });
175 *
176 * const cur = yield* AWS.BCMDataExports.Export("Cur2", {
177 * dataQuery: {
178 * queryStatement:
179 * "SELECT identity_line_item_id, line_item_unblended_cost FROM COST_AND_USAGE_REPORT",
180 * tableConfigurations: {
181 * COST_AND_USAGE_REPORT: { TIME_GRANULARITY: "HOURLY" },
182 * },
183 * },
184 * s3Destination: {
185 * s3Bucket: bucket.bucketName,
186 * s3Prefix: "cur2",
187 * s3Region: "us-west-2",
188 * },
189 * });
190 * ```
191 *
192 * @resource
193 */
194export const Export = Resource<Export>("AWS.BCMDataExports.Export");
195
196const retryThrottling = <A, E extends { readonly _tag: string }, R>(
197 self: Effect.Effect<A, E, R>,
198): Effect.Effect<A, E, R> =>
199 Effect.retry(self, {
200 while: (e) => e._tag === "ThrottlingException",
201 schedule: Schedule.max([
202 Schedule.exponential("1 second"),
203 Schedule.recurs(5),
204 ]),
205 });
206
207export const ExportProvider = () =>
208 Provider.effect(
209 Export,
210 Effect.gen(function* () {
211 const createName = Effect.fn(function* (
212 id: string,
213 props: { exportName?: string | undefined },
214 ) {
215 return (
216 props.exportName ??
217 (yield* createPhysicalName({ id, maxLength: 100 }))
218 );
219 });
220
221 const buildExport = (name: string, props: ExportProps): bcm.Export => ({
222 Name: name,
223 Description: props.description,
224 DataQuery: {
225 QueryStatement: props.dataQuery.queryStatement,
226 TableConfigurations: props.dataQuery.tableConfigurations,
227 },
228 DestinationConfigurations: {
229 S3Destination: {
230 S3Bucket: props.s3Destination.s3Bucket,
231 S3Prefix: props.s3Destination.s3Prefix,
232 S3Region: props.s3Destination.s3Region,
233 S3BucketOwner: props.s3Destination.s3BucketOwner,
234 S3OutputConfigurations: {
235 OutputType:
236 props.s3Destination.s3OutputConfigurations?.outputType ??
237 "CUSTOM",
238 Format:
239 props.s3Destination.s3OutputConfigurations?.format ??
240 "TEXT_OR_CSV",
241 Compression:
242 props.s3Destination.s3OutputConfigurations?.compression ??
243 "GZIP",
244 Overwrite:
245 props.s3Destination.s3OutputConfigurations?.overwrite ??
246 "OVERWRITE_REPORT",
247 },
248 },
249 },
250 RefreshCadence: {
251 Frequency: props.refreshCadence?.frequency ?? "SYNCHRONOUS",
252 },
253 });
254
255 // Project both the observed and the desired definition through the
256 // same shape (fixed key order, defaults filled) so wire echoes — key
257 // ordering, an echoed S3BucketOwner — never register as drift.
258 const normalizeExport = (e: bcm.Export): string => {
259 const dest = e.DestinationConfigurations.S3Destination;
260 return JSON.stringify({
261 Name: e.Name,
262 Description: e.Description ?? "",
263 QueryStatement: e.DataQuery.QueryStatement,
264 TableConfigurations: Object.fromEntries(
265 Object.entries(e.DataQuery.TableConfigurations ?? {})
266 .sort(([l], [r]) => l.localeCompare(r))
267 .map(([table, props]) => [
268 table,
269 Object.fromEntries(
270 Object.entries(props ?? {}).sort(([l], [r]) =>
271 l.localeCompare(r),
272 ),
273 ),
274 ]),
275 ),
276 S3Bucket: dest.S3Bucket,
277 S3Prefix: dest.S3Prefix,
278 S3Region: dest.S3Region,
279 OutputType: dest.S3OutputConfigurations.OutputType,
280 Format: dest.S3OutputConfigurations.Format,
281 Compression: dest.S3OutputConfigurations.Compression,
282 Overwrite: dest.S3OutputConfigurations.Overwrite,
283 Frequency: e.RefreshCadence.Frequency,
284 });
285 };
286 // S3BucketOwner is compared only when the user sets it — AWS may echo
287 // the effective owner account on reads even when it was never supplied.
288 const sameExport = (live: bcm.Export, desired: bcm.Export): boolean =>
289 normalizeExport(live) === normalizeExport(desired) &&
290 (desired.DestinationConfigurations.S3Destination.S3BucketOwner ===
291 undefined ||
292 desired.DestinationConfigurations.S3Destination.S3BucketOwner ===
293 live.DestinationConfigurations.S3Destination.S3BucketOwner);
294
295 const findArnByName = Effect.fn(function* (name: string) {
296 const matches = yield* bcm.listExports.items({}).pipe(
297 Stream.filter((e) => e.ExportName === name),
298 Stream.take(1),
299 Stream.runCollect,
300 );
301 return Array.from(matches)[0]?.ExportArn;
302 });
303
304 // Observe the live export definition. `arnHint` (from cached output)
305 // is verified, not trusted — a stale ARN falls through to name lookup.
306 const observe = Effect.fn(function* (
307 name: string,
308 arnHint: string | undefined,
309 ) {
310 const arns = arnHint !== undefined ? [arnHint] : [];
311 const byName = yield* findArnByName(name);
312 if (byName !== undefined && byName !== arnHint) arns.push(byName);
313 for (const arn of arns) {
314 const found = yield* bcm
315 .getExport({ ExportArn: arn })
316 .pipe(
317 Effect.catchTag("ResourceNotFoundException", () =>
318 Effect.succeed(undefined),
319 ),
320 );
321 if (found?.Export !== undefined) {
322 return { exportArn: arn, export: found.Export };
323 }
324 }
325 return undefined;
326 });
327
328 const fetchObservedTags = Effect.fn(function* (arn: string) {
329 return yield* bcm.listTagsForResource({ ResourceArn: arn }).pipe(
330 Effect.map((r) =>
331 Object.fromEntries(
332 (r.ResourceTags ?? []).map((t) => [t.Key, t.Value]),
333 ),
334 ),
335 Effect.catchTag("ResourceNotFoundException", () =>
336 Effect.succeed({} as Record<string, string>),
337 ),
338 );
339 });
340
341 return Export.Provider.of({
342 stables: ["exportName", "exportArn"],
343 list: () =>
344 bcm.listExports.items({}).pipe(
345 Stream.runCollect,
346 Effect.map((chunk) =>
347 Array.from(chunk).map((e) => ({
348 exportName: e.ExportName,
349 exportArn: e.ExportArn,
350 })),
351 ),
352 ),
353 read: Effect.fn(function* ({ id, olds, output }) {
354 const name =
355 output?.exportName ?? (yield* createName(id, olds ?? {}));
356 const live = yield* observe(name, output?.exportArn);
357 if (live === undefined) return undefined;
358 const attrs = { exportName: name, exportArn: live.exportArn };
359 const tags = yield* fetchObservedTags(live.exportArn);
360 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
361 }),
362 // Replacement detection only — the export name is immutable; every
363 // other property updates in place via updateExport.
364 diff: Effect.fn(function* ({ id, news, olds }) {
365 if (!isResolved(news)) return undefined;
366 const oldName = yield* createName(id, olds);
367 const newName = yield* createName(id, news);
368 if (oldName !== newName) return { action: "replace" } as const;
369 }),
370 reconcile: Effect.fn(function* ({ id, news, output, session }) {
371 const name = output?.exportName ?? (yield* createName(id, news));
372 const internalTags = yield* createInternalTags(id);
373 const desiredTags = { ...news.tags, ...internalTags };
374 const desired = buildExport(name, news);
375
376 // 1. OBSERVE — cloud state is authoritative; output only caches
377 // the ARN.
378 const live = yield* observe(name, output?.exportArn);
379
380 let exportArn: string;
381 if (live === undefined) {
382 // 2. ENSURE — create the export with all tags attached.
383 yield* session.note(`creating data export ${name}`);
384 const created = yield* retryThrottling(
385 bcm.createExport({
386 Export: desired,
387 ResourceTags: Object.entries(desiredTags).map(
388 ([Key, Value]) => ({ Key, Value }),
389 ),
390 }),
391 );
392 exportArn =
393 created.ExportArn ?? (yield* findArnByName(name)) ?? name;
394 } else {
395 exportArn = live.exportArn;
396 // 3. SYNC — updateExport overwrites the full definition; skip
397 // the call when the observed definition already matches.
398 if (!sameExport(live.export, desired)) {
399 yield* session.note(`updating data export ${name}`);
400 yield* retryThrottling(
401 bcm.updateExport({ ExportArn: exportArn, Export: desired }),
402 );
403 }
404 }
405
406 // 3b. SYNC TAGS — diff against observed cloud tags.
407 const observedTags = yield* fetchObservedTags(exportArn);
408 const { upsert, removed } = diffTags(observedTags, desiredTags);
409 if (upsert.length > 0) {
410 yield* bcm.tagResource({
411 ResourceArn: exportArn,
412 ResourceTags: upsert,
413 });
414 }
415 if (removed.length > 0) {
416 yield* bcm.untagResource({
417 ResourceArn: exportArn,
418 ResourceTagKeys: removed,
419 });
420 }
421
422 yield* session.note(name);
423 return { exportName: name, exportArn };
424 }),
425 delete: Effect.fn(function* ({ output }) {
426 yield* bcm
427 .deleteExport({ ExportArn: output.exportArn })
428 .pipe(
429 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
430 );
431 }),
432 });
433 }),
434 );