1import { loadInternalWorker } from "../internal/internal-worker.ts";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Stream from "effect/Stream";
5import * as HttpBody from "effect/http/HttpBody";
6import * as HttpClient from "effect/http/HttpClient";
7import * as NodeCrypto from "node:crypto";
8const RegistryProxyWorker = {
9 worker: () =>
10 loadInternalWorker(
11 "#cloudflare-runtime-core-worker/registry/RegistryProxy.worker",
12 ),
13};
14import {
15 defaultDurableObjectUniqueKey,
16 DEFAULT_COMPATIBILITY_DATE,
17 SERVICE_USER_WORKER,
18} from "../internal/constants.ts";
19import { formatInternalWorkerModules } from "../internal/internal-modules.ts";
20import * as Plugin from "../Plugin.ts";
21import { PluginContext } from "../PluginContext.ts";
22import { SystemError } from "../RuntimeError.shared.ts";
23import * as WorkerdConfig from "../workerd/Config.ts";
24import * as Registry from "./Registry.ts";
25import type { ResolvedTargetMap, Subscriber } from "./RegistryTypes.shared.ts";
26
27export class RegistryProxy extends Plugin.Service<
28 RegistryProxy,
29 {
30 /**
31 * Declares a service, belonging to the current worker, that will be published to the registry proxy.
32 */
33 readonly publish: (service: PublishedService) => Effect.Effect<void>;
34 /**
35 * Declares an external service that will be consumed by the current worker.
36 * Returns a service designator that proxies to the external target.
37 */
38 readonly subscribe: <T extends Subscriber>(
39 subscriber: T,
40 ) => Effect.Effect<ServiceDesignatorFromSubscriber<T>>;
41 }
42>()("cloudflare-runtime/plugin/RegistryProxy") {}
43
44type PublishedService =
45 | Subscriber.DurableObject
46 | (Subscriber.QueueConsumer & { service: string })
47 | (Omit<Subscriber.Workflow, "scriptName"> & { service: string });
48
49type ServiceDesignatorFromSubscriber<T extends Subscriber> = T extends {
50 kind: "durable-object";
51}
52 ? WorkerdConfig.Worker_DurableObjectNamespace
53 : WorkerdConfig.ServiceDesignator;
54
55const SERVICE_REGISTRY_PROXY = "cloudflare-runtime:registry-proxy";
56const SOCKET_REGISTRY_PROXY = "registry-proxy-socket";
57const SOCKET_DEBUG_PORT = "debug-port";
58
59export const RegistryProxyLive = Layer.effect(
60 RegistryProxy,
61 Effect.gen(function* () {
62 const registry = yield* Registry.Registry;
63 const http = yield* HttpClient.HttpClient;
64
65 return RegistryProxy.of(
66 Effect.gen(function* () {
67 const { worker } = yield* PluginContext;
68
69 const published: Array<PublishedService> =
70 worker.durableObjectNamespaces?.map((namespace) => ({
71 kind: "durable-object",
72 scriptName: worker.name,
73 className: namespace.className,
74 uniqueKey:
75 namespace.uniqueKey ??
76 defaultDurableObjectUniqueKey(worker.name, namespace.className),
77 })) ?? [];
78 const subscribed: Array<Subscriber> = [];
79
80 return {
81 defer: Effect.gen(function* () {
82 if (subscribed.length === 0) {
83 return {};
84 }
85 const durableObjects = subscribed.filter(
86 (subscriber) => subscriber.kind === "durable-object",
87 );
88 const service: WorkerdConfig.Service = {
89 name: SERVICE_REGISTRY_PROXY,
90 worker: {
91 compatibilityDate: DEFAULT_COMPATIBILITY_DATE,
92 compatibilityFlags: ["service_binding_extra_handlers"],
93 modules: yield* registry
94 .read(subscribed)
95 .pipe(
96 Effect.flatMap((targets) =>
97 Effect.promise(() =>
98 buildWorkerModules(targets, durableObjects),
99 ),
100 ),
101 ),
102 bindings: [
103 {
104 name: "REGISTRY_DEBUG_PORT",
105 workerdDebugPort: WorkerdConfig.kVoid,
106 },
107 ],
108 durableObjectStorage: { inMemory: WorkerdConfig.kVoid },
109 durableObjectNamespaces: durableObjects.map((subscriber) => ({
110 className: externalDurableObjectClassName(
111 subscriber.scriptName,
112 subscriber.className,
113 ),
114 uniqueKey: subscriber.uniqueKey,
115 })),
116 },
117 };
118 const socket: WorkerdConfig.Socket = {
119 name: SOCKET_REGISTRY_PROXY,
120 address: "127.0.0.1:0",
121 service: { name: SERVICE_REGISTRY_PROXY },
122 };
123 return { sockets: [socket], services: [service] };
124 }),
125 start: (ports) =>
126 // When the worker starts, publish it to the registry.
127 // If we have external services, subscribe to the registry and push updates to the proxy.
128 Effect.all(
129 [
130 Effect.gen(function* () {
131 const debugPort = ports[SOCKET_DEBUG_PORT];
132 if (!debugPort) {
133 return yield* new SystemError({
134 subtag: "RegistryProxyDebugPort",
135 message: "Debug port not found.",
136 detail: { debugPort },
137 });
138 }
139 yield* registry.write({
140 scriptName: worker.name,
141 debugPortAddress: `127.0.0.1:${debugPort}`,
142 services: [
143 {
144 kind: "worker",
145 // TODO: The fetch service should be the service name for the top of the middleware chain.
146 fetchService: SERVICE_USER_WORKER,
147 rpcService: SERVICE_USER_WORKER,
148 },
149 ...published.map((service) => ({
150 ...service,
151 service:
152 "service" in service
153 ? service.service
154 : SERVICE_USER_WORKER,
155 })),
156 ],
157 });
158 }),
159 Effect.gen(function* () {
160 if (subscribed.length === 0) {
161 return;
162 }
163 const proxyPort = ports[SOCKET_REGISTRY_PROXY];
164 if (!proxyPort) {
165 return yield* new SystemError({
166 subtag: "RegistryProxyPort",
167 message: "Registry proxy port not found.",
168 detail: { proxyPort },
169 });
170 }
171 // A replacement process starts with the original module
172 // snapshot. Restore current targets before announcing it as
173 // ready, then subscribe afresh for subsequent changes.
174 const update = (targets: ResolvedTargetMap) =>
175 HttpClient.filterStatusOk(http)
176 .post(`http://127.0.0.1:${proxyPort}/`, {
177 body: HttpBody.jsonUnsafe(targets),
178 })
179 .pipe(
180 Effect.mapError(
181 (cause) =>
182 new SystemError({
183 subtag: "RegistryProxyUpdate",
184 message:
185 "Failed to update the registry proxy targets.",
186 cause,
187 }),
188 ),
189 );
190 yield* registry.read(subscribed).pipe(Effect.flatMap(update));
191 yield* registry
192 .subscribe(subscribed)
193 .pipe(Stream.runForEach(update), Effect.forkScoped);
194 }),
195 ],
196 { concurrency: "unbounded", discard: true },
197 ),
198 api: {
199 publish: (entry) =>
200 Effect.sync(() => {
201 published.push(entry);
202 }),
203 subscribe: <T extends Subscriber>(subscriber: T) =>
204 Effect.sync<ServiceDesignatorFromSubscriber<T>>(() => {
205 subscribed.push(subscriber);
206 switch (subscriber.kind) {
207 case "worker":
208 return {
209 name: SERVICE_REGISTRY_PROXY,
210 entrypoint: "ExternalService",
211 props: {
212 json: JSON.stringify(subscriber),
213 },
214 };
215 case "durable-object":
216 return {
217 serviceName: SERVICE_REGISTRY_PROXY,
218 className: externalDurableObjectClassName(
219 subscriber.scriptName,
220 subscriber.className,
221 ),
222 };
223 case "queue-consumer":
224 return {
225 name: SERVICE_REGISTRY_PROXY,
226 entrypoint: "ExternalQueueConsumer",
227 props: {
228 json: JSON.stringify(subscriber),
229 },
230 };
231 case "workflow":
232 return {
233 name: SERVICE_REGISTRY_PROXY,
234 entrypoint: "ExternalWorkflow",
235 props: {
236 json: JSON.stringify(subscriber),
237 },
238 };
239 }
240 }),
241 },
242 };
243 }),
244 );
245 }),
246);
247
248/** Generates a stable, variable-safe class name for an external Durable Object. */
249const externalDurableObjectClassName = (
250 scriptName: string,
251 className: string,
252) =>
253 `ExternalDurableObject_${NodeCrypto.createHash("sha256").update(scriptName).update(className).digest("hex").slice(0, 16)}`;
254
255/** Builds the worker modules for the registry proxy, including a pre-populated registry and proxy classes for each external Durable Object. */
256const buildWorkerModules = async (
257 targets: ResolvedTargetMap,
258 durableObjects: ReadonlyArray<Subscriber.DurableObject>,
259) => {
260 const worker = await RegistryProxyWorker.worker();
261 const main = [
262 `import { makeExternalDurableObject, Target } from "./${worker.main}";`,
263 `export { ExternalService, ExternalWorkflow, ExternalQueueConsumer } from "./${worker.main}";`,
264 `Target.set(${JSON.stringify(targets)});`,
265 `export default {`,
266 ` async fetch(request) {`,
267 ` if (request.method === "POST") {`,
268 ` const data = await request.json();`,
269 ` Target.set(data);`,
270 ` return new Response("ok");`,
271 ` }`,
272 ` return new Response("not found", { status: 404 });`,
273 ` }`,
274 `};`,
275 ...durableObjects.map(
276 (subscriber) =>
277 `export const ${externalDurableObjectClassName(subscriber.scriptName, subscriber.className)} = makeExternalDurableObject(${JSON.stringify(subscriber)});`,
278 ),
279 ].join("\n");
280 return formatInternalWorkerModules({
281 modules: {
282 "index.worker.mjs": main,
283 ...worker.modules,
284 },
285 });
286};