"""Two storage tiers for a datasource (Master) or a form's submissions.

"Flexible" — the default — keeps each record as a JSONB row in
master_records / form_submissions: any shape, any time, no migrations.

"Table" adds a real Postgres table per datasource or form, in the
entity_data schema, generated from its fields:
  * a typed column per field (numeric, date, boolean, uuid, ...);
  * for a datasource: NOT NULL for required fields, a CHECK for a choice's
    options and a text field's max length, a foreign key for a link to
    another datasource that is also a table (a deleted record that's still
    linked to is refused), and an index behind each link;
  * an index behind each field marked Indexed.
A form's table follows its live (published) version and holds its
submitted, not-deleted submissions only — a draft is never in it. A form
can hide a required field, so "required" stays the form's own rule there.

The table is kept in step with the records by triggers (one function per
source, generated here) in the same transaction as the write — so a value
the table can't hold, a missing required value or a broken link refuses
the whole write, whichever path made it (API, import, task). Queries read
the typed columns where they mirror the stored value exactly
(app/crud/record_sql.py); the JSONB row stays the copy everything else
reads — an expand/contract migration.

Every change runs as DDL inside the request's transaction (Postgres DDL is
transactional): promote, the column changes an edit needs (migrate) and
demote either apply completely or not at all, and each is logged with its
statements in entity_migrations. Before any of them touch a column, check()
looks for existing records that wouldn't fit and the change is refused
with those records listed instead.

A large source is switched online (see "Online switching" below): the
request only creates the table and its sync triggers — the one moment the
records table's writes pause — and the existing records are copied in
batches in the background, each its own short transaction, while new writes
reach the table through the triggers. Until the copy is reconciled and
constrained the table is "building" and nothing reads it.
"""

import hashlib
import logging
import re
import uuid
from dataclasses import asdict, dataclass
from dataclasses import field as dc_field
from typing import Any

from sqlalchemy import or_, select, text
from sqlalchemy.exc import DBAPIError
from sqlalchemy.ext.asyncio import AsyncSession

from app.core import jobs
from app.crud import field_indexes
from app.models.entity_storage import EntityMigration, EntityTable
from app.models.master import Master

SCHEMA = "entity_data"
log = logging.getLogger(__name__)

# A source with this many records or more is switched online, in batches.
ONLINE_THRESHOLD = 20_000
BATCH_SIZE = 5_000

# Master field type -> (SQL type, conversion function suffix, what a value must be).
# "strict_" conversions take only the value's own JSON type, so the column
# mirrors the stored value exactly and queries can read it instead of the
# JSON (app/crud/record_sql.py) — the same types the app's own checks
# require (app/crud/validation.py). The rest convert leniently and are read
# from the JSON.
_TYPES: dict[str, tuple[str, str, str]] = {
    "text": ("text", "strict_text", "text"),
    "varchar": ("text", "strict_text", "text"),
    "char": ("text", "strict_text", "text"),
    "number": ("numeric", "strict_numeric", "number"),
    "numeric": ("numeric", "strict_numeric", "number"),
    "smallint": ("smallint", "strict_smallint", "whole number"),
    "integer": ("integer", "strict_integer", "whole number"),
    "bigint": ("bigint", "strict_bigint", "whole number"),
    "real": ("real", "real", "number"),
    "double_precision": ("double precision", "double", "number"),
    "boolean": ("boolean", "strict_boolean", "yes/no value"),
    "date": ("date", "date", "date"),
    "time": ("time", "time", "time"),
    "timestamp": ("timestamp", "timestamp", "date and time"),
    "timestamptz": ("timestamptz", "timestamptz", "date and time"),
    "interval": ("interval", "interval", "duration"),
    "uuid": ("uuid", "uuid", "id"),
    "json": ("jsonb", "jsonb", "JSON value"),
    "jsonb": ("jsonb", "jsonb", "JSON value"),
    "inet": ("inet", "inet", "IP address"),
    "single_select": ("text", "strict_text", "option"),
    "reference": ("uuid", "uuid", "record id"),
    # A formula may give a number or text — jsonb keeps whichever it is.
    "formula": ("jsonb", "jsonb", "value"),
    "autonumber": ("text", "strict_text", "text"),
    "file": ("uuid", "uuid", "file id"),
    "user": ("uuid", "uuid", "user id"),
    "geolocation": ("text", "text", "location"),
}

# Form field type -> the same, after app/crud/form_integrity.py's shaping:
# text-like fields hold strings, number fields numbers, a checkbox a
# boolean, a single choice a string. A cleared number or checkbox can stay
# "" (the integrity check only shapes non-empty values), so those columns
# take "" as empty (ColumnSpec.blank_ok). Lists, documents and line items
# stay jsonb.
_FORM_TEXT = {"text", "textarea", "password", "email", "url", "rich_text", "markdown", "html", "code", "color_picker", "signature", "combobox", "phone", "barcode", "autonumber"}
_FORM_NUMBER = {"number", "currency", "percentage", "slider", "rating"}
_FORM_SINGLE_CHOICE = {"radio", "dropdown", "autocomplete", "lookup"}
_FORM_NO_VALUE = {"section", "panel", "tabs", "accordion", "wizard", "row", "column", "group_end", "button"}
_FORM_TYPES: dict[str, tuple[str, str, str]] = {
    **{t: ("text", "strict_text", "text") for t in _FORM_TEXT},
    **{t: ("numeric", "strict_numeric", "number") for t in _FORM_NUMBER},
    **{t: ("text", "strict_text", "choice") for t in _FORM_SINGLE_CHOICE},
    **{t: ("boolean", "strict_boolean", "yes/no value") for t in ("checkbox", "toggle", "consent")},
    "date": ("date", "date", "date"),
    "time": ("time", "time", "time"),
    "datetime": ("timestamptz", "timestamptz", "date and time"),
    "user": ("uuid", "uuid", "user id"),
    "file_upload": ("text", "text", "file"),
    "image_upload": ("text", "text", "file"),
    "attachment": ("text", "text", "file"),
    "camera": ("text", "text", "file"),
    "geolocation": ("text", "text", "location"),
}
_FORM_BLANK_OK = {"strict_numeric", "strict_boolean"}


def column_type(kind: str, field: dict) -> tuple[str, str, str] | None:
    """(SQL type, conversion, what a value must be) for a field, or None for
    one that holds no value."""
    ftype = field.get("type") or ""
    if kind == "master":
        if ftype == "reference" and field.get("multi"):
            return "uuid[]", "uuid_list", "list of record ids"
        return _TYPES.get(ftype)
    if ftype in _FORM_NO_VALUE:
        return None
    return _FORM_TYPES.get(ftype, ("jsonb", "jsonb", "value"))


class StorageError(Exception):
    """A storage change that can't be made — `status` 409 (conflict with the
    current state) or 422 (records that don't fit, listed in `issues`)."""

    def __init__(self, message: str, status: int = 409, issues: list[dict] | None = None):
        super().__init__(message)
        self.message = message
        self.status = status
        self.issues = issues or []


# ---- the source: a datasource or a form ------------------------------------------------------------


@dataclass
class Source:
    kind: str  # "master" | "form"
    id: uuid.UUID
    name: str
    # A datasource's field schema, or the form's live version's field definitions.
    fields: list[dict]

    @property
    def records(self) -> str:
        return "master_records" if self.kind == "master" else "form_submissions"

    @property
    def key(self) -> str:
        return "master_id" if self.kind == "master" else "form_id"

    @property
    def noun(self) -> str:
        return "datasource" if self.kind == "master" else "form"


def master_source(master: Master) -> Source:
    return Source("master", master.id, master.name, master.field_schema)


async def form_source(db: AsyncSession, tenant_id: uuid.UUID, form_id: uuid.UUID) -> tuple[Source, bool] | None:
    """The form as a storage source — its live version's fields (the draft,
    for a form never published) — and whether that version is published."""
    from app.crud.form_versions import live_form

    live = await live_form(db, tenant_id, form_id)
    if live is None:
        return None
    return Source("form", form_id, live.name, live.field_definitions), live.status == "published"


def _records_of(kind: str) -> str:
    return "master_records" if kind == "master" else "form_submissions"


# Columns every table has, from the record row itself: (column, type, value from the row r).
def _system_columns(kind: str) -> list[tuple[str, str, str]]:
    common = [
        ("id", "uuid PRIMARY KEY", "{r}.id"),
        ("tenant_id", "uuid NOT NULL", "{r}.tenant_id"),
        ("created_at", "timestamptz NOT NULL", "{r}.created_at"),
        ("updated_at", "timestamptz NOT NULL", "{r}.updated_at"),
    ]
    if kind == "master":
        return [*common, ("created_by", "uuid", "{r}.created_by"), ("updated_by", "uuid", "{r}.updated_by")]
    return [
        *common,
        ("created_by", "uuid", "{r}.submitted_by"),
        ("updated_by", "uuid", "{r}.updated_by"),
        ("submitted_at", "timestamptz NOT NULL", "{r}.submitted_at"),
        ("vendor_id", "uuid", "{r}.vendor_id"),
        ("form_version", "integer NOT NULL", "{r}.form_version"),
    ]


def system_column_names(kind: str) -> list[str]:
    return [c for c, _, _ in _system_columns(kind)]


def _live(kind: str, alias: str = "r") -> str:
    """A record that belongs in the table: not deleted (and, for a form, submitted)."""
    if kind == "master":
        return f"NOT {alias}.is_deleted"
    return f"NOT {alias}.is_deleted AND {alias}.status = 'submitted'"


@dataclass
class ColumnSpec:
    field: str
    label: str
    column: str
    sql_type: str
    conv: str
    kind: str
    required: bool = False
    options: list[str] | None = None
    max_length: int | None = None
    # A link to another datasource's records; `fk` when that datasource is a
    # table too (then the link is a real foreign key).
    ref_master_id: str | None = None
    ref_table: str | None = None
    indexed: bool = False
    # "" counts as empty (a form's cleared number or checkbox) — stored as NULL.
    blank_ok: bool = False

    @property
    def fk(self) -> bool:
        return self.ref_table is not None

    @property
    def strict(self) -> bool:
        return self.conv.startswith("strict_")

    @property
    def query_kind(self) -> str | None:
        """How queries read this column in place of the JSON value — see
        record_sql.typed_* — or None to keep reading the JSON. "_blank" =
        the JSON may hold "" where the column holds NULL."""
        base = {"strict_text": "text", "strict_numeric": "numeric", "strict_boolean": "boolean"}.get(
            self.conv, "integer" if self.conv in ("strict_smallint", "strict_integer", "strict_bigint") else None
        )
        return f"{base}_blank" if base and self.blank_ok and base != "text" else base

    def shape(self) -> tuple:
        """What the column itself is — a change here rebuilds the column."""
        return (self.sql_type, self.conv, self.required, tuple(self.options or ()), self.max_length, self.ref_table, self.blank_ok)


# ---- naming ----------------------------------------------------------------------------------


def _q(identifier: str) -> str:
    return '"' + identifier.replace('"', '""') + '"'


def _lit(value: str) -> str:
    return "'" + value.replace("'", "''") + "'"


def _slug(name: str, limit: int) -> str:
    s = re.sub(r"[^a-z0-9]+", "_", name.lower()).strip("_") or "field"
    if s[0].isdigit():
        s = "f_" + s
    return s[:limit].rstrip("_")


def table_name_for(source: Source) -> str:
    return f"{_slug(source.name, 40)}__{source.id.hex[:12]}"


def _hex(source_id: uuid.UUID | str) -> str:
    return uuid.UUID(str(source_id)).hex


def function_name(source_id: uuid.UUID | str) -> str:
    return f"{SCHEMA}.{_q('sync_' + _hex(source_id))}"


def trigger_names(source_id: uuid.UUID | str) -> tuple[str, str]:
    h = _hex(source_id)
    return f"ff_ent_{h}_iu", f"ff_ent_{h}_d"


def qualified(table: str) -> str:
    return f"{SCHEMA}.{_q(table)}"


def _index_name(table: str, column: str) -> str:
    digest = hashlib.sha1(f"{table}.{column}".encode()).hexdigest()[:8]
    return f"ix_{column[:40]}_{digest}"


# ---- the registry --------------------------------------------------------------------------------------


async def tables_by_master(db: AsyncSession, tenant_id: uuid.UUID) -> dict[str, EntityTable]:
    """The datasources stored as tables, by datasource id."""
    rows = (await db.execute(select(EntityTable).where(EntityTable.tenant_id == tenant_id, EntityTable.source_kind == "master"))).scalars().all()
    return {str(r.master_id): r for r in rows}


async def tables_by_source(db: AsyncSession, tenant_id: uuid.UUID) -> dict[str, EntityTable]:
    """Every datasource and form stored as a table, by its id."""
    rows = (await db.execute(select(EntityTable).where(EntityTable.tenant_id == tenant_id))).scalars().all()
    return {str(r.source_id): r for r in rows}


async def entity_table(db: AsyncSession, tenant_id: uuid.UUID, source_id: uuid.UUID) -> EntityTable | None:
    """The table of a datasource or form, if it's stored as one."""
    return (
        await db.execute(
            select(EntityTable).where(EntityTable.tenant_id == tenant_id, or_(EntityTable.master_id == source_id, EntityTable.form_id == source_id))
        )
    ).scalar_one_or_none()


# ---- column specs ---------------------------------------------------------------------------------


def _column_name(name: str, keep: dict[str, str], used: set[str]) -> str:
    column = keep.get(name)
    if column is None:
        base = _slug(name, 55)
        column, i = base, 2
        while column in used:
            column = f"{base}_{i}"
            i += 1
        used.add(column)
    return column


def column_specs(source: Source, tables: dict[str, EntityTable], keep: dict[str, str] | None = None) -> list[ColumnSpec]:
    """The columns a source's fields become. `keep` holds the column names
    already in use (field -> column), so existing columns keep theirs."""
    keep = keep or {}
    used = set(system_column_names(source.kind)) | set(keep.values())
    own_table = tables.get(str(source.id))
    specs: list[ColumnSpec] = []
    for f in source.fields:
        name = f.get("name")
        typed = column_type(source.kind, f) if name else None
        if typed is None:
            continue
        sql_type, conv, kind = typed
        column = _column_name(name, keep, used)
        if source.kind == "form":
            specs.append(
                ColumnSpec(
                    field=name, label=f.get("label") or name, column=column, sql_type=sql_type, conv=conv, kind=kind,
                    indexed=bool(f.get("indexed")) and sql_type != "jsonb", blank_ok=conv in _FORM_BLANK_OK,
                )
            )
            continue
        ftype = f.get("type")
        multi = sql_type == "uuid[]"
        ref_master = str(f["reference_master_id"]) if ftype == "reference" and f.get("reference_master_id") else None
        ref_table = None
        if ref_master and not multi:
            target = own_table if ref_master == str(source.id) else tables.get(ref_master)
            if target is not None and target is not own_table and target.state == "building":
                target = None
            ref_table = target.table_name if target else None
        specs.append(
            ColumnSpec(
                field=name,
                label=f.get("alias") or name,
                column=column,
                sql_type=sql_type,
                conv=conv,
                kind=kind,
                # A formula is computed — it's never "missing" in the table's sense.
                required=bool(f.get("required")) and ftype != "formula",
                options=list(f["options"]) if ftype == "single_select" and f.get("options") else None,
                max_length=f.get("max_length") if ftype in ("varchar", "char") else None,
                ref_master_id=ref_master,
                ref_table=ref_table,
                indexed=bool(f.get("indexed")) and sql_type not in ("jsonb", "uuid[]"),
            )
        )
    return specs


def _specs_from_registry(row: EntityTable) -> list[ColumnSpec]:
    return [ColumnSpec(**c) for c in row.columns.values()]


# ---- SQL pieces --------------------------------------------------------------------------------------


def _converted(spec: ColumnSpec, raw: str) -> str:
    return f"{SCHEMA}.ff_{spec.conv}({raw})" if spec.strict else f"{SCHEMA}.ff_try_{spec.conv}({raw})"


def _need_function(spec: ColumnSpec) -> str:
    if not spec.strict:
        return "ff_need"
    return "ff_need_blank" if spec.blank_ok else "ff_need_strict"


def _value_expr(spec: ColumnSpec, row: str) -> str:
    raw = f"{row}.field_values->{_lit(spec.field)}"
    return f"{SCHEMA}.{_need_function(spec)}({_converted(spec, raw)}, {raw}, {_lit(spec.field)}, {_lit(spec.kind)})"


def _column_constraints(table: str, spec: ColumnSpec) -> list[str]:
    t, c = qualified(table), _q(spec.column)
    out = []
    if spec.required:
        out.append(f"ALTER TABLE {t} ALTER COLUMN {c} SET NOT NULL")
        if spec.conv == "strict_text":
            # Text keeps "" as stored — required means not that either.
            out.append(f"ALTER TABLE {t} ADD CONSTRAINT {_q(spec.column + '__required')} CHECK ({c} <> '')")
    if spec.options:
        allowed = ", ".join(_lit(o) for o in spec.options)
        out.append(f"ALTER TABLE {t} ADD CONSTRAINT {_q(spec.column + '__options')} CHECK ({c} IN ({allowed}))")
    if spec.max_length:
        out.append(f"ALTER TABLE {t} ADD CONSTRAINT {_q(spec.column + '__length')} CHECK (char_length({c}) <= {int(spec.max_length)})")
    if spec.fk:
        out.append(f"ALTER TABLE {t} ADD CONSTRAINT {_q(spec.column + '__link')} FOREIGN KEY ({c}) REFERENCES {qualified(spec.ref_table)}(id)")
    if spec.fk or spec.indexed:
        out.append(f"CREATE INDEX {_q(_index_name(table, spec.column))} ON {t} ({c})")
    return out


def _row_values(kind: str, specs: list[ColumnSpec], row: str) -> tuple[list[str], list[str]]:
    system = _system_columns(kind)
    cols = [c for c, _, _ in system] + [s.column for s in specs]
    values = [expr.format(r=row) for _, _, expr in system] + [_value_expr(s, row) for s in specs]
    return cols, values


def _sync_function(source: Source, table: str, specs: list[ColumnSpec]) -> str:
    t = qualified(table)
    cols, values = _row_values(source.kind, specs, "NEW")
    updated = [c for c in cols if c not in ("id", "tenant_id", "created_at")]
    leaves = "NEW.is_deleted" if source.kind == "master" else "NEW.is_deleted OR NEW.status <> 'submitted'"
    return f"""CREATE OR REPLACE FUNCTION {function_name(source.id)}() RETURNS trigger LANGUAGE plpgsql AS $fn$
BEGIN
  IF TG_OP = 'DELETE' THEN
    DELETE FROM {t} WHERE id = OLD.id;
    RETURN NULL;
  END IF;
  IF {leaves} THEN
    DELETE FROM {t} WHERE id = NEW.id;
    RETURN NULL;
  END IF;
  INSERT INTO {t} ({", ".join(_q(c) for c in cols)})
  VALUES ({", ".join(values)})
  ON CONFLICT (id) DO UPDATE SET ({", ".join(_q(c) for c in updated)}) = ROW({", ".join("EXCLUDED." + _q(c) for c in updated)});
  RETURN NULL;
END $fn$"""


async def _triggers(db: AsyncSession, source: Source) -> list[str]:
    """On the source's own partition, so other sources' writes don't check them."""
    iu, d = trigger_names(source.id)
    m = f"'{uuid.UUID(str(source.id))}'::uuid"
    partition = await field_indexes.record_partition(db, source.kind, source.id)
    return [
        f"CREATE TRIGGER {_q(iu)} AFTER INSERT OR UPDATE ON {partition} FOR EACH ROW WHEN (NEW.{source.key} = {m}) EXECUTE FUNCTION {function_name(source.id)}()",
        f"CREATE TRIGGER {_q(d)} AFTER DELETE ON {partition} FOR EACH ROW WHEN (OLD.{source.key} = {m}) EXECUTE FUNCTION {function_name(source.id)}()",
    ]


async def _drop_sync(db: AsyncSession, source_id: uuid.UUID, kind: str) -> list[str]:
    iu, d = trigger_names(source_id)
    records = await field_indexes.record_partition(db, kind, source_id)
    return [
        f"DROP TRIGGER IF EXISTS {_q(iu)} ON {records}",
        f"DROP TRIGGER IF EXISTS {_q(d)} ON {records}",
        f"DROP FUNCTION IF EXISTS {function_name(source_id)}()",
    ]


def _live_rows(source: Source, tenant_id: uuid.UUID) -> str:
    return f"mr.tenant_id = '{tenant_id}'::uuid AND mr.{source.key} = '{uuid.UUID(str(source.id))}'::uuid AND {_live(source.kind, 'mr')}"


def _backfill(source: Source, tenant_id: uuid.UUID, table: str, specs: list[ColumnSpec]) -> str:
    cols, values = _row_values(source.kind, specs, "mr")
    return (
        f"INSERT INTO {qualified(table)} ({', '.join(_q(c) for c in cols)}) "
        f"SELECT {', '.join(values)} FROM {source.records} mr WHERE {_live_rows(source, tenant_id)}"
    )


# ---- checks ---------------------------------------------------------------------------------------------

_SAMPLE = 5


async def _problem(db: AsyncSession, source: Source, where: str, raw: str | None) -> tuple[int, list[str], list[Any]]:
    sample = f", (array_agg({raw} ORDER BY mr.created_at))[1:{_SAMPLE}]" if raw else ""
    row = (
        await db.execute(text(f"SELECT count(*), (array_agg(mr.id::text ORDER BY mr.created_at))[1:{_SAMPLE}]{sample} FROM {source.records} mr WHERE {where}"))
    ).one()
    return int(row[0]), list(row[1] or []), list(row[2] or []) if raw else []


async def check(db: AsyncSession, tenant_id: uuid.UUID, source: Source, specs: list[ColumnSpec]) -> list[dict]:
    """Records that wouldn't fit these columns: values that don't convert,
    missing required values, values outside a choice's options or over a max
    length, and links to records that don't exist. Empty = they all fit."""
    await _no_time_limit(db)
    issues: list[dict] = []
    live = _live_rows(source, tenant_id)
    for s in specs:
        raw = f"mr.field_values->{_lit(s.field)}"
        present = f"NOT {SCHEMA}.ff_blank({raw})"
        converted = _converted(s, raw)
        # A strict column takes only its own JSON type — "" in a number field
        # is wrong too, unless "" counts as empty there.
        given = (f"NOT {SCHEMA}.ff_absent_or_blank({raw})" if s.blank_ok else f"NOT {SCHEMA}.ff_absent({raw})") if s.strict else present
        checks: list[tuple[str, str, str]] = [
            ("type", f"{given} AND {converted} IS NULL", f"isn't a valid {s.kind}"),
        ]
        if s.required:
            checks.append(("required", f"NOT ({present})", "is empty but required"))
        if s.options:
            allowed = ", ".join(_lit(o) for o in s.options)
            checks.append(("options", f"{present} AND {SCHEMA}.ff_try_text({raw}) NOT IN ({allowed})", "isn't one of the options"))
        if s.max_length:
            checks.append(("length", f"char_length({SCHEMA}.ff_try_text({raw})) > {int(s.max_length)}", f"is longer than {int(s.max_length)} characters"))
        if s.fk:
            checks.append(
                (
                    "link",
                    f"{converted} IS NOT NULL AND NOT EXISTS (SELECT 1 FROM {qualified(s.ref_table)} target WHERE target.id = {converted})",
                    "links to a record that doesn't exist",
                )
            )
        for problem, cond, message in checks:
            count, ids, samples = await _problem(db, source, f"{live} AND {cond}", None if problem == "required" else f"{raw} #>> '{{}}'")
            if count:
                issues.append({"field": s.field, "label": s.label, "problem": problem, "message": f"{s.label} {message}", "count": count, "record_ids": ids, "samples": samples})
    return issues


# ---- applying -----------------------------------------------------------------------------------------------


async def _no_time_limit(db: AsyncSession) -> None:
    """A storage change has to finish, however long its source takes: lifts
    the request's limit (core/database.py) for the rest of this transaction."""
    await db.execute(text("SET LOCAL statement_timeout = 0"))


async def _run(db: AsyncSession, statements: list[str]) -> None:
    await _no_time_limit(db)
    for stmt in statements:
        await db.execute(text(stmt))


async def _run_locking(db: AsyncSession, statements: list[str]) -> None:
    """Statements that create or drop the sync triggers on a records table.
    Another change to that table under way — an index being built, most
    often — would make them wait, or deadlock; a short lock timeout turns
    that into "busy, try again" (the transaction is rolled back either way,
    so nothing is half-done)."""
    await db.execute(text("SET LOCAL lock_timeout = '5s'"))
    try:
        await _run(db, statements)
    except DBAPIError as exc:
        state = getattr(getattr(exc.orig, "__cause__", None), "sqlstate", None)
        if state in ("55P03", "40P01"):  # lock_not_available, deadlock_detected
            raise StorageError("Records are being re-indexed right now — try again in a moment.") from exc
        raise


async def _log(db: AsyncSession, tenant_id: uuid.UUID, source_id: uuid.UUID, user_id: uuid.UUID | None, action: str, summary: str, statements: list[str]) -> None:
    # entity_migrations.master_id holds the source's id — a datasource's or a form's.
    db.add(EntityMigration(tenant_id=tenant_id, master_id=source_id, action=action, summary=summary, statements=statements, applied_by=user_id))
    await db.flush()


def _registry_columns(specs: list[ColumnSpec]) -> dict:
    return {s.field: asdict(s) for s in specs}


def _new_registry_row(source: Source, tenant_id: uuid.UUID, table: str, specs: list[ColumnSpec] | None = None, user_id: uuid.UUID | None = None) -> EntityTable:
    return EntityTable(
        tenant_id=tenant_id,
        source_kind=source.kind,
        master_id=source.id if source.kind == "master" else None,
        form_id=source.id if source.kind == "form" else None,
        table_name=table,
        columns=_registry_columns(specs or []),
        created_by=user_id,
    )


async def promote(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, source: Source) -> EntityTable:
    """Store a flexible datasource or form as a table — refused
    (StorageError 422, with the offending records) if any record wouldn't fit."""
    if await entity_table(db, tenant_id, source.id):
        raise StorageError(f"This {source.noun} is already stored as a table.")
    tables = await tables_by_master(db, tenant_id)
    table = table_name_for(source)
    # A datasource's own links point at itself once it's a table.
    tables[str(source.id)] = _new_registry_row(source, tenant_id, table)
    specs = column_specs(source, tables)
    # A self-link can't be checked against a table that doesn't exist yet;
    # check it against the records themselves instead.
    issues = await check(db, tenant_id, source, [s if s.ref_table != table else _self_link_check(s) for s in specs])
    issues += await _self_link_issues(db, tenant_id, source, [s for s in specs if s.ref_table == table])
    if issues:
        raise StorageError("Some records don't fit the table yet — fix them, then try again.", status=422, issues=issues)

    count = await live_count(db, tenant_id, source)
    if count >= ONLINE_THRESHOLD:
        return await _start_build(db, tenant_id, user_id, source, table, specs, count)
    statements = [
        *_create_statements(source, table, specs),
        # Triggers before the backfill: they wait for writes already under
        # way, so the backfill below sees every record and none is missed.
        *await _triggers(db, source),
        _backfill(source, tenant_id, table, specs),
        *(c for s in specs for c in _column_constraints(table, s)),
        *_finishing_statements(source, table),
    ]
    await _run_locking(db, statements)
    row = _new_registry_row(source, tenant_id, table, specs, user_id)
    db.add(row)
    await db.flush()
    await _log(db, tenant_id, source.id, user_id, "promote", f"Stored as table {SCHEMA}.{table} ({len(specs)} columns)", statements)
    if source.kind == "master":
        await link_referencing(db, tenant_id, user_id, source.id)
    return row


def _create_statements(source: Source, table: str, specs: list[ColumnSpec]) -> list[str]:
    columns = ", ".join([*(f"{c} {t}" for c, t, _ in _system_columns(source.kind)), *(f"{_q(s.column)} {s.sql_type}" for s in specs)])
    return [
        f"CREATE TABLE {qualified(table)} ({columns})",
        f"COMMENT ON TABLE {qualified(table)} IS {_lit(source.noun.capitalize() + ': ' + source.name)}",
        _sync_function(source, table, specs),
    ]


def _finishing_statements(source: Source, table: str) -> list[str]:
    order = "submitted_at" if source.kind == "form" else "created_at"
    return [
        # Insertion order — what every unsorted page of records is read in.
        f"CREATE INDEX {_q(_index_name(table, order))} ON {qualified(table)} ({order}, id)",
        # Statistics now, so the first queries already get good plans.
        f"ANALYZE {qualified(table)}",
    ]


async def live_count(db: AsyncSession, tenant_id: uuid.UUID, source: Source) -> int:
    await _no_time_limit(db)
    return (await db.execute(text(f"SELECT count(*) FROM {source.records} mr WHERE {_live_rows(source, tenant_id)}"))).scalar_one()


# ---- online switching ---------------------------------------------------------------------------------------

BUILD_JOB = "storage.build"


async def _start_build(
    db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, source: Source, table: str, specs: list[ColumnSpec], count: int
) -> EntityTable:
    """The table and its sync triggers, in the request — the records are
    copied by a job (app/core/jobs.py) once it has committed (_continue_build)."""
    statements = [*_create_statements(source, table, specs), *await _triggers(db, source)]
    await _run_locking(db, statements)
    row = _new_registry_row(source, tenant_id, table, specs, user_id)
    row.state = "building"
    db.add(row)
    await db.flush()
    await _log(db, tenant_id, source.id, user_id, "promote", f"Building table {SCHEMA}.{table} — copying {count:,} records in the background", statements)
    await _schedule(db, tenant_id, source.id)
    return row


async def _schedule(db: AsyncSession, tenant_id: uuid.UUID, source_id: uuid.UUID) -> None:
    await jobs.enqueue(db, BUILD_JOB, {"source_id": str(source_id)}, tenant_id=tenant_id, dedupe_key=f"{BUILD_JOB}:{source_id}")


async def wait_for_builds() -> None:
    """Runs the queued table builds now, in this process (tests, scripts)."""
    await jobs.run_due([BUILD_JOB])


async def _source_of(db: AsyncSession, tenant_id: uuid.UUID, row: EntityTable) -> Source | None:
    if row.source_kind == "master":
        master = await db.get(Master, row.master_id)
        return master_source(master) if master else None
    found = await form_source(db, tenant_id, row.form_id)
    return found[0] if found else None


async def _copy_batch(db: AsyncSession, tenant_id: uuid.UUID, source: Source, table: str, specs: list[ColumnSpec], last: uuid.UUID | None) -> tuple[int, uuid.UUID | None]:
    """The next BATCH_SIZE records by id. One the triggers already wrote is
    newer than the copy, so it's left as it is."""
    after = f" AND mr.id > '{last}'::uuid" if last else ""
    ids = [r[0] for r in await db.execute(text(f"SELECT mr.id FROM {source.records} mr WHERE {_live_rows(source, tenant_id)}{after} ORDER BY mr.id LIMIT {BATCH_SIZE}"))]
    if not ids:
        return 0, last
    cols, values = _row_values(source.kind, specs, "mr")
    await db.execute(
        text(
            f"INSERT INTO {qualified(table)} ({', '.join(_q(c) for c in cols)}) SELECT {', '.join(values)} FROM {source.records} mr "
            f"WHERE mr.id = ANY(:ids) AND mr.{source.key} = '{uuid.UUID(str(source.id))}'::uuid AND {_live(source.kind, 'mr')} ON CONFLICT (id) DO NOTHING"
        ),
        {"ids": ids},
    )
    return len(ids), ids[-1]


async def _finish_build(db: AsyncSession, tenant_id: uuid.UUID, source: Source) -> None:
    row = await entity_table(db, tenant_id, source.id)
    if row is None or row.state != "building":
        return
    t = qualified(row.table_name)
    statements = [
        # Anything deleted — or, for a form, back to a draft — while the copy ran.
        f"DELETE FROM {t} e WHERE NOT EXISTS (SELECT 1 FROM {source.records} mr WHERE mr.{source.key} = '{uuid.UUID(str(source.id))}'::uuid AND mr.id = e.id AND {_live(source.kind, 'mr')})",
        *(c for s in _specs_from_registry(row) for c in _column_constraints(row.table_name, s)),
        *_finishing_statements(source, row.table_name),
    ]
    await _run(db, statements)
    row.state = "ready"
    await db.flush()
    rows = (await db.execute(text(f"SELECT count(*) FROM {t}"))).scalar_one()
    await _log(db, tenant_id, source.id, row.created_by, "promote", f"Table {SCHEMA}.{row.table_name} ready — {rows:,} records copied", statements)
    if source.kind == "master":
        await link_referencing(db, tenant_id, row.created_by, source.id)


def _sqlstate(exc: BaseException) -> str | None:
    return getattr(getattr(getattr(exc, "orig", None), "__cause__", None), "sqlstate", None)


def _is_data_problem(exc: BaseException) -> bool:
    """A record the table can't hold or a constraint it breaks — copying
    again won't change that (a dropped connection or a lock would)."""
    state = _sqlstate(exc) or ""
    return state.startswith(("22", "23"))


async def _continue_build(tenant_id: uuid.UUID, source_id: uuid.UUID) -> str:
    """Copies a building table's records in batches, then finishes it. A
    data problem undoes the build at once (jobs.Permanent); anything else is
    retried by the job queue, and undone only once it has failed for good."""
    async with jobs.tenant_session(tenant_id) as db:
        row = await entity_table(db, tenant_id, source_id)
        if row is None or row.state != "building":
            return "nothing to build"
        source = await _source_of(db, tenant_id, row)
    if source is None:
        return "source is gone"
    specs = _specs_from_registry(row)
    last: uuid.UUID | None = None
    try:
        while True:
            async with jobs.tenant_session(tenant_id) as db:
                current = await entity_table(db, tenant_id, source_id)
                if current is None or current.state != "building":
                    return "switched back to flexible storage meanwhile"
                copied, last = await _copy_batch(db, tenant_id, source, row.table_name, specs, last)
            if copied < BATCH_SIZE:
                break
        async with jobs.tenant_session(tenant_id) as db:
            await _finish_build(db, tenant_id, source)
    except Exception as exc:
        if not _is_data_problem(exc):
            raise
        await _fail_build(tenant_id, source_id, exc)
        raise jobs.Permanent(_failure_text(exc)) from exc
    return f"{SCHEMA}.{row.table_name} ready"


@jobs.handler(BUILD_JOB, max_attempts=3)
async def _build_job(job: jobs.Job) -> str:
    return await _continue_build(job.tenant_id, uuid.UUID(job.payload["source_id"]))


@jobs.on_failure(BUILD_JOB)
async def _build_failed(job: jobs.Job, exc: BaseException) -> None:
    await _fail_build(job.tenant_id, uuid.UUID(job.payload["source_id"]), exc.__cause__ or exc)


def _failure_text(exc: BaseException) -> str:
    pg = getattr(getattr(exc, "orig", None), "__cause__", None) or exc
    message = str(getattr(pg, "message", None) or pg)
    if message.startswith("FF:type:"):
        _, _, field, kind = message.split(":", 3)
        return f"a record's {field} isn't a valid {kind}"
    return message.splitlines()[0][:300]


async def _fail_build(tenant_id: uuid.UUID, source_id: uuid.UUID, exc: BaseException) -> None:
    """A record written during the copy that the table can't hold (or a
    constraint it breaks) undoes the build — back to flexible, and why."""
    async with jobs.tenant_session(tenant_id) as db:
        row = await entity_table(db, tenant_id, source_id)
        if row is None:
            return
        statements = [*await _drop_sync(db, source_id, row.source_kind), f"DROP TABLE IF EXISTS {qualified(row.table_name)}"]
        await _run(db, statements)
        await db.delete(row)
        await db.flush()
        await _log(db, tenant_id, source_id, row.created_by, "demote", f"Building the table failed and was undone: {_failure_text(exc)}", statements)


def _self_link_check(spec: ColumnSpec) -> ColumnSpec:
    return ColumnSpec(**{**asdict(spec), "ref_table": None})


async def _self_link_issues(db: AsyncSession, tenant_id: uuid.UUID, source: Source, specs: list[ColumnSpec]) -> list[dict]:
    issues = []
    live = _live_rows(source, tenant_id)
    for s in specs:
        raw = f"mr.field_values->{_lit(s.field)}"
        converted = f"{SCHEMA}.ff_try_uuid({raw})"
        cond = (
            f"{live} AND {converted} IS NOT NULL AND NOT EXISTS (SELECT 1 FROM master_records t WHERE t.id = {converted} "
            f"AND t.master_id = mr.master_id AND NOT t.is_deleted)"
        )
        count, ids, samples = await _problem(db, source, cond, f"{raw} #>> '{{}}'")
        if count:
            issues.append({"field": s.field, "label": s.label, "problem": "link", "message": f"{s.label} links to a record that doesn't exist", "count": count, "record_ids": ids, "samples": samples})
    return issues


async def migrate(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, source: Source) -> list[str]:
    """Bring a table's columns in line with its source's (just edited or
    just published) fields: add, rebuild or drop the columns that changed.
    Refused (StorageError 422) if existing records don't fit a new or
    changed column — the caller's transaction then rolls the edit back too."""
    row = await entity_table(db, tenant_id, source.id)
    if row is None:
        return []
    if row.state == "building":
        raise StorageError(f"This {source.noun}'s table is still being built — try again once it's ready.")
    tables = await tables_by_master(db, tenant_id)
    old = {s.field: s for s in _specs_from_registry(row)}
    new = column_specs(source, tables, keep={f: s.column for f, s in old.items()})
    # A link whose target has since become a table gains a foreign key only
    # if its records allow it — an edit to something else isn't refused
    # because of it (see link_referencing).
    for s in new:
        before = old.get(s.field)
        if s.fk and before and not before.fk and before.shape()[:5] == s.shape()[:5] and await check(db, tenant_id, source, [s]):
            s.ref_table = None
    new_by_field = {s.field: s for s in new}
    table = row.table_name
    t = qualified(table)

    removed = [s for f, s in old.items() if f not in new_by_field]
    added = [s for s in new if s.field not in old]
    rebuilt = [s for s in new if s.field in old and s.shape() != old[s.field].shape()]
    # A column whose type changes comes back under a new name: a statement a
    # connection has already prepared over the old column then never meets
    # a different result type under the same name.
    used = set(system_column_names(source.kind)) | {s.column for s in new} | {s.column for s in old.values()}
    for s in rebuilt:
        if s.sql_type != old[s.field].sql_type:
            s.column = _column_name(s.field + "_v", {}, used)
    # A link's column is always indexed for its foreign key; "Indexed" only matters for the rest.
    reindexed = [s for s in new if s.field in old and s.shape() == old[s.field].shape() and s.indexed != old[s.field].indexed and not s.fk]

    issues = await check(db, tenant_id, source, added + rebuilt)
    if issues:
        raise StorageError("Some records don't fit the changed fields — fix them, then try again.", status=422, issues=issues)

    statements: list[str] = []
    for s in removed:
        statements.append(f"ALTER TABLE {t} DROP COLUMN {_q(s.column)}")
    for s in rebuilt:
        statements.append(f"ALTER TABLE {t} DROP COLUMN {_q(old[s.field].column)}")
    for s in added + rebuilt:
        statements.append(f"ALTER TABLE {t} ADD COLUMN {_q(s.column)} {s.sql_type}")
        statements.append(f"UPDATE {t} AS e SET {_q(s.column)} = {_value_expr(s, 'mr')} FROM {source.records} mr WHERE mr.{source.key} = '{uuid.UUID(str(source.id))}'::uuid AND mr.id = e.id")
        statements.extend(_column_constraints(table, s))
    for s in reindexed:
        index = _q(_index_name(table, s.column))
        statements.append(f"CREATE INDEX {index} ON {t} ({_q(s.column)})" if s.indexed else f"DROP INDEX IF EXISTS {SCHEMA}.{index}")

    if removed or added or rebuilt:
        statements.append(_sync_function(source, table, new))
    if not statements:
        # Nothing structural — but a renamed label still updates the registry.
        if _registry_columns(new) != row.columns:
            row.columns = _registry_columns(new)
            await db.flush()
        return []
    await _run(db, statements)
    row.columns = _registry_columns(new)
    await db.flush()
    parts = [f"{len(added)} added" if added else "", f"{len(rebuilt)} changed" if rebuilt else "", f"{len(removed)} removed" if removed else "", f"{len(reindexed)} re-indexed" if reindexed else ""]
    await _log(db, tenant_id, source.id, user_id, "migrate", "Columns: " + ", ".join(p for p in parts if p), statements)
    return statements


async def sync_form(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, form_id: uuid.UUID) -> None:
    """A form stored as a table follows its live version — after a publish,
    or a change made in place to the live version (fields added, indexes)."""
    if await entity_table(db, tenant_id, form_id) is None:
        return
    found = await form_source(db, tenant_id, form_id)
    if found is not None:
        await migrate(db, tenant_id, user_id, found[0])


async def link_referencing(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, target_master_id: uuid.UUID) -> None:
    """Now that a datasource is a table, links to it from other table
    datasources become foreign keys — where their records allow it (a link
    to a missing record leaves that one link unenforced until it's fixed
    and the datasource is saved again)."""
    tables = await tables_by_master(db, tenant_id)
    for master_id, row in tables.items():
        if master_id == str(target_master_id):
            continue
        olds = _specs_from_registry(row)
        old_by_field = {o.field: o for o in olds}
        if not any(s.ref_master_id == str(target_master_id) and not s.fk and s.sql_type == "uuid" for s in olds):
            continue
        master = await db.get(Master, uuid.UUID(master_id))
        if master is None:
            continue
        source = master_source(master)
        candidates = column_specs(source, tables, keep={s.field: s.column for s in olds})
        linkable = [s for s in candidates if s.ref_master_id == str(target_master_id) and s.fk and s.field in old_by_field and not old_by_field[s.field].fk]
        clean = [s for s in linkable if not await check(db, tenant_id, source, [s])]
        if not clean:
            continue
        t = qualified(row.table_name)
        statements = [
            f"ALTER TABLE {t} ADD CONSTRAINT {_q(s.column + '__link')} FOREIGN KEY ({_q(s.column)}) REFERENCES {qualified(s.ref_table)}(id)" for s in clean
        ] + [f"CREATE INDEX IF NOT EXISTS {_q(_index_name(row.table_name, s.column))} ON {t} ({_q(s.column)})" for s in clean]
        await _run(db, statements)
        by_field = {s.field: s for s in clean}
        row.columns = _registry_columns([by_field.get(o.field, o) for o in olds])
        await db.flush()
        await _log(db, tenant_id, master.id, user_id, "migrate", f"Links to {len(clean)} field(s) now enforced", statements)


async def referencing_tables(db: AsyncSession, table: str) -> list[str]:
    """Other entity tables with a foreign key into this one."""
    rows = await db.execute(
        text(
            "SELECT DISTINCT src.relname FROM pg_constraint c "
            "JOIN pg_class src ON src.oid = c.conrelid JOIN pg_class dst ON dst.oid = c.confrelid "
            "JOIN pg_namespace n ON n.oid = dst.relnamespace "
            "WHERE c.contype = 'f' AND n.nspname = :schema AND dst.relname = :table AND src.oid <> dst.oid"
        ),
        {"schema": SCHEMA, "table": table},
    )
    return [r[0] for r in rows]


async def demote(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, source_id: uuid.UUID, reason: str = "Back to flexible storage") -> None:
    """Drop the table (the JSONB copy holds every record, so nothing is
    lost) — refused while another table datasource's links point into it."""
    row = await entity_table(db, tenant_id, source_id)
    if row is None:
        raise StorageError("This isn't stored as a table.")
    inbound = await referencing_tables(db, row.table_name)
    if inbound:
        names = await _master_names_for_tables(db, tenant_id, inbound)
        raise StorageError(f"Links from {', '.join(names)} depend on this table — switch those to flexible storage first.")
    statements = [*await _drop_sync(db, source_id, row.source_kind), f"DROP TABLE IF EXISTS {qualified(row.table_name)}"]
    await _run_locking(db, statements)
    await db.delete(row)
    await db.flush()
    await _log(db, tenant_id, source_id, user_id, "demote", reason, statements)


async def _master_names_for_tables(db: AsyncSession, tenant_id: uuid.UUID, table_names: list[str]) -> list[str]:
    rows = await db.execute(
        select(Master.name).join(EntityTable, EntityTable.master_id == Master.id).where(EntityTable.tenant_id == tenant_id, EntityTable.table_name.in_(table_names))
    )
    return [r[0] for r in rows] or table_names


async def drop_tenant_storage(db: AsyncSession, tenant_id: uuid.UUID) -> None:
    await field_indexes.drop_tenant_indexes(db, tenant_id)
    """Every table of a tenant that's going away (tests, tenant deletion)."""
    rows = (await db.execute(select(EntityTable).where(EntityTable.tenant_id == tenant_id))).scalars().all()
    for row in rows:
        await _run(db, [*await _drop_sync(db, row.source_id, row.source_kind), f"DROP TABLE IF EXISTS {qualified(row.table_name)} CASCADE"])
        await db.delete(row)
    await db.flush()


# ---- describing ---------------------------------------------------------------------------------------------


def _describe_column(s: ColumnSpec) -> dict:
    return {
        "field": s.field, "label": s.label, "column": s.column, "sql_type": s.sql_type, "required": s.required,
        "options": s.options, "max_length": s.max_length, "is_link": bool(s.ref_master_id),
        "link_table": f"{SCHEMA}.{s.ref_table}" if s.ref_table else None, "link_enforced": s.fk, "indexed": s.indexed or s.fk,
    }


def _in_field_order(specs: list[ColumnSpec], fields: list[dict]) -> list[ColumnSpec]:
    """The registry is JSONB, which doesn't keep key order — show the
    columns in the source's own field order."""
    order = {f.get("name"): i for i, f in enumerate(fields)}
    return sorted(specs, key=lambda s: order.get(s.field, len(order)))


async def describe(db: AsyncSession, tenant_id: uuid.UUID, source: Source) -> dict:
    """The source's storage: its tier, and as a table its name, columns
    (with which constraints apply), row count and migration history."""
    row = await entity_table(db, tenant_id, source.id)
    history = (
        await db.execute(
            select(EntityMigration).where(EntityMigration.tenant_id == tenant_id, EntityMigration.master_id == source.id).order_by(EntityMigration.applied_at.desc()).limit(50)
        )
    ).scalars().all()
    out: dict[str, Any] = {
        "tier": ("building" if row.state == "building" else "table") if row else "flexible",
        "table": None,
        "columns": [],
        "row_count": None,
        "progress": None,
        "history": [
            {"id": str(h.id), "action": h.action, "summary": h.summary, "statements": h.statements, "applied_by": str(h.applied_by) if h.applied_by else None, "applied_at": h.applied_at.isoformat()}
            for h in history
        ],
    }
    if row:
        out["table"] = f"{SCHEMA}.{row.table_name}"
        out["columns"] = [_describe_column(s) for s in _in_field_order(_specs_from_registry(row), source.fields)]
        out["row_count"] = (await db.execute(text(f"SELECT count(*) FROM {qualified(row.table_name)}"))).scalar_one()
        if row.state == "building":
            out["progress"] = {"copied": out["row_count"], "total": await live_count(db, tenant_id, source)}
    return out


async def preview(db: AsyncSession, tenant_id: uuid.UUID, source: Source) -> dict:
    """What storing this source as a table would create, and which records
    stop it today."""
    tables = await tables_by_master(db, tenant_id)
    table = table_name_for(source)
    tables.setdefault(str(source.id), _new_registry_row(source, tenant_id, table))
    specs = column_specs(source, tables)
    own = tables[str(source.id)].table_name
    issues = await check(db, tenant_id, source, [s if s.ref_table != own else _self_link_check(s) for s in specs])
    issues += await _self_link_issues(db, tenant_id, source, [s for s in specs if s.ref_table == own])
    return {
        "table": f"{SCHEMA}.{own}",
        "columns": [_describe_column(s) for s in specs],
        "issues": issues,
    }


# ---- database errors ------------------------------------------------------------------------------------------


@dataclass
class StorageViolation:
    status: int
    message: str
    field: str | None = None
    extra: dict = dc_field(default_factory=dict)


async def explain_violation(db_factory, exc: BaseException) -> StorageViolation | None:
    """A database error raised by an entity table (or its sync trigger),
    in words a person filling a form can act on. None = not one of ours."""
    orig = getattr(exc, "orig", None)
    pg = getattr(orig, "__cause__", None) or orig
    message = str(getattr(pg, "message", None) or pg or "")
    sqlstate = getattr(pg, "sqlstate", None)
    if message.startswith("FF:type:"):
        _, _, fieldname, kind = message.split(":", 3)
        return StorageViolation(422, f"{fieldname} must be a valid {kind}.", fieldname)
    schema = getattr(pg, "schema_name", None)
    table = getattr(pg, "table_name", None)
    if schema != SCHEMA or not table:
        return None
    constraint = getattr(pg, "constraint_name", None) or ""
    column = getattr(pg, "column_name", None)
    detail = str(getattr(pg, "detail", "") or "")
    async with db_factory() as db:
        row = (await db.execute(select(EntityTable).where(EntityTable.table_name == table))).scalar_one_or_none()
        by_column = {c["column"]: c for c in (row.columns.values() if row else [])}
        if sqlstate == "23503" and "is still referenced from table" in detail:
            src = re.search(r'referenced from table "([^"]+)"', detail)
            names = await _master_names_for_tables(db, row.tenant_id, [src.group(1)]) if (row and src) else []
            used_by = names[0] if names else "another datasource"
            return StorageViolation(409, f"This record is still linked from {used_by} — remove those links first.")
    if sqlstate == "23502" and column:
        label = by_column.get(column, {}).get("label", column)
        return StorageViolation(422, f"{label} is required.", by_column.get(column, {}).get("field"))
    col = constraint.rsplit("__", 1)[0] if "__" in constraint else None
    spec = by_column.get(col or "", {})
    label = spec.get("label", col or "A value")
    if sqlstate == "23514" and constraint.endswith("__required"):
        return StorageViolation(422, f"{label} is required.", spec.get("field"))
    if sqlstate == "23514" and constraint.endswith("__options"):
        return StorageViolation(422, f"{label} must be one of: {', '.join(spec.get('options') or [])}.", spec.get("field"))
    if sqlstate == "23514" and constraint.endswith("__length"):
        return StorageViolation(422, f"{label} is longer than {spec.get('max_length')} characters.", spec.get("field"))
    if sqlstate == "23503":
        return StorageViolation(422, f"{label} links to a record that doesn't exist.", spec.get("field"))
    return None