Move thread search SQLite reads off the serving loop

apps/server/src/routes/threads/base.ts

@@ -8,7 +8,7 @@ import {
   listThreadMentionRowsByIds,
   listThreadsWithPendingInteractionStateOffThread,
   markThreadDeleted,
-  searchThreadsWithPendingInteractionState,
+  searchThreadsWithPendingInteractionStateOffThread,
   updateThread,
   type ThreadSearchResultGroup as DbThreadSearchResultGroup,
   type UpdateThreadInput,
@@ -309,7 +309,7 @@ export function registerThreadBaseRoutes(app: Hono, deps: AppDeps): void {
     );
   });
 
-  get(routes.search, (context, query) => {
+  get(routes.search, async (context, query) => {
     const searchQuery = query.query.trim();
     if (countNonWhitespaceChars(searchQuery) < 2) {
       throw new ApiError(
@@ -321,10 +321,10 @@ export function registerThreadBaseRoutes(app: Hono, deps: AppDeps): void {
     const limitPerGroup = parseSearchLimitPerGroup(query.limitPerGroup);
     return context.json(
       buildThreadSearchResponse(deps, {
-        ...searchThreadsWithPendingInteractionState(deps.db, {
+        ...(await searchThreadsWithPendingInteractionStateOffThread(deps.db, {
           query: searchQuery,
           limitPerGroup,
-        }),
+        })),
       }) satisfies ThreadSearchResponse,
     );
   });

packages/db/src/index.ts

@@ -2,6 +2,7 @@ export { createConnection } from "./connection.js";
 export {
   listThreadsWithPendingInteractionStateForProjectsOffThread,
   listThreadsWithPendingInteractionStateOffThread,
+  searchThreadsWithPendingInteractionStateOffThread,
   startSqliteReadWorker,
   stopSqliteReadWorker,
 } from "./sqlite-read-queue.js";

packages/db/src/sqlite-read-queue.ts

@@ -5,8 +5,11 @@ import type { DbConnection } from "./connection.js";
 import {
   listThreadsWithPendingInteractionState,
   listThreadsWithPendingInteractionStateForProjects,
+  searchThreadsWithPendingInteractionState,
   type ListThreadsForProjectsOptions,
   type ListThreadsOptions,
+  type SearchThreadsWithPendingInteractionStateArgs,
+  type ThreadSearchResults,
   type ThreadWithPendingInteractionState,
 } from "./data/threads.js";
 import type {
@@ -16,7 +19,7 @@ import type {
 
 type PendingRead = {
   reject: (error: Error) => void;
-  resolve: (result: ThreadWithPendingInteractionState[]) => void;
+  resolve: (result: unknown) => void;
 };
 
 let worker: Worker | null = null;
@@ -74,7 +77,7 @@ export function startSqliteReadWorker(args: {
     }
     pending.delete(response.id);
     if (response.ok) {
-      request.resolve(response.result as ThreadWithPendingInteractionState[]);
+      request.resolve(response.result);
       return;
     }
     request.reject(new Error(response.error));
@@ -102,9 +105,9 @@ export async function stopSqliteReadWorker(): Promise<void> {
   await current.terminate();
 }
 
-async function request(
+async function request<T>(
   message: Omit<SqliteReadRequest, "id">,
-): Promise<ThreadWithPendingInteractionState[]> {
+): Promise<T> {
   const current = worker;
   if (current === null) {
     throw new Error("sqlite read worker is not running");
@@ -124,7 +127,7 @@ export async function listThreadsWithPendingInteractionStateOffThread(
   if (worker === null) {
     return listThreadsWithPendingInteractionState(db, options);
   }
-  return request({
+  return request<ThreadWithPendingInteractionState[]>({
     name: "listThreadsWithPendingInteractionState",
     args: options,
   });
@@ -137,8 +140,21 @@ export async function listThreadsWithPendingInteractionStateForProjectsOffThread
   if (worker === null) {
     return listThreadsWithPendingInteractionStateForProjects(db, options);
   }
-  return request({
+  return request<ThreadWithPendingInteractionState[]>({
     name: "listThreadsWithPendingInteractionStateForProjects",
     args: options,
   });
 }
+
+export async function searchThreadsWithPendingInteractionStateOffThread(
+  db: DbConnection,
+  args: SearchThreadsWithPendingInteractionStateArgs,
+): Promise<ThreadSearchResults> {
+  if (worker === null) {
+    return searchThreadsWithPendingInteractionState(db, args);
+  }
+  return request<ThreadSearchResults>({
+    name: "searchThreadsWithPendingInteractionState",
+    args,
+  });
+}

packages/db/src/sqlite-read-worker.ts

@@ -3,8 +3,10 @@ import { createConnection } from "./connection.js";
 import {
   listThreadsWithPendingInteractionState,
   listThreadsWithPendingInteractionStateForProjects,
+  searchThreadsWithPendingInteractionState,
   type ListThreadsForProjectsOptions,
   type ListThreadsOptions,
+  type SearchThreadsWithPendingInteractionStateArgs,
 } from "./data/threads.js";
 
 export type SqliteReadRequest =
@@ -17,6 +19,11 @@ export type SqliteReadRequest =
       id: number;
       name: "listThreadsWithPendingInteractionStateForProjects";
       args: ListThreadsForProjectsOptions;
+    }
+  | {
+      id: number;
+      name: "searchThreadsWithPendingInteractionState";
+      args: SearchThreadsWithPendingInteractionStateArgs;
     };
 
 export type SqliteReadResponse =
@@ -36,7 +43,9 @@ port.on("message", (request: SqliteReadRequest) => {
     const result =
       request.name === "listThreadsWithPendingInteractionState"
         ? listThreadsWithPendingInteractionState(db, request.args)
-        : listThreadsWithPendingInteractionStateForProjects(db, request.args);
+        : request.name === "listThreadsWithPendingInteractionStateForProjects"
+          ? listThreadsWithPendingInteractionStateForProjects(db, request.args)
+          : searchThreadsWithPendingInteractionState(db, request.args);
     const response: SqliteReadResponse = { id: request.id, ok: true, result };
     port.postMessage(response);
   } catch (error) {

packages/db/test/sqlite-read-queue.test.ts

@@ -11,13 +11,16 @@ import {
   createThread,
   listThreadsWithPendingInteractionState,
   listThreadsWithPendingInteractionStateForProjects,
+  searchThreadsWithPendingInteractionState,
+  updateThread,
 } from "../src/data/threads.js";
 import { migrate } from "../src/migrate.js";
 import { noopNotifier } from "../src/notifier.js";
 import {
   isSqliteReadWorkerActive,
   listThreadsWithPendingInteractionStateForProjectsOffThread,
   listThreadsWithPendingInteractionStateOffThread,
+  searchThreadsWithPendingInteractionStateOffThread,
   startSqliteReadWorker,
   stopSqliteReadWorker,
 } from "../src/sqlite-read-queue.js";
@@ -101,4 +104,18 @@ describe("sqlite read queue", () => {
     expect(fromWorker).toEqual(fromServing);
     expect(fromProjectsWorker).toEqual(fromProjectsServing);
   });
+
+  it("returns the same thread search from a file-backed worker as the serving connection", async () => {
+    const { db, source, thread } = createFileDatabase();
+    updateThread(db, noopNotifier, thread.id, {
+      title: "alpha search target",
+    });
+    startSqliteReadWorker({ source });
+    expect(isSqliteReadWorkerActive()).toBe(true);
+
+    const args = { query: "alpha", limitPerGroup: 10 };
+    await expect(
+      searchThreadsWithPendingInteractionStateOffThread(db, args),
+    ).resolves.toEqual(searchThreadsWithPendingInteractionState(db, args));
+  });
 });