/**
 * TASK-0096 — search indexer unit tests (Docker-free).
 *
 * Fakes: a recording PrismaClient, a recording/fetch-backed SearchClient, and
 * a mocked `enqueue` (the processor's permission.changed → resync path).
 * The StreamConsumer transport itself is integration-covered (Docker-gated).
 */
import { describe, expect, it, vi, beforeEach } from "vitest";
import type { PrismaClient } from "@prisma/client";
import { SearchClient, chunkText, DEFAULT_CHUNK_OPTIONS, type SearchDoc } from "@factory/search";
import { ApiError } from "@factory/shared/errors";
import type { WorkerEnv } from "../config.js";
import {
  buildGrantSnapshot,
  workspaceFloorSnapshot,
  pageScopes,
  type GrantContext
} from "../processors/search-index/grants.js";
import {
  buildPageDoc,
  buildBlockDocs,
  buildUserDoc,
  contentHash as builderHash,
  upsertMirror,
  deleteMirror,
  mirrorRefId,
  blockText,
  renderPropertyValue,
  expandChunkDocs,
  CHUNK_BODY_THRESHOLD
} from "../processors/search-index/builders.js";
import {
  handleSearchEvent,
  removeDoc,
  runReconcile,
  runReindex,
  ensureContentIndex,
  recordSearchDlq,
  indexDocs,
  type SearchIndexOps
} from "../processors/search-index/processor.js";

vi.mock("../queues.js", () => {
  const calls: { queue: string; data: Record<string, unknown> }[] = [];
  return {
    QUEUES: {
      files: "files",
      searchIndex: "search-index",
      webhooks: "webhooks",
      automations: "automations",
      email: "email",
      ai: "ai",
      maintenance: "maintenance"
    },
    enqueue: vi.fn(async (_c: unknown, queue: string, data: Record<string, unknown>) => {
      calls.push({ queue, data });
      return { id: "job-1" };
    }),
    enqueueWithId: vi.fn(async () => ({ id: "job-2" })),
    __calls: calls
  };
});
import { enqueue, QUEUES } from "../queues.js";

// ---------------------------------------------------------------------------
// fakes
// ---------------------------------------------------------------------------

const ENV = {
  SEARCH_VECTOR_DIMENSION: 1536,
  SEARCH_KNN_M: 16,
  SEARCH_KNN_EF_CONSTRUCTION: 100,
  WORKER_CONCURRENCY: 2
} as unknown as WorkerEnv;

const NOW = new Date("2026-01-02T03:04:05.000Z");

interface FakePrisma {
  prisma: PrismaClient;
  page: { findUnique: ReturnType<typeof vi.fn>; findMany: ReturnType<typeof vi.fn> };
  searchDocument: {
    findMany: ReturnType<typeof vi.fn>;
    upsert: ReturnType<typeof vi.fn>;
    deleteMany: ReturnType<typeof vi.fn>;
  };
  permission: { findMany: ReturnType<typeof vi.fn> };
  user: { findUnique: ReturnType<typeof vi.fn> };
  workspace: { findMany: ReturnType<typeof vi.fn> };
}

function fakePrisma(): FakePrisma {
  const page = {
    findUnique: vi.fn(async () => null),
    findMany: vi.fn(async () => [])
  };
  const searchDocument = {
    findMany: vi.fn(async () => []),
    upsert: vi.fn(async (a: { create: Record<string, unknown> }) => ({ id: "sd-1", ...a.create })),
    deleteMany: vi.fn(async () => ({ count: 1 }))
  };
  const permission = { findMany: vi.fn(async () => []) };
  const user = { findUnique: vi.fn(async () => null) };
  const workspace = { findMany: vi.fn(async () => []) };
  const prisma = {
    page,
    block: { findUnique: vi.fn(async () => null) },
    database: { findUnique: vi.fn(async () => null), findMany: vi.fn(async () => []) },
    databaseRecord: { findUnique: vi.fn(async () => null), findMany: vi.fn(async () => []) },
    comment: { findUnique: vi.fn(async () => null), findMany: vi.fn(async () => []) },
    file: { findUnique: vi.fn(async () => null), findMany: vi.fn(async () => []) },
    user,
    share: { findMany: vi.fn(async () => []) },
    permission,
    searchDocument,
    workspace,
    workspaceMember: { findMany: vi.fn(async () => []) }
  };
  return {
    prisma: prisma as unknown as PrismaClient,
    page,
    searchDocument,
    permission,
    user,
    workspace
  };
}

interface FakeSearch {
  ops: SearchIndexOps;
  indexed: SearchDoc[][];
  deleted: string[];
  updated: { docId: string; doc: Record<string, unknown> }[];
  xadd: ReturnType<typeof vi.fn>;
  docExists: ReturnType<typeof vi.fn>;
  deleteChunksOf: ReturnType<typeof vi.fn>;
}

function fakeSearch(prisma: PrismaClient): FakeSearch {
  const indexed: SearchDoc[][] = [];
  const deleted: string[] = [];
  const updated: { docId: string; doc: Record<string, unknown> }[] = [];
  const xadd = vi.fn(async () => "1-1");
  const docExists = vi.fn(async (ids: string[]) => new Set(ids)); // all present by default
  const deleteChunksOf = vi.fn(async () => 0); // default: no stale chunks
  const search = {
    indexDocs: vi.fn(async (docs: SearchDoc[]) => {
      indexed.push(docs);
      return { indexed: docs.length };
    }),
    deleteDoc: vi.fn(async (id: string) => {
      deleted.push(id);
    }),
    deleteChunksOf,
    updateDoc: vi.fn(async (docId: string, doc: Record<string, unknown>) => {
      updated.push({ docId, doc });
    }),
    docExists,
    indexExists: vi.fn(async () => false),
    // TASK-0100 S5 self-heal: ensureContentIndex probes whether the bare
    // CONTENT_INDEX_ALIAS name is a real index (not an alias) and deletes it.
    isAlias: vi.fn(async () => true),
    deleteIndex: vi.fn(async () => {}),
    createIndexIfMissing: vi.fn(async () => true),
    createAlias: vi.fn(async () => {}),
    getVectorDimension: vi.fn(async () => null),
    refresh: vi.fn(async () => {})
  } as unknown as SearchClient;
  const ops: SearchIndexOps = {
    search,
    prisma,
    connection: { xadd } as unknown as SearchIndexOps["connection"],
    env: ENV
  };
  return { ops, indexed, deleted, updated, xadd, docExists, deleteChunksOf };
}

function allIndexed(fs: FakeSearch): SearchDoc[] {
  return fs.indexed.flat();
}

const basePage = () => ({
  id: "pg1",
  workspaceId: "ws1",
  teamId: null,
  title: "Roadmap",
  icon: null,
  description: "The plan",
  isPrivate: false,
  archivedAt: null,
  trashedAt: null,
  deletedAt: null,
  createdBy: "u1",
  createdAt: NOW,
  updatedAt: NOW,
  blocks: [
    {
      id: "b1",
      type: "PARAGRAPH",
      content: { text: "Ship search" },
      properties: {},
      position: "a0",
      deletedAt: null,
      createdBy: "u1",
      createdAt: NOW,
      updatedAt: NOW
    }
  ]
});

beforeEach(() => {
  vi.clearAllMocks();
});

// ---------------------------------------------------------------------------
// grants.ts — the security core
// ---------------------------------------------------------------------------

describe("buildGrantSnapshot (mirrors effectiveLevel ≥ VIEW)", () => {
  const ctx: GrantContext = {
    workspaceId: "ws1",
    teamId: null,
    scopes: pageScopes("ws1", "pg1", null),
    isPrivate: false,
    shares: []
  };

  it("public page: workspace-visible, no public", async () => {
    const prisma = fakePrisma();
    const snap = await buildGrantSnapshot(prisma.prisma as never, ctx, NOW);
    expect(snap).toEqual({ public: false, workspace: true, user: false, teams: [], users: [] });
  });

  it("private page: workspace-invisible unless an explicit user grant applies", async () => {
    const prisma = fakePrisma();
    prisma.permission.findMany.mockResolvedValue([
      { granteeType: "user", granteeId: "u9", resourceType: "page", resourceId: "pg1", level: "VIEW" },
      // scope NOT applicable to this page → ignored
      { granteeType: "user", granteeId: "u7", resourceType: "page", resourceId: "other", level: "EDIT" },
      // granteeType "share" is never counted (guard filters user|team)
      { granteeType: "share", granteeId: "s1", resourceType: "page", resourceId: "pg1", level: "VIEW" }
    ]);
    const snap = await buildGrantSnapshot(prisma.prisma as never, { ...ctx, isPrivate: true }, NOW);
    expect(snap.workspace).toBe(false); // intended model: private ⇒ workspace-invisible
    expect(snap.users).toEqual(["u9"]);
    expect(snap.user).toBe(true);
  });

  it("private page with NO explicit grants: the CREATOR is still a viewer (TASK-0100 S5)", async () => {
    // The owner has no Permission row (ownership lives in Page.createdBy), so
    // without creatorId the snapshot was users:[] and the creator's own
    // private page was unfindable in search.
    const prisma = fakePrisma();
    prisma.permission.findMany.mockResolvedValue([]);
    const snap = await buildGrantSnapshot(
      prisma.prisma as never,
      { ...ctx, isPrivate: true, creatorId: "owner1" },
      NOW
    );
    expect(snap.workspace).toBe(false);
    expect(snap.users).toEqual(["owner1"]);
    expect(snap.user).toBe(true);
  });

  it("active public share (allowSearchIndexing, unexpired) ⇒ public", async () => {
    const prisma = fakePrisma();
    const snap = await buildGrantSnapshot(
      prisma.prisma as never,
      { ...ctx, shares: [{ scope: "PUBLIC", allowSearchIndexing: true, expiresAt: new Date("2030-01-01") }] },
      NOW
    );
    expect(snap.public).toBe(true);
  });

  it("expired / non-indexable public share ⇒ not public", async () => {
    const prisma = fakePrisma();
    const snap = await buildGrantSnapshot(
      prisma.prisma as never,
      {
        ...ctx,
        shares: [
          { scope: "PUBLIC", allowSearchIndexing: true, expiresAt: new Date("2020-01-01") },
          { scope: "PUBLIC", allowSearchIndexing: false, expiresAt: null }
        ]
      },
      NOW
    );
    expect(snap.public).toBe(false);
  });

  it("workspaceFloorSnapshot: workspace-visible unless private", () => {
    expect(workspaceFloorSnapshot({ isPrivate: false, shares: [] }).workspace).toBe(true);
    expect(workspaceFloorSnapshot({ isPrivate: true, shares: [] }).workspace).toBe(false);
  });
});

// ---------------------------------------------------------------------------
// builders.ts — doc shape
// ---------------------------------------------------------------------------

describe("buildPageDoc / buildBlockDocs", () => {
  it("composes title + description + block text, grants, url, docId", async () => {
    const { prisma, page } = fakePrisma();
    page.findUnique.mockResolvedValue(basePage());
    const deps = { prisma, now: () => NOW };
    const doc = await buildPageDoc(deps, "pg1");
    expect(doc).not.toBeNull();
    expect(doc!.docId).toBe("page:pg1");
    expect(doc!.title).toBe("Roadmap");
    expect(doc!.body).toBe("Roadmap\nThe plan\nShip search");
    expect(doc!.grants.workspace).toBe(true);
    expect(doc!.archived).toBe(false);
    expect(doc!.url).toBe("/w/ws1/page/pg1");
    expect(doc!.workspaceId).toBe("ws1");

    const blocks = await buildBlockDocs(deps, "pg1");
    expect(blocks).toHaveLength(1);
    expect(blocks[0]!.docId).toBe("block:b1");
    expect(blocks[0]!.body).toBe("Ship search");
    expect(blocks[0]!.pageId).toBe("pg1");
    expect(blocks[0]!.blockId).toBe("b1");
  });

  it("trashed page ⇒ null (removed from the index; arch §1.1 / TASK-0096 §5)", async () => {
    const { prisma, page } = fakePrisma();
    page.findUnique.mockResolvedValue({ ...basePage(), trashedAt: NOW });
    const doc = await buildPageDoc({ prisma, now: () => NOW }, "pg1");
    expect(doc).toBeNull();
    const blocks = await buildBlockDocs({ prisma, now: () => NOW }, "pg1");
    expect(blocks).toEqual([]);
  });

  it("purged page (deletedAt) ⇒ null (caller deletes)", async () => {
    const { prisma, page } = fakePrisma();
    page.findUnique.mockResolvedValue({ ...basePage(), deletedAt: NOW });
    await expect(buildPageDoc({ prisma, now: () => NOW }, "pg1")).resolves.toBeNull();
    await expect(buildBlockDocs({ prisma, now: () => NOW }, "pg1")).resolves.toEqual([]);
  });

  it("blocks with no text are not separate docs", async () => {
    const { prisma, page } = fakePrisma();
    page.findUnique.mockResolvedValue({
      ...basePage(),
      blocks: [{ ...basePage().blocks[0]!, content: {} }, { ...basePage().blocks[0]!, id: "b2", content: { text: "x" } }]
    });
    const blocks = await buildBlockDocs({ prisma, now: () => NOW }, "pg1");
    expect(blocks.map((b) => b.id)).toEqual(["b2"]);
  });
});

describe("chunk emission (TASK-0099 §3 — page/record bodies over threshold)", () => {
  // A paragraph-based body (NOT uniform long text): merged pieces stay
  // sub-maxChars, so the chunker applies its full `overlapChars` tail copy to
  // every consecutive chunk (the hard-split budget case would drop the tail).
  const para = "lorem ipsum dolor sit amet consectetur adipiscing ".repeat(9).trim();
  const LONG = Array.from({ length: 18 }, (_, i) => `Section ${i}: ${para}`).join("\n\n"); // ~5.2k chars
  const bigPage = () => ({ ...basePage(), blocks: [{ ...basePage().blocks[0]!, content: { text: LONG } }] });

  it("a long page body ⇒ 1 whole doc + N chunk docs (N ≥ 5, overlapping)", async () => {
    const { prisma, page } = fakePrisma();
    page.findUnique.mockResolvedValue(bigPage());
    const whole = (await buildPageDoc({ prisma, now: () => NOW }, "pg1"))!;
    expect(whole.body.length).toBeGreaterThan(CHUNK_BODY_THRESHOLD);
    expect(whole.body.length).toBeGreaterThan(5000);

    const expanded = expandChunkDocs(whole);
    // whole doc first, then one chunk doc per chunker chunk
    const expectedChunks = chunkText(whole.body, DEFAULT_CHUNK_OPTIONS);
    expect(expectedChunks.length).toBeGreaterThanOrEqual(5);
    expect(expanded).toHaveLength(expectedChunks.length + 1);
    expect(expanded[0]).toBe(whole); // whole doc unchanged (keyword stays whole-doc)

    for (let i = 0; i < expectedChunks.length; i++) {
      const c = expanded[i + 1]!;
      expect(c.type).toBe("chunk");
      expect(c.docId).toBe(`page:pg1#c${i}`);
      expect(c.id).toBe("pg1"); // parent's ref id
      expect(c.chunkId).toBe(`c${i}`);
      expect(c.chunkIndex).toBe(i);
      expect(c.parentDocId).toBe("page:pg1");
      expect(c.body).toBe(expectedChunks[i]!.text);
      // same grants/facet snapshot as the parent
      expect(c.grants).toEqual(whole.grants);
      expect(c.workspaceId).toBe(whole.workspaceId);
      expect(c.url).toBe(whole.url);
    }

    // Consecutive chunks overlap (so matches near boundaries are not lost):
    // each chunk OPENS with a copy of the previous chunk's `overlapChars` tail.
    // NOTE: the chunker's `charStart`/`charEnd` are CONTIGUOUS source offsets
    // (no source-overlap), so overlap is asserted on the chunk TEXT — the
    // brief's `charStart[i+1] < charEnd[i]` check does not hold for this
    // chunker (recorded reconciliation).
    const chunkDocs = expanded.slice(1);
    const overlap = DEFAULT_CHUNK_OPTIONS.overlapChars;
    for (let i = 0; i < chunkDocs.length - 1; i++) {
      const tail = chunkDocs[i]!.body.slice(-overlap);
      expect(chunkDocs[i + 1]!.body.startsWith(tail)).toBe(true);
    }
  });

  it("a short body ⇒ NO chunk docs (whole doc only)", async () => {
    const whole = {
      docId: "page:pg1",
      type: "page" as const,
      id: "pg1",
      workspaceId: "ws1",
      title: "T",
      body: "x".repeat(200),
      grants: { public: false, workspace: true, user: false, teams: [], users: [] },
      archived: false,
      url: "/w/ws1/page/pg1",
      createdAt: "t",
      updatedAt: "t"
    };
    expect(expandChunkDocs(whole)).toEqual([whole]); // exactly one, no chunks
  });

  it("non-chunkable types (block/comment/…) ⇒ unchanged, single doc", async () => {
    const doc = {
      docId: "comment:c1",
      type: "comment" as const,
      id: "c1",
      workspaceId: "ws1",
      title: "C",
      body: "y".repeat(5000), // over threshold but NOT a page/record
      grants: { public: false, workspace: true, user: false, teams: [], users: [] },
      archived: false,
      url: "/w/ws1",
      createdAt: "t",
      updatedAt: "t"
    };
    expect(expandChunkDocs(doc)).toEqual([doc]);
  });

  it("indexDocs writes chunk docs to the index but NOT to the mirror", async () => {
    const { prisma, page, searchDocument } = fakePrisma();
    page.findUnique.mockResolvedValue(bigPage());
    const whole = (await buildPageDoc({ prisma, now: () => NOW }, "pg1"))!;
    const fs = fakeSearch(prisma);

    await indexDocs(fs.ops, [whole]);
    const indexed = allIndexed(fs);
    expect(indexed.filter((d) => d.type === "chunk").length).toBeGreaterThan(0);
    expect(indexed.filter((d) => d.type === "chunk").every((d) => d.parentDocId === "page:pg1")).toBe(true);
    // the whole doc IS mirrored; the chunk docs are NOT (mirror rows = 1, the page)
    expect(searchDocument.upsert).toHaveBeenCalledTimes(1);
    const mirrored = searchDocument.upsert.mock.calls[0]![0] as { create: { type: string; refId: string } };
    expect(mirrored.create.type).toBe("page");
    expect(mirrored.create.refId).toBe("pg1");
  });

  it("indexDocs clears the parent's stale chunks before writing (shrink/grow safe)", async () => {
    const { prisma, page } = fakePrisma();
    page.findUnique.mockResolvedValue(bigPage());
    const whole = (await buildPageDoc({ prisma, now: () => NOW }, "pg1"))!;
    const fs = fakeSearch(prisma);
    await indexDocs(fs.ops, [whole]);
    expect(fs.deleteChunksOf).toHaveBeenCalledTimes(1);
    expect(fs.deleteChunksOf).toHaveBeenCalledWith(["page:pg1"]);
  });

  it("TASK-0125: indexDocs enqueues embed:chunks for page/record only — and survives a failed enqueue", async () => {
    const { prisma, page } = fakePrisma();
    page.findUnique.mockResolvedValue(bigPage());
    const whole = (await buildPageDoc({ prisma, now: () => NOW }, "pg1"))!;
    const blockDoc = {
      docId: "block:b1",
      type: "block" as const,
      id: "b1",
      workspaceId: "ws1",
      title: "b",
      body: "x",
      grants: { public: false, workspace: true, user: false, teams: [], users: [] },
      archived: false,
      url: "/w/ws1",
      createdAt: "t",
      updatedAt: "t"
    };
    const enqueueEmbed = vi.fn(async () => {});
    const fs = fakeSearch(prisma);
    fs.ops.enqueueEmbed = enqueueEmbed;
    await indexDocs(fs.ops, [whole, blockDoc]);
    // the page → one embed job (refId parsed from the docId); the block → none
    expect(enqueueEmbed).toHaveBeenCalledTimes(1);
    expect(enqueueEmbed).toHaveBeenCalledWith({ workspaceId: "ws1", docType: "page", refId: "pg1" });

    // a failed enqueue must NOT fail the index write (nightly backstop)
    const fs2 = fakeSearch(prisma);
    fs2.ops.enqueueEmbed = vi.fn(async () => {
      throw new Error("redis down");
    });
    await expect(indexDocs(fs2.ops, [whole])).resolves.toMatchObject({ indexed: expect.any(Number) });
  });

  it("a record body over threshold is chunked too (parent-driven, #c<n>)", async () => {
    const whole = {
      docId: "record:r1",
      type: "record" as const,
      id: "r1",
      workspaceId: "ws1",
      title: "R",
      body: LONG,
      grants: { public: false, workspace: true, user: false, teams: [], users: [] },
      archived: false,
      url: "/w/ws1/databases/d1/r1",
      createdAt: "t",
      updatedAt: "t"
    };
    const expanded = expandChunkDocs(whole);
    expect(expanded.length).toBe(chunkText(LONG, DEFAULT_CHUNK_OPTIONS).length + 1);
    expect(expanded[1]!.type).toBe("chunk");
    expect(expanded[1]!.docId).toBe("record:r1#c0");
    expect(expanded[1]!.parentDocId).toBe("record:r1");
  });
});

describe("buildUserDoc (per-workspace doc)", () => {
  it("stamps docId user:<ws>:<uid> and grants every workspace member", async () => {
    const { prisma, user } = fakePrisma();
    user.findUnique.mockResolvedValue({
      id: "u1",
      name: "Ada",
      displayName: "Ada L.",
      email: "ada@x.io",
      status: "ACTIVE",
      updatedAt: NOW
    });
    const doc = await buildUserDoc({ prisma, now: () => NOW }, "u1", "ws1");
    expect(doc!.docId).toBe("user:ws1:u1");
    expect(doc!.id).toBe("u1");
    expect(doc!.title).toBe("Ada L.");
    expect(doc!.grants.workspace).toBe(true);
    expect(doc!.url).toBe("/w/ws1");
    expect(mirrorRefId(doc!)).toBe("ws1:u1");
  });

  it("suspended account ⇒ null", async () => {
    const { prisma, user } = fakePrisma();
    user.findUnique.mockResolvedValue({ id: "u1", name: "Ada", displayName: null, email: "a@x.io", status: "SUSPENDED", updatedAt: NOW });
    await expect(buildUserDoc({ prisma, now: () => NOW }, "u1", "ws1")).resolves.toBeNull();
  });
});

describe("contentHash / mirrorRefId", () => {
  const g = { public: false, workspace: true, user: false, teams: [], users: [] };
  it("is deterministic and sensitive to content/grants", () => {
    const a = builderHash({ title: "t", body: "b", grants: g, archived: false });
    const b = builderHash({ title: "t", body: "b", grants: g, archived: false });
    const c = builderHash({ title: "t", body: "B", grants: g, archived: false });
    const d = builderHash({ title: "t", body: "b", grants: { ...g, public: true }, archived: false });
    expect(a).toBe(b);
    expect(a).not.toBe(c);
    expect(a).not.toBe(d);
    expect(a).toMatch(/^[0-9a-f]{40}$/);
  });
  it("mirrorRefId: plain id except user (workspace-scoped)", () => {
    expect(mirrorRefId({ type: "page", id: "p1", workspaceId: "ws1" })).toBe("p1");
    expect(mirrorRefId({ type: "user", id: "u1", workspaceId: "ws1" })).toBe("ws1:u1");
    expect(mirrorRefId({ type: "user", id: "u1", workspaceId: undefined })).toBe("u1");
  });
});

describe("blockText / renderPropertyValue", () => {
  it("blockText: text | parts[] | none", () => {
    expect(blockText({ text: "hi" })).toBe("hi");
    expect(blockText({ parts: [{ text: "a" }, { text: "b" }] })).toBe("ab");
    expect(blockText({})).toBe("");
    expect(blockText(null)).toBe("");
  });
  it("renderPropertyValue: relations → linked titles", () => {
    const rel = new Map([["r1", "Linked Task"]]);
    expect(renderPropertyValue("RELATION", ["r1", "r2"], rel)).toBe("Linked Task, r2");
    expect(renderPropertyValue("TEXT", null)).toBe("");
    expect(renderPropertyValue("NUMBER", 42)).toBe("42");
  });
});

describe("upsertMirror / deleteMirror", () => {
  it("upsert stores the grant snapshot + contentHash (create)", async () => {
    const { prisma, searchDocument } = fakePrisma();
    const doc: SearchDoc = {
      docId: "page:pg1",
      type: "page",
      id: "pg1",
      workspaceId: "ws1",
      title: "Roadmap",
      body: "b",
      grants: { public: false, workspace: true, user: false, teams: [], users: [] },
      archived: false,
      url: "/w/ws1/page/pg1",
      createdAt: NOW.toISOString(),
      updatedAt: NOW.toISOString()
    };
    const res = await upsertMirror(prisma as never, doc);
    expect(res.contentHash).toBe(builderHash(doc));
    expect(searchDocument.upsert).toHaveBeenCalledWith(
      expect.objectContaining({
        where: { type_refId: { type: "page", refId: "pg1" } },
        create: expect.objectContaining({
          permissions: doc.grants,
          contentHash: res.contentHash,
          metadata: expect.objectContaining({ contentHash: res.contentHash, url: doc.url })
        })
      })
    );
  });

  it("deleteMirror uses (type, refId) with workspace-scoped user refId", async () => {
    const { prisma, searchDocument } = fakePrisma();
    await deleteMirror(prisma as never, "user", "u1", "ws1");
    expect(searchDocument.deleteMany).toHaveBeenCalledWith({
      where: { type: "user", refId: "ws1:u1" }
    });
    await deleteMirror(prisma as never, "page", "pg1");
    expect(searchDocument.deleteMany).toHaveBeenLastCalledWith({
      where: { type: "page", refId: "pg1" }
    });
  });
});

// ---------------------------------------------------------------------------
// processor.ts — event dispatch
// ---------------------------------------------------------------------------

describe("handleSearchEvent", () => {
  it("page.created indexes the page doc + block docs", async () => {
    const { prisma, page } = fakePrisma();
    const fs = fakeSearch(prisma);
    page.findUnique.mockResolvedValue(basePage());
    await handleSearchEvent(fs.ops, { event: "page.created", payload: { workspaceId: "ws1", pageId: "pg1" } });
    const docs = allIndexed(fs);
    expect(docs.map((d) => d.docId).sort()).toEqual(["block:b1", "page:pg1"]);
    expect(docs.find((d) => d.docId === "page:pg1")!.grants.workspace).toBe(true);
  });

  it("page.deleted permanent ⇒ deletes page + mirror-enumerated blocks", async () => {
    const { prisma, page, searchDocument } = fakePrisma();
    const fs = fakeSearch(prisma);
    // source rows are gone (purged) — the mirror still lists the block docs
    page.findUnique.mockResolvedValue(null);
    searchDocument.findMany.mockResolvedValue([{ refId: "b1" }, { refId: "b2" }]);
    await handleSearchEvent(fs.ops, {
      event: "page.deleted",
      payload: { workspaceId: "ws1", pageId: "pg1", permanent: true, pageIds: ["pg1", "pg2"] }
    });
    expect(fs.deleted).toContain("page:pg1");
    expect(fs.deleted).toContain("page:pg2");
    expect(fs.deleted).toContain("block:b1");
    expect(fs.deleted).toContain("block:b2");
  });

  it("page.deleted (trash) removes the doc from the index (restore re-adds via reconcile)", async () => {
    const { prisma, page, searchDocument } = fakePrisma();
    const fs = fakeSearch(prisma);
    // The source row is trashed when the event arrives → no doc to rebuild.
    page.findUnique.mockResolvedValue({ ...basePage(), trashedAt: NOW });
    // The mirror still lists the block docs (block source rows survive trash)
    // — they must be removed too.
    searchDocument.findMany.mockResolvedValue([{ refId: "b1" }]);
    await handleSearchEvent(fs.ops, {
      event: "page.deleted",
      payload: { workspaceId: "ws1", pageId: "pg1", permanent: false, pageIds: ["pg1"] }
    });
    expect(fs.deleted).toContain("page:pg1");
    expect(fs.deleted).toContain("block:b1");
    expect(allIndexed(fs)).toEqual([]); // nothing re-upserted
  });

  it("permission.changed ⇒ workspace resync enqueued (never targeted)", async () => {
    const { prisma } = fakePrisma();
    const fs = fakeSearch(prisma);
    await handleSearchEvent(fs.ops, {
      event: "permission.changed",
      payload: { workspaceId: "ws1", userId: "u2", role: "MEMBER" }
    });
    expect(enqueue).toHaveBeenCalledWith(fs.ops.connection, QUEUES.maintenance, {
      type: "workspace-resync",
      workspaceId: "ws1"
    });
    expect(fs.indexed).toEqual([]); // no direct doc write
  });

  it("comment.deleted ⇒ doc removed", async () => {
    const { prisma, searchDocument } = fakePrisma();
    const fs = fakeSearch(prisma);
    searchDocument.deleteMany.mockResolvedValue({ count: 1 });
    await handleSearchEvent(fs.ops, {
      event: "comment.deleted",
      payload: { workspaceId: "ws1", commentId: "c1" }
    });
    expect(fs.deleted).toEqual(["comment:c1"]);
    expect(searchDocument.deleteMany).toHaveBeenCalled();
  });

  it("user.register stamps the per-workspace user doc", async () => {
    const { prisma, user } = fakePrisma();
    const fs = fakeSearch(prisma);
    user.findUnique.mockResolvedValue({
      id: "u1",
      name: "Ada",
      displayName: null,
      email: "ada@x.io",
      status: "ACTIVE",
      updatedAt: NOW
    });
    await handleSearchEvent(fs.ops, {
      event: "user.register",
      payload: { userId: "u1", workspaceId: "ws1" }
    });
    expect(allIndexed(fs).map((d) => d.docId)).toEqual(["user:ws1:u1"]);
  });

  it("unknown source row (poison) ⇒ logs, does not throw", async () => {
    const { prisma } = fakePrisma();
    const fs = fakeSearch(prisma);
    await expect(
      handleSearchEvent(fs.ops, { event: "database.created", payload: { workspaceId: "ws1", databaseId: "gone" } })
    ).resolves.toBeUndefined();
    expect(fs.indexed).toEqual([]);
  });

  it("transport failure ⇒ rethrows (entry stays pending)", async () => {
    const { prisma, page } = fakePrisma();
    const fs = fakeSearch(prisma);
    page.findUnique.mockResolvedValue(basePage());
    fs.ops.search.indexDocs = vi.fn(async () => {
      throw new Error("opensearch down");
    }) as never;
    await expect(
      handleSearchEvent(fs.ops, { event: "page.updated", payload: { pageId: "pg1" } })
    ).rejects.toThrow("opensearch down");
  });

  it("event not in the search subscription ⇒ no-op", async () => {
    const { prisma } = fakePrisma();
    const fs = fakeSearch(prisma);
    await handleSearchEvent(fs.ops, { event: "notification.created", payload: { id: "n1" } });
    expect(fs.indexed).toEqual([]);
    expect(fs.deleted).toEqual([]);
  });
});

describe("removeDoc", () => {
  it("tolerates 404 (idempotent delete)", async () => {
    const { prisma } = fakePrisma();
    const fs = fakeSearch(prisma);
    fs.ops.search.deleteDoc = vi.fn(async () => {
      const { ApiError, ErrorCodes } = await import("@factory/shared/errors");
      throw new ApiError(ErrorCodes.SEARCH_UNAVAILABLE, { message: "404", status: 404 });
    }) as never;
    await expect(removeDoc(fs.ops, "page", "pg1", "ws1")).resolves.toBeUndefined();
  });
});

// ---------------------------------------------------------------------------
// reconcile + index management
// ---------------------------------------------------------------------------

describe("runReconcile (contentHash sweep)", () => {
  it("repairs drifted docs and removes orphan mirror rows", async () => {
    const { prisma, page, searchDocument } = fakePrisma();
    const fs = fakeSearch(prisma);
    // one in-sync row, one drifted, one orphan (source gone)
    const inSyncDoc = await (async () => {
      page.findUnique.mockResolvedValueOnce(basePage());
      return buildPageDoc({ prisma, now: () => NOW }, "pg1");
    })();
    page.findUnique.mockReset();
    page.findUnique.mockImplementation(async (a: { where: { id: string } }) =>
      a.where.id === "pg1" ? basePage() : a.where.id === "pg2" ? { ...basePage(), id: "pg2", title: "Changed" } : null
    );
    const inSyncHash = builderHash(inSyncDoc!);
    searchDocument.findMany.mockResolvedValue([
      { type: "page", refId: "pg1", contentHash: inSyncHash, indexedAt: NOW },
      { type: "page", refId: "pg2", contentHash: "stale", indexedAt: NOW },
      { type: "page", refId: "pg3", contentHash: "stale", indexedAt: NOW }
    ]);
    const res = await runReconcile(fs.ops, "ws1");
    // pg1 in sync (skipped), pg2 repaired (rebuilt from the fake), pg3 removed
    expect(res.checked).toBe(3);
    expect(res.repaired).toBe(1);
    expect(res.removed).toBe(1);
    expect(fs.deleted).toContain("page:pg3");
  });

  it("re-indexes a doc LOST from the index (healthy mirror, missing in OpenSearch)", async () => {
    const { prisma, page, searchDocument } = fakePrisma();
    const fs = fakeSearch(prisma);
    page.findUnique.mockResolvedValue(basePage());
    // Mirror says the doc is current, but the index lost it (deleted directly).
    const hash = builderHash((await buildPageDoc({ prisma, now: () => NOW }, "pg1"))!);
    searchDocument.findMany.mockResolvedValue([
      { type: "page", refId: "pg1", contentHash: hash, indexedAt: NOW }
    ]);
    // docExists reports page:pg1 MISSING.
    fs.docExists.mockImplementation(async (ids: string[]) => new Set(ids.filter((id) => id !== "page:pg1")));
    const res = await runReconcile(fs.ops, "ws1");
    expect(res.repaired).toBe(1);
    expect(allIndexed(fs).map((d) => d.docId)).toContain("page:pg1");
  });

  it("re-adds a RESTORED page via the source→index direction (no mirror row)", async () => {
    const { prisma, page, searchDocument } = fakePrisma();
    const fs = fakeSearch(prisma);
    // The trash dropped the mirror row; the RESTORE published no domain event,
    // so only the source table has the page again (trashedAt null).
    searchDocument.findMany.mockResolvedValue([]);
    page.findMany.mockResolvedValue([{ id: "pg1" }]);
    page.findUnique.mockResolvedValue(basePage()); // trashedAt null
    // The index lost it (docExists says not present).
    fs.docExists.mockResolvedValue(new Set());
    const res = await runReconcile(fs.ops, "ws1");
    // Page + its block doc both re-added from the source tables.
    expect(res.repaired).toBe(2);
    expect(allIndexed(fs).map((d) => d.docId).sort()).toEqual(["block:b1", "page:pg1"]);
  });
});

describe("runReindex (full (re)index, T-156 org scoping)", () => {
  it("no orgId → reindexes ALL workspaces", async () => {
    const { prisma, workspace } = fakePrisma();
    const fs = fakeSearch(prisma);
    workspace.findMany.mockResolvedValue([
      { id: "ws1" },
      { id: "ws2" }
    ]);
    const res = await runReindex(fs.ops);
    expect(workspace.findMany).toHaveBeenCalledWith({
      where: {},
      select: { id: true }
    });
    expect(res.workspaces).toBe(2);
  });

  it("orgId → ONLY that org's workspaces are reindexed (other orgs untouched)", async () => {
    const { prisma, workspace } = fakePrisma();
    const fs = fakeSearch(prisma);
    workspace.findMany.mockResolvedValue([
      { id: "ws1" },
      { id: "ws2" }
    ]);
    const res = await runReindex(fs.ops, undefined, "org1");
    expect(workspace.findMany).toHaveBeenCalledWith({
      where: { organizationId: "org1" },
      select: { id: true }
    });
    // The where clause is the scoping guarantee: a second org's workspaces
    // (ws3) would be filtered out before any doc is collected.
    workspace.findMany.mockResolvedValue([{ id: "ws3" }]);
    await runReindex(fs.ops, undefined, "org2");
    expect(workspace.findMany).toHaveBeenLastCalledWith({
      where: { organizationId: "org2" },
      select: { id: true }
    });
    // and an org with NO workspaces reindexes nothing (the loop is a no-op)
    workspace.findMany.mockResolvedValue([]);
    const empty = await runReindex(fs.ops, undefined, "org-empty");
    expect(empty.workspaces).toBe(0);
    expect(empty.docs).toBe(0);
  });
});

describe("ensureContentIndex", () => {
  it("creates the index + alias when missing (real SearchClient, fake fetch)", async () => {
    const calls: { path: string; method: string }[] = [];
    const fetchImpl = (vi.fn(async (url: string | URL | Request) => {
      const u = String(url);
      const method = u.includes("_bulk") ? "POST" : u.endsWith("_aliases") ? "POST" : u.includes("factory-content") ? "PUT" : "GET";
      calls.push({ path: u.replace("http://os:9200", ""), method });
      if (u.endsWith("/factory-content-v1")) {
        // first GET: exists?
        if (calls.filter((c) => c.path === "/factory-content-v1").length === 1) {
          return new Response("nf", { status: 404 });
        }
        return new Response(JSON.stringify({ mappings: { properties: { vector: { dimension: 1536 } } } }), { status: 200 });
      }
      return new Response(JSON.stringify({}), { status: 200 });
    }) as unknown) as typeof fetch;
    const { prisma } = fakePrisma();
    const search = new SearchClient({ baseUrl: "http://os:9200", fetchImpl });
    const ops: SearchIndexOps = { search, prisma, connection: { xadd: vi.fn() } as never, env: ENV };
    const res = await ensureContentIndex(ops);
    expect(res.created).toBe(true);
    const paths = calls.map((c) => c.path);
    expect(paths).toContain("/factory-content-v1"); // PUT create
    expect(paths).toContain("/_aliases"); // alias swap
  });

  it("leaves a dimension-mismatched index alone (rollover warning path)", async () => {
    const { prisma } = fakePrisma();
    const search = {
      indexExists: vi.fn(async () => true),
      getVectorDimension: vi.fn(async () => 768),
      createIndexIfMissing: vi.fn(),
      createAlias: vi.fn()
    } as unknown as SearchClient;
    const ops: SearchIndexOps = { search, prisma, connection: { xadd: vi.fn() } as never, env: ENV };
    const res = await ensureContentIndex(ops);
    expect(res.created).toBe(false);
    expect(search.createIndexIfMissing).not.toHaveBeenCalled();
    expect(search.createAlias).not.toHaveBeenCalled();
  });
});

describe("recordSearchDlq", () => {
  it("xadds a MAXLEN-capped DLQ entry", async () => {
    const { prisma } = fakePrisma();
    const fs = fakeSearch(prisma);
    const job = { id: "job-9", data: { type: "reindex", workspaceId: "ws1" } } as never;
    await recordSearchDlq(fs.ops.connection, job, new Error("boom"));
    expect(fs.xadd).toHaveBeenCalledTimes(1);
    // xadd(key, "MAXLEN", "~", "1000", "*", "jobId", ..., "error", ...)
    // The "*" stream id is MANDATORY — omitting it made Redis reject the
    // command ("ERR Invalid stream ID") and crash the whole worker process.
    const args = fs.xadd.mock.calls[0]! as unknown as string[];
    expect(args[0]).toBe("search-index:dlq");
    expect(args).toContain("MAXLEN");
    expect(args).toContain("~");
    expect(args[4]).toBe("*"); // stream id position (after MAXLEN ~ 1000)
    expect(args).toContain("job-9");
    expect(args).toContain("boom");
  });
});