import { Inject, Injectable } from "@nestjs/common";
import type { Request } from "express";
import { Prisma, PrismaClient } from "@prisma/client";
import { z } from "zod";
import { optionIdForName, normalizeSelectOptions } from "@factory/shared";
import { PrismaService } from "../../common/prisma.service.js";
import { AuditService } from "../../common/audit.service.js";
import { EventBus } from "../../common/event-bus.js";
import { CacheService } from "../../common/cache.js";
import { ApiError } from "../../common/errors.js";
import { assertCan } from "../../lib/authorize.js";
import { buildRecordFilter, type PropertyFilter } from "../../lib/filter-builder.js";
import { MAX_PRECISION, between, compare, rebalancePositions } from "../../lib/positions.js";
import { parsePagination } from "../../lib/pagination.js";
import { parseCsv } from "../../lib/csv-parser.js";
import { EVENTS, MAX_CSV_IMPORT_ROWS, MAX_EXPORT_ROWS } from "../../constants.js";
import { NotificationsService } from "../notifications/notifications.service.js";
import type { BulkRecordsInput, CreateRecordInput, CsvImportInput, DatabaseQueryInput, ListRecordsInput, UpdateRecordInput } from "./dto.js";
import { PROPERTY_CACHE } from "./property-cache-schema.js";

/** Read-only computed property types — writes are rejected with 422. */
const READ_ONLY_TYPES = new Set([
  "CREATED_TIME", "CREATED_BY", "LAST_EDITED_TIME", "LAST_EDITED_BY",
  "ROLLUP", "FORMULA", "BUTTON", "UNIQUE_ID"
]);

/**
 * Docs §5.3 condition ops → FILTER_OPS (lib/filter-builder.ts) aliases.
 * Conflict recorded per AGENTS.md rule 29: the docs use `is` / `equals` /
 * `greater_than` / `does_not_contain` while the filter builder accepts
 * `eq` / `gte` / `ncontains`. We normalize at the service boundary; the lib
 * is left untouched. `contains_any` / `contains_all` have no SQL
 * fragment equivalent and are evaluated in JS only.
 */
const OP_ALIASES: Record<string, string> = {
  is: "eq",
  is_not: "neq",
  equals: "eq",
  not_equals: "neq",
  greater_than: "gt",
  greater_or_equal: "gte",
  less_than: "lt",
  less_or_equal: "lte",
  does_not_contain: "ncontains",
  starts_with: "startsWith",
  ends_with: "endsWith",
  is_empty: "isEmpty",
  is_not_empty: "isNotEmpty",
  between: "between",
  is_this_week: "isThisWeek",
  is_this_month: "isThisMonth",
  within_next_days: "withinNextDays",
  is_past: "isPast",
  is_future: "isFuture",
  is_current_user: "isCurrentUser",
  on_or_after: "gte",
  on_or_before: "lte"
};
// Ops whose filter value is an envelope that `simplifyValue` would mangle
// before the validator sees it. `between` sends {start, end} but
// simplifyValue unwraps {start} → a bare start string, silently dropping
// `end` (so a bad `end` would slip past validation). These keep the RAW
// envelope so buildValueCondition can validate the full value. All ops are
// still validated — none skip it — and the JS pass (applyDisjunction) is the
// exact evaluator; SQL is only emitted for CHECKBOX eq/neq (RAW_SCALAR_TYPES).
const RAW_ENVELOPE_OPS = new Set(["between"]);

interface PropertyMeta {
  id: string;
  name: string;
  type: string;
  isTitle: boolean;
  /** Stored `config` (Prisma JsonValue) — select/status options carry
   *  `{ id, name, color }`, used to canonicalize legacy name-based
   *  `optionId` values at the read and write boundaries (no-op for ids
   *  that already resolve). */
  config?: unknown;
}

/** One failed CSV import row (T-165). `row` is the 1-based CSV line
 *  (header = line 1, so the first data row is line 2). */
interface CsvImportRowError {
  row: number;
  column: string;
  error: string;
}

interface RowT {
  id: string;
  workspaceId: string;
  databaseId: string;
  position: string;
  pageId: string | null;
  icon: string | null;
  coverUrl: string | null;
  archivedAt: Date | null;
  version: number;
  createdTime: Date;
  lastEditedTime: Date;
  createdAt: Date;
  lastEditedBy: string | null;
  createdBy: string | null;
  page: { title: string; icon: string | null } | null;
  properties: { propertyId: string; value: unknown }[];
}

interface SortClause {
  propertyId: string;
  direction: "asc" | "desc";
}

interface GroupClause {
  propertyId: string;
}

interface FilterLeaf {
  propertyId: string;
  op: string;
  value?: unknown;
}

/**
 * A normalized §5.3 filter node — either a leaf condition
 * (`{ propertyId, op, value? }`) or a branch
 * (`{ op: "and" | "or" | "not", conditions }`). Both producers
 * (`parseFilterTree` for the query param, `asFilters` for stored view
 * filters) return ONE root tree, so `filters` below is a single
 * `FilterNode | null` (the `FilterNode[]` array annotation was a
 * leftover of an earlier flat-list design).
 */
interface FilterNode {
  propertyId?: string;
  op?: string;
  value?: unknown;
  conditions?: FilterNode[];
}

/**
 * Record query engine: filter → sort → paginate over DatabaseRecord rows,
 * joining RecordProperty values by propertyId, and expanding relations.
 * Also record CRUD (record = page, 1:1) + bulk ops + CSV export.
 *
 * Sorting/grouping are applied in JS after a bounded fetch (take ≤ 5000):
 * property values live in one JSONB column (RecordProperty.value), so SQL
 * cannot order by them without a per-type case ladder.
 *
 * Filter trees (docs/api/databases.md §5.3) compile as follows:
 *  - The whole tree is always evaluated in JS over the bounded row set
 *    (≤ 5000 rows) — `nodeMatches`/`leafMatches` are the exact evaluator
 *    (title comes from the page relation; envelope values are unwrapped).
 *  - `and` leaves get an ADDITIONAL SQL pre-filter via buildRecordFilter,
 *    but only when Prisma can express it (raw-scalar value + exact eq/neq —
 *    i.e. CHECKBOX; envelope-typed values live under the JSONB `value`
 *    column and only whole-value equals is valid there). Pushing any other
 *    JSONB operator (string_contains, gte, …) at the top level made Prisma
 *    throw PrismaClientValidationError → 400 "Invalid database query." for
 *    every filtered list (search-in-view, quick filters, saved view filters).
 *
 * Events: record.created / record.updated carry
 * `{ recordId, workspaceId, databaseId, fields }` — the worker re-reads
 * the row from Prisma for search indexing (no denormalized payload).
 * Record deletion is hard (schema cascades); there is no `record.deleted`
 * event in the worker catalog, so deletions emit no event — search docs
 * are dropped by the periodic reconcile (recorded per rule 29).
 */
@Injectable()
export class DatabaseQueryService {
  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
  ) {}

  // ---------- listing ----------

  /**
   * List records: view defaults → filter → bounded fetch → JS sort →
   * cursor pagination (or grouped response when `group` is set).
   */
  async list(databaseId: string, userId: string, query: ListRecordsInput) {
    await this.getDbChecked(databaseId, userId, "VIEW");

    // View defaults (explicit params override — docs §6.1).
    let filters: FilterNode | null = null;
    let sort: SortClause[] | null = null;
    let group: GroupClause[] | null = null;
    let pageSize = parsePagination({ cursor: query.cursor, limit: query.limit }).limit;
    if (query.viewId) {
      const view = await this.prisma.databaseView.findFirst({ where: { id: query.viewId, databaseId } });
      if (!view) throw ApiError.notFound("View not found.");
      filters = this.asFilters(view.filter);
      sort = this.asSort(view.sort);
      group = this.asGroup(view.group);
      if (view.pageSize && view.pageSize > 0) pageSize = Math.min(view.pageSize, 200);
    }
    if (query.filter) {
      const parsed = this.parseFilterTree(query.filter);
      if (parsed) filters = parsed;
    }
    if (query.sort) sort = this.parseSort(query.sort);
    if (query.group) {
      const g = this.parseGroup(query.group);
      group = g.length > 0 ? g : null;
    }
    if (group && group.length === 0) group = null;

    const properties = await this.prisma.databaseProperty.findMany({
      where: { databaseId, deletedAt: null },
      orderBy: { position: "asc" },
      select: { id: true, name: true, type: true, isTitle: true, config: true }
    });
    const byId = new Map(properties.map((p) => [p.id, p]));

    const where = this.compileWhere(filters, byId);
    // TASK-0135 (replica routing): the record LISTING is a spec-named replica
    // read (`db.read()`). Listing staleness (a tiny replica-lag window) is the
    // documented, spec-acceptable trade-off; writes + the create path stay on
    // the primary (the k6 `database-load` create → read-your-own-create path is
    // a separate request, not a same-request write-then-read).
    const rows = await this.prisma.read().databaseRecord.findMany({
      where: { ...where, databaseId, archivedAt: null },
      include: { page: { select: { title: true, icon: true } }, properties: true },
      orderBy: { position: "asc" },
      take: 5000
    }) as RowT[];

    let rowsF = this.applyDisjunction(rows, filters, byId, userId);
    rowsF = this.applySort(rowsF, sort, byId);
    // Relation values live in `RelationLink` (not `DatabaseRecordProperty`);
    // batch-load once for the full filtered set so `toDto` synthesizes
    // `properties[relPropId] = { recordIds }`.
    const relLinkRecords = await this.buildRelationLinkRecords(rowsF, byId);

    // Grouped listing: groups in first-seen (post-sort) order, rows
    // pre-sorted within groups; no cross-group cursor (page within group).
    if (group) {
      const groupProp = group[0]!;
      const groups: { key: { propertyId: string; value: unknown; label: string }; count: number; records: Record<string, unknown>[] }[] = [];
      const index = new Map<string, (typeof groups)[number]>();
      const groupMeta = byId.get(groupProp.propertyId);
      for (const row of rowsF) {
        // Group by the CANONICAL value (legacy name refs → deterministic ids)
        // so the group key + the web's column value both resolve by id.
        const value = this.canonicalOptionValue(
          this.propertyValue(row, byId, groupProp.propertyId),
          groupMeta?.type ?? "TEXT",
          groupMeta?.config
        );
        const gid = `${groupProp.propertyId}::${this.groupKey(value)}`;
        let g = index.get(gid);
        if (!g) {
          g = { key: { propertyId: groupProp.propertyId, value, label: this.groupLabel(groupProp.propertyId, value, byId) }, count: 0, records: [] };
          index.set(gid, g);
          groups.push(g);
        }
        g.records.push(this.toDto(row, byId, undefined, relLinkRecords));
        g.count += 1;
      }
      return { data: { groups }, nextCursor: null };
    }

    const offset = query.cursor ? this.cursorOffset(rowsF, query.cursor) : 0;
    const slice = rowsF.slice(offset, offset + pageSize);
    const more = offset + pageSize < rowsF.length;
    const last = slice.length > 0 ? slice[slice.length - 1] : null;
    const nextCursor = more && last ? this.makeCursor(last.id) : null;
    const relations = query.include === "relations" ? await this.buildRelations(slice) : undefined;

    return {
      data: slice.map((r) => this.toDto(r, byId, relations, relLinkRecords)),
      nextCursor
    };
  }

  /**
   * TASK-0124 (repair): docs §6 query DSL — `POST /databases/:id/query`.
   * Same engine as `list()` (filter → bounded fetch → JS sort → cursor
   * pagination) but takes JSON bodies and returns the §6 envelope the web
   * `useDatabaseRecords` expects: flat `{ data: { records, totalCount,
   * nextCursor } }` or `{ data: { groups, totalCount, nextCursor } }` when
   * grouped. `viewId` seeds filter/sort/group/pageSize from the stored view;
   * explicit body fields override. `select` limits the returned
   * `properties` map to the given property ids (plus the title property).
   * `aggregate` is accepted for contract completeness but returns no
   * `aggregates` object (no client sends it — recorded per rule 29).
   */
  async query(
    databaseId: string,
    userId: string,
    body: DatabaseQueryInput
  ): Promise<Record<string, unknown>> {
    await this.getDbChecked(databaseId, userId, "VIEW");

    let filters: FilterNode | null = null;
    let sort: SortClause[] | null = null;
    let group: GroupClause[] | null = null;
    let pageSize = body.pageSize;
    if (body.viewId) {
      const view = await this.prisma.databaseView.findFirst({ where: { id: body.viewId, databaseId } });
      if (!view) throw ApiError.notFound("View not found.");
      filters = this.asFilters(view.filter);
      sort = this.asSort(view.sort);
      group = this.asGroup(view.group);
      if (view.pageSize && view.pageSize > 0) pageSize = Math.min(view.pageSize, 200);
    }
    if (body.filter !== undefined && body.filter !== null) {
      filters = this.normalizeTree(body.filter);
    }
    if (body.sort && body.sort.length > 0) {
      sort = body.sort.map((s) => ({ propertyId: s.propertyId, direction: s.direction }));
    }
    if (body.group && body.group.length > 0) {
      group = body.group.map((g) => ({ propertyId: g.propertyId }));
    }
    if (group && group.length === 0) group = null;

    const properties = await this.prisma.databaseProperty.findMany({
      where: { databaseId, deletedAt: null },
      orderBy: { position: "asc" },
      select: { id: true, name: true, type: true, isTitle: true, config: true }
    });
    const byId = new Map(properties.map((p) => [p.id, p]));

    // `select` (docs §6): restrict the DTO's properties map; the title
    // property is always kept (row identity) even when not selected.
    const selectSet = body.select && body.select.length > 0 ? new Set(body.select) : null;

    const where = this.compileWhere(filters, byId);
    // TASK-0135 (replica routing): the table-view record QUERY (filter/sort/
    // page over records — the "listing" read) is a spec-named replica read.
    // Same staleness contract as `list()` above.
    const rows = await this.prisma.read().databaseRecord.findMany({
      where: {
        ...where,
        databaseId,
        ...(body.includeArchived ? {} : { archivedAt: null })
      },
      include: { page: { select: { title: true, icon: true } }, properties: true },
      orderBy: { position: "asc" },
      take: 5000
    }) as RowT[];

    let rowsF = this.applyDisjunction(rows, filters, byId, userId);
    rowsF = this.applySort(rowsF, sort, byId);
    const totalCount = rowsF.length;

    // Relation values live in `RelationLink`, not `DatabaseRecordProperty` —
    // batch-load them once for the full filtered set so `toDto` can synthesize
    // `properties[relPropId] = { recordIds }`. (Skipping this makes every
    // relation cell read `null` and created links are invisible in the UI.)
    const relLinkRecords = await this.buildRelationLinkRecords(rowsF, byId);

    const mapRow = (row: RowT) => {
      const dto = this.toDto(row, byId, undefined, relLinkRecords) as Record<
        string,
        unknown
      > & { properties: Record<string, unknown> };
      if (!selectSet) return dto;
      const kept: Record<string, unknown> = {};
      for (const [pid, value] of Object.entries(dto.properties)) {
        const prop = byId.get(pid);
        if (selectSet.has(pid) || prop?.isTitle || prop?.type === "TITLE") kept[pid] = value;
      }
      return { ...dto, properties: kept };
    };

    if (group) {
      const groupProp = group[0]!;
      const groupMeta = byId.get(groupProp.propertyId);
      const groups: { key: { propertyId: string; value: unknown; label: string }; count: number; records: Record<string, unknown>[] }[] = [];
      const index = new Map<string, (typeof groups)[number]>();
      for (const row of rowsF) {
        const value = this.canonicalOptionValue(
          this.propertyValue(row, byId, groupProp.propertyId),
          groupMeta?.type ?? "TEXT",
          groupMeta?.config
        );
        const gid = `${groupProp.propertyId}::${this.groupKey(value)}`;
        let g = index.get(gid);
        if (!g) {
          g = {
            key: { propertyId: groupProp.propertyId, value, label: this.groupLabel(groupProp.propertyId, value, byId) },
            count: 0,
            records: []
          };
          index.set(gid, g);
          groups.push(g);
        }
        g.records.push(mapRow(row));
        g.count += 1;
      }
      return { data: { groups, totalCount, nextCursor: null } };
    }

    const offset = body.cursor ? this.cursorOffset(rowsF, body.cursor) : 0;
    const slice = rowsF.slice(offset, offset + pageSize);
    const more = offset + pageSize < rowsF.length;
    const last = slice.length > 0 ? slice[slice.length - 1] : null;
    const nextCursor = more && last ? this.makeCursor(last.id) : null;

    return { data: { records: slice.map(mapRow), totalCount, nextCursor } };
  }

  // ---------- CRUD ----------

  async createRecord(databaseId: string, userId: string, body: CreateRecordInput, req: Request) {
    const db = await this.getDbChecked(databaseId, userId, "EDIT");
    const properties = await this.prisma.databaseProperty.findMany({
      where: { databaseId, deletedAt: null },
      select: { id: true, name: true, type: true, isTitle: true, config: true }
    });
    const byId = new Map(properties.map((p) => [p.id, p]));
    this.validateWritable(properties, body.properties, body.links);

    const last = await this.prisma.databaseRecord.findFirst({
      where: { databaseId },
      orderBy: { position: "desc" },
      select: { position: true }
    });
    const position = between(last?.position, undefined);
    const title = body.title?.trim() || "Untitled";

    const created = await this.prisma.$transaction(async (tx) => {
      const page = await tx.page.create({
        data: { workspaceId: db.workspaceId, title, position, createdBy: userId, updatedBy: userId }
      });
      const record = await tx.databaseRecord.create({
        data: {
          workspaceId: db.workspaceId,
          databaseId,
          pageId: page.id,
          position,
          createdBy: userId,
          lastEditedBy: userId
        }
      });
      await this.writeProperties(tx, record.id, byId, body.properties);
      await this.writeLinks(tx, db.id, record.id, byId, body.links);
      return { page, record };
    });

    const full = (await this.prisma.databaseRecord.findUniqueOrThrow({
      where: { id: created.record.id },
      include: { page: { select: { title: true, icon: true } }, properties: true }
    })) as RowT;

    await this.audit.audit(req, req?.auth, {
      action: "record.created",
      resourceType: "record",
      resourceId: created.record.id,
      workspaceId: db.workspaceId,
      pageId: created.page.id,
      metadata: { databaseId }
    });
    // TASK-0110: the Phase-5 record event carries `changedProperties`
    // (the field the Phase-11 automation matcher + PROPERTY_CHANGED trigger
    // read). Record creation changes every field, hence `["*"]`.
    await this.events.publish(EVENTS.RECORD_CREATED, {
      workspaceId: db.workspaceId,
      databaseId,
      recordId: created.record.id,
      changedProperties: ["*"]
    });
    return this.toDto(full, byId);
  }

  async getRecord(recordId: string, userId: string) {
    // TASK-0135 query-shape fix (record-open, hot path #5): the row is read
    // ONCE in a NARROW select (no unused relation include) — the two used
    // relations (the page title + the property rows) come in the same
    // round-trip, and the profiled double-fetch (a wide pre-permission
    // `findUnique` + a post-permission `include`) is gone. The permission
    // check still runs BEFORE anything is returned; the read is a plain
    // fetch (no side effect) of data the response never exposes to a
    // non-member (the 403 is thrown from `getDbChecked` before the
    // assembly/return).
    const row = (await this.prisma.databaseRecord.findUnique({
      where: { id: recordId },
      select: {
        id: true,
        workspaceId: true,
        databaseId: true,
        pageId: true,
        icon: true,
        coverUrl: true,
        archivedAt: true,
        version: true,
        position: true,
        createdTime: true,
        lastEditedTime: true,
        createdBy: true,
        lastEditedBy: true,
        createdAt: true,
        page: { select: { title: true, icon: true } },
        properties: { select: { propertyId: true, value: true } }
      }
    })) as RowT | null;
    if (!row || row.archivedAt !== null) throw ApiError.notFound("Record not found.");

    const db = await this.getDbChecked(row.databaseId, userId, "VIEW", row.workspaceId);

    // The column metadata is the shared 60 s cache (busted on
    // `database.updated`) — the SAME key + bust semantics + entry shape as
    // `listProperties`, so the record-open and the table-view property list
    // never disagree (a narrower entry here would corrupt the shared key —
    // zod-on-read would treat it as a miss on every cross read).
    const { data } = await this.cache.getMetadata(
      db.id,
      "properties",
      async () => {
        const properties = await this.prisma.databaseProperty.findMany({
          where: { databaseId: db.id, deletedAt: null },
          orderBy: { position: "asc" }
        });
        // The SAME legacy-id normalization as `listProperties` (and `get`):
        // both loaders write to the ONE shared `…:properties` cache key, so
        // they must store the SAME normalized shape — whichever fills first
        // wins for the 60 s TTL, and a raw (id-less) entry filled by a
        // record-open would make the table view's `findOption` miss every
        // canonical `{ optionId }` value → blank select/status cells.
        const normalized = properties.map((p) =>
          p.type === "SELECT" || p.type === "MULTI_SELECT" || p.type === "STATUS"
            ? { ...p, config: normalizeSelectOptions(p.config) as object }
            : p
        );
        return { data: normalized };
      },
      z.object({ data: z.array(PROPERTY_CACHE) })
    );
    const byId = new Map(
      (data as { data: Array<PropertyMeta & { databaseId: string }> }).data.map((p) => [p.id, p])
    );

    const relations = await this.buildRelations([row]);
    const relLinkRecords = await this.buildRelationLinkRecords([row], byId);
    return this.toDto(row, byId, relations, relLinkRecords);
  }

  async updateRecord(recordId: string, userId: string, body: UpdateRecordInput, req: Request) {
    const record = await this.prisma.databaseRecord.findUnique({ where: { id: recordId } });
    if (!record) throw ApiError.notFound("Record not found.");
    const db = await this.getDbChecked(record.databaseId, userId, "EDIT", record.workspaceId);
    const properties = await this.prisma.databaseProperty.findMany({
      where: { databaseId: db.id, deletedAt: null },
      select: { id: true, name: true, type: true, isTitle: true, config: true }
    });
    const byId = new Map(properties.map((p) => [p.id, p]));
    this.validateWritable(properties, body.properties, body.links);

    // TASK-0094 R4 — the ASSIGNMENT hook: capture the PRE-UPDATE values of
    // any PERSON properties being changed, so the new assignees (new minus
    // old, minus the caller) get an ASSIGNMENT notification after the write.
    const personPropertyIds =
      body.properties != null
        ? Object.entries(body.properties)
            .filter(([pid]) => byId.get(pid)?.type === "PERSON")
            .map(([pid]) => pid)
        : [];
    const oldPersonValues = new Map<string, string[]>();
    if (personPropertyIds.length > 0) {
      const oldRows = await this.prisma.recordProperty.findMany({
        where: { recordId, propertyId: { in: personPropertyIds } },
        select: { propertyId: true, value: true }
      });
      for (const row of oldRows) {
        const v = row.value as { userIds?: unknown } | null;
        oldPersonValues.set(
          row.propertyId,
          Array.isArray(v?.userIds) ? (v!.userIds as string[]) : []
        );
      }
    }

    await this.prisma.$transaction(async (tx) => {
      await tx.databaseRecord.update({
        where: { id: recordId },
        data: {
          ...(body.coverUrl !== undefined ? { coverUrl: body.coverUrl } : {}),
          ...(body.icon !== undefined ? { icon: body.icon } : {}),
          ...(body.archived !== undefined ? { archivedAt: body.archived ? new Date() : null } : {}),
          version: { increment: 1 },
          lastEditedBy: userId
        }
      });
      if (record.pageId && (body.title !== undefined || body.archived !== undefined)) {
        await tx.page.update({
          where: { id: record.pageId },
          data: {
            ...(body.title !== undefined ? { title: body.title } : {}),
            ...(body.archived !== undefined ? { archivedAt: body.archived ? new Date() : null } : {}),
            updatedBy: userId
          }
        });
      }
      if (body.properties) await this.writeProperties(tx, recordId, byId, body.properties);
      if (body.links) await this.writeLinks(tx, db.id, recordId, byId, body.links);
    });

    const full = (await this.prisma.databaseRecord.findUniqueOrThrow({
      where: { id: recordId },
      include: { page: { select: { title: true, icon: true } }, properties: true }
    })) as RowT;

    await this.audit.audit(req, req?.auth, {
      action: "record.updated",
      resourceType: "record",
      resourceId: recordId,
      workspaceId: db.workspaceId,
      pageId: record.pageId ?? undefined,
      metadata: { databaseId: db.id, changedProperties: this.changedFields(body) }
    });
    // TASK-0110: `changedProperties` (not `fields`) — the field the Phase-11
    // automation matcher + PROPERTY_CHANGED trigger read. An event lacking
    // it never fires PROPERTY_CHANGED (no false positives).
    await this.events.publish(EVENTS.RECORD_UPDATED, {
      workspaceId: db.workspaceId,
      databaseId: db.id,
      recordId,
      changedProperties: this.changedFields(body)
    });
    // The ASSIGNMENT notifications (the PERSON-property changes) — after
    // the write + event, best-effort (a notification failure never fails
    // the record update). `db` may be the assert-only stub (workspaceId
    // override path), so fetch the title explicitly for the notification.
    if (body.properties) {
      const dbTitle =
        (await this.prisma.database.findUnique({
          where: { id: record.databaseId },
          select: { title: true }
        }))?.title ?? "a database";
      await this.notifyAssignments(record, dbTitle, body.properties, byId, oldPersonValues, userId);
    }
    // Read-your-own-write: the mutation response must reflect the link that
    // was JUST written, so read the junction from the primary (not the
    // replica — a lagging replica could return a stale relation value).
    const relLinkRecords = await this.buildRelationLinkRecords([full], byId, this.prisma);
    return this.toDto(full, byId, undefined, relLinkRecords);
  }

  /**
   * TASK-0094 R4 — PERSON property gained users → an ASSIGNMENT
   * notification for each NEW assignee (new − old, minus the caller — the
   * self-notification suppression; a user assigning themselves isn't
   * "assigned" in the product sense, recorded). The deep link opens the
   * record's 1:1 page (records are pages — the `record` deep-link type).
   */
  private async notifyAssignments(
    record: { id: string; workspaceId: string; pageId?: string | null },
    databaseTitle: string,
    properties: Record<string, unknown>,
    byId: Map<string, PropertyMeta>,
    oldValues: Map<string, string[]>,
    actorId: string
  ): Promise<void> {
    for (const [propertyId, raw] of Object.entries(properties)) {
      const prop = byId.get(propertyId);
      if (!prop || prop.type !== "PERSON") continue;
      const newIds = new Set(this.asStringArray(raw, ["userIds"], "userIds"));
      const oldIds = new Set(oldValues.get(propertyId) ?? []);
      const assigned = [...newIds].filter((id) => !oldIds.has(id) && id !== actorId);
      if (assigned.length === 0) continue;
      for (const assigneeId of assigned) {
        await this.notifications
          .notify({
            workspaceId: record.workspaceId,
            userId: assigneeId,
            type: "ASSIGNMENT",
            title: `You were assigned on “${databaseTitle}”`,
            body: `${prop.name} → you`,
            // `pageId` ride-along (TASK-0094 R5): records are 1:1 pages, so
            // the web can open the record's page directly.
            link: { type: "record", id: record.id, pageId: record.pageId ?? undefined },
            actorId
          })
          .catch(() => undefined);
      }
    }
  }

  async deleteRecord(recordId: string, userId: string, req: Request): Promise<{ deleted: true }> {
    const record = await this.prisma.databaseRecord.findUnique({
      where: { id: recordId },
      select: { id: true, pageId: true, workspaceId: true, databaseId: true }
    });
    if (!record) throw ApiError.notFound("Record not found.");
    const db = await this.getDbChecked(record.databaseId, userId, "EDIT", record.workspaceId);

    // Hard delete (properties/links cascade per schema); the 1:1 page and
    // its blocks are removed explicitly — mirrors DatabaseService.remove.
    await this.prisma.$transaction(async (tx) => {
      if (record.pageId) {
        await tx.block.deleteMany({ where: { pageId: record.pageId } });
        await tx.page.delete({ where: { id: record.pageId } });
      }
      await tx.databaseRecord.delete({ where: { id: recordId } });
    });

    await this.audit.audit(req, req?.auth, {
      action: "record.deleted",
      resourceType: "record",
      resourceId: recordId,
      workspaceId: db.workspaceId,
      pageId: record.pageId ?? undefined,
      metadata: { databaseId: db.id }
    });
    // No record.deleted event: the worker catalog only indexes
    // record.created / record.updated and re-reads rows from Prisma, so a
    // publish would be a no-op (schema constraint, recorded per rule 29).
    // Search docs are dropped by the periodic reconcile.
    return { deleted: true };
  }

  async bulk(databaseId: string, userId: string, body: BulkRecordsInput, req: Request) {
    const db = await this.getDbChecked(databaseId, userId, "EDIT");
    const targets = await this.prisma.databaseRecord.findMany({
      where: { id: { in: body.recordIds }, databaseId },
      select: { id: true, pageId: true }
    });
    if (targets.length === 0) throw ApiError.notFound("No matching records found.");
    const properties = await this.prisma.databaseProperty.findMany({
      where: { databaseId, deletedAt: null },
      select: { id: true, name: true, type: true, isTitle: true, config: true }
    });
    const byId = new Map(properties.map((p) => [p.id, p]));
    if (body.op === "update") this.validateWritable(properties, body.properties, undefined);

    await this.prisma.$transaction(async (tx) => {
      if (body.op === "update") {
        await this.bulkUpdate(tx, targets, byId, body.properties ?? {}, userId);
      } else {
        const pageIds = targets.map((t) => t.pageId).filter((p): p is string => Boolean(p));
        if (pageIds.length > 0) {
          await tx.block.deleteMany({ where: { pageId: { in: pageIds } } });
          await tx.page.deleteMany({ where: { id: { in: pageIds } } });
        }
        await tx.databaseRecord.deleteMany({ where: { id: { in: targets.map((t) => t.id) } } });
      }
    });

    const changedProperties = body.op === "update" ? this.changedFields(body) : ["*"];
    await this.audit.audit(req, req?.auth, {
      action: body.op === "update" ? "record.bulk_updated" : "record.bulk_deleted",
      resourceType: "record",
      resourceId: targets.map((t) => t.id).join(","),
      workspaceId: db.workspaceId,
      metadata: { databaseId, count: targets.length }
    });
    if (body.op === "update") {
      // Cap stream fan-out: individual events for the first 50 records
      // (the worker re-reads each by id); the audit row covers the tail.
      const MAX_EVENTS = 50;
      for (let i = 0; i < Math.min(targets.length, MAX_EVENTS); i += 1) {
        const id = targets[i]?.id;
        if (!id) continue;
        await this.events.publish(EVENTS.RECORD_UPDATED, { workspaceId: db.workspaceId, databaseId, recordId: id, changedProperties });
      }
    }
    return { affected: targets.length };
  }

  // ---------- CSV export ----------

  async exportCsv(databaseId: string, userId: string, query: ListRecordsInput): Promise<{ csv: string; rows: number; truncated: boolean }> {
    await this.getDbChecked(databaseId, userId, "VIEW");
    const properties = await this.prisma.databaseProperty.findMany({
      where: { databaseId, deletedAt: null },
      orderBy: { position: "asc" },
      select: { id: true, name: true, type: true, isTitle: true, config: true }
    });
    const byId = new Map(properties.map((p) => [p.id, p]));

    // Reuse the listing pipeline for filter/sort parity (group ignored —
    // a CSV is flat).
    let filters: FilterNode | null = null;
    let sort: SortClause[] | null = null;
    if (query.viewId) {
      const view = await this.prisma.databaseView.findFirst({ where: { id: query.viewId, databaseId } });
      if (!view) throw ApiError.notFound("View not found.");
      filters = this.asFilters(view.filter);
      sort = this.asSort(view.sort);
    }
    if (query.filter) {
      const parsed = this.parseFilterTree(query.filter);
      if (parsed) filters = parsed;
    }
    if (query.sort) sort = this.parseSort(query.sort);

    const where = this.compileWhere(filters, byId);
    const rows = (await this.prisma.databaseRecord.findMany({
      where: { ...where, databaseId, archivedAt: null },
      include: { page: { select: { title: true, icon: true } }, properties: true },
      orderBy: { position: "asc" },
      take: MAX_EXPORT_ROWS + 1
    })) as RowT[];
    const rowsF = this.applySort(this.applyDisjunction(rows, filters, byId, userId), sort, byId);
    const truncated = rowsF.length > MAX_EXPORT_ROWS;
    const page = rowsF.slice(0, MAX_EXPORT_ROWS);

    const columns: { label: string; get: (r: RowT) => string }[] = [
      { label: "Title", get: (r) => r.page?.title ?? "" },
      ...properties.filter((p) => !p.isTitle).map((p) => ({ label: p.name, get: (r: RowT) => this.exportValue(r, byId, p.id) }))
    ];
    const lines = [
      columns.map((c) => this.csvCell(c.label)).join(","),
      ...page.map((r) => columns.map((c) => this.csvCell(c.get(r))).join(","))
    ];
    return { csv: lines.join("\n") + "\n", rows: page.length, truncated };
  }

  // ---------- CSV import (T-165) ----------

  /**
   * Import CSV rows as records. `dryRun` computes the header → property
   * mapping and returns a sample without writing; the follow-up call with
   * `dryRun: false` performs a single-transaction import (page + record +
   * property rows per row, same shape as `createRecord`).
   *
   * Caps: the request body is limited by the route-scoped body parser
   * (`MAX_CSV_IMPORT_BODY_BYTES`, see `common/csv-import-body.ts`) and the
   * parsed table by `MAX_CSV_IMPORT_ROWS`. Read-only / relation / title
   * properties are never importable — they are reported unmapped in the
   * dry run, and a mapping that targets one is a 400.
   */
  async importCsv(databaseId: string, userId: string, input: CsvImportInput, req: Request) {
    const db = await this.getDbChecked(databaseId, userId, "EDIT");
    const properties = await this.prisma.databaseProperty.findMany({
      where: { databaseId, deletedAt: null },
      select: { id: true, name: true, type: true, isTitle: true, config: true }
    });
    const byId = new Map(properties.map((p) => [p.id, p]));
    const titleProp = properties.find((p) => p.isTitle || p.type === "TITLE");
    if (!titleProp) throw ApiError.validation("The database has no title property.", [{ issue: "no_title_property" }]);

    const rows = parseCsv(input.csv);
    if (rows.length === 0) {
      throw ApiError.validation("The CSV is empty (a header row is required).", [{ issue: "empty_csv" }]);
    }
    const [headerRow, ...dataRows] = rows;
    const header: string[] = (headerRow ?? []).map((h) => h.trim());

    // Row cap (the body-size cap is enforced upstream by the body parser).
    // A hard 400 — not a silent truncation — because an over-cap file would
    // otherwise drop rows the client expects to be imported.
    if (dataRows.length > MAX_CSV_IMPORT_ROWS) {
      throw ApiError.validation(
        `The CSV has ${dataRows.length} data rows; the limit is ${MAX_CSV_IMPORT_ROWS}.`,
        [{ issue: "too_many_rows", received: dataRows.length, limit: MAX_CSV_IMPORT_ROWS }]
      );
    }
    const limited = dataRows;

    // ---- column mapping (name-based, then explicit overrides) ----
    // Importable = writable value properties. Read-only (FORMULA/ROLLUP/
    // timestamps/…), RELATION (written via links, needs a target record id)
    // and PERSON/FILES (values are user/file ids — no round-trip from a
    // human-readable CSV) are reported unmapped with a reason.
    const importable = properties.filter((p) => !p.isTitle && p.type !== "TITLE" && !READ_ONLY_TYPES.has(p.type) && p.type !== "RELATION" && p.type !== "PERSON" && p.type !== "FILES");
    const notImportableReason = (p: PropertyMeta): string =>
      p.isTitle || p.type === "TITLE" ? "title" : READ_ONLY_TYPES.has(p.type) ? "read_only" : p.type === "RELATION" ? "relation" : p.type === "PERSON" || p.type === "FILES" ? "record_ids_only" : "not_importable";
    const headerIndex = new Map<string, number>();
    header.forEach((h, i) => {
      if (h !== "" && !headerIndex.has(h)) headerIndex.set(h, i);
    });

    // The title column is written via the page title, not a property row.
    const wantedTitle = (input.titleHeader ?? titleProp.name).trim();
    const titleCol = headerIndex.get(wantedTitle);
    if (input.titleHeader !== undefined && titleCol === undefined) {
      throw ApiError.validation(`No column named "${wantedTitle}" for the title.`, [
        { field: "titleHeader", issue: "unknown_header", received: wantedTitle }
      ]);
    }

    // First match wins: a property whose name appears in several headers is
    // claimed by the FIRST one only (deterministic; later headers report
    // `duplicate_header`).
    const columns: { index: number; header: string; propertyId: string | null; reason: string }[] = [];
    const seenProp = new Map<string, number>(); // propertyId → claiming header index
    const claimedPropIds = new Set<string>();
    header.forEach((h, i) => {
      if (i === titleCol) {
        columns.push({ index: i, header: h, propertyId: titleProp.id, reason: "title" });
        seenProp.set(titleProp.id, i);
        return;
      }
      const prop = importable.find((p) => p.name === h);
      if (prop && !claimedPropIds.has(prop.id)) {
        claimedPropIds.add(prop.id);
        seenProp.set(prop.id, i);
        columns.push({ index: i, header: h, propertyId: prop.id, reason: "" });
      } else {
        const reason = h === "" ? "no_header" : prop ? "duplicate_header" : "no_matching_property";
        columns.push({ index: i, header: h, propertyId: null, reason });
      }
    });
    // Explicit overrides: header → propertyId (must be a real, importable property).
    for (const [h, pid] of Object.entries(input.mappings ?? {})) {
      const idx = headerIndex.get(h);
      if (idx === undefined) {
        throw ApiError.validation(`mapping references unknown header "${h}".`, [
          { field: "mappings", issue: "unknown_header", received: h }
        ]);
      }
      const prop = byId.get(pid);
      if (!prop) {
        throw ApiError.validation(`mapping propertyId "${pid}" is not a property of this database.`, [
          { field: "mappings", issue: "unknown_property", received: pid }
        ]);
      }
      if (prop.isTitle || prop.type === "TITLE") {
        throw ApiError.validation(`"${h}" maps to the title property; use titleHeader instead.`, [
          { field: "mappings", issue: "title_via_title_header", received: h }
        ]);
      }
      if (READ_ONLY_TYPES.has(prop.type)) {
        throw ApiError.validation(`"${h}" cannot map to read-only property ${prop.name} (${prop.type}).`, [
          { field: "mappings", issue: "read_only_property", received: h }
        ]);
      }
      if (prop.type === "RELATION") {
        throw ApiError.validation(`"${h}" cannot map to a relation property.`, [
          { field: "mappings", issue: "relation_not_importable", received: h }
        ]);
      }
      if (prop.type === "PERSON" || prop.type === "FILES") {
        throw ApiError.validation(`"${h}" cannot map to ${prop.type} (stored as ids — not importable).`, [
          { field: "mappings", issue: "record_ids_only", received: h }
        ]);
      }
      // Re-point the column: drop the property's previous claim first.
      const prevIdx = seenProp.get(pid);
      if (prevIdx !== undefined && prevIdx !== idx) {
        const prev = columns[prevIdx]!;
        prev.propertyId = null;
        prev.reason = "reassigned";
      }
      const col = columns[idx]!;
      col.propertyId = pid;
      col.reason = "";
      seenProp.set(pid, idx);
    }
    // Drop duplicate claims (a property matched by two headers): keep the
    // first. The title column keeps its claim (handled by `titleCol`).
    for (const c of columns) {
      if (!c.propertyId) continue;
      if (c.index !== seenProp.get(c.propertyId)) {
        c.propertyId = null;
        c.reason = "duplicate_header";
      }
    }
    // Final propertyId → header index (title included — written via page.title).
    const columnOf = new Map<string, number>();
    for (const c of columns) if (c.propertyId) columnOf.set(c.propertyId, c.index);

    if (!input.dryRun && columnOf.size === 0) {
      throw ApiError.validation("No columns map to writable properties — nothing to import.", [
        { issue: "no_importable_columns" }
      ]);
    }

    // ---- dry run: report the mapping + a sample, write nothing ----
    if (input.dryRun) {
      return {
        dryRun: true,
        databaseId,
        rowCount: limited.length,
        columns: columns.map((c) => ({ ...c })),
        unmappedProperties: properties
          .filter((p) => p !== titleProp && !columnOf.has(p.id))
          .map((p) => ({ id: p.id, name: p.name, type: p.type, reason: notImportableReason(p) })),
        sampleRows: limited.slice(0, 5).map((r) => {
          const o: Record<string, string> = {};
          columns.forEach((c) => {
            o[c.header || `col_${c.index}`] = (r as string[])[c.index] ?? "";
          });
          return o;
        })
      };
    }

    // ---- perform the import (single transaction) ----
    const errors: CsvImportRowError[] = [];
    const coerceCell = (prop: PropertyMeta, raw: string): unknown => {
      const text = raw.trim();
      if (text === "") return null;
      switch (prop.type) {
        case "TEXT":
        case "RICH_TEXT":
        case "TITLE":
          return { text, marks: [] };
        case "NUMBER": {
          const n = Number(text);
          if (Number.isNaN(n)) throw ApiError.validation(`not a number: "${text}"`, [{ issue: "bad_number" }]);
          return { value: n };
        }
        case "CHECKBOX": {
          const t = text.toLowerCase();
          if (["true", "1", "yes", "on"].includes(t)) return { value: true };
          if (["false", "0", "no", "off", ""].includes(t)) return { value: false };
          throw ApiError.validation(`not a boolean: "${text}"`, [{ issue: "bad_checkbox" }]);
        }
        case "SELECT":
        case "STATUS": {
          const id = this.resolveOptionRef(text, this.optionsOf(prop.config));
          if (!this.optionExists(id, this.optionsOf(prop.config))) {
            throw ApiError.validation(`unknown option: "${text}"`, [{ issue: "unknown_option", received: text }]);
          }
          return { optionId: id };
        }
        case "MULTI_SELECT": {
          const ids = text.split("|").map((s) => s.trim()).filter(Boolean);
          for (const s of ids) {
            const id = this.resolveOptionRef(s, this.optionsOf(prop.config));
            if (!this.optionExists(id, this.optionsOf(prop.config))) {
              throw ApiError.validation(`unknown option: "${s}"`, [{ issue: "unknown_option", received: s }]);
            }
          }
          return { optionIds: ids.map((s) => this.resolveOptionRef(s, this.optionsOf(prop.config))) };
        }
        case "DATE":
        case "DATE_RANGE": {
          const d = new Date(text);
          if (Number.isNaN(d.getTime())) throw ApiError.validation(`not a date: "${text}"`, [{ issue: "bad_date", received: text }]);
          return { start: d.toISOString(), end: null };
        }
        case "EMAIL":
        case "PHONE":
        case "URL":
          return this.coerceAndCanonical(text, prop);
        default:
          // FILES (and anything unmapped-by-accident): canonical passthrough.
          return this.coerceAndCanonical(text, prop);
      }
    };

    const recordIds: string[] = [];
    await this.prisma.$transaction(async (tx) => {
      const last = await tx.databaseRecord.findFirst({
        where: { databaseId },
        orderBy: { position: "desc" },
        select: { position: true }
      });
      const positions = this.appendPositions(last?.position, limited.length);
      if (positions.length === 0) return;
      if (!positions.every((p) => p.length <= MAX_PRECISION)) {
        // The tail is at (or past) MAX_PRECISION: re-space the whole
        // database to an even grid in-transaction (the house rebalance
        // pattern from block.service) so the new rows append after it.
        const all = await tx.databaseRecord.findMany({
          where: { databaseId },
          orderBy: { position: "asc" },
          select: { id: true, position: true }
        });
        const fresh = rebalancePositions(all.map((r) => r.position).concat(limited.map(() => "A")));
        for (let i = 0; i < all.length; i += 1) {
          const r = all[i]!;
          if (r.position !== fresh[i]) {
            await tx.databaseRecord.update({ where: { id: r.id }, data: { position: fresh[i] } });
          }
        }
        // The imported rows take the grid slots after the existing rows.
        positions.length = 0;
        positions.push(...fresh.slice(all.length));
      }
      for (let i = 0; i < limited.length; i += 1) {
        const r = limited[i] as string[];
        const line = i + 2; // header is line 1
        // Per-cell validation: a bad cell fails THAT row only.
        const propsPatch: Record<string, unknown> = {};
        let rowFailed = false;
        for (const [pid, hIdx] of columnOf) {
          if (pid === titleProp.id) continue; // title handled below
          const prop = byId.get(pid);
          if (!prop) continue;
          const raw = (r as string[])[hIdx] ?? "";
          try {
            propsPatch[pid] = coerceCell(prop, raw);
          } catch (e) {
            errors.push({ row: line, column: prop.name, error: e instanceof Error ? e.message : "invalid value" });
            rowFailed = true;
            break;
          }
        }
        if (rowFailed) continue;

        const titleRaw = titleCol !== undefined ? ((r as string[])[titleCol] ?? "").trim() : "";
        const title = titleRaw || "Untitled";
        const position = positions[i]!;
        const page = await tx.page.create({
          data: { workspaceId: db.workspaceId, title, position, createdBy: userId, updatedBy: userId }
        });
        const record = await tx.databaseRecord.create({
          data: {
            workspaceId: db.workspaceId,
            databaseId,
            pageId: page.id,
            position,
            createdBy: userId,
            lastEditedBy: userId
          }
        });
        await this.writeProperties(tx, record.id, byId, propsPatch);
        recordIds.push(record.id);
      }
    });

    await this.audit.audit(req, req?.auth, {
      action: "database.records_imported",
      resourceType: "database",
      resourceId: databaseId,
      workspaceId: db.workspaceId,
      metadata: { imported: recordIds.length, failed: errors.length, rows: limited.length, databaseId }
    });
    // Cap stream fan-out like `bulk`: individual events for the first 50
    // records (the worker re-reads each by id); the audit row covers the tail.
    const MAX_EVENTS = 50;
    for (let i = 0; i < Math.min(recordIds.length, MAX_EVENTS); i += 1) {
      await this.events.publish(EVENTS.RECORD_CREATED, {
        workspaceId: db.workspaceId,
        databaseId,
        recordId: recordIds[i]!,
        changedProperties: ["*"]
      });
    }

    return {
      dryRun: false,
      databaseId,
      imported: recordIds.length,
      failed: errors.length,
      rows: limited.length,
      errors
    };
  }

  /**
   * Generate N positions that sort AFTER `last` (all records are appended to
   * the end). `rebalancePositions` builds an even grid over `n` cells; we
   * prefix each with `last` so every new position is `> last` and the list
   * stays evenly spaced. (The `a + midpoint` sequential-append of
   * `between(last)` would grow one char per row and exhaust MAX_PRECISION
   * around row 128.)
   */
  private appendPositions(last: string | undefined, n: number): string[] {
    if (n === 0) return [];
    const grid = rebalancePositions(Array.from({ length: n }, () => "A"));
    const base = last ?? "";
    return grid.map((g) => base + g);
  }

  /** True when `ref` resolves to an existing option (id, legacy name, or
   *  deterministic id) — used to reject unknown SELECT/STATUS labels in the
   *  import (per-row error) instead of silently storing the raw label. */
  private optionExists(ref: string, options: unknown[]): boolean {
    if (ref === "") return false;
    for (const o of options) {
      const opt = o as { id?: unknown; name?: unknown } | null;
      const id = this.effectiveOptionId(opt);
      if (!id) continue;
      if (id === ref) return true;
      if (typeof opt?.name === "string" && opt.name === ref) return true;
    }
    return false;
  }

  // ---------- internals ----------

  private async getDbChecked(
    databaseId: string,
    userId: string,
    need: "VIEW" | "EDIT" | "FULL_ACCESS",
    workspaceIdOverride?: string
  ) {
    const db = workspaceIdOverride
      ? { id: databaseId, workspaceId: workspaceIdOverride, isArchived: false }
      : await this.prisma.database.findUnique({ where: { id: databaseId } });
    if (!db) throw ApiError.notFound("Database not found.");
    if (!workspaceIdOverride && db.isArchived) throw ApiError.notFound("Database not found.");
    await assertCan(this.prisma, { workspaceId: db.workspaceId, userId, resource: { databaseId: db.id } }, need);
    return db;
  }

  /** Validate that only writable properties/relations are in the payload. */
  private validateWritable(
    properties: PropertyMeta[],
    propertiesPatch?: Record<string, unknown>,
    links?: Record<string, string[]>
  ): void {
    for (const key of [...(propertiesPatch ? Object.keys(propertiesPatch) : []), ...(links ? Object.keys(links) : [])]) {
      const prop = properties.find((p) => p.id === key);
      if (!prop) {
        throw ApiError.validation(`Unknown property: ${key}.`, [
          { field: "properties", issue: "unknown_property", received: key }
        ]);
      }
      if (READ_ONLY_TYPES.has(prop.type)) {
        throw new ApiError("UNPROCESSABLE_ENTITY", 422, `Property ${prop.name} is read-only and cannot be written.`, [
          { issue: "read_only_property", property: prop.name }
        ]);
      }
      if (prop.isTitle || prop.type === "TITLE") {
        throw ApiError.validation("The title is set via the `title` field, not properties.", [
          { field: "properties", issue: "title_via_title_field", received: key }
        ]);
      }
    }
    if (links) {
      for (const [propId, targetIds] of Object.entries(links)) {
        const prop = properties.find((p) => p.id === propId);
        if (!prop || prop.type !== "RELATION") {
          throw ApiError.validation("links keys must be relation property ids.", [
            { field: "links", issue: "not_a_relation", received: propId }
          ]);
        }
        if (!Array.isArray(targetIds)) {
          throw ApiError.validation("links values must be arrays of record ids.", [
            { field: "links", issue: "array_required", received: propId }
          ]);
        }
      }
    }
  }

  /** Upsert RecordProperty rows for the given patch (title handled by caller). */
  private async writeProperties(
    tx: Prisma.TransactionClient,
    recordId: string,
    byId: Map<string, PropertyMeta>,
    patch: Record<string, unknown>
  ): Promise<void> {
    for (const [propertyId, raw] of Object.entries(patch)) {
      const prop = byId.get(propertyId);
      if (!prop) {
        throw ApiError.validation(`Unknown property: ${propertyId}.`, [
          { field: "properties", issue: "unknown_property", received: propertyId }
        ]);
      }
      // Canonicalize select-family refs (a legacy option NAME → the
      // deterministic id) so NEW rows store the id, not the name.
      const value = this.coerceAndCanonical(raw, prop);
      const json = value === null ? Prisma.JsonNull : (value as Prisma.InputJsonValue);
      await tx.recordProperty.upsert({
        where: { recordId_propertyId: { recordId, propertyId } },
        update: { value: json },
        create: { recordId, propertyId, value: json }
      });
    }
  }

  /**
   * Replace the record's relation links to each target database named by
   * the given relation properties (the target db is resolved via the
   * Relation row that owns the property).
   */
  private async writeLinks(
    tx: Prisma.TransactionClient,
    databaseId: string,
    recordId: string,
    byId: Map<string, PropertyMeta>,
    links: Record<string, string[]>
  ): Promise<void> {
    for (const [propId, targetIds] of Object.entries(links)) {
      const prop = byId.get(propId);
      if (!prop || prop.type !== "RELATION") continue;
      const rel = await tx.relation.findFirst({
        where: { databaseId, propertyId: propId },
        select: { targetDatabaseId: true }
      });
      if (!rel) continue; // relation row missing — nothing to link against
      const targetDatabaseId = rel.targetDatabaseId;
      await tx.relationLink.deleteMany({ where: { recordId, databaseId, targetDatabaseId } });
      const existing = await tx.databaseRecord.findMany({
        where: { id: { in: targetIds }, databaseId: targetDatabaseId },
        select: { id: true }
      });
      const found = new Set(existing.map((r) => r.id));
      for (const targetRecordId of targetIds) {
        if (!found.has(targetRecordId)) {
          throw ApiError.validation(`Link target record not found: ${targetRecordId}.`, [
            { field: "links", issue: "unknown_record", received: targetRecordId }
          ]);
        }
        await tx.relationLink.upsert({
          where: {
            databaseId_recordId_targetDatabaseId_targetRecordId: {
              databaseId,
              recordId,
              targetDatabaseId,
              targetRecordId
            }
          },
          update: {},
          create: { databaseId, recordId, targetDatabaseId, targetRecordId }
        });
      }
    }
  }

  /**
   * Resolve one stored option REF (an `optionId`) to the option's real id, or
   * `null` when the ref is not an option reference. LEGACY records created by
   * the pre-fix AI path stored the option NAME as the ref (`{ optionId:
   * "To Do" }` on options that carried no ids). The property options (after
   * the service-level `normalizeSelectOptions`) are keyed by
   * `optionIdForName(name)`, so a name ref resolves against BOTH the id and
   * the name — modern id refs resolve via the id, name refs via the name,
   * and everything else passes through untouched.
   */
  private optionsOf(config?: unknown): unknown[] {
    const cfg = (config ?? {}) as Record<string, unknown>;
    return Array.isArray(cfg.options) ? (cfg.options as unknown[]) : [];
  }

  /**
   * The option's effective id: its stored `id`, or — for a LEGACY AI option
   * that has none — the deterministic `optionIdForName(name)`. This is the
   * same id the service's `normalizeSelectOptions` assigns, so a legacy name
   * ref resolves to it even before the row's options are back-filled.
   */
  private effectiveOptionId(opt: { id?: unknown; name?: unknown } | null): string | null {
    if (!opt) return null;
    if (typeof opt.id === "string" && opt.id) return opt.id;
    if (typeof opt.name === "string" && opt.name) return optionIdForName(opt.name);
    return null;
  }

  private resolveOptionRef(ref: unknown, options: unknown[]): string {
    if (typeof ref !== "string" || ref.length === 0) return String(ref);
    for (const o of options) {
      const opt = o as { id?: unknown; name?: unknown } | null;
      const id = this.effectiveOptionId(opt);
      if (!id) continue;
      if (id === ref) return id;
      if (typeof opt?.name === "string" && opt.name === ref) return id;
    }
    return ref;
  }

  /**
   * The stored refs that a filter reference (an option id or, legacy, an
   * option NAME) should match: the canonical id PLUS the ref itself, so a
   * filter survives whichever side was written before the id fix (row stored
   * by name vs filter by id, or the inverse). Id refs collapse to the id.
   */
  private optionMatchSet(ref: string, config?: unknown): string[] {
    const resolved = this.resolveOptionRef(ref, this.optionsOf(config));
    return resolved === ref ? [ref] : [resolved, ref];
  }

  /** The display label (option NAME) of a stored select/status/multi-select value. */
  private selectLabelText(value: unknown, type: string, config?: unknown): string | null {
    if (type !== "SELECT" && type !== "STATUS" && type !== "MULTI_SELECT") return this.toText(value, type);
    const options = this.optionsOf(config);
    const nameOf = (ref: unknown): string => {
      if (typeof ref !== "string" || ref === "") return "";
      for (const o of options) {
        const opt = o as { id?: unknown; name?: unknown } | null;
        // Match the stored id, the EFFECTIVE id (a legacy option without an
        // id resolves to optionIdForName(name)), or the name itself.
        if (opt && typeof opt.name === "string" && (opt.id === ref || this.effectiveOptionId(opt) === ref || opt.name === ref)) {
          return opt.name;
        }
      }
      return ref;
    };
    if (type === "MULTI_SELECT") {
      const ids = this.toIdArray(value);
      const names = ids.map(nameOf).filter(Boolean);
      return names.length > 0 ? names.join(", ") : null;
    }
    const ref = (value as { optionId?: unknown } | null)?.optionId;
    if (typeof ref !== "string" || ref === "") return null;
    return nameOf(ref);
  }

  /** Canonicalize a stored value's option ref(s) for a select-family type. */
  private canonicalOptionValue(value: unknown, type: string, config?: unknown): unknown {
    if (value === null || value === undefined || typeof value !== "object") return value;
    const options = this.optionsOf(config);
    const v = value as Record<string, unknown>;
    if (type === "SELECT" || type === "STATUS") {
      if (typeof v.optionId === "string" && v.optionId !== "") {
        const id = this.resolveOptionRef(v.optionId, options);
        return id === v.optionId ? value : { ...v, optionId: id };
      }
      return value;
    }
    if (type === "MULTI_SELECT" && Array.isArray(v.optionIds)) {
      const ids = v.optionIds.map((x) => this.resolveOptionRef(x, options));
      if (ids.every((x, i) => x === (v.optionIds as unknown[])[i])) return value;
      return { ...v, optionIds: ids };
    }
    return value;
  }

  /**
   * Shape a raw API value to the stored per-type JSON value (or null), then
   * canonicalize select-family option refs (a legacy option NAME → the
   * deterministic id) so a write stores the id, not the name. Reads back
   * through `canonicalOptionValue` anyway, but writing the id keeps the DB
   * clean for the next reader (and makes `actualScalar`/sorts stable).
   */
  private coerceAndCanonical(raw: unknown, prop: PropertyMeta): unknown {
    const value = this.coerceValue(raw, prop.type);
    return this.canonicalOptionValue(value, prop.type, prop.config);
  }

  /** Shape a raw API value to the stored per-type JSON value (or null). */
  private coerceValue(raw: unknown, type: string): unknown {
    if (raw === undefined || raw === null) return null;
    switch (type) {
      case "TEXT":
      case "RICH_TEXT":
      case "TITLE": {
        if (typeof raw === "string") return { text: raw, marks: [] };
        if (typeof raw === "object" && raw !== null && "text" in raw) return raw;
        throw ApiError.validation("text value required.", [{ issue: "text_required", type }]);
      }
      case "NUMBER": {
        if (typeof raw === "number") return { value: raw };
        if (typeof raw === "string" && raw !== "" && !Number.isNaN(Number(raw))) return { value: Number(raw) };
        if (typeof raw === "object" && raw !== null && "value" in raw) return raw;
        throw ApiError.validation("number value required.", [{ issue: "number_required", type }]);
      }
      case "SELECT":
      case "STATUS": {
        if (typeof raw === "string") return { optionId: raw };
        if (typeof raw === "object" && raw !== null && "optionId" in raw) return raw;
        throw ApiError.validation("optionId value required.", [{ issue: "option_required", type }]);
      }
      case "MULTI_SELECT": {
        const ids = this.asStringArray(raw, ["optionIds"], "optionIds");
        return { optionIds: ids };
      }
      case "DATE":
      case "DATE_RANGE": {
        if (typeof raw === "string") return { start: raw, end: null };
        if (typeof raw === "object" && raw !== null && "start" in raw) return raw;
        throw ApiError.validation("date value required.", [{ issue: "date_required", type }]);
      }
      case "PERSON": {
        const ids = this.asStringArray(raw, ["userIds"], "userIds");
        return { userIds: ids };
      }
      case "EMAIL":
      case "PHONE": {
        if (typeof raw === "string") return { value: raw };
        if (typeof raw === "object" && raw !== null && "value" in raw) return raw;
        throw ApiError.validation(`${type} value required.`, [{ issue: "value_required", type }]);
      }
      case "URL": {
        if (typeof raw === "string") return { value: raw, title: null };
        if (typeof raw === "object" && raw !== null && "value" in raw) return raw;
        throw ApiError.validation("url value required.", [{ issue: "value_required", type }]);
      }
      case "CHECKBOX": {
        if (typeof raw === "boolean") return { value: raw };
        if (typeof raw === "object" && raw !== null && "value" in raw) return raw;
        throw ApiError.validation("boolean value required.", [{ issue: "boolean_required", type }]);
      }
      case "FILES": {
        const ids = this.asStringArray(raw, ["fileIds"], "fileIds");
        return { fileIds: ids };
      }
      case "RELATION":
        // relations are written via the `links` field, never properties.
        return null;
      default:
        throw new ApiError("UNPROCESSABLE_ENTITY", 422, `Property type ${type} does not accept direct writes.`, [
          { issue: "read_only_property", property: type }
        ]);
    }
  }

  private asStringArray(raw: unknown, objectKeys: string[], label: string): string[] {
    let ids: unknown;
    if (Array.isArray(raw)) ids = raw;
    else if (typeof raw === "string") ids = [raw];
    else if (typeof raw === "object" && raw !== null) {
      ids = null;
      for (const k of objectKeys) {
        const v = (raw as Record<string, unknown>)[k];
        if (Array.isArray(v)) {
          ids = v;
          break;
        }
      }
    }
    if (ids === null || ids === undefined) ids = [];
    if (!Array.isArray(ids) || !ids.every((i) => typeof i === "string")) {
      throw ApiError.validation(`${label} array required.`, [{ issue: `${label}_required`, received: raw }]);
    }
    return ids;
  }

  private changedFields(body: UpdateRecordInput | BulkRecordsInput): string[] {
    const fields: string[] = [];
    if ("title" in body && body.title !== undefined) fields.push("title");
    if ("icon" in body && body.icon !== undefined) fields.push("icon");
    if ("coverUrl" in body && body.coverUrl !== undefined) fields.push("coverUrl");
    if ("archived" in body && body.archived !== undefined) fields.push("archived");
    if (body.properties) fields.push(...Object.keys(body.properties));
    if ("links" in body && body.links) fields.push(...Object.keys(body.links));
    return fields.length > 0 ? fields : ["*"];
  }

  // ---------- filter parsing ----------

  /** Parse a URL-encoded filter tree (§5.3) into a normalized tree. */
  private parseFilterTree(json: string): FilterNode | null {
    let tree: unknown;
    try {
      tree = JSON.parse(json);
    } catch {
      throw ApiError.validation("filter must be valid JSON.", [{ field: "filter", issue: "invalid_json" }]);
    }
    const root = this.normalizeTree(tree);
    const leaves: FilterNode[] = [];
    this.collectLeaves(root, leaves);
    if (leaves.length === 0) return null;
    return root;
  }

  /** Accept a bare leaf, a leaf array, or a §5.3 and/or/not tree. */
  private normalizeTree(tree: unknown): FilterNode {
    if (Array.isArray(tree)) {
      if (tree.length === 0) throw ApiError.validation("filter must not be empty.", [{ field: "filter", issue: "empty" }]);
      return { op: "and", conditions: tree.map((t) => this.normalizeTree(t)) };
    }
    if (typeof tree !== "object" || tree === null) {
      throw ApiError.validation("filter must be an object or array.", [{ field: "filter", issue: "shape" }]);
    }
    const t = tree as Record<string, unknown>;
    if (t.op === "and" || t.op === "or") {
      if (!Array.isArray(t.conditions)) {
        throw ApiError.validation(`${t.op} node requires a conditions array.`, [
          { field: "filter", issue: "conditions_required", received: t.op }
        ]);
      }
      return { op: t.op, conditions: t.conditions.map((c) => this.normalizeTree(c)) };
    }
    if (t.op === "not") {
      if (!Array.isArray(t.conditions) || t.conditions.length !== 1) {
        throw ApiError.validation("not node requires exactly one condition.", [
          { field: "filter", issue: "not_requires_one" }
        ]);
      }
      return { op: "not", conditions: [this.normalizeTree(t.conditions[0])] };
    }
    if (typeof t.propertyId === "string" && typeof t.op === "string") {
      return {
        propertyId: t.propertyId,
        op: this.normalizeOp(t.op),
        ...(t.value !== undefined ? { value: t.value } : {})
      };
    }
    throw ApiError.validation("filter node must be a condition or and/or/not node.", [
      { field: "filter", issue: "unknown_node" }
    ]);
  }

  /** Map docs §5.3 op names onto the filter builder's FILTER_OPS. */
  private normalizeOp(op: string): string {
    return OP_ALIASES[op] ?? op;
  }

  /** Count leaf conditions (any branch) — used to reject empty trees. */
  private collectLeaves(node: FilterNode, out: FilterNode[]): void {
    if (node.propertyId !== undefined) {
      out.push(node);
      return;
    }
    for (const child of node.conditions ?? []) this.collectLeaves(child, out);
  }

  /**
   * Compile the SQL-pre-filterable subset: `and`-reachable leaves, with
   * values unwrapped to the lib's raw shape. `or`/`not` subtrees and
   * JS-only ops are skipped here (re-checked in JS over the row set).
   */
  private compileWhere(root: FilterNode | null, byId: Map<string, PropertyMeta>): Record<string, unknown> {
    if (!root) return {};
    const leaves: FilterLeaf[] = [];
    this.collectAndLeaves(root, leaves);
    if (leaves.length === 0) return {};
    const propertyTypes: Record<string, string> = {};
    for (const leaf of leaves) propertyTypes[leaf.propertyId] = byId.get(leaf.propertyId)?.type ?? "TEXT";
    return buildRecordFilter(leaves as PropertyFilter[], propertyTypes);
  }

  private collectAndLeaves(node: FilterNode, out: FilterLeaf[]): void {
    if (node.propertyId !== undefined) {
      const op = node.op ?? "eq"; // same default as evalLeaf
      // Every leaf is validated by buildValueCondition (unknown op / bad
      // type / bad value → 400 with a message). No op skips validation — a
      // filter that the SQL fragment can't express still has to be a well-
      // formed one. Envelopes that simplifyValue would mangle (between's
      // {start,end}) are passed through raw so the validator sees the full
      // value; everything else is unwrapped to the raw scalar.
      out.push({
        propertyId: node.propertyId,
        op,
        value: RAW_ENVELOPE_OPS.has(op) ? node.value : this.simplifyValue(node.value)
      });
      return;
    }
    if ((node.op ?? "and") !== "and") return; // or/not handled in JS
    for (const child of node.conditions ?? []) this.collectAndLeaves(child, out);
  }

  /**
   * Unwrap docs §5.3 value envelopes ({text}, {value}, {optionId},
   * {start}, …) to the raw shape buildValueCondition expects.
   */
  private simplifyValue(value: unknown): unknown {
    if (typeof value !== "object" || value === null || Array.isArray(value)) return value;
    const v = value as Record<string, unknown>;
    if ("text" in v) return String(v.text);
    if ("optionId" in v) return String(v.optionId);
    if ("start" in v) return v.start as unknown;
    if ("value" in v) return v.value;
    return value;
  }

  /**
   * JS evaluation of the parts SQL cannot express (or/not trees,
   * contains_any/all, and re-validation of the and-part so the bounded
   * row set is exact).
   */
  private applyDisjunction(rows: RowT[], root: FilterNode | null, byId: Map<string, PropertyMeta>, userId: string): RowT[] {
    // `root` is the single normalized §5.3 tree (or null) — NOT an array.
    if (!root) return rows;
    return rows.filter((row) => this.nodeMatches(row, root, byId, userId));
  }

  private nodeMatches(row: RowT, node: FilterNode, byId: Map<string, PropertyMeta>, userId: string): boolean {
    if (node.propertyId !== undefined) {
      const prop = byId.get(node.propertyId);
      // Legacy name-based option refs resolve to the deterministic ids (the
      // toDto side does the same) so a filter on an option NAME written before
      // the id fix still matches records stored with the id (and vice versa).
      const value = this.canonicalOptionValue(
        this.propertyValue(row, byId, node.propertyId),
        prop?.type ?? "TEXT",
        prop?.config
      );
      return this.leafMatches(node, value, prop?.type ?? "TEXT", userId, prop?.config);
    }
    const op = node.op ?? "and";
    const children = node.conditions ?? [];
    if (op === "and") return children.every((c) => this.nodeMatches(row, c, byId, userId));
    if (op === "or") return children.length > 0 && children.some((c) => this.nodeMatches(row, c, byId, userId));
    if (op === "not") return children.length > 0 && !this.nodeMatches(row, children[0]!, byId, userId);
    return true;
  }

  /** JS mirror of buildValueCondition semantics for the docs' op set. */
  private leafMatches(node: FilterNode, value: unknown, type: string, userId: string, config?: unknown): boolean {
    const op = node.op;
    const expected = this.simplifyValue(node.value);
    const options = this.optionsOf(config);
    // A filter written with an option NAME (legacy) must match records stored
    // with the id, and vice versa — canonicalize the expected scalar the same
    // way the row value was canonicalized (nodeMatches).
    const expectedRef =
      (type === "SELECT" || type === "STATUS" || type === "MULTI_SELECT") &&
      (typeof expected === "string" || Array.isArray(expected))
        ? type === "MULTI_SELECT"
          ? (expected as unknown[]).map((x) => this.resolveOptionRef(x, options))
          : this.resolveOptionRef(expected, options)
        : expected;
    switch (op) {
      case "eq": {
        if (this.isEmptyValue(value)) return false;
        return this.scalarEquals(this.actualScalar(value, type), expectedRef);
      }
      case "neq": {
        if (this.isEmptyValue(value)) return false; // lib semantics: neq excludes null
        return !this.scalarEquals(this.actualScalar(value, type), expectedRef);
      }
      case "isEmpty":
        return this.isEmptyValue(value);
      case "isNotEmpty":
        return !this.isEmptyValue(value);
      case "contains":
      case "ncontains": {
        // Select-family values: "contains" on the option NAME(s) (the UI
        // shows names); every other type compares the plain text rendering.
        let mine = this.toText(value, type) ?? "";
        if (type === "SELECT" || type === "STATUS" || type === "MULTI_SELECT") mine = this.selectLabelText(value, type, config) ?? "";
        const needle = typeof expectedRef === "string" ? expectedRef : String(expectedRef ?? "");
        const hit = mine.toLowerCase().includes(needle.toLowerCase());
        return op === "contains" ? hit : !hit;
      }
      case "contains_any":
      case "contains_all": {
        const mine = this.toIdArray(value);
        const raw = Array.isArray(expectedRef) ? (expectedRef as string[]) : typeof expectedRef === "string" ? [expectedRef] : [];
        const target = raw.flatMap((t) => this.optionMatchSet(t, config));
        if (target.length === 0) return op === "contains_any" ? false : true;
        return op === "contains_any"
          ? target.some((t) => mine.includes(t))
          : target.every((t) => mine.includes(t));
      }
      case "gte":
      case "gt":
      case "lte":
      case "lt":
      case "after":
      case "before": {
        if (type === "DATE" || type === "DATE_RANGE") {
          const a = this.toDateMs(value, type);
          const b = typeof expected === "string" || typeof expected === "number" ? new Date(expected).getTime() : NaN;
          if (a === null || Number.isNaN(b)) return false;
          return this.compareByOp(op, a, b);
        }
        const a = this.toNumber(value);
        const b = typeof expected === "number" ? expected : Number(expected);
        if (a === null || Number.isNaN(b)) return false;
        return this.compareByOp(op, a, b);
      }
      case "startsWith":
      case "endsWith": {
        const mine = this.toText(value, type) ?? "";
        const needle = typeof expected === "string" ? expected : String(expected ?? "");
        const hay = mine.toLowerCase();
        const n = needle.toLowerCase();
        if (n.length === 0) return true;
        return op === "startsWith" ? hay.startsWith(n) : hay.endsWith(n);
      }
      case "between": {
        if (type !== "DATE" && type !== "DATE_RANGE") return false;
        const a = this.toDateMs(value, type);
        // The web sends a raw {start, end} envelope; simplifyValue would drop
        // `end`, so read the raw node value for the range bounds.
        const range = (node.value ?? {}) as { start?: unknown; end?: unknown };
        const s = typeof range.start === "string" || typeof range.start === "number" ? new Date(String(range.start)).getTime() : NaN;
        const e = typeof range.end === "string" || typeof range.end === "number" ? new Date(String(range.end)).getTime() : NaN;
        if (a === null || Number.isNaN(s) || Number.isNaN(e)) return false;
        return a >= s && a <= e;
      }
      case "isThisWeek":
      case "isThisMonth":
      case "isPast":
      case "isFuture": {
        if (type !== "DATE" && type !== "DATE_RANGE") return false;
        const a = this.toDateMs(value, type);
        if (a === null) return false;
        const now = Date.now();
        if (op === "isPast") return a < now;
        if (op === "isFuture") return a > now;
        if (op === "isThisMonth") {
          const d = new Date(a);
          const n = new Date(now);
          return d.getFullYear() === n.getFullYear() && d.getMonth() === n.getMonth();
        }
        // isThisWeek — ISO week, Monday-start (matches the UI's mental model).
        const d = new Date(a);
        const day = (d.getDay() + 6) % 7; // Mon=0
        const weekStart = new Date(d);
        weekStart.setHours(0, 0, 0, 0);
        weekStart.setDate(d.getDate() - day);
        return a >= weekStart.getTime() && a < weekStart.getTime() + 7 * 86400000;
      }
      case "withinNextDays": {
        if (type !== "DATE" && type !== "DATE_RANGE") return false;
        const a = this.toDateMs(value, type);
        // The web sends {days: N}; simplifyValue leaves that envelope intact,
        // so unwrap it here (also accept a bare number for the docs' shape).
        const nRaw = (expected as { days?: unknown })?.days ?? expected;
        const n = typeof nRaw === "number" ? nRaw : Number(nRaw);
        if (a === null || Number.isNaN(n)) return false;
        const now = Date.now();
        return a >= now && a <= now + n * 86400000;
      }
      case "isCurrentUser": {
        if (type !== "PERSON") return false;
        const mine = this.toIdArray(value);
        return mine.includes(userId);
      }
      default:
        return true; // unknown ops are rejected at normalize/compile time
    }
  }

  private compareByOp(op: string, a: number, b: number): boolean {
    switch (op) {
      case "gte":
      case "after":
        return a >= b;
      case "gt":
        return a > b;
      case "lte":
      case "before":
        return a <= b;
      default:
        return a < b;
    }
  }

  private scalarEquals(a: unknown, b: unknown): boolean {
    if (a === b) return true;
    if (typeof a === "number" && typeof b === "string") return a === Number(b);
    if (typeof b === "number" && typeof a === "string") return Number(a) === b;
    if (Array.isArray(a) && Array.isArray(b)) {
      return JSON.stringify([...a].sort()) === JSON.stringify([...b].sort());
    }
    return JSON.stringify(a) === JSON.stringify(b);
  }

  /**
   * Reduce a stored JSONB value to the comparison-relevant scalar for its
   * property type (text → string, number → number, select → optionId, …).
   */
  private actualScalar(value: unknown, type: string): unknown {
    if (value === null || value === undefined) return null;
    switch (type) {
      case "TITLE":
      case "TEXT":
      case "RICH_TEXT":
      case "EMAIL":
      case "PHONE":
      case "URL":
        return this.toText(value, type);
      case "NUMBER":
        return this.toNumber(value);
      case "CHECKBOX":
        return typeof value === "boolean" ? value : (value as { value?: boolean })?.value ?? null;
      case "SELECT":
      case "STATUS":
        return (value as { optionId?: string })?.optionId ?? null;
      case "MULTI_SELECT":
      case "PERSON":
      case "FILES":
      case "RELATION":
        return this.toIdArray(value);
      case "DATE":
      case "DATE_RANGE": {
        const start = (value as { start?: string })?.start;
        return typeof start === "string" ? start : null;
      }
      default:
        return value;
    }
  }

  private isEmptyValue(value: unknown): boolean {
    if (value === null || value === undefined) return true;
    if (typeof value === "string") return value === "";
    if (Array.isArray(value)) return value.length === 0;
    if (typeof value === "object") {
      for (const v of Object.values(value as Record<string, unknown>)) {
        if (!this.isEmptyValue(v)) return false;
      }
      return true;
    }
    return false;
  }

  private toIdArray(value: unknown): string[] {
    if (Array.isArray(value)) return value as string[];
    if (typeof value === "object" && value !== null) {
      for (const k of ["optionIds", "userIds", "fileIds", "recordIds", "ids"]) {
        const v = (value as Record<string, unknown>)[k];
        if (Array.isArray(v)) return v as string[];
      }
    }
    return typeof value === "string" ? [value] : [];
  }

  private toNumber(value: unknown): number | null {
    if (typeof value === "number") return value;
    if (typeof value === "string" && value !== "" && !Number.isNaN(Number(value))) return Number(value);
    if (typeof value === "object" && value !== null) {
      const v = (value as Record<string, unknown>).value;
      if (typeof v === "number") return v;
      if (typeof v === "string" && v !== "" && !Number.isNaN(Number(v))) return Number(v);
    }
    return null;
  }

  private toDateMs(value: unknown, type: string): number | null {
    let raw: unknown;
    if (type === "DATE_RANGE" || (typeof value === "object" && value !== null)) {
      raw = (value as { start?: unknown })?.start;
    } else {
      raw = value;
    }
    if (typeof raw === "string" && !Number.isNaN(new Date(raw).getTime())) return new Date(raw).getTime();
    if (typeof raw === "number" && !Number.isNaN(new Date(raw).getTime())) return new Date(raw).getTime();
    return null;
  }

  private toText(value: unknown, type: string, config?: unknown): string | null {
    if (value === null || value === undefined) return null;
    // Select-family values store `{ optionId }` — the plain rendering is the
    // option NAME (without this, CSV export / sort fell through to
    // `String(value)` → "[object Object]").
    if (type === "SELECT" || type === "STATUS" || type === "MULTI_SELECT") {
      return this.selectLabelText(value, type, config);
    }
    if (typeof value === "string") return value;
    if (typeof value === "object") {
      const v = value as Record<string, unknown>;
      if (type === "TITLE" || type === "TEXT" || type === "RICH_TEXT") {
        if (typeof v.text === "string") return v.text;
      }
      if (typeof v.value === "string") return v.value;
    }
    return String(value);
  }

  // ---------- sorting / grouping / values ----------

  private parseSort(json: string): SortClause[] {
    const parsed = this.parseJsonArray(json, "sort");
    return parsed.map((s) => {
      if (typeof s !== "object" || s === null || typeof (s as { propertyId?: unknown }).propertyId !== "string") {
        throw ApiError.validation("sort entries need a propertyId.", [
          { field: "sort", issue: "propertyId_required", received: s }
        ]);
      }
      const dir = (s as { direction?: string; dir?: string }).direction ?? (s as { dir?: string }).dir ?? "asc";
      if (dir !== "asc" && dir !== "desc") {
        throw ApiError.validation("sort direction must be asc or desc.", [{ field: "sort", issue: "direction", received: dir }]);
      }
      return { propertyId: (s as { propertyId: string }).propertyId, direction: dir };
    });
  }

  private parseGroup(json: string): GroupClause[] {
    let parsed: unknown;
    try {
      parsed = JSON.parse(json);
    } catch {
      throw ApiError.validation("group must be valid JSON.", [{ field: "group", issue: "invalid_json" }]);
    }
    const arr = Array.isArray(parsed) ? parsed : [parsed];
    return arr
      .map((g) => {
        if (typeof g !== "object" || g === null || typeof (g as { propertyId?: unknown }).propertyId !== "string") {
          throw ApiError.validation("group entries need a propertyId.", [
            { field: "group", issue: "propertyId_required", received: g }
          ]);
        }
        return { propertyId: (g as { propertyId: string }).propertyId };
      })
      .filter((g) => Boolean(g.propertyId));
  }

  private parseJsonArray(json: string, field: string): unknown[] {
    let parsed: unknown;
    try {
      parsed = JSON.parse(json);
    } catch {
      throw ApiError.validation(`${field} must be valid JSON.`, [{ field, issue: "invalid_json" }]);
    }
    if (!Array.isArray(parsed) || parsed.length === 0) {
      throw ApiError.validation(`${field} must be a non-empty array.`, [{ field, issue: "array_required" }]);
    }
    return parsed;
  }

  private asFilters(json: unknown): FilterNode | null {
    if (!json) return null;
    try {
      return this.normalizeTree(typeof json === "string" ? JSON.parse(json) : json);
    } catch {
      return null;
    }
  }

  private asSort(json: unknown): SortClause[] | null {
    if (!json) return null;
    try {
      const parsed: unknown = typeof json === "string" ? JSON.parse(json) : json;
      if (!Array.isArray(parsed) || parsed.length === 0) return null;
      return (parsed as { propertyId: string; direction?: string }[]).map((s) => ({
        propertyId: s.propertyId,
        direction: s.direction === "desc" ? "desc" : "asc"
      }));
    } catch {
      return null;
    }
  }

  private asGroup(json: unknown): GroupClause[] | null {
    if (!json) return null;
    try {
      const parsed: unknown = typeof json === "string" ? JSON.parse(json) : json;
      const arr = Array.isArray(parsed) ? parsed : [parsed];
      return arr
        .map((g) => ({ propertyId: (g as { propertyId?: string }).propertyId ?? "" }))
        .filter((g) => Boolean(g.propertyId));
    } catch {
      return null;
    }
  }

  private propertyValue(row: RowT, byId: Map<string, PropertyMeta>, propertyId: string): unknown {
    const prop = byId.get(propertyId);
    if (prop?.isTitle || prop?.type === "TITLE") return row.page?.title ?? "";
    const rp = row.properties.find((p) => p.propertyId === propertyId);
    return rp ? rp.value : null;
  }

  /** Sort comparator: missing values always last, direction-aware. */
  private applySort(rows: RowT[], sort: SortClause[] | null, byId: Map<string, PropertyMeta>): RowT[] {
    if (!sort || sort.length === 0) return rows;
    return [...rows].sort((a, b) => {
      for (const clause of sort) {
        const va = this.propertyValue(a, byId, clause.propertyId);
        const vb = this.propertyValue(b, byId, clause.propertyId);
        const ma = this.isEmptyValue(va);
        const mb = this.isEmptyValue(vb);
        let cmp = 0;
        if (ma && mb) cmp = 0;
        else if (ma) cmp = 1; // missing last regardless of direction
        else if (mb) cmp = -1;
        else {
          const na = this.toNumber(va);
          const nb = this.toNumber(vb);
          if (na !== null && nb !== null) {
            cmp = na < nb ? -1 : na > nb ? 1 : 0;
          } else {
            const meta = byId.get(clause.propertyId);
            const type = meta?.type ?? "TEXT";
            const ta = this.toText(va, type, meta?.config) ?? "";
            const tb = this.toText(vb, type, meta?.config) ?? "";
            cmp = ta < tb ? -1 : ta > tb ? 1 : 0;
          }
        }
        if (cmp !== 0) return clause.direction === "desc" ? -cmp : cmp;
      }
      return compare(a.position, b.position);
    });
  }

  private groupKey(value: unknown): string {
    if (value === null || value === undefined) return "__empty__";
    return JSON.stringify(value);
  }

  private groupLabel(propertyId: string, value: unknown, byId: Map<string, PropertyMeta>): string {
    const type = byId.get(propertyId)?.type ?? "TEXT";
    if (this.isEmptyValue(value)) return "(empty)";
    // Select-family values: the option NAME (the id is internal); everything
    // else keeps the plain-text rendering.
    const text =
      type === "SELECT" || type === "STATUS" || type === "MULTI_SELECT"
        ? this.selectLabelText(value, type, byId.get(propertyId)?.config)
        : this.toText(value, type);
    if (text) return text.length > 60 ? text.slice(0, 60) + "…" : text;
    return this.groupKey(value);
  }

  private exportValue(row: RowT, byId: Map<string, PropertyMeta>, propertyId: string): string {
    const type = byId.get(propertyId)?.type ?? "TEXT";
    const value = this.propertyValue(row, byId, propertyId);
    if (this.isEmptyValue(value)) return "";
    if (type === "PERSON" || type === "FILES" || type === "RELATION" || type === "MULTI_SELECT") {
      const ids = this.toIdArray(value);
      if (ids.length > 0) return ids.join(" | ");
    }
    return this.toText(value, type) ?? this.groupKey(value);
  }

  private csvCell(value: string): string {
    if (/[",\n\r]/.test(value)) return `"${value.replace(/"/g, '""')}"`;
    return value;
  }

  // ---------- relations ----------

  /**
   * Expand each row's RelationLinks into `{ [targetDatabaseTitle]: targetRecordTitles[] }`,
   * keyed by the target database's title (RelationLink rows carry the target
   * database/record ids only). Rows are expected to come from one database.
   */
  private async buildRelations(rows: RowT[]): Promise<Map<string, Record<string, string[]>>> {
    const result = new Map<string, Record<string, string[]>>();
    for (const row of rows) result.set(row.id, {});
    const recordIds = rows.map((r) => r.id);
    if (recordIds.length === 0) return result;

    const links = await this.prisma.relationLink.findMany({
      where: { recordId: { in: recordIds } },
      select: { recordId: true, targetRecordId: true, targetDatabaseId: true }
    });
    if (links.length === 0) return result;

    const targetDbIds = [...new Set(links.map((l) => l.targetDatabaseId))];
    const targetDbs = await this.prisma.database.findMany({
      where: { id: { in: targetDbIds } },
      select: { id: true, title: true }
    });
    const dbTitle = new Map(targetDbs.map((d) => [d.id, d.title || "Linked"]));

    const targetRecordIds = [...new Set(links.map((l) => l.targetRecordId))];
    const targets = await this.prisma.databaseRecord.findMany({
      where: { id: { in: targetRecordIds } },
      include: { page: { select: { title: true } } }
    });
    const targetTitle = new Map(targets.map((t) => [t.id, t.page?.title || "(untitled)"]));

    for (const link of links) {
      const name = dbTitle.get(link.targetDatabaseId) ?? "Linked";
      const title = targetTitle.get(link.targetRecordId) ?? "(deleted)";
      const bucket = result.get(link.recordId);
      if (!bucket) continue;
      (bucket[name] ??= []).push(title);
    }
    return result;
  }

  // ---------- DTO ----------

  private toDto(
    row: RowT,
    byId: Map<string, PropertyMeta>,
    relations?: Map<string, Record<string, string[]>>,
    relLinkRecords?: Map<string, Record<string, string[]>>
  ): Record<string, unknown> {
    const properties: Record<string, unknown> = {};
    for (const p of byId.values()) {
      if (p.isTitle || p.type === "TITLE") {
        properties[p.id] = { text: row.page?.title ?? "", marks: [] };
      } else if (p.type === "RELATION") {
        // Relation values are NOT stored in `DatabaseRecordProperty` — they
        // live in the `RelationLink` junction table (written via the `links`
        // field by create/update). Without this synthesis every relation
        // property would read back `null`, so created links were invisible in
        // the table (and to `getRecord` / a post-link update). The link map
        // keys recordId → { propertyId → targetRecordId[] }.
        const links = relLinkRecords?.get(row.id);
        properties[p.id] =
          links && Array.isArray(links[p.id]) && links[p.id].length > 0
            ? { recordIds: links[p.id] }
            : null;
      } else {
        const rp = row.properties.find((x) => x.propertyId === p.id);
        const value = rp ? rp.value : null;
        // Legacy name-based option refs resolve to the deterministic ids the
        // service assigns on read (chips / preselect / board columns then
        // match by id). No-op for modern id refs.
        properties[p.id] = this.canonicalOptionValue(value, p.type, p.config);
      }
    }
    return {
      id: row.id,
      workspaceId: row.workspaceId,
      databaseId: row.databaseId,
      pageId: row.pageId,
      title: row.page?.title ?? "",
      icon: row.icon,
      coverUrl: row.coverUrl,
      position: row.position,
      version: row.version,
      archivedAt: row.archivedAt,
      createdTime: row.createdTime,
      lastEditedTime: row.lastEditedTime,
      createdBy: row.createdBy,
      lastEditedBy: row.lastEditedBy,
      properties,
      ...(relations?.has(row.id) ? { relations: relations.get(row.id) } : {})
    };
  }

  /**
   * Batch-load the relation links for a set of records and map them to their
   * property ids. Relation values live in the `RelationLink` junction table,
   * NOT in `DatabaseRecordProperty` (the `properties` include). Without this
   * synthesis every relation property reads back `null`, so a freshly created
   * relation + link was invisible in the table and in `getRecord`. The
   * `Relation` row is the authoritative propertyId ↔ targetDatabaseId link
   * (the `reversePropertyId` resolves the two-way counterpart, which shares
   * the same junction rows). One indexed query for the links + one for the
   * relations — no per-record N+1.
   */
  private async buildRelationLinkRecords(
    rows: RowT[],
    byId: Map<string, PropertyMeta>,
    client: PrismaClient = this.prisma.read()
  ): Promise<Map<string, Record<string, string[]>>> {
    const result = new Map<string, Record<string, string[]>>();
    const recordIds = rows.map((r) => r.id);
    if (recordIds.length === 0) return result;
    // Only databases that actually have a RELATION property need the query.
    const relationPropIds = [...byId.values()]
      .filter((p) => p.type === "RELATION")
      .map((p) => p.id);
    if (relationPropIds.length === 0) return result;
    const databaseId = rows[0]!.databaseId;
    const links = await client.relationLink.findMany({
      where: { databaseId, recordId: { in: recordIds } },
      select: { recordId: true, targetRecordId: true, targetDatabaseId: true }
    });
    if (links.length === 0) return result;
    const targetDbIds = [...new Set(links.map((l) => l.targetDatabaseId))];
    const relations = await client.relation.findMany({
      where: { databaseId, targetDatabaseId: { in: targetDbIds } },
      select: { propertyId: true, reversePropertyId: true, targetDatabaseId: true }
    });
    const propForTarget = new Map<string, string[]>();
    for (const rel of relations) {
      for (const pid of [rel.propertyId, rel.reversePropertyId]) {
        if (!pid) continue;
        const arr = propForTarget.get(rel.targetDatabaseId);
        if (arr) arr.push(pid);
        else propForTarget.set(rel.targetDatabaseId, [pid]);
      }
    }
    for (const link of links) {
      const propIds = propForTarget.get(link.targetDatabaseId) ?? [];
      if (!result.has(link.recordId)) result.set(link.recordId, {});
      const bucket = result.get(link.recordId)!;
      for (const pid of propIds) {
        const arr = bucket[pid];
        if (arr) arr.push(link.targetRecordId);
        else bucket[pid] = [link.targetRecordId];
      }
    }
    return result;
  }
  // ---------- cursor (row-id based; stable under any sort) ----------

  private makeCursor(recordId: string): string {
    return Buffer.from(`id:${recordId}`, "utf8").toString("base64url");
  }

  /** Index after the cursor row (0 when the row no longer exists). */
  private cursorOffset(rows: RowT[], cursor: string): number {
    let recordId: string | null = null;
    try {
      const decoded = Buffer.from(cursor, "base64url").toString("utf8");
      recordId = decoded.startsWith("id:") ? decoded.slice(3) : null;
    } catch {
      recordId = null;
    }
    if (recordId === null) return 0;
    const idx = rows.findIndex((r) => r.id === recordId);
    return idx === -1 ? 0 : idx + 1;
  }

  // ---------- bulk ----------

  private async bulkUpdate(
    tx: Prisma.TransactionClient,
    targets: { id: string; pageId: string | null }[],
    byId: Map<string, PropertyMeta>,
    properties: Record<string, unknown>,
    userId: string
  ): Promise<void> {
    const patchKeys = Object.keys(properties);
    if (patchKeys.length === 0) {
      await tx.databaseRecord.updateMany({
        where: { id: { in: targets.map((t) => t.id) } },
        data: { version: { increment: 1 }, lastEditedBy: userId }
      });
      return;
    }
    for (const target of targets) {
      for (const propertyId of patchKeys) {
        const prop = byId.get(propertyId);
        if (!prop) continue;
        const value = this.coerceAndCanonical(properties[propertyId], prop);
        const json = value === null ? Prisma.JsonNull : (value as Prisma.InputJsonValue);
        await tx.recordProperty.upsert({
          where: { recordId_propertyId: { recordId: target.id, propertyId } },
          update: { value: json },
          create: { recordId: target.id, propertyId, value: json }
        });
      }
      await tx.databaseRecord.update({
        where: { id: target.id },
        data: { version: { increment: 1 }, lastEditedBy: userId }
      });
      if (target.pageId) {
        await tx.page.update({ where: { id: target.pageId }, data: { updatedBy: userId } });
      }
    }
  }
}