Commit 7a047d0136de606accc8270dce64f2260529c120 | the-knife

apps/server/src/cloud/CloudPreferences.test.ts created+219

...	@@ -0,0 +1,219 @@
1	import { EnvironmentId } from "@t3tools/contracts";
2	import { assert, it } from "@effect/vitest";
3	import * as Effect from "effect/Effect";
4	import * as Fiber from "effect/Fiber";
5	import * as Layer from "effect/Layer";
6	import * as Option from "effect/Option";
7	import * as FetchHttpClient from "effect/http/FetchHttpClient";
8
9	import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
10	import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
11	import * as AgentAwarenessRelay from "../relay/AgentAwarenessRelay.ts";
12	import * as CloudPreferences from "./CloudPreferences.ts";
13	import {
14	HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET,
15	PUBLISH_AGENT_ACTIVITY_SECRET,
16	RELAY_ENVIRONMENT_CREDENTIAL_SECRET,
17	RELAY_URL_SECRET,
18	} from "./config.ts";
19
20	const encode = (value: string) => new TextEncoder().encode(value);
21
22	/** A linked environment whose secret store can refuse writes, and the relay calls it made. */
23	const withService = <A, E>(
24	options: {
25	readonly failHoldWrite?: boolean;
26	readonly failActivityWrite?: boolean;
27	readonly relayFails?: boolean;
28	readonly failActivityRead?: boolean;
29	readonly failHoldRead?: boolean;
30	/** The first relay call reports itself, then waits for this before answering. */
31	readonly holdFirstRelayCall?: {
32	readonly started: () => void;
33	readonly released: Promise<void>;
34	};
35	},
36	body: (input: {
37	readonly preferences: CloudPreferences.CloudPreferences["Service"];
38	readonly stored: Map<string, Uint8Array>;
39	readonly relayCalls: Array<boolean>;
40	}) => Effect.Effect<A, E, never>,
41	) =>
42	Effect.gen(function* () {
43	const stored = new Map<string, Uint8Array>([
44	[RELAY_URL_SECRET, encode("https://relay.test")],
45	[RELAY_ENVIRONMENT_CREDENTIAL_SECRET, encode("credential")],
46	[HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET, encode("false")],
47	[PUBLISH_AGENT_ACTIVITY_SECRET, encode("false")],
48	]);
49	const relayCalls: Array<boolean> = [];
50	const fetch: typeof globalThis.fetch = Object.assign(
51	(_input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
52	const payload = JSON.parse(String(init?.body)) as { holdWebhooksWhileOffline: boolean };
53	relayCalls.push(payload.holdWebhooksWhileOffline);
54	if (relayCalls.length === 1 && options.holdFirstRelayCall) {
55	options.holdFirstRelayCall.started();
56	return options.holdFirstRelayCall.released.then(() => Response.json(payload));
57	}
58	return options.relayFails
59	? Promise.resolve(new Response("unavailable", { status: 503 }))
60	: Promise.resolve(Response.json(payload));
61	},
62	{ preconnect: () => {} },
63	);
64	const dependencies = Layer.mergeAll(
65	Layer.mock(ServerSecretStore.ServerSecretStore)({
66	get: (name) =>
67	(options.failActivityRead && name === PUBLISH_AGENT_ACTIVITY_SECRET) \|\|
68	(options.failHoldRead && name === HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)
69	? Effect.fail(
70	new ServerSecretStore.SecretStoreReadError({
71	resource: name,
72	cause: new Error("busy"),
73	}),
74	)
75	: Effect.succeed(Option.fromNullishOr(stored.get(name))),
76	set: (name, value) =>
77	(options.failHoldWrite && name === HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET) \|\|
78	(options.failActivityWrite && name === PUBLISH_AGENT_ACTIVITY_SECRET)
79	? Effect.fail(
80	new ServerSecretStore.SecretStorePersistError({
81	resource: name,
82	cause: new Error("read-only"),
83	}),
84	)
85	: Effect.sync(() => void stored.set(name, value)),
86	}),
87	Layer.mock(ServerEnvironment.ServerEnvironment)({
88	getEnvironmentId: Effect.succeed(EnvironmentId.make("environment-1")),
89	}),
90	Layer.mock(AgentAwarenessRelay.AgentAwarenessRelay)({ requestCatchUp: () => Effect.void }),
91	);
92	return yield* Effect.gen(function* () {
93	const preferences = yield* CloudPreferences.CloudPreferences;
94	return yield* body({ preferences, stored, relayCalls });
95	}).pipe(
96	Effect.provide(CloudPreferences.layer.pipe(Layer.provide(dependencies))),
97	Effect.provideService(FetchHttpClient.Fetch, fetch),
98	);
99	});
100
101	it.effect("tells the relay before saving the hold setting locally", () =>
102	withService({}, ({ preferences, stored, relayCalls }) =>
103	Effect.gen(function* () {
104	yield* preferences.update({ publishAgentActivity: true, holdWebhooksWhileOffline: true });
105	assert.deepEqual(relayCalls, [true]);
106	assert.equal(
107	new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
108	"true",
109	);
110	}),
111	),
112	);
113
114	it.effect("puts the relay back when the local save fails", () =>
115	withService({ failHoldWrite: true }, ({ preferences, stored, relayCalls }) =>
116	Effect.gen(function* () {
117	const error = yield* preferences
118	.update({ publishAgentActivity: true, holdWebhooksWhileOffline: true })
119	.pipe(Effect.flip);
120	assert.equal(error._tag, "EnvironmentHttpInternalServerError");
121	assert.deepEqual(relayCalls, [true, false]);
122	assert.equal(
123	new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
124	"false",
125	);
126	}),
127	),
128	);
129
130	it.effect("leaves the relay untouched when the activity setting can't be saved", () =>
131	withService({ failActivityWrite: true }, ({ preferences, stored, relayCalls }) =>
132	Effect.gen(function* () {
133	const error = yield* preferences
134	.update({ publishAgentActivity: true, holdWebhooksWhileOffline: true })
135	.pipe(Effect.flip);
136	assert.equal(error._tag, "EnvironmentHttpInternalServerError");
137	assert.deepEqual(relayCalls, []);
138	assert.equal(
139	new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
140	"false",
141	);
142	}),
143	),
144	);
145
146	it.effect("keeps both settings unchanged when the relay refuses the hold change", () =>
147	withService({ relayFails: true }, ({ preferences, stored }) =>
148	Effect.gen(function* () {
149	yield* preferences
150	.update({ publishAgentActivity: true, holdWebhooksWhileOffline: true })
151	.pipe(Effect.flip);
152	assert.equal(new TextDecoder().decode(stored.get(PUBLISH_AGENT_ACTIVITY_SECRET)), "false");
153	assert.equal(
154	new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
155	"false",
156	);
157	}),
158	),
159	);
160
161	it.effect("changes nothing when the current activity setting can't be read", () =>
162	withService({ failActivityRead: true, relayFails: true }, ({ preferences, stored, relayCalls }) =>
163	Effect.gen(function* () {
164	const error = yield* preferences
165	.update({ publishAgentActivity: true, holdWebhooksWhileOffline: true })
166	.pipe(Effect.flip);
167	assert.equal(error._tag, "EnvironmentHttpInternalServerError");
168	assert.deepEqual(relayCalls, []);
169	assert.equal(new TextDecoder().decode(stored.get(PUBLISH_AGENT_ACTIVITY_SECRET)), "false");
170	}),
171	),
172	);
173
174	it.effect("two overlapping updates are applied one after the other", () => {
175	let release!: () => void;
176	let started!: () => void;
177	const released = new Promise<void>((resolve) => {
178	release = resolve;
179	});
180	const firstAtRelay = new Promise<void>((resolve) => {
181	started = resolve;
182	});
183	return withService({ holdFirstRelayCall: { started, released } }, ({ preferences, stored }) =>
184	Effect.gen(function* () {
185	const first = yield* preferences
186	.update({ publishAgentActivity: true, holdWebhooksWhileOffline: true })
187	.pipe(Effect.forkChild);
188	yield* Effect.promise(() => firstAtRelay);
189	const second = yield* preferences
190	.update({ publishAgentActivity: false, holdWebhooksWhileOffline: false })
191	.pipe(Effect.forkChild);
192	// Give the second update every chance to run; it must wait for the first.
193	yield* Effect.yieldNow;
194	assert.equal(new TextDecoder().decode(stored.get(PUBLISH_AGENT_ACTIVITY_SECRET)), "true");
195	release();
196	yield* Fiber.join(first);
197	yield* Fiber.join(second);
198	assert.equal(new TextDecoder().decode(stored.get(PUBLISH_AGENT_ACTIVITY_SECRET)), "false");
199	assert.equal(
200	new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
201	"false",
202	);
203	}),
204	);
205	});
206
207	it.effect("changes nothing when the current hold setting can't be read", () =>
208	withService({ failHoldRead: true, failHoldWrite: true }, ({ preferences, stored, relayCalls }) =>
209	Effect.gen(function* () {
210	stored.set(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET, encode("true"));
211	const error = yield* preferences
212	.update({ publishAgentActivity: true, holdWebhooksWhileOffline: false })
213	.pipe(Effect.flip);
214	assert.equal(error._tag, "EnvironmentHttpInternalServerError");
215	assert.deepEqual(relayCalls, []);
216	assert.equal(new TextDecoder().decode(stored.get(PUBLISH_AGENT_ACTIVITY_SECRET)), "false");
217	}),
218	),
219	);

apps/server/src/cloud/CloudPreferences.ts created+130

...	@@ -0,0 +1,130 @@
1	import {
2	type EnvironmentCloudPreferencesRequest,
3	EnvironmentHttpBadRequestError,
4	EnvironmentHttpInternalServerError,
5	} from "@t3tools/contracts";
6	import * as Context from "effect/Context";
7	import * as Effect from "effect/Effect";
8	import * as Layer from "effect/Layer";
9	import * as Option from "effect/Option";
10	import * as Semaphore from "effect/Semaphore";
11
12	import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
13	import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
14	import * as AgentAwarenessRelay from "../relay/AgentAwarenessRelay.ts";
15	import { makeRelayEnvironmentClient } from "../relay/relayEnvironmentClient.ts";
16	import {
17	HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET,
18	PUBLISH_AGENT_ACTIVITY_SECRET,
19	readRelayConnection,
20	} from "./config.ts";
21
22	const encode = (value: boolean) => new TextEncoder().encode(String(value));
23
24	/** A failed rollback leaves a setting changed; it is logged, not hidden. */
25	const rollbackFailed = (cause: unknown) =>
26	Effect.logWarning("Could not roll back a T3 Connect preference", { cause });
27
28	const internalError = (message: string) => (cause: unknown) =>
29	Effect.logError(message, { cause }).pipe(
30	Effect.andThen(Effect.fail(new EnvironmentHttpInternalServerError({ message }))),
31	);
32
33	export class CloudPreferences extends Context.Service<
34	CloudPreferences,
35	{
36	/**
37	* Saves this environment's T3 Connect preferences, all or nothing. The
38	* activity setting is saved first. Holding webhooks while offline is decided
39	* by the relay, so the relay is told before the local copy is saved. If
40	* either step fails, the activity setting is put back, and so is the relay
41	* when only the local save failed.
42	*/
43	readonly update: (
44	input: EnvironmentCloudPreferencesRequest,
45	) => Effect.Effect<void, EnvironmentHttpBadRequestError \| EnvironmentHttpInternalServerError>;
46	}
47	>()("t3/cloud/CloudPreferences") {}
48
49	const make = Effect.gen(function* () {
50	const secrets = yield* ServerSecretStore.ServerSecretStore;
51	const environment = yield* ServerEnvironment.ServerEnvironment;
52	const awarenessRelay = yield* AgentAwarenessRelay.AgentAwarenessRelay;
53	const withSecrets = Effect.provideService(ServerSecretStore.ServerSecretStore, secrets);
54
55	const pushHoldWebhooksWhileOffline = Effect.fn("CloudPreferences.pushHoldWebhooksWhileOffline")(
56	function* (holdWebhooksWhileOffline: boolean) {
57	const connection = yield* readRelayConnection.pipe(withSecrets);
58	if (connection === null) {
59	return yield* new EnvironmentHttpBadRequestError({
60	message: "Link this environment to T3 Connect first.",
61	});
62	}
63	const environmentId = yield* environment.getEnvironmentId;
64	const client = yield* makeRelayEnvironmentClient(connection);
65	yield* client.server
66	.updateLinkPreferences({
67	params: { environmentId },
68	payload: { holdWebhooksWhileOffline },
69	})
70	.pipe(
71	Effect.timeout("10 seconds"),
72	Effect.catch(internalError("Could not update T3 Connect webhook settings.")),
73	);
74	},
75	);
76
77	const save = (name: string, value: boolean) =>
78	secrets
79	.set(name, encode(value))
80	.pipe(Effect.catch(internalError("Could not persist environment cloud preferences.")));
81
82	// One update at a time, so two requests can't each leave one setting behind.
83	const updateLock = yield* Semaphore.make(1);
84
85	const update: CloudPreferences["Service"]["update"] = Effect.fn("CloudPreferences.update")(
86	function* (input) {
87	// All or nothing: the activity setting is saved first, before the relay
88	// is told anything, and put back if the hold change then fails. Both
89	// current values are read up front; a failed read stops here, before
90	// anything changes, because a guessed value would be the rollback target.
91	const readCurrent = (name: string) =>
92	secrets
93	.get(name)
94	.pipe(Effect.catch(internalError("Could not read environment cloud preferences.")));
95	const previousActivity = yield* readCurrent(PUBLISH_AGENT_ACTIVITY_SECRET);
96	const previousHold = yield* readCurrent(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET);
97	yield* save(PUBLISH_AGENT_ACTIVITY_SECRET, input.publishAgentActivity);
98	if (input.holdWebhooksWhileOffline !== undefined) {
99	const next = input.holdWebhooksWhileOffline;
100	const previous = Option.match(previousHold, {
101	onNone: () => false,
102	onSome: (bytes) => new TextDecoder().decode(bytes) === "true",
103	});
104	yield* pushHoldWebhooksWhileOffline(next).pipe(
105	Effect.andThen(
106	save(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET, next).pipe(
107	Effect.tapError(() =>
108	previous === next
109	? Effect.void
110	: pushHoldWebhooksWhileOffline(previous).pipe(Effect.catch(rollbackFailed)),
111	),
112	),
113	),
114	Effect.tapError(() =>
115	Option.match(previousActivity, {
116	onNone: () => secrets.remove(PUBLISH_AGENT_ACTIVITY_SECRET),
117	onSome: (bytes) => secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, bytes),
118	}).pipe(Effect.catch(rollbackFailed)),
119	),
120	);
121	}
122	yield* awarenessRelay.requestCatchUp();
123	},
124	updateLock.withPermits(1),
125	);
126
127	return CloudPreferences.of({ update });
128	});
129
130	export const layer = Layer.effect(CloudPreferences, make);

apps/server/src/cloud/config.ts+16-21

...	@@ -6,7 +6,7 @@ import * as Effect from "effect/Effect";	...	@@ -6,7 +6,7 @@ import * as Effect from "effect/Effect";
6	import * as Option from "effect/Option";	6	import * as Option from "effect/Option";
7	import * as Schema from "effect/Schema";	7	import * as Schema from "effect/Schema";
8		8
9	import type * as ServerSecretStore from "../auth/ServerSecretStore.ts";	9	import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
10		10
11	export const CLOUD_MINT_PUBLIC_KEY = "cloud-mint-ed25519-public-key";	11	export const CLOUD_MINT_PUBLIC_KEY = "cloud-mint-ed25519-public-key";
12	export const CLOUD_ENDPOINT_RUNTIME_CONFIG = "cloud-endpoint-runtime-config";	12	export const CLOUD_ENDPOINT_RUNTIME_CONFIG = "cloud-endpoint-runtime-config";
...	@@ -75,11 +75,10 @@ export const readAgentActivityPublishingActive = (	...	@@ -75,11 +75,10 @@ export const readAgentActivityPublishingActive = (
75	);	75	);
76	}).pipe(Effect.orElseSucceed(() => false));	76	}).pipe(Effect.orElseSucceed(() => false));
77		77
78	const readSecretString = (	78	/** A non-empty secret as text, or null when it is missing, empty or unreadable. */
79	secrets: ServerSecretStore.ServerSecretStore["Service"],	79	const readSecretString = (name: string) =>
80	name: string,	80	ServerSecretStore.ServerSecretStore.pipe(
81	): Effect.Effect<string \| null> =>	81	Effect.flatMap((secrets) => secrets.get(name)),
82	secrets.get(name).pipe(
83	Effect.map((bytes) =>	82	Effect.map((bytes) =>
84	Option.isSome(bytes) && bytes.value.length > 0 ? new TextDecoder().decode(bytes.value) : null,	83	Option.isSome(bytes) && bytes.value.length > 0 ? new TextDecoder().decode(bytes.value) : null,
85	),	84	),
...	@@ -87,20 +86,16 @@ const readSecretString = (	...	@@ -87,20 +86,16 @@ const readSecretString = (
87	);	86	);
88		87
89	/** The relay URL and environment credential, or null when not linked to T3 Connect. */	88	/** The relay URL and environment credential, or null when not linked to T3 Connect. */
90	export const readRelayConnection = (secrets: ServerSecretStore.ServerSecretStore["Service"]) =>	89	export const readRelayConnection = Effect.all([
91	Effect.all([	90	readSecretString(RELAY_URL_SECRET),
92	readSecretString(secrets, RELAY_URL_SECRET),	91	readSecretString(RELAY_ENVIRONMENT_CREDENTIAL_SECRET),
93	readSecretString(secrets, RELAY_ENVIRONMENT_CREDENTIAL_SECRET),	92	]).pipe(
94	]).pipe(	93	Effect.map(([url, environmentCredential]) =>
95	Effect.map(([url, environmentCredential]) =>	94	url && environmentCredential ? { url, environmentCredential } : null,
96	url && environmentCredential ? { url, environmentCredential } : null,	95	),
97	),	96	);
98	);
99		97
100	/** Whether this environment opted in to T3 Connect holding webhooks while it is offline. */	98	/** Whether this environment opted in to T3 Connect holding webhooks while it is offline. */
101	export const readHoldWebhooksWhileOffline = (	99	export const readHoldWebhooksWhileOffline = readSecretString(
102	secrets: ServerSecretStore.ServerSecretStore["Service"],	100	HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET,
103	) =>	101	).pipe(Effect.map((value) => value === "true"));
104	readSecretString(secrets, HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET).pipe(
105	Effect.map((value) => value === "true"),
106	);

apps/server/src/cloud/http.ts+16-64

...	@@ -8,7 +8,6 @@ import {	...	@@ -8,7 +8,6 @@ import {
8	EnvironmentCloudRelayConfigResult,	8	EnvironmentCloudRelayConfigResult,
9	EnvironmentHttpApi,	9	EnvironmentHttpApi,
10	EnvironmentHttpBadRequestError,	10	EnvironmentHttpBadRequestError,
11	type EnvironmentCloudPreferencesRequest,
12	EnvironmentHttpConflictError,	11	EnvironmentHttpConflictError,
13	EnvironmentHttpInternalServerError,	12	EnvironmentHttpInternalServerError,
14	EnvironmentHttpUnauthorizedError,	13	EnvironmentHttpUnauthorizedError,
...	@@ -70,7 +69,6 @@ import * as ServerSecretStore from "../auth/ServerSecretStore.ts";	...	@@ -70,7 +69,6 @@ import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
70	import { requireEnvironmentScope } from "../auth/http.ts";	69	import { requireEnvironmentScope } from "../auth/http.ts";
71	import * as ServerConfig from "../config.ts";	70	import * as ServerConfig from "../config.ts";
72	import * as ServerEnvironment from "../environment/ServerEnvironment.ts";	71	import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
73	import { makeRelayEnvironmentClient } from "../relay/relayEnvironmentClient.ts";
74	import * as AgentAwarenessRelay from "../relay/AgentAwarenessRelay.ts";	72	import * as AgentAwarenessRelay from "../relay/AgentAwarenessRelay.ts";
75	import * as ManagedEndpointRuntime from "./ManagedEndpointRuntime.ts";	73	import * as ManagedEndpointRuntime from "./ManagedEndpointRuntime.ts";
76	import {	74	import {
...	@@ -89,8 +87,6 @@ import {	...	@@ -89,8 +87,6 @@ import {
89	encodeConfirmedOriginJson,	87	encodeConfirmedOriginJson,
90	PUBLISH_AGENT_ACTIVITY_SECRET,	88	PUBLISH_AGENT_ACTIVITY_SECRET,
91	HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET,	89	HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET,
92	readHoldWebhooksWhileOffline,
93	readRelayConnection,
94	RELAY_ENVIRONMENT_CREDENTIAL_SECRET,	90	RELAY_ENVIRONMENT_CREDENTIAL_SECRET,
95	RELAY_ISSUER_SECRET,	91	RELAY_ISSUER_SECRET,
96	RELAY_URL_SECRET,	92	RELAY_URL_SECRET,
...	@@ -102,6 +98,7 @@ import {	...	@@ -102,6 +98,7 @@ import {
102	setCliDesiredCloudLink,	98	setCliDesiredCloudLink,
103	} from "./CliState.ts";	99	} from "./CliState.ts";
104	import * as CliTokenManager from "./CliTokenManager.ts";	100	import * as CliTokenManager from "./CliTokenManager.ts";
		101	import * as CloudPreferences from "./CloudPreferences.ts";
105	import { getOrCreateEnvironmentKeyPairFromSecretStore } from "./environmentKeys.ts";	102	import { getOrCreateEnvironmentKeyPairFromSecretStore } from "./environmentKeys.ts";
106	import { traceRelayRequest } from "./traceRelayRequest.ts";	103	import { traceRelayRequest } from "./traceRelayRequest.ts";
107	import { filterRelayResponse, relayRequestError, shouldRetryCloudLink } from "./relayResponse.ts";	104	import { filterRelayResponse, relayRequestError, shouldRetryCloudLink } from "./relayResponse.ts";
...	@@ -1358,65 +1355,6 @@ const cloudUnlinkHandler = Effect.fn("environment.cloud.unlink")(	...	@@ -1358,65 +1355,6 @@ const cloudUnlinkHandler = Effect.fn("environment.cloud.unlink")(
1358	),	1355	),
1359	);	1356	);
1360		1357
1361	const pushHoldWebhooksWhileOffline = Effect.fn("environment.cloud.pushHoldWebhooksWhileOffline")(
1362	function* (dependencies: CloudHttpDependencies, holdWebhooksWhileOffline: boolean) {
1363	const connection = yield* readRelayConnection(dependencies.secrets);
1364	if (connection === null) {
1365	return yield* new EnvironmentHttpBadRequestError({
1366	message: "Link this environment to T3 Connect first.",
1367	});
1368	}
1369	const environmentId = yield* dependencies.environment.getEnvironmentId;
1370	const client = yield* makeRelayEnvironmentClient(connection);
1371	yield* client.server
1372	.updateLinkPreferences({
1373	params: { environmentId },
1374	payload: { holdWebhooksWhileOffline },
1375	})
1376	.pipe(
1377	Effect.timeout("10 seconds"),
1378	Effect.catch(
1379	failEnvironmentCloudInternalError("Could not update T3 Connect webhook settings."),
1380	),
1381	);
1382	},
1383	);
1384
1385	const cloudPreferencesHandler = Effect.fn("environment.cloud.preferences")(
1386	function* (dependencies: CloudHttpDependencies, payload: EnvironmentCloudPreferencesRequest) {
1387	yield* requireEnvironmentScope(AuthRelayWriteScope);
1388	if (payload.holdWebhooksWhileOffline !== undefined) {
1389	// The relay decides whether to hold a request, so it is told first; the
1390	// local copy is only saved once the relay has the same value, and the
1391	// relay is put back if that save fails.
1392	const previous = yield* readHoldWebhooksWhileOffline(dependencies.secrets);
1393	yield* pushHoldWebhooksWhileOffline(dependencies, payload.holdWebhooksWhileOffline);
1394	yield* dependencies.secrets
1395	.set(
1396	HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET,
1397	stringToBytes(String(payload.holdWebhooksWhileOffline)),
1398	)
1399	.pipe(
1400	Effect.tapError(() =>
1401	previous === payload.holdWebhooksWhileOffline
1402	? Effect.void
1403	: pushHoldWebhooksWhileOffline(dependencies, previous).pipe(Effect.ignore),
1404	),
1405	);
1406	}
1407	yield* dependencies.secrets.set(
1408	PUBLISH_AGENT_ACTIVITY_SECRET,
1409	stringToBytes(String(payload.publishAgentActivity)),
1410	);
1411	yield* dependencies.awarenessRelay.requestCatchUp();
1412	return yield* readCloudLinkState(dependencies);
1413	},
1414	Effect.catchIf(
1415	ServerSecretStore.isSecretStoreError,
1416	failEnvironmentCloudInternalError("Could not persist environment cloud preferences."),
1417	),
1418	);
1419
1420	const cloudEnvironmentHealthHandler = Effect.fn("environment.cloud.health")(	1358	const cloudEnvironmentHealthHandler = Effect.fn("environment.cloud.health")(
1421	function* (dependencies: CloudHttpDependencies, request: RelayCloudEnvironmentHealthRequest) {	1359	function* (dependencies: CloudHttpDependencies, request: RelayCloudEnvironmentHealthRequest) {
1422	const cloudMintPublicKey = yield* dependencies.secrets	1360	const cloudMintPublicKey = yield* dependencies.secrets
...	@@ -1661,12 +1599,26 @@ export const connectHttpApiLayer = HttpApiBuilder.group(	...	@@ -1661,12 +1599,26 @@ export const connectHttpApiLayer = HttpApiBuilder.group(
1661	"connect",	1599	"connect",
1662	Effect.fnUntraced(function* (handlers) {	1600	Effect.fnUntraced(function* (handlers) {
1663	const dependencies = yield* cloudHttpDependencies;	1601	const dependencies = yield* cloudHttpDependencies;
		1602	const preferences = yield* CloudPreferences.CloudPreferences;
1664	return handlers	1603	return handlers
1665	.handle("linkProof", ({ payload }) => cloudLinkProofHandler(dependencies, payload))	1604	.handle("linkProof", ({ payload }) => cloudLinkProofHandler(dependencies, payload))
1666	.handle("relayConfig", ({ payload }) => cloudRelayConfigHandler(dependencies, payload))	1605	.handle("relayConfig", ({ payload }) => cloudRelayConfigHandler(dependencies, payload))
1667	.handle("linkState", () => cloudLinkStateHandler(dependencies))	1606	.handle("linkState", () => cloudLinkStateHandler(dependencies))
1668	.handle("unlink", () => cloudUnlinkHandler(dependencies))	1607	.handle("unlink", () => cloudUnlinkHandler(dependencies))
1669	.handle("preferences", ({ payload }) => cloudPreferencesHandler(dependencies, payload))	1608	.handle(
		1609	"preferences",
		1610	Effect.fn("environment.cloud.preferences")(
		1611	function* ({ payload }) {
		1612	yield* requireEnvironmentScope(AuthRelayWriteScope);
		1613	yield* preferences.update(payload);
		1614	return yield* readCloudLinkState(dependencies);
		1615	},
		1616	Effect.catchIf(
		1617	ServerSecretStore.isSecretStoreError,
		1618	failEnvironmentCloudInternalError("Could not read environment cloud preferences."),
		1619	),
		1620	),
		1621	)
1670	.handle("health", ({ payload }) => cloudEnvironmentHealthHandler(dependencies, payload))	1622	.handle("health", ({ payload }) => cloudEnvironmentHealthHandler(dependencies, payload))
1671	.handle("mintCredential", ({ payload }) => cloudMintCredentialHandler(dependencies, payload))	1623	.handle("mintCredential", ({ payload }) => cloudMintCredentialHandler(dependencies, payload))
1672	.handle("t3MintCredential", ({ payload }) =>	1624	.handle("t3MintCredential", ({ payload }) =>

apps/server/src/relay/HeldHooksWaker.ts+2-4

apps/server/src/scheduledTasks/RelayDeliveryProof.ts created+91

...	@@ -0,0 +1,91 @@
1	/**
2	* Whether a webhook request really came through this environment's relay.
3	*
4	* The relay's delivery id, receive time and trace context arrive as plain
5	* headers, and the webhook URL can also be called directly, so they are only
6	* trusted alongside a proof the relay signed with its mint key. A request
7	* without a valid proof is handled as a direct request.
8	*/
9	import { RelayHookDeliveryProofPayload } from "@t3tools/contracts/relay";
10	import {
11	normalizeRelayIssuer,
12	RELAY_HOOK_DELIVERY_HEADER,
13	RELAY_HOOK_DELIVERY_TYP,
14	verifyRelayJwt,
15	} from "@t3tools/shared/relayJwt";
16	import * as Clock from "effect/Clock";
17	import * as Context from "effect/Context";
18	import * as Effect from "effect/Effect";
19	import * as Layer from "effect/Layer";
20	import * as Option from "effect/Option";
21	import * as Schema from "effect/Schema";
22
23	import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
24	import { CLOUD_MINT_PUBLIC_KEY, RELAY_ISSUER_SECRET, RELAY_URL_SECRET } from "../cloud/config.ts";
25	import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
26
27	/** Covers the relay's 24-hour hold plus a margin. */
28	const MAX_PROOF_AGE_SECONDS = 25 * 60 * 60;
29
30	const decodePayload = Schema.decodeUnknownOption(RelayHookDeliveryProofPayload);
31	const text = (bytes: Option.Option<Uint8Array>) =>
32	Option.map(bytes, (value) => new TextDecoder().decode(value));
33
34	export class RelayDeliveryProof extends Context.Service<
35	RelayDeliveryProof,
36	{
37	/** The relay's own delivery id and receive time, when the request proves it came from the relay. */
38	readonly verify: (input: {
39	readonly headers: Readonly<Record<string, string>>;
40	readonly hookId: string;
41	}) => Effect.Effect<
42	Option.Option<{ readonly deliveryId: string; readonly receivedAt: string }>
43	>;
44	}
45	>()("t3/scheduledTasks/RelayDeliveryProof") {}
46
47	const make = Effect.gen(function* () {
48	const secrets = yield* ServerSecretStore.ServerSecretStore;
49	const environment = yield* ServerEnvironment.ServerEnvironment;
50	const verify: RelayDeliveryProof["Service"]["verify"] = (input) =>
51	Effect.gen(function* () {
52	const proof = input.headers[RELAY_HOOK_DELIVERY_HEADER];
53	const deliveryId = input.headers["x-t3-relay-delivery-id"];
54	const receivedAt = input.headers["x-t3-relay-received-at"];
55	if (proof === undefined \|\| deliveryId === undefined \|\| receivedAt === undefined) {
56	return Option.none();
57	}
58	const publicKey = text(yield* secrets.get(CLOUD_MINT_PUBLIC_KEY));
59	const issuer = Option.orElse(text(yield* secrets.get(RELAY_ISSUER_SECRET)), () =>
60	Option.none<string>(),
61	);
62	const relayUrl = text(yield* secrets.get(RELAY_URL_SECRET));
63	const relayIssuer = Option.isSome(issuer) ? issuer : relayUrl;
64	// Not linked to T3 Connect: no relay can be delivering to us.
65	if (Option.isNone(publicKey) \|\| Option.isNone(relayIssuer)) return Option.none();
66	const environmentId = yield* environment.getEnvironmentId;
67	const payload = yield* verifyRelayJwt({
68	publicKey: publicKey.value,
69	token: proof,
70	typ: RELAY_HOOK_DELIVERY_TYP,
71	issuer: normalizeRelayIssuer(relayIssuer.value),
72	audience: `t3-env:${environmentId}`,
73	nowEpochSeconds: Math.floor((yield* Clock.currentTimeMillis) / 1_000),
74	maxTokenAge: MAX_PROOF_AGE_SECONDS,
75	}).pipe(Effect.map(decodePayload), Effect.orElseSucceed(Option.none));
76	// The proof must name exactly this delivery, so it cannot be lifted onto another one.
77	if (
78	Option.isNone(payload) \|\|
79	payload.value.environmentId !== environmentId \|\|
80	payload.value.deliveryId !== deliveryId \|\|
81	payload.value.receivedAt !== receivedAt \|\|
82	payload.value.hookId !== input.hookId
83	) {
84	return Option.none();
85	}
86	return Option.some({ deliveryId, receivedAt });
87	}).pipe(Effect.orElseSucceed(Option.none), Effect.withSpan("webhook.verifyRelayDelivery"));
88	return RelayDeliveryProof.of({ verify });
89	});
90
91	export const layer = Layer.effect(RelayDeliveryProof, make);

apps/server/src/scheduledTasks/relayDeliveryProof.ts deleted-74

...	@@ -1,74 +0,0 @@
1	/**
2	* Whether a webhook request really came through this environment's relay.
3	*
4	* The relay's delivery id, receive time and trace context arrive as plain
5	* headers, and the webhook URL can also be called directly, so they are only
6	* trusted alongside a proof the relay signed with its mint key. A request
7	* without a valid proof is handled as a direct request.
8	*/
9	import { RelayHookDeliveryProofPayload } from "@t3tools/contracts/relay";
10	import {
11	normalizeRelayIssuer,
12	RELAY_HOOK_DELIVERY_HEADER,
13	RELAY_HOOK_DELIVERY_TYP,
14	verifyRelayJwt,
15	} from "@t3tools/shared/relayJwt";
16	import * as Clock from "effect/Clock";
17	import * as Effect from "effect/Effect";
18	import * as Option from "effect/Option";
19	import * as Schema from "effect/Schema";
20
21	import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
22	import { CLOUD_MINT_PUBLIC_KEY, RELAY_ISSUER_SECRET, RELAY_URL_SECRET } from "../cloud/config.ts";
23	import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
24
25	/** Covers the relay's 24-hour hold plus a margin. */
26	const MAX_PROOF_AGE_SECONDS = 25 * 60 * 60;
27
28	const decodePayload = Schema.decodeUnknownOption(RelayHookDeliveryProofPayload);
29	const text = (bytes: Option.Option<Uint8Array>) =>
30	Option.map(bytes, (value) => new TextDecoder().decode(value));
31
32	/** The relay's own delivery id and receive time, when the request proves it came from the relay. */
33	export const makeRelayDeliveryVerifier = Effect.gen(function* () {
34	const secrets = yield* ServerSecretStore.ServerSecretStore;
35	const environment = yield* ServerEnvironment.ServerEnvironment;
36	return (input: { readonly headers: Readonly<Record<string, string>>; readonly hookId: string }) =>
37	Effect.gen(function* () {
38	const proof = input.headers[RELAY_HOOK_DELIVERY_HEADER];
39	const deliveryId = input.headers["x-t3-relay-delivery-id"];
40	const receivedAt = input.headers["x-t3-relay-received-at"];
41	if (proof === undefined \|\| deliveryId === undefined \|\| receivedAt === undefined) {
42	return Option.none();
43	}
44	const publicKey = text(yield* secrets.get(CLOUD_MINT_PUBLIC_KEY));
45	const issuer = Option.orElse(text(yield* secrets.get(RELAY_ISSUER_SECRET)), () =>
46	Option.none<string>(),
47	);
48	const relayUrl = text(yield* secrets.get(RELAY_URL_SECRET));
49	const relayIssuer = Option.isSome(issuer) ? issuer : relayUrl;
50	// Not linked to T3 Connect: no relay can be delivering to us.
51	if (Option.isNone(publicKey) \|\| Option.isNone(relayIssuer)) return Option.none();
52	const environmentId = yield* environment.getEnvironmentId;
53	const payload = yield* verifyRelayJwt({
54	publicKey: publicKey.value,
55	token: proof,
56	typ: RELAY_HOOK_DELIVERY_TYP,
57	issuer: normalizeRelayIssuer(relayIssuer.value),
58	audience: `t3-env:${environmentId}`,
59	nowEpochSeconds: Math.floor((yield* Clock.currentTimeMillis) / 1_000),
60	maxTokenAge: MAX_PROOF_AGE_SECONDS,
61	}).pipe(Effect.map(decodePayload), Effect.orElseSucceed(Option.none));
62	// The proof must name exactly this delivery, so it cannot be lifted onto another one.
63	if (
64	Option.isNone(payload) \|\|
65	payload.value.environmentId !== environmentId \|\|
66	payload.value.deliveryId !== deliveryId \|\|
67	payload.value.receivedAt !== receivedAt \|\|
68	payload.value.hookId !== input.hookId
69	) {
70	return Option.none();
71	}
72	return Option.some({ deliveryId, receivedAt });
73	}).pipe(Effect.orElseSucceed(Option.none), Effect.withSpan("webhook.verifyRelayDelivery"));
74	});

apps/server/src/scheduledTasks/webhookRoute.test.ts+2

apps/server/src/scheduledTasks/webhookRoute.ts+104-112

 @@ -11,7 +11,7 @@ import { withRelayClientTracing } from "@t3tools/shared/relayTracing";
 import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
 import * as Metrics from "../observability/Metrics.ts";
-import { makeRelayDeliveryVerifier } from "./relayDeliveryProof.ts";
+import * as RelayDeliveryProof from "./RelayDeliveryProof.ts";
 import * as ScheduledTaskService from "./ScheduledTaskService.ts";
 /** Largest request body a webhook accepts. The relay enforces the same cap. */
 @@ -23,121 +23,113 @@ const WEBHOOK_OUTCOME_HEADER = "x-t3-hook-outcome";
 const json = (status: number, body: Record<string, string>, outcome: string) =>
   HttpServerResponse.jsonUnsafe(body, { status, headers: { [WEBHOOK_OUTCOME_HEADER]: outcome } });
-/**
- * Handles `/api/hooks/:hookId/:token` for every accepted method. The endpoint
- * is raw so the signature is checked over the exact body bytes; the service
- * checks the token and signature. It is reachable directly, over the managed
- * tunnel, or through the relay's stable `/v1/hooks/:endpointKey/:hookId/:token`
- * URL, where the endpoint key is this environment's managed tunnel key.
- */
-const handleWebhook =
-  (
-    scheduledTasks: ScheduledTaskService.ScheduledTaskService["Service"],
-    verifyRelayDelivery: Effect.Success<typeof makeRelayDeliveryVerifier>,
-  ) =>
-  ({
-    params,
-    request,
-  }: {
-    readonly params: { readonly hookId: string; readonly token: string };
-    readonly request: HttpServerRequest.HttpServerRequest;
-  }) =>
-    Effect.gen(function* () {
-      const contentLength = Number(request.headers["content-length"] ?? "0");
-      if (!Number.isFinite(contentLength) || contentLength > WEBHOOK_MAX_BODY_BYTES) {
-        return json(413, { error: "body_too_large" }, "body_too_large");
-      }
-      // Chunked requests carry no content-length, so the reader itself is capped.
-      const body = yield* request.arrayBuffer.pipe(
-        Effect.map((buffer) => new Uint8Array(buffer)),
-        Effect.provideService(
-          HttpIncomingMessage.MaxBodySize,
-          ByteSize.bytes(WEBHOOK_MAX_BODY_BYTES),
-        ),
-        Effect.option,
-      );
-      // Refused before a task is looked up, so the service never sees them.
-      const tooLarge = (error: string) =>
-        Metrics.increment(Metrics.webhookDeliveriesTotal, {
-          outcome: "body_too_large",
-          source: request.headers["x-t3-relay-delivery-id"] ? "relay" : "direct",
-        }).pipe(Effect.as(json(413, { error }, "body_too_large")));
-      if (Option.isNone(body)) return yield* tooLarge("body_too_large_or_unreadable");
-      if (body.value.byteLength > WEBHOOK_MAX_BODY_BYTES) {
-        return yield* tooLarge("body_too_large");
-      }
-      const headers: Record<string, string> = {};
-      for (const [name, value] of Object.entries(request.headers)) {
-        if (typeof value === "string") headers[name.toLowerCase()] = value;
-      }
-      const queryIndex = request.url.indexOf("?");
-      // The relay's delivery id and receive time count only with its signed
-      // proof; the URL can also be called directly. The receive time matters
-      // for requests the relay held while we were offline.
-      const relay = yield* verifyRelayDelivery({ headers, hookId: params.hookId });
-      const relayDeliveryId = Option.isSome(relay) ? relay.value.deliveryId : undefined;
-      const relayReceivedAt = Option.isSome(relay) ? relay.value.receivedAt : undefined;
-      // A relay delivery joins the relay's trace, and goes to the T3 Connect
-      // tracer with it. Anyone else's traceparent is never trusted.
-      const relayParent = Option.isSome(relay)
-        ? HttpTraceContext.fromHeaders(request.headers)
-        : Option.none<Tracer.ExternalSpan>();
-      const result = yield* scheduledTasks
-        .triggerWebhook({
-          hookId: params.hookId,
-          token: params.token,
-          method: request.method,
-          path: `${ScheduledTaskService.WEBHOOK_ROUTE_PREFIX}/${encodeURIComponent(params.hookId)}`,
-          query: queryIndex === -1 ? "" : request.url.slice(queryIndex + 1),
-          headers,
-          body: body.value,
-          bodyText: new TextDecoder().decode(body.value),
-          ...(relayDeliveryId ? { relayDeliveryId } : {}),
-          ...(relayReceivedAt ? { receivedAt: relayReceivedAt } : {}),
-        })
-        .pipe(
-          Option.isSome(relayParent)
-            ? (effect) =>
-                effect.pipe(Effect.withParentSpan(relayParent.value), withRelayClientTracing)
-            : (effect) => effect,
-          // Defects too, so the sender only ever sees the fixed error body.
-          Effect.catchCause((cause) =>
-            Effect.logWarning("Webhook delivery failed").pipe(
-              Effect.annotateLogs({ hookId: params.hookId }),
-              Effect.andThen(Effect.logDebug("Webhook delivery failure cause", { cause })),
-              Effect.as({ _tag: "error" as const }),
-            ),
-          ),
-        );
-      switch (result._tag) {
-        case "accepted":
-          return json(202, { deliveryId: result.deliveryId }, result.outcome);
-        case "not_found":
-          return json(404, { error: "hook_not_found" }, "not_found");
-        case "rejected_signature":
-          return json(401, { error: "invalid_signature" }, "rejected_signature");
-        case "disabled":
-          return json(409, { error: "hook_disabled" }, "disabled");
-        case "rate_limited":
-          return json(429, { error: "rate_limited" }, result.outcome);
-        case "expired":
-          return json(410, { error: "delivery_too_old" }, "expired");
-        case "error":
-          return json(500, { error: "internal_error" }, "error");
-      }
-    });
 export const webhookHttpApiLayer = HttpApiBuilder.group(
   EnvironmentHttpApi,
   "webhooks",
   Effect.fnUntraced(function* (handlers) {
-    const handler = handleWebhook(
+    const scheduledTasks = yield* ScheduledTaskService.ScheduledTaskService;
-      yield* ScheduledTaskService.ScheduledTaskService,
+    const relayDeliveryProof = yield* RelayDeliveryProof.RelayDeliveryProof;
-      yield* makeRelayDeliveryVerifier,
+    /**
-    );
+     * Handles `/api/hooks/:hookId/:token` for every accepted method. The endpoint
+     * is raw so the signature is checked over the exact body bytes; the service
+     * checks the token and signature. It is reachable directly, over the managed
+     * tunnel, or through the relay's stable `/v1/hooks/:endpointKey/:hookId/:token`
+     * URL, where the endpoint key is this environment's managed tunnel key.
+     */
+    const handler = ({
+      params,
+      request,
+    }: {
+      readonly params: { readonly hookId: string; readonly token: string };
+      readonly request: HttpServerRequest.HttpServerRequest;
+    }) =>
+      Effect.gen(function* () {
+        const contentLength = Number(request.headers["content-length"] ?? "0");
+        if (!Number.isFinite(contentLength) || contentLength > WEBHOOK_MAX_BODY_BYTES) {
+          return json(413, { error: "body_too_large" }, "body_too_large");
+        }
+        // Chunked requests carry no content-length, so the reader itself is capped.
+        const body = yield* request.arrayBuffer.pipe(
+          Effect.map((buffer) => new Uint8Array(buffer)),
+          Effect.provideService(
+            HttpIncomingMessage.MaxBodySize,
+            ByteSize.bytes(WEBHOOK_MAX_BODY_BYTES),
+          ),
+          Effect.option,
+        );
+        // Refused before a task is looked up, so the service never sees them.
+        const tooLarge = (error: string) =>
+          Metrics.increment(Metrics.webhookDeliveriesTotal, {
+            outcome: "body_too_large",
+            source: request.headers["x-t3-relay-delivery-id"] ? "relay" : "direct",
+          }).pipe(Effect.as(json(413, { error }, "body_too_large")));
+        if (Option.isNone(body)) return yield* tooLarge("body_too_large_or_unreadable");
+        if (body.value.byteLength > WEBHOOK_MAX_BODY_BYTES) {
+          return yield* tooLarge("body_too_large");
+        }
+        const headers: Record<string, string> = {};
+        for (const [name, value] of Object.entries(request.headers)) {
+          if (typeof value === "string") headers[name.toLowerCase()] = value;
+        }
+        const queryIndex = request.url.indexOf("?");
+        // The relay's delivery id and receive time count only with its signed
+        // proof; the URL can also be called directly. The receive time matters
+        // for requests the relay held while we were offline.
+        const relay = yield* relayDeliveryProof.verify({ headers, hookId: params.hookId });
+        const relayDeliveryId = Option.isSome(relay) ? relay.value.deliveryId : undefined;
+        const relayReceivedAt = Option.isSome(relay) ? relay.value.receivedAt : undefined;
+        // A relay delivery joins the relay's trace, and goes to the T3 Connect
+        // tracer with it. Anyone else's traceparent is never trusted.
+        const relayParent = Option.isSome(relay)
+          ? HttpTraceContext.fromHeaders(request.headers)
+          : Option.none<Tracer.ExternalSpan>();
+        const result = yield* scheduledTasks
+          .triggerWebhook({
+            hookId: params.hookId,
+            token: params.token,
+            method: request.method,
+            path: `${ScheduledTaskService.WEBHOOK_ROUTE_PREFIX}/${encodeURIComponent(params.hookId)}`,
+            query: queryIndex === -1 ? "" : request.url.slice(queryIndex + 1),
+            headers,
+            body: body.value,
+            bodyText: new TextDecoder().decode(body.value),
+            ...(relayDeliveryId ? { relayDeliveryId } : {}),
+            ...(relayReceivedAt ? { receivedAt: relayReceivedAt } : {}),
+          })
+          .pipe(
+            Option.isSome(relayParent)
+              ? (effect) =>
+                  effect.pipe(Effect.withParentSpan(relayParent.value), withRelayClientTracing)
+              : (effect) => effect,
+            // Defects too, so the sender only ever sees the fixed error body.
+            Effect.catchCause((cause) =>
+              Effect.logWarning("Webhook delivery failed").pipe(
+                Effect.annotateLogs({ hookId: params.hookId }),
+                Effect.andThen(Effect.logDebug("Webhook delivery failure cause", { cause })),
+                Effect.as({ _tag: "error" as const }),
+              ),
+            ),
+          );
+        switch (result._tag) {
+          case "accepted":
+            return json(202, { deliveryId: result.deliveryId }, result.outcome);
+          case "not_found":
+            return json(404, { error: "hook_not_found" }, "not_found");
+          case "rejected_signature":
+            return json(401, { error: "invalid_signature" }, "rejected_signature");
+          case "disabled":
+            return json(409, { error: "hook_disabled" }, "disabled");
+          case "rate_limited":
+            return json(429, { error: "rate_limited" }, result.outcome);
+          case "expired":
+            return json(410, { error: "delivery_too_old" }, "expired");
+          case "error":
+            return json(500, { error: "internal_error" }, "error");
+        }
+      });
     return handlers
       .handleRaw("webhookPost", handler)
       .handleRaw("webhookPut", handler)

apps/server/src/secrets/SecretRequests.ts+30-17

 @@ -12,7 +12,6 @@ import {
   CommandId,
   SecretRef,
   SecretRequestError,
-  type SecretRequestFailureReason,
   type ProjectId,
   type SecretRequestAnswerInput,
   type ThreadId,
 @@ -62,9 +61,6 @@ const StoredSecret = Schema.fromJsonString(
 const encodeStored = Schema.encodeEffect(StoredSecret);
 const decodeStored = Schema.decodeUnknownOption(StoredSecret);
-const fail = (reason: SecretRequestFailureReason, cause?: unknown) =>
-  new SecretRequestError({ reason, ...(cause === undefined ? {} : { cause }) });
 export class SecretRequests extends Context.Service<
   SecretRequests,
   {
 @@ -109,19 +105,23 @@ const make = Effect.gen(function* () {
           turnItemTypes: ["secret_request"],
           messageRoles: [],
         })
-        .pipe(Effect.mapError((cause) => fail("load_failed", cause)));
+        .pipe(
+          Effect.mapError(
+            (cause) => new SecretRequestError({ reason: "load_failed", cause: cause }),
+          ),
+        );
       const item = records.turnItems.find((candidate) => candidate.id === input.turnItemId);
       if (item?.type !== "secret_request" || item.runId === null || item.nodeId === null) {
-        return yield* fail("not_found");
+        return yield* new SecretRequestError({ reason: "not_found" });
       }
       if (item.secretStatus !== "pending") {
-        return yield* fail("already_answered");
+        return yield* new SecretRequestError({ reason: "already_answered" });
       }
       // The agent is waiting inside the run that asked; once it has ended,
       // nobody will ever receive the ref, so a value saved now would be lost.
       const run = records.runs.find((candidate) => candidate.id === item.runId);
       if (run === undefined || ThreadManagementService.isTerminalRunStatus(run.status)) {
-        return yield* fail("agent_stopped");
+        return yield* new SecretRequestError({ reason: "agent_stopped" });
       }
       // Store first: the card only says saved once the value is kept. Create,
       // not set: a second answer racing this one must not replace the value.
 @@ -140,7 +140,9 @@ const make = Effect.gen(function* () {
           )
           .pipe(
             Effect.catchIf(ServerSecretStore.isSecretAlreadyExistsError, () => Effect.void),
-            Effect.mapError((error) => fail("store_failed", error)),
+            Effect.mapError(
+              (error) => new SecretRequestError({ reason: "store_failed", cause: error }),
+            ),
           );
       }
       const secretStatus = input.answer.type === "save" ? "saved" : "declined";
 @@ -158,7 +160,9 @@ const make = Effect.gen(function* () {
           secretStatus,
         })
         .pipe(
-          Effect.mapError((cause) => fail("record_failed", cause)),
+          Effect.mapError(
+            (cause) => new SecretRequestError({ reason: "record_failed", cause: cause }),
+          ),
           // The card still says pending, so the user can save again; the value
           // stored above would make that retry look already answered.
           Effect.tapError(() =>
 @@ -176,11 +180,15 @@ const make = Effect.gen(function* () {
           turnItemTypes: ["secret_request"],
           messageRoles: [],
         })
-        .pipe(Effect.mapError((cause) => fail("record_failed", cause)));
+        .pipe(
+          Effect.mapError(
+            (cause) => new SecretRequestError({ reason: "record_failed", cause: cause }),
+          ),
+        );
       const card = recorded.turnItems.find((candidate) => candidate.id === item.id);
       if (card?.type === "secret_request" && card.secretStatus !== "saved") {
         yield* removeLogged(storeName(refFor(salt, input.threadId, item.id)));
-        return yield* fail("agent_stopped");
+        return yield* new SecretRequestError({ reason: "agent_stopped" });
       }
     }).pipe(Effect.withSpan("SecretRequests.answer"));
 @@ -217,25 +225,30 @@ const make = Effect.gen(function* () {
   const consumeRef = (input: { readonly ref: SecretRef; readonly projectId: ProjectId }) =>
     Effect.gen(function* () {
-      if (!REF_PATTERN.test(input.ref)) return yield* fail("invalid_ref");
+      if (!REF_PATTERN.test(input.ref))
+        return yield* new SecretRequestError({ reason: "invalid_ref" });
       const stored = yield* store
         .get(storeName(input.ref))
-        .pipe(Effect.mapError((cause) => fail("read_failed", cause)));
+        .pipe(
+          Effect.mapError(
+            (cause) => new SecretRequestError({ reason: "read_failed", cause: cause }),
+          ),
+        );
       const decoded = Option.flatMap(stored, (bytes) =>
         decodeStored(new TextDecoder().decode(bytes)),
       );
       if (Option.isNone(decoded) || decoded.value.projectId !== input.projectId) {
-        return yield* fail("ref_unavailable");
+        return yield* new SecretRequestError({ reason: "ref_unavailable" });
       }
       if ((yield* Clock.currentTimeMillis) - decoded.value.savedAt > SECRET_REF_TTL_MS) {
         // The hourly sweep retries a removal that fails here.
         yield* removeLogged(storeName(input.ref));
-        return yield* fail("ref_expired");
+        return yield* new SecretRequestError({ reason: "ref_expired" });
       }
       // One use: the value moves into whatever consumed it. If it cannot be
       // deleted, it is not handed out, so a ref is never used twice.
       if (!(yield* removeLogged(storeName(input.ref)))) {
-        return yield* fail("consume_failed");
+        return yield* new SecretRequestError({ reason: "consume_failed" });
       }
       return decoded.value.value;
     });

apps/server/src/server.ts+4-2

 @@ -119,6 +119,8 @@ import { authHttpApiLayer, environmentAuthenticatedAuthLayer } from "./auth/http
 import * as ReplayMarkers from "./auth/replayMarkers.ts";
 import * as ServerSecretStore from "./auth/ServerSecretStore.ts";
 import { webhookHttpApiLayer } from "./scheduledTasks/webhookRoute.ts";
+import * as RelayDeliveryProof from "./scheduledTasks/RelayDeliveryProof.ts";
+import * as CloudPreferences from "./cloud/CloudPreferences.ts";
 import * as HeldHooksWaker from "./relay/HeldHooksWaker.ts";
 import {
   relayHookBaseUrl,
 @@ -679,12 +681,12 @@ const makeRoutesLayer = Layer.mergeAll(
   Layer.mergeAll(
     HttpApiBuilder.layer(EnvironmentHttpApi).pipe(
       Layer.provide(authHttpApiLayer),
-      Layer.provide(connectHttpApiLayer),
+      Layer.provide(connectHttpApiLayer.pipe(Layer.provide(CloudPreferences.layer))),
       Layer.provide(orchestrationHttpApiLayer),
       Layer.provide(pullRequestHttpApiLayer),
       Layer.provide(projectHttpApiLayer),
       Layer.provide(serverEnvironmentHttpApiLayer),
-      Layer.provide(webhookHttpApiLayer),
+      Layer.provide(webhookHttpApiLayer.pipe(Layer.provide(RelayDeliveryProof.layer))),
       Layer.provide(environmentAuthenticatedAuthLayer),
     ),
     otlpTracesProxyRouteLayer,

infra/relay/src/hooks/HeldHooks.test.ts created+157

...	@@ -0,0 +1,157 @@
1	import { describe, expect, it } from "@effect/vitest";
2	import * as Effect from "effect/Effect";
3	import * as Layer from "effect/Layer";
4	import * as Redacted from "effect/Redacted";
5
6	import * as RelayConfiguration from "../Config.ts";
7	import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts";
8	import * as ManagedEndpointAllocations from "../environments/ManagedEndpointAllocations.ts";
9	import * as HeldHooks from "./HeldHooks.ts";
10	import * as HookInbox from "./HookInbox.ts";
11
12	const settings = {
13	managedEndpointBaseDomain: "example.test",
14	managedEndpointNamespace: "dev",
15	cloudMintPrivateKey: Redacted.make("unused"),
16	} as RelayConfiguration.RelayConfiguration["Service"];
17
18	const environmentId = "env-hook";
19	const ownKey = "0123456789abcdef";
20	const otherKey = "fedcba9876543210";
21
22	const allocationFor = (
23	userId: string,
24	key: string,
25	ready: boolean,
26	): ManagedEndpointAllocations.ManagedEndpointAllocation => ({
27	userId,
28	environmentId,
29	hostname: `${key}.example.test`,
30	tunnelId: `tunnel-${key}`,
31	tunnelName: `t3coderelay-managedendpoint-dev-${key}`,
32	dnsRecordId: "dns-record-id",
33	readyAt: ready ? "2026-05-25T00:00:00.000Z" : null,
34	origin: { localHttpHost: "127.0.0.1", localHttpPort: 3773 },
35	updatedAt: "2026-05-25T00:00:00.000Z",
36	generation: 1,
37	});
38
39	const linkFor = (userId: string, key: string, environmentPublicKey: string) => ({
40	userId,
41	environmentId: environmentId as never,
42	label: "Hook env",
43	endpoint: {
44	httpBaseUrl: `https://${key}.example.test/`,
45	wsBaseUrl: `wss://${key}.example.test/ws`,
46	providerKind: "cloudflare_tunnel" as const,
47	},
48	environmentPublicKey,
49	linkedAt: "2026-05-25T00:00:00.000Z",
50	holdWebhooksWhileOffline: true,
51	});
52
53	/** This environment's own link (user_1) and another account's link of the same environment. */
54	const withService = <A, E>(
55	options: { readonly ownReady?: boolean; readonly pendingKeys?: ReadonlyArray<string> },
56	body: (input: {
57	readonly heldHooks: HeldHooks.HeldHooks["Service"];
58	readonly cleared: Array<string>;
59	readonly woken: Array<string>;
60	}) => Effect.Effect<A, E>,
61	) => {
62	const cleared: Array<string> = [];
63	const woken: Array<string> = [];
64	const links = [linkFor("user_1", ownKey, "own-key"), linkFor("user_2", otherKey, "other-key")];
65	const allocations = [
66	allocationFor("user_1", ownKey, options.ownReady ?? true),
67	allocationFor("user_2", otherKey, true),
68	];
69	const dependencies = Layer.mergeAll(
70	Layer.succeed(RelayConfiguration.RelayConfiguration, settings),
71	Layer.mock(EnvironmentLinks.EnvironmentLinks, {
72	findActiveManagedForEnvironment: (input) =>
73	Effect.succeed(
74	links.filter(
75	(link) =>
76	link.environmentId === input.environmentId &&
77	(input.userId === undefined \|\| link.userId === input.userId) &&
78	(input.environmentPublicKey === undefined \|\|
79	link.environmentPublicKey === input.environmentPublicKey),
80	),
81	),
82	setHoldWebhooksWhileOffline: () => Effect.void,
83	}),
84	Layer.mock(ManagedEndpointAllocations.ManagedEndpointAllocations, {
85	get: (input) =>
86	Effect.succeed(
87	allocations.find((allocation) => allocation.userId === input.userId) ?? null,
88	),
89	getByTunnelName: (tunnelName) =>
90	Effect.succeed(
91	allocations.find((allocation) => allocation.tunnelName === tunnelName) ?? null,
92	),
93	}),
94	Layer.mock(HookInbox.HookInbox, {
95	clear: ({ endpointKey }) => Effect.sync(() => void cleared.push(endpointKey)),
96	wake: ({ endpointKey }) =>
97	Effect.sync(() => {
98	woken.push(endpointKey);
99	return (options.pendingKeys ?? []).includes(endpointKey);
100	}),
101	}),
102	);
103	return Effect.gen(function* () {
104	const heldHooks = yield* HeldHooks.HeldHooks;
105	return yield* body({ heldHooks, cleared, woken });
106	}).pipe(Effect.provide(HeldHooks.layer.pipe(Layer.provide(dependencies))));
107	};
108
109	describe("HeldHooks", () => {
110	it.effect("opting out clears only the caller's own endpoints", () =>
111	withService({}, ({ heldHooks, cleared }) =>
112	Effect.gen(function* () {
113	yield* heldHooks.setHoldWhileOffline({
114	environmentId,
115	environmentPublicKey: "own-key",
116	holdWebhooksWhileOffline: false,
117	});
118	expect(cleared).toEqual([ownKey]);
119	}),
120	),
121	);
122
123	it.effect("opting in keeps what is held", () =>
124	withService({}, ({ heldHooks, cleared }) =>
125	Effect.gen(function* () {
126	yield* heldHooks.setHoldWhileOffline({
127	environmentId,
128	environmentPublicKey: "own-key",
129	holdWebhooksWhileOffline: true,
130	});
131	expect(cleared).toEqual([]);
132	}),
133	),
134	);
135
136	it.effect("wakes the caller's ready endpoints and reports whether anything was waiting", () =>
137	withService({ pendingKeys: [ownKey] }, ({ heldHooks, woken }) =>
138	Effect.gen(function* () {
139	expect(yield* heldHooks.wake({ environmentId, environmentPublicKey: "own-key" })).toBe(
140	true,
141	);
142	expect(woken).toEqual([ownKey]);
143	}),
144	),
145	);
146
147	it.effect("does not wake an endpoint that is not ready", () =>
148	withService({ ownReady: false, pendingKeys: [ownKey] }, ({ heldHooks, woken }) =>
149	Effect.gen(function* () {
150	expect(yield* heldHooks.wake({ environmentId, environmentPublicKey: "own-key" })).toBe(
151	false,
152	);
153	expect(woken).toEqual([]);
154	}),
155	),
156	);
157	});

infra/relay/src/hooks/HeldHooks.ts created+143

...	@@ -0,0 +1,143 @@
1	import * as Context from "effect/Context";
2	import * as Effect from "effect/Effect";
3	import * as Layer from "effect/Layer";
4	import * as Result from "effect/Result";
5
6	import * as RelayConfiguration from "../Config.ts";
7	import {
8	MANAGED_ENDPOINT_KEY_PATTERN,
9	managedEndpointTunnelNameForKey,
10	} from "../deploymentConfig.ts";
11	import { validateManagedEndpoint } from "../environments/EnvironmentConnector.ts";
12	import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts";
13	import * as ManagedEndpointAllocations from "../environments/ManagedEndpointAllocations.ts";
14	import * as HookInbox from "./HookInbox.ts";
15
16	/** The endpoint key of an environment's own managed endpoint, for authenticated callers. */
17	export const endpointKeyForTunnelName = (namespace: string, tunnelName: string): string \| null => {
18	const prefix = managedEndpointTunnelNameForKey(namespace, "");
19	const key = tunnelName.startsWith(prefix) ? tunnelName.slice(prefix.length) : "";
20	return MANAGED_ENDPOINT_KEY_PATTERN.test(key) ? key : null;
21	};
22
23	type PersistenceError =
24	\| EnvironmentLinks.EnvironmentLinkEnvironmentLookupPersistenceError
25	\| ManagedEndpointAllocations.ManagedEndpointAllocationPersistenceError;
26
27	interface HookEndpoint {
28	readonly httpBaseUrl: string;
29	readonly environmentId: string;
30	readonly holdWhileOffline: boolean;
31	}
32
33	export class HeldHooks extends Context.Service<
34	HeldHooks,
35	{
36	/**
37	* The ready managed endpoint a webhook URL's endpoint key names, with whether
38	* its link opted in to holding webhooks while offline. The key is the tunnel
39	* name's hash of user and environment, so it names exactly one allocation and
40	* at most one active link; nobody else can link their way onto it.
41	*/
42	readonly resolveEndpoint: (
43	endpointKey: string,
44	) => Effect.Effect<HookEndpoint \| null, PersistenceError>;
45	/**
46	* Saves an environment's opt-in. Opting out also drops what is already held,
47	* rather than delivering it later to an environment that said it does not
48	* want it. Only the links the caller's environment key proved are touched.
49	*/
50	readonly setHoldWhileOffline: (input: {
51	readonly environmentId: string;
52	readonly environmentPublicKey: string;
53	readonly holdWebhooksWhileOffline: boolean;
54	}) => Effect.Effect<void, PersistenceError \| HookInbox.HookInboxError>;
55	/** Delivers what the environment's endpoints hold now; true when anything was waiting. */
56	readonly wake: (input: {
57	readonly environmentId: string;
58	readonly environmentPublicKey: string;
59	}) => Effect.Effect<boolean, PersistenceError \| HookInbox.HookInboxError>;
60	}
61	>()("t3code-relay/hooks/HeldHooks") {}
62
63	const make = Effect.gen(function* () {
64	const links = yield* EnvironmentLinks.EnvironmentLinks;
65	const allocations = yield* ManagedEndpointAllocations.ManagedEndpointAllocations;
66	const settings = yield* RelayConfiguration.RelayConfiguration;
67	const inbox = yield* HookInbox.HookInbox;
68
69	const resolveEndpoint: HeldHooks["Service"]["resolveEndpoint"] = Effect.fn(
70	"relay.hooks.resolve_endpoint",
71	)(function* (endpointKey) {
72	if (!settings.managedEndpointNamespace) return null;
73	const allocation = yield* allocations.getByTunnelName(
74	managedEndpointTunnelNameForKey(settings.managedEndpointNamespace, endpointKey),
75	);
76	if (allocation === null) return null;
77	const [link] = yield* links.findActiveManagedForEnvironment({
78	environmentId: allocation.environmentId,
79	userId: allocation.userId,
80	});
81	if (!link) return null;
82	const result = validateManagedEndpoint({
83	link,
84	allocation,
85	baseDomain: settings.managedEndpointBaseDomain,
86	});
87	if (Result.isFailure(result)) return null;
88	return {
89	...result.success,
90	environmentId: allocation.environmentId,
91	holdWhileOffline: link.holdWebhooksWhileOffline,
92	};
93	});
94
95	/**
96	* Endpoint keys of the managed links the calling environment key proved.
97	* The environment id alone would also match other accounts' links of it.
98	*/
99	const ownEndpointKeys = Effect.fn("relay.hooks.own_endpoint_keys")(function* (input: {
100	readonly environmentId: string;
101	readonly environmentPublicKey: string;
102	}) {
103	const namespace = settings.managedEndpointNamespace;
104	if (!namespace) return [];
105	const ownLinks = yield* links.findActiveManagedForEnvironment(input);
106	const keys: Array<string> = [];
107	for (const link of ownLinks) {
108	const allocation = yield* allocations.get({
109	userId: link.userId,
110	environmentId: input.environmentId,
111	});
112	const key = allocation ? endpointKeyForTunnelName(namespace, allocation.tunnelName) : null;
113	if (key !== null) keys.push(key);
114	}
115	return keys;
116	});
117
118	const setHoldWhileOffline: HeldHooks["Service"]["setHoldWhileOffline"] = Effect.fn(
119	"relay.hooks.set_hold_while_offline",
120	)(function* (input) {
121	yield* links.setHoldWebhooksWhileOffline(input);
122	if (input.holdWebhooksWhileOffline) return;
123	for (const endpointKey of yield* ownEndpointKeys(input)) {
124	yield* inbox.clear({ endpointKey });
125	}
126	});
127
128	const wake: HeldHooks["Service"]["wake"] = Effect.fn("relay.hooks.wake")(function* (input) {
129	let pending = false;
130	for (const endpointKey of yield* ownEndpointKeys(input)) {
131	const endpoint = yield* resolveEndpoint(endpointKey);
132	if (endpoint === null) continue;
133	if (yield* inbox.wake({ endpointKey, baseUrl: endpoint.httpBaseUrl })) {
134	pending = true;
135	}
136	}
137	return pending;
138	});
139
140	return HeldHooks.of({ resolveEndpoint, setHoldWhileOffline, wake });
141	});
142
143	export const layer = Layer.effect(HeldHooks, make);

infra/relay/src/hooks/HookForwarder.test.ts+2

...	@@ -37,6 +37,7 @@ import {	...	@@ -37,6 +37,7 @@ import {
37	} from "../http/Api.ts";	37	} from "../http/Api.ts";
38	import * as HookForwarder from "./HookForwarder.ts";	38	import * as HookForwarder from "./HookForwarder.ts";
39	import { RELAY_HOOK_DELIVERY_TYP, verifyRelayJwt } from "@t3tools/shared/relayJwt";	39	import { RELAY_HOOK_DELIVERY_TYP, verifyRelayJwt } from "@t3tools/shared/relayJwt";
		40	import * as HeldHooks from "./HeldHooks.ts";
40	import * as HookInbox from "./HookInbox.ts";	41	import * as HookInbox from "./HookInbox.ts";
41	import type { HeldHook } from "./HookInboxStore.ts";	42	import type { HeldHook } from "./HookInboxStore.ts";
42	import { RELAY_HOOK_UPSTREAM_TIMEOUT_MS } from "./upstream.ts";	43	import { RELAY_HOOK_UPSTREAM_TIMEOUT_MS } from "./upstream.ts";
...	@@ -114,6 +115,7 @@ function makeHarness(options: Harness = {}) {	...	@@ -114,6 +115,7 @@ function makeHarness(options: Harness = {}) {
114	),	115	),
115	));	116	));
116	const forwarderLayer = HookForwarder.layer.pipe(	117	const forwarderLayer = HookForwarder.layer.pipe(
		118	Layer.provideMerge(HeldHooks.layer),
117	Layer.provide(	119	Layer.provide(
118	Layer.mergeAll(	120	Layer.mergeAll(
119	Layer.succeed(RelayConfiguration.RelayConfiguration, settings),	121	Layer.succeed(RelayConfiguration.RelayConfiguration, settings),

infra/relay/src/hooks/HookForwarder.ts+31-84

...	@@ -23,13 +23,8 @@ import {	...	@@ -23,13 +23,8 @@ import {
23	} from "@t3tools/shared/relayJwt";	23	} from "@t3tools/shared/relayJwt";
24		24
25	import * as RelayConfiguration from "../Config.ts";	25	import * as RelayConfiguration from "../Config.ts";
26	import {	26	import { MANAGED_ENDPOINT_KEY_PATTERN } from "../deploymentConfig.ts";
27	MANAGED_ENDPOINT_KEY_PATTERN,	27	import * as HeldHooks from "./HeldHooks.ts";
28	managedEndpointTunnelNameForKey,
29	} from "../deploymentConfig.ts";
30	import { validateManagedEndpoint } from "../environments/EnvironmentConnector.ts";
31	import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts";
32	import * as ManagedEndpointAllocations from "../environments/ManagedEndpointAllocations.ts";
33	import * as HookInbox from "./HookInbox.ts";	28	import * as HookInbox from "./HookInbox.ts";
34	import { sendUpstream, TUNNEL_OFFLINE_STATUS } from "./upstream.ts";	29	import { sendUpstream, TUNNEL_OFFLINE_STATUS } from "./upstream.ts";
35		30
...	@@ -143,34 +138,6 @@ const hookNotFound = () => errorResponse(404, "hook_not_found");	...	@@ -143,34 +138,6 @@ const hookNotFound = () => errorResponse(404, "hook_not_found");
143	/** Longer than the inbox holds a request, so a held delivery's proof still verifies. */	138	/** Longer than the inbox holds a request, so a held delivery's proof still verifies. */
144	const DELIVERY_PROOF_LIFETIME_SECONDS = 25 * 60 * 60;	139	const DELIVERY_PROOF_LIFETIME_SECONDS = 25 * 60 * 60;
145		140
146	const signDeliveryProof = (input: {
147	readonly settings: RelayConfiguration.RelayConfiguration["Service"];
148	readonly environmentId: string;
149	readonly deliveryId: string;
150	readonly receivedAt: string;
151	readonly hookId: string;
152	readonly jti: string;
153	}) =>
154	Effect.gen(function* () {
155	const now = Math.floor((yield* Clock.currentTimeMillis) / 1_000);
156	return yield* signRelayJwt({
157	privateKey: Redacted.value(input.settings.cloudMintPrivateKey),
158	typ: RELAY_HOOK_DELIVERY_TYP,
159	payload: {
160	iss: normalizeRelayIssuer(input.settings.relayIssuer),
161	aud: `t3-env:${input.environmentId}`,
162	sub: input.environmentId,
163	jti: input.jti,
164	iat: now,
165	exp: now + DELIVERY_PROOF_LIFETIME_SECONDS,
166	environmentId: EnvironmentId.make(input.environmentId),
167	deliveryId: input.deliveryId,
168	receivedAt: input.receivedAt,
169	hookId: input.hookId,
170	} satisfies RelayHookDeliveryProofPayload,
171	});
172	}).pipe(Effect.orDie);
173
174	/** Methods a webhook can arrive with; HEAD reaches the GET route and is refused. */	141	/** Methods a webhook can arrive with; HEAD reaches the GET route and is refused. */
175	const FORWARDED_METHODS = new Set(["GET", "POST", "PUT", "PATCH"]);	142	const FORWARDED_METHODS = new Set(["GET", "POST", "PUT", "PATCH"]);
176		143
...	@@ -266,57 +233,41 @@ const readCappedBody = (request: HttpServerRequest.HttpServerRequest) =>	...	@@ -266,57 +233,41 @@ const readCappedBody = (request: HttpServerRequest.HttpServerRequest) =>
266	);	233	);
267	});	234	});
268		235
269	/**
270	* The ready managed endpoint a webhook URL's endpoint key names, with whether
271	* its link opted in to holding webhooks while offline. The key is the tunnel
272	* name's hash of user and environment, so it names exactly one allocation and
273	* at most one active link; nobody else can link their way onto it.
274	*/
275	export const resolveHookEndpoint = Effect.fn("relay.hooks.resolve_endpoint")(function* (
276	endpointKey: string,
277	) {
278	const links = yield* EnvironmentLinks.EnvironmentLinks;
279	const allocations = yield* ManagedEndpointAllocations.ManagedEndpointAllocations;
280	const settings = yield* RelayConfiguration.RelayConfiguration;
281	if (!settings.managedEndpointNamespace) return null;
282	const allocation = yield* allocations.getByTunnelName(
283	managedEndpointTunnelNameForKey(settings.managedEndpointNamespace, endpointKey),
284	);
285	if (allocation === null) return null;
286	const [link] = yield* links.findActiveManagedForEnvironment({
287	environmentId: allocation.environmentId,
288	userId: allocation.userId,
289	});
290	if (!link) return null;
291	const result = validateManagedEndpoint({
292	link,
293	allocation,
294	baseDomain: settings.managedEndpointBaseDomain,
295	});
296	if (Result.isFailure(result)) return null;
297	return {
298	...result.success,
299	environmentId: allocation.environmentId,
300	holdWhileOffline: link.holdWebhooksWhileOffline,
301	};
302	});
303
304	/** The endpoint key of an environment's own managed endpoint, for authenticated callers. */
305	export const endpointKeyForTunnelName = (namespace: string, tunnelName: string): string \| null => {
306	const prefix = managedEndpointTunnelNameForKey(namespace, "");
307	const key = tunnelName.startsWith(prefix) ? tunnelName.slice(prefix.length) : "";
308	return MANAGED_ENDPOINT_KEY_PATTERN.test(key) ? key : null;
309	};
310
311	const make = Effect.gen(function* () {	236	const make = Effect.gen(function* () {
312	const links = yield* EnvironmentLinks.EnvironmentLinks;	237	const heldHooks = yield* HeldHooks.HeldHooks;
313	const allocations = yield* ManagedEndpointAllocations.ManagedEndpointAllocations;
314	const settings = yield* RelayConfiguration.RelayConfiguration;	238	const settings = yield* RelayConfiguration.RelayConfiguration;
315	const httpClient = yield* HttpClient.HttpClient;	239	const httpClient = yield* HttpClient.HttpClient;
316	const rateLimiter = yield* HookRateLimiter;	240	const rateLimiter = yield* HookRateLimiter;
317	const inbox = yield* HookInbox.HookInbox;	241	const inbox = yield* HookInbox.HookInbox;
318	const crypto = yield* Crypto.Crypto;	242	const crypto = yield* Crypto.Crypto;
319		243
		244	const signDeliveryProof = (input: {
		245	readonly environmentId: string;
		246	readonly deliveryId: string;
		247	readonly receivedAt: string;
		248	readonly hookId: string;
		249	readonly jti: string;
		250	}) =>
		251	Effect.gen(function* () {
		252	const now = Math.floor((yield* Clock.currentTimeMillis) / 1_000);
		253	return yield* signRelayJwt({
		254	privateKey: Redacted.value(settings.cloudMintPrivateKey),
		255	typ: RELAY_HOOK_DELIVERY_TYP,
		256	payload: {
		257	iss: normalizeRelayIssuer(settings.relayIssuer),
		258	aud: `t3-env:${input.environmentId}`,
		259	sub: input.environmentId,
		260	jti: input.jti,
		261	iat: now,
		262	exp: now + DELIVERY_PROOF_LIFETIME_SECONDS,
		263	environmentId: EnvironmentId.make(input.environmentId),
		264	deliveryId: input.deliveryId,
		265	receivedAt: input.receivedAt,
		266	hookId: input.hookId,
		267	} satisfies RelayHookDeliveryProofPayload,
		268	});
		269	}).pipe(Effect.orDie);
		270
320	const handle = Effect.fn("relay.hooks.forward")(function* (	271	const handle = Effect.fn("relay.hooks.forward")(function* (
321	request: HttpServerRequest.HttpServerRequest,	272	request: HttpServerRequest.HttpServerRequest,
322	) {	273	) {
...	@@ -357,10 +308,7 @@ const make = Effect.gen(function* () {	...	@@ -357,10 +308,7 @@ const make = Effect.gen(function* () {
357	return errorResponse(413, "payload_too_large");	308	return errorResponse(413, "payload_too_large");
358	}	309	}
359		310
360	const endpoint = yield* resolveHookEndpoint(parsed.endpointKey).pipe(	311	const endpoint = yield* heldHooks.resolveEndpoint(parsed.endpointKey).pipe(
361	Effect.provideService(EnvironmentLinks.EnvironmentLinks, links),
362	Effect.provideService(ManagedEndpointAllocations.ManagedEndpointAllocations, allocations),
363	Effect.provideService(RelayConfiguration.RelayConfiguration, settings),
364	Effect.catch((error) =>	312	Effect.catch((error) =>
365	Effect.logWarning("Failed to resolve hook endpoint", {	313	Effect.logWarning("Failed to resolve hook endpoint", {
366	endpointKey: parsed.endpointKey,	314	endpointKey: parsed.endpointKey,
...	@@ -398,7 +346,6 @@ const make = Effect.gen(function* () {	...	@@ -398,7 +346,6 @@ const make = Effect.gen(function* () {
398	// trace context came from the relay. Signed once here and stored with a	346	// trace context came from the relay. Signed once here and stored with a
399	// held request, so the inbox never needs the signing key.	347	// held request, so the inbox never needs the signing key.
400	const proof = yield* signDeliveryProof({	348	const proof = yield* signDeliveryProof({
401	settings,
402	environmentId: endpoint.environmentId,	349	environmentId: endpoint.environmentId,
403	deliveryId,	350	deliveryId,
404	receivedAt,	351	receivedAt,

infra/relay/src/http/Api.test.ts+2-8

 @@ -61,6 +61,7 @@ import {
 import * as RelayConfiguration from "../Config.ts";
 import * as RelayDb from "../db.ts";
 import * as EnvironmentCredentials from "../environments/EnvironmentCredentials.ts";
+import * as HeldHooks from "../hooks/HeldHooks.ts";
 import * as HookInbox from "../hooks/HookInbox.ts";
 import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts";
 import * as ManagedEndpointAllocations from "../environments/ManagedEndpointAllocations.ts";
 @@ -1269,14 +1270,7 @@ describe("relay routing fallback", () => {
                 Layer.mock(ManagedEndpointProvider.ManagedEndpointProvider, {}),
               ),
             ),
-            Layer.provide([
+            Layer.provide([publisher, signatures, Layer.mock(HeldHooks.HeldHooks, {})]),
-              publisher,
-              signatures,
-              Layer.mock(EnvironmentLinks.EnvironmentLinks, {}),
-              Layer.mock(HookInbox.HookInbox, {}),
-              Layer.mock(ManagedEndpointAllocations.ManagedEndpointAllocations, {}),
-              Layer.succeed(RelayConfiguration.RelayConfiguration, relaySettings),
-            ]),
           ),
         ),
         Layer.provide(auth),

infra/relay/src/http/Api.ts+9-51

 @@ -67,6 +67,7 @@ import * as RelayTokens from "../auth/RelayTokens.ts";
 import * as EnvironmentCredentials from "../environments/EnvironmentCredentials.ts";
 import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts";
 import * as HookForwarder from "../hooks/HookForwarder.ts";
+import * as HeldHooks from "../hooks/HeldHooks.ts";
 import * as HookInbox from "../hooks/HookInbox.ts";
 import * as LiveActivities from "../agentActivity/LiveActivities.ts";
 import * as RelayConfiguration from "../Config.ts";
 @@ -554,7 +555,7 @@ export const unlinkEnvironmentRecord = Effect.fn("relay.api.client.unlinkEnviron
     // is already gone, and its inbox drops anything left after its TTL.
     const endpointKey =
       deprovisionTarget && input.managedEndpointNamespace
-        ? HookForwarder.endpointKeyForTunnelName(
+        ? HeldHooks.endpointKeyForTunnelName(
             input.managedEndpointNamespace,
             deprovisionTarget.tunnelName,
           )
 @@ -1176,10 +1177,7 @@ export const serverApi = HttpApiBuilder.group(
   Effect.fnUntraced(function* (handlers) {
     const publisher = yield* AgentActivityPublisher.AgentActivityPublisher;
     const publishSignatures = yield* EnvironmentPublishSignatures.EnvironmentPublishSignatures;
-    const links = yield* EnvironmentLinks.EnvironmentLinks;
+    const heldHooks = yield* HeldHooks.HeldHooks;
-    const inbox = yield* HookInbox.HookInbox;
-    const allocations = yield* ManagedEndpointAllocations.ManagedEndpointAllocations;
-    const settings = yield* RelayConfiguration.RelayConfiguration;
     const requireOwnEnvironment = (environmentId: string) =>
       Effect.gen(function* () {
         const principal = yield* RelayEnvironmentPrincipal;
 @@ -1187,29 +1185,6 @@ export const serverApi = HttpApiBuilder.group(
           return yield* new HttpApiError.Unauthorized({});
         }
       });
-    /**
-     * Endpoint keys of the managed links the calling environment key proved.
-     * The environment id alone would also match other accounts' links of it.
-     */
-    const ownEndpointKeys = (environmentId: string) =>
-      Effect.gen(function* () {
-        const principal = yield* RelayEnvironmentPrincipal;
-        const namespace = settings.managedEndpointNamespace;
-        if (!namespace) return [];
-        const ownLinks = yield* links.findActiveManagedForEnvironment({
-          environmentId,
-          environmentPublicKey: principal.environmentPublicKey,
-        });
-        const keys: Array<string> = [];
-        for (const link of ownLinks) {
-          const allocation = yield* allocations.get({ userId: link.userId, environmentId });
-          const key = allocation
-            ? HookForwarder.endpointKeyForTunnelName(namespace, allocation.tunnelName)
-            : null;
-          if (key !== null) keys.push(key);
-        }
-        return keys;
-      });
     const activityHandlers = handlers.handle(
       "publishAgentActivity",
       Effect.fn("relay.api.server.publishAgentActivity")(
 @@ -1425,18 +1400,11 @@ export const serverApi = HttpApiBuilder.group(
         Effect.fn("relay.api.server.updateLinkPreferences")(function* ({ params, payload }) {
           yield* requireOwnEnvironment(params.environmentId);
           const principal = yield* RelayEnvironmentPrincipal;
-          yield* links.setHoldWebhooksWhileOffline({
+          yield* heldHooks.setHoldWhileOffline({
             environmentId: params.environmentId,
             environmentPublicKey: principal.environmentPublicKey,
             holdWebhooksWhileOffline: payload.holdWebhooksWhileOffline,
           });
-          // Opting out also drops what is already held, rather than delivering
-          // it later to an environment that said it does not want it.
-          if (!payload.holdWebhooksWhileOffline) {
-            for (const endpointKey of yield* ownEndpointKeys(params.environmentId)) {
-              yield* inbox.clear({ endpointKey });
-            }
-          }
           return payload;
         }, mapRelayCommonApiErrors("not_authorized")),
       )
 @@ -1444,21 +1412,11 @@ export const serverApi = HttpApiBuilder.group(
         "wakeHeldHooks",
         Effect.fn("relay.api.server.wakeHeldHooks")(function* ({ params }) {
           yield* requireOwnEnvironment(params.environmentId);
-          let pending = false;
+          const principal = yield* RelayEnvironmentPrincipal;
-          for (const endpointKey of yield* ownEndpointKeys(params.environmentId)) {
+          const pending = yield* heldHooks.wake({
-            const endpoint = yield* HookForwarder.resolveHookEndpoint(endpointKey).pipe(
+            environmentId: params.environmentId,
-              Effect.provideService(EnvironmentLinks.EnvironmentLinks, links),
+            environmentPublicKey: principal.environmentPublicKey,
-              Effect.provideService(
+          });
-                ManagedEndpointAllocations.ManagedEndpointAllocations,
-                allocations,
-              ),
-              Effect.provideService(RelayConfiguration.RelayConfiguration, settings),
-            );
-            if (endpoint === null) continue;
-            if (yield* inbox.wake({ endpointKey, baseUrl: endpoint.httpBaseUrl })) {
-              pending = true;
-            }
-          }
           return { pending };
         }, mapRelayCommonApiErrors("not_authorized")),
       );

infra/relay/src/worker.ts+6-1

 @@ -76,6 +76,7 @@ import * as ManagedEndpointReaper from "./environments/ManagedEndpointReaper.ts"
 import * as ManagedTunnelLimits from "./environments/ManagedTunnelLimits.ts";
 import * as MobileRegistrations from "./agentActivity/MobileRegistrations.ts";
 import * as HookForwarder from "./hooks/HookForwarder.ts";
+import * as HeldHooks from "./hooks/HeldHooks.ts";
 import * as HookInbox from "./hooks/HookInbox.ts";
 import { HookInboxObject, HookInboxObjectLive } from "./hooks/HookInboxObject.ts";
 @@ -273,7 +274,11 @@ export const ApiLive = Api.make(
       Layer.provideMerge(EnvironmentConnector.layer),
       Layer.provideMerge(EnvironmentLinker.layer),
       Layer.provideMerge(
-        Layer.merge(EnvironmentPublishSignatures.layer, ManagedEndpointReaper.layer),
+        Layer.mergeAll(
+          EnvironmentPublishSignatures.layer,
+          ManagedEndpointReaper.layer,
+          HeldHooks.layer,
+        ),
       ),
       Layer.provideMerge(
         ManagedEndpointProvider.layerCloudflareBindings(

author	Julius Marminge <julius0216@outlook.com> 2026-10-05 16:58:19-07:00
committer	GitHub <noreply@github.com> 2026-10-05 16:58:19-07:00
log	7a047d0136de606accc8270dce64f2260529c120
tree	4409010d81711e1f996fdcf8099fa7ccdaeff3fd
parent	cfa4f765ec05950a032b6c1cf9cdfff0c2391545
signature	Signed by PGP key B5690EEEBB952194

...	@@ -3,7 +3,6 @@ import * as Layer from "effect/Layer";	...	@@ -3,7 +3,6 @@ import * as Layer from "effect/Layer";
3	import * as Schedule from "effect/Schedule";	3	import * as Schedule from "effect/Schedule";
4	import * as Stream from "effect/Stream";	4	import * as Stream from "effect/Stream";
5		5
6	import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
7	import * as CloudManagedEndpointRuntime from "../cloud/ManagedEndpointRuntime.ts";	6	import * as CloudManagedEndpointRuntime from "../cloud/ManagedEndpointRuntime.ts";
8	import { readHoldWebhooksWhileOffline, readRelayConnection } from "../cloud/config.ts";	7	import { readHoldWebhooksWhileOffline, readRelayConnection } from "../cloud/config.ts";
9	import * as ServerEnvironment from "../environment/ServerEnvironment.ts";	8	import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
...	@@ -15,9 +14,8 @@ import { makeRelayEnvironmentClient } from "./relayEnvironmentClient.ts";	...	@@ -15,9 +14,8 @@ import { makeRelayEnvironmentClient } from "./relayEnvironmentClient.ts";
15	* next backoff step. Nothing happens unless the environment opted in.	14	* next backoff step. Nothing happens unless the environment opted in.
16	*/	15	*/
17	const wakeHeldHooks = Effect.fn("HeldHooksWaker.wake")(function* () {	16	const wakeHeldHooks = Effect.fn("HeldHooksWaker.wake")(function* () {
18	const secrets = yield* ServerSecretStore.ServerSecretStore;	17	if (!(yield* readHoldWebhooksWhileOffline)) return false;
19	if (!(yield* readHoldWebhooksWhileOffline(secrets))) return false;	18	const connection = yield* readRelayConnection;
20	const connection = yield* readRelayConnection(secrets);
21	if (connection === null) return false;	19	if (connection === null) return false;
22	const environmentId = yield* (yield* ServerEnvironment.ServerEnvironment).getEnvironmentId;	20	const environmentId = yield* (yield* ServerEnvironment.ServerEnvironment).getEnvironmentId;
23	const client = yield* makeRelayEnvironmentClient(connection);	21	const client = yield* makeRelayEnvironmentClient(connection);

...	@@ -26,6 +26,7 @@ import {	...	@@ -26,6 +26,7 @@ import {
26	import * as ServerSecretStore from "../auth/ServerSecretStore.ts";	26	import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
27	import { CLOUD_MINT_PUBLIC_KEY, RELAY_ISSUER_SECRET } from "../cloud/config.ts";	27	import { CLOUD_MINT_PUBLIC_KEY, RELAY_ISSUER_SECRET } from "../cloud/config.ts";
28	import * as ServerEnvironment from "../environment/ServerEnvironment.ts";	28	import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
		29	import * as RelayDeliveryProof from "./RelayDeliveryProof.ts";
29	import { WEBHOOK_MAX_BODY_BYTES, webhookHttpApiLayer } from "./webhookRoute.ts";	30	import { WEBHOOK_MAX_BODY_BYTES, webhookHttpApiLayer } from "./webhookRoute.ts";
30		31
31	class WebhookTestApi extends HttpApi.make("environment").add(EnvironmentHttpApi.groups.webhooks) {}	32	class WebhookTestApi extends HttpApi.make("environment").add(EnvironmentHttpApi.groups.webhooks) {}
...	@@ -77,6 +78,7 @@ const handlerFor = (	...	@@ -77,6 +78,7 @@ const handlerFor = (
77	HttpRouter.toWebHandler(	78	HttpRouter.toWebHandler(
78	HttpApiBuilder.layer(WebhookTestApi).pipe(	79	HttpApiBuilder.layer(WebhookTestApi).pipe(
79	Layer.provide(webhookHttpApiLayer),	80	Layer.provide(webhookHttpApiLayer),
		81	Layer.provide(RelayDeliveryProof.layer),
80	Layer.provide(Layer.mock(ScheduledTaskService)({ triggerWebhook: trigger })),	82	Layer.provide(Layer.mock(ScheduledTaskService)({ triggerWebhook: trigger })),
81	Layer.provide(	83	Layer.provide(
82	Layer.mock(ServerSecretStore.ServerSecretStore)({	84	Layer.mock(ServerSecretStore.ServerSecretStore)({

refactor(server,relay): webhook capabilities live in services, not handlers (#16232)