"""
Audience "databases" — a parsed upload (CSV/XLSX/JSON), cleaned, classified, and normalized
(see backend/normalize.py for exactly what "normalized" means here and why it stops short of
blind full 5NF). Replaces the old flat backend/audiences.py contact list: a row now belongs to
a named, typed database rather than floating in one flat per-org list.

Column types: email, contact (phone), name, tag, bool, link, text (default). The auto-
classifier runs once at upload time; a column's type can be changed later via
update_column_type, which requires at least one existing value to conform to the new type's
syntax and flags (not blocks) every row that doesn't.

Encryption follows the same posture as the old contacts module (SRS FR-15): row data and
lookup-table data are both encrypted at rest via backend.crypto_utils, same mechanism as
mailbox credentials.
"""

from __future__ import annotations

import csv
import io
import json
import re
import uuid
from datetime import datetime, timezone

from backend import firestore_client as db
from backend import normalize
from backend.templating import split_items
from backend import orgs
from backend.crypto_utils import decrypt, encrypt

COLUMN_TYPES: tuple[str, ...] = ("email", "contact", "name", "tag", "bool", "link", "date", "text")

EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
# Sheets exports often prefix phone numbers with a stray apostrophe ('+234...) to stop them
# being read as numbers; tolerate it rather than call the whole column text.
PHONE_RE = re.compile(r"^'?\+?[\d][\d\s\-().]{5,}$")
# Dates and timestamps: 2004-05-17, 2024/03/09 14:22:01 GMT+1, 17/05/2004, 17 May 2004,
# 2025-01-02T10:00:00.000Z. Checked before PHONE_RE, which would otherwise swallow digits-and-dashes.
_TIME = r"\d{1,2}:\d{2}(:\d{2}(\.\d+)?)?(\s*(am|pm))?(\s*(z|gmt|utc|[a-z]{3})?\s*([+-]\d{1,2}(:?\d{2})?)?)?"
DATE_RE = re.compile(
    rf"^(\d{{4}}[-/.]\d{{1,2}}[-/.]\d{{1,2}}([ T]{_TIME})?"
    rf"|\d{{1,2}}[-/.]\d{{1,2}}[-/.]\d{{2,4}}([ T]{_TIME})?"
    r"|\d{1,2}\s+[A-Za-z]{3,9}\.?,?\s+\d{4}"
    r"|[A-Za-z]{3,9}\.?\s+\d{1,2},?\s+\d{4})$",
    re.IGNORECASE,
)
# "18 - 24", "25-34": a range of small numbers, not a phone number.
RANGE_RE = re.compile(r"^\d{1,3}\s*(-|to|–)\s*\d{1,3}\+?$|^\d{1,3}\+$", re.IGNORECASE)
URL_RE = re.compile(r"^(https?://|www\.)\S+$", re.IGNORECASE)
BOOL_VALUES = {"true", "false", "yes", "no", "y", "n", "1", "0"}

# None of these three ever raise for an empty cell — an unfilled field isn't a conformance
# failure, only a filled-but-malformed one is.
TYPE_VALIDATORS = {
    "email": lambda v: bool(EMAIL_RE.match(v.strip())),
    "contact": lambda v: bool(PHONE_RE.match(v.strip())) and not DATE_RE.match(v.strip()),
    "date": lambda v: bool(DATE_RE.match(v.strip())),
    "link": lambda v: bool(URL_RE.match(v.strip())),
    "bool": lambda v: v.strip().lower() in BOOL_VALUES,
}


class ValidationError(Exception):
    pass


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _database_path(org_id: str, database_id: str) -> str:
    return f"orgs/{org_id}/databases/{database_id}"


def _row_path(org_id: str, database_id: str, row_id: str) -> str:
    return f"orgs/{org_id}/databases/{database_id}/rows/{row_id}"


def _rows_prefix(org_id: str, database_id: str) -> str:
    return f"orgs/{org_id}/databases/{database_id}/rows/"


def _lookup_path(org_id: str, database_id: str, lookup_name: str) -> str:
    return f"orgs/{org_id}/databases/{database_id}/lookups/{lookup_name}"


# --- Parsing ------------------------------------------------------------------

def _parse_csv(file_bytes: bytes) -> tuple[list[str], list[dict]]:
    text = file_bytes.decode("utf-8-sig")
    reader = csv.DictReader(io.StringIO(text))
    headers = list(reader.fieldnames or [])
    return headers, [dict(row) for row in reader]


def _parse_json(file_bytes: bytes) -> tuple[list[str], list[dict]]:
    data = json.loads(file_bytes.decode("utf-8"))
    if isinstance(data, dict):
        data = data.get("rows") or data.get("data") or list(data.values())
    if not isinstance(data, list) or not data or not all(isinstance(r, dict) for r in data):
        raise ValidationError("JSON upload must be a non-empty array of objects")
    headers = list(dict.fromkeys(k for row in data for k in row.keys()))
    return headers, [{h: row.get(h) for h in headers} for row in data]


def _parse_xlsx(file_bytes: bytes) -> tuple[list[str], list[dict]]:
    import openpyxl  # only needed for this one branch — keep the import lazy

    workbook = openpyxl.load_workbook(io.BytesIO(file_bytes), read_only=True, data_only=True)
    sheet = workbook.active
    rows_iter = sheet.iter_rows(values_only=True)
    try:
        header_row = next(rows_iter)
    except StopIteration:
        raise ValidationError("Spreadsheet has no rows")
    headers = [str(h) if h is not None else "" for h in header_row]
    rows = [
        {headers[i]: (r[i] if i < len(r) else None) for i in range(len(headers))}
        for r in rows_iter
    ]
    return headers, rows


_PARSERS = {"csv": _parse_csv, "json": _parse_json, "xlsx": _parse_xlsx, "xls": _parse_xlsx}


def _clean(headers: list[str], rows: list[dict]) -> tuple[list[str], list[dict]]:
    """Trims whitespace, drops blank/duplicate headers, drops fully-empty rows and columns."""
    deduped: list[str] = []
    seen: set[str] = set()
    for h in headers:
        h = (h or "").strip()
        if not h:
            continue
        base, i = h, 1
        while h in seen:
            i += 1
            h = f"{base}_{i}"
        seen.add(h)
        deduped.append(h)

    cleaned_rows = []
    for row in rows:
        new_row = {h: (str(row.get(h)).strip() if row.get(h) not in (None, "") else "") for h in deduped}
        if any(new_row.values()):
            cleaned_rows.append(new_row)

    nonempty_cols = [h for h in deduped if any(r[h] for r in cleaned_rows)]
    cleaned_rows = [{h: r[h] for h in nonempty_cols} for r in cleaned_rows]
    return nonempty_cols, cleaned_rows


# --- Classification -------------------------------------------------------------

def _match_ratio(values: list[str], predicate) -> float:
    nonempty = [v for v in values if v]
    if not nonempty:
        return 0.0
    return sum(1 for v in nonempty if predicate(v)) / len(nonempty)


_PERSON_HEADER_RE = re.compile(r"^(full |first |last |sur|given |your |participant |contact )?(name|surname|fullname)\b", re.IGNORECASE)


def _looks_like_person_name_header(header: str) -> bool:
    """A header that merely *contains* "name" is often not a person ("Name of School /
    Institution", "Username", "Company name"): only headers that lead with name-ish words count."""
    return bool(_PERSON_HEADER_RE.match(header.strip())) and not re.search(
        r"\b(of|school|institution|organi[sz]ation|company|business|user|file|team|group)\b", header, re.IGNORECASE
    )


def classify_column(header: str, values: list[str]) -> str:
    if _match_ratio(values, lambda v: bool(EMAIL_RE.match(v))) >= 0.6:
        return "email"
    if _match_ratio(values, lambda v: bool(URL_RE.match(v))) >= 0.6:
        return "link"
    if _match_ratio(values, lambda v: v.lower() in BOOL_VALUES) >= 0.9:
        return "bool"
    if _match_ratio(values, lambda v: bool(DATE_RE.match(v))) >= 0.6:
        return "date"
    if _match_ratio(values, lambda v: bool(PHONE_RE.match(v)) and not RANGE_RE.match(v)) >= 0.6:
        return "contact"
    if _match_ratio(values, lambda v: ";" in v) >= 0.3:
        return "tag"
    if _looks_like_person_name_header(header):
        return "name"
    return "text"


LIST_DELIMITERS = (";", ",", "|", "\n", "mixed")


def detect_list_delimiter(values: list[str]) -> str | None:
    """Looks for a separator that makes this column a list of short items ("Acme, Beta Ltd").
    Returns the delimiter or None. Deliberately conservative: it only ever feeds a *suggestion*
    the user confirms, because commas also appear in names ("Smith, John") and sentences."""
    nonempty = [v for v in values if v]
    if len(nonempty) < 2:
        return None

    def qualifies(delim: str) -> bool:
        parts = [split_items(v, delim) for v in nonempty if (delim in v if delim != "mixed" else re.search(r"[;,|\n]", v))]
        if len(parts) < 2 or len(parts) / len(nonempty) < 0.3:
            return False
        items = [i for p in parts for i in p]
        if not items or any(len(i) > 60 for i in items):
            return False
        # Sentences have long "items"; lists of names/events/tags have short ones.
        return sum(len(i.split()) for i in items) / len(items) <= 4

    ratios = {d: sum(1 for v in nonempty if d in v) / len(nonempty) for d in (";", ",", "|", "\n")}
    if ratios[";"] >= 0.2 and ratios[","] >= 0.2 and qualifies("mixed"):
        return "mixed"
    for delim in sorted(ratios, key=lambda d: -ratios[d]):
        if ratios[delim] >= 0.3 and qualifies(delim):
            return delim
    return None


def _has_conforming_value(values: list[str], new_type: str) -> bool:
    validator = TYPE_VALIDATORS.get(new_type)
    if validator is None:
        return True  # name/tag/text have no syntax to conform to
    nonempty = [v for v in values if v]
    return not nonempty or any(validator(v) for v in nonempty)


def _flagged_values(values: list[str], col_type: str) -> list[int]:
    validator = TYPE_VALIDATORS.get(col_type)
    if validator is None:
        return []
    return [i for i, v in enumerate(values) if v and not validator(v)]


def _assign_row_ids(rows: list[dict]) -> list[tuple[str, dict]]:
    return [(uuid.uuid4().hex, row) for row in rows]


def _columns_with_flags(column_types: dict[str, str], headers: list[str], id_rows: list[tuple[str, dict]]) -> list[dict]:
    """Flags non-conforming values up front, at classification time — not only when a column's
    type is changed later. An auto-classified "email" column with one bad row should show that
    immediately, the same as a manual type change would."""
    columns = []
    for h in headers:
        values = [row.get(h, "") for _, row in id_rows]
        flagged_indices = _flagged_values(values, column_types[h])
        col = {
            "name": h,
            "type": column_types[h],
            "flagged_row_ids": [id_rows[i][0] for i in flagged_indices],
        }
        if col["type"] == "tag":
            col["delimiter"] = detect_list_delimiter(values) or ";"
        elif col["type"] in ("text", "name"):
            suggested = detect_list_delimiter(values)
            if suggested:
                col["list_suggestion"] = suggested
        columns.append(col)
    return columns


# --- Upload / create ------------------------------------------------------------

def create_database_from_upload(
    org_id: str, actor_uid: str, name: str, file_bytes: bytes, filename: str,
    save_copy=None,
) -> str:
    """save_copy(database_id, ext, file_bytes) -> None is injected rather than hardcoded so
    this stays unit-testable without touching disk; the real implementation lives in app.py's
    route (writes to uploads/{org_id}/{database_id}.{ext})."""
    orgs.require_permission(org_id, actor_uid, "edit_lists")

    ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
    parser = _PARSERS.get(ext)
    if parser is None:
        raise ValidationError(f"Unsupported file type {filename!r} — use .csv, .xlsx, or .json")

    headers, rows = parser(file_bytes)
    headers, rows = _clean(headers, rows)
    if not headers or not rows:
        raise ValidationError("No usable data found in that file after cleaning")

    column_types = {h: classify_column(h, [r[h] for r in rows]) for h in headers}
    identity_columns = {h for h in headers if column_types[h] in ("email", "contact")}
    result = normalize.normalize(rows, headers, exclude_determinants=identity_columns, auto_split=False)

    database_id = uuid.uuid4().hex
    if save_copy:
        save_copy(database_id, ext, file_bytes)

    id_rows = _assign_row_ids(result.root.rows)
    columns = _columns_with_flags(column_types, result.root.columns, id_rows)

    db.set_doc(_database_path(org_id, database_id), {
        "name": name,
        "source_filename": filename,
        "source_kind": "xlsx" if ext in ("xlsx", "xls") else ext,
        "created_at": _now(),
        "row_count": len(result.root.rows),
        "columns": columns,
        "lookup_tables": [{"name": t.name, "columns": t.columns} for t in result.lookups],
        "suggestions": result.suggestions,
    })

    for lookup in result.lookups:
        db.set_doc(_lookup_path(org_id, database_id, lookup.name), {
            "columns": lookup.columns,
            "encrypted_rows": encrypt(json.dumps(lookup.rows)),
        })

    for row_id, row in id_rows:
        db.set_doc(_row_path(org_id, database_id, row_id), {"encrypted_data": encrypt(json.dumps(row))})

    return database_id


def get_database(org_id: str, database_id: str) -> dict:
    doc = db.get_doc(_database_path(org_id, database_id))
    if doc is None:
        raise orgs.NotFoundError(f"No database {database_id!r} in org {org_id!r}")
    return doc


def list_databases(org_id: str) -> dict[str, dict]:
    prefix = f"orgs/{org_id}/databases/"
    docs = db.query_prefix(prefix)
    return {path[len(prefix):]: data for path, data in docs.items()}


def list_rows(org_id: str, database_id: str, search: str | None = None, sort: str | None = None) -> list[dict]:
    """Every row, decrypted, with its row_id — client-side-sized (prototype scale), so search
    and sort both happen here in Python rather than needing a real query engine."""
    prefix = _rows_prefix(org_id, database_id)
    docs = db.query_prefix(prefix)
    rows = []
    for path, data in docs.items():
        row = json.loads(decrypt(data["encrypted_data"]))
        row["row_id"] = path[len(prefix):]
        rows.append(row)

    if search:
        needle = search.strip().lower()
        rows = [r for r in rows if any(needle in str(v).lower() for v in r.values())]

    if sort:
        rows.sort(key=lambda r: (r.get(sort) is None, str(r.get(sort, "")).lower()))

    return rows


# Operators a segment condition can use. Which ones make sense depends on the column's type —
# SEGMENT_OPERATORS_BY_TYPE below is what the frontend uses to only offer valid ones, but
# _row_matches tolerates any operator against any column rather than erroring, since a changed
# column type shouldn't break a condition written before the change.
SEGMENT_OPERATORS_BY_TYPE: dict[str, tuple[str, ...]] = {
    "bool": ("is_true", "is_false", "is_empty", "is_not_empty"),
    "tag": ("has_tag", "lacks_tag", "is_empty", "is_not_empty"),
}
_DEFAULT_OPERATORS = ("equals", "not_equals", "contains", "not_contains", "is_empty", "is_not_empty")


def _row_matches(row: dict, col_type: str, field: str, op: str, value: str, delimiter: str | None = None) -> bool:
    text = str(row.get(field, "") or "")

    if op == "is_empty":
        return text.strip() == ""
    if op == "is_not_empty":
        return text.strip() != ""

    if col_type == "bool":
        truthy = text.strip().lower() in {"true", "yes", "y", "1"}
        if op == "is_true":
            return truthy
        if op == "is_false":
            return not truthy

    if col_type == "tag":
        tags = {t.lower() for t in split_items(text, delimiter)}
        needle = value.strip().lower()
        if op == "has_tag":
            return needle in tags
        if op == "lacks_tag":
            return needle not in tags

    needle = value.strip().lower()
    hay = text.strip().lower()
    if op == "equals":
        return hay == needle
    if op == "not_equals":
        return hay != needle
    if op == "contains":
        return needle in hay
    if op == "not_contains":
        return needle not in hay
    return True


def apply_segment(rows: list[dict], columns: list[dict], conditions: list[dict]) -> list[dict]:
    """Narrows `rows` to only those matching every condition (AND). Each condition is
    {field, op, value}; unknown fields match nothing (a stale condition from a since-removed
    column shouldn't silently widen a send back out to everyone)."""
    col_types = {c["name"]: c["type"] for c in columns}
    delimiters = {c["name"]: c.get("delimiter") for c in columns}
    matched = rows
    for cond in conditions:
        field = cond.get("field", "")
        if field not in col_types:
            return []
        op = cond.get("op", "")
        value = str(cond.get("value", ""))
        col_type = col_types[field]
        matched = [r for r in matched if _row_matches(r, col_type, field, op, value, delimiters[field])]
    return matched


def preview_segment(org_id: str, database_id: str, conditions: list[dict]) -> dict:
    """Live "how many recipients does this match" count for the campaign segment builder, plus
    a small sample so the UI can show who, not just how many."""
    database = get_database(org_id, database_id)
    rows = list_rows(org_id, database_id)
    matched = apply_segment(rows, database["columns"], conditions)
    return {"matched": len(matched), "sample": matched[:5]}


def add_row(org_id: str, actor_uid: str, database_id: str, values: dict) -> str:
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    return insert_row(org_id, database_id, values)


def insert_row(org_id: str, database_id: str, values: dict) -> str:
    """add_row without the role check — for callers that have already authorized the insert
    some other way (backend/collect.py: a confirmed double opt-in, not a signed-in user)."""
    get_database(org_id, database_id)  # raises NotFoundError if missing
    row_id = uuid.uuid4().hex
    db.set_doc(_row_path(org_id, database_id, row_id), {"encrypted_data": encrypt(json.dumps(values))})
    database = get_database(org_id, database_id)
    db.update_doc(_database_path(org_id, database_id), {"row_count": database["row_count"] + 1})
    known = {c["name"] for c in database["columns"]}
    new_cols = [k for k in values if k not in known]
    if new_cols:
        updated_columns = database["columns"] + [{"name": k, "type": "text", "flagged_row_ids": []} for k in new_cols]
        db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns})
    return row_id


def update_row(org_id: str, actor_uid: str, database_id: str, row_id: str, tags: dict) -> None:
    """Merges `tags` into the row's existing values. Any key not already a known column is
    appended to the database's column list (type "text") — a manually-added tag becomes a
    real, placeholder-usable column for the whole database, matching how the CSV-derived
    columns work, not a side channel."""
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    path = _row_path(org_id, database_id, row_id)
    doc = db.get_doc(path)
    if doc is None:
        raise orgs.NotFoundError(f"No row {row_id!r} in database {database_id!r}")
    row = json.loads(decrypt(doc["encrypted_data"]))
    row.update(tags)
    db.set_doc(path, {"encrypted_data": encrypt(json.dumps(row))})

    database = get_database(org_id, database_id)
    known = {c["name"] for c in database["columns"]}
    new_cols = [k for k in tags if k not in known]
    if new_cols:
        updated_columns = database["columns"] + [{"name": k, "type": "text", "flagged_row_ids": []} for k in new_cols]
        db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns})


def bulk_update_rows(org_id: str, actor_uid: str, database_id: str, row_ids: list[str], tags: dict) -> list[str]:
    """Same merge semantics as update_row, applied to many rows in one call — the mass "fill a
    column" / "add a tag to the selection" grid actions need this, since looping update_row
    client-side would mean one HTTP round trip per selected row (unusable at a few thousand
    rows). Fetches the database doc once up front and updates it once at the end, rather than
    once per row like update_row does. Silently skips a row_id that no longer exists (e.g. a
    stale selection from before a delete) instead of failing the whole batch over it."""
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    database = get_database(org_id, database_id)

    updated: list[str] = []
    for row_id in row_ids:
        path = _row_path(org_id, database_id, row_id)
        doc = db.get_doc(path)
        if doc is None:
            continue
        row = json.loads(decrypt(doc["encrypted_data"]))
        row.update(tags)
        db.set_doc(path, {"encrypted_data": encrypt(json.dumps(row))})
        updated.append(row_id)

    known = {c["name"] for c in database["columns"]}
    new_cols = [k for k in tags if k not in known]
    if new_cols:
        updated_columns = database["columns"] + [{"name": k, "type": "text", "flagged_row_ids": []} for k in new_cols]
        db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns})

    return updated


def bulk_append_tag(
    org_id: str, actor_uid: str, database_id: str, row_ids: list[str], tag: str, tag_column: str = "Tags",
) -> list[str]:
    """Appends `tag` to the existing semicolon-delimited tag_column value for each row, rather
    than overwriting it like bulk_update_rows does — each row keeps whatever tags it already
    had, so this can't be expressed as one uniform PATCH value the way a column "fill" can.
    Still a single HTTP call for the whole selection, looping server-side rather than once per
    row over the network."""
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    database = get_database(org_id, database_id)

    updated: list[str] = []
    for row_id in row_ids:
        path = _row_path(org_id, database_id, row_id)
        doc = db.get_doc(path)
        if doc is None:
            continue
        row = json.loads(decrypt(doc["encrypted_data"]))
        existing = [t for t in str(row.get(tag_column, "")).split(";") if t]
        if tag not in existing:
            existing.append(tag)
        row[tag_column] = ";".join(existing)
        db.set_doc(path, {"encrypted_data": encrypt(json.dumps(row))})
        updated.append(row_id)

    known = {c["name"] for c in database["columns"]}
    if tag_column not in known:
        updated_columns = database["columns"] + [{"name": tag_column, "type": "tag", "flagged_row_ids": []}]
        db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns})

    return updated


def delete_rows(org_id: str, actor_uid: str, database_id: str, row_ids: list[str]) -> int:
    """Bulk delete for the grid's mass-selection "Delete" action — same one-call-not-N
    reasoning as bulk_update_rows. Returns the number actually deleted (stale ids are skipped,
    not errors) so the caller can show an accurate count."""
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    database = get_database(org_id, database_id)

    deleted = 0
    for row_id in row_ids:
        path = _row_path(org_id, database_id, row_id)
        if db.get_doc(path) is None:
            continue
        db.delete_doc(path)
        deleted += 1

    if deleted:
        db.update_doc(_database_path(org_id, database_id), {"row_count": max(0, database["row_count"] - deleted)})

    return deleted


def bulk_set_values(org_id: str, actor_uid: str, database_id: str, updates: dict[str, dict]) -> list[str]:
    """Like bulk_update_rows, but each row gets its OWN value map rather than one uniform
    value for everyone. This is what undoing a "fill column for N rows" needs — each row's
    prior value was (generally) different before the fill overwrote it, so restoring them
    can't be expressed as a single uniform PATCH the way the fill itself was."""
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    database = get_database(org_id, database_id)

    updated: list[str] = []
    all_new_keys: set[str] = set()
    for row_id, values in updates.items():
        path = _row_path(org_id, database_id, row_id)
        doc = db.get_doc(path)
        if doc is None:
            continue
        row = json.loads(decrypt(doc["encrypted_data"]))
        row.update(values)
        db.set_doc(path, {"encrypted_data": encrypt(json.dumps(row))})
        updated.append(row_id)
        all_new_keys.update(values.keys())

    known = {c["name"] for c in database["columns"]}
    new_cols = [k for k in all_new_keys if k not in known]
    if new_cols:
        updated_columns = database["columns"] + [{"name": k, "type": "text", "flagged_row_ids": []} for k in new_cols]
        db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns})
    return updated


def remove_column(org_id: str, actor_uid: str, database_id: str, column: str) -> dict:
    """Drops a column from the schema and strips it from every row. Returns a snapshot
    ({column def, values_by_row_id}) so the caller (the grid's undo stack) can restore it
    exactly via restore_column without a second round trip to reconstruct what was deleted."""
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    database = get_database(org_id, database_id)
    col_def = next((c for c in database["columns"] if c["name"] == column), None)
    if col_def is None:
        raise orgs.NotFoundError(f"No column {column!r} in database {database_id!r}")

    values_by_row_id: dict[str, str] = {}
    for row in list_rows(org_id, database_id):
        row_id = row["row_id"]
        if column not in row:
            continue
        values_by_row_id[row_id] = row[column]
        remaining = {k: v for k, v in row.items() if k not in (column, "row_id")}
        db.set_doc(_row_path(org_id, database_id, row_id), {"encrypted_data": encrypt(json.dumps(remaining))})

    updated_columns = [c for c in database["columns"] if c["name"] != column]
    db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns})
    return {"column": col_def, "values_by_row_id": values_by_row_id}


def restore_column(org_id: str, actor_uid: str, database_id: str, column_def: dict, values_by_row_id: dict[str, str]) -> None:
    """The inverse of remove_column — undo target for the grid's "Remove column" action."""
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    database = get_database(org_id, database_id)
    for row_id, value in values_by_row_id.items():
        path = _row_path(org_id, database_id, row_id)
        doc = db.get_doc(path)
        if doc is None:
            continue
        row = json.loads(decrypt(doc["encrypted_data"]))
        row[column_def["name"]] = value
        db.set_doc(path, {"encrypted_data": encrypt(json.dumps(row))})
    if not any(c["name"] == column_def["name"] for c in database["columns"]):
        db.update_doc(_database_path(org_id, database_id), {"columns": database["columns"] + [column_def]})


def rename_column(org_id: str, actor_uid: str, database_id: str, column: str, new_name: str) -> str:
    """Renames a column in the schema and re-keys it in every row, keeping column order. No
    values change, so renaming back is an exact undo. Saved presets and campaigns that use
    {OldName} keep the old text; the UI warns about that."""
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    new_name = " ".join(new_name.split())
    if not new_name:
        raise ValidationError("A column name can't be empty")
    if new_name == "row_id":
        raise ValidationError("That name is reserved")
    database = get_database(org_id, database_id)
    if not any(c["name"] == column for c in database["columns"]):
        raise orgs.NotFoundError(f"No column {column!r} in database {database_id!r}")
    if new_name == column:
        return new_name
    if any(c["name"].lower() == new_name.lower() for c in database["columns"] if c["name"] != column):
        raise ValidationError(f"There is already a column called {new_name!r}")

    for row in list_rows(org_id, database_id):
        if column not in row:
            continue
        rekeyed = {(new_name if k == column else k): v for k, v in row.items() if k != "row_id"}
        db.set_doc(_row_path(org_id, database_id, row["row_id"]), {"encrypted_data": encrypt(json.dumps(rekeyed))})

    updated_columns = [{**c, "name": new_name} if c["name"] == column else c for c in database["columns"]]
    db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns})
    return new_name


def column_rename_suggestions(org_id: str, database_id: str) -> list[dict]:
    from backend import rename_suggest

    return rename_suggest.suggest_renames(get_database(org_id, database_id)["columns"])


def column_type_suggestions(org_id: str, database_id: str) -> list[dict]:
    """Columns whose stored type no longer matches what their values say (e.g. a date column
    that an older classifier called "contact"). Only ever a suggestion; never applied unasked.
    Only recommends moving between the plain, low-stakes types, so a column the person
    deliberately set to email, list or yes/no is left alone."""
    database = get_database(org_id, database_id)
    rows = list_rows(org_id, database_id)
    suggestions = []
    for col in database["columns"]:
        if col["type"] not in ("text", "contact", "name"):
            continue
        guess = classify_column(col["name"], [r.get(col["name"], "") for r in rows])
        if guess != col["type"] and guess in ("text", "date", "contact", "name", "email", "link", "bool"):
            suggestions.append({"column": col["name"], "current": col["type"], "suggested": guess})
    return suggestions


def rename_database(org_id: str, actor_uid: str, database_id: str, new_name: str) -> None:
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    get_database(org_id, database_id)
    if not new_name.strip():
        raise ValidationError("Name can't be empty")
    db.update_doc(_database_path(org_id, database_id), {"name": new_name.strip()})


def set_database_disabled(org_id: str, actor_uid: str, database_id: str, disabled: bool) -> None:
    """"Disabled" means excluded from campaign sends only — still fully visible and editable
    here. Doesn't cascade into blocking merges or anything else; that would be a separate,
    more surprising rule than what the name implies."""
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    get_database(org_id, database_id)
    db.update_doc(_database_path(org_id, database_id), {"disabled": disabled})


def delete_database(org_id: str, actor_uid: str, database_id: str) -> None:
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    get_database(org_id, database_id)
    for path in list(db.query_prefix(_rows_prefix(org_id, database_id)).keys()):
        db.delete_doc(path)
    lookups_prefix = f"orgs/{org_id}/databases/{database_id}/lookups/"
    for path in list(db.query_prefix(lookups_prefix).keys()):
        db.delete_doc(path)
    db.delete_doc(_database_path(org_id, database_id))


def duplicate_database(org_id: str, actor_uid: str, database_id: str, new_name: str) -> str:
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    database = get_database(org_id, database_id)
    rows = list_rows(org_id, database_id)
    new_id = uuid.uuid4().hex
    id_map = {r["row_id"]: uuid.uuid4().hex for r in rows}

    db.set_doc(_database_path(org_id, new_id), {**database, "name": new_name, "created_at": _now()})
    for r in rows:
        payload = {k: v for k, v in r.items() if k != "row_id"}
        db.set_doc(_row_path(org_id, new_id, id_map[r["row_id"]]), {"encrypted_data": encrypt(json.dumps(payload))})

    lookups_prefix = f"orgs/{org_id}/databases/{database_id}/lookups/"
    for path, data in db.query_prefix(lookups_prefix).items():
        db.set_doc(_lookup_path(org_id, new_id, path[len(lookups_prefix):]), data)

    # re-point flagged_row_ids at the new row ids — leaving them pointing at the OLD
    # database's ids would silently flag the wrong rows (or none, once those ids never match).
    updated_columns = [
        {**c, "flagged_row_ids": [id_map[rid] for rid in c["flagged_row_ids"] if rid in id_map]}
        for c in database["columns"]
    ]
    db.update_doc(_database_path(org_id, new_id), {"columns": updated_columns})
    return new_id


def get_lookup_table(org_id: str, database_id: str, lookup_name: str) -> dict:
    """Read-only access to a normalized lookup table's actual rows — the database doc only
    ever stores the lookup's NAME and column list (see create_database_from_upload), never its
    rows, so the grid's lookup-table viewer needs a dedicated fetch rather than reading it off
    the database doc it already has."""
    doc = db.get_doc(_lookup_path(org_id, database_id, lookup_name))
    if doc is None:
        raise orgs.NotFoundError(f"No lookup table {lookup_name!r} in database {database_id!r}")
    return {"columns": doc["columns"], "rows": json.loads(decrypt(doc["encrypted_rows"]))}


def set_column_list(org_id: str, actor_uid: str, database_id: str, column: str, delimiter: str | None) -> None:
    """Makes `column` a list column split on `delimiter` (or, with None, dismisses the
    suggestion to do so and leaves the column as it is)."""
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    if delimiter is not None and delimiter not in LIST_DELIMITERS:
        raise ValidationError(f"Unknown list separator {delimiter!r}")
    database = get_database(org_id, database_id)
    if not any(c["name"] == column for c in database["columns"]):
        raise orgs.NotFoundError(f"No column {column!r} in database {database_id!r}")
    updated = []
    for c in database["columns"]:
        if c["name"] == column:
            c = {k: v for k, v in c.items() if k != "list_suggestion"}
            if delimiter is not None:
                c = {**c, "type": "tag", "delimiter": delimiter, "flagged_row_ids": []}
        updated.append(c)
    db.update_doc(_database_path(org_id, database_id), {"columns": updated})


def update_column_type(org_id: str, database_id: str, column: str, new_type: str) -> list[str]:
    if new_type not in COLUMN_TYPES:
        raise ValidationError(f"Unknown column type {new_type!r}")
    database = get_database(org_id, database_id)
    if not any(c["name"] == column for c in database["columns"]):
        raise orgs.NotFoundError(f"No column {column!r} in database {database_id!r}")

    rows = list_rows(org_id, database_id)
    values = [r.get(column, "") for r in rows]
    if not _has_conforming_value(values, new_type):
        raise ValidationError(
            f"No existing value in {column!r} looks like a valid {new_type} — refusing the change"
        )

    flagged_indices = _flagged_values(values, new_type)
    flagged_row_ids = [rows[i]["row_id"] for i in flagged_indices]

    updated_columns = [
        {**{k: v for k, v in c.items() if k != "list_suggestion"}, "type": new_type, "flagged_row_ids": flagged_row_ids,
         **({"delimiter": c.get("delimiter") or ";"} if new_type == "tag" else {})}
        if c["name"] == column else c
        for c in database["columns"]
    ]
    db.update_doc(_database_path(org_id, database_id), {"columns": updated_columns})
    return flagged_row_ids


def merge_databases(org_id: str, actor_uid: str, database_id_a: str, database_id_b: str, new_name: str) -> str:
    """Concatenates both databases' rows under a brand-new database, reconciling columns by
    name (union) and re-running classification + normalization over the combined set — a
    merge isn't just a row concat, it's a fresh upload of the combined data."""
    orgs.require_permission(org_id, actor_uid, "edit_lists")
    rows_a = list_rows(org_id, database_id_a)
    rows_b = list_rows(org_id, database_id_b)
    headers = list(dict.fromkeys([k for r in rows_a + rows_b for k in r if k != "row_id"]))

    # Exact-duplicate rows (e.g. the same contact present in both databases) are dropped
    # before normalizing — besides being the obviously-wanted merge behavior on its own, an
    # un-deduplicated duplicate row makes EVERY column pair look like a valid functional
    # dependency (they trivially agree with their own duplicate), which would otherwise send
    # the FD-discovery heuristic down a false trail.
    seen_rows: set[tuple] = set()
    combined: list[dict] = []
    for r in rows_a + rows_b:
        row = {h: r.get(h, "") for h in headers}
        key = tuple(row[h] for h in headers)
        if key in seen_rows:
            continue
        seen_rows.add(key)
        combined.append(row)

    column_types = {h: classify_column(h, [r[h] for r in combined]) for h in headers}
    identity_columns = {h for h in headers if column_types[h] in ("email", "contact")}
    result = normalize.normalize(combined, headers, exclude_determinants=identity_columns, auto_split=False)

    database_id = uuid.uuid4().hex
    id_rows = _assign_row_ids(result.root.rows)
    columns = _columns_with_flags(column_types, result.root.columns, id_rows)

    db.set_doc(_database_path(org_id, database_id), {
        "name": new_name,
        "source_filename": f"merge({database_id_a[:8]}, {database_id_b[:8]})",
        "source_kind": "merge",
        "created_at": _now(),
        "row_count": len(result.root.rows),
        "columns": columns,
        "lookup_tables": [{"name": t.name, "columns": t.columns} for t in result.lookups],
        "suggestions": result.suggestions,
    })
    for lookup in result.lookups:
        db.set_doc(_lookup_path(org_id, database_id, lookup.name), {
            "columns": lookup.columns,
            "encrypted_rows": encrypt(json.dumps(lookup.rows)),
        })
    for row_id, row in id_rows:
        db.set_doc(_row_path(org_id, database_id, row_id), {"encrypted_data": encrypt(json.dumps(row))})
    return database_id