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

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

@@ -0,0 +1,160 @@
+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,
+  PUBLISH_AGENT_ACTIVITY_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;
+    readonly failActivityWrite?: boolean;
+    readonly relayFails?: boolean;
+    readonly failActivityRead?: 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")],
+      [PUBLISH_AGENT_ACTIVITY_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 options.relayFails
+          ? Promise.resolve(new Response("unavailable", { status: 503 }))
+          : Promise.resolve(Response.json(payload));
+      },
+      { preconnect: () => {} },
+    );
+    const dependencies = Layer.mergeAll(
+      Layer.mock(ServerSecretStore.ServerSecretStore)({
+        get: (name) =>
+          options.failActivityRead && name === PUBLISH_AGENT_ACTIVITY_SECRET
+            ? Effect.fail(
+                new ServerSecretStore.SecretStoreReadError({
+                  resource: name,
+                  cause: new Error("busy"),
+                }),
+              )
+            : Effect.succeed(Option.fromNullishOr(stored.get(name))),
+        set: (name, value) =>
+          (options.failHoldWrite && name === HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET) ||
+          (options.failActivityWrite && name === PUBLISH_AGENT_ACTIVITY_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",
+      );
+    }),
+  ),
+);
+
+it.effect("leaves the relay untouched when the activity setting can't be saved", () =>
+  withService({ failActivityWrite: 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, []);
+      assert.equal(
+        new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
+        "false",
+      );
+    }),
+  ),
+);
+
+it.effect("keeps both settings unchanged when the relay refuses the hold change", () =>
+  withService({ relayFails: true }, ({ preferences, stored }) =>
+    Effect.gen(function* () {
+      yield* preferences
+        .update({ publishAgentActivity: true, holdWebhooksWhileOffline: true })
+        .pipe(Effect.flip);
+      assert.equal(new TextDecoder().decode(stored.get(PUBLISH_AGENT_ACTIVITY_SECRET)), "false");
+      assert.equal(
+        new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
+        "false",
+      );
+    }),
+  ),
+);
+
+it.effect("changes nothing when the current activity setting can't be read", () =>
+  withService({ failActivityRead: true, relayFails: 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, []);
+      assert.equal(new TextDecoder().decode(stored.get(PUBLISH_AGENT_ACTIVITY_SECRET)), "false");
+    }),
+  ),
+);

apps/server/src/cloud/CloudPreferences.ts

@@ -0,0 +1,120 @@
+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 Option from "effect/Option";
+
+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));
+
+/** A failed rollback leaves a setting changed; it is logged, not hidden. */
+const rollbackFailed = (cause: unknown) =>
+  Effect.logWarning("Could not roll back a T3 Connect preference", { cause });
+
+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, all or nothing. The
+     * activity setting is saved first. Holding webhooks while offline is decided
+     * by the relay, so the relay is told before the local copy is saved. If
+     * either step fails, the activity setting is put back, and so is the relay
+     * when only the local save failed.
+     */
+    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 withSecrets = Effect.provideService(ServerSecretStore.ServerSecretStore, secrets);
+
+  const pushHoldWebhooksWhileOffline = Effect.fn("CloudPreferences.pushHoldWebhooksWhileOffline")(
+    function* (holdWebhooksWhileOffline: boolean) {
+      const connection = yield* readRelayConnection.pipe(withSecrets);
+      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) {
+      // All or nothing: the activity setting is saved first, before the relay
+      // is told anything, and put back if the hold change then fails.
+      // A failed read stops here, before anything changes: guessing "unset"
+      // would make a later rollback delete the real setting.
+      const previousActivity = yield* secrets
+        .get(PUBLISH_AGENT_ACTIVITY_SECRET)
+        .pipe(Effect.catch(internalError("Could not read environment cloud preferences.")));
+      yield* save(PUBLISH_AGENT_ACTIVITY_SECRET, input.publishAgentActivity);
+      if (input.holdWebhooksWhileOffline !== undefined) {
+        const next = input.holdWebhooksWhileOffline;
+        const previous = yield* readHoldWebhooksWhileOffline.pipe(withSecrets);
+        yield* pushHoldWebhooksWhileOffline(next).pipe(
+          Effect.andThen(
+            save(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET, next).pipe(
+              Effect.tapError(() =>
+                previous === next
+                  ? Effect.void
+                  : pushHoldWebhooksWhileOffline(previous).pipe(Effect.catch(rollbackFailed)),
+              ),
+            ),
+          ),
+          Effect.tapError(() =>
+            Option.match(previousActivity, {
+              onNone: () => secrets.remove(PUBLISH_AGENT_ACTIVITY_SECRET),
+              onSome: (bytes) => secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, bytes),
+            }).pipe(Effect.catch(rollbackFailed)),
+          ),
+        );
+      }
+      yield* awarenessRelay.requestCatchUp();
+    },
+  );
+
+  return CloudPreferences.of({ update });
+});
+
+export const layer = Layer.effect(CloudPreferences, make);

apps/server/src/cloud/config.ts

@@ -6,7 +6,7 @@ import * as Effect from "effect/Effect";
 import * as Option from "effect/Option";
 import * as Schema from "effect/Schema";
 
-import type * as ServerSecretStore from "../auth/ServerSecretStore.ts";
+import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
 
 export const CLOUD_MINT_PUBLIC_KEY = "cloud-mint-ed25519-public-key";
 export const CLOUD_ENDPOINT_RUNTIME_CONFIG = "cloud-endpoint-runtime-config";
@@ -75,32 +75,27 @@ export const readAgentActivityPublishingActive = (
     );
   }).pipe(Effect.orElseSucceed(() => false));
 
-const readSecretString = (
-  secrets: ServerSecretStore.ServerSecretStore["Service"],
-  name: string,
-): Effect.Effect<string | null> =>
-  secrets.get(name).pipe(
+/** A non-empty secret as text, or null when it is missing, empty or unreadable. */
+const readSecretString = (name: string) =>
+  ServerSecretStore.ServerSecretStore.pipe(
+    Effect.flatMap((secrets) => secrets.get(name)),
     Effect.map((bytes) =>
       Option.isSome(bytes) && bytes.value.length > 0 ? new TextDecoder().decode(bytes.value) : null,
     ),
     Effect.orElseSucceed(() => null),
   );
 
 /** The relay URL and environment credential, or null when not linked to T3 Connect. */
-export const readRelayConnection = (secrets: ServerSecretStore.ServerSecretStore["Service"]) =>
-  Effect.all([
-    readSecretString(secrets, RELAY_URL_SECRET),
-    readSecretString(secrets, RELAY_ENVIRONMENT_CREDENTIAL_SECRET),
-  ]).pipe(
-    Effect.map(([url, environmentCredential]) =>
-      url && environmentCredential ? { url, environmentCredential } : null,
-    ),
-  );
+export const readRelayConnection = Effect.all([
+  readSecretString(RELAY_URL_SECRET),
+  readSecretString(RELAY_ENVIRONMENT_CREDENTIAL_SECRET),
+]).pipe(
+  Effect.map(([url, environmentCredential]) =>
+    url && environmentCredential ? { url, environmentCredential } : null,
+  ),
+);
 
 /** Whether this environment opted in to T3 Connect holding webhooks while it is offline. */
-export const readHoldWebhooksWhileOffline = (
-  secrets: ServerSecretStore.ServerSecretStore["Service"],
-) =>
-  readSecretString(secrets, HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET).pipe(
-    Effect.map((value) => value === "true"),
-  );
+export const readHoldWebhooksWhileOffline = readSecretString(
+  HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET,
+).pipe(Effect.map((value) => value === "true"));