1import * as Deadline from "@/AWS/Deadline";
2import * as Lambda from "@/AWS/Lambda";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Stream from "effect/Stream";
7import { HttpServerRequest } from "effect/http/HttpServerRequest";
8import * as HttpServerResponse from "effect/http/HttpServerResponse";
9import path from "pathe";
10
11const main = path.resolve(import.meta.dirname, "handler.ts");
12
13/**
14 * Minimal valid Open Job Description template. With no fleet associated to
15 * the queue the job simply sits READY — nothing ever runs (or costs money).
16 */
17const JOB_TEMPLATE = JSON.stringify({
18 specificationVersion: "jobtemplate-2023-09",
19 name: "AlchemyDeadlineBindingsJob",
20 steps: [
21 {
22 name: "Echo",
23 script: {
24 actions: { onRun: { command: "/bin/echo", args: ["hello"] } },
25 },
26 },
27 ],
28});
29
30export class DeadlineTestFunction extends Lambda.Function<Lambda.Function>()(
31 "DeadlineTestFunction",
32) {}
33
34export default DeadlineTestFunction.make(
35 {
36 main,
37 functionUrl: true,
38 timeout: Duration.seconds(30),
39 },
40 Effect.gen(function* () {
41 // The farm + queue the bindings are bound to. No fleet is associated,
42 // so submitted jobs stay READY and never provision workers.
43 const farm = yield* Deadline.Farm("BindingsFarm", {
44 description: "alchemy deadline bindings fixture farm",
45 });
46 const queue = yield* Deadline.Queue("BindingsQueue", {
47 farmId: farm.farmId,
48 description: "alchemy deadline bindings fixture queue",
49 jobRunAsUser: { runAs: "WORKER_AGENT_USER" },
50 });
51 const QueueId = yield* queue.queueId;
52
53 // Event source: subscribe the host to Deadline job status changes. The
54 // deploy proves the EventBridge rule + invoke permission wiring.
55 yield* Deadline.consumeFarmEvents(
56 { kinds: ["job-run", "job-lifecycle"] },
57 (events) =>
58 Stream.runForEach(events, (event) =>
59 Effect.log(
60 `deadline job event: ${event.detail.jobId} -> ${event.detail.status}`,
61 ),
62 ),
63 );
64
65 const createJob = yield* Deadline.CreateJob(queue);
66 const getJob = yield* Deadline.GetJob(queue);
67 const updateJob = yield* Deadline.UpdateJob(queue);
68 const listJobs = yield* Deadline.ListJobs(queue);
69 const listJobParameterDefinitions =
70 yield* Deadline.ListJobParameterDefinitions(queue);
71 const searchJobs = yield* Deadline.SearchJobs(queue);
72 const searchSteps = yield* Deadline.SearchSteps(queue);
73 const searchTasks = yield* Deadline.SearchTasks(queue);
74 const getSession = yield* Deadline.GetSession(queue);
75 const listSessions = yield* Deadline.ListSessions(queue);
76 const getSessionAction = yield* Deadline.GetSessionAction(queue);
77 const listSessionActions = yield* Deadline.ListSessionActions(queue);
78 const getStep = yield* Deadline.GetStep(queue);
79 const listSteps = yield* Deadline.ListSteps(queue);
80 const updateStep = yield* Deadline.UpdateStep(queue);
81 const getTask = yield* Deadline.GetTask(queue);
82 const listTasks = yield* Deadline.ListTasks(queue);
83 const updateTask = yield* Deadline.UpdateTask(queue);
84 const startAggregation =
85 yield* Deadline.StartSessionsStatisticsAggregation(farm);
86 const getAggregation =
87 yield* Deadline.GetSessionsStatisticsAggregation(farm);
88
89 const bound = {
90 createJob,
91 getJob,
92 updateJob,
93 listJobs,
94 listJobParameterDefinitions,
95 searchJobs,
96 searchSteps,
97 searchTasks,
98 getSession,
99 listSessions,
100 getSessionAction,
101 listSessionActions,
102 getStep,
103 listSteps,
104 updateStep,
105 getTask,
106 listTasks,
107 updateTask,
108 startAggregation,
109 getAggregation,
110 };
111
112 return {
113 fetch: Effect.gen(function* () {
114 const request = yield* HttpServerRequest;
115 const url = new URL(request.originalUrl);
116 const pathname = url.pathname;
117 const param = (name: string) => url.searchParams.get(name) ?? "";
118
119 if (request.method === "GET" && pathname === "/bindings") {
120 return yield* HttpServerResponse.json({
121 bound: Object.keys(bound),
122 });
123 }
124
125 if (request.method === "GET" && pathname === "/queue") {
126 return yield* HttpServerResponse.json({ queueId: yield* QueueId });
127 }
128
129 // Submit a render job from the fixture template (farm/queue ids are
130 // injected by the binding).
131 if (request.method === "POST" && pathname === "/jobs") {
132 const { jobId } = yield* createJob({
133 // The outer HTTP fixture retries transient 5xx responses. Keep
134 // CreateJob idempotent if AWS accepted the first request but the
135 // Lambda response was lost.
136 clientToken: "alchemy-deadline-bindings-job",
137 template: JOB_TEMPLATE,
138 templateType: "JSON",
139 priority: 50,
140 });
141 return yield* HttpServerResponse.json({ jobId });
142 }
143
144 if (request.method === "GET" && pathname === "/job") {
145 const job = yield* getJob({ jobId: param("jobId") });
146 return yield* HttpServerResponse.json({
147 jobId: job.jobId,
148 name: job.name,
149 lifecycleStatus: job.lifecycleStatus,
150 priority: job.priority,
151 });
152 }
153
154 if (request.method === "POST" && pathname === "/priority") {
155 yield* updateJob({
156 jobId: param("jobId"),
157 priority: Number(param("priority")),
158 });
159 return yield* HttpServerResponse.json({ ok: true });
160 }
161
162 if (request.method === "GET" && pathname === "/jobs") {
163 const { jobs } = yield* listJobs();
164 return yield* HttpServerResponse.json({
165 ids: (jobs ?? []).map((job) => job.jobId),
166 });
167 }
168
169 // Search the queue's jobs; summaries include the queueId, which the
170 // test feeds back into the statistics aggregation route.
171 if (request.method === "POST" && pathname === "/search") {
172 const { jobs } = yield* searchJobs({ itemOffset: 0 });
173 return yield* HttpServerResponse.json({
174 ids: (jobs ?? []).map((job) => job.jobId),
175 queueIds: (jobs ?? []).map((job) => job.queueId),
176 });
177 }
178
179 if (request.method === "POST" && pathname === "/search/steps") {
180 const { steps } = yield* searchSteps({
181 itemOffset: 0,
182 jobId: param("jobId"),
183 });
184 return yield* HttpServerResponse.json({
185 ids: (steps ?? []).map((step) => step.stepId),
186 });
187 }
188
189 if (request.method === "POST" && pathname === "/search/tasks") {
190 const { tasks } = yield* searchTasks({
191 itemOffset: 0,
192 jobId: param("jobId"),
193 });
194 return yield* HttpServerResponse.json({
195 ids: (tasks ?? []).map((task) => task.taskId),
196 });
197 }
198
199 if (request.method === "GET" && pathname === "/params") {
200 const { jobParameterDefinitions } =
201 yield* listJobParameterDefinitions({ jobId: param("jobId") });
202 return yield* HttpServerResponse.json({
203 count: (jobParameterDefinitions ?? []).length,
204 });
205 }
206
207 if (request.method === "GET" && pathname === "/steps") {
208 const { steps } = yield* listSteps({ jobId: param("jobId") });
209 return yield* HttpServerResponse.json({
210 ids: (steps ?? []).map((step) => step.stepId),
211 names: (steps ?? []).map((step) => step.name),
212 });
213 }
214
215 if (request.method === "GET" && pathname === "/step") {
216 const step = yield* getStep({
217 jobId: param("jobId"),
218 stepId: param("stepId"),
219 });
220 return yield* HttpServerResponse.json({
221 stepId: step.stepId,
222 name: step.name,
223 lifecycleStatus: step.lifecycleStatus,
224 });
225 }
226
227 if (request.method === "GET" && pathname === "/tasks") {
228 const { tasks } = yield* listTasks({
229 jobId: param("jobId"),
230 stepId: param("stepId"),
231 });
232 return yield* HttpServerResponse.json({
233 ids: (tasks ?? []).map((task) => task.taskId),
234 });
235 }
236
237 if (request.method === "GET" && pathname === "/task") {
238 const task = yield* getTask({
239 jobId: param("jobId"),
240 stepId: param("stepId"),
241 taskId: param("taskId"),
242 });
243 return yield* HttpServerResponse.json({
244 taskId: task.taskId,
245 runStatus: task.runStatus,
246 });
247 }
248
249 // Cancel a single READY task (nothing runs — no fleet).
250 if (request.method === "POST" && pathname === "/task/cancel") {
251 yield* updateTask({
252 jobId: param("jobId"),
253 stepId: param("stepId"),
254 taskId: param("taskId"),
255 targetRunStatus: "CANCELED",
256 });
257 return yield* HttpServerResponse.json({ ok: true });
258 }
259
260 // Requeue the whole step's tasks back to READY.
261 if (request.method === "POST" && pathname === "/step/requeue") {
262 yield* updateStep({
263 jobId: param("jobId"),
264 stepId: param("stepId"),
265 targetTaskRunStatus: "READY",
266 });
267 return yield* HttpServerResponse.json({ ok: true });
268 }
269
270 // With no fleet there are never sessions — proves the op + IAM wire
271 // up and returns an empty page.
272 if (request.method === "GET" && pathname === "/sessions") {
273 const { sessions } = yield* listSessions({ jobId: param("jobId") });
274 return yield* HttpServerResponse.json({
275 ids: (sessions ?? []).map((session) => session.sessionId),
276 });
277 }
278
279 if (request.method === "GET" && pathname === "/session") {
280 const session = yield* getSession({
281 jobId: param("jobId"),
282 sessionId: param("sessionId"),
283 });
284 return yield* HttpServerResponse.json({
285 sessionId: session.sessionId,
286 });
287 }
288
289 // No workers means no sessions and therefore no session actions —
290 // proves the op + IAM wire up and returns an empty page. The API
291 // requires a sessionId or taskId to scope the query.
292 if (request.method === "GET" && pathname === "/session-actions") {
293 const { sessionActions } = yield* listSessionActions({
294 jobId: param("jobId"),
295 taskId: param("taskId"),
296 });
297 return yield* HttpServerResponse.json({
298 ids: (sessionActions ?? []).map((action) => action.sessionActionId),
299 });
300 }
301
302 if (request.method === "GET" && pathname === "/session-action") {
303 const action = yield* getSessionAction({
304 jobId: param("jobId"),
305 sessionActionId: param("sessionActionId"),
306 });
307 return yield* HttpServerResponse.json({
308 sessionActionId: action.sessionActionId,
309 status: action.status,
310 });
311 }
312
313 if (request.method === "POST" && pathname === "/stats") {
314 // The API rejects timestamps whose minutes/seconds are not zero —
315 // align the window to whole hours.
316 const hourMs = 3600 * 1000;
317 const endTime = new Date(Math.floor(Date.now() / hourMs) * hourMs);
318 const startTime = new Date(endTime.getTime() - 24 * hourMs);
319 const { aggregationId } = yield* startAggregation({
320 resourceIds: { queueIds: [param("queueId")] },
321 startTime,
322 endTime,
323 groupBy: ["QUEUE_ID"],
324 statistics: ["SUM"],
325 });
326 return yield* HttpServerResponse.json({ aggregationId });
327 }
328
329 if (request.method === "GET" && pathname === "/stats") {
330 const result = yield* getAggregation({
331 aggregationId: param("aggregationId"),
332 });
333 return yield* HttpServerResponse.json({
334 status: result.status,
335 count: (result.statistics ?? []).length,
336 });
337 }
338
339 return yield* HttpServerResponse.json(
340 { error: "Not found", method: request.method, pathname },
341 { status: 404 },
342 );
343 }).pipe(
344 // Surface typed operation errors as a JSON 500 so the test log
345 // shows the real cause instead of an opaque crash.
346 Effect.catch((error) =>
347 HttpServerResponse.json({ error: String(error) }, { status: 500 }),
348 ),
349 Effect.orDie,
350 ),
351 };
352 }).pipe(
353 Effect.provide(
354 Layer.mergeAll(
355 Lambda.EventSource,
356 Deadline.CreateJobHttp,
357 Deadline.GetJobHttp,
358 Deadline.UpdateJobHttp,
359 Deadline.ListJobsHttp,
360 Deadline.ListJobParameterDefinitionsHttp,
361 Deadline.SearchJobsHttp,
362 Deadline.SearchStepsHttp,
363 Deadline.SearchTasksHttp,
364 Deadline.GetSessionHttp,
365 Deadline.ListSessionsHttp,
366 Deadline.GetSessionActionHttp,
367 Deadline.ListSessionActionsHttp,
368 Deadline.GetStepHttp,
369 Deadline.ListStepsHttp,
370 Deadline.UpdateStepHttp,
371 Deadline.GetTaskHttp,
372 Deadline.ListTasksHttp,
373 Deadline.UpdateTaskHttp,
374 Deadline.StartSessionsStatisticsAggregationHttp,
375 Deadline.GetSessionsStatisticsAggregationHttp,
376 ),
377 ),
378 ),
379);