1import * as athena from "@distilled.cloud/aws/athena";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { isResolved } from "../../Diff.ts";
5import * as Provider from "../../Provider.ts";
6import { Resource } from "../../Resource.ts";
7import { createInternalTags, diffTags } from "../../Tags.ts";
8import { AWSEnvironment, type AccountID } from "../Environment.ts";
9import type { Providers } from "../Providers.ts";
10import type { RegionID } from "../Region.ts";
11
12export type DataCatalogArn =
13 `arn:aws:athena:${RegionID}:${AccountID}:datacatalog/${string}`;
14
15export interface DataCatalogProps {
16 /**
17 * Name of the data catalog. This is the catalog's identity — changing it
18 * replaces the resource. Up to 256 characters.
19 */
20 name: string;
21 /**
22 * Catalog type. Changing this replaces the resource.
23 * - `LAMBDA` — a federated connector backed by a Lambda metadata function.
24 * - `GLUE` — an AWS Glue Data Catalog (typically in another account).
25 * - `HIVE` — an external Apache Hive metastore via a Lambda function.
26 * - `FEDERATED` — a managed federated connector.
27 */
28 type: "LAMBDA" | "GLUE" | "HIVE" | "FEDERATED";
29 /**
30 * Optional description.
31 */
32 description?: string;
33 /**
34 * Type-specific parameters. For `LAMBDA`/`HIVE` this carries the
35 * `metadata-function` / `record-function` (or `function`) ARNs; for `GLUE`
36 * the `catalog-id`.
37 */
38 parameters?: Record<string, string>;
39 /**
40 * User-defined tags to apply to the data catalog.
41 */
42 tags?: Record<string, string>;
43}
44
45export interface DataCatalog extends Resource<
46 "AWS.Athena.DataCatalog",
47 DataCatalogProps,
48 {
49 /**
50 * Name of the data catalog.
51 */
52 name: string;
53 /**
54 * ARN of the data catalog.
55 */
56 dataCatalogArn: DataCatalogArn;
57 /**
58 * Catalog type (`LAMBDA`, `GLUE`, or `HIVE`).
59 */
60 type: string;
61 /**
62 * Type-specific connection parameters.
63 */
64 parameters: Record<string, string>;
65 /**
66 * Description of the data catalog.
67 */
68 description: string | undefined;
69 /**
70 * Tags on the data catalog.
71 */
72 tags: Record<string, string>;
73 },
74 never,
75 Providers
76> {}
77
78/**
79 * An Amazon Athena data catalog — registers an external metadata source (a
80 * federated Lambda connector, an external Hive metastore, or a cross-account
81 * Glue Data Catalog) that Athena queries can reference as a `catalog`.
82 *
83 * ### Registering Catalogs
84 * **Example:** A federated Lambda-backed catalog
85 * ```typescript
86 * const catalog = yield* AWS.Athena.DataCatalog("Cmdb", {
87 * name: "cmdb_connector",
88 * type: "LAMBDA",
89 * parameters: {
90 * "metadata-function": connector.functionArn,
91 * "record-function": connector.functionArn,
92 * },
93 * });
94 * ```
95 *
96 * @resource
97 */
98export const DataCatalog = Resource<DataCatalog>("AWS.Athena.DataCatalog");
99
100const observedTagsOf = (tags: readonly athena.Tag[] | undefined) =>
101 Object.fromEntries(
102 (tags ?? []).flatMap((t) =>
103 t.Key !== undefined && t.Value !== undefined ? [[t.Key, t.Value]] : [],
104 ),
105 );
106
107const paramsOf = (params: { [key: string]: string | undefined } | undefined) =>
108 Object.fromEntries(
109 Object.entries(params ?? {}).flatMap((e) =>
110 typeof e[1] === "string" ? [[e[0], e[1]]] : [],
111 ),
112 ) as Record<string, string>;
113
114export const DataCatalogProvider = () =>
115 Provider.effect(
116 DataCatalog,
117 Effect.gen(function* () {
118 const getOne = (name: string) =>
119 // GetDataCatalog resolves catalogs through a workgroup — without one it
120 // reports even existing catalogs as not-found, so scope to `primary`.
121 athena.getDataCatalog({ Name: name, WorkGroup: "primary" }).pipe(
122 Effect.map((res) => res.DataCatalog),
123 Effect.catchTag("DataCatalogNotFound", () =>
124 Effect.succeed(undefined),
125 ),
126 );
127
128 const fetchTags = (arn: string) =>
129 athena.listTagsForResource.items({ ResourceARN: arn }).pipe(
130 Stream.runCollect,
131 Effect.map((chunk) => observedTagsOf(Array.from(chunk))),
132 Effect.catch(() => Effect.succeed({} as Record<string, string>)),
133 );
134
135 return {
136 stables: ["name", "dataCatalogArn"],
137 diff: Effect.fn(function* ({ olds, news }) {
138 if (!isResolved(news)) return;
139 if ((olds?.name ?? undefined) !== news.name) {
140 return { action: "replace" } as const;
141 }
142 if ((olds?.type ?? undefined) !== news.type) {
143 return { action: "replace" } as const;
144 }
145 }),
146 read: Effect.fn(function* ({ olds, output }) {
147 const { accountId, region } = yield* AWSEnvironment.current;
148 const name = output?.name ?? olds?.name;
149 if (!name) return undefined;
150 const dc = yield* getOne(name);
151 if (!dc) return undefined;
152 const arn =
153 `arn:aws:athena:${region}:${accountId}:datacatalog/${name}` as DataCatalogArn;
154 return {
155 name,
156 dataCatalogArn: arn,
157 type: dc.Type,
158 parameters: paramsOf(dc.Parameters),
159 description: dc.Description,
160 tags: yield* fetchTags(arn),
161 };
162 }),
163 list: () =>
164 Effect.gen(function* () {
165 const { accountId, region } = yield* AWSEnvironment.current;
166 const pages = yield* athena.listDataCatalogs
167 .pages({})
168 .pipe(Stream.runCollect);
169 return Array.from(pages)
170 .flatMap((page) => page.DataCatalogsSummary ?? [])
171 .flatMap((dc) =>
172 dc.CatalogName
173 ? [
174 {
175 name: dc.CatalogName,
176 dataCatalogArn:
177 `arn:aws:athena:${region}:${accountId}:datacatalog/${dc.CatalogName}` as DataCatalogArn,
178 type: dc.Type ?? "LAMBDA",
179 parameters: {} as Record<string, string>,
180 description: undefined,
181 tags: {} as Record<string, string>,
182 },
183 ]
184 : [],
185 );
186 }),
187 reconcile: Effect.fn(function* ({ id, news, output, session }) {
188 const { accountId, region } = yield* AWSEnvironment.current;
189 const name = news.name ?? output?.name!;
190 const arn =
191 `arn:aws:athena:${region}:${accountId}:datacatalog/${name}` as DataCatalogArn;
192 const internalTags = yield* createInternalTags(id);
193 const desiredTags = { ...internalTags, ...news.tags };
194 const desiredParams = news.parameters ?? {};
195
196 // Observe — cloud state is authoritative.
197 let dc = yield* getOne(name);
198
199 // Ensure — create if missing.
200 if (!dc) {
201 yield* athena.createDataCatalog({
202 Name: name,
203 Type: news.type,
204 Description: news.description,
205 Parameters: desiredParams,
206 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
207 Key,
208 Value,
209 })),
210 });
211 dc = yield* getOne(name);
212 } else {
213 // Sync — AWS derives extra parameter keys (e.g. `catalog`) beyond
214 // what we sent, so only update when a DESIRED key is missing or
215 // differs from the observed value (ignore AWS-added keys).
216 const observedParams = paramsOf(dc.Parameters);
217 const paramsDrift = Object.entries(desiredParams).some(
218 ([k, v]) => observedParams[k] !== v,
219 );
220 const descDrift =
221 news.description !== undefined &&
222 dc.Description !== news.description;
223 if (paramsDrift || descDrift) {
224 yield* athena.updateDataCatalog({
225 Name: name,
226 Type: news.type,
227 Description: news.description,
228 Parameters: desiredParams,
229 });
230 }
231 }
232
233 // Sync tags — diff against OBSERVED cloud tags.
234 const observed = yield* fetchTags(arn);
235 const { upsert, removed } = diffTags(observed, desiredTags);
236 if (upsert.length > 0) {
237 yield* athena.tagResource({
238 ResourceARN: arn,
239 Tags: upsert.map((t) => ({ Key: t.Key, Value: t.Value })),
240 });
241 }
242 if (removed.length > 0) {
243 yield* athena.untagResource({ ResourceARN: arn, TagKeys: removed });
244 }
245
246 const final = (yield* getOne(name)) ?? dc;
247 yield* session.note(arn);
248 return {
249 name,
250 dataCatalogArn: arn,
251 type: final?.Type ?? news.type,
252 parameters: paramsOf(final?.Parameters),
253 description: final?.Description ?? news.description,
254 tags: desiredTags,
255 };
256 }),
257 delete: Effect.fn(function* ({ output }) {
258 yield* athena
259 .deleteDataCatalog({ Name: output.name })
260 .pipe(Effect.catchTag("DataCatalogNotFound", () => Effect.void));
261 }),
262 };
263 }),
264 );