authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-07-27 17:59:55+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-07-27 15:59:55+00:00
log180620e43688843e694c6fd155d4e113886698a0
treef24e30ba8dfb0b9777382f1fc3f5a8f1d222f7ee
parent96398e3775fc6c74525e4a211e75e580c6cf4225
signature Signed by PGP key B5690EEEBB952194

Fix Relay Worker RuntimeContext wiring (#4653)

Co-authored-by: codex <codex@users.noreply.github.com>

5 files changed, 218 insertions(+), 140 deletions(-)

infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts+36
......@@ -1,5 +1,6 @@
11import * as NodeCryptoLayer from "@effect/platform-node/NodeCrypto";
22import { describe, expect, it } from "@effect/vitest";
3import * as Alchemy from "alchemy";
34import * as Cloudflare from "alchemy/Cloudflare";
45import * as Effect from "effect/Effect";
56import * as Layer from "effect/Layer";
......@@ -28,6 +29,41 @@ const config: RelayConfiguration.RelayConfiguration["Service"] = {
2829};
2930
3031describe("ApnsDeliveryQueue", () => {
32 it.effect("does not require the deployment RuntimeContext when building the Worker layer", () => {
33 const sent: unknown[] = [];
34 const sender: Cloudflare.Queues.WriteQueueClient = {
35 raw: Effect.die("raw queue binding is not used"),
36 send: (body) =>
37 Effect.sync(() => {
38 sent.push(body);
39 }),
40 sendBatch: () => Effect.die("batch queue binding is not used"),
41 };
42 const runtimeContext = {} as Alchemy.BaseRuntimeContext;
43 const layer = ApnsDeliveryQueue.layerCloudflareQueues(sender, runtimeContext).pipe(
44 Layer.provide(NodeCryptoLayer.layer),
45 Layer.provide(RelayConfiguration.layer(config)),
46 );
47
48 return Effect.gen(function* () {
49 const queue = yield* ApnsDeliveryQueue.ApnsDeliveryQueue;
50 yield* queue.enqueuePushNotification({
51 userId: "user-1",
52 deviceId: "device-1",
53 token: "push-token",
54 notification: {
55 title: "Thread",
56 body: "Input: Project",
57 environmentId: "env-1",
58 threadId: "thread-1",
59 deepLink: "/threads/env-1/thread-1",
60 },
61 });
62
63 expect(sent).toHaveLength(1);
64 }).pipe(Effect.provide(layer));
65 });
66
3167 it.effect("preserves job identity and the queue sender cause", () => {
3268 const cause = new Error("queue unavailable");
3369 const senderCause = new Cloudflare.Queues.SendError({
infra/relay/src/agentActivity/ApnsDeliveryQueue.ts+11-12
......@@ -208,21 +208,20 @@ export const make = Effect.gen(function* () {
208208
209209export const layer = Layer.effect(ApnsDeliveryQueue, make);
210210
211export const layerCloudflareQueues = (sender: Cloudflare.Queues.WriteQueueClient) =>
211export const layerCloudflareQueues = (
212 sender: Cloudflare.Queues.WriteQueueClient,
213 alchemyRuntimeContext: Alchemy.BaseRuntimeContext,
214) =>
212215 layer.pipe(
213216 Layer.provide(
214 Layer.effect(
217 Layer.succeed(
215218 ApnsDeliveryQueueSender,
216 Alchemy.RuntimeContext.pipe(
217 Effect.map((alchemyRuntimeContext) =>
218 ApnsDeliveryQueueSender.of({
219 send: (body) =>
220 sender
221 .send(body)
222 .pipe(Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext)),
223 }),
224 ),
225 ),
219 ApnsDeliveryQueueSender.of({
220 send: (body) =>
221 sender
222 .send(body)
223 .pipe(Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext)),
224 }),
226225 ),
227226 ),
228227 );
infra/relay/src/environments/ManagedEndpointProvider.test.ts+43
......@@ -2,6 +2,8 @@ import * as NodeCrypto from "node:crypto";
22import * as NodeServices from "@effect/platform-node/NodeServices";
33
44import { describe, expect, it } from "@effect/vitest";
5import * as Alchemy from "alchemy";
6import * as Cloudflare from "alchemy/Cloudflare";
57import * as Effect from "effect/Effect";
68import * as Layer from "effect/Layer";
79import * as Redacted from "effect/Redacted";
......@@ -280,6 +282,47 @@ function expectedManagedTunnelName(environmentId: string, userId = "user_ABC"):
280282}
281283
282284describe("ManagedEndpointProvider", () => {
285 it.effect("does not require the deployment RuntimeContext when building the Worker layer", () => {
286 const tunnelClient = {
287 list: () => Effect.succeed({ result: [] }),
288 create: (request: { readonly name: string }) =>
289 Effect.succeed({ id: "tunnel-id", name: request.name }),
290 putConfiguration: () => Effect.void,
291 getToken: () => Effect.succeed("connector-token"),
292 delete: () => Effect.void,
293 } as unknown as Cloudflare.Tunnel.ReadWriteTunnelClient;
294 const dnsClient = {
295 listDnsRecords: () => Effect.succeed({ result: [] }),
296 createDnsRecord: () => Effect.succeed({ id: "dns-record-id" }),
297 updateDnsRecord: () => Effect.void,
298 deleteDnsRecord: () => Effect.void,
299 } as unknown as Cloudflare.DNS.ReadWriteDnsClient;
300 const runtimeContext = {} as Alchemy.BaseRuntimeContext;
301 const layer = ManagedEndpointProvider.layerCloudflareBindings(
302 tunnelClient,
303 dnsClient,
304 runtimeContext,
305 ).pipe(
306 Layer.provideMerge(NodeServices.layer),
307 Layer.provide(RelayConfiguration.layer(config)),
308 Layer.provide(
309 Layer.succeed(ManagedEndpointAllocations.ManagedEndpointAllocations, makeAllocations()),
310 ),
311 Layer.provide(Layer.succeed(ManagedTunnelLimits.ManagedTunnelLimits, makeTunnelLimits())),
312 );
313
314 return Effect.gen(function* () {
315 const provider = yield* ManagedEndpointProvider.ManagedEndpointProvider;
316 const result = yield* provider.provision({
317 userId: "user_ABC",
318 environmentId: "env_ABC",
319 origin: { localHttpHost: "127.0.0.1", localHttpPort: 3773 },
320 });
321
322 expect(result.runtime.connectorToken).toBe("connector-token");
323 }).pipe(Effect.provide(layer));
324 });
325
283326 it.effect("provisions a Cloudflare tunnel endpoint and connector token", () => {
284327 const tunnelCalls: TunnelCall[] = [];
285328 const dnsCalls: DnsCall[] = [];
infra/relay/src/environments/ManagedEndpointProvider.ts+122-127
......@@ -811,137 +811,132 @@ export const layer = Layer.effect(ManagedEndpointProvider, make);
811811export const layerCloudflareBindings = (
812812 tunnelClient: Cloudflare.Tunnel.ReadWriteTunnelClient,
813813 dnsClient: Cloudflare.DNS.ReadWriteDnsClient,
814 alchemyRuntimeContext: Alchemy.BaseRuntimeContext,
814815) =>
815816 layer.pipe(
816817 Layer.provide(
817 Layer.unwrap(
818 Alchemy.RuntimeContext.pipe(
819 Effect.map((alchemyRuntimeContext) =>
820 Layer.mergeAll(
821 layerTunnelClient({
822 list: (request) =>
823 tunnelClient.list(request).pipe(
824 Effect.mapError(
825 (cause) =>
826 new ManagedEndpointTunnelClientError({
827 operation: "list",
828 tunnelName: request.name,
829 cause,
830 }),
831 ),
832 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
833 ),
834 create: (request) =>
835 tunnelClient.create(request).pipe(
836 Effect.mapError(
837 (cause) =>
838 new ManagedEndpointTunnelClientError({
839 operation: "create",
840 tunnelName: request.name,
841 cause,
842 }),
843 ),
844 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
845 ),
846 putConfiguration: (tunnelId, config) =>
847 tunnelClient.putConfiguration(tunnelId, config).pipe(
848 Effect.mapError(
849 (cause) =>
850 new ManagedEndpointTunnelClientError({
851 operation: "put-configuration",
852 tunnelId,
853 cause,
854 }),
855 ),
856 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
857 ),
858 getToken: (tunnelId) =>
859 tunnelClient.getToken(tunnelId).pipe(
860 Effect.mapError(
861 (cause) =>
862 new ManagedEndpointTunnelClientError({
863 operation: "get-token",
864 tunnelId,
865 cause,
866 }),
867 ),
868 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
869 ),
870 delete: (tunnelId) =>
871 tunnelClient.delete(tunnelId).pipe(
872 Effect.mapError(
873 (cause) =>
874 new ManagedEndpointTunnelClientError({
875 operation: "delete",
876 tunnelId,
877 cause,
878 }),
879 ),
880 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
881 ),
882 }),
883 layerDnsClient({
884 listRecords: (hostname) =>
885 dnsClient.listDnsRecords({ search: hostname }).pipe(
886 Effect.map((response) =>
887 response.result.filter(
888 (record): record is typeof record & { readonly id: string } =>
889 typeof record.id === "string" &&
890 normalizeHostname(record.name) === normalizeHostname(hostname),
891 ),
892 ),
893 Effect.mapError(
894 (cause) =>
895 new ManagedEndpointDnsClientError({
896 operation: "list-records",
897 hostname,
898 cause,
899 }),
900 ),
901 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
902 ),
903 createRecord: (request) =>
904 dnsClient.createDnsRecord(request).pipe(
905 Effect.map((response) => ({ id: response.id })),
906 Effect.mapError(
907 (cause) =>
908 new ManagedEndpointDnsClientError({
909 operation: "create-record",
910 hostname: request.name,
911 cause,
912 }),
913 ),
914 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
915 ),
916 updateRecord: (dnsRecordId, request) =>
917 dnsClient.updateDnsRecord(dnsRecordId, request).pipe(
918 Effect.mapError(
919 (cause) =>
920 new ManagedEndpointDnsClientError({
921 operation: "update-record",
922 hostname: request.name,
923 dnsRecordId,
924 cause,
925 }),
926 ),
927 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
928 ),
929 deleteRecord: (dnsRecordId) =>
930 dnsClient.deleteDnsRecord(dnsRecordId).pipe(
931 Effect.mapError(
932 (cause) =>
933 new ManagedEndpointDnsClientError({
934 operation: "delete-record",
935 dnsRecordId,
936 cause,
937 }),
938 ),
939 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
940 ),
941 }),
818 Layer.mergeAll(
819 layerTunnelClient({
820 list: (request) =>
821 tunnelClient.list(request).pipe(
822 Effect.mapError(
823 (cause) =>
824 new ManagedEndpointTunnelClientError({
825 operation: "list",
826 tunnelName: request.name,
827 cause,
828 }),
829 ),
830 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
942831 ),
943 ),
944 ),
832 create: (request) =>
833 tunnelClient.create(request).pipe(
834 Effect.mapError(
835 (cause) =>
836 new ManagedEndpointTunnelClientError({
837 operation: "create",
838 tunnelName: request.name,
839 cause,
840 }),
841 ),
842 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
843 ),
844 putConfiguration: (tunnelId, config) =>
845 tunnelClient.putConfiguration(tunnelId, config).pipe(
846 Effect.mapError(
847 (cause) =>
848 new ManagedEndpointTunnelClientError({
849 operation: "put-configuration",
850 tunnelId,
851 cause,
852 }),
853 ),
854 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
855 ),
856 getToken: (tunnelId) =>
857 tunnelClient.getToken(tunnelId).pipe(
858 Effect.mapError(
859 (cause) =>
860 new ManagedEndpointTunnelClientError({
861 operation: "get-token",
862 tunnelId,
863 cause,
864 }),
865 ),
866 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
867 ),
868 delete: (tunnelId) =>
869 tunnelClient.delete(tunnelId).pipe(
870 Effect.mapError(
871 (cause) =>
872 new ManagedEndpointTunnelClientError({
873 operation: "delete",
874 tunnelId,
875 cause,
876 }),
877 ),
878 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
879 ),
880 }),
881 layerDnsClient({
882 listRecords: (hostname) =>
883 dnsClient.listDnsRecords({ search: hostname }).pipe(
884 Effect.map((response) =>
885 response.result.filter(
886 (record): record is typeof record & { readonly id: string } =>
887 typeof record.id === "string" &&
888 normalizeHostname(record.name) === normalizeHostname(hostname),
889 ),
890 ),
891 Effect.mapError(
892 (cause) =>
893 new ManagedEndpointDnsClientError({
894 operation: "list-records",
895 hostname,
896 cause,
897 }),
898 ),
899 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
900 ),
901 createRecord: (request) =>
902 dnsClient.createDnsRecord(request).pipe(
903 Effect.map((response) => ({ id: response.id })),
904 Effect.mapError(
905 (cause) =>
906 new ManagedEndpointDnsClientError({
907 operation: "create-record",
908 hostname: request.name,
909 cause,
910 }),
911 ),
912 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
913 ),
914 updateRecord: (dnsRecordId, request) =>
915 dnsClient.updateDnsRecord(dnsRecordId, request).pipe(
916 Effect.mapError(
917 (cause) =>
918 new ManagedEndpointDnsClientError({
919 operation: "update-record",
920 hostname: request.name,
921 dnsRecordId,
922 cause,
923 }),
924 ),
925 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
926 ),
927 deleteRecord: (dnsRecordId) =>
928 dnsClient.deleteDnsRecord(dnsRecordId).pipe(
929 Effect.mapError(
930 (cause) =>
931 new ManagedEndpointDnsClientError({
932 operation: "delete-record",
933 dnsRecordId,
934 cause,
935 }),
936 ),
937 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
938 ),
939 }),
945940 ),
946941 ),
947942 );
infra/relay/src/worker.ts+6-1
......@@ -154,6 +154,8 @@ export const ApiLive = Api.make(
154154 //
155155 // 3. Runtime layers and app construction
156156 //
157 const alchemyRuntimeContext: Alchemy.BaseRuntimeContext = yield* Cloudflare.Worker;
158
157159 const loadSettings = Effect.gen(function* () {
158160 return RelayConfiguration.RelayConfiguration.of({
159161 relayIssuer: relayPublicOrigin,
......@@ -193,12 +195,15 @@ export const ApiLive = Api.make(
193195 ManagedEndpointProvider.layerCloudflareBindings(
194196 managedEndpointTunnelBinding,
195197 managedEndpointDnsBinding,
198 alchemyRuntimeContext,
196199 ),
197200 ),
198201 Layer.provideMerge(DpopProofs.layer),
199202 Layer.provideMerge(ApnsDeliveries.layer),
200203 Layer.provideMerge(ApnsClient.layer.pipe(Layer.provideMerge(ApnsProviderTokens.layer))),
201 Layer.provideMerge(ApnsDeliveryQueue.layerCloudflareQueues(apnsDeliveryQueueSender)),
204 Layer.provideMerge(
205 ApnsDeliveryQueue.layerCloudflareQueues(apnsDeliveryQueueSender, alchemyRuntimeContext),
206 ),
202207 Layer.provideMerge(AgentActivityRows.layer),
203208 Layer.provideMerge(Devices.layer),
204209 Layer.provideMerge(EnvironmentCredentials.layer),