1import * as Alchemy from "alchemy";
2import * as Axiom from "alchemy/Axiom";
3import * as Cloudflare from "alchemy/Cloudflare";
4import * as Drizzle from "alchemy/Drizzle/Postgres";
5import * as Config from "effect/Config";
6import * as Cause from "effect/Cause";
7import * as DateTime from "effect/DateTime";
8import * as Crypto from "effect/Crypto";
9import * as Effect from "effect/Effect";
10import * as Layer from "effect/Layer";
11import * as Option from "effect/Option";
12import * as Redacted from "effect/Redacted";
13import * as Stream from "effect/Stream";
14import * as Etag from "effect/http/Etag";
15import * as HttpMiddleware from "effect/http/HttpMiddleware";
16import * as HttpPlatform from "effect/http/HttpPlatform";
17import * as HttpRouter from "effect/http/HttpRouter";
18import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
19import * as HttpApiScalar from "effect/http-api/HttpApiScalar";
20
21import { RelayApi } from "@t3tools/contracts/relay";
22
23import {
24 RELAY_HTTP_ROUTER_CONFIG,
25 traceRelayHttpRequestWith,
26 withoutCapturedParentSpan,
27} from "./http/Api.ts";
28import * as RelayHttpApi from "./http/Api.ts";
29import { ManagedEndpointZone, RelayApiZone, RelayDeploymentConfig } from "./zone.ts";
30import { RelayObservability } from "./observability.ts";
31import * as Observability from "./observability.ts";
32import * as DeliveryAttempts from "./agentActivity/DeliveryAttempts.ts";
33import * as AgentActivityRows from "./agentActivity/AgentActivityRows.ts";
34import * as Devices from "./agentActivity/Devices.ts";
35import * as DpopProofs from "./auth/DpopProofs.ts";
36import * as RelayTokens from "./auth/RelayTokens.ts";
37import * as EnvironmentCredentials from "./environments/EnvironmentCredentials.ts";
38import * as EnvironmentLinks from "./environments/EnvironmentLinks.ts";
39import * as ManagedEndpointAllocations from "./environments/ManagedEndpointAllocations.ts";
40import * as LiveActivities from "./agentActivity/LiveActivities.ts";
41import * as RelayDb from "./db.ts";
42import {
43 RelayApnsDeliveryDeadLetterQueue,
44 RelayApnsDeliveryQueue,
45 RelayFcmDeliveryQueue,
46 RelayFcmDeliveryDeadLetterQueue,
47} from "./queues.ts";
48import * as WebCrypto from "./WebCrypto.ts";
49import * as FcmAssertionSigner from "./agentActivity/FcmAssertionSigner.ts";
50import * as FcmClient from "./agentActivity/FcmClient.ts";
51import * as FcmDeliveryQueueSender from "./agentActivity/FcmDeliveryQueueSender.ts";
52import * as FcmDeliveries from "./agentActivity/FcmDeliveries.ts";
53import * as FcmDeliveryQueueConsumer from "./agentActivity/FcmDeliveryQueueConsumer.ts";
54import * as RelayConfiguration from "./Config.ts";
55import * as AgentActivityPublisher from "./agentActivity/AgentActivityPublisher.ts";
56import * as ApnsClient from "./agentActivity/ApnsClient.ts";
57import * as ApnsProviderTokens from "./agentActivity/ApnsProviderTokens.ts";
58import * as ApnsDeliveryQueue from "./agentActivity/ApnsDeliveryQueue.ts";
59import * as ApnsDeliveries from "./agentActivity/ApnsDeliveries.ts";
60import * as EnvironmentConnector from "./environments/EnvironmentConnector.ts";
61import * as EnvironmentLinker from "./environments/EnvironmentLinker.ts";
62import * as EnvironmentPublishSignatures from "./environments/EnvironmentPublishSignatures.ts";
63import * as ManagedEndpointProvider from "./environments/ManagedEndpointProvider.ts";
64import * as ManagedEndpointReaper from "./environments/ManagedEndpointReaper.ts";
65import * as ManagedTunnelLimits from "./environments/ManagedTunnelLimits.ts";
66import * as MobileRegistrations from "./agentActivity/MobileRegistrations.ts";
67import * as HookForwarder from "./hooks/HookForwarder.ts";
68import * as HeldHooks from "./hooks/HeldHooks.ts";
69import * as HookInbox from "./hooks/HookInbox.ts";
70import * as HookInboxObject from "./hooks/HookInboxObject.ts";
71
72const layerWebcrypto = Layer.succeed(
73 Crypto.Crypto,
74 Crypto.make({
75 randomBytes: (size) => globalThis.crypto.getRandomValues(new Uint8Array(size)),
76 digest: (algorithm, data) =>
77 Effect.promise(async () => {
78 const input = new Uint8Array(data.length);
79 input.set(data);
80 return new Uint8Array(await globalThis.crypto.subtle.digest(algorithm, input.buffer));
81 }),
82 }),
83);
84
85const layerHttpPlatformNotSupported = Layer.succeed(HttpPlatform.HttpPlatform, {
86 platform: "web",
87 compression: {
88 algorithms: new Set<HttpPlatform.CompressionAlgorithm>(),
89 compressResponse: (response) => Effect.succeed(response),
90 },
91 fileResponse: () => Effect.die("Relay API does not serve filesystem responses"),
92 fileWebResponse: () => Effect.die("Relay API does not serve file responses"),
93});
94
95const layerRelayApi = Layer.mergeAll(
96 RelayHttpApi.layerHealthApi,
97 RelayHttpApi.layerMetadataApi,
98 RelayHttpApi.layerMobileApi,
99 RelayHttpApi.layerClientApi,
100 RelayHttpApi.layerTokenApi,
101 RelayHttpApi.layerDpopClientApi,
102 RelayHttpApi.layerServerApi,
103);
104
105const CloudMintKeyPair = Alchemy.KeyPair("CloudMintKeyPair");
106const ApnsDeliveryJobSigningSecret = Alchemy.makeRandom("ApnsDeliveryJobSigningSecret", {
107 bytes: 32,
108});
109
110export class Api extends Cloudflare.Worker<Api, {}>()("Api") {}
111
112export const layer = Api.make(
113 RelayDeploymentConfig.pipe(
114 Effect.map(({ relayPublicDomain }) => ({
115 main: import.meta.filename,
116 compatibility: {
117 date: "2026-05-22",
118 flags: ["nodejs_compat"],
119 },
120 domain: relayPublicDomain,
121 })),
122 Effect.orDie,
123 ),
124 Effect.gen(function* () {
125 //
126 // 1. Provision Infrastructure for the Worker to use
127 //
128 const { relayPublicOrigin, stage } = yield* RelayDeploymentConfig;
129 const apnsDeliveryQueue = yield* RelayApnsDeliveryQueue;
130 const apnsDeliveryDeadLetterQueue = yield* RelayApnsDeliveryDeadLetterQueue;
131 const fcmDeliveryQueue = yield* RelayFcmDeliveryQueue;
132 const fcmDeliveryDeadLetterQueue = yield* RelayFcmDeliveryDeadLetterQueue;
133 const cloudMintKeyPair = yield* CloudMintKeyPair;
134 const relayApiZone = yield* RelayApiZone;
135 const managedEndpointZone = yield* ManagedEndpointZone;
136 const randomApnsDeliveryJobSigningSecret = yield* ApnsDeliveryJobSigningSecret;
137 yield* RelayObservability;
138
139 //
140 // 2. Create bindings
141 //
142 const apnsEnabled = yield* Config.Boolean("APNS_ENABLED").pipe(Config.withDefault(true));
143 const apnsCredentials = apnsEnabled
144 ? {
145 environment: yield* Config.schema(RelayConfiguration.ApnsEnvironment, "APNS_ENVIRONMENT"),
146 teamId: yield* Config.String("APNS_TEAM_ID"),
147 keyId: yield* Config.String("APNS_KEY_ID"),
148 bundleId: yield* Config.String("APNS_BUNDLE_ID"),
149 privateKey: yield* Config.Redacted("APNS_PRIVATE_KEY"),
150 }
151 : null;
152 const fcmServiceAccount = Option.getOrUndefined(
153 Option.filter(
154 yield* Config.option(Config.Redacted("FCM_SERVICE_ACCOUNT")),
155 (value) => Redacted.value(value).trim().length > 0,
156 ),
157 );
158 const apnsDeliveryJobSigningSecret = yield* randomApnsDeliveryJobSigningSecret;
159 const apnsDeliveryQueueSender = yield* Cloudflare.Queues.WriteQueue(apnsDeliveryQueue);
160 const fcmDeliveryQueueSender = yield* Cloudflare.Queues.WriteQueue(fcmDeliveryQueue);
161
162 const clerkSecretKey = yield* Config.Redacted("CLERK_SECRET_KEY");
163 const clerkPublishableKey = yield* Config.String("CLERK_PUBLISHABLE_KEY");
164 const clerkJwtAudience = yield* Config.String("CLERK_JWT_AUDIENCE");
165
166 const cloudMintPrivateKey = yield* cloudMintKeyPair.privateKey;
167 const cloudMintPublicKey = yield* cloudMintKeyPair.publicKey;
168 const hyperdrive = yield* Cloudflare.Hyperdrive.Connect(yield* RelayDb.RelayHyperdrive);
169 // Named prepared statements collide behind Hyperdrive's transaction-mode
170 // pool: sql-pg < 4.0.0-rc.117 names them `effect1..N` per connection, and
171 // inside a transaction a Bind can reach another request's statement
172 // (#14070, Effect-TS/effect#8320). Unnamed statements cannot collide.
173 const db = yield* Drizzle.Postgres(hyperdrive.connectionString, undefined, {
174 prepare: false,
175 });
176
177 const managedEndpointTunnelBinding = yield* Cloudflare.Tunnel.ReadWriteTunnel();
178 // Keep Worker custom-domain reconciliation ordered after API zone provisioning.
179 yield* yield* relayApiZone.zoneId;
180 const managedEndpointDnsBinding = yield* Cloudflare.DNS.ReadWriteDns(managedEndpointZone);
181 const managedEndpointZoneName = yield* managedEndpointZone.name;
182 const managedEndpointCleanupMode = yield* RelayConfiguration.managedEndpointCleanupModeConfig;
183 const legacyManagedEndpointCleanupMode =
184 yield* RelayConfiguration.legacyManagedEndpointCleanupModeConfig;
185 const legacyTunnelGraceMinutes = Option.getOrUndefined(
186 yield* RelayConfiguration.legacyTunnelGraceMinutesConfig,
187 );
188 // Keys are endpoint keys or hashes over them, which already differ per
189 // stage, so stages sharing an account cannot collide in these namespaces.
190 const hookRateLimit = yield* Cloudflare.RateLimit("HOOK_RATE_LIMIT", {
191 namespaceId: 1001,
192 simple: {
193 limit: HookForwarder.RELAY_HOOK_RATE_LIMIT.limit,
194 period: HookForwarder.RELAY_HOOK_RATE_LIMIT.periodSeconds,
195 },
196 });
197 const hookEndpointRateLimit = yield* Cloudflare.RateLimit("HOOK_ENDPOINT_RATE_LIMIT", {
198 namespaceId: 1002,
199 simple: {
200 limit: HookForwarder.RELAY_HOOK_ENDPOINT_RATE_LIMIT.limit,
201 period: HookForwarder.RELAY_HOOK_ENDPOINT_RATE_LIMIT.periodSeconds,
202 },
203 });
204 const hookInboxes = yield* HookInboxObject.HookInboxObject;
205
206 //
207 // 3. Runtime layers and app construction
208 //
209 const alchemyRuntimeContext: Alchemy.BaseRuntimeContext = yield* Cloudflare.Worker;
210
211 const loadSettings = Effect.gen(function* () {
212 return RelayConfiguration.RelayConfiguration.of({
213 relayIssuer: relayPublicOrigin,
214 ...(fcmServiceAccount ? { fcmServiceAccount } : {}),
215 apns: apnsCredentials,
216 apnsDeliveryJobSigningSecret: yield* apnsDeliveryJobSigningSecret,
217 clerkSecretKey,
218 clerkPublishableKey,
219 clerkJwtAudience,
220 cloudMintPrivateKey: yield* cloudMintPrivateKey,
221 cloudMintPublicKey: yield* cloudMintPublicKey,
222 managedEndpointBaseDomain: yield* managedEndpointZoneName,
223 managedEndpointNamespace: stage,
224 managedEndpointCleanupMode,
225 legacyManagedEndpointCleanupMode,
226 ...(legacyTunnelGraceMinutes === undefined ? {} : { legacyTunnelGraceMinutes }),
227 });
228 });
229
230 // Each managed endpoint's held webhook requests live in its own Durable Object.
231 const inboxCall =
232 <A>(operation: HookInbox.HookInboxError["operation"], endpointKey: string) =>
233 (effect: Effect.Effect<A, never, Alchemy.RuntimeContext>) =>
234 effect.pipe(
235 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
236 Effect.catchCause((cause) =>
237 Effect.fail(
238 new HookInbox.HookInboxError({
239 operation,
240 endpointKey,
241 cause: Cause.squash(cause),
242 }),
243 ),
244 ),
245 );
246 const layerHookInbox = Layer.succeed(HookInbox.HookInbox, {
247 hold: ({ endpointKey, baseUrl, hook }) =>
248 hookInboxes.getByName(endpointKey).hold(hook, baseUrl).pipe(inboxCall("hold", endpointKey)),
249 wake: ({ endpointKey, baseUrl }) =>
250 hookInboxes.getByName(endpointKey).wake(baseUrl).pipe(inboxCall("wake", endpointKey)),
251 clear: ({ endpointKey }) =>
252 hookInboxes.getByName(endpointKey).clear().pipe(inboxCall("clear", endpointKey)),
253 });
254
255 const layerRuntime = Layer.empty.pipe(
256 Layer.provideMerge(MobileRegistrations.layer),
257 Layer.provideMerge(AgentActivityPublisher.layer),
258 Layer.provideMerge(EnvironmentConnector.layer),
259 Layer.provideMerge(EnvironmentLinker.layer),
260 Layer.provideMerge(
261 Layer.mergeAll(
262 EnvironmentPublishSignatures.layer,
263 ManagedEndpointReaper.layer,
264 HeldHooks.layer,
265 ),
266 ),
267 Layer.provideMerge(
268 ManagedEndpointProvider.layerCloudflareBindings(
269 managedEndpointTunnelBinding,
270 managedEndpointDnsBinding,
271 alchemyRuntimeContext,
272 ),
273 ),
274 Layer.provideMerge(DpopProofs.layer),
275 Layer.provideMerge(ApnsDeliveries.layer),
276 Layer.provideMerge(
277 FcmDeliveries.layer.pipe(
278 Layer.provide(
279 Layer.succeed(FcmDeliveryQueueSender.FcmDeliveryQueueSender, {
280 send: (body) =>
281 fcmDeliveryQueueSender
282 .send(body)
283 .pipe(Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext)),
284 }),
285 ),
286 Layer.provideMerge(
287 FcmClient.layer.pipe(
288 Layer.provide(FcmAssertionSigner.layer),
289 Layer.provide(
290 Layer.succeed(WebCrypto.WebCrypto, { subtle: globalThis.crypto.subtle }),
291 ),
292 ),
293 ),
294 ),
295 ),
296 Layer.provideMerge(ApnsClient.layer.pipe(Layer.provideMerge(ApnsProviderTokens.layer))),
297 Layer.provideMerge(
298 ApnsDeliveryQueue.layerCloudflareQueues(apnsDeliveryQueueSender, alchemyRuntimeContext),
299 ),
300 Layer.provideMerge(Layer.mergeAll(AgentActivityRows.layer, Devices.layer, layerHookInbox)),
301 Layer.provideMerge(EnvironmentCredentials.layer),
302 Layer.provideMerge(
303 Layer.mergeAll(
304 EnvironmentLinks.layer,
305 ManagedEndpointAllocations.layer,
306 ManagedTunnelLimits.layer,
307 ),
308 ),
309 Layer.provideMerge(LiveActivities.layer),
310 Layer.provideMerge(DeliveryAttempts.layer),
311 Layer.provideMerge(RelayTokens.layer),
312 Layer.provideMerge(
313 RelayDb.RelayTransactions.layer.pipe(
314 Layer.provideMerge(Layer.succeed(RelayDb.RelayDb, db)),
315 ),
316 ),
317 Layer.provideMerge(Layer.effect(RelayConfiguration.RelayConfiguration, loadSettings)),
318 Layer.provideMerge(layerWebcrypto),
319 );
320
321 // Fails open: a limiter outage must not drop webhooks the environment would accept.
322 const allowWith =
323 (limiter: typeof hookRateLimit) =>
324 (key: string): Effect.Effect<boolean> =>
325 limiter.limit({ key }).pipe(
326 Effect.map((result) => result.success),
327 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
328 Effect.catch((error) =>
329 Effect.logWarning("Hook rate limiter unavailable", { error: error.message }).pipe(
330 // Visible on the forward span, so an outage that disables limits shows up.
331 Effect.andThen(
332 Effect.annotateCurrentSpan({ "relay.hook.rate_limiter_failed_open": true }),
333 ),
334 Effect.as(true),
335 ),
336 ),
337 );
338 const layerHookRateLimiter = Layer.succeed(HookForwarder.HookRateLimiter, {
339 allowHook: allowWith(hookRateLimit),
340 allowEndpoint: allowWith(hookEndpointRateLimit),
341 });
342
343 const layerApp = Layer.merge(
344 layerRelayApi,
345 HookForwarder.layerApi.pipe(
346 Layer.provide(HookForwarder.layer),
347 Layer.provide(layerHookRateLimiter),
348 ),
349 ).pipe(
350 Layer.provideMerge(RelayHttpApi.layerClientAuth),
351 Layer.provideMerge(RelayHttpApi.layerDpopClientAuth),
352 Layer.provideMerge(RelayHttpApi.layerEnvironmentAuth),
353 Layer.provide(layerRuntime),
354 );
355
356 yield* Cloudflare.Queues.consumeQueueMessages<unknown>(
357 apnsDeliveryQueue,
358 {
359 batchSize: 10,
360 maxRetries: 5,
361 maxWaitTime: "5 seconds",
362 retryDelay: "30 seconds",
363 deadLetterQueue: apnsDeliveryDeadLetterQueue.queueName as unknown as string,
364 },
365 (stream) =>
366 stream.pipe(
367 Stream.withSpan("relay.apn_delivery_queue.process_batch"),
368 Stream.runForEach((message) =>
369 ApnsDeliveries.ApnsDeliveries.pipe(
370 Effect.flatMap((deliveries) => deliveries.processSignedJob(message.body)),
371 Effect.withSpan("relay.apn_delivery_queue.process_message"),
372 ),
373 ),
374 Effect.provide(layerRuntime),
375 ),
376 );
377
378 yield* Cloudflare.Queues.consumeQueueMessages<unknown>(
379 fcmDeliveryQueue,
380 {
381 batchSize: 10,
382 maxRetries: 5,
383 maxWaitTime: "1 second",
384 retryDelay: "30 seconds",
385 deadLetterQueue: fcmDeliveryDeadLetterQueue.queueName as unknown as string,
386 },
387 (stream) =>
388 stream.pipe(
389 Stream.withSpan("relay.fcm_delivery_queue.process_batch"),
390 Stream.runForEach(FcmDeliveryQueueConsumer.processMessage),
391 Effect.provide(layerRuntime),
392 ),
393 );
394
395 yield* Cloudflare.Workers.cron("*/5 * * * *", () =>
396 Effect.all(
397 [
398 DpopProofs.DpopProofReplay.pipe(
399 Effect.flatMap((dpopProofs) => dpopProofs.pruneExpired),
400 // Keep completed thread rows long enough to show their final state.
401 Effect.andThen(
402 Effect.all([AgentActivityRows.AgentActivityRows, DateTime.now]).pipe(
403 Effect.flatMap(([activityRows, now]) =>
404 activityRows.pruneTerminal({
405 updatedBefore: DateTime.formatIso(DateTime.subtract(now, { minutes: 30 })),
406 }),
407 ),
408 ),
409 ),
410 Effect.catchCause((cause) =>
411 Cause.hasInterrupts(cause)
412 ? Effect.interrupt
413 : Effect.logWarning("Failed to prune expired relay state", { cause }),
414 ),
415 ),
416 ManagedEndpointReaper.ManagedEndpointReaper.pipe(
417 Effect.flatMap((reaper) => reaper.sweep.pipe(Effect.timeout("2 minutes"))),
418 Effect.tap((result) =>
419 result.scanned > 0
420 ? Effect.logInfo("Finished managed tunnel cleanup", result)
421 : Effect.void,
422 ),
423 Effect.catchCause((cause) =>
424 Cause.hasInterrupts(cause)
425 ? Effect.interrupt
426 : Effect.logWarning("Failed to clean up inactive managed tunnels", { cause }),
427 ),
428 ),
429 ],
430 { concurrency: 2, discard: true },
431 ).pipe(
432 Effect.withSpan("relay.cron.prune_expired_state"),
433 Observability.withSchemaErrorSpanAttributes,
434 Effect.provide(layerRuntime),
435 ),
436 );
437
438 const fetch = Layer.merge(
439 Layer.mergeAll(
440 HttpApiBuilder.layer(RelayApi, { openapiPath: "/openapi.json" }).pipe(
441 Layer.provide(layerApp),
442 ),
443 HttpApiScalar.layer(RelayApi, { path: "/docs" }),
444 RelayHttpApi.layerDocsRedirectRoute,
445 ).pipe(
446 Layer.provide([Etag.layerWeak, layerHttpPlatformNotSupported, RelayHttpApi.layerCors]),
447 ),
448 RelayHttpApi.layerNotFoundRoute,
449 ).pipe(
450 HttpRouter.toHttpEffect,
451 Effect.provideService(HttpRouter.RouterConfig, RELAY_HTTP_ROUTER_CONFIG),
452 withoutCapturedParentSpan,
453 Effect.map((httpEffect) =>
454 traceRelayHttpRequestWith(httpEffect, Layer.empty).pipe(
455 Observability.withSchemaErrorSpanAttributes,
456 ),
457 ),
458 );
459
460 return { fetch };
461 }).pipe(
462 Effect.provide(
463 Layer.empty.pipe(
464 Layer.provideMerge(Cloudflare.Hyperdrive.ConnectBinding),
465 Layer.provideMerge(Cloudflare.Workers.CronEventSourceLive),
466 Layer.provideMerge(Cloudflare.Queues.WriteQueueBinding),
467 Layer.provideMerge(Cloudflare.Queues.EventSourceLive),
468 Layer.provideMerge(Cloudflare.Tunnel.ReadWriteTunnelBinding),
469 Layer.provideMerge(Cloudflare.DNS.ReadWriteDnsHttp),
470 Layer.provideMerge(Cloudflare.Workers.RateLimitBinding),
471 Layer.provideMerge(HookInboxObject.layer),
472 // The worker runtime opens its own HTTP span around ours. Ours carries
473 // the route, header redaction and webhook URL redaction, and drops a
474 // webhook sender's traceparent, so the runtime's span is off for every
475 // request rather than duplicating it. Registered as telemetry:
476 // request-time context is assembled per event, and only these layers
477 // are built into it.
478 Layer.provideMerge(
479 Alchemy.Telemetry.layer(Layer.succeed(HttpMiddleware.TracerDisabledWhen)(() => true)),
480 ),
481 // The Worker's only trace exporter: every event (fetch, queue, cron,
482 // HookInboxObject calls and alarms) exports through it to Axiom.
483 Layer.provideMerge(
484 Layer.unwrap(
485 Effect.map(RelayObservability, (observability) =>
486 Axiom.Telemetry({
487 serviceName: "t3code-relay",
488 token: observability.workerIngestToken,
489 traces: observability.traces,
490 }),
491 ),
492 ),
493 ),
494 ),
495 ),
496 ),
497);
498
499export default layer;