1import * as Effect from "effect/Effect";
2import * as Layer from "effect/Layer";
3import * as HttpServerResponse from "effect/http/HttpServerResponse";
4import * as Namespace from "../../Namespace.ts";
5import * as Output from "../../Output.ts";
6import { Job } from "../CloudScheduler/Job.ts";
7import {
8 ScheduleEventSource as SchedulerScheduleEventSource,
9 type ScheduledEvent,
10 type ScheduleEventSourceProps,
11 type ScheduleEventSourceService,
12} from "../CloudScheduler/ScheduleEventSource.ts";
13import {
14 deliveryAudience,
15 grantSelfInvoker,
16 hostEndpoint,
17 listenForDeliveries,
18 pathSegment,
19 pushHost,
20} from "../PushDelivery.ts";
21
22/** Default delivery path for a schedule. */
23export const schedulePushPath = (id: string, props: ScheduleEventSourceProps) =>
24 props.path ?? `/__alchemy/scheduler/${pathSegment(id)}`;
25
26/**
27 * Implementation of `GCP.CloudScheduler.ScheduleEventSource` for HTTP
28 * hosts (`GCP.Run.Service` / `GCP.Function`, `GCP.CloudFunctions.Function`).
29 *
30 * Deploy-time: grants the host's runtime service account
31 * `roles/run.invoker` on the host and creates a Cloud Scheduler job that
32 * `POST`s to the host's URL with an OIDC token for that account. Runtime:
33 * claims the job's deliveries on its path, verifies the token, and runs
34 * the handler with the `X-CloudScheduler-*` headers. 204 marks the run
35 * successful; a failed handler answers 500 and the job's retry policy
36 * applies.
37 *
38 * ### Running on a schedule
39 * **Example:** Every 5 minutes
40 * ```typescript
41 * Effect.gen(function* () {
42 * yield* GCP.CloudScheduler.consumeSchedule(
43 * "Sweep",
44 * { schedule: "every 5 minutes" },
45 * (event) => Effect.log(`sweep ${event.scheduleTime}`),
46 * );
47 * }).pipe(Effect.provide(GCP.Run.ScheduleEventSource));
48 * ```
49 *
50 * @layer
51 * @provides GCP.CloudScheduler.ScheduleEventSource
52 * @category Run
53 */
54export const ScheduleEventSource = Layer.effect(
55 SchedulerScheduleEventSource,
56 Effect.gen(function* () {
57 const jobs = yield* Job;
58
59 return Effect.fn(function* <Req = never>(
60 id: string,
61 props: ScheduleEventSourceProps,
62 process: (event: ScheduledEvent) => Effect.Effect<void, never, Req>,
63 ) {
64 const host = yield* pushHost("GCP.CloudScheduler.ScheduleEventSource");
65 const path = schedulePushPath(id, props);
66
67 if (!globalThis.__ALCHEMY_RUNTIME__) {
68 const endpoint = hostEndpoint(host);
69 yield* Namespace.push(
70 host.LogicalId,
71 Effect.gen(function* () {
72 yield* grantSelfInvoker(host);
73 yield* jobs(`${id}-Schedule`, {
74 location: props.location,
75 schedule: props.schedule,
76 timeZone: props.timeZone,
77 retryConfig: props.retryConfig ?? {
78 retryCount: 5,
79 minBackoffDuration: "10s",
80 maxBackoffDuration: "120s",
81 },
82 httpTarget: {
83 uri: Output.interpolate`${endpoint.url}${path}`,
84 httpMethod: "POST",
85 body: props.body,
86 oidcToken: {
87 serviceAccountEmail: endpoint.serviceAccount,
88 audience: deliveryAudience(endpoint.url, path),
89 },
90 },
91 });
92 }),
93 );
94 }
95
96 yield* listenForDeliveries(host, path, (request) =>
97 Effect.gen(function* () {
98 const body = yield* request.text.pipe(Effect.orElseSucceed(() => ""));
99 yield* process({
100 scheduleTime:
101 request.headers["x-cloudscheduler-scheduletime"] ?? "",
102 jobName: request.headers["x-cloudscheduler-jobname"] ?? "",
103 body: body.length > 0 ? body : undefined,
104 }).pipe(Effect.orDie);
105 return HttpServerResponse.empty({ status: 204 });
106 }),
107 );
108 }) as ScheduleEventSourceService;
109 }),
110);