/**
 * apps/worker — search indexer orchestrator (TASK-0096).
 *
 * The single OpenSearch writer for the content index (search §0 "Single search
 * writer"). Two entry points into the same builders:
 *
 *  1. STREAM — a StreamConsumer on `factory:events` (group `worker-search-index`)
 *     reacting to content/permission/user/workspace events and upserting or
 *     deleting the matching docs. The primary path. `permission.changed`
 *     enqueues a workspace resync — the API never carries resource ids on that
 *     event, so a targeted reindex is impossible by design (the 15-minute
 *     reconcile sweep is the backstop).
 *
 *  2. QUEUE — BullMQ `search-index` for explicit ops:
 *        {type:"reindex",   workspaceId?}   full (re)index in 500-doc batches
 *        {type:"reconcile", workspaceId?}   contentHash sweep (source vs mirror)
 *        {type:"delete",    docType, refId, workspaceId?}   explicit removal
 *
 * The index is ONE: `factory-content` (alias over `factory-content-v1`),
 * created at startup with the `@factory/search` `contentMapping()` (dimension
 * from env). Deletes (arch §1.1 + TASK-0096 §5): soft/trash REMOVES the doc
 * from the index (search must not surface trashed content); a RESTORE re-adds
 * it — the API publishes no restore domain event, so the re-add comes from the
 * reconciliation sweep's source→index direction, not the stream. Purge removes
 * doc + mirror rows outright.
 *
 * Idempotency: events are at-least-once; upserts are keyed by doc id
 * (`<type>:<refId>`, user: `user:<ws>:<uid>`) so replays converge. A missing
 * source row (poison) is logged + acked; a transport/DB failure RE-THROWS so
 * the entry stays PENDING for XAUTOCLAIM redelivery — never ack an unwritten
 * doc. After the BullMQ 8-attempt cap a failed job is recorded to the
 * `search-index:dlq` stream (TASK-0096 §8).
 */
import { Worker, type Job } from "bullmq";
import { withJobOtel } from "../../lib/job-otel.js";
import type { Redis } from "ioredis";
import { z } from "zod";
import type { PrismaClient } from "@prisma/client";
import { ApiError } from "@factory/shared/errors";
import {
  CONTENT_INDEX_ALIAS,
  CHUNK_DOC_TYPE,
  contentMapping,
  vectorDimensionMatches,
  type IndexDocInput,
  type SearchClient,
  type SearchType,
} from "@factory/search";
import { QUEUES, enqueue } from "../../queues.js";
import { getLogger } from "../../logger.js";
import type { WorkerEnv } from "../../config.js";
import { StreamConsumer, type StreamEntry } from "../../lib/redis-stream.js";
import { tryParseEvent, SEARCH_INDEX_EVENTS } from "../../events.js";
import {
  buildPageDoc,
  buildBlockDocs,
  buildDatabaseDoc,
  buildRecordDoc,
  buildCommentDoc,
  buildFileDoc,
  buildUserDoc,
  contentHash,
  upsertMirror,
  deleteMirror,
  mirrorRefId,
  docIdFor,
  expandChunkDocs,
  type IndexerDeps,
} from "./builders.js";

const log = getLogger("search-index");

const BATCH = 500; // TASK-0096 §7: index in batches of 500 (log per batch)
const DLQ_STREAM = "search-index:dlq";
const INDEX_VERSION = "v1"; // physical index suffix; dimension rollover = S5

export interface SearchIndexOps {
  /** The @factory/search client (injectable fetch for tests). */
  search: SearchClient;
  prisma: PrismaClient;
  /** Redis connection (DLQ stream + enqueues). */
  connection: Redis;
  env: WorkerEnv;
  /**
   * TASK-0125 — incremental RAG embedding hook: after a page/record is
   * (re)indexed, enqueue its `embed:chunks` job on the `ai` queue. Wired at
   * boot (index.ts); optional so the Docker-free unit tests pass plain ops.
   * The hook is fire-and-forget (a failed enqueue must not fail the index
   * write — the nightly `embed:reembed` reconciles any gap).
   */
  enqueueEmbed?: (job: { workspaceId: string; docType: "page" | "record"; refId: string }) => Promise<unknown> | unknown;
}

const deps = (ops: SearchIndexOps): IndexerDeps => ({ prisma: ops.prisma });

// ---------------------------------------------------------------------------
// Index management
// ---------------------------------------------------------------------------

const contentIndexName = () => `factory-content-${INDEX_VERSION}`;

/**
 * Startup: create the content index (env-driven dimension) behind the alias
 * `factory-content`, idempotently. If the alias already points at an index
 * with a DIFFERENT vector dimension, leave it alone and warn — that is the
 * S5 rollover script's job (`dims` cannot change in place).
 */
export async function ensureContentIndex(ops: SearchIndexOps): Promise<{ created: boolean }> {
  const { search, env } = ops;
  const mapping = contentMapping({
    dimension: env.SEARCH_VECTOR_DIMENSION,
    m: env.SEARCH_KNN_M,
    efConstruction: env.SEARCH_KNN_EF_CONSTRUCTION,
  });
  const name = contentIndexName();
  const exists = await search.indexExists(name).catch(() => false);
  if (!exists) {
    // TASK-0100 S5 self-heal: if the ALIAS name is already taken by a REAL
    // index (no mapping — OpenSearch was down at boot, so this boot's create
    // was skipped, and the first `_bulk` auto-created a bare real index
    // named `factory-content`), it has NO vector mapping and `createAlias`
    // would reject it (an index cannot be aliased onto itself). Delete it
    // and rebuild the versioned index + alias. This only ever runs when the
    // versioned index is missing, so it is safe: the bare index holds no
    // queryable state (reconcile/reindex rebuild from source).
    if (!(await search.isAlias(CONTENT_INDEX_ALIAS).catch(() => false))) {
      const bareReal = await search.indexExists(CONTENT_INDEX_ALIAS).catch(() => false);
      if (bareReal) {
        log.warn(
          { index: CONTENT_INDEX_ALIAS },
          "content alias name is a real (mapping-less) index — removing to rebuild versioned index + alias"
        );
        await search.deleteIndex(CONTENT_INDEX_ALIAS).catch((err: unknown) => {
          log.error({ err: (err as Error).message }, "failed to delete bare content index — manual cleanup required");
          throw err;
        });
      }
    }
    const created = await search.createIndexIfMissing(name, mapping);
    if (created) await search.createAlias(name, CONTENT_INDEX_ALIAS);
    log.info(
      { index: name, alias: CONTENT_INDEX_ALIAS, dimension: env.SEARCH_VECTOR_DIMENSION },
      "content index ensured"
    );
    return { created };
  }
  const liveDim = await search.getVectorDimension(name).catch(() => null);
  if (liveDim !== null && !vectorDimensionMatches({ mappings: { properties: { vector: { dimension: liveDim } } } }, env.SEARCH_VECTOR_DIMENSION)) {
    log.warn(
      { index: name, liveDim, wanted: env.SEARCH_VECTOR_DIMENSION },
      "content index dimension mismatch — alias rollover required (scripts/opensearch-reindex.ts)"
    );
  }
  return { created: false };
}

/**
 * TASK-0100 S5: the boot-time `ensureContentIndex` is the ONLY create path,
 * and it is SKIPPED when OpenSearch is down at boot (degrade-and-continue).
 * The old boot comment promised the index would be "created lazily by the
 * first (re)index" — but no such lazy path existed, so after a boot-with-
 * OpenSearch-down, the first `_bulk` (targeting the alias name) auto-created
 * a bare real index and every search was SEARCH_UNAVAILABLE forever.
 * This guard makes the write paths self-heal: if the versioned index is
 * missing, run `ensureContentIndex` first (idempotent, cheap when it exists).
 */
let contentIndexEnsured = false;
async function ensureContentIndexOnce(ops: SearchIndexOps): Promise<void> {
  if (contentIndexEnsured) return;
  await ensureContentIndex(ops);
  contentIndexEnsured = true;
}

// ---------------------------------------------------------------------------
// Write paths (doc → index + mirror)
// ---------------------------------------------------------------------------

/**
 * Bulk-index docs in 500-doc batches, mirroring each into Postgres.
 *
 * Chunk expansion (TASK-0099 §3): before writing, each page/record whole-doc
 * is expanded via {@link expandChunkDocs} — bodies over the threshold also
 * emit per-chunk docs (the kNN leg's target). Chunk docs are derived from the
 * parent (they have NO `search_documents` mirror rows), so:
 *   - the parent's PRE-EXISTING chunks are cleared first (a `_delete_by_query`
 *     on `parentDocId`), because a shrunk body changes the chunk count and the
 *     old ids can't be enumerated from the mirror. Every call passes the
 *     COMPLETE per-parent set, so clear-then-write keeps the chunk set exact
 *     (shrink/grow); it's an idempotent no-op when the parent has no chunks.
 *   - chunk docs are NOT mirrored (only whole docs are).
 */
export async function indexDocs(ops: SearchIndexOps, docs: IndexDocInput[]): Promise<{ indexed: number }> {
  await ensureContentIndexOnce(ops); // TASK-0100 S5 self-heal (index may not exist yet)
  const chunkableParents = docs
    .filter((d) => d.type === "page" || d.type === "record")
    .map((d) => d.docId);
  if (chunkableParents.length > 0) {
    await ops.search.deleteChunksOf(chunkableParents);
  }
  const expanded = docs.flatMap((d) => expandChunkDocs(d));
  let indexed = 0;
  for (let i = 0; i < expanded.length; i += BATCH) {
    const chunk = expanded.slice(i, i + BATCH);
    const res = await ops.search.indexDocs(chunk);
    for (const doc of chunk) {
      if (doc.type !== CHUNK_DOC_TYPE) await upsertMirror(ops.prisma, doc); // chunks: no mirror row
    }
    indexed += res.indexed;
    if (i + BATCH < expanded.length) log.info({ batch: i / BATCH + 1 }, "index batch");
  }
  // TASK-0125 — incremental RAG embedding: every page/record that was just
  // (re)indexed gets its vectors refreshed on the `ai` queue. The event
  // dispatch, the workspace reindex AND the 15-min reconcile all funnel
  // through indexDocs, so this single hook covers the whole incremental
  // surface (spec: "incremental (event-driven)"). Fire-and-forget: an
  // enqueue failure must not fail the index write — the nightly re-embed
  // closes any gap. The embed job re-derives the chunk set from the source,
  // so the refId here is only a routing key.
  if (ops.enqueueEmbed) {
    for (const d of docs) {
      if (d.type !== "page" && d.type !== "record") continue;
      const refId = d.docId.split(":")[1] ?? d.docId;
      try {
        await ops.enqueueEmbed({ workspaceId: d.workspaceId, docType: d.type, refId });
      } catch (err) {
        log.warn(
          { err: (err as Error).message, docId: d.docId },
          "embed:chunks enqueue failed (nightly re-embed backstops)",
        );
      }
    }
  }
  return { indexed };
}

/**
 * Remove one doc (index + mirror). A 404 on delete is fine (idempotent). For
 * a page/record parent, its chunk docs are removed in a parent-driven pass
 * (`deleteChunksOf` — chunks have no mirror rows to enumerate, TASK-0099 §3).
 */
export async function removeDoc(
  ops: SearchIndexOps,
  type: SearchType,
  refId: string,
  workspaceId?: string
): Promise<void> {
  const docId = type === "user" && workspaceId ? `user:${workspaceId}:${refId}` : `${type}:${refId}`;
  await ensureContentIndexOnce(ops); // TASK-0100 S5 self-heal (index may not exist yet)
  try {
    await ops.search.deleteDoc(docId);
  } catch (err) {
    if (!(err instanceof ApiError && err.status === 404)) throw err;
  }
  if (type === "page" || type === "record") {
    try {
      await ops.search.deleteChunksOf(docId);
    } catch (err) {
      if (!(err instanceof ApiError && err.status === 404)) throw err;
    }
  }
  await deleteMirror(ops.prisma, type, refId, workspaceId);
}

// ---------------------------------------------------------------------------
// Stream consumer
// ---------------------------------------------------------------------------

/**
 * Rebuild a page doc + its block docs (null-safe). Returns the docs to
 * upsert plus the docIds to DELETE: a trashed OR purged page has no doc
 * (arch §1.1 "trashed = deleted from index"; TASK-0096 §5 "page.deleted
 * (soft/trash) removes the doc"). Block docs of a vanished page are
 * enumerated from the MIRROR (their source rows may already be hard-deleted).
 * Used by create/update/share/trash events.
 */
async function pageDocs(
  ops: SearchIndexOps,
  pageId: string,
  workspaceId?: string
): Promise<{ docs: IndexDocInput[]; deleteIds: string[] }> {
  const d = deps(ops);
  const [pageDoc, blockDocs] = await Promise.all([buildPageDoc(d, pageId), buildBlockDocs(d, pageId)]);
  if (pageDoc || blockDocs.length > 0) {
    return { docs: [pageDoc, ...blockDocs].filter((x): x is IndexDocInput => x !== null), deleteIds: [] };
  }
  // Vanished (trashed/purged): remove the page doc + its surviving block docs.
  const orphanBlocks = await ops.prisma.searchDocument.findMany({
    where: { ...(workspaceId ? { workspaceId } : {}), type: "block", metadata: { path: ["pageId"], equals: pageId } },
    select: { refId: true },
  });
  const deleteIds = [`page:${pageId}`, ...orphanBlocks.map((b) => `block:${b.refId}`)];
  return { docs: [], deleteIds };
}

/**
 * The event dispatch for one parsed `factory:events` entry (exported for
 * Docker-free unit tests). Rethrows transport/DB failures so the StreamConsumer
 * keeps the entry PENDING (never ack an unwritten doc). Missing source rows
 * (poison) log + ack; the 15-min reconcile backstops.
 */
export async function handleSearchEvent(
  ops: SearchIndexOps,
  envelope: { event: string; payload: Record<string, unknown> }
): Promise<void> {
  if (!SEARCH_INDEX_EVENTS.includes(envelope.event as (typeof SEARCH_INDEX_EVENTS)[number])) {
    return; // not mine — XACK (the group is per-queue, not per-event)
  }
  const p = envelope.payload;
  const t0 = Date.now();
  switch (envelope.event) {
    case "page.created":
    case "page.updated":
    case "page.shared": {
      const ws = str(p.workspaceId);
      const pageId = str(p.pageId ?? p.id);
      if (!pageId) return;
      const { docs, deleteIds } = await pageDocs(ops, pageId, ws || undefined);
      if (deleteIds.length > 0) {
        // Trashed (soft delete) or purged: the doc is REMOVED from the index
        // (TASK-0096 §5 / arch §1.1). Restore re-adds it via the
        // reconciliation sweep (the API publishes no restore domain event).
        for (const id of deleteIds) {
          const [type, ...rest] = id.split(":");
          await removeDoc(ops, type as SearchType, rest.join(":"), ws || undefined);
        }
        log.info({ pageId, removed: deleteIds.length, ms: Date.now() - t0 }, "page removed from search (trashed/purged)");
        break;
      }
      await indexDocs(ops, docs);
      log.info({ event: envelope.event, pageId, docs: docs.length, ms: Date.now() - t0 }, "page indexed");
      break;
    }
    case "page.deleted": {
      // Both soft (trash) and hard (purge) REMOVE the doc(s) from the index —
      // the source rows are already trashed/purged by the API when this event
      // arrives, so `pageDocs` returns deleteIds in both cases (unified with
      // the page.created/updated handler). Purge additionally lists the
      // descendant pageIds explicitly.
      const ws = str(p.workspaceId);
      const rootPage = str(p.pageId ?? p.id);
      const pageIds: string[] = Array.isArray(p.pageIds)
        ? (p.pageIds as unknown[]).filter((x): x is string => typeof x === "string")
        : [];
      const all = [...new Set([rootPage, ...pageIds].filter(Boolean))];
      for (const pageId of all) {
        const { deleteIds } = await pageDocs(ops, pageId, ws || undefined);
        for (const id of deleteIds) {
          const [type, ...rest] = id.split(":");
          await removeDoc(ops, type as SearchType, rest.join(":"), ws || undefined);
        }
      }
      log.info({ pageIds: all, permanent: p.permanent === true, ms: Date.now() - t0 }, "page(s) removed from search");
      break;
    }
    case "database.created":
    case "database.updated": {
      const databaseId = str(p.databaseId ?? p.id);
      if (!databaseId) return;
      const doc = await buildDatabaseDoc(deps(ops), databaseId);
      if (doc) {
        await indexDocs(ops, [doc]);
        log.info({ databaseId, ms: Date.now() - t0 }, "database indexed");
      } else {
        log.warn({ databaseId }, "database source row missing — poison, ack (reconcile backstops)");
      }
      break;
    }
    case "record.created":
    case "record.updated": {
      const recordId = str(p.recordId ?? p.id);
      if (!recordId) return;
      const doc = await buildRecordDoc(deps(ops), recordId);
      if (doc) {
        await indexDocs(ops, [doc]);
        log.info({ recordId, ms: Date.now() - t0 }, "record indexed");
      } else {
        log.warn({ recordId }, "record source row missing — poison, ack (reconcile backstops)");
      }
      break;
    }
    case "comment.created": {
      const commentId = str(p.commentId ?? p.id);
      if (!commentId) return;
      const doc = await buildCommentDoc(deps(ops), commentId);
      if (doc) {
        await indexDocs(ops, [doc]);
        log.info({ commentId, ms: Date.now() - t0 }, "comment indexed");
      } else {
        log.warn({ commentId }, "comment source row missing — poison, ack (reconcile backstops)");
      }
      break;
    }
    case "comment.deleted": {
      // Hard delete in the API → delete the doc now.
      const commentId = str(p.commentId ?? p.id);
      if (!commentId) return;
      await removeDoc(ops, "comment", commentId, str(p.workspaceId) || undefined);
      log.info({ commentId, ms: Date.now() - t0 }, "comment deleted from search");
      break;
    }
    case "file.uploaded":
    case "file.finalized":
    case "file.updated": {
      // `file.uploaded` fires at PRESIGN (the File row may still be UPLOADING
      // with null mime/size); `file.finalized` is published by the files worker
      // at S3 finalization with the real mime/size; `file.updated` fires on a
      // rename (TASK-0104 §8) so the doc's title converges immediately. The
      // doc converges on whichever arrives (reconcile backstops).
      const fileId = str(p.fileId ?? p.id);
      if (!fileId) return;
      const doc = await buildFileDoc(deps(ops), fileId);
      if (doc) {
        await indexDocs(ops, [doc]);
        log.info({ fileId, ms: Date.now() - t0 }, "file indexed (mime/size may converge via resync)");
      } else {
        log.warn({ fileId }, "file source row missing — poison, ack (reconcile backstops)");
      }
      break;
    }
    case "file.deleted": {
      // TASK-0104 §3: the file went to the trash → remove the search doc now
      // (the acceptance criterion is "index doc gone within 30 s"). An
      // EXPLICIT removal is required, not a re-upsert: `buildFileDoc` returns
      // null for a `deletedAt` row (the upsert path would be a no-op), so the
      // doc has to be deleted by id. Restore re-indexes via the
      // `file.uploaded {restored:true}` re-emit; a purge's row delete is
      // backstopped by the 15-min contentHash reconcile (the doc's source row
      // is gone → the sweep removes it).
      const fileId = str(p.fileId ?? p.id);
      if (!fileId) return;
      await removeDoc(ops, "file", fileId, str(p.workspaceId) || undefined);
      log.info({ fileId, ms: Date.now() - t0 }, "file removed from search (trashed)");
      break;
    }
    case "permission.changed": {
      // The API publishes {workspaceId, userId?, role?|removed?} — never
      // resource ids — so a targeted reindex is impossible by design (search
      // §0; recorded in TASK-0096). Reindex the whole workspace via the
      // maintenance queue; the 15-min reconcile sweep backstops.
      const ws = str(p.workspaceId);
      if (!ws) return;
      await enqueue(ops.connection, QUEUES.maintenance, { type: "workspace-resync", workspaceId: ws });
      log.info({ workspaceId: ws, ms: Date.now() - t0 }, "permission changed → workspace resync enqueued");
      break;
    }
    case "user.register":
    case "user.login": {
      const userId = str(p.userId ?? p.id);
      if (!userId) return;
      const ws = str(p.workspaceId);
      if (!ws) {
        log.warn({ userId }, "user event without workspaceId — nothing to stamp (reconcile covers)");
        return;
      }
      const doc = await buildUserDoc(deps(ops), userId, ws);
      if (doc) await indexDocs(ops, [doc]);
      break;
    }
    case "workspace.created": {
      const ws = str(p.workspaceId);
      if (!ws) return;
      await runReindex(ops, ws);
      log.info({ workspaceId: ws, ms: Date.now() - t0 }, "workspace created → initial index");
      break;
    }
    default:
      return;
  }
}

export function createSearchIndexConsumer(opts: {
  redis: Redis;
  ops: SearchIndexOps;
  consumerId: string;
}): StreamConsumer {
  const { redis, ops, consumerId } = opts;
  return new StreamConsumer(
    redis,
    "factory:events",
    "worker-search-index",
    consumerId,
    async (entry: StreamEntry) => {
      const parsed = tryParseEvent(entry);
      if (!parsed) return; // poison envelope: logged, XACK
      try {
        await handleSearchEvent(ops, parsed.envelope);
      } catch (err) {
        // Transport/DB failure: rethrow → entry stays PENDING (XAUTOCLAIM
        // redelivers). Never ack a doc we failed to write.
        log.error({ err: (err as Error).message, event: parsed.envelope.event, id: entry.id }, "search index handler failed — entry pending");
        throw err;
      }
    }
  );
}

// ---------------------------------------------------------------------------
// Full (re)index — all types of a workspace (or every workspace)
// ---------------------------------------------------------------------------

export interface ReindexCounts {
  workspaces: number;
  docs: number;
  orphansRemoved: number;
}

/**
 * Collect the FULL set of searchable docs for a workspace straight from the
 * source tables (source→index direction). Trashed/purged pages and deleted
 * files build to null (they contribute nothing); archived records build with
 * `archived: true`. Shared by the full reindex and the reconciliation sweep.
 */
async function collectWorkspaceDocs(ops: SearchIndexOps, ws: string): Promise<IndexDocInput[]> {
  const { prisma } = ops;
  const d = deps(ops);
  const [pages, databases, records, comments, files, members] = await Promise.all([
    prisma.page.findMany({ where: { workspaceId: ws, deletedAt: null }, select: { id: true } }),
    prisma.database.findMany({ where: { workspaceId: ws }, select: { id: true } }),
    prisma.databaseRecord.findMany({ where: { workspaceId: ws }, select: { id: true } }),
    prisma.comment.findMany({ where: { workspaceId: ws, pageId: { not: null } }, select: { id: true } }),
    prisma.file.findMany({ where: { workspaceId: ws, deletedAt: null }, select: { id: true } }),
    prisma.workspaceMember.findMany({ where: { workspaceId: ws }, select: { userId: true } }),
  ]);

  const docs: IndexDocInput[] = [];
  for (const pg of pages) docs.push(...(await pageDocs(ops, pg.id, ws).then((r) => r.docs)));
  for (const db of databases) {
    const doc = await buildDatabaseDoc(d, db.id);
    if (doc) docs.push(doc);
  }
  for (const r of records) {
    const doc = await buildRecordDoc(d, r.id);
    if (doc) docs.push(doc);
  }
  for (const c of comments) {
    const doc = await buildCommentDoc(d, c.id);
    if (doc) docs.push(doc);
  }
  for (const f of files) {
    const doc = await buildFileDoc(d, f.id);
    if (doc) docs.push(doc);
  }
  // User docs: every active member (collaborator search, per-workspace doc).
  for (const m of members) {
    const doc = await buildUserDoc(d, m.userId, ws);
    if (doc) docs.push(doc);
  }
  return docs;
}

export async function runReindex(ops: SearchIndexOps, workspaceId?: string, orgId?: string): Promise<ReindexCounts> {
  await ensureContentIndexOnce(ops); // TASK-0100 S5 self-heal (index may not exist yet)
  const { prisma } = ops;
  // T-156: org-scoped reindex (the org-admin trigger). The workspace set is
  // filtered by `organizationId` — one org's admin can never reindex (and
  // thus rewrite the grant snapshots of) another org's workspaces.
  const workspaces = workspaceId
    ? [workspaceId]
    : (
        await prisma.workspace.findMany({
          where: orgId ? { organizationId: orgId } : {},
          select: { id: true }
        })
      ).map((w) => w.id);
  const counts: ReindexCounts = { workspaces: 0, docs: 0, orphansRemoved: 0 };

  for (const ws of workspaces) {
    const docs = await collectWorkspaceDocs(ops, ws);
    await indexDocs(ops, docs);

    // Orphans: mirror rows whose source row is gone (purged / deleted /
    // demoted member) → remove from index + mirror. Content DRIFT (source
    // changed without an event) is the reconcile sweep's job, not this one.
    const live = new Set(docs.map((doc) => `${doc.type}:${mirrorRefId(doc)}`));
    const mirror = await prisma.searchDocument.findMany({
      where: { workspaceId: ws },
      select: { type: true, refId: true },
    });
    for (const row of mirror) {
      if (live.has(`${row.type}:${row.refId}`)) continue;
      const refId = row.type === "user" ? row.refId.split(":").pop() ?? row.refId : row.refId;
      await removeDoc(ops, row.type as SearchType, refId, ws);
      counts.orphansRemoved++;
    }

    counts.docs += docs.length;
    counts.workspaces++;
    log.info({ workspaceId: ws, docs: docs.length, orphans: counts.orphansRemoved }, "reindex workspace done");
  }
  log.info(counts, "reindex complete");
  return counts;
}

// ---------------------------------------------------------------------------
// ContentHash reconciliation sweep (search §1.3/§7.3 — supersedes the count
// heuristic; subsumes the old `reconcileSearch`).
// ---------------------------------------------------------------------------

export interface ReconcileResult {
  workspaces: number;
  checked: number;
  repaired: number;
  removed: number;
}

/** Rebuild one doc from its mirror row (null = source row gone). */
async function docFromMirrorRow(
  ops: SearchIndexOps,
  row: { type: string; refId: string },
  workspaceId: string
): Promise<IndexDocInput | null> {
  const d = deps(ops);
  switch (row.type as SearchType) {
    case "page":
      return buildPageDoc(d, row.refId);
    case "block":
      return blockDocById(d, row.refId);
    case "database":
      return buildDatabaseDoc(d, row.refId);
    case "record":
      return buildRecordDoc(d, row.refId);
    case "comment":
      return buildCommentDoc(d, row.refId);
    case "file":
      return buildFileDoc(d, row.refId);
    case "user": {
      const uid = row.refId.split(":").pop() ?? row.refId;
      return buildUserDoc(d, uid, workspaceId);
    }
    default:
      return null; // unknown type (legacy mirror row) → caller removes
  }
}

/** A single block doc by block id (reconcile path). */
async function blockDocById(d: IndexerDeps, blockId: string): Promise<IndexDocInput | null> {
  const block = await d.prisma.block.findUnique({
    where: { id: blockId },
    select: { id: true, pageId: true, deletedAt: true },
  });
  if (!block || block.deletedAt !== null || !block.pageId) return null;
  const docs = await buildBlockDocs(d, block.pageId);
  return docs.find((x) => x.blockId === blockId) ?? null;
}

export async function runReconcile(ops: SearchIndexOps, workspaceId?: string): Promise<ReconcileResult> {
  await ensureContentIndexOnce(ops); // TASK-0100 S5 self-heal (index may not exist yet)
  const { prisma } = ops;
  const workspaces = workspaceId
    ? [workspaceId]
    : (await prisma.workspace.findMany({ select: { id: true } })).map((w) => w.id);
  const result: ReconcileResult = { workspaces: 0, checked: 0, repaired: 0, removed: 0 };

  for (const ws of workspaces) {
    const rows = await prisma.searchDocument.findMany({
      where: { workspaceId: ws },
      select: { type: true, refId: true, contentHash: true, indexedAt: true },
    });
    for (let i = 0; i < rows.length; i += BATCH) {
      const batch = rows.slice(i, i + BATCH);
      // Index-loss check: which docs are MISSING from OpenSearch? The
      // contentHash sweep alone only catches SOURCE drift (changed content);
      // a doc deleted/lost directly in the index keeps a healthy mirror row.
      // Batched `_mget` (docExists) so a full sweep costs ~rows/500 requests.
      let present = new Set<string>();
      try {
        present = await ops.search.docExists(batch.map((r) => docIdFor(r.type as SearchType, r.refId)));
      } catch (err) {
        // OpenSearch unreachable: treat everything as present (rethrowing here
        // would abort the whole sweep; the next run / DLQ covers transport).
        log.warn({ err: (err as Error).message, workspaceId: ws }, "docExists failed — index-loss check skipped for batch");
        present = new Set(batch.map((r) => docIdFor(r.type as SearchType, r.refId)));
      }
      for (const row of batch) {
        result.checked++;
        const doc = await docFromMirrorRow(ops, row, ws);
        if (doc === null) {
          const refId = row.type === "user" ? row.refId.split(":").pop() ?? row.refId : row.refId;
          await removeDoc(ops, row.type as SearchType, refId, ws);
          result.removed++;
          continue;
        }
        const docId = docIdFor(row.type as SearchType, row.refId);
        if (!present.has(docId)) {
          // Lost in the index → reindex regardless of hash.
          await indexDocs(ops, [doc]);
          result.repaired++;
          continue;
        }
        if (row.contentHash !== null && row.contentHash === contentHash(doc)) continue; // in sync
        await indexDocs(ops, [doc]);
        result.repaired++;
      }
      if (i + BATCH < rows.length) {
        log.info({ workspaceId: ws, batch: i / BATCH + 1, totalBatches: Math.ceil(rows.length / BATCH) }, "reconcile batch");
      }
    }

    // Direction B (source → index): re-add docs that exist in the SOURCE but
    // not in the index AND not in the mirror. The mirror-only direction above
    // cannot see these: e.g. a RESTORED page (the API publishes no restore
    // domain event, so the mirror row was dropped with the trash) or a page
    // created whose index write failed before the mirror upsert.
    const live = await collectWorkspaceDocs(ops, ws);
    const covered = new Set(
      rows.map((r) => `${r.type}:${r.type === "user" ? (r.refId.split(":").pop() ?? r.refId) : r.refId}`)
    );
    const missing: IndexDocInput[] = [];
    for (const doc of live) {
      if (covered.has(`${doc.type}:${mirrorRefId(doc)}`)) continue;
      let present = false;
      try {
        // `live` holds whole docs only (collectWorkspaceDocs never emits
        // chunk docs), so `doc.type` is a public SearchType here.
        const id = docIdFor(doc.type as SearchType, doc.id);
        present = (await ops.search.docExists([id])).has(id);
      } catch {
        present = true; // unreachable → don't re-add what we can't verify
      }
      if (!present) missing.push(doc);
    }
    if (missing.length > 0) {
      await indexDocs(ops, missing);
      result.repaired += missing.length;
      log.info({ workspaceId: ws, readded: missing.length }, "reconcile: source→index re-adds (restored/lost before mirror)");
    }

    result.workspaces++;
  }
  log.info(result, "reconcile complete");
  return result;
}

// ---------------------------------------------------------------------------
// DLQ (TASK-0096 §8)
// ---------------------------------------------------------------------------

/** Record a permanently-failed search-index job to the DLQ stream. */
export async function recordSearchDlq(connection: Redis, job: Job, err: Error): Promise<void> {
  const data = (job.data ?? {}) as Record<string, unknown>;
  await connection.xadd(
    DLQ_STREAM,
    "MAXLEN",
    "~",
    "1000",
    "*",
    "jobId",
    job.id ?? "",
    "jobType",
    str(data.type) || "unknown",
    "workspaceId",
    str(data.workspaceId),
    "error",
    err.message,
    "ts",
    new Date().toISOString()
  );
  log.error({ jobId: job.id, err: err.message }, "search-index job dead-lettered");
}

// ---------------------------------------------------------------------------
// BullMQ worker (explicit ops)
// ---------------------------------------------------------------------------

export const searchIndexJobSchema = z.discriminatedUnion("type", [
  z.object({ type: z.literal("reindex"), workspaceId: z.string().min(1).optional(), orgId: z.string().min(1).optional() }),
  z.object({ type: z.literal("reconcile"), workspaceId: z.string().min(1).optional() }),
  z.object({
    type: z.literal("delete"),
    docType: z.enum(["page", "block", "database", "record", "comment", "file", "user"]),
    refId: z.string().min(1),
    workspaceId: z.string().min(1).optional(),
  }),
]);
export type SearchIndexJobData = z.infer<typeof searchIndexJobSchema>;

export function createSearchIndexWorker(opts: {
  connection: Redis;
  ops: SearchIndexOps;
  concurrency?: number;
}): Worker {
  const { connection, ops, concurrency } = opts;
  return new Worker(
    QUEUES.searchIndex,
    withJobOtel(QUEUES.searchIndex, async (job: Job) => {
      const data = searchIndexJobSchema.parse(job.data);
      const started = Date.now();
      switch (data.type) {
        case "reindex": {
          const counts = await runReindex(ops, data.workspaceId, data.orgId);
          log.info({ jobId: job.id, counts, ms: Date.now() - started }, "reindex done");
          return counts;
        }
        case "reconcile": {
          const result = await runReconcile(ops, data.workspaceId);
          log.info({ jobId: job.id, result, ms: Date.now() - started }, "reconcile done");
          // T-177 (search health tab): stamp the last-reconcile time for the
          // org-admin console (best-effort — a stamp failure must never fail
          // an otherwise-successful sweep; the 15-minute cadence makes a
          // missed stamp self-healing).
          try {
            await connection.set("search:reconcile:last", new Date().toISOString(), "EX", 30 * 86_400);
          } catch (err) {
            log.warn({ jobId: job.id, err: String(err) }, "search:reconcile:last stamp failed (non-fatal)");
          }
          return result;
        }
        case "delete": {
          await removeDoc(ops, data.docType, data.refId, data.workspaceId);
          return { deleted: true };
        }
      }
    }),
    {
      connection,
      ...(concurrency !== undefined ? { concurrency } : {})
    }
  );
}

function str(v: unknown): string {
  return typeof v === "string" ? v : "";
}