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

apps/server/src/cloud/CloudPreferences.test.ts

@@ -0,0 +1,98 @@
+import { EnvironmentId } from "@t3tools/contracts";
+import { assert, it } from "@effect/vitest";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
+import * as FetchHttpClient from "effect/http/FetchHttpClient";
+
+import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
+import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
+import * as AgentAwarenessRelay from "../relay/AgentAwarenessRelay.ts";
+import * as CloudPreferences from "./CloudPreferences.ts";
+import {
+  HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET,
+  RELAY_ENVIRONMENT_CREDENTIAL_SECRET,
+  RELAY_URL_SECRET,
+} from "./config.ts";
+
+const encode = (value: string) => new TextEncoder().encode(value);
+
+/** A linked environment whose secret store can refuse writes, and the relay calls it made. */
+const withService = <A, E>(
+  options: { readonly failHoldWrite?: boolean },
+  body: (input: {
+    readonly preferences: CloudPreferences.CloudPreferences["Service"];
+    readonly stored: Map<string, Uint8Array>;
+    readonly relayCalls: Array<boolean>;
+  }) => Effect.Effect<A, E, never>,
+) =>
+  Effect.gen(function* () {
+    const stored = new Map<string, Uint8Array>([
+      [RELAY_URL_SECRET, encode("https://relay.test")],
+      [RELAY_ENVIRONMENT_CREDENTIAL_SECRET, encode("credential")],
+      [HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET, encode("false")],
+    ]);
+    const relayCalls: Array<boolean> = [];
+    const fetch: typeof globalThis.fetch = Object.assign(
+      (_input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
+        const payload = JSON.parse(String(init?.body)) as { holdWebhooksWhileOffline: boolean };
+        relayCalls.push(payload.holdWebhooksWhileOffline);
+        return Promise.resolve(Response.json(payload));
+      },
+      { preconnect: () => {} },
+    );
+    const dependencies = Layer.mergeAll(
+      Layer.mock(ServerSecretStore.ServerSecretStore)({
+        get: (name) => Effect.succeed(Option.fromNullishOr(stored.get(name))),
+        set: (name, value) =>
+          options.failHoldWrite && name === HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET
+            ? Effect.fail(
+                new ServerSecretStore.SecretStorePersistError({
+                  resource: name,
+                  cause: new Error("read-only"),
+                }),
+              )
+            : Effect.sync(() => void stored.set(name, value)),
+      }),
+      Layer.mock(ServerEnvironment.ServerEnvironment)({
+        getEnvironmentId: Effect.succeed(EnvironmentId.make("environment-1")),
+      }),
+      Layer.mock(AgentAwarenessRelay.AgentAwarenessRelay)({ requestCatchUp: () => Effect.void }),
+    );
+    return yield* Effect.gen(function* () {
+      const preferences = yield* CloudPreferences.CloudPreferences;
+      return yield* body({ preferences, stored, relayCalls });
+    }).pipe(
+      Effect.provide(CloudPreferences.layer.pipe(Layer.provide(dependencies))),
+      Effect.provideService(FetchHttpClient.Fetch, fetch),
+    );
+  });
+
+it.effect("tells the relay before saving the hold setting locally", () =>
+  withService({}, ({ preferences, stored, relayCalls }) =>
+    Effect.gen(function* () {
+      yield* preferences.update({ publishAgentActivity: true, holdWebhooksWhileOffline: true });
+      assert.deepEqual(relayCalls, [true]);
+      assert.equal(
+        new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
+        "true",
+      );
+    }),
+  ),
+);
+
+it.effect("puts the relay back when the local save fails", () =>
+  withService({ failHoldWrite: true }, ({ preferences, stored, relayCalls }) =>
+    Effect.gen(function* () {
+      const error = yield* preferences
+        .update({ publishAgentActivity: true, holdWebhooksWhileOffline: true })
+        .pipe(Effect.flip);
+      assert.equal(error._tag, "EnvironmentHttpInternalServerError");
+      assert.deepEqual(relayCalls, [true, false]);
+      assert.equal(
+        new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
+        "false",
+      );
+    }),
+  ),
+);

apps/server/src/cloud/CloudPreferences.ts

@@ -0,0 +1,96 @@
+import {
+  type EnvironmentCloudPreferencesRequest,
+  EnvironmentHttpBadRequestError,
+  EnvironmentHttpInternalServerError,
+} from "@t3tools/contracts";
+import * as Context from "effect/Context";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+
+import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
+import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
+import * as AgentAwarenessRelay from "../relay/AgentAwarenessRelay.ts";
+import { makeRelayEnvironmentClient } from "../relay/relayEnvironmentClient.ts";
+import {
+  HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET,
+  PUBLISH_AGENT_ACTIVITY_SECRET,
+  readHoldWebhooksWhileOffline,
+  readRelayConnection,
+} from "./config.ts";
+
+const encode = (value: boolean) => new TextEncoder().encode(String(value));
+
+const internalError = (message: string) => (cause: unknown) =>
+  Effect.logError(message, { cause }).pipe(
+    Effect.andThen(Effect.fail(new EnvironmentHttpInternalServerError({ message }))),
+  );
+
+export class CloudPreferences extends Context.Service<
+  CloudPreferences,
+  {
+    /**
+     * Saves this environment's T3 Connect preferences. Holding webhooks while
+     * offline is decided by the relay, so it is told first and put back if the
+     * local save fails.
+     */
+    readonly update: (
+      input: EnvironmentCloudPreferencesRequest,
+    ) => Effect.Effect<void, EnvironmentHttpBadRequestError | EnvironmentHttpInternalServerError>;
+  }
+>()("t3/cloud/CloudPreferences") {}
+
+const make = Effect.gen(function* () {
+  const secrets = yield* ServerSecretStore.ServerSecretStore;
+  const environment = yield* ServerEnvironment.ServerEnvironment;
+  const awarenessRelay = yield* AgentAwarenessRelay.AgentAwarenessRelay;
+
+  const pushHoldWebhooksWhileOffline = Effect.fn("CloudPreferences.pushHoldWebhooksWhileOffline")(
+    function* (holdWebhooksWhileOffline: boolean) {
+      const connection = yield* readRelayConnection(secrets);
+      if (connection === null) {
+        return yield* new EnvironmentHttpBadRequestError({
+          message: "Link this environment to T3 Connect first.",
+        });
+      }
+      const environmentId = yield* environment.getEnvironmentId;
+      const client = yield* makeRelayEnvironmentClient(connection);
+      yield* client.server
+        .updateLinkPreferences({
+          params: { environmentId },
+          payload: { holdWebhooksWhileOffline },
+        })
+        .pipe(
+          Effect.timeout("10 seconds"),
+          Effect.catch(internalError("Could not update T3 Connect webhook settings.")),
+        );
+    },
+  );
+
+  const save = (name: string, value: boolean) =>
+    secrets
+      .set(name, encode(value))
+      .pipe(Effect.catch(internalError("Could not persist environment cloud preferences.")));
+
+  const update: CloudPreferences["Service"]["update"] = Effect.fn("CloudPreferences.update")(
+    function* (input) {
+      if (input.holdWebhooksWhileOffline !== undefined) {
+        const next = input.holdWebhooksWhileOffline;
+        const previous = yield* readHoldWebhooksWhileOffline(secrets);
+        yield* pushHoldWebhooksWhileOffline(next);
+        yield* save(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET, next).pipe(
+          Effect.tapError(() =>
+            previous === next
+              ? Effect.void
+              : pushHoldWebhooksWhileOffline(previous).pipe(Effect.ignore),
+          ),
+        );
+      }
+      yield* save(PUBLISH_AGENT_ACTIVITY_SECRET, input.publishAgentActivity);
+      yield* awarenessRelay.requestCatchUp();
+    },
+  );
+
+  return CloudPreferences.of({ update });
+});
+
+export const layer = Layer.effect(CloudPreferences, make);

apps/server/src/cloud/http.ts

@@ -70,7 +70,6 @@ import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
 import { requireEnvironmentScope } from "../auth/http.ts";
 import * as ServerConfig from "../config.ts";
 import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
-import { makeRelayEnvironmentClient } from "../relay/relayEnvironmentClient.ts";
 import * as AgentAwarenessRelay from "../relay/AgentAwarenessRelay.ts";
 import * as ManagedEndpointRuntime from "./ManagedEndpointRuntime.ts";
 import {
@@ -102,6 +101,7 @@ import {
   setCliDesiredCloudLink,
 } from "./CliState.ts";
 import * as CliTokenManager from "./CliTokenManager.ts";
+import * as CloudPreferences from "./CloudPreferences.ts";
 import { getOrCreateEnvironmentKeyPairFromSecretStore } from "./environmentKeys.ts";
 import { traceRelayRequest } from "./traceRelayRequest.ts";
 import { filterRelayResponse, relayRequestError, shouldRetryCloudLink } from "./relayResponse.ts";
@@ -1358,57 +1358,14 @@ const cloudUnlinkHandler = Effect.fn("environment.cloud.unlink")(
   ),
 );
 
-const pushHoldWebhooksWhileOffline = Effect.fn("environment.cloud.pushHoldWebhooksWhileOffline")(
-  function* (dependencies: CloudHttpDependencies, holdWebhooksWhileOffline: boolean) {
-    const connection = yield* readRelayConnection(dependencies.secrets);
-    if (connection === null) {
-      return yield* new EnvironmentHttpBadRequestError({
-        message: "Link this environment to T3 Connect first.",
-      });
-    }
-    const environmentId = yield* dependencies.environment.getEnvironmentId;
-    const client = yield* makeRelayEnvironmentClient(connection);
-    yield* client.server
-      .updateLinkPreferences({
-        params: { environmentId },
-        payload: { holdWebhooksWhileOffline },
-      })
-      .pipe(
-        Effect.timeout("10 seconds"),
-        Effect.catch(
-          failEnvironmentCloudInternalError("Could not update T3 Connect webhook settings."),
-        ),
-      );
-  },
-);
-
 const cloudPreferencesHandler = Effect.fn("environment.cloud.preferences")(
-  function* (dependencies: CloudHttpDependencies, payload: EnvironmentCloudPreferencesRequest) {
+  function* (
+    dependencies: CloudHttpDependencies,
+    preferences: CloudPreferences.CloudPreferences["Service"],
+    payload: EnvironmentCloudPreferencesRequest,
+  ) {
     yield* requireEnvironmentScope(AuthRelayWriteScope);
-    if (payload.holdWebhooksWhileOffline !== undefined) {
-      // The relay decides whether to hold a request, so it is told first; the
-      // local copy is only saved once the relay has the same value, and the
-      // relay is put back if that save fails.
-      const previous = yield* readHoldWebhooksWhileOffline(dependencies.secrets);
-      yield* pushHoldWebhooksWhileOffline(dependencies, payload.holdWebhooksWhileOffline);
-      yield* dependencies.secrets
-        .set(
-          HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET,
-          stringToBytes(String(payload.holdWebhooksWhileOffline)),
-        )
-        .pipe(
-          Effect.tapError(() =>
-            previous === payload.holdWebhooksWhileOffline
-              ? Effect.void
-              : pushHoldWebhooksWhileOffline(dependencies, previous).pipe(Effect.ignore),
-          ),
-        );
-    }
-    yield* dependencies.secrets.set(
-      PUBLISH_AGENT_ACTIVITY_SECRET,
-      stringToBytes(String(payload.publishAgentActivity)),
-    );
-    yield* dependencies.awarenessRelay.requestCatchUp();
+    yield* preferences.update(payload);
     return yield* readCloudLinkState(dependencies);
   },
   Effect.catchIf(
@@ -1661,12 +1618,15 @@ export const connectHttpApiLayer = HttpApiBuilder.group(
   "connect",
   Effect.fnUntraced(function* (handlers) {
     const dependencies = yield* cloudHttpDependencies;
+    const preferences = yield* CloudPreferences.CloudPreferences;
     return handlers
       .handle("linkProof", ({ payload }) => cloudLinkProofHandler(dependencies, payload))
       .handle("relayConfig", ({ payload }) => cloudRelayConfigHandler(dependencies, payload))
       .handle("linkState", () => cloudLinkStateHandler(dependencies))
       .handle("unlink", () => cloudUnlinkHandler(dependencies))
-      .handle("preferences", ({ payload }) => cloudPreferencesHandler(dependencies, payload))
+      .handle("preferences", ({ payload }) =>
+        cloudPreferencesHandler(dependencies, preferences, payload),
+      )
       .handle("health", ({ payload }) => cloudEnvironmentHealthHandler(dependencies, payload))
       .handle("mintCredential", ({ payload }) => cloudMintCredentialHandler(dependencies, payload))
       .handle("t3MintCredential", ({ payload }) =>

apps/server/src/scheduledTasks/relayDeliveryProof.ts → apps/server/src/scheduledTasks/RelayDeliveryProof.ts

@@ -14,7 +14,9 @@ import {
   verifyRelayJwt,
 } from "@t3tools/shared/relayJwt";
 import * as Clock from "effect/Clock";
+import * as Context from "effect/Context";
 import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
 import * as Schema from "effect/Schema";
 
@@ -29,11 +31,23 @@ const decodePayload = Schema.decodeUnknownOption(RelayHookDeliveryProofPayload);
 const text = (bytes: Option.Option<Uint8Array>) =>
   Option.map(bytes, (value) => new TextDecoder().decode(value));
 
-/** The relay's own delivery id and receive time, when the request proves it came from the relay. */
-export const makeRelayDeliveryVerifier = Effect.gen(function* () {
+export class RelayDeliveryProof extends Context.Service<
+  RelayDeliveryProof,
+  {
+    /** The relay's own delivery id and receive time, when the request proves it came from the relay. */
+    readonly verify: (input: {
+      readonly headers: Readonly<Record<string, string>>;
+      readonly hookId: string;
+    }) => Effect.Effect<
+      Option.Option<{ readonly deliveryId: string; readonly receivedAt: string }>
+    >;
+  }
+>()("t3/scheduledTasks/RelayDeliveryProof") {}
+
+const make = Effect.gen(function* () {
   const secrets = yield* ServerSecretStore.ServerSecretStore;
   const environment = yield* ServerEnvironment.ServerEnvironment;
-  return (input: { readonly headers: Readonly<Record<string, string>>; readonly hookId: string }) =>
+  const verify: RelayDeliveryProof["Service"]["verify"] = (input) =>
     Effect.gen(function* () {
       const proof = input.headers[RELAY_HOOK_DELIVERY_HEADER];
       const deliveryId = input.headers["x-t3-relay-delivery-id"];
@@ -71,4 +85,7 @@ export const makeRelayDeliveryVerifier = Effect.gen(function* () {
       }
       return Option.some({ deliveryId, receivedAt });
     }).pipe(Effect.orElseSucceed(Option.none), Effect.withSpan("webhook.verifyRelayDelivery"));
+  return RelayDeliveryProof.of({ verify });
 });
+
+export const layer = Layer.effect(RelayDeliveryProof, make);

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

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

apps/server/src/scheduledTasks/webhookRoute.ts

@@ -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. */
@@ -33,7 +33,7 @@ const json = (status: number, body: Record<string, string>, outcome: string) =>
 const handleWebhook =
   (
     scheduledTasks: ScheduledTaskService.ScheduledTaskService["Service"],
-    verifyRelayDelivery: Effect.Success<typeof makeRelayDeliveryVerifier>,
+    relayDeliveryProof: RelayDeliveryProof.RelayDeliveryProof["Service"],
   ) =>
   ({
     params,
@@ -75,7 +75,7 @@ const handleWebhook =
       // 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 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;
 
@@ -136,7 +136,7 @@ export const webhookHttpApiLayer = HttpApiBuilder.group(
   Effect.fnUntraced(function* (handlers) {
     const handler = handleWebhook(
       yield* ScheduledTaskService.ScheduledTaskService,
-      yield* makeRelayDeliveryVerifier,
+      yield* RelayDeliveryProof.RelayDeliveryProof,
     );
     return handlers
       .handleRaw("webhookPost", handler)