import { Inject, Injectable } from "@nestjs/common";
import { Prisma } from "@prisma/client";
import { z } from "zod";
import { PrismaService } from "../../common/prisma.service.js";
import { AuditService } from "../../common/audit.service.js";
import { EventBus } from "../../common/event-bus.js";
import { NotificationsService } from "../notifications/notifications.service.js";
import { CacheService } from "../../common/cache.js";
import { ApiError } from "../../common/errors.js";
import { assertCan } from "../../lib/authorize.js";
import { between, compare, rebalancePositions } from "../../lib/positions.js";
import { EVENTS, MAX_BLOCK_TREE } from "../../constants.js";
import type {
  BatchBlocksInput,
  CreateBlockInput,
  MoveBlockInput,
  UpdateBlockInput
} from "./dto.js";

const BLOCK_SELECT = {
  id: true, pageId: true, workspaceId: true, parentId: true, type: true,
  position: true, content: true, properties: true, isSynced: true,
  syncedSourceId: true, version: true, createdAt: true, updatedAt: true,
  deletedAt: true
} as const;
type BlockRow = Prisma.BlockGetPayload<{ select: typeof BLOCK_SELECT }>;

/**
 * TASK-0135: the full-page block tree (the editor-open load) is the hottest
 * read. It is cached under a version-key (see common/cache.ts): the key embeds
 * the workspace content-version, which every page/block write bumps — so a
 * write always changes the key and the next read is fresh. Entries are
 * zod-validated on read (corrupt entry = miss, never a 500). The shape below
 * matches BLOCK_SELECT as JSON (Prisma dates serialize to ISO strings).
 */
const BLOCK_ROW_CACHE = z.object({
  id: z.string(),
  pageId: z.string(),
  workspaceId: z.string(),
  parentId: z.string().nullable(),
  type: z.string(),
  position: z.string(),
  content: z.unknown(),
  properties: z.unknown(),
  isSynced: z.boolean(),
  syncedSourceId: z.string().nullable(),
  version: z.number(),
  createdAt: z.string(),
  updatedAt: z.string(),
  deletedAt: z.string().nullable()
});
const BLOCK_ROWS_CACHE = z.array(BLOCK_ROW_CACHE);

/**
 * Block service: CRUD on the block tree, fractional positioning, batch ops,
 * duplicate, and flat text export. Every mutation emits `page.updated` with
 * the affected `blockIds` (the realtime sync contract).
 */
@Injectable()
export class BlockService {
  constructor(
    @Inject(PrismaService) private readonly prisma: PrismaService,
    @Inject(AuditService) private readonly audit: AuditService,
    @Inject(EventBus) private readonly events: EventBus,
    @Inject(CacheService) private readonly cache: CacheService,
    @Inject(NotificationsService) private readonly notifications: NotificationsService
  ) {}

  private async getBlock(blockId: string) {
    const block = await this.prisma.block.findUnique({ where: { id: blockId }, select: BLOCK_SELECT });
    if (!block || block.deletedAt) throw ApiError.notFound("Block not found.");
    return block;
  }

  private async requirePageAccess(pageId: string, userId: string, need: "VIEW" | "EDIT") {
    const page = await this.prisma.page.findUnique({
      where: { id: pageId },
      select: { id: true, workspaceId: true, deletedAt: true, version: true }
    });
    if (!page || page.deletedAt) throw ApiError.notFound("Page not found.");
    await assertCan(this.prisma, { workspaceId: page.workspaceId, userId, resource: { pageId: page.id } }, need);
    return page;
  }

  /** Siblings of a parent (or root), ordered. */
  private async siblings(tx: Prisma.TransactionClient, pageId: string, parentId: string | null): Promise<BlockRow[]> {
    return tx.block.findMany({
      where: { pageId, parentId, deletedAt: null },
      select: BLOCK_SELECT,
      orderBy: { position: "asc" }
    });
  }

  // ---------- list ----------

  async list(pageId: string, userId: string, query: { parentId?: string | null; depth?: string | number }) {
    const page = await this.requirePageAccess(pageId, userId, "VIEW");
    const depth = typeof query.depth === "string" ? query.depth : query.depth ? String(query.depth) : "1";

    if (depth === "all") {
      // TASK-0135: cache the full tree (the editor-open load) under a
      // version-key. `pageId` scopes the key; the version is the workspace
      // content-version (bumped on every page/block write). The loader is the
      // real PG read — on a cache hit we skip it entirely.
      const { data: rowsData } = await this.cache.getTree(page.workspaceId, pageId, async () => {
        // TASK-0135 (replica routing): the tree load is a spec-named replica
        // read (`db.read()`). The version-key cache above already guards
        // concurrent-write staleness; a tiny replica-lag window on a cache
        // MISS is the documented, spec-acceptable staleness for tree loads.
        const r = await this.prisma.read().block.findMany({
          where: { pageId, deletedAt: null },
          select: BLOCK_SELECT,
          orderBy: { position: "asc" }
        });
        return r as unknown[];
      }, BLOCK_ROWS_CACHE);
      const flat = rowsData as unknown as BlockRow[];
      if (flat.length > MAX_BLOCK_TREE) {
        throw new ApiError("PAYLOAD_TOO_LARGE", 413, `Block tree exceeds ${MAX_BLOCK_TREE} blocks.`, [
          { maxBlocks: MAX_BLOCK_TREE, received: flat.length }
        ]);
      }
      return { data: flat, tree: this.buildTree(flat), total: flat.length };
    }

    const parentId = query.parentId ?? null;
    const rows = await this.prisma.block.findMany({
      where: { pageId, parentId, deletedAt: null },
      select: BLOCK_SELECT,
      orderBy: { position: "asc" },
      take: 1000
    });
    return { data: rows };
  }

  private buildTree(rows: BlockRow[]) {
    const byId = new Map<string, BlockRow & { children: unknown[] }>();
    for (const r of rows) byId.set(r.id, { ...r, children: [] });
    const roots: (BlockRow & { children: unknown[] })[] = [];
    for (const r of rows) {
      const node = byId.get(r.id)!;
      if (r.parentId && byId.has(r.parentId)) byId.get(r.parentId)!.children.push(node);
      else roots.push(node);
    }
    return roots;
  }

  // ---------- position helpers ----------

  private async resolvePosition(
    tx: Prisma.TransactionClient,
    pageId: string,
    opts: { parentId?: string | null; afterBlockId?: string; beforeBlockId?: string; position?: string }
  ): Promise<string> {
    const parentId = opts.parentId ?? null;
    if (opts.position) return opts.position;

    let before: string | undefined;
    let after: string | undefined;
    if (opts.beforeBlockId) {
      const b = await tx.block.findUnique({ where: { id: opts.beforeBlockId }, select: { position: true, parentId: true, pageId: true, deletedAt: true } });
      if (!b || b.deletedAt || b.pageId !== pageId) throw ApiError.notFound("beforeBlockId not found.");
      before = b.position;
    }
    if (opts.afterBlockId) {
      const a = await tx.block.findUnique({ where: { id: opts.afterBlockId }, select: { position: true, parentId: true, pageId: true, deletedAt: true } });
      if (!a || a.deletedAt || a.pageId !== pageId) throw ApiError.notFound("afterBlockId not found.");
      after = a.position;
    }
    // `after` = the block the new position sits BELOW (lower bound, sorts
    // earlier); `before` = the block it sits ABOVE (upper bound, sorts later).
    // `between(a, b)` requires a to sort before b, so pass (lower, upper) —
    // the same (after, before) ordering database.service uses. The previous
    // `between(before, after)` always threw (before sorts after after) → 500
    // on every move that sent both anchors (moving a middle block).
    if (before && after) return between(after, before);
    if (after) {
      // A single AFTER anchor must be bounded by the anchor's NEXT LIVE
      // sibling (in the target parent), not by the grid top: `between(a)`
      // is deterministic (midpoint of a and top), so re-inserting after the
      // same anchor returns the SAME position the anchor's next sibling
      // already occupies → duplicate position (ordering corruption). With
      // the neighbour bound the new position always lands strictly between
      // two live siblings. No next sibling → the append case (after top).
      const next = await tx.block.findFirst({
        where: { pageId, parentId, deletedAt: null, position: { gt: after } },
        orderBy: { position: "asc" },
        select: { position: true }
      });
      return between(after, next?.position);
    }
    if (before) {
      // Mirror image: bound a single BEFORE anchor from below by its
      // previous live sibling (previously `between(undefined, before)` — the
      // deterministic bottom-midpoint, same collision class).
      const prev = await tx.block.findFirst({
        where: { pageId, parentId, deletedAt: null, position: { lt: before } },
        orderBy: { position: "desc" },
        select: { position: true }
      });
      return between(prev?.position, before);
    }
    const last = await tx.block.findFirst({
      where: { pageId, parentId, deletedAt: null },
      orderBy: { position: "desc" },
      select: { position: true }
    });
    return between(last?.position);
  }

  /**
   * Re-space a sibling list that exhausted its fractional headroom (the
   * historical "position too close to the top" RangeError on large pastes):
   * rewrite every row to an even grid (rebalancePositions) and return the
   * fresh positions by id. Runs inside the caller's transaction — the
   * pending updates are visible to subsequent reads in the same tx.
   */
  private async rebalanceSiblingList(tx: Prisma.TransactionClient, pageId: string, parentId: string | null): Promise<Map<string, string>> {
    const siblings = await this.siblings(tx, pageId, parentId);
    const fresh = rebalancePositions(siblings.map((s) => s.position));
    const byId = new Map<string, string>();
    for (let i = 0; i < siblings.length; i++) {
      const s = siblings[i]!;
      const pos = fresh[i]!;
      if (s.position !== pos) {
        await tx.block.update({ where: { id: s.id }, data: { position: pos } });
      }
      byId.set(s.id, pos);
    }
    return byId;
  }

  /**
   * `resolvePosition` with a last-resort rebalance: when the sibling list
   * has no splittable gap left, re-space the whole list in-transaction and
   * recompute against the fresh grid. The happy path costs nothing extra.
   */
  private async resolvePositionSafe(tx: Prisma.TransactionClient, pageId: string, opts: { parentId?: string | null; afterBlockId?: string; beforeBlockId?: string; position?: string }): Promise<string> {
    try {
      return await this.resolvePosition(tx, pageId, opts);
    } catch (err) {
      if (!(err instanceof RangeError)) throw err;
      await this.rebalanceSiblingList(tx, pageId, opts.parentId ?? null);
      return this.resolvePosition(tx, pageId, opts);
    }
  }

  /** Fill the positional gap left by a removed sibling. */
  private async fillGap(tx: Prisma.TransactionClient, pageId: string, parentId: string | null, removedPosition: string): Promise<void> {
    const siblings = await this.siblings(tx, pageId, parentId);
    const prev = [...siblings].reverse().find((s) => compare(s.position, removedPosition) < 0);
    const next = siblings.find((s) => compare(s.position, removedPosition) > 0);
    if (prev && next) {
      await tx.block.update({ where: { id: prev.id }, data: { position: between(prev.position, next.position) } });
    } else if (prev) {
      await tx.block.update({ where: { id: prev.id }, data: { position: between(prev.position) } });
    }
  }

  // ---------- create / update ----------

  /**
   * A `column_list` created through the API (slash menu, paste import, AI)
   * ships as a Notion-style 2-column grid: two `column` children, each with
   * one empty `paragraph` inside. Without this the list is an uneditable
   * dead end (the renderer has no "add column" affordance and `column` rows
   * render nothing standalone).
   */
  private async seedColumnChildren(tx: Prisma.TransactionClient, block: { id: string; pageId: string; workspaceId: string }, userId: string): Promise<void> {
    const workspaceId = block.workspaceId;
    for (const colPos of ["A", between("A")]) {
      const column = await tx.block.create({
        data: {
          pageId: block.pageId,
          workspaceId,
          parentId: block.id,
          // The BlockType column is the UPPERCASE DB enum (the REST DTO
          // transform stores `type` upper-cased), so the seed children must
          // use the enum literals or the insert fails.
          type: "COLUMN",
          position: colPos,
          content: {} as Prisma.InputJsonValue,
          properties: {} as object,
          createdBy: userId
        }
      });
      await tx.block.create({
        data: {
          pageId: block.pageId,
          workspaceId,
          parentId: column.id,
          type: "PARAGRAPH",
          position: "A",
          content: {} as Prisma.InputJsonValue,
          properties: {} as object,
          createdBy: userId
        }
      });
    }
  }

  async create(pageId: string, userId: string, input: CreateBlockInput, req: any) {
    const page = await this.requirePageAccess(pageId, userId, "EDIT");
    const block = await this.prisma.$transaction(async (tx) => {
      const parentId = input.parentId ?? null;
      if (parentId) {
        const parent = await tx.block.findUnique({ where: { id: parentId }, select: { pageId: true, deletedAt: true } });
        if (!parent || parent.deletedAt || parent.pageId !== pageId) throw ApiError.notFound("parentId not found.");
      }
      const position = await this.resolvePositionSafe(tx, pageId, input);
      const created = await tx.block.create({
        data: {
          pageId,
          workspaceId: page.workspaceId,
          parentId,
          type: input.type,
          position,
          content: input.content as Prisma.InputJsonValue,
          properties: input.properties as object,
          createdBy: userId
        },
        select: BLOCK_SELECT
      });
      if (created.type === "COLUMN_LIST") {
        // UPPERCASE: the REST DTO transform stores `type` as the DB enum
        // (the client sends codec-lowercase `column_list`).
        await this.seedColumnChildren(tx, created, userId);
      }
      return created;
    });
    await this.audit.audit(req, req?.auth, {
      action: "block.created",
      resourceType: "block",
      resourceId: block.id,
      workspaceId: page.workspaceId,
      pageId,
      metadata: { type: block.type, parentId: block.parentId }
    });
    await this.events.publish(EVENTS.PAGE_UPDATED, {
      workspaceId: page.workspaceId,
      pageId,
      blockIds: [block.id],
      changed: "created"
    });
    return block;
  }

  async update(blockId: string, userId: string, input: UpdateBlockInput, req: any) {
    const block = await this.getBlock(blockId);
    await this.requirePageAccess(block.pageId, userId, "EDIT");
    const data: Prisma.BlockUpdateInput = { version: { increment: 1 }, updatedBy: userId };
    if (input.content !== undefined) data.content = input.content as object;
    if (input.type !== undefined) {
      data.type = input.type;
      // type conversion: unsupported props are cleared
      data.properties = (input.properties ?? {}) as object;
    } else if (input.properties !== undefined) {
      data.properties = input.properties as object;
    }
    if (input.archived !== undefined) {
      // soft-archive via deletedAt (blocks have no archivedAt column)
      data.deletedAt = input.archived ? new Date() : null;
    }
    // TASK-0123: optional optimistic concurrency. When `baseVersion` is given
    // the write only lands if the block is STILL at that version; a stale
    // version is a 409 CONFLICT carrying the actual version (the writing
    // apply path uses this to offer a rebase). Omitted → today's behavior
    // (unconditional increment) — full backward compatibility.
    let updated: BlockRow | null;
    if (input.baseVersion !== undefined) {
      const res = await this.prisma.block.updateMany({
        where: { id: blockId, version: input.baseVersion, deletedAt: null },
        data,
      });
      if (res.count === 0) {
        const current = await this.prisma.block.findUnique({
          where: { id: blockId },
          select: { version: true, deletedAt: true }
        });
        if (!current || current.deletedAt) throw ApiError.notFound("Block not found.");
        throw ApiError.conflict("Block version conflict (stale version).", [
          { reason: "stale_version", blockId, expected: input.baseVersion, actualVersion: current.version }
        ]);
      }
      updated = await this.prisma.block.findUnique({ where: { id: blockId }, select: BLOCK_SELECT });
    } else {
      updated = await this.prisma.block.update({ where: { id: blockId }, data, select: BLOCK_SELECT });
    }
    if (!updated) throw ApiError.notFound("Block not found.");
    await this.audit.audit(req, req?.auth, {
      action: "block.updated",
      resourceType: "block",
      resourceId: blockId,
      workspaceId: block.workspaceId,
      pageId: block.pageId,
      metadata: { fields: Object.keys(input) }
    });
    await this.events.publish(EVENTS.PAGE_UPDATED, {
      workspaceId: block.workspaceId,
      pageId: block.pageId,
      blockIds: [blockId],
      changed: "updated"
    });
    return updated;
  }

  // ---------- move / delete / duplicate ----------

  async move(blockId: string, userId: string, input: MoveBlockInput, req: any) {
    const block = await this.getBlock(blockId);
    await this.requirePageAccess(block.pageId, userId, "EDIT");
    const result = await this.prisma.$transaction(async (tx) => {
      const newParentId = input.parentId ?? block.parentId;
      if (newParentId) {
        const parent = await tx.block.findUnique({ where: { id: newParentId }, select: { pageId: true, deletedAt: true } });
        if (!parent || parent.deletedAt || parent.pageId !== block.pageId) throw ApiError.notFound("Target parent not found.");
        if (this.isDescendant(await tx.block.findMany({ where: { pageId: block.pageId, deletedAt: null }, select: { id: true, parentId: true } }), blockId, newParentId)) {
          throw ApiError.validation("Cannot move a block into its own subtree.", [{ issue: "circular" }]);
        }
      }
      // remove from old position
      await tx.block.update({ where: { id: blockId }, data: { parentId: newParentId } });
      await this.fillGap(tx, block.pageId, block.parentId, block.position);
      const position = await this.resolvePositionSafe(tx, block.pageId, {
        parentId: newParentId,
        afterBlockId: input.afterBlockId,
        beforeBlockId: input.beforeBlockId,
        position: input.position
      });
      return tx.block.update({ where: { id: blockId }, data: { position }, select: BLOCK_SELECT });
    });
    await this.audit.audit(req, req?.auth, {
      action: "block.moved",
      resourceType: "block",
      resourceId: blockId,
      workspaceId: block.workspaceId,
      pageId: block.pageId
    });
    await this.events.publish(EVENTS.PAGE_UPDATED, {
      workspaceId: block.workspaceId,
      pageId: block.pageId,
      blockIds: [blockId],
      moved: true
    });
    return result;
  }

  private isDescendant(allBlocks: { id: string; parentId: string | null }[], ancestorId: string, candidateId: string): boolean {
    let cursor: string | null = candidateId;
    const seen = new Set<string>();
    while (cursor) {
      if (cursor === ancestorId) return true;
      if (seen.has(cursor)) return false;
      seen.add(cursor);
      cursor = allBlocks.find((b) => b.id === cursor)?.parentId ?? null;
    }
    return false;
  }

  async delete(blockId: string, userId: string, req: any) {
    const block = await this.getBlock(blockId);
    await this.requirePageAccess(block.pageId, userId, "EDIT");
    const subtreeIds = await this.subtreeIds(block.pageId, [blockId]);
    await this.prisma.$transaction(async (tx) => {
      await tx.block.deleteMany({ where: { pageId: block.pageId, id: { in: subtreeIds } } });
      await this.fillGap(tx, block.pageId, block.parentId, block.position);
    });
    await this.audit.audit(req, req?.auth, {
      action: "block.deleted",
      resourceType: "block",
      resourceId: blockId,
      workspaceId: block.workspaceId,
      pageId: block.pageId,
      metadata: { subtreeSize: subtreeIds.length }
    });
    await this.events.publish(EVENTS.PAGE_UPDATED, {
      workspaceId: block.workspaceId,
      pageId: block.pageId,
      blockIds: [blockId],
      changed: "deleted"
    });
    return { deleted: true, blockIds: subtreeIds };
  }

  private async subtreeIds(pageId: string, roots: string[]): Promise<string[]> {
    const rows = await this.prisma.block.findMany({
      where: { pageId, deletedAt: null },
      select: { id: true, parentId: true }
    });
    const childrenOf = new Map<string | null, string[]>();
    for (const r of rows) {
      const list = childrenOf.get(r.parentId) ?? [];
      list.push(r.id);
      childrenOf.set(r.parentId, list);
    }
    const ids: string[] = [];
    const queue = [...roots];
    while (queue.length > 0) {
      const id = queue.shift()!;
      ids.push(id);
      queue.push(...(childrenOf.get(id) ?? []));
    }
    return ids;
  }

  /** Deep-copy a block subtree, inserted right after the source. */
  async duplicate(blockId: string, userId: string, req: any, input?: { parentId?: string | null; afterBlockId?: string }) {
    const source = await this.getBlock(blockId);
    await this.requirePageAccess(source.pageId, userId, "EDIT");
    const rows = await this.prisma.block.findMany({
      where: { pageId: source.pageId, deletedAt: null },
      select: { id: true, parentId: true, type: true, content: true, properties: true, position: true }
    });
    const subtree = await this.subtreeIds(source.pageId, [blockId]);
    const set = new Set(subtree);
    const nodes = rows.filter((r) => set.has(r.id));
    const idMap = new Map<string, string>();
    for (const n of nodes) idMap.set(n.id, `blk_${Math.random().toString(36).slice(2, 14)}`);
    // Target slot: by default directly BELOW the source in its own parent
    // list; an explicit `afterBlockId` anchors the copy somewhere else (the
    // web "Paste below" / "Copy" path) and `parentId` re-parents the copy.
    const targetParentId = input?.parentId === undefined ? source.parentId : input.parentId;
    const afterId = input?.afterBlockId ?? blockId;
    await this.prisma.$transaction(async (tx) => {
      const position = await this.resolvePositionSafe(tx, source.pageId, {
        parentId: targetParentId,
        afterBlockId: afterId
      });
      for (const n of nodes) {
        await tx.block.create({
          data: {
            id: idMap.get(n.id)!,
            pageId: source.pageId,
            workspaceId: source.workspaceId,
            // The ROOT lands in the (possibly re-parented) target; inner
            // nodes keep their subtree-relative parents via the id map.
            parentId: n.id === blockId ? targetParentId : (n.parentId ? (idMap.get(n.parentId) ?? n.parentId) : null),
            type: n.type,
            // keep intra-tree relative order; the root gets the computed position
            position: n.id === blockId ? position : n.position,
            content: n.content as Prisma.InputJsonValue,
            properties: n.properties as Prisma.InputJsonValue,
            createdBy: userId
          }
        });
      }
    });
    await this.audit.audit(req, req?.auth, {
      action: "block.duplicated",
      resourceType: "block",
      resourceId: blockId,
      workspaceId: source.workspaceId,
      pageId: source.pageId,
      metadata: { subtreeSize: subtree.length }
    });
    await this.events.publish(EVENTS.PAGE_UPDATED, {
      workspaceId: source.workspaceId,
      pageId: source.pageId,
      blockIds: [...subtree],
      changed: "duplicated"
    });
    return { duplicated: true, newRootId: idMap.get(blockId), blockIds: subtree };
  }

  // ---------- batch ----------

  async batch(pageId: string, userId: string, input: BatchBlocksInput, req: any) {
    const page = await this.requirePageAccess(pageId, userId, "EDIT");
    // Stale-base fast path: a pre-check is an optimization, not the guard —
    // the authoritative check is the CAS UPDATE inside the transaction
    // below. A stale baseVersion 409s immediately with the CURRENT version
    // (the "CONFLICT carries the new version" contract — the REST client
    // re-reads and retries).
    if (input.baseVersion !== undefined && page.version !== input.baseVersion) {
      throw ApiError.conflict(
        "The page changed since you last read it (a concurrent edit — including a realtime flush). Re-read and retry.",
        [
          { issue: "stale_base_version", received: input.baseVersion },
          { currentVersion: page.version },
          { reason: "flush_wins" }
        ]
      );
    }
    const touched = new Set<string>();
    let newVersion: number;
    try {
      newVersion = await this.prisma.$transaction(async (tx) => {
        // The `pages.version` CAS (TASK-0092, backend-architecture §12):
        // the REST batch serializes against a concurrent realtime flush —
        // the flush's `POST /internal/blocks/flush` runs
        // `UPDATE pages SET version = v+1 WHERE version = base`; if a REST
        // write advanced the version first, that CAS fails and the flush
        // rebases over the REST write ("flush wins; the doc absorbs REST
        // writes, never clobbers them"). The bump is the FIRST statement
        // of the transaction (a race between the pre-check above and the
        // UPDATE aborts everything — nothing written, the 409 path below).
        const base = input.baseVersion ?? page.version;
        const cas = await tx.page.updateMany({
          where: { id: pageId, version: base },
          data: { version: base + 1, updatedAt: new Date() }
        });
        if (cas.count !== 1) throw new BatchVersionCasConflict();
        for (const op of input.ops) {
          if (op.op === "update") {
            // The web editor's debounced save (block-save-queue.ts) flushes one
            // `op:"update"` per typed block to this endpoint. Inlined here (NOT
            // `this.update`) so the write stays inside the transaction — batch
            // atomicity ("one invalid op → 400 and NO partial application") and
            // the pages.version CAS hold for update ops too. Semantics mirror
            // the single `update()`: version increment + updatedBy, type
            // conversion clears properties, re-parenting guards against
            // circular moves.
            const block = await tx.block.findUnique({ where: { id: op.blockId }, select: { id: true, pageId: true, parentId: true, position: true, type: true, deletedAt: true } });
            if (!block || block.deletedAt || block.pageId !== pageId) throw ApiError.notFound(`block ${op.blockId} not found.`);
            const patch = op.patch;
            const data: Prisma.BlockUpdateInput = { version: { increment: 1 }, updatedBy: userId };
            if (patch.content !== undefined) data.content = patch.content as unknown as object;
            if (patch.type !== undefined && patch.type !== block.type) {
              data.type = patch.type;
              data.properties = (patch.properties ?? {}) as object;
            } else if (patch.properties !== undefined) {
              data.properties = patch.properties as object;
            }
            if (patch.parentId !== undefined && patch.parentId !== block.parentId) {
              const newParentId = patch.parentId;
              if (newParentId) {
                const parent = await tx.block.findUnique({ where: { id: newParentId }, select: { pageId: true, deletedAt: true } });
                if (!parent || parent.deletedAt || parent.pageId !== pageId) throw ApiError.notFound(`update parent ${newParentId} not found.`);
                if (this.isDescendant(await tx.block.findMany({ where: { pageId, deletedAt: null }, select: { id: true, parentId: true } }), op.blockId, newParentId)) {
                  throw ApiError.validation("Cannot move a block into its own subtree.", [{ issue: "circular" }]);
                }
              }
              data.parentId = newParentId;
            }
            await tx.block.update({ where: { id: op.blockId }, data });
            touched.add(op.blockId);
        } else if (op.op === "upsert") {
          const b = op.block;
          const parentId = b.parentId ?? null;
          if (parentId) {
            const parent = await tx.block.findUnique({ where: { id: parentId }, select: { pageId: true, deletedAt: true } });
            if (!parent || parent.deletedAt || parent.pageId !== pageId) throw ApiError.notFound(`parentId ${parentId} not found.`);
          }
          const position = await this.resolvePositionSafe(tx, pageId, b);
          const created = await tx.block.create({
            data: {
              // Client-chosen ids (the web paste path references nested
              // parents by id within the SAME batch): the DB primary key
              // enforces uniqueness — a collision aborts the whole batch.
              ...(b.id ? { id: b.id } : {}),
              pageId,
              workspaceId: page.workspaceId,
              parentId,
              type: b.type,
              position,
              content: b.content as Prisma.InputJsonValue,
              properties: b.properties as object,
              createdBy: userId
            }
          });
          touched.add(created.id);
        } else if (op.op === "delete") {
          const block = await tx.block.findUnique({ where: { id: op.blockId }, select: { id: true, pageId: true, parentId: true, position: true, deletedAt: true } });
          if (!block || block.deletedAt || block.pageId !== pageId) throw ApiError.notFound(`block ${op.blockId} not found.`);
          const ids = await this.subtreeTx(tx, pageId, [op.blockId]);
          await tx.block.deleteMany({ where: { pageId, id: { in: ids } } });
          await this.fillGap(tx, pageId, block.parentId, block.position);
          touched.add(op.blockId);
        } else if (op.op === "move") {
          const block = await tx.block.findUnique({ where: { id: op.blockId }, select: { id: true, pageId: true, parentId: true, position: true, deletedAt: true } });
          if (!block || block.deletedAt || block.pageId !== pageId) throw ApiError.notFound(`block ${op.blockId} not found.`);
          const mv = op.move;
          const newParentId = mv.parentId ?? block.parentId;
          if (newParentId) {
            const parent = await tx.block.findUnique({ where: { id: newParentId }, select: { pageId: true, deletedAt: true } });
            if (!parent || parent.deletedAt || parent.pageId !== pageId) throw ApiError.notFound(`move parent ${newParentId} not found.`);
          }
          await tx.block.update({ where: { id: op.blockId }, data: { parentId: newParentId } });
          await this.fillGap(tx, pageId, block.parentId, block.position);
          const position = await this.resolvePositionSafe(tx, pageId, mv);
          await tx.block.update({ where: { id: op.blockId }, data: { position } });
          touched.add(op.blockId);
        }
      }
      return base + 1;
    });
    } catch (err) {
      if (err instanceof BatchVersionCasConflict) {
        // The authoritative CAS failed inside the transaction (a concurrent
        // writer advanced pages.version between the pre-check and the
        // UPDATE) → nothing was written; report the CURRENT version.
        const fresh = await this.prisma.page.findUnique({ where: { id: pageId }, select: { version: true } });
        const current = fresh?.version ?? page.version + 1;
        throw ApiError.conflict(
          "The page changed since you last read it (a concurrent edit — including a realtime flush). Re-read and retry.",
          [
            { issue: "stale_base_version", received: input.baseVersion ?? page.version },
            { currentVersion: current },
            { reason: "flush_wins" }
          ]
        );
      }
      throw err;
    }
    await this.audit.audit(req, req?.auth, {
      action: "block.batch",
      resourceType: "block",
      resourceId: pageId,
      workspaceId: page.workspaceId,
      pageId,
      metadata: { ops: input.ops.length, version: newVersion }
    });
    await this.events.publish(EVENTS.PAGE_UPDATED, {
      workspaceId: page.workspaceId,
      pageId,
      version: newVersion,
      blockIds: [...touched]
    });
    // T-157: watched-page fan-out (batch is the single block-write path); awaited
    // + guarded (deterministic for tests, never fails the request).
    try { await this.notifications.notifyWatchersForPageUpdate(pageId, userId); } catch { /* best-effort */ }
    return { applied: input.ops.length, blockIds: [...touched], version: newVersion };
  }

  private async subtreeTx(tx: Prisma.TransactionClient, pageId: string, roots: string[]): Promise<string[]> {
    const rows = await tx.block.findMany({
      where: { pageId, deletedAt: null },
      select: { id: true, parentId: true }
    });
    const childrenOf = new Map<string | null, string[]>();
    for (const r of rows) {
      const list = childrenOf.get(r.parentId) ?? [];
      list.push(r.id);
      childrenOf.set(r.parentId, list);
    }
    const ids: string[] = [];
    const queue = [...roots];
    while (queue.length > 0) {
      const id = queue.shift()!;
      ids.push(id);
      queue.push(...(childrenOf.get(id) ?? []));
    }
    return ids;
  }

  // ---------- export ----------

  /** Flat text of a page (search fallback / export). */
  async content(pageId: string, userId: string): Promise<{ pageId: string; text: string; blocks: number }> {
    await this.requirePageAccess(pageId, userId, "VIEW");
    const rows = await this.prisma.block.findMany({
      where: { pageId, deletedAt: null },
      select: { type: true, content: true, position: true },
      orderBy: { position: "asc" },
      take: MAX_BLOCK_TREE
    });
    const lines: string[] = [];
    for (const r of rows) {
      const c = r.content as { text?: string; title?: string };
      const text = c.text ?? c.title ?? "";
      if (text) lines.push(text);
    }
    return { pageId, text: lines.join("\n"), blocks: rows.length };
  }
}

/**
 * The REST batch's `pages.version` CAS guard failed inside the transaction
 * (TASK-0092): a concurrent writer advanced the version between the
 * pre-check and the CAS UPDATE → the transaction aborts (nothing written)
 * and `batch` converts this into the 409 CONFLICT carrying the new version.
 */
class BatchVersionCasConflict extends Error {
  constructor() {
    super("pages.version CAS conflict (concurrent external writer)");
  }
}