perf(server): restore the fork's batched projection bootstrap

.t3-turbo/customizations.json

@@ -1381,6 +1381,32 @@
           ]
         }
       ]
+    },
+    {
+      "id": "batched-projection-bootstrap",
+      "status": "implemented",
+      "summary": "Cold start replays projection history in batches instead of one transaction per event. Upstream's ProjectionPipeline runs runProjectorForEvent per replayed event, so a multi-thousand-event history pays one sqlite transaction (and one thread shell-summary recompute per dirtying event) each time. The fork keeps upstream's live path and its shouldRefreshThreadShellSummary gate untouched, and adds a bootstrap-only batch path: PROJECTION_BOOTSTRAP_BATCH_SIZE (500) events per sql.withTransaction, shell-summary refreshes deferred into a per-batch Set and replayed once per thread when the batch closes, and a single projection-state upsert per batch. This was lost once already in the 0.0.45 ingest because it was never registered.",
+      "checks": [
+        {
+          "path": "apps/server/src/orchestration/Layers/ProjectionPipeline.ts",
+          "markers": [
+            "PROJECTION_BOOTSTRAP_BATCH_SIZE",
+            "runProjectorBatch",
+            "refreshOrDeferThreadShellSummary",
+            "deferredThreadShellSummaryIds",
+            "ProjectionApplyContext"
+          ]
+        },
+        {
+          "path": "apps/server/src/orchestration/Layers/ProjectionPipeline.turbo.test.ts",
+          "markers": [
+            "replays history in batched transactions with deduped shell refreshes",
+            "the threads projector should commit once per batch",
+            "shell summaries must refresh at most once per thread per batch",
+            "the live path must refresh the shell summary without waiting for a batch"
+          ]
+        }
+      ]
     }
   ]
 }

SEAM.md

@@ -464,6 +464,42 @@ On a nightly-sync conflict: keep upstream's `onExit` shape and re-apply the succ
 the `Effect.catchTags` recovery, and the `Effect.repeat` wrapper. Retire this seam if upstream merges
 its own durable config subscription — PR pingdotgg/t3code#7233 (issue #7231) is the candidate.
 
+## Batched projection bootstrap (fork perf)
+
+Upstream's `apps/server/src/orchestration/Layers/ProjectionPipeline.ts` bootstraps every projector by
+streaming history through `runProjectorForEvent`, which opens one `sql.withTransaction` per replayed
+event and recomputes a thread's shell summary inline for every event that dirties it. Cold start on a
+real database therefore pays a commit (and an fsync) per event, plus O(events) shell-summary
+recomputes on the same handful of threads.
+
+The fork leaves upstream's live path alone — `projectEvent` still calls `runProjectorForEvent`, and
+upstream's `shouldRefreshThreadShellSummary` gate still decides which events dirty a summary — and
+adds a bootstrap-only batch path:
+
+- `PROJECTION_BOOTSTRAP_BATCH_SIZE = 500`. `bootstrapProjector` reads from `lastAppliedSequence` in
+  pages of that size and stops when a page comes back empty.
+- `runProjectorBatch` runs the whole page inside **one** `sql.withTransaction`, applying the
+  projector per event with a `ProjectionApplyContext`.
+- The context carries `deferredThreadShellSummaryIds`, a `Set<ThreadId>`. While it is present,
+  `refreshOrDeferThreadShellSummary` records the thread instead of recomputing; each recorded thread
+  is refreshed exactly once at the end of the batch. With no context (the live path) it refreshes
+  immediately, which is what upstream's `ProjectionPipeline.test.ts` shell-update counts assert.
+- One `projectionStateRepository.upsert` per batch, stamped with the batch's last event. Attachment
+  side-effects still run outside the transaction, exactly as upstream does per event.
+
+This seam was **lost once already**: the 0.0.45 ingest took upstream's per-event pipeline wholesale
+because upstream's `ProjectionPipeline.test.ts` asserts exact shell-update counts and the fork code
+was not registered here. It is registered now, and
+`apps/server/src/orchestration/Layers/ProjectionPipeline.turbo.test.ts` pins both halves: 1200 events
+across three threads bootstrap in `ceil(n / 500)` projection-state commits with at most one shell
+refresh per thread per batch, and a single live event still refreshes its summary in the same call.
+
+On a nightly-sync conflict: take upstream's file, then re-add `ProjectionApplyContext`, the
+`context?` parameter on `ProjectorDefinition["apply"]`, `PROJECTION_BOOTSTRAP_BATCH_SIZE`,
+`refreshOrDeferThreadShellSummary` (wrapping, never replacing, upstream's
+`shouldRefreshThreadShellSummary` gate), `runProjectorBatch`, and the paging `bootstrapProjector`.
+Both test files must pass.
+
 ## Nightly sync conflicts
 
 Resolve against the new upstream file first, then reapply only the behavior above; never take the

apps/desktop/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@t3tools/desktop",
-  "version": "0.0.47",
+  "version": "0.0.48",
   "private": true,
   "type": "module",
   "main": "dist-electron/main.cjs",

apps/server/package.json

@@ -1,6 +1,6 @@
 {
   "name": "t3",
-  "version": "0.0.47",
+  "version": "0.0.48",
   "license": "MIT",
   "repository": {
     "type": "git",

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -98,6 +98,7 @@ interface ProjectorDefinition {
   readonly apply: (
     event: OrchestrationEvent,
     attachmentSideEffects: AttachmentSideEffects,
+    context?: ProjectionApplyContext,
   ) => Effect.Effect<void, ProjectionRepositoryError>;
 }
 
@@ -106,6 +107,26 @@ interface AttachmentSideEffects {
   readonly prunedThreadRelativePaths: Map<string, Set<string>>;
 }
 
+/**
+ * Batch-scoped bookkeeping handed to a projector while it replays history.
+ *
+ * Present only on the bootstrap path. When `deferredThreadShellSummaryIds` is
+ * set, a projector records the threads whose shell summary went stale instead
+ * of recomputing each one inline; `runProjectorBatch` refreshes every recorded
+ * thread exactly once when the batch closes. The live path passes no context,
+ * so `projectEvent` keeps refreshing immediately.
+ */
+interface ProjectionApplyContext {
+  readonly deferredThreadShellSummaryIds?: Set<ThreadId>;
+}
+
+/**
+ * Events replayed per bootstrap transaction. Replaying one event per
+ * transaction makes cold start pay an fsync per event; batching turns a
+ * multi-thousand-event history into a handful of commits.
+ */
+const PROJECTION_BOOTSTRAP_BATCH_SIZE = 500;
+
 const materializeAttachmentsForProjection = Effect.fn("materializeAttachmentsForProjection")(
   (input: { readonly attachments: ReadonlyArray<ChatAttachment> }) =>
     Effect.succeed(input.attachments.length === 0 ? [] : input.attachments),
@@ -620,9 +641,25 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
       });
     });
 
+    /**
+     * Refresh a thread's shell summary now, or record it for the end of the
+     * batch when replaying. Deferring collapses the N stale marks a thread
+     * accumulates inside one batch into a single recompute.
+     */
+    const refreshOrDeferThreadShellSummary = (
+      threadId: ThreadId,
+      context: ProjectionApplyContext | undefined,
+    ) => {
+      if (context?.deferredThreadShellSummaryIds !== undefined) {
+        context.deferredThreadShellSummaryIds.add(threadId);
+        return Effect.void;
+      }
+      return refreshThreadShellSummary(threadId);
+    };
+
     const applyThreadsProjection: ProjectorDefinition["apply"] = Effect.fn(
       "applyThreadsProjection",
-    )(function* (event, attachmentSideEffects) {
+    )(function* (event, attachmentSideEffects, context) {
       switch (event.type) {
         case "thread.created":
           yield* projectionThreadRepository.upsert({
@@ -913,7 +950,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
             updatedAt: event.occurredAt,
           });
           if (shouldRefreshThreadShellSummary(event)) {
-            yield* refreshThreadShellSummary(event.payload.threadId);
+            yield* refreshOrDeferThreadShellSummary(event.payload.threadId, context);
           }
           return;
         }
@@ -931,7 +968,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
             latestTurnId: event.payload.session.activeTurnId ?? existingRow.value.latestTurnId,
             updatedAt: event.occurredAt,
           });
-          yield* refreshThreadShellSummary(event.payload.threadId);
+          yield* refreshOrDeferThreadShellSummary(event.payload.threadId, context);
           return;
         }
 
@@ -947,7 +984,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
             latestTurnId: event.payload.turnId,
             updatedAt: event.occurredAt,
           });
-          yield* refreshThreadShellSummary(event.payload.threadId);
+          yield* refreshOrDeferThreadShellSummary(event.payload.threadId, context);
           return;
         }
 
@@ -985,7 +1022,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
             latestTurnId,
             updatedAt: event.occurredAt,
           });
-          yield* refreshThreadShellSummary(event.payload.threadId);
+          yield* refreshOrDeferThreadShellSummary(event.payload.threadId, context);
           return;
         }
 
@@ -1780,20 +1817,85 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
       );
     });
 
+    /**
+     * Bootstrap counterpart to `runProjectorForEvent`: one transaction for a
+     * whole batch, with every thread shell summary the batch dirtied refreshed
+     * once at the end and a single projection-state upsert for the batch's last
+     * event. Attachment side-effects still run outside the transaction.
+     */
+    const runProjectorBatch = Effect.fn("runProjectorBatch")(function* (
+      projector: ProjectorDefinition,
+      events: ReadonlyArray<OrchestrationEvent>,
+    ) {
+      const lastEvent = events.at(-1);
+      if (lastEvent === undefined) {
+        return;
+      }
+
+      const attachmentSideEffects: AttachmentSideEffects = {
+        deletedThreadIds: new Set<string>(),
+        prunedThreadRelativePaths: new Map<string, Set<string>>(),
+      };
+      const deferredThreadShellSummaryIds = new Set<ThreadId>();
+      const context: ProjectionApplyContext = { deferredThreadShellSummaryIds };
+
+      yield* sql.withTransaction(
+        Effect.forEach(events, (event) => projector.apply(event, attachmentSideEffects, context), {
+          concurrency: 1,
+          discard: true,
+        }).pipe(
+          Effect.andThen(
+            Effect.forEach(deferredThreadShellSummaryIds, refreshThreadShellSummary, {
+              concurrency: 1,
+              discard: true,
+            }),
+          ),
+          Effect.andThen(
+            projectionStateRepository.upsert({
+              projector: projector.name,
+              lastAppliedSequence: lastEvent.sequence,
+              updatedAt: lastEvent.occurredAt,
+            }),
+          ),
+        ),
+      );
+
+      yield* runAttachmentSideEffects(attachmentSideEffects).pipe(
+        Effect.catch((cause) =>
+          Effect.logWarning("failed to apply projected attachment side-effects", {
+            projector: projector.name,
+            sequence: lastEvent.sequence,
+            eventType: lastEvent.type,
+            cause,
+          }),
+        ),
+      );
+    });
+
     const bootstrapProjector = (projector: ProjectorDefinition) =>
       projectionStateRepository
         .getByProjector({
           projector: projector.name,
         })
         .pipe(
           Effect.flatMap((stateRow) =>
-            Stream.runForEach(
-              eventStore.readFromSequence(
-                Option.isSome(stateRow) ? stateRow.value.lastAppliedSequence : 0,
-                Number.MAX_SAFE_INTEGER,
-              ),
-              (event) => runProjectorForEvent(projector, event),
-            ),
+            Effect.gen(function* () {
+              let cursor = Option.isSome(stateRow) ? stateRow.value.lastAppliedSequence : 0;
+              while (true) {
+                const events = yield* eventStore
+                  .readFromSequence(cursor, PROJECTION_BOOTSTRAP_BATCH_SIZE)
+                  .pipe(
+                    Stream.runCollect,
+                    Effect.map((chunk) => Array.from(chunk)),
+                  );
+                const lastEvent = events.at(-1);
+                if (lastEvent === undefined) {
+                  return;
+                }
+                yield* runProjectorBatch(projector, events);
+                cursor = lastEvent.sequence;
+              }
+            }),
           ),
         );