processor.ts · Source

processor.tsapps/worker/src/processors/search-index/processor.tsraw ← search-index34,780 B← Back to the app
1/**
2 * apps/worker — search indexer orchestrator (TASK-0096).
3 *
4 * The single OpenSearch writer for the content index (search §0 "Single search
5 * writer"). Two entry points into the same builders:
6 *
7 *  1. STREAM — a StreamConsumer on `factory:events` (group `worker-search-index`)
8 *     reacting to content/permission/user/workspace events and upserting or
9 *     deleting the matching docs. The primary path. `permission.changed`
10 *     enqueues a workspace resync — the API never carries resource ids on that
11 *     event, so a targeted reindex is impossible by design (the 15-minute
12 *     reconcile sweep is the backstop).
13 *
14 *  2. QUEUE — BullMQ `search-index` for explicit ops:
15 *        {type:"reindex",   workspaceId?}   full (re)index in 500-doc batches
16 *        {type:"reconcile", workspaceId?}   contentHash sweep (source vs mirror)
17 *        {type:"delete",    docType, refId, workspaceId?}   explicit removal
18 *
19 * The index is ONE: `factory-content` (alias over `factory-content-v1`),
20 * created at startup with the `@factory/search` `contentMapping()` (dimension
21 * from env). Deletes (arch §1.1 + TASK-0096 §5): soft/trash REMOVES the doc
22 * from the index (search must not surface trashed content); a RESTORE re-adds
23 * it — the API publishes no restore domain event, so the re-add comes from the
24 * reconciliation sweep's source→index direction, not the stream. Purge removes
25 * doc + mirror rows outright.
26 *
27 * Idempotency: events are at-least-once; upserts are keyed by doc id
28 * (`<type>:<refId>`, user: `user:<ws>:<uid>`) so replays converge. A missing
29 * source row (poison) is logged + acked; a transport/DB failure RE-THROWS so
30 * the entry stays PENDING for XAUTOCLAIM redelivery — never ack an unwritten
31 * doc. After the BullMQ 8-attempt cap a failed job is recorded to the
32 * `search-index:dlq` stream (TASK-0096 §8).
33 */
34import { Worker, type Job } from "bullmq";
35import { withJobOtel } from "../../lib/job-otel.js";
36import type { Redis } from "ioredis";
37import { z } from "zod";
38import type { PrismaClient } from "@prisma/client";
39import { ApiError } from "@factory/shared/errors";
40import {
41  CONTENT_INDEX_ALIAS,
42  CHUNK_DOC_TYPE,
43  contentMapping,
44  vectorDimensionMatches,
45  type IndexDocInput,
46  type SearchClient,
47  type SearchType,
48} from "@factory/search";
49import { QUEUES, enqueue } from "../../queues.js";
50import { getLogger } from "../../logger.js";
51import type { WorkerEnv } from "../../config.js";
52import { StreamConsumer, type StreamEntry } from "../../lib/redis-stream.js";
53import { tryParseEvent, SEARCH_INDEX_EVENTS } from "../../events.js";
54import {
55  buildPageDoc,
56  buildBlockDocs,
57  buildDatabaseDoc,
58  buildRecordDoc,
59  buildCommentDoc,
60  buildFileDoc,
61  buildUserDoc,
62  contentHash,
63  upsertMirror,
64  deleteMirror,
65  mirrorRefId,
66  docIdFor,
67  expandChunkDocs,
68  type IndexerDeps,
69} from "./builders.js";
70 
71const log = getLogger("search-index");
72 
73const BATCH = 500; // TASK-0096 §7: index in batches of 500 (log per batch)
74const DLQ_STREAM = "search-index:dlq";
75const INDEX_VERSION = "v1"; // physical index suffix; dimension rollover = S5
76 
77export interface SearchIndexOps {
78  /** The @factory/search client (injectable fetch for tests). */
79  search: SearchClient;
80  prisma: PrismaClient;
81  /** Redis connection (DLQ stream + enqueues). */
82  connection: Redis;
83  env: WorkerEnv;
84  /**
85   * TASK-0125 — incremental RAG embedding hook: after a page/record is
86   * (re)indexed, enqueue its `embed:chunks` job on the `ai` queue. Wired at
87   * boot (index.ts); optional so the Docker-free unit tests pass plain ops.
88   * The hook is fire-and-forget (a failed enqueue must not fail the index
89   * write — the nightly `embed:reembed` reconciles any gap).
90   */
91  enqueueEmbed?: (job: { workspaceId: string; docType: "page" | "record"; refId: string }) => Promise<unknown> | unknown;
92}
93 
94const deps = (ops: SearchIndexOps): IndexerDeps => ({ prisma: ops.prisma });
95 
96// ---------------------------------------------------------------------------
97// Index management
98// ---------------------------------------------------------------------------
99 
100const contentIndexName = () => `factory-content-${INDEX_VERSION}`;
101 
102/**
103 * Startup: create the content index (env-driven dimension) behind the alias
104 * `factory-content`, idempotently. If the alias already points at an index
105 * with a DIFFERENT vector dimension, leave it alone and warn — that is the
106 * S5 rollover script's job (`dims` cannot change in place).
107 */
108export async function ensureContentIndex(ops: SearchIndexOps): Promise<{ created: boolean }> {
109  const { search, env } = ops;
110  const mapping = contentMapping({
111    dimension: env.SEARCH_VECTOR_DIMENSION,
112    m: env.SEARCH_KNN_M,
113    efConstruction: env.SEARCH_KNN_EF_CONSTRUCTION,
114  });
115  const name = contentIndexName();
116  const exists = await search.indexExists(name).catch(() => false);
117  if (!exists) {
118    // TASK-0100 S5 self-heal: if the ALIAS name is already taken by a REAL
119    // index (no mapping — OpenSearch was down at boot, so this boot's create
120    // was skipped, and the first `_bulk` auto-created a bare real index
121    // named `factory-content`), it has NO vector mapping and `createAlias`
122    // would reject it (an index cannot be aliased onto itself). Delete it
123    // and rebuild the versioned index + alias. This only ever runs when the
124    // versioned index is missing, so it is safe: the bare index holds no
125    // queryable state (reconcile/reindex rebuild from source).
126    if (!(await search.isAlias(CONTENT_INDEX_ALIAS).catch(() => false))) {
127      const bareReal = await search.indexExists(CONTENT_INDEX_ALIAS).catch(() => false);
128      if (bareReal) {
129        log.warn(
130          { index: CONTENT_INDEX_ALIAS },
131          "content alias name is a real (mapping-less) index — removing to rebuild versioned index + alias"
132        );
133        await search.deleteIndex(CONTENT_INDEX_ALIAS).catch((err: unknown) => {
134          log.error({ err: (err as Error).message }, "failed to delete bare content index — manual cleanup required");
135          throw err;
136        });
137      }
138    }
139    const created = await search.createIndexIfMissing(name, mapping);
140    if (created) await search.createAlias(name, CONTENT_INDEX_ALIAS);
141    log.info(
142      { index: name, alias: CONTENT_INDEX_ALIAS, dimension: env.SEARCH_VECTOR_DIMENSION },
143      "content index ensured"
144    );
145    return { created };
146  }
147  const liveDim = await search.getVectorDimension(name).catch(() => null);
148  if (liveDim !== null && !vectorDimensionMatches({ mappings: { properties: { vector: { dimension: liveDim } } } }, env.SEARCH_VECTOR_DIMENSION)) {
149    log.warn(
150      { index: name, liveDim, wanted: env.SEARCH_VECTOR_DIMENSION },
151      "content index dimension mismatch — alias rollover required (scripts/opensearch-reindex.ts)"
152    );
153  }
154  return { created: false };
155}
156 
157/**
158 * TASK-0100 S5: the boot-time `ensureContentIndex` is the ONLY create path,
159 * and it is SKIPPED when OpenSearch is down at boot (degrade-and-continue).
160 * The old boot comment promised the index would be "created lazily by the
161 * first (re)index" — but no such lazy path existed, so after a boot-with-
162 * OpenSearch-down, the first `_bulk` (targeting the alias name) auto-created
163 * a bare real index and every search was SEARCH_UNAVAILABLE forever.
164 * This guard makes the write paths self-heal: if the versioned index is
165 * missing, run `ensureContentIndex` first (idempotent, cheap when it exists).
166 */
167let contentIndexEnsured = false;
168async function ensureContentIndexOnce(ops: SearchIndexOps): Promise<void> {
169  if (contentIndexEnsured) return;
170  await ensureContentIndex(ops);
171  contentIndexEnsured = true;
172}
173 
174// ---------------------------------------------------------------------------
175// Write paths (doc → index + mirror)
176// ---------------------------------------------------------------------------
177 
178/**
179 * Bulk-index docs in 500-doc batches, mirroring each into Postgres.
180 *
181 * Chunk expansion (TASK-0099 §3): before writing, each page/record whole-doc
182 * is expanded via {@link expandChunkDocs} — bodies over the threshold also
183 * emit per-chunk docs (the kNN leg's target). Chunk docs are derived from the
184 * parent (they have NO `search_documents` mirror rows), so:
185 *   - the parent's PRE-EXISTING chunks are cleared first (a `_delete_by_query`
186 *     on `parentDocId`), because a shrunk body changes the chunk count and the
187 *     old ids can't be enumerated from the mirror. Every call passes the
188 *     COMPLETE per-parent set, so clear-then-write keeps the chunk set exact
189 *     (shrink/grow); it's an idempotent no-op when the parent has no chunks.
190 *   - chunk docs are NOT mirrored (only whole docs are).
191 */
192export async function indexDocs(ops: SearchIndexOps, docs: IndexDocInput[]): Promise<{ indexed: number }> {
193  await ensureContentIndexOnce(ops); // TASK-0100 S5 self-heal (index may not exist yet)
194  const chunkableParents = docs
195    .filter((d) => d.type === "page" || d.type === "record")
196    .map((d) => d.docId);
197  if (chunkableParents.length > 0) {
198    await ops.search.deleteChunksOf(chunkableParents);
199  }
200  const expanded = docs.flatMap((d) => expandChunkDocs(d));
201  let indexed = 0;
202  for (let i = 0; i < expanded.length; i += BATCH) {
203    const chunk = expanded.slice(i, i + BATCH);
204    const res = await ops.search.indexDocs(chunk);
205    for (const doc of chunk) {
206      if (doc.type !== CHUNK_DOC_TYPE) await upsertMirror(ops.prisma, doc); // chunks: no mirror row
207    }
208    indexed += res.indexed;
209    if (i + BATCH < expanded.length) log.info({ batch: i / BATCH + 1 }, "index batch");
210  }
211  // TASK-0125 — incremental RAG embedding: every page/record that was just
212  // (re)indexed gets its vectors refreshed on the `ai` queue. The event
213  // dispatch, the workspace reindex AND the 15-min reconcile all funnel
214  // through indexDocs, so this single hook covers the whole incremental
215  // surface (spec: "incremental (event-driven)"). Fire-and-forget: an
216  // enqueue failure must not fail the index write — the nightly re-embed
217  // closes any gap. The embed job re-derives the chunk set from the source,
218  // so the refId here is only a routing key.
219  if (ops.enqueueEmbed) {
220    for (const d of docs) {
221      if (d.type !== "page" && d.type !== "record") continue;
222      const refId = d.docId.split(":")[1] ?? d.docId;
223      try {
224        await ops.enqueueEmbed({ workspaceId: d.workspaceId, docType: d.type, refId });
225      } catch (err) {
226        log.warn(
227          { err: (err as Error).message, docId: d.docId },
228          "embed:chunks enqueue failed (nightly re-embed backstops)",
229        );
230      }
231    }
232  }
233  return { indexed };
234}
235 
236/**
237 * Remove one doc (index + mirror). A 404 on delete is fine (idempotent). For
238 * a page/record parent, its chunk docs are removed in a parent-driven pass
239 * (`deleteChunksOf` — chunks have no mirror rows to enumerate, TASK-0099 §3).
240 */
241export async function removeDoc(
242  ops: SearchIndexOps,
243  type: SearchType,
244  refId: string,
245  workspaceId?: string
246): Promise<void> {
247  const docId = type === "user" && workspaceId ? `user:${workspaceId}:${refId}` : `${type}:${refId}`;
248  await ensureContentIndexOnce(ops); // TASK-0100 S5 self-heal (index may not exist yet)
249  try {
250    await ops.search.deleteDoc(docId);
251  } catch (err) {
252    if (!(err instanceof ApiError && err.status === 404)) throw err;
253  }
254  if (type === "page" || type === "record") {
255    try {
256      await ops.search.deleteChunksOf(docId);
257    } catch (err) {
258      if (!(err instanceof ApiError && err.status === 404)) throw err;
259    }
260  }
261  await deleteMirror(ops.prisma, type, refId, workspaceId);
262}
263 
264// ---------------------------------------------------------------------------
265// Stream consumer
266// ---------------------------------------------------------------------------
267 
268/**
269 * Rebuild a page doc + its block docs (null-safe). Returns the docs to
270 * upsert plus the docIds to DELETE: a trashed OR purged page has no doc
271 * (arch §1.1 "trashed = deleted from index"; TASK-0096 §5 "page.deleted
272 * (soft/trash) removes the doc"). Block docs of a vanished page are
273 * enumerated from the MIRROR (their source rows may already be hard-deleted).
274 * Used by create/update/share/trash events.
275 */
276async function pageDocs(
277  ops: SearchIndexOps,
278  pageId: string,
279  workspaceId?: string
280): Promise<{ docs: IndexDocInput[]; deleteIds: string[] }> {
281  const d = deps(ops);
282  const [pageDoc, blockDocs] = await Promise.all([buildPageDoc(d, pageId), buildBlockDocs(d, pageId)]);
283  if (pageDoc || blockDocs.length > 0) {
284    return { docs: [pageDoc, ...blockDocs].filter((x): x is IndexDocInput => x !== null), deleteIds: [] };
285  }
286  // Vanished (trashed/purged): remove the page doc + its surviving block docs.
287  const orphanBlocks = await ops.prisma.searchDocument.findMany({
288    where: { ...(workspaceId ? { workspaceId } : {}), type: "block", metadata: { path: ["pageId"], equals: pageId } },
289    select: { refId: true },
290  });
291  const deleteIds = [`page:${pageId}`, ...orphanBlocks.map((b) => `block:${b.refId}`)];
292  return { docs: [], deleteIds };
293}
294 
295/**
296 * The event dispatch for one parsed `factory:events` entry (exported for
297 * Docker-free unit tests). Rethrows transport/DB failures so the StreamConsumer
298 * keeps the entry PENDING (never ack an unwritten doc). Missing source rows
299 * (poison) log + ack; the 15-min reconcile backstops.
300 */
301export async function handleSearchEvent(
302  ops: SearchIndexOps,
303  envelope: { event: string; payload: Record<string, unknown> }
304): Promise<void> {
305  if (!SEARCH_INDEX_EVENTS.includes(envelope.event as (typeof SEARCH_INDEX_EVENTS)[number])) {
306    return; // not mine — XACK (the group is per-queue, not per-event)
307  }
308  const p = envelope.payload;
309  const t0 = Date.now();
310  switch (envelope.event) {
311    case "page.created":
312    case "page.updated":
313    case "page.shared": {
314      const ws = str(p.workspaceId);
315      const pageId = str(p.pageId ?? p.id);
316      if (!pageId) return;
317      const { docs, deleteIds } = await pageDocs(ops, pageId, ws || undefined);
318      if (deleteIds.length > 0) {
319        // Trashed (soft delete) or purged: the doc is REMOVED from the index
320        // (TASK-0096 §5 / arch §1.1). Restore re-adds it via the
321        // reconciliation sweep (the API publishes no restore domain event).
322        for (const id of deleteIds) {
323          const [type, ...rest] = id.split(":");
324          await removeDoc(ops, type as SearchType, rest.join(":"), ws || undefined);
325        }
326        log.info({ pageId, removed: deleteIds.length, ms: Date.now() - t0 }, "page removed from search (trashed/purged)");
327        break;
328      }
329      await indexDocs(ops, docs);
330      log.info({ event: envelope.event, pageId, docs: docs.length, ms: Date.now() - t0 }, "page indexed");
331      break;
332    }
333    case "page.deleted": {
334      // Both soft (trash) and hard (purge) REMOVE the doc(s) from the index —
335      // the source rows are already trashed/purged by the API when this event
336      // arrives, so `pageDocs` returns deleteIds in both cases (unified with
337      // the page.created/updated handler). Purge additionally lists the
338      // descendant pageIds explicitly.
339      const ws = str(p.workspaceId);
340      const rootPage = str(p.pageId ?? p.id);
341      const pageIds: string[] = Array.isArray(p.pageIds)
342        ? (p.pageIds as unknown[]).filter((x): x is string => typeof x === "string")
343        : [];
344      const all = [...new Set([rootPage, ...pageIds].filter(Boolean))];
345      for (const pageId of all) {
346        const { deleteIds } = await pageDocs(ops, pageId, ws || undefined);
347        for (const id of deleteIds) {
348          const [type, ...rest] = id.split(":");
349          await removeDoc(ops, type as SearchType, rest.join(":"), ws || undefined);
350        }
351      }
352      log.info({ pageIds: all, permanent: p.permanent === true, ms: Date.now() - t0 }, "page(s) removed from search");
353      break;
354    }
355    case "database.created":
356    case "database.updated": {
357      const databaseId = str(p.databaseId ?? p.id);
358      if (!databaseId) return;
359      const doc = await buildDatabaseDoc(deps(ops), databaseId);
360      if (doc) {
361        await indexDocs(ops, [doc]);
362        log.info({ databaseId, ms: Date.now() - t0 }, "database indexed");
363      } else {
364        log.warn({ databaseId }, "database source row missing — poison, ack (reconcile backstops)");
365      }
366      break;
367    }
368    case "record.created":
369    case "record.updated": {
370      const recordId = str(p.recordId ?? p.id);
371      if (!recordId) return;
372      const doc = await buildRecordDoc(deps(ops), recordId);
373      if (doc) {
374        await indexDocs(ops, [doc]);
375        log.info({ recordId, ms: Date.now() - t0 }, "record indexed");
376      } else {
377        log.warn({ recordId }, "record source row missing — poison, ack (reconcile backstops)");
378      }
379      break;
380    }
381    case "comment.created": {
382      const commentId = str(p.commentId ?? p.id);
383      if (!commentId) return;
384      const doc = await buildCommentDoc(deps(ops), commentId);
385      if (doc) {
386        await indexDocs(ops, [doc]);
387        log.info({ commentId, ms: Date.now() - t0 }, "comment indexed");
388      } else {
389        log.warn({ commentId }, "comment source row missing — poison, ack (reconcile backstops)");
390      }
391      break;
392    }
393    case "comment.deleted": {
394      // Hard delete in the API → delete the doc now.
395      const commentId = str(p.commentId ?? p.id);
396      if (!commentId) return;
397      await removeDoc(ops, "comment", commentId, str(p.workspaceId) || undefined);
398      log.info({ commentId, ms: Date.now() - t0 }, "comment deleted from search");
399      break;
400    }
401    case "file.uploaded":
402    case "file.finalized":
403    case "file.updated": {
404      // `file.uploaded` fires at PRESIGN (the File row may still be UPLOADING
405      // with null mime/size); `file.finalized` is published by the files worker
406      // at S3 finalization with the real mime/size; `file.updated` fires on a
407      // rename (TASK-0104 §8) so the doc's title converges immediately. The
408      // doc converges on whichever arrives (reconcile backstops).
409      const fileId = str(p.fileId ?? p.id);
410      if (!fileId) return;
411      const doc = await buildFileDoc(deps(ops), fileId);
412      if (doc) {
413        await indexDocs(ops, [doc]);
414        log.info({ fileId, ms: Date.now() - t0 }, "file indexed (mime/size may converge via resync)");
415      } else {
416        log.warn({ fileId }, "file source row missing — poison, ack (reconcile backstops)");
417      }
418      break;
419    }
420    case "file.deleted": {
421      // TASK-0104 §3: the file went to the trash → remove the search doc now
422      // (the acceptance criterion is "index doc gone within 30 s"). An
423      // EXPLICIT removal is required, not a re-upsert: `buildFileDoc` returns
424      // null for a `deletedAt` row (the upsert path would be a no-op), so the
425      // doc has to be deleted by id. Restore re-indexes via the
426      // `file.uploaded {restored:true}` re-emit; a purge's row delete is
427      // backstopped by the 15-min contentHash reconcile (the doc's source row
428      // is gone → the sweep removes it).
429      const fileId = str(p.fileId ?? p.id);
430      if (!fileId) return;
431      await removeDoc(ops, "file", fileId, str(p.workspaceId) || undefined);
432      log.info({ fileId, ms: Date.now() - t0 }, "file removed from search (trashed)");
433      break;
434    }
435    case "permission.changed": {
436      // The API publishes {workspaceId, userId?, role?|removed?} — never
437      // resource ids — so a targeted reindex is impossible by design (search
438      // §0; recorded in TASK-0096). Reindex the whole workspace via the
439      // maintenance queue; the 15-min reconcile sweep backstops.
440      const ws = str(p.workspaceId);
441      if (!ws) return;
442      await enqueue(ops.connection, QUEUES.maintenance, { type: "workspace-resync", workspaceId: ws });
443      log.info({ workspaceId: ws, ms: Date.now() - t0 }, "permission changed → workspace resync enqueued");
444      break;
445    }
446    case "user.register":
447    case "user.login": {
448      const userId = str(p.userId ?? p.id);
449      if (!userId) return;
450      const ws = str(p.workspaceId);
451      if (!ws) {
452        log.warn({ userId }, "user event without workspaceId — nothing to stamp (reconcile covers)");
453        return;
454      }
455      const doc = await buildUserDoc(deps(ops), userId, ws);
456      if (doc) await indexDocs(ops, [doc]);
457      break;
458    }
459    case "workspace.created": {
460      const ws = str(p.workspaceId);
461      if (!ws) return;
462      await runReindex(ops, ws);
463      log.info({ workspaceId: ws, ms: Date.now() - t0 }, "workspace created → initial index");
464      break;
465    }
466    default:
467      return;
468  }
469}
470 
471export function createSearchIndexConsumer(opts: {
472  redis: Redis;
473  ops: SearchIndexOps;
474  consumerId: string;
475}): StreamConsumer {
476  const { redis, ops, consumerId } = opts;
477  return new StreamConsumer(
478    redis,
479    "factory:events",
480    "worker-search-index",
481    consumerId,
482    async (entry: StreamEntry) => {
483      const parsed = tryParseEvent(entry);
484      if (!parsed) return; // poison envelope: logged, XACK
485      try {
486        await handleSearchEvent(ops, parsed.envelope);
487      } catch (err) {
488        // Transport/DB failure: rethrow → entry stays PENDING (XAUTOCLAIM
489        // redelivers). Never ack a doc we failed to write.
490        log.error({ err: (err as Error).message, event: parsed.envelope.event, id: entry.id }, "search index handler failed — entry pending");
491        throw err;
492      }
493    }
494  );
495}
496 
497// ---------------------------------------------------------------------------
498// Full (re)index — all types of a workspace (or every workspace)
499// ---------------------------------------------------------------------------
500 
501export interface ReindexCounts {
502  workspaces: number;
503  docs: number;
504  orphansRemoved: number;
505}
506 
507/**
508 * Collect the FULL set of searchable docs for a workspace straight from the
509 * source tables (source→index direction). Trashed/purged pages and deleted
510 * files build to null (they contribute nothing); archived records build with
511 * `archived: true`. Shared by the full reindex and the reconciliation sweep.
512 */
513async function collectWorkspaceDocs(ops: SearchIndexOps, ws: string): Promise<IndexDocInput[]> {
514  const { prisma } = ops;
515  const d = deps(ops);
516  const [pages, databases, records, comments, files, members] = await Promise.all([
517    prisma.page.findMany({ where: { workspaceId: ws, deletedAt: null }, select: { id: true } }),
518    prisma.database.findMany({ where: { workspaceId: ws }, select: { id: true } }),
519    prisma.databaseRecord.findMany({ where: { workspaceId: ws }, select: { id: true } }),
520    prisma.comment.findMany({ where: { workspaceId: ws, pageId: { not: null } }, select: { id: true } }),
521    prisma.file.findMany({ where: { workspaceId: ws, deletedAt: null }, select: { id: true } }),
522    prisma.workspaceMember.findMany({ where: { workspaceId: ws }, select: { userId: true } }),
523  ]);
524 
525  const docs: IndexDocInput[] = [];
526  for (const pg of pages) docs.push(...(await pageDocs(ops, pg.id, ws).then((r) => r.docs)));
527  for (const db of databases) {
528    const doc = await buildDatabaseDoc(d, db.id);
529    if (doc) docs.push(doc);
530  }
531  for (const r of records) {
532    const doc = await buildRecordDoc(d, r.id);
533    if (doc) docs.push(doc);
534  }
535  for (const c of comments) {
536    const doc = await buildCommentDoc(d, c.id);
537    if (doc) docs.push(doc);
538  }
539  for (const f of files) {
540    const doc = await buildFileDoc(d, f.id);
541    if (doc) docs.push(doc);
542  }
543  // User docs: every active member (collaborator search, per-workspace doc).
544  for (const m of members) {
545    const doc = await buildUserDoc(d, m.userId, ws);
546    if (doc) docs.push(doc);
547  }
548  return docs;
549}
550 
551export async function runReindex(ops: SearchIndexOps, workspaceId?: string, orgId?: string): Promise<ReindexCounts> {
552  await ensureContentIndexOnce(ops); // TASK-0100 S5 self-heal (index may not exist yet)
553  const { prisma } = ops;
554  // T-156: org-scoped reindex (the org-admin trigger). The workspace set is
555  // filtered by `organizationId` — one org's admin can never reindex (and
556  // thus rewrite the grant snapshots of) another org's workspaces.
557  const workspaces = workspaceId
558    ? [workspaceId]
559    : (
560        await prisma.workspace.findMany({
561          where: orgId ? { organizationId: orgId } : {},
562          select: { id: true }
563        })
564      ).map((w) => w.id);
565  const counts: ReindexCounts = { workspaces: 0, docs: 0, orphansRemoved: 0 };
566 
567  for (const ws of workspaces) {
568    const docs = await collectWorkspaceDocs(ops, ws);
569    await indexDocs(ops, docs);
570 
571    // Orphans: mirror rows whose source row is gone (purged / deleted /
572    // demoted member) → remove from index + mirror. Content DRIFT (source
573    // changed without an event) is the reconcile sweep's job, not this one.
574    const live = new Set(docs.map((doc) => `${doc.type}:${mirrorRefId(doc)}`));
575    const mirror = await prisma.searchDocument.findMany({
576      where: { workspaceId: ws },
577      select: { type: true, refId: true },
578    });
579    for (const row of mirror) {
580      if (live.has(`${row.type}:${row.refId}`)) continue;
581      const refId = row.type === "user" ? row.refId.split(":").pop() ?? row.refId : row.refId;
582      await removeDoc(ops, row.type as SearchType, refId, ws);
583      counts.orphansRemoved++;
584    }
585 
586    counts.docs += docs.length;
587    counts.workspaces++;
588    log.info({ workspaceId: ws, docs: docs.length, orphans: counts.orphansRemoved }, "reindex workspace done");
589  }
590  log.info(counts, "reindex complete");
591  return counts;
592}
593 
594// ---------------------------------------------------------------------------
595// ContentHash reconciliation sweep (search §1.3/§7.3 — supersedes the count
596// heuristic; subsumes the old `reconcileSearch`).
597// ---------------------------------------------------------------------------
598 
599export interface ReconcileResult {
600  workspaces: number;
601  checked: number;
602  repaired: number;
603  removed: number;
604}
605 
606/** Rebuild one doc from its mirror row (null = source row gone). */
607async function docFromMirrorRow(
608  ops: SearchIndexOps,
609  row: { type: string; refId: string },
610  workspaceId: string
611): Promise<IndexDocInput | null> {
612  const d = deps(ops);
613  switch (row.type as SearchType) {
614    case "page":
615      return buildPageDoc(d, row.refId);
616    case "block":
617      return blockDocById(d, row.refId);
618    case "database":
619      return buildDatabaseDoc(d, row.refId);
620    case "record":
621      return buildRecordDoc(d, row.refId);
622    case "comment":
623      return buildCommentDoc(d, row.refId);
624    case "file":
625      return buildFileDoc(d, row.refId);
626    case "user": {
627      const uid = row.refId.split(":").pop() ?? row.refId;
628      return buildUserDoc(d, uid, workspaceId);
629    }
630    default:
631      return null; // unknown type (legacy mirror row) → caller removes
632  }
633}
634 
635/** A single block doc by block id (reconcile path). */
636async function blockDocById(d: IndexerDeps, blockId: string): Promise<IndexDocInput | null> {
637  const block = await d.prisma.block.findUnique({
638    where: { id: blockId },
639    select: { id: true, pageId: true, deletedAt: true },
640  });
641  if (!block || block.deletedAt !== null || !block.pageId) return null;
642  const docs = await buildBlockDocs(d, block.pageId);
643  return docs.find((x) => x.blockId === blockId) ?? null;
644}
645 
646export async function runReconcile(ops: SearchIndexOps, workspaceId?: string): Promise<ReconcileResult> {
647  await ensureContentIndexOnce(ops); // TASK-0100 S5 self-heal (index may not exist yet)
648  const { prisma } = ops;
649  const workspaces = workspaceId
650    ? [workspaceId]
651    : (await prisma.workspace.findMany({ select: { id: true } })).map((w) => w.id);
652  const result: ReconcileResult = { workspaces: 0, checked: 0, repaired: 0, removed: 0 };
653 
654  for (const ws of workspaces) {
655    const rows = await prisma.searchDocument.findMany({
656      where: { workspaceId: ws },
657      select: { type: true, refId: true, contentHash: true, indexedAt: true },
658    });
659    for (let i = 0; i < rows.length; i += BATCH) {
660      const batch = rows.slice(i, i + BATCH);
661      // Index-loss check: which docs are MISSING from OpenSearch? The
662      // contentHash sweep alone only catches SOURCE drift (changed content);
663      // a doc deleted/lost directly in the index keeps a healthy mirror row.
664      // Batched `_mget` (docExists) so a full sweep costs ~rows/500 requests.
665      let present = new Set<string>();
666      try {
667        present = await ops.search.docExists(batch.map((r) => docIdFor(r.type as SearchType, r.refId)));
668      } catch (err) {
669        // OpenSearch unreachable: treat everything as present (rethrowing here
670        // would abort the whole sweep; the next run / DLQ covers transport).
671        log.warn({ err: (err as Error).message, workspaceId: ws }, "docExists failed — index-loss check skipped for batch");
672        present = new Set(batch.map((r) => docIdFor(r.type as SearchType, r.refId)));
673      }
674      for (const row of batch) {
675        result.checked++;
676        const doc = await docFromMirrorRow(ops, row, ws);
677        if (doc === null) {
678          const refId = row.type === "user" ? row.refId.split(":").pop() ?? row.refId : row.refId;
679          await removeDoc(ops, row.type as SearchType, refId, ws);
680          result.removed++;
681          continue;
682        }
683        const docId = docIdFor(row.type as SearchType, row.refId);
684        if (!present.has(docId)) {
685          // Lost in the index → reindex regardless of hash.
686          await indexDocs(ops, [doc]);
687          result.repaired++;
688          continue;
689        }
690        if (row.contentHash !== null && row.contentHash === contentHash(doc)) continue; // in sync
691        await indexDocs(ops, [doc]);
692        result.repaired++;
693      }
694      if (i + BATCH < rows.length) {
695        log.info({ workspaceId: ws, batch: i / BATCH + 1, totalBatches: Math.ceil(rows.length / BATCH) }, "reconcile batch");
696      }
697    }
698 
699    // Direction B (source → index): re-add docs that exist in the SOURCE but
700    // not in the index AND not in the mirror. The mirror-only direction above
701    // cannot see these: e.g. a RESTORED page (the API publishes no restore
702    // domain event, so the mirror row was dropped with the trash) or a page
703    // created whose index write failed before the mirror upsert.
704    const live = await collectWorkspaceDocs(ops, ws);
705    const covered = new Set(
706      rows.map((r) => `${r.type}:${r.type === "user" ? (r.refId.split(":").pop() ?? r.refId) : r.refId}`)
707    );
708    const missing: IndexDocInput[] = [];
709    for (const doc of live) {
710      if (covered.has(`${doc.type}:${mirrorRefId(doc)}`)) continue;
711      let present = false;
712      try {
713        // `live` holds whole docs only (collectWorkspaceDocs never emits
714        // chunk docs), so `doc.type` is a public SearchType here.
715        const id = docIdFor(doc.type as SearchType, doc.id);
716        present = (await ops.search.docExists([id])).has(id);
717      } catch {
718        present = true; // unreachable → don't re-add what we can't verify
719      }
720      if (!present) missing.push(doc);
721    }
722    if (missing.length > 0) {
723      await indexDocs(ops, missing);
724      result.repaired += missing.length;
725      log.info({ workspaceId: ws, readded: missing.length }, "reconcile: source→index re-adds (restored/lost before mirror)");
726    }
727 
728    result.workspaces++;
729  }
730  log.info(result, "reconcile complete");
731  return result;
732}
733 
734// ---------------------------------------------------------------------------
735// DLQ (TASK-0096 §8)
736// ---------------------------------------------------------------------------
737 
738/** Record a permanently-failed search-index job to the DLQ stream. */
739export async function recordSearchDlq(connection: Redis, job: Job, err: Error): Promise<void> {
740  const data = (job.data ?? {}) as Record<string, unknown>;
741  await connection.xadd(
742    DLQ_STREAM,
743    "MAXLEN",
744    "~",
745    "1000",
746    "*",
747    "jobId",
748    job.id ?? "",
749    "jobType",
750    str(data.type) || "unknown",
751    "workspaceId",
752    str(data.workspaceId),
753    "error",
754    err.message,
755    "ts",
756    new Date().toISOString()
757  );
758  log.error({ jobId: job.id, err: err.message }, "search-index job dead-lettered");
759}
760 
761// ---------------------------------------------------------------------------
762// BullMQ worker (explicit ops)
763// ---------------------------------------------------------------------------
764 
765export const searchIndexJobSchema = z.discriminatedUnion("type", [
766  z.object({ type: z.literal("reindex"), workspaceId: z.string().min(1).optional(), orgId: z.string().min(1).optional() }),
767  z.object({ type: z.literal("reconcile"), workspaceId: z.string().min(1).optional() }),
768  z.object({
769    type: z.literal("delete"),
770    docType: z.enum(["page", "block", "database", "record", "comment", "file", "user"]),
771    refId: z.string().min(1),
772    workspaceId: z.string().min(1).optional(),
773  }),
774]);
775export type SearchIndexJobData = z.infer<typeof searchIndexJobSchema>;
776 
777export function createSearchIndexWorker(opts: {
778  connection: Redis;
779  ops: SearchIndexOps;
780  concurrency?: number;
781}): Worker {
782  const { connection, ops, concurrency } = opts;
783  return new Worker(
784    QUEUES.searchIndex,
785    withJobOtel(QUEUES.searchIndex, async (job: Job) => {
786      const data = searchIndexJobSchema.parse(job.data);
787      const started = Date.now();
788      switch (data.type) {
789        case "reindex": {
790          const counts = await runReindex(ops, data.workspaceId, data.orgId);
791          log.info({ jobId: job.id, counts, ms: Date.now() - started }, "reindex done");
792          return counts;
793        }
794        case "reconcile": {
795          const result = await runReconcile(ops, data.workspaceId);
796          log.info({ jobId: job.id, result, ms: Date.now() - started }, "reconcile done");
797          // T-177 (search health tab): stamp the last-reconcile time for the
798          // org-admin console (best-effort — a stamp failure must never fail
799          // an otherwise-successful sweep; the 15-minute cadence makes a
800          // missed stamp self-healing).
801          try {
802            await connection.set("search:reconcile:last", new Date().toISOString(), "EX", 30 * 86_400);
803          } catch (err) {
804            log.warn({ jobId: job.id, err: String(err) }, "search:reconcile:last stamp failed (non-fatal)");
805          }
806          return result;
807        }
808        case "delete": {
809          await removeDoc(ops, data.docType, data.refId, data.workspaceId);
810          return { deleted: true };
811        }
812      }
813    }),
814    {
815      connection,
816      ...(concurrency !== undefined ? { concurrency } : {})
817    }
818  );
819}
820 
821function str(v: unknown): string {
822  return typeof v === "string" ? v : "";
823}
824