fix(server): project updates reach folded settings

apps/server/src/device/DeviceService.test.ts

@@ -203,14 +203,17 @@ const fixture = Effect.fn("fixture")(function* (
         ready: Effect.void,
         getSettings: Ref.get(settings),
         updateSettings: (patch) =>
-          Ref.updateAndGet(settings, (current) => ({
-            ...current,
-            enableDeviceSupport: patch.enableDeviceSupport ?? current.enableDeviceSupport,
-            enableAgentDeviceAccess:
-              patch.enableAgentDeviceAccess ?? current.enableAgentDeviceAccess,
-            deviceOnboardingCompleted:
-              patch.deviceOnboardingCompleted ?? current.deviceOnboardingCompleted,
-          })),
+          Ref.updateAndGet(settings, (current) => {
+            const resolvedPatch = typeof patch === "function" ? patch(current) : patch;
+            return {
+              ...current,
+              enableDeviceSupport: resolvedPatch.enableDeviceSupport ?? current.enableDeviceSupport,
+              enableAgentDeviceAccess:
+                resolvedPatch.enableAgentDeviceAccess ?? current.enableAgentDeviceAccess,
+              deviceOnboardingCompleted:
+                resolvedPatch.deviceOnboardingCompleted ?? current.deviceOnboardingCompleted,
+            };
+          }),
         streamChanges: Stream.empty,
         subscribeChanges: Effect.succeed(Stream.empty),
       }),

apps/server/src/orchestration-v2/EventSink.ts

@@ -12,6 +12,7 @@ import {
   type ProjectId,
   ThreadId,
 } from "@t3tools/contracts";
+import * as KeyedLock from "@t3tools/shared/KeyedLock";
 import * as Context from "effect/Context";
 import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
@@ -177,6 +178,7 @@ export interface EventSinkV2Shape {
     readonly commandType: string;
     readonly acceptedAt: DateTime.Utc;
     readonly event: UnsequencedProjectEvent;
+    readonly beforeCommit?: Effect.Effect<void, unknown>;
   }) => Effect.Effect<
     { readonly receipt: CommandReceiptStore.ProjectCommandReceiptV2; readonly committed: boolean },
     EventSinkV2Error
@@ -222,6 +224,7 @@ const layerBase: Layer.Layer<
   Effect.gen(function* () {
     const sql = yield* SqlClient.SqlClient;
     const commandReceipts = yield* CommandReceiptStore.CommandReceiptStoreV2;
+    const commandLocks = yield* KeyedLock.make<CommandId>();
     const effectOutbox = yield* EffectOutbox.EffectOutboxV2;
     const eventStore = yield* EventStore.EventStoreV2;
     const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
@@ -718,6 +721,16 @@ const layerBase: Layer.Layer<
 
     const commitProjectCommandEffect = Effect.fn("orchestrationV2.EventSink.commitProjectCommand")(
       function* (input: Parameters<EventSinkV2Shape["commitProjectCommand"]>[0]) {
+        const existing = yield* commandReceipts.getProjectByCommandId(input.commandId);
+        if (Option.isSome(existing)) return { receipt: existing.value, committed: false };
+        if (Option.isSome(yield* commandReceipts.getByCommandId(input.commandId))) {
+          return yield* Effect.fail(
+            `Command ${input.commandId} was already used by a thread command.`,
+          );
+        }
+        // Settings takes its own lock before reading SQL. Never hold the SQL writer while
+        // waiting for it. The command lock keeps concurrent collisions from rerunning this write.
+        if (input.beforeCommit !== undefined) yield* input.beforeCommit;
         const result = yield* commitThenPublish(
           Effect.gen(function* () {
             const reserved: CommandReceiptStore.ProjectCommandReceiptV2 = {
@@ -899,7 +912,7 @@ const layerBase: Layer.Layer<
           ),
         ),
       commitCommand: (input) =>
-        commitCommandEffect(input).pipe(
+        commandLocks.withLock(input.commandId, commitCommandEffect(input)).pipe(
           Effect.mapError(
             (cause) =>
               new EventSinkWriteError({
@@ -910,7 +923,7 @@ const layerBase: Layer.Layer<
           ),
         ),
       commitRejectedCommand: (input) =>
-        commitRejectedCommandEffect(input).pipe(
+        commandLocks.withLock(input.commandId, commitRejectedCommandEffect(input)).pipe(
           Effect.mapError(
             (cause) =>
               new EventSinkWriteError({
@@ -921,19 +934,23 @@ const layerBase: Layer.Layer<
           ),
         ),
       commitProjectCommand: (input) =>
-        commitProjectCommandEffect(input).pipe(
-          Effect.mapError(
-            (cause) =>
-              new EventSinkWriteError({ commandId: input.commandId, eventCount: 1, cause }),
+        commandLocks
+          .withLock(input.commandId, commitProjectCommandEffect(input))
+          .pipe(
+            Effect.mapError(
+              (cause) =>
+                new EventSinkWriteError({ commandId: input.commandId, eventCount: 1, cause }),
+            ),
           ),
-        ),
       commitRejectedProjectCommand: (input) =>
-        commitRejectedProjectCommandEffect(input).pipe(
-          Effect.mapError(
-            (cause) =>
-              new EventSinkWriteError({ commandId: input.commandId, eventCount: 0, cause }),
+        commandLocks
+          .withLock(input.commandId, commitRejectedProjectCommandEffect(input))
+          .pipe(
+            Effect.mapError(
+              (cause) =>
+                new EventSinkWriteError({ commandId: input.commandId, eventCount: 0, cause }),
+            ),
           ),
-        ),
       stream: (input) =>
         stream(input).pipe(
           Stream.mapError(

apps/server/src/orchestration-v2/ThreadSettlementService.test.ts

@@ -15,7 +15,6 @@ import {
   type OrchestrationV2ThreadShell,
   type PullRequestSummary,
   type ServerSettings as ContractServerSettings,
-  type ServerSettingsPatch,
 } from "@t3tools/contracts";
 import { applyServerSettingsPatch } from "@t3tools/shared/serverSettings";
 import * as Crypto from "effect/Crypto";
@@ -505,9 +504,15 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
   const scriptRuns =
     yield* Queue.unbounded<ProjectSetupScriptRunner.ProjectSetupScriptRunnerInput>();
 
-  const updateSettings = (patch: ServerSettingsPatch) =>
+  const updateSettings: ServerSettings.ServerSettingsService["Service"]["updateSettings"] = (
+    patch,
+  ) =>
     Effect.gen(function* () {
-      const next = applyServerSettingsPatch(yield* Ref.get(settings), patch);
+      const current = yield* Ref.get(settings);
+      const next = applyServerSettingsPatch(
+        current,
+        typeof patch === "function" ? patch(current) : patch,
+      );
       yield* Ref.set(settings, next);
       yield* PubSub.publish(settingsChanges, next);
       return next;