Merge workbench settings patches under a row lock

packages/chat/src/routes.ts

@@ -3407,23 +3407,19 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono<TenantEnv> {
       // `chat/participants` is normalized to records on write even when
       // a caller PATCHes it with bare addresses (as the settings-control
       // wire path does) — settings always hold records, never strings.
-      const merged: Record<string, unknown> = {
-        ...existing.settings,
-        ...patch,
-      };
+      // Merge happens under a row lock against the live snapshot so a
+      // PATCH that omits `chat/participants` cannot revert a concurrent
+      // invite.
+      const writePatch: Record<string, unknown> = { ...patch };
       if (patch["chat/participants"] !== undefined) {
-        merged["chat/participants"] = parseParticipants(
+        writePatch["chat/participants"] = parseParticipants(
           patch["chat/participants"],
         );
       }
-      // The settings record itself is the durable source of truth; it
-      // is updated before anything else here fires, so a failure
-      // below never leaves the record unwritten and the audit trail
-      // silently ahead of it.
-      const row = await deps.store.updateWorkbenchSettings({
+      const row = await deps.store.patchWorkbenchSettings({
         tenantId: tenant.id,
         workbenchId,
-        settings: merged,
+        patch: writePatch,
         updatedBy: principal.id,
       });
 

packages/chat/src/store.ts

@@ -1,9 +1,12 @@
 // Persistence for the two chat product tables, kept apart from route
 // wiring so the HTTP layer never touches drizzle directly. `settings`
-// is record-as-truth: callers read and write the whole namespaced
-// jsonb blob, and this module never interprets any `chat/*` key —
-// that parsing lives in `routes.ts`, next to the request boundary it
-// guards.
+// is record-as-truth: callers read and write namespaced jsonb keys, and
+// this module never interprets any `chat/*` key — that parsing lives in
+// `routes.ts`, next to the request boundary it guards. Whole-blob replace
+// (`updateWorkbenchSettings`) remains for callers that already hold the
+// full record; `patchWorkbenchSettings` locks the row and merges only the
+// keys the caller sent so concurrent PATCHes of different keys cannot
+// clobber each other.
 //
 // `ChatStore` is the seam `routes.ts` actually depends on; `createDrizzleChatStore`
 // is its one production implementation, over the two tables in `./schema.ts`.
@@ -73,6 +76,13 @@ export interface MutateWorkbenchParticipantsInput {
   ) => ParticipantRecord[];
 }
 
+export interface PatchWorkbenchSettingsInput {
+  readonly tenantId: string;
+  readonly workbenchId: string;
+  readonly patch: Record<string, unknown>;
+  readonly updatedBy: string;
+}
+
 export interface ChatBenchSettingsRow {
   readonly tenantId: string;
   readonly settings: Record<string, unknown>;
@@ -146,6 +156,15 @@ export interface ChatStore {
   mutateWorkbenchParticipants(
     input: MutateWorkbenchParticipantsInput,
   ): Promise<WorkbenchSettingsRow>;
+  /**
+   * Merges `patch` onto the workbench's settings under a row lock
+   * (`SELECT ... FOR UPDATE` in Postgres). Only keys present in `patch`
+   * are written; omitted keys keep the locked snapshot's values, so a
+   * concurrent PATCH of a different key cannot be reverted.
+   */
+  patchWorkbenchSettings(
+    input: PatchWorkbenchSettingsInput,
+  ): Promise<WorkbenchSettingsRow>;
   getBenchSettings(tenantId: string): Promise<ChatBenchSettingsRow | undefined>;
   upsertBenchSettings(
     input: UpsertBenchSettingsInput,
@@ -200,6 +219,18 @@ export interface ChatStore {
   ): Promise<WorkbenchByParticipantAddress | undefined>;
 }
 
+/** Top-level JSONB merge: only keys present in `patch` overwrite. */
+function mergeSettingsPatch(
+  existing: Record<string, unknown>,
+  patch: Record<string, unknown>,
+): Record<string, unknown> {
+  const merged: Record<string, unknown> = { ...existing };
+  for (const key of Object.keys(patch)) {
+    merged[key] = patch[key];
+  }
+  return merged;
+}
+
 /**
  * The production `ChatStore`, backed by the `workbench_settings` and
  * `workbench_read_state` tables declared in `./schema.ts`.
@@ -338,6 +369,49 @@ export function createDrizzleChatStore<TSchema extends Record<string, unknown>>(
       });
     },
 
+    async patchWorkbenchSettings(input) {
+      return db.transaction(async (tx) => {
+        const [selected] = await tx
+          .select()
+          .from(workbenchSettings)
+          .where(
+            and(
+              eq(workbenchSettings.tenantId, input.tenantId),
+              eq(workbenchSettings.workbenchId, input.workbenchId),
+            ),
+          )
+          .for("update")
+          .limit(1);
+        if (selected === undefined) {
+          throw new Error(
+            `patchWorkbenchSettings: no workbench_settings row for workbench ${input.workbenchId}`,
+          );
+        }
+        const existing = selected as WorkbenchSettingsRow;
+        const merged = mergeSettingsPatch(existing.settings, input.patch);
+        const [row] = await tx
+          .update(workbenchSettings)
+          .set({
+            settings: merged,
+            updatedBy: input.updatedBy,
+            updatedAt: new Date(),
+          })
+          .where(
+            and(
+              eq(workbenchSettings.tenantId, input.tenantId),
+              eq(workbenchSettings.workbenchId, input.workbenchId),
+            ),
+          )
+          .returning();
+        if (row === undefined) {
+          throw new Error(
+            `patchWorkbenchSettings: no workbench_settings row for workbench ${input.workbenchId}`,
+          );
+        }
+        return row as WorkbenchSettingsRow;
+      });
+    },
+
     async getBenchSettings(tenantId) {
       const [selected] = await db
         .select()
@@ -467,6 +541,7 @@ export function createInMemoryChatStore(): ChatStore {
   const readStateByKey = new Map<string, ReadStateRow>();
   const benchSettingsByTenant = new Map<string, ChatBenchSettingsRow>();
   const launchedByKey = new Set<string>();
+  const rowLocks = new Map<string, Promise<void>>();
 
   const settingsKey = (tenantId: string, workbenchId: string) =>
     `${tenantId}:${workbenchId}`;
@@ -476,6 +551,27 @@ export function createInMemoryChatStore(): ChatStore {
     principalId: string,
   ) => `${tenantId}:${workbenchId}:${principalId}`;
 
+  const withRowLock = async <T>(
+    key: string,
+    fn: () => Promise<T>,
+  ): Promise<T> => {
+    const previous = rowLocks.get(key) ?? Promise.resolve();
+    let release: () => void = () => undefined;
+    const held = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    rowLocks.set(
+      key,
+      previous.then(() => held),
+    );
+    await previous;
+    try {
+      return await fn();
+    } finally {
+      release();
+    }
+  };
+
   return {
     async createWorkbenchSettings(input) {
       const row: WorkbenchSettingsRow = {
@@ -548,6 +644,26 @@ export function createInMemoryChatStore(): ChatStore {
       return row;
     },
 
+    async patchWorkbenchSettings(input) {
+      const key = settingsKey(input.tenantId, input.workbenchId);
+      return withRowLock(key, async () => {
+        const existing = settingsByKey.get(key);
+        if (existing === undefined) {
+          throw new Error(
+            `patchWorkbenchSettings: no workbench_settings row for workbench ${input.workbenchId}`,
+          );
+        }
+        const row: WorkbenchSettingsRow = {
+          ...existing,
+          settings: mergeSettingsPatch(existing.settings, input.patch),
+          updatedBy: input.updatedBy,
+          updatedAt: new Date(),
+        };
+        settingsByKey.set(key, row);
+        return row;
+      });
+    },
+
     async getBenchSettings(tenantId) {
       return benchSettingsByTenant.get(tenantId);
     },

packages/chat/test/settings-patch.drizzle.test.ts

@@ -0,0 +1,129 @@
+// DB-gated: skipped when no DATABASE_URL is reachable (a fresh
+// checkout still runs the unit gates), mirroring `read-state.drizzle.test.ts`.
+// Two concurrent `patchWorkbenchSettings` calls on one row must both land:
+// a participants write and a `chat/pinned` write cannot clobber each other.
+import { afterAll, beforeAll, describe, expect, test } from "bun:test";
+import { drizzle } from "drizzle-orm/postgres-js";
+import postgres from "postgres";
+
+import { e2eDatabaseUrl } from "../../../scripts/e2e/harness";
+import { applyChatMigrations } from "../src/migrations";
+import { createDrizzleChatStore } from "../src/store";
+
+function scratchUrlFor(e2eUrl: string): string {
+  const url = new URL(e2eUrl);
+  const database = url.pathname.replace(/^\//, "");
+  url.pathname = `/${database}_chat_settings_patch_drizzle_test`;
+  return url.toString();
+}
+
+const databaseUrl = e2eDatabaseUrl();
+const describeIfDb = databaseUrl === undefined ? describe.skip : describe;
+
+const TENANT = "tnt_1";
+const WORKBENCH = "run_workbench1";
+const ALICE = { address: "prn_alice@acme.example", handle: "alice" };
+const BOB = { address: "prn_bob@acme.example", handle: "bob" };
+
+describeIfDb("createDrizzleChatStore: patchWorkbenchSettings", () => {
+  const scratchUrl = scratchUrlFor(
+    databaseUrl ?? "postgres://localhost:5432/unused",
+  );
+  const scratchTarget = new URL(scratchUrl);
+  const scratchDatabase = scratchTarget.pathname.replace(/^\//, "");
+
+  beforeAll(async () => {
+    const maintenanceUrl = new URL(scratchUrl);
+    maintenanceUrl.pathname = "/postgres";
+    const maintenance = postgres(maintenanceUrl.toString(), {
+      max: 1,
+      onnotice: () => undefined,
+    });
+    try {
+      await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`);
+      await maintenance.unsafe(`CREATE DATABASE "${scratchDatabase}"`);
+    } finally {
+      await maintenance.end();
+    }
+    await applyChatMigrations(scratchUrl);
+  });
+
+  afterAll(async () => {
+    const maintenanceUrl = new URL(scratchUrl);
+    maintenanceUrl.pathname = "/postgres";
+    const maintenance = postgres(maintenanceUrl.toString(), {
+      max: 1,
+      onnotice: () => undefined,
+    });
+    try {
+      await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`);
+    } finally {
+      await maintenance.end();
+    }
+  });
+
+  test("concurrent patches of participants and another key both land", async () => {
+    const sql = postgres(scratchUrl, { max: 5, onnotice: () => undefined });
+    try {
+      const store = createDrizzleChatStore(drizzle(sql));
+      await store.createWorkbenchSettings({
+        tenantId: TENANT,
+        workbenchId: WORKBENCH,
+        settings: {
+          "chat/pinned": false,
+          "chat/participants": [ALICE],
+        },
+        updatedBy: "prn_1",
+      });
+
+      await Promise.all([
+        store.patchWorkbenchSettings({
+          tenantId: TENANT,
+          workbenchId: WORKBENCH,
+          patch: { "chat/pinned": true },
+          updatedBy: "prn_2",
+        }),
+        store.patchWorkbenchSettings({
+          tenantId: TENANT,
+          workbenchId: WORKBENCH,
+          patch: { "chat/participants": [ALICE, BOB] },
+          updatedBy: "prn_3",
+        }),
+      ]);
+
+      const row = await store.getWorkbenchSettings(TENANT, WORKBENCH);
+      expect(row?.settings["chat/pinned"]).toBe(true);
+      expect(row?.settings["chat/participants"]).toEqual([ALICE, BOB]);
+    } finally {
+      await sql.end();
+    }
+  });
+
+  test("a patch that omits chat/participants keeps the locked row's list", async () => {
+    const sql = postgres(scratchUrl, { max: 5, onnotice: () => undefined });
+    try {
+      const store = createDrizzleChatStore(drizzle(sql));
+      const workbenchId = "run_workbench2";
+      await store.createWorkbenchSettings({
+        tenantId: TENANT,
+        workbenchId,
+        settings: {
+          "chat/pinned": false,
+          "chat/participants": [ALICE, BOB],
+        },
+        updatedBy: "prn_1",
+      });
+
+      const updated = await store.patchWorkbenchSettings({
+        tenantId: TENANT,
+        workbenchId,
+        patch: { "chat/pinned": true },
+        updatedBy: "prn_2",
+      });
+      expect(updated.settings["chat/pinned"]).toBe(true);
+      expect(updated.settings["chat/participants"]).toEqual([ALICE, BOB]);
+    } finally {
+      await sql.end();
+    }
+  });
+});