1import { loadInternalWorker } from "../internal/internal-worker.ts";
2import * as Cause from "effect/Cause";
3import * as Cron from "effect/Cron";
4import * as Effect from "effect/Effect";
5import * as FileSystem from "effect/FileSystem";
6import * as Layer from "effect/Layer";
7import * as Path from "effect/Path";
8import * as Result from "effect/Result";
9import * as Schedule from "effect/Schedule";
10const EntryWorker = {
11 worker: () =>
12 loadInternalWorker("#cloudflare-runtime-core-worker/globals/entry.worker"),
13};
14import {
15 DEFAULT_COMPATIBILITY_DATE,
16 SERVICE_USER_WORKER,
17 SOCKET_USER_ENTRY,
18} from "../internal/constants.ts";
19import { formatInternalWorkerModules } from "../internal/internal-modules.ts";
20import * as Plugin from "../Plugin.ts";
21import { PluginContext } from "../PluginContext.ts";
22import { ConfigError } from "../RuntimeError.shared.ts";
23import type * as WorkerdConfig from "../workerd/Config.ts";
24import { BINDING_PROXY_SHARED_SECRET } from "./ProxyHeaders.shared.ts";
25import * as Cf from "./Cf.ts";
26import {
27 BINDING_EMAIL_DIRECTORY,
28 BINDING_EMAIL_DISK,
29 SERVICE_EMAIL_STORAGE,
30} from "./EmailOptions.shared.ts";
31import { BINDING_USER_WORKER_DIRECT } from "./EntryOptions.shared.ts";
32import * as Internet from "./Internet.ts";
33import { PATH_SCHEDULED } from "./ScheduledOptions.shared.ts";
34import * as Storage from "./Storage.ts";
35
36export class Globals extends Plugin.Service<Globals>()(
37 "cloudflare-runtime/plugin/Globals",
38) {}
39
40/**
41 * Fire one cron expression against the entry socket's scheduled route.
42 *
43 * Driven by `Schedule.cron`: each iteration sleeps until the expression's
44 * next match, then triggers the route — no polling, no manual clock math.
45 * Trigger failures are logged and the loop continues. The loop runs on the
46 * worker's start `Scope`, so it dies with the worker.
47 */
48const cronLoop = (
49 workerName: string,
50 expression: string,
51 cron: Cron.Cron,
52 port: number,
53) => {
54 const fire = Effect.gen(function* () {
55 const time = yield* Effect.sync(() => Date.now());
56 const url =
57 `http://127.0.0.1:${port}${PATH_SCHEDULED}` +
58 `?cron=${encodeURIComponent(expression)}&time=${time}`;
59 const response = yield* Effect.tryPromise(() =>
60 fetch(url, { method: "POST" }),
61 );
62 const body = yield* Effect.tryPromise(() => response.text());
63 if (response.ok) {
64 yield* Effect.logInfo(
65 `[cron:${workerName}] "${expression}" triggered scheduled(): ${body}`,
66 );
67 } else {
68 yield* Effect.logWarning(
69 `[cron:${workerName}] "${expression}" scheduled() handler failed: ${body}`,
70 );
71 }
72 }).pipe(
73 Effect.catchCause((cause) =>
74 Effect.logWarning(
75 `[cron:${workerName}] "${expression}" trigger failed`,
76 Cause.squash(cause),
77 ),
78 ),
79 );
80 // `Schedule.cron` over a pre-parsed `Cron` cannot fail to parse, so the
81 // loop's error channel is `never`.
82 return fire.pipe(Effect.schedule(Schedule.cron(cron)), Effect.asVoid);
83};
84
85export const GlobalsLive = Layer.effect(
86 Globals,
87 Effect.gen(function* () {
88 const fs = yield* FileSystem.FileSystem;
89 const path = yield* Path.Path;
90 const internet = yield* Internet.Internet;
91 const storage = yield* Storage.Storage;
92 const cf = yield* Cf.Cf;
93 const modules = formatInternalWorkerModules(
94 yield* Effect.promise(EntryWorker.worker),
95 );
96
97 // Inbound email replies (`message.reply(...)` from the user worker's
98 // `email()` handler) are persisted under `{storage}/email` — the same
99 // directory the send-email simulator writes to. The entry middleware
100 // gets a disk service binding plus the node-side absolute path so its
101 // logged file paths point at real files. Miniflare persists replies via
102 // its loopback `store-temp-file` endpoint instead.
103 const storageDiskPath = "disk" in storage ? storage.disk?.path : undefined;
104 const email =
105 storageDiskPath === undefined
106 ? undefined
107 : yield* Effect.gen(function* () {
108 const persistPath = path.join(storageDiskPath, "email");
109 yield* fs.makeDirectory(persistPath, { recursive: true }).pipe(
110 Effect.mapError(
111 (cause) =>
112 new ConfigError({
113 subtag: "Globals",
114 message: `Failed to create email persistence directory "${persistPath}": ${cause.message}`,
115 hint: "Ensure the storage directory is writable.",
116 detail: { persistPath },
117 cause,
118 }),
119 ),
120 );
121 return {
122 persistPath,
123 service: {
124 name: SERVICE_EMAIL_STORAGE,
125 disk: { path: persistPath, writable: true },
126 } satisfies WorkerdConfig.Service,
127 };
128 });
129 return Globals.of(
130 Effect.gen(function* () {
131 const { worker } = yield* PluginContext;
132 // Per-worker `request.cf` override; falls back to the runtime-wide
133 // `Cf` reference (Miniflare's static placeholder by default).
134 const blob = worker.cf ?? cf;
135 // Validate cron expressions up front so a typo is a `ConfigError` at
136 // plan/config time rather than a dead timer at runtime. Cloudflare
137 // evaluates crons in UTC; pin the parse to UTC so `Cron.next`
138 // matches production timing.
139 const crons = yield* Effect.forEach(
140 worker.crons ?? [],
141 (expression) => {
142 const parsed = Cron.parse(expression, "UTC");
143 return Result.isSuccess(parsed)
144 ? Effect.succeed({ expression, cron: parsed.success })
145 : Effect.fail(
146 new ConfigError({
147 subtag: "InvalidCron",
148 message: `Invalid cron expression "${expression}": ${parsed.failure.message}`,
149 hint: 'Use a standard cron expression, e.g. "*/5 * * * *".',
150 detail: { workerName: worker.name, expression },
151 }),
152 );
153 },
154 );
155 return {
156 middlewares: [
157 {
158 name: "plugin:entry",
159 // Accept and route internal requests before assets or other middleware.
160 order: -2,
161 worker: {
162 compatibilityDate: DEFAULT_COMPATIBILITY_DATE,
163 compatibilityFlags: [
164 "experimental",
165 "enable_request_signal",
166 "service_binding_extra_handlers",
167 ],
168 modules,
169 bindings: [
170 { name: "CF_BLOB", json: JSON.stringify(blob) },
171 {
172 name: BINDING_PROXY_SHARED_SECRET,
173 text: worker.proxySharedSecret ?? "",
174 },
175 // Non-fetch dispatch (queue/scheduled/email JSRPC) goes
176 // straight to the raw user worker: the `USER_WORKER`
177 // upstream binding points at the next middleware in the
178 // chain, and middlewares are fetch-only HTTP interceptors
179 // (see EntryOptions.shared.ts).
180 {
181 name: BINDING_USER_WORKER_DIRECT,
182 service: { name: SERVICE_USER_WORKER },
183 },
184 ...(email === undefined
185 ? []
186 : [
187 {
188 name: BINDING_EMAIL_DISK,
189 service: { name: SERVICE_EMAIL_STORAGE },
190 },
191 {
192 name: BINDING_EMAIL_DIRECTORY,
193 json: JSON.stringify(email.persistPath),
194 },
195 ]),
196 ],
197 },
198 upstreamBindingName: "USER_WORKER",
199 },
200 ],
201 services: [
202 internet,
203 storage,
204 ...(email === undefined ? [] : [email.service]),
205 ],
206 start:
207 crons.length === 0
208 ? undefined
209 : (ports) =>
210 Effect.gen(function* () {
211 const port = ports[SOCKET_USER_ENTRY];
212 if (port === undefined) return;
213 yield* Effect.forEach(
214 crons,
215 ({ expression, cron }) =>
216 Effect.forkScoped(
217 cronLoop(worker.name, expression, cron, port),
218 ),
219 { discard: true },
220 );
221 }),
222 };
223 }),
224 );
225 }),
226);