1import * as Glue from "@/AWS/Glue";
2import { Role } from "@/AWS/IAM";
3import * as Lambda from "@/AWS/Lambda";
4import { Bucket } from "@/AWS/S3";
5import * as Output from "@/Output";
6import * as Duration from "effect/Duration";
7import * as Effect from "effect/Effect";
8import * as Layer from "effect/Layer";
9import * as Schedule from "effect/Schedule";
10import * as Stream from "effect/Stream";
11import { HttpServerRequest } from "effect/http/HttpServerRequest";
12import * as HttpServerResponse from "effect/http/HttpServerResponse";
13import path from "pathe";
14
15const main = path.resolve(import.meta.dirname, "handler.ts");
16
17export class GlueTestFunction extends Lambda.Function<Lambda.Function>()(
18 "GlueTestFunction",
19) {}
20
21export default GlueTestFunction.make(
22 {
23 main,
24 functionUrl: true,
25 timeout: Duration.seconds(90),
26 },
27 Effect.gen(function* () {
28 const bucket = yield* Bucket("GlueBindingsBucket", { forceDestroy: true });
29
30 const role = yield* Role("GlueBindingsRole", {
31 assumeRolePolicyDocument: {
32 Version: "2012-10-17",
33 Statement: [
34 {
35 Effect: "Allow",
36 Principal: { Service: "glue.amazonaws.com" },
37 Action: ["sts:AssumeRole"],
38 },
39 ],
40 },
41 managedPolicyArns: [
42 "arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole",
43 "arn:aws:iam::aws:policy/AmazonS3FullAccess",
44 ],
45 });
46
47 const database = yield* Glue.Database("BindingsDatabase", {});
48
49 const table = yield* Glue.Table("BindingsEvents", {
50 databaseName: database.databaseName,
51 storageDescriptor: {
52 location: Output.interpolate`s3://${bucket.bucketName}/events/`,
53 columns: [
54 { name: "id", type: "string" },
55 { name: "amount", type: "double" },
56 ],
57 },
58 partitionKeys: [{ name: "dt", type: "string" }],
59 parameters: { classification: "json" },
60 });
61
62 const job = yield* Glue.Job("BindingsJob", {
63 role: role.roleArn,
64 command: {
65 name: "pythonshell",
66 pythonVersion: "3.9",
67 scriptLocation: Output.interpolate`s3://${bucket.bucketName}/scripts/noop.py`,
68 },
69 maxCapacity: 0.0625,
70 glueVersion: "3.0",
71 timeout: "10 minutes",
72 });
73
74 const crawler = yield* Glue.Crawler("BindingsCrawler", {
75 role: role.roleArn,
76 databaseName: database.databaseName,
77 targets: {
78 s3Targets: [
79 { path: Output.interpolate`s3://${bucket.bucketName}/crawl/` },
80 ],
81 },
82 });
83
84 // Event sources: the deploys prove the EventBridge rule + invoke
85 // permission wiring for both Glue detail types.
86 yield* Glue.consumeJobEvents({ states: ["FAILED", "TIMEOUT"] }, (events) =>
87 Stream.runForEach(events, (event) =>
88 Effect.log(`glue job event: ${event.detail.jobName}`),
89 ),
90 );
91 yield* Glue.consumeCrawlerEvents({ states: ["Succeeded"] }, (events) =>
92 Stream.runForEach(events, (event) =>
93 Effect.log(`glue crawler event: ${event.detail.crawlerName}`),
94 ),
95 );
96
97 // Job-scoped bindings.
98 const startJobRun = yield* Glue.StartJobRun(job);
99 const getJobRun = yield* Glue.GetJobRun(job);
100 const getJobRuns = yield* Glue.GetJobRuns(job);
101 const batchStopJobRun = yield* Glue.BatchStopJobRun(job);
102 const getJobBookmark = yield* Glue.GetJobBookmark(job);
103 const resetJobBookmark = yield* Glue.ResetJobBookmark(job);
104
105 // Crawler-scoped bindings.
106 const startCrawler = yield* Glue.StartCrawler(crawler);
107 const stopCrawler = yield* Glue.StopCrawler(crawler);
108 const getCrawler = yield* Glue.GetCrawler(crawler);
109
110 // Catalog bindings.
111 const getTables = yield* Glue.GetTables(database);
112 const getTable = yield* Glue.GetTable(table);
113 const getPartitions = yield* Glue.GetPartitions(table);
114 const getPartition = yield* Glue.GetPartition(table);
115 const createPartition = yield* Glue.CreatePartition(table);
116 const batchCreatePartition = yield* Glue.BatchCreatePartition(table);
117 const batchGetPartition = yield* Glue.BatchGetPartition(table);
118 const updatePartition = yield* Glue.UpdatePartition(table);
119 const batchUpdatePartition = yield* Glue.BatchUpdatePartition(table);
120 const deletePartition = yield* Glue.DeletePartition(table);
121 const batchDeletePartition = yield* Glue.BatchDeletePartition(table);
122
123 const bound = {
124 startJobRun,
125 getJobRun,
126 getJobRuns,
127 batchStopJobRun,
128 getJobBookmark,
129 resetJobBookmark,
130 startCrawler,
131 stopCrawler,
132 getCrawler,
133 getTables,
134 getTable,
135 getPartitions,
136 getPartition,
137 createPartition,
138 batchCreatePartition,
139 batchGetPartition,
140 updatePartition,
141 batchUpdatePartition,
142 deletePartition,
143 batchDeletePartition,
144 };
145
146 return {
147 fetch: Effect.gen(function* () {
148 const request = yield* HttpServerRequest;
149 const url = new URL(request.originalUrl);
150 const pathname = url.pathname;
151
152 if (request.method === "GET" && pathname === "/bindings") {
153 return yield* HttpServerResponse.json({ bound: Object.keys(bound) });
154 }
155
156 // List the job's runs (empty on a fresh job) — proves GetJobRuns.
157 if (request.method === "GET" && pathname === "/job-runs") {
158 const { JobRuns } = yield* getJobRuns({ MaxResults: 10 });
159 return yield* HttpServerResponse.json({
160 count: (JobRuns ?? []).length,
161 });
162 }
163
164 // Bookmarks: a job that has never run with bookmarks enabled has no
165 // entry — both operations surface the typed EntityNotFoundException.
166 if (request.method === "GET" && pathname === "/job-bookmark") {
167 const bookmark = yield* getJobBookmark().pipe(
168 Effect.map((r) => r.JobBookmarkEntry ?? "none"),
169 Effect.catchTag("EntityNotFoundException", () =>
170 Effect.succeed("none" as const),
171 ),
172 );
173 const reset = yield* resetJobBookmark().pipe(
174 Effect.map(() => "reset" as const),
175 Effect.catchTag("EntityNotFoundException", () =>
176 Effect.succeed("none" as const),
177 ),
178 );
179 return yield* HttpServerResponse.json({ bookmark, reset });
180 }
181
182 // Start a real (cheap pythonshell) run, read it back, then stop it —
183 // proves StartJobRun + GetJobRun + BatchStopJobRun end to end.
184 if (request.method === "POST" && pathname === "/job-run") {
185 const { JobRunId } = yield* startJobRun({}).pipe(
186 Effect.retry({
187 while: (e): boolean => e._tag === "GlueRoleNotAssumable",
188 schedule: Schedule.max([
189 Schedule.fixed("5 seconds"),
190 Schedule.recurs(6),
191 ]),
192 }),
193 );
194 const { JobRun } = yield* getJobRun({ RunId: JobRunId! });
195 const stopped = yield* batchStopJobRun({ JobRunIds: [JobRunId!] });
196 return yield* HttpServerResponse.json({
197 runId: JobRunId,
198 state: JobRun?.JobRunState,
199 stopSubmitted: (stopped.SuccessfulSubmissions ?? []).length,
200 stopErrors: (stopped.Errors ?? []).length,
201 });
202 }
203
204 // GetCrawler reads the crawler's live state — proves the op + grant.
205 if (request.method === "GET" && pathname === "/crawler") {
206 const { Crawler } = yield* getCrawler();
207 return yield* HttpServerResponse.json({
208 name: Crawler?.Name,
209 state: Crawler?.State,
210 });
211 }
212
213 // StopCrawler on an idle crawler surfaces the typed
214 // CrawlerNotRunningException — proves the op + IAM grant cheaply.
215 if (request.method === "POST" && pathname === "/crawler-stop") {
216 const result = yield* stopCrawler().pipe(
217 Effect.map(() => "stopped" as const),
218 Effect.catchTag("CrawlerNotRunningException", () =>
219 Effect.succeed("not-running" as const),
220 ),
221 Effect.catchTag("CrawlerStoppingException", () =>
222 Effect.succeed("stopping" as const),
223 ),
224 );
225 return yield* HttpServerResponse.json({ result });
226 }
227
228 // Start a crawl (empty S3 prefix) and immediately cancel it —
229 // proves StartCrawler + StopCrawler live.
230 if (request.method === "POST" && pathname === "/crawler-run") {
231 const started = yield* startCrawler().pipe(
232 Effect.map(() => "started" as const),
233 Effect.catchTag("CrawlerRunningException", () =>
234 Effect.succeed("already-running" as const),
235 ),
236 );
237 const stopped = yield* stopCrawler().pipe(
238 Effect.map(() => "stopped" as const),
239 Effect.catchTag("CrawlerNotRunningException", () =>
240 Effect.succeed("not-running" as const),
241 ),
242 Effect.catchTag("CrawlerStoppingException", () =>
243 Effect.succeed("stopping" as const),
244 ),
245 );
246 return yield* HttpServerResponse.json({ started, stopped });
247 }
248
249 // Catalog reads — proves GetTable + GetTables.
250 if (request.method === "GET" && pathname === "/table") {
251 const { Table } = yield* getTable();
252 const { TableList } = yield* getTables();
253 return yield* HttpServerResponse.json({
254 name: Table?.Name,
255 columns: (Table?.StorageDescriptor?.Columns ?? []).map(
256 (c) => c.Name,
257 ),
258 partitionKeys: (Table?.PartitionKeys ?? []).map((k) => k.Name),
259 tables: (TableList ?? []).map((t) => t.Name),
260 });
261 }
262
263 // Full partition lifecycle — proves CreatePartition, GetPartition,
264 // UpdatePartition, BatchCreatePartition, GetPartitions,
265 // DeletePartition.
266 if (request.method === "POST" && pathname === "/partitions") {
267 yield* createPartition({
268 PartitionInput: { Values: ["2026-01-01"] },
269 }).pipe(
270 Effect.catchTag("AlreadyExistsException", () => Effect.succeed({})),
271 );
272 const created = yield* getPartition({
273 PartitionValues: ["2026-01-01"],
274 });
275 yield* updatePartition({
276 PartitionValueList: ["2026-01-01"],
277 PartitionInput: {
278 Values: ["2026-01-01"],
279 Parameters: { updated: "true" },
280 },
281 });
282 const updated = yield* getPartition({
283 PartitionValues: ["2026-01-01"],
284 });
285 const batch = yield* batchCreatePartition({
286 PartitionInputList: [
287 { Values: ["2026-01-02"] },
288 { Values: ["2026-01-03"] },
289 ],
290 });
291 // Bulk update the batch-created partitions' parameters.
292 const batchUpdated = yield* batchUpdatePartition({
293 Entries: ["2026-01-02", "2026-01-03"].map((dt) => ({
294 PartitionValueList: [dt],
295 PartitionInput: {
296 Values: [dt],
297 Parameters: { bulkUpdated: "true" },
298 },
299 })),
300 });
301 // Bulk read all three back (one miss to prove UnprocessedKeys).
302 const bulk = yield* batchGetPartition({
303 PartitionsToGet: [
304 { Values: ["2026-01-01"] },
305 { Values: ["2026-01-02"] },
306 { Values: ["2026-01-03"] },
307 ],
308 });
309 const all = yield* getPartitions();
310 // Bulk delete two, single-delete the third.
311 const batchDeleted = yield* batchDeletePartition({
312 PartitionsToDelete: [
313 { Values: ["2026-01-02"] },
314 { Values: ["2026-01-03"] },
315 ],
316 });
317 yield* deletePartition({ PartitionValues: ["2026-01-01"] }).pipe(
318 Effect.catchTag("EntityNotFoundException", () => Effect.void),
319 );
320 const after = yield* getPartitions();
321 return yield* HttpServerResponse.json({
322 created: created.Partition?.Values,
323 updatedParam: updated.Partition?.Parameters?.updated,
324 batchErrors: (batch.Errors ?? []).length,
325 batchUpdateErrors: (batchUpdated.Errors ?? []).length,
326 bulkRead: (bulk.Partitions ?? []).length,
327 bulkUpdatedParams: (bulk.Partitions ?? []).filter(
328 (p) => p.Parameters?.bulkUpdated === "true",
329 ).length,
330 count: (all.Partitions ?? []).length,
331 batchDeleteErrors: (batchDeleted.Errors ?? []).length,
332 remaining: (after.Partitions ?? []).length,
333 });
334 }
335
336 return yield* HttpServerResponse.json(
337 { error: "Not found", method: request.method, pathname },
338 { status: 404 },
339 );
340 }).pipe(Effect.orDie),
341 };
342 }).pipe(
343 Effect.provide(
344 Layer.mergeAll(
345 Lambda.EventSource,
346 Glue.StartJobRunHttp,
347 Glue.GetJobRunHttp,
348 Glue.GetJobRunsHttp,
349 Glue.BatchStopJobRunHttp,
350 Glue.GetJobBookmarkHttp,
351 Glue.ResetJobBookmarkHttp,
352 Glue.StartCrawlerHttp,
353 Glue.StopCrawlerHttp,
354 Glue.GetCrawlerHttp,
355 Glue.GetTablesHttp,
356 Glue.GetTableHttp,
357 Glue.GetPartitionsHttp,
358 Glue.GetPartitionHttp,
359 Glue.CreatePartitionHttp,
360 Glue.BatchCreatePartitionHttp,
361 Glue.BatchGetPartitionHttp,
362 Glue.UpdatePartitionHttp,
363 Glue.BatchUpdatePartitionHttp,
364 Glue.DeletePartitionHttp,
365 Glue.BatchDeletePartitionHttp,
366 ),
367 ),
368 ),
369);