"""
CA Automator — database layer.

Plain SQL over sqlite3 from the standard library. No ORM, so the PostgreSQL port
is mechanical (see the notes at the foot of schema.sql).
"""
import base64
import contextvars
import hmac
import json
import os
import re
import sqlite3
import hashlib
import secrets
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone

import nepal_calendar

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.environ.get("CA_DATA_DIR", os.path.join(os.path.dirname(BASE_DIR), "data"))
DB_PATH = os.environ.get("CA_DB_PATH", os.path.join(DATA_DIR, "ca_automator.db"))
STORAGE_DIR = os.path.join(DATA_DIR, "storage")

# ---------------------------------------------------------------- engine
# Everything above and below this block is unchanged, unconditional SQLite
# behavior — the default, and what every existing deployment keeps running
# exactly as before. Setting CA_DB_ENGINE=mysql (plus CA_MYSQL_HOST/PORT/
# USER/PASSWORD/DATABASE) switches every function in this module onto the
# MySQL path below instead, built for cheap-shared-hosting/cPanel
# deployments that offer MySQL but not SQLite-friendly persistent disk. The
# two paths share every function signature in this file — nothing outside
# db.py (1,200+ call sites in app.py/engines.py/etc.) needs to know or care
# which engine is live.
DB_ENGINE = os.environ.get("CA_DB_ENGINE", "sqlite").strip().lower()
if DB_ENGINE not in ("sqlite", "mysql"):
    raise RuntimeError(f"Unknown CA_DB_ENGINE: {DB_ENGINE!r} (expected 'sqlite' or 'mysql')")

if DB_ENGINE == "mysql":
    import pymysql
    import pymysql.err

    MYSQL_CONFIG = {
        "host": os.environ.get("CA_MYSQL_HOST", "localhost"),
        "port": int(os.environ.get("CA_MYSQL_PORT", "3306")),
        "user": os.environ.get("CA_MYSQL_USER", "root"),
        "password": os.environ.get("CA_MYSQL_PASSWORD", ""),
        "database": os.environ.get("CA_MYSQL_DATABASE", "ca_automator"),
        "charset": "utf8mb4",
    }

    class MySQLRow:
        """Stands in for sqlite3.Row — supports both row["col"] and row[0]
        (app.py mixes both styles throughout, and some code unpacks a row
        by position, e.g. "for firm_id, raw in rows:"), plus the .keys()
        method dict(row) relies on internally."""
        __slots__ = ("_map", "_values")

        def __init__(self, cols, values):
            self._values = values
            self._map = dict(zip(cols, values))

        def __getitem__(self, key):
            return self._values[key] if isinstance(key, int) else self._map[key]

        def __iter__(self):
            return iter(self._values)

        def __len__(self):
            return len(self._values)

        def __contains__(self, key):
            return key in self._map

        def __repr__(self):
            return f"MySQLRow({self._map!r})"

        def keys(self):
            return self._map.keys()

        def get(self, key, default=None):
            return self._map.get(key, default)

    # SQLite has no real types — a column declared TEXT happily stores an
    # int, and every value the app already has round-trips as the same
    # Python type it was written as. MySQL is strict: a DATETIME column
    # (used for every *_at timestamp — see server/schema_mysql.sql's
    # DEFAULT CURRENT_TIMESTAMP notes) comes back as a real datetime.datetime,
    # a DATE as a date, and DECIMAL as a Decimal. Converting each back to
    # exactly the string/float shape db.now_iso() and ordinary float() math
    # already produce is what lets every one of app.py's 1,200+ call sites
    # (string-slicing a date, comparing two "created_at" strings, doing
    # arithmetic on a money value) stay completely unaware the underlying
    # column type changed.
    def _normalize_mysql_value(v):
        import datetime as _dt
        import decimal
        if isinstance(v, _dt.datetime):
            return v.strftime("%Y-%m-%d %H:%M:%S")
        if isinstance(v, _dt.date):
            return v.strftime("%Y-%m-%d")
        if isinstance(v, decimal.Decimal):
            return float(v)
        if isinstance(v, (bytes, bytearray)):
            return v.decode("utf-8", "replace")
        return v

    # Translates the handful of SQLite-only SQL fragments that appear
    # inline across app.py/db.py/notify.py/oauth_login.py into their MySQL
    # equivalents, plus the universal "?" -> "%s" placeholder style — done
    # centrally, here, so none of those 1,200+ call sites need a MySQL-
    # specific variant of their query string. Every substitution here is
    # safe as a blind text replace because each SQLite fragment (datetime
    # ('now'), a bare "?" placeholder, "INSERT OR IGNORE") is SQL syntax the
    # developer wrote, never data a user could have typed into a string
    # literal that this might accidentally corrupt.
    def _translate_sql(sql):
        sql = sql.replace("datetime('now')", "NOW()")
        sql = sql.replace("date('now')", "CURDATE()")
        sql = sql.replace("INSERT OR IGNORE", "INSERT IGNORE")
        # SQLite/Postgres-style UPSERT ("... ON CONFLICT(cols) DO UPDATE SET
        # a=excluded.a, ..." or "... DO NOTHING") -> MySQL's "... ON
        # DUPLICATE KEY UPDATE a=VALUES(a), ..." (MySQL infers the conflict
        # target from whichever unique/primary key was violated, so the
        # explicit column list simply drops). DO NOTHING has no direct
        # MySQL equivalent attached to an INSERT's tail — "INSERT IGNORE"
        # (prefixed on the statement, same net effect: silently skip a
        # conflicting row) is the idiomatic substitute.
        if "ON CONFLICT" in sql and re.search(r"ON CONFLICT\s*\([^)]*\)\s*DO NOTHING", sql):
            sql = re.sub(r"ON CONFLICT\s*\([^)]*\)\s*DO NOTHING\s*", "", sql)
            sql = re.sub(r"^(\s*)INSERT INTO", r"\1INSERT IGNORE INTO", sql)
        else:
            sql = re.sub(r"ON CONFLICT\s*\([^)]*\)\s*DO UPDATE SET", "ON DUPLICATE KEY UPDATE", sql)
            sql = re.sub(r"\bexcluded\.(\w+)", r"VALUES(\1)", sql)
        sql = sql.replace("?", "%s")
        return sql

    class _MySQLCursorWrapper:
        def __init__(self, cursor):
            self._cursor = cursor

        def _cols(self):
            return [d[0] for d in self._cursor.description] if self._cursor.description else []

        def fetchall(self):
            cols = self._cols()
            return [MySQLRow(cols, tuple(_normalize_mysql_value(v) for v in row)) for row in self._cursor.fetchall()]

        def fetchone(self):
            row = self._cursor.fetchone()
            if row is None:
                return None
            cols = self._cols()
            return MySQLRow(cols, tuple(_normalize_mysql_value(v) for v in row))

        @property
        def lastrowid(self):
            return self._cursor.lastrowid

    class _MySQLConnWrapper:
        """Enough of sqlite3.Connection's surface (.execute() on the
        connection itself, .commit()/.rollback()/.close()) that every
        existing "with conn() as c: c.execute(...)" call site in this file
        and the 3 in app.py/run.py works completely unchanged."""
        def __init__(self, raw):
            self._raw = raw

        def execute(self, sql, args=()):
            cur = self._raw.cursor()
            # PyMySQL substitutes params via old-style %-formatting — pass
            # None (not an empty tuple) when there truly are none, or it
            # mogrifies the query text regardless and breaks on any literal
            # "%" already in the SQL (e.g. a LIKE '%...%' pattern with no
            # bound arguments at all, as in _sanitize_maps_embed below).
            # Real placeholders are unaffected either way: SQLite's "?"
            # (translated to "%s" in _translate_sql above) only appears in a
            # query that actually has args to go with it.
            cur.execute(_translate_sql(sql), tuple(args) if args else None)
            return _MySQLCursorWrapper(cur)

        def executescript(self, script):
            # No PyMySQL/MySQL equivalent of sqlite3's executescript — split
            # into individual statements. Safe here because every script
            # this is ever called with (schema_mysql.sql, seed.sql) is
            # authored by this app, not user input, so a naive split on
            # ";\n" (never split mid-statement by design — see the schema
            # generator) doesn't risk cutting a string literal in half.
            for stmt in _split_sql_statements(script):
                try:
                    self._raw.cursor().execute(_translate_sql(stmt))
                except pymysql.err.OperationalError as e:
                    # 1050 = table already exists ("CREATE TABLE IF NOT
                    # EXISTS" on a table MySQL still had to touch to check),
                    # 1061 = duplicate key/index name (MySQL has no "CREATE
                    # INDEX IF NOT EXISTS" — see schema_mysql.sql's own
                    # generation notes). Both mean "already applied,"
                    # exactly what running this idempotently is supposed to
                    # tolerate. Anything else is a real problem.
                    if e.args and e.args[0] in (1050, 1061):
                        continue
                    raise

        def commit(self):
            self._raw.commit()

        def rollback(self):
            self._raw.rollback()

        def close(self):
            self._raw.close()

    def _split_sql_statements(script):
        """Splits a .sql file's text into individual statements on ';' at
        the top level, skipping '--' line comments and ';' inside a quoted
        string — good enough for schema_mysql.sql/seed.sql specifically
        (this app's own generated/authored SQL, never arbitrary input)."""
        statements = []
        buf = []
        in_string = None
        i, n = 0, len(script)
        while i < n:
            ch = script[i]
            if in_string:
                buf.append(ch)
                if ch == in_string and script[i - 1:i] != "\\":
                    in_string = None
                i += 1
                continue
            if ch in ("'", '"'):
                in_string = ch
                buf.append(ch)
                i += 1
                continue
            if ch == "-" and script[i:i + 2] == "--":
                nl = script.find("\n", i)
                i = n if nl == -1 else nl + 1
                continue
            if ch == ";":
                stmt = "".join(buf).strip()
                if stmt:
                    statements.append(stmt)
                buf = []
                i += 1
                continue
            buf.append(ch)
            i += 1
        tail = "".join(buf).strip()
        if tail:
            statements.append(tail)
        return statements

# A shared name every "except db.IntegrityError:" call site across app.py
# (duplicate slug, unique constraint violation, etc.) can catch regardless
# of which engine is live — sqlite3.IntegrityError and pymysql's are
# different classes, so those call sites can't reference either directly.
IntegrityError = pymysql.err.IntegrityError if DB_ENGINE == "mysql" else sqlite3.IntegrityError

# Days a firm's staff keep full access after firm.subscription_expires_at
# passes before the firm is actually suspended. Lives here (not app.py or
# notify.py) so both can reference the same constant without a circular
# import — app.py enforces it, notify.py's sweep_subscriptions() uses it to
# time the grace/suspension notices.
SUBSCRIPTION_GRACE_DAYS = 3

# Every module_key a user_module_grant/firm_module_grant row can legally
# name — the 21 classic engines + 9 relational modules web/app.js's
# MODULES/EXTRA_MODULES dicts grant by (their dict key, not always their
# route slug — e.g. Transfer Pricing is granted as "transfer_pricing", its
# route is "tp"), plus 5 admin content-management areas that used to be
# all-or-nothing behind is_super_admin and are now individually grantable
# the same way (site_settings/blog_admin/career_admin/chatbot_admin/
# billing). Keep this in sync with app.js's MODULE_CATALOG + those 5 keys —
# server/app.py's admin_create_user and admin_update_user reject any
# module_key not in this set before it ever reaches either grant table.
ALL_GRANTABLE_MODULE_KEYS = {
    # classic engines (MODULES dict)
    "audit_report", "biz_tax", "capital_advisory", "dd", "depreciation", "emi", "fdi",
    "fin_statements", "fixed_assets", "grant", "internal_audit", "inventory", "lease",
    "penalty", "personal_tax", "projection", "salary_tds", "stat_audit", "tds", "valuation", "vat",
    # UAE-specific — see server/uae_engines.py. Real module pages like the
    # classic engines above (unlike the India keys below, which are still
    # API-only) — country_service.MODULE_COUNTRY_APPLICABILITY scopes them
    # to country_code='AE' so a Nepal/India firm never sees them granted.
    "uae_corporate_tax", "uae_penalties",
    # relational modules (EXTRA_MODULES dict — granted by dict key, not route)
    "transfer_pricing", "practice_mgmt", "forensic_audit", "corporate_advisory",
    "business_advisory", "ipo_advisory", "tax_advisory", "audit_execution", "acct_automation",
    # admin content areas — individually grantable, see RAIL_HELP/buildRail() in app.js
    "site_settings", "blog_admin", "career_admin", "chatbot_admin", "billing",
    # organization/practice administration — branches, departments, teams,
    # positions, employee profiles, skills taxonomy, workflow rules. Tasks,
    # client contacts, engagement team rosters and the calendar are NOT
    # gated behind this (or any) module — same precedent as client_followup/
    # client_credential: ordinary work-management actions any staff member
    # with client/engagement access already has, not a specialized add-on.
    "org_admin",
    # India-specific engines — see server/india_engines.py. Callable via the
    # same POST /api/compute every other engine uses; no dedicated
    # module-page UI yet (see PROJECT_STATE.md), granted for API access.
    "india_depreciation", "india_gst", "india_interest_penalty",
}

# Seeded once as global (firm_id NULL) built-in presets — see
# _seed_role_templates. Editable or deletable by a platform admin
# afterward; any firm can also add its own on top. (key, name,
# description, [module_key, ...]).
_DEFAULT_ROLE_TEMPLATES = [
    ("hr", "HR", "Manages job postings and applications.", ["career_admin"]),
    ("accountant", "Accountant", "Time & billing plus the core accounting engines.",
     ["billing", "fin_statements", "depreciation", "fixed_assets", "vat", "tds"]),
    # blog_admin deliberately excluded — the blog has no per-firm isolation
    # (blog_post/blog_category have no firm_id at all, unlike everything
    # else site_settings covers), so it's platform-admin-only regardless of
    # any grant (see _require_platform_admin on the blog routes in app.py).
    ("content_creator", "Content Creator", "Manages the landing-page chatbot's leads.",
     ["chatbot_admin"]),
    ("auditor", "Auditor", "Internal audit, statutory audit and audit execution.",
     ["internal_audit", "stat_audit", "audit_execution"]),
    ("tax_preparer", "Tax Preparer", "The tax-computation modules.",
     ["biz_tax", "personal_tax", "tds", "vat", "penalty"]),
    ("site_manager", "Site Manager", "Site settings — services, pages, firm profile, integrations.",
     ["site_settings"]),
]


def now_iso():
    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")


@contextmanager
def _sqlite_conn():
    os.makedirs(DATA_DIR, exist_ok=True)
    os.makedirs(STORAGE_DIR, exist_ok=True)
    c = sqlite3.connect(DB_PATH, timeout=20)
    c.row_factory = sqlite3.Row
    c.execute("PRAGMA foreign_keys = ON")
    c.execute("PRAGMA journal_mode = WAL")
    try:
        yield c
        c.commit()
    except Exception:
        c.rollback()
        raise
    finally:
        c.close()


@contextmanager
def _mysql_conn():
    raw = pymysql.connect(**MYSQL_CONFIG)
    c = _MySQLConnWrapper(raw)
    try:
        yield c
        c.commit()
    except Exception:
        c.rollback()
        raise
    finally:
        c.close()


def conn():
    return _mysql_conn() if DB_ENGINE == "mysql" else _sqlite_conn()


def query(sql, args=(), one=False):
    with conn() as c:
        cur = c.execute(sql, args)
        rows = [dict(r) for r in cur.fetchall()]
    return (rows[0] if rows else None) if one else rows


def execute(sql, args=()):
    with conn() as c:
        cur = c.execute(sql, args)
        return cur.lastrowid


_NEW_COLUMNS = {
    "client": {
        "vat_filing_frequency": "TEXT",     # monthly | four_monthly | six_monthly | not_registered
        "stage": "TEXT NOT NULL DEFAULT 'active'",
        "urgent": "INTEGER NOT NULL DEFAULT 0",
        "service_types": "TEXT",            # JSON array, e.g. ["audit","tax","valuation","fdi"]
        "assigned_to": "INTEGER REFERENCES app_user(id)",
        # A client's own jurisdiction — deliberately independent of
        # firm.country_code, not defaulted to it at the schema level (a
        # Nepal-based firm can still have an India- or UAE-incorporated
        # client). No DB-level default here on purpose; create_client sets
        # it to the firm's own country_code only as a starting suggestion,
        # never silently. Drives which country's engines a client's own
        # workfiles resolve to, same as firm.country_code does for the
        # firm-wide default.
        "country_code": "TEXT",
        "logo_path": "TEXT",
    },
    "time_entry": {
        "invoice_id": "INTEGER REFERENCES invoice(id) ON DELETE SET NULL",
        # Optional per-task hours, alongside the existing engagement-level
        # entries — nullable so every pre-existing row (engagement-level
        # only) is unaffected. "Actual hours" for a task is SUM(hours) WHERE
        # task_id = ?, computed at read time rather than a stored column on
        # task itself — same "capture once" convention as client_directory's
        # own roll-up.
        "task_id": "INTEGER REFERENCES task(id) ON DELETE SET NULL",
    },
    "task": {
        "parent_task_id": "INTEGER REFERENCES task(id) ON DELETE CASCADE",
        "reviewer_id": "INTEGER REFERENCES app_user(id) ON DELETE SET NULL",
        "estimated_hours": "REAL",
        # Same "notified once per day" convention as deadline.last_notified_at
        # — see notify.sweep_overdue_tasks(), which fires workflow_engine's
        # task.overdue event.
        "overdue_notified_at": "TEXT",
    },
    "app_user": {
        "is_super_admin": "INTEGER NOT NULL DEFAULT 0",
        # Platform-wide, not firm-scoped — CA-Automator's own operator role,
        # distinct from is_super_admin (the top user WITHIN one firm/tenant).
        # Only a platform admin can create a new firm, change its seat
        # limit, or change which modules its plan includes.
        "is_platform_admin": "INTEGER NOT NULL DEFAULT 0",
    },
    "appointment_booking": {
        "service": "TEXT",                  # which of the firm's own services the visitor wants to discuss
        "service_id": "INTEGER REFERENCES service(id) ON DELETE SET NULL",
        "client_id": "INTEGER REFERENCES client(id) ON DELETE SET NULL",   # set when booked from inside the client portal
    },
    "user_module_grant": {
        "expires_at": "TEXT",               # 'YYYY-MM-DD'; NULL = permanent, the pre-existing default behaviour
    },
    "engagement": {
        "scope": "TEXT NOT NULL DEFAULT 'standalone'",  # standalone | integrated
    },
    "lead": {
        "subject": "TEXT",
        "service_id": "INTEGER REFERENCES service(id) ON DELETE SET NULL",
        "assigned_to": "INTEGER REFERENCES app_user(id)",
        "follow_up_date": "TEXT",           # 'YYYY-MM-DD'
        "notes": "TEXT",                    # internal CRM notes, distinct from the visitor's own `message`
        "client_id": "INTEGER REFERENCES client(id) ON DELETE SET NULL",  # set once converted, or if a portal client raised this themselves
        # Which firm this inquiry belongs to — NULL on every pre-existing
        # row until _backfill_lead_firm_id assigns one. Without this, every
        # tenant firm's staff saw (and could even convert into their own
        # client!) every other firm's "Request access" signups, contact-form
        # messages and client-portal service requests.
        "firm_id": "INTEGER REFERENCES firm(id) ON DELETE CASCADE",
    },
    "document": {
        "visible_to_client": "INTEGER NOT NULL DEFAULT 0",  # staff opts a document into the client portal explicitly — never on by default
    },
    "service": {
        # The dedicated /services/<slug> page's extra sections — description_html
        # already covers "what we deliver"; these are the rest of the
        # Problem -> Solution -> Deliver -> Process -> Who it's for ->
        # Benefits -> Documents flow. All optional: a section only renders
        # on the public page when its field is actually filled in.
        # Despite the "_html" name (kept as-is to avoid a column rename/data
        # migration), every one of these — including description_html — is
        # plain text/Markdown from Admin's point of view: the admin editor
        # is a plain textarea, never a raw-HTML field, and server/public_pages.py
        # renders all of them through markdown_lite before they reach a
        # visitor. A firm typing an ordinary sentence in here (which is all
        # that existed in these columns before this comment was written) has
        # never had to know what HTML is.
        "problem_html": "TEXT",
        "process_html": "TEXT",
        "who_for_html": "TEXT",
        "benefits_html": "TEXT",
        "documents_html": "TEXT",
    },
    "page": {
        # content_md is plain text/Markdown, rendered through markdown_lite —
        # the admin-facing replacement for content_html, which stays as a
        # column (nobody types into it anymore, but a page saved before this
        # existed keeps rendering exactly as it always has: raw HTML,
        # untouched, since reinterpreting real markup as Markdown would
        # mangle it rather than help).
        "content_md": "TEXT",
    },
    "firm": {
        # Jurisdiction the firm operates under — drives which rate_master /
        # tp_regulatory_rule rows the country-aware engines read (see
        # country_service.py). 'NP' keeps every existing installation's
        # behaviour identical to today; multi-country firms can add more
        # jurisdictions later without this column changing shape.
        "country_code": "TEXT NOT NULL DEFAULT 'NP'",
        "tagline": "TEXT",
        "google_maps_url": "TEXT",
        "google_maps_embed": "TEXT",
        "office_hours": "TEXT",
        "facebook_url": "TEXT",
        "linkedin_url": "TEXT",
        "twitter_url": "TEXT",
        "instagram_url": "TEXT",
        # SaaS seat licensing — how many active staff accounts this firm's
        # plan allows (enforced in admin_create_user), and an informational
        # label for which plan that number corresponds to. is_active lets a
        # platform admin suspend a tenant without deleting its data.
        "max_seats": "INTEGER NOT NULL DEFAULT 5",
        "plan_label": "TEXT",
        "is_active": "INTEGER NOT NULL DEFAULT 1",
        # Term subscription lifecycle. NULL = perpetual (no expiry at all —
        # what every pre-existing/grandfathered firm gets, so nothing changes
        # for them). Once set, a firm gets GRACE_DAYS (app.py) past this date
        # before staff access is paused; the *_notified_at columns make each
        # transition notice (see notify.sweep_subscriptions) fire exactly
        # once instead of every worker cycle. Client data is never touched by
        # any of this — suspension only affects whether staff can sign in to
        # use the app.
        "subscription_expires_at": "TEXT",
        "subscription_grace_notified_at": "TEXT",
        "subscription_suspended_notified_at": "TEXT",
        # "Our Clients" showcase on the main public landing page — a tenant
        # firm's own business registration number (collected at onboarding
        # alongside the pre-existing pan/phone/logo_path), and a platform-
        # admin-controlled on/off switch for whether this firm's name+logo
        # appears in that rolling strip. Defaults to 0 (off) — a platform
        # admin opts a firm in explicitly rather than every signup
        # auto-appearing on the homepage. The site's own default firm
        # (_default_firm_id()) is excluded from the showcase query
        # regardless of this flag — a firm doesn't showcase itself as its
        # own client.
        "registration_no": "TEXT",
        "show_on_landing": "INTEGER NOT NULL DEFAULT 0",
    },
    # Every existing row is Nepal rate data; defaulting to 'NP' preserves
    # today's lookups (db.rates/db.rate_dict) exactly. A future country's
    # rates get their own rows scoped by this column, following the same
    # pattern tp_regulatory_rule already uses for Transfer Pricing.
    "rate_master": {
        "country_code": "TEXT NOT NULL DEFAULT 'NP'",
    },
    # Gives each team member their own public URL (/team/<slug>) — see
    # _backfill_team_member_slugs for the one-time backfill on existing
    # rows and the UNIQUE index it creates once every row has one.
    "team_member": {
        # VARCHAR, not bare TEXT — this gets a UNIQUE index once every
        # existing row has one (see _backfill_team_member_slugs), and MySQL
        # can't index a bare TEXT/BLOB column without a prefix length.
        # Identical "TEXT affinity" behavior in SQLite either way.
        "slug": "VARCHAR(100)",
    },
}


def _create_index_if_not_exists(c, index_sql):
    """A handful of one-time migration functions create a UNIQUE index only
    after backfilling/deduping the data it will enforce — "CREATE ... INDEX
    IF NOT EXISTS" on SQLite, run as authored; MySQL has no IF NOT EXISTS
    for indexes at all, so strip it and treat "duplicate key name" as the
    same already-applied no-op schema_mysql.sql's own loader does."""
    if DB_ENGINE != "mysql":
        c.execute(index_sql)
        return
    stripped = index_sql.replace(" IF NOT EXISTS", "")
    try:
        c.execute(stripped)
    except pymysql.err.OperationalError as e:
        if not (e.args and e.args[0] == 1061):
            raise


def _mysql_column_decl(decl):
    """_NEW_COLUMNS declarations are written once, in SQLite syntax, and
    used for both engines' ALTER TABLE ADD COLUMN — same translation rules
    schema_mysql.sql's own generator applies to the initial CREATE TABLEs
    (see convert_schema.py): INTEGER/REAL -> MySQL's native names, and a
    literal-string-defaulted TEXT column (MySQL can't default a TEXT/BLOB
    column at all) or a datetime('now')-defaulted one widened/retyped the
    same way. A bare, default-less TEXT column needs no change."""
    decl = decl.replace("TEXT NOT NULL DEFAULT (datetime('now'))", "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP")
    decl = re.sub(r"\bTEXT\b(?=(\s+NOT NULL)?\s+DEFAULT\s+')", "VARCHAR(100)", decl)
    decl = re.sub(r"\bINTEGER\b", "INT", decl)
    decl = re.sub(r"\bREAL\b", "DOUBLE", decl)
    return decl


def _migrate_columns(c):
    """schema.sql's CREATE TABLE IF NOT EXISTS never adds a column to a table
    that already exists, so a column added after the firm already has data has
    to go through ALTER TABLE — which neither engine makes idempotent either,
    hence the existing-columns check rather than a bare try/except swallowing
    every OperationalError."""
    for table, cols in _NEW_COLUMNS.items():
        if DB_ENGINE == "mysql":
            existing = {row["COLUMN_NAME"] for row in c.execute(
                "SELECT COLUMN_NAME FROM information_schema.columns"
                " WHERE table_schema = DATABASE() AND table_name = ?", (table,)).fetchall()}
        else:
            existing = {row["name"] for row in c.execute(f"PRAGMA table_info({table})").fetchall()}
        for col, decl in cols.items():
            if col not in existing:
                if DB_ENGINE == "mysql":
                    # `table` backtick-quoted: "lead" collides with the
                    # SQL:2003 LEAD() window function (see schema_mysql.sql
                    # generation notes) — harmless to quote every table name
                    # here, not just that one.
                    c.execute(f"ALTER TABLE `{table}` ADD COLUMN {col} {_mysql_column_decl(decl)}")
                else:
                    c.execute(f"ALTER TABLE {table} ADD COLUMN {col} {decl}")


def _ensure_super_admin(c):
    """Every firm must always have at least one active super admin, or nobody
    can reach the Admin screen to fix it. Adding is_super_admin defaults every
    existing user to 0, which would zero out an install that predates this
    column — so on each boot, any firm with zero active super admins gets its
    earliest-created active user promoted automatically. A no-op once a real
    promotion/demotion has happened, since that firm then has exactly one (or
    more)."""
    firms = c.execute("SELECT DISTINCT firm_id FROM app_user").fetchall()
    for row in firms:
        firm_id = row[0]
        has_admin = c.execute(
            "SELECT 1 FROM app_user WHERE firm_id = ? AND is_super_admin = 1 AND is_active = 1 LIMIT 1",
            (firm_id,)).fetchone()
        if has_admin:
            continue
        earliest = c.execute(
            "SELECT id FROM app_user WHERE firm_id = ? AND is_active = 1 ORDER BY created_at LIMIT 1",
            (firm_id,)).fetchone()
        if earliest:
            c.execute("UPDATE app_user SET is_super_admin = 1 WHERE id = ?", (earliest[0],))


def _ensure_platform_admin(c):
    """Same reasoning as _ensure_super_admin, one level up: is_platform_admin
    is a brand-new column defaulting every existing user to 0, which would
    leave a system with no one able to reach Platform Admin (create a firm,
    set seat limits, grant firm-level module entitlements). If nobody
    anywhere is a platform admin yet, the earliest-created active super
    admin across all firms is promoted. A no-op forever after that."""
    if c.execute("SELECT 1 FROM app_user WHERE is_platform_admin = 1 LIMIT 1").fetchone():
        return
    earliest = c.execute(
        "SELECT id FROM app_user WHERE is_super_admin = 1 AND is_active = 1 ORDER BY created_at LIMIT 1"
    ).fetchone()
    if earliest:
        c.execute("UPDATE app_user SET is_platform_admin = 1 WHERE id = ?", (earliest[0],))


def _sanitize_maps_embed(c):
    """One-time cleanup for firm.google_maps_embed rows saved before
    app.py's admin_update_firm_profile started validating this field. The
    Contact section drops this value straight into an <iframe src="…">
    (web/landing.js renderContactInfo) — a row saved as a chunk of raw HTML
    from a third-party "map embed generator" (found in this app's own
    production data, complete with a hidden tracking backlink to another
    site) never renders a map and ships that markup to every site visitor.
    Extracts the real src="…" URL out of any such row when there is one, or
    clears the field when there isn't — never leaves it as unescaped HTML."""
    rows = c.execute("SELECT id, google_maps_embed FROM firm WHERE google_maps_embed LIKE '%<%'").fetchall()
    for firm_id, raw in rows:
        match = re.search(r'src=["\']([^"\']+)["\']', raw or "", re.IGNORECASE)
        clean = match.group(1) if match else None
        if clean and not re.match(r'^https://(www\.)?(google\.[a-z.]+/maps|maps\.google\.[a-z.]+)/', clean, re.IGNORECASE):
            clean = None
        c.execute("UPDATE firm SET google_maps_embed = ? WHERE id = ?", (clean, firm_id))


def _backfill_team_member_slugs(c):
    """team_member.slug is new — every row that predates it (or was created
    without one) needs one before the unique index below can be created.
    Idempotent: only touches rows with no slug yet, and the unique index
    creation is itself a no-op once it exists."""
    rows = c.execute("SELECT id, name FROM team_member WHERE slug IS NULL OR slug = ''").fetchall()
    if rows:
        existing = {r[0] for r in c.execute("SELECT slug FROM team_member WHERE slug IS NOT NULL AND slug != ''").fetchall()}
        for member_id, name in rows:
            base = re.sub(r"[^a-z0-9]+", "-", (name or "").strip().lower()).strip("-") or f"team-{member_id}"
            slug, n = base, 2
            while slug in existing:
                slug = f"{base}-{n}"; n += 1
            existing.add(slug)
            c.execute("UPDATE team_member SET slug = ? WHERE id = ?", (slug, member_id))
    _create_index_if_not_exists(c, "CREATE UNIQUE INDEX IF NOT EXISTS ux_team_member_slug ON team_member(slug)")


def _backfill_lead_firm_id(c):
    """lead.firm_id is new — every row that predates it belongs to whichever
    firm it actually came from: a client-portal service request (client_id
    set) belongs to that client's own firm; everything else (public
    "Request access"/contact-form signups, chatbot conversions) belongs to
    the site's own default firm, the only firm that has ever operated the
    public marketing site. Idempotent — only touches rows still NULL."""
    default_id = c.execute("SELECT id FROM firm ORDER BY id LIMIT 1").fetchone()
    if not default_id:
        return
    default_id = default_id[0]
    for lead_id, client_id in c.execute("SELECT id, client_id FROM `lead` WHERE firm_id IS NULL").fetchall():
        firm_id = default_id
        if client_id:
            row = c.execute("SELECT firm_id FROM client WHERE id = ?", (client_id,)).fetchone()
            if row:
                firm_id = row[0]
        c.execute("UPDATE `lead` SET firm_id = ? WHERE id = ?", (firm_id, lead_id))


def _grandfather_firm_module_grants(c):
    """firm_module_grant is a new ceiling on what a firm's own super admin
    can grant its staff — without this, every existing firm would suddenly
    be entitled to nothing, breaking every module grant already in use.
    Any firm with zero rows here (never had entitlements set, i.e. every
    firm that existed before this feature) gets every module its own staff
    already actually use granted at the firm level, so nothing already
    working breaks. A brand new firm a platform admin creates going
    forward starts with zero rows here on purpose — a deliberate choice,
    not an oversight — until the platform admin actually grants something."""
    firms = c.execute("SELECT id FROM firm").fetchall()
    for row in firms:
        firm_id = row[0]
        if c.execute("SELECT 1 FROM firm_module_grant WHERE firm_id = ? LIMIT 1", (firm_id,)).fetchone():
            continue
        keys = set(k[0] for k in c.execute(
            "SELECT DISTINCT g.module_key FROM user_module_grant g JOIN app_user u ON u.id = g.user_id"
            " WHERE u.firm_id = ?", (firm_id,)).fetchall())
        keys |= ALL_GRANTABLE_MODULE_KEYS
        for key in keys:
            c.execute("INSERT OR IGNORE INTO firm_module_grant (firm_id, module_key) VALUES (?, ?)", (firm_id, key))


def _grant_new_module_to_every_firm(c, module_key):
    """_grandfather_firm_module_grants only back-fills a firm that has ZERO
    firm_module_grant rows at all (the one-time old->new transition) — a
    firm that already has some rows (every real firm by now) never
    automatically gets a module_key introduced after that transition, so
    org_admin would 403 for everyone but a platform admin without this.
    INSERT OR IGNORE keeps it idempotent and never touches a firm that has
    already had this key explicitly revoked."""
    for row in c.execute("SELECT id FROM firm").fetchall():
        c.execute("INSERT OR IGNORE INTO firm_module_grant (firm_id, module_key) VALUES (?, ?)",
                 (row[0], module_key))


def _seed_role_templates(c):
    """Global, built-in starting presets — a platform admin can edit or
    remove these, and any firm can add its own on top (role_template.firm_id
    set). INSERT OR IGNORE on (a made-up-for-this key uniqueness check)
    isn't available since key isn't declared UNIQUE (a firm's own custom
    template might reuse a builtin's key on purpose), so this checks
    explicitly instead — idempotent the same way _ensure_email_templates is,
    just without a DB-level constraint to lean on."""
    for key, name, description, grants in _DEFAULT_ROLE_TEMPLATES:
        row = c.execute(
            "SELECT module_grants FROM role_template WHERE firm_id IS NULL AND `key` = ?", (key,)).fetchone()
        if row is None:
            c.execute(
                "INSERT INTO role_template (firm_id, `key`, name, description, module_grants, is_builtin)"
                " VALUES (NULL, ?, ?, ?, ?, 1)",
                (key, name, description, json.dumps(grants)))
            continue
        # A builtin already seeded before blog_admin was locked down
        # platform-admin-only (see app.py's blog routes) may still carry it
        # in an already-live database — drop it so the checkbox stops
        # promising access that's now always refused. Only ever removes
        # blog_admin, never touches anything else a platform admin may have
        # customized on this builtin since.
        try:
            existing_grants = json.loads(row["module_grants"] or "[]")
        except (TypeError, ValueError):
            existing_grants = []
        if "blog_admin" in existing_grants:
            c.execute("UPDATE role_template SET module_grants = ? WHERE firm_id IS NULL AND `key` = ?",
                     (json.dumps([g for g in existing_grants if g != "blog_admin"]), key))


_DEFAULT_EMAIL_TEMPLATES = {
    "career_received": (
        "We've received your application — {{position}}",
        "Dear {{name}},\n\nThank you for applying for {{position}} at {{firm_name}}. "
        "We've received your application and will be in touch as it moves through review.\n\n— {{firm_name}}"),
    "career_screening": (
        "Your application is being screened — {{position}}",
        "Dear {{name}},\n\nYour application for {{position}} is now being screened. "
        "We'll update you as soon as there's a decision on the next step.\n\n— {{firm_name}}"),
    "career_under_review": (
        "Your application is under review — {{position}}",
        "Dear {{name}},\n\nYour application for {{position}} is currently under review. "
        "Thank you for your patience — we'll be in touch shortly.\n\n— {{firm_name}}"),
    "career_shortlisted": (
        "You've been shortlisted — {{position}}",
        "Dear {{name}},\n\nCongratulations — you've been shortlisted for {{position}}. "
        "We'll be in touch soon with next steps.\n\n— {{firm_name}}"),
    "career_interview": (
        "Interview invitation — {{position}}",
        "Dear {{name}},\n\nWe'd like to invite you to interview for {{position}}. "
        "Someone from our team will reach out shortly to arrange a time.\n\n— {{firm_name}}"),
    "career_selected": (
        "You've been selected — {{position}}",
        "Dear {{name}},\n\nWe are pleased to inform you that your application for {{position}} has been "
        "selected. We'll be in touch shortly with next steps.\n\nCongratulations, and welcome.\n\n— {{firm_name}}"),
    "career_rejected": (
        "Update on your application — {{position}}",
        "Dear {{name}},\n\nThank you for applying for {{position}} and for the time you invested in your "
        "application. After careful consideration, we will not be moving forward on this occasion. "
        "We wish you the best in your search.\n\n— {{firm_name}}"),
}


def _ensure_email_templates(c):
    """Seeds the fixed set of well-known template keys idempotently on every
    boot (INSERT OR IGNORE — an admin's own edit is never overwritten), so an
    upgrade never leaves a career-stage email with no template to send."""
    for key, (subject, body) in _DEFAULT_EMAIL_TEMPLATES.items():
        c.execute("INSERT OR IGNORE INTO email_template (`key`, subject, body) VALUES (?,?,?)", (key, subject, body))


_DEFAULT_MENU_ITEMS = {
    "primary": [
        ("Home", "#platform"), ("About", "#about"), ("Services", "#services"),
        ("CA-Automator", "#modules"), ("Insights", "/blog"), ("Careers", "/careers"),
        ("Downloads", "/downloads"), ("Contact", "#contact"),
    ],
    "footer": [
        ("Insights", "/blog"), ("Careers", "/careers"), ("Downloads", "/downloads"), ("Contact", "#contact"),
    ],
}


def _seed_default_menu(c, firm_id=1):
    """Seeded once, only if this firm has no menu rows at all — an admin
    who deliberately clears a section down to nothing (wants a bare-bones
    nav) never has it silently repopulated on the next server restart,
    unlike _ensure_email_templates' INSERT OR IGNORE (there, missing a
    well-known template key is always a bug to heal; an empty nav section
    might be exactly what was intended)."""
    n = c.execute("SELECT COUNT(*) FROM nav_menu_item WHERE firm_id = ?", (firm_id,)).fetchone()[0]
    if n:
        return
    for section, items in _DEFAULT_MENU_ITEMS.items():
        for i, (label, url) in enumerate(items):
            c.execute("INSERT INTO nav_menu_item (firm_id, section, label, url, sort_order) VALUES (?,?,?,?,?)",
                     (firm_id, section, label, url, i))


_DEFAULT_JOB_POSITIONS = [
    # (name, hierarchy_level) — lower level = more senior. This is a
    # starting skeleton, not a fixed enum: a firm can rename, add or delete
    # any of these from the Positions screen. Never hard-coded into engine
    # logic — this table only ever feeds employee_profile.job_position_id
    # and, later, seniority/approval-authority comparisons via
    # hierarchy_level, exactly the "configurable, not hard-coded" hierarchy
    # the org-structure design calls for.
    ("Managing Partner", 0), ("Partner", 1), ("Director", 2), ("Manager", 3),
    ("Assistant Manager", 4), ("Senior", 5), ("Associate", 6),
    ("Assistant", 7), ("Intern", 8),
]


def _seed_default_job_positions(c):
    """Same tradeoff as _seed_default_menu, not _ensure_email_templates:
    seeded once per firm, only if that firm has zero job_position rows at
    all. Renaming or deleting a builtin position is an explicitly supported
    admin action (unlike an email template's key), so an INSERT OR IGNORE
    keyed on name would silently re-create "Partner" every boot after an
    admin renamed it to "Senior Partner" — seeding only into a genuinely
    empty table avoids that."""
    for firm_id in [r[0] for r in c.execute("SELECT id FROM firm").fetchall()]:
        n = c.execute("SELECT COUNT(*) FROM job_position WHERE firm_id = ?", (firm_id,)).fetchone()[0]
        if n:
            continue
        for name, level in _DEFAULT_JOB_POSITIONS:
            c.execute(
                "INSERT INTO job_position (firm_id, name, hierarchy_level, is_builtin) VALUES (?,?,?,1)",
                (firm_id, name, level))


def _sync_current_fiscal_year(c):
    """Whichever fiscal_year row's date range actually covers today becomes
    is_current — recomputed on every boot instead of trusting a flag someone
    has to remember to flip at year-end. A year not seeded yet (today falls
    outside every row's range) leaves the flag as it was rather than
    clearing it, so there's always a fallback 'current' row to point at."""
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    row = c.execute(
        "SELECT id FROM fiscal_year WHERE start_ad <= ? AND end_ad >= ? LIMIT 1", (today, today)
    ).fetchone()
    if row:
        c.execute("UPDATE fiscal_year SET is_current = 0")
        c.execute("UPDATE fiscal_year SET is_current = 1 WHERE id = ?", (row[0],))


def _fix_appointment_booking_slot_constraint(c):
    """The first cut of the appointments feature declared slot_id UNIQUE —
    which SQLite enforces unconditionally, so a single cancelled booking
    permanently blocked that slot from ever being booked again (the exact
    bug a real booking attempt hit in testing: 'UNIQUE constraint failed:
    appointment_booking.slot_id' on a slot whose only prior booking had
    already been cancelled). schema.sql now creates the table without that
    constraint and adds a partial unique index instead (only NON-cancelled
    bookings need be unique per slot) — but SQLite's CREATE TABLE IF NOT
    EXISTS never retrofits an already-existing table, so an install that
    already has the old appointment_booking table needs an explicit
    rebuild. Detected by checking for the old column-level UNIQUE's
    auto-generated index rather than a version flag, so this is safe to
    run on every boot and a true no-op once the fix has applied."""
    tables = {row[0] for row in c.execute(
        "SELECT name FROM sqlite_master WHERE type='table' AND name='appointment_booking'").fetchall()}
    if not tables:
        return
    has_old_constraint = False
    for idx in c.execute("PRAGMA index_list('appointment_booking')").fetchall():
        if not idx["unique"] or idx["origin"] != "u":
            continue
        cols = [r["name"] for r in c.execute(f"PRAGMA index_info('{idx['name']}')").fetchall()]
        if cols == ["slot_id"]:
            has_old_constraint = True
            break
    if not has_old_constraint:
        return
    c.execute("""
        CREATE TABLE appointment_booking_rebuild (
            id             INTEGER PRIMARY KEY,
            slot_id        INTEGER NOT NULL REFERENCES appointment_slot(id) ON DELETE CASCADE,
            firm_id        INTEGER NOT NULL REFERENCES firm(id) ON DELETE CASCADE,
            name           TEXT NOT NULL,
            email          TEXT NOT NULL,
            phone          TEXT,
            company        TEXT,
            method         TEXT NOT NULL DEFAULT 'video',
            message        TEXT,
            price          REAL NOT NULL,
            status         TEXT NOT NULL DEFAULT 'pending',
            cancel_reason  TEXT,
            created_at     TEXT NOT NULL DEFAULT (datetime('now')),
            service        TEXT
        )
    """)
    existing_cols = {row["name"] for row in c.execute("PRAGMA table_info(appointment_booking)").fetchall()}
    copy_cols = [col for col in ("id", "slot_id", "firm_id", "name", "email", "phone", "company", "method",
                                  "message", "price", "status", "cancel_reason", "created_at", "service")
                 if col in existing_cols]
    cols_sql = ", ".join(copy_cols)
    c.execute(f"INSERT INTO appointment_booking_rebuild ({cols_sql}) SELECT {cols_sql} FROM appointment_booking")
    c.execute("DROP TABLE appointment_booking")
    c.execute("ALTER TABLE appointment_booking_rebuild RENAME TO appointment_booking")
    c.execute("CREATE UNIQUE INDEX IF NOT EXISTS ux_appt_booking_active_slot ON appointment_booking(slot_id) WHERE status != 'cancelled'")
    c.execute("CREATE INDEX IF NOT EXISTS ix_appt_booking_firm ON appointment_booking(firm_id, status)")


def _fix_document_report_client_delete(c):
    """document.client_id/engagement_id and report.engagement_id used to be
    ON DELETE CASCADE, so deleting a client silently deleted every document
    and report that had ever been generated for them. schema.sql now declares
    both as ON DELETE SET NULL — a document or report is only ever removed by
    an explicit delete action, never as a side effect of deleting the client
    or engagement it happened to be filed under. CREATE TABLE IF NOT EXISTS
    never retrofits an already-existing table, so an install with the old
    cascading FK needs an explicit rebuild, detected via PRAGMA foreign_key_list
    rather than a version flag — a true no-op once the fix has applied."""
    def has_cascade(table, fk_from):
        if not c.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table,)).fetchone():
            return False
        for fk in c.execute(f"PRAGMA foreign_key_list('{table}')").fetchall():
            if fk["from"] == fk_from and fk["on_delete"] == "CASCADE":
                return True
        return False

    if has_cascade("document", "client_id") or has_cascade("document", "engagement_id"):
        c.execute("""
            CREATE TABLE document_rebuild (
                id            INTEGER PRIMARY KEY,
                engagement_id INTEGER REFERENCES engagement(id) ON DELETE SET NULL,
                client_id     INTEGER REFERENCES client(id) ON DELETE SET NULL,
                category      TEXT,
                filename      TEXT NOT NULL,
                mime          TEXT,
                byte_size     INTEGER,
                sha256        TEXT,
                storage_path  TEXT NOT NULL,
                uploaded_by   INTEGER REFERENCES app_user(id),
                uploaded_at   TEXT NOT NULL DEFAULT (datetime('now'))
            )
        """)
        cols = [r["name"] for r in c.execute("PRAGMA table_info(document)").fetchall()]
        cols_sql = ", ".join(cols)
        c.execute(f"INSERT INTO document_rebuild ({cols_sql}) SELECT {cols_sql} FROM document")
        c.execute("DROP TABLE document")
        c.execute("ALTER TABLE document_rebuild RENAME TO document")
        c.execute("CREATE INDEX IF NOT EXISTS ix_doc_eng ON document(engagement_id)")

    if has_cascade("report", "engagement_id"):
        c.execute("""
            CREATE TABLE report_rebuild (
                id            INTEGER PRIMARY KEY,
                workfile_id   INTEGER REFERENCES workfile(id) ON DELETE SET NULL,
                engagement_id INTEGER REFERENCES engagement(id) ON DELETE SET NULL,
                module_key    TEXT NOT NULL,
                format        TEXT NOT NULL,
                template_id   INTEGER REFERENCES report_template(id),
                filename      TEXT NOT NULL,
                storage_path  TEXT,
                params        TEXT,
                status        TEXT NOT NULL DEFAULT 'ready',
                error_text    TEXT,
                generated_by  INTEGER REFERENCES app_user(id),
                generated_at  TEXT NOT NULL DEFAULT (datetime('now')),
                byte_size     INTEGER
            )
        """)
        cols = [r["name"] for r in c.execute("PRAGMA table_info(report)").fetchall()]
        cols_sql = ", ".join(cols)
        c.execute(f"INSERT INTO report_rebuild ({cols_sql}) SELECT {cols_sql} FROM report")
        c.execute("DROP TABLE report")
        c.execute("ALTER TABLE report_rebuild RENAME TO report")
        c.execute("CREATE INDEX IF NOT EXISTS ix_report_eng ON report(engagement_id, generated_at DESC)")


def _dedupe_audit_criteria(c):
    """audit_criterion had no uniqueness constraint, and seed.sql's INSERT
    OR IGNORE had nothing to conflict against — so if seed.sql ever ran
    against a database that already had these ~102 standard checklist
    items (e.g. the data file was recreated more than once during setup,
    each time looking "fresh"), the whole bank piled up duplicated: this
    firm's real database had it 7x over (714 rows for 102 distinct
    checklist items) before this fix. Internal Audit and Statutory Audit
    recompute their checklist from this table fresh on every request
    rather than caching a specific row's id anywhere durable except a
    workfile's own criteria_responses (keyed by criterion id) — confirmed
    against this firm's actual workfiles that none had any recorded
    responses yet, so keeping the lowest id per (area, criterion) and
    dropping the rest is safe here. The unique index created below makes
    this permanently a one-time fix: once applied, seed.sql's INSERT OR
    IGNORE has something to actually conflict against, so duplicates can't
    accumulate again on any future reseed."""
    if DB_ENGINE == "mysql":
        # MySQL rejects a DELETE whose subquery reads the same table it's
        # deleting from ("Error 1093: You can't specify target table ... for
        # update in FROM clause") — SQLite has no such restriction. Wrapping
        # the subquery in a derived table sidesteps it: MySQL materializes
        # the inner SELECT before the DELETE runs, so it's no longer reading
        # "live" from the table being modified.
        c.execute("""
            DELETE FROM audit_criterion
            WHERE id NOT IN (SELECT id FROM (SELECT MIN(id) id FROM audit_criterion GROUP BY area, criterion) keep)
        """)
    else:
        c.execute("""
            DELETE FROM audit_criterion
            WHERE id NOT IN (SELECT MIN(id) FROM audit_criterion GROUP BY area, criterion)
        """)
    # criterion is real checklist sentence text (unbounded on purpose, so
    # nothing here truncates a preparer-visible label) — MySQL can't index a
    # bare TEXT column without a prefix length, unlike area (already
    # VARCHAR — see schema_mysql.sql's own generation, which widens it via
    # the static ix_audit_criterion_area index already in schema.sql). A
    # 255-char prefix is dedup-equivalent in practice (no two distinct
    # checklist criteria in this app share a 255-character prefix) without
    # having to shrink the column itself.
    criterion_key = "criterion(255)" if DB_ENGINE == "mysql" else "criterion"
    _create_index_if_not_exists(c, "CREATE UNIQUE INDEX IF NOT EXISTS ux_audit_criterion_area_criterion"
                                   f" ON audit_criterion(area, {criterion_key})")


def _fix_rate_master_dedup(c):
    """Same bug, same fix, as _dedupe_audit_criteria just above — rate_master's
    original UNIQUE (domain, code, fiscal_year_label, applies_to) never
    dedupes a standing rule (NULL fiscal_year_label — the convention every
    India row uses, see seed.sql) against itself, because SQL treats every
    NULL as distinct from every other NULL for uniqueness purposes. It also
    never included country_code. Confirmed live in a rehearsal database:
    init_db(force_seed=True) doubled every India row before this fix
    existed. Dedupes first (keeping the lowest id, same as
    _dedupe_audit_criteria — rate_master rows are looked up by domain/code/
    country/fy/applies_to, never cached by a specific row id anywhere
    durable), then creates the corrected expression index so future
    reseeds can't duplicate again."""
    if DB_ENGINE == "mysql":
        c.execute("""
            DELETE FROM rate_master
            WHERE id NOT IN (SELECT id FROM (
                SELECT MIN(id) id FROM rate_master GROUP BY domain, code, country_code, fiscal_year_label, applies_to
            ) keep)
        """)
        _create_index_if_not_exists(c, "CREATE UNIQUE INDEX IF NOT EXISTS ux_rate_master_dedup ON rate_master"
                                       "(domain, code, country_code, (COALESCE(fiscal_year_label, '')), (COALESCE(applies_to, '')))")
    else:
        c.execute("""
            DELETE FROM rate_master
            WHERE id NOT IN (SELECT MIN(id) FROM rate_master GROUP BY domain, code, country_code, fiscal_year_label, applies_to)
        """)
        _create_index_if_not_exists(c, "CREATE UNIQUE INDEX IF NOT EXISTS ux_rate_master_dedup ON rate_master"
                                       "(domain, code, country_code, COALESCE(fiscal_year_label, ''), COALESCE(applies_to, ''))")


def init_db(force_seed=False):
    """Create the schema and load seed data if the database is empty."""
    if DB_ENGINE == "mysql":
        with conn() as c:
            fresh = not c.execute(
                "SELECT 1 FROM information_schema.tables"
                " WHERE table_schema = DATABASE() AND table_name = 'firm'").fetchone()
    else:
        fresh = not os.path.exists(DB_PATH)
    os.makedirs(DATA_DIR, exist_ok=True)
    with conn() as c:
        schema_file = "schema_mysql.sql" if DB_ENGINE == "mysql" else "schema.sql"
        with open(os.path.join(BASE_DIR, schema_file), encoding="utf-8") as f:
            c.executescript(f.read())
        _migrate_columns(c)
        if DB_ENGINE != "mysql":
            # Both are one-time SQLite table-rebuild workarounds for
            # constraints SQLite can't ALTER TABLE into an existing table —
            # schema_mysql.sql already has the correct final shape (see
            # convert_schema.py's appointment_booking generated-column
            # note), so MySQL never had the "old" constraint to fix.
            _fix_appointment_booking_slot_constraint(c)
            _fix_document_report_client_delete(c)
        _dedupe_audit_criteria(c)
        _fix_rate_master_dedup(c)
        _sanitize_maps_embed(c)
        _backfill_team_member_slugs(c)
        _backfill_lead_firm_id(c)
        if fresh or force_seed:
            with open(os.path.join(BASE_DIR, "seed.sql"), encoding="utf-8") as f:
                c.executescript(f.read())
        _sync_current_fiscal_year(c)
        _ensure_super_admin(c)
        _ensure_platform_admin(c)
        _grandfather_firm_module_grants(c)
        _grant_new_module_to_every_firm(c, "org_admin")
        for key in ("india_depreciation", "india_gst", "india_interest_penalty",
                    "uae_corporate_tax", "uae_penalties"):
            _grant_new_module_to_every_firm(c, key)
        _seed_role_templates(c)
        _ensure_email_templates(c)
        _seed_default_menu(c)
        _seed_default_job_positions(c)
    return fresh


def current_fy_label():
    """Today's fiscal year label, computed from the fiscal_year table's own
    date ranges (falling back to the mid-July heuristic if today isn't
    covered by any row yet) — never a hardcoded '2082-83'-style literal, so
    this doesn't need a code change every Shrawan."""
    fys = query("SELECT * FROM fiscal_year")
    return nepal_calendar.fy_label_for_ad(now_iso()[:10], fys)


def sync_current_fiscal_year():
    """Public entry point for _sync_current_fiscal_year — recomputes which
    row is_current right after a new fiscal_year is added, so a firm doesn't
    have to restart the server for a freshly-added year to take effect."""
    with conn() as c:
        _sync_current_fiscal_year(c)


def create_fiscal_year(label_bs, start_bs, end_bs, start_ad, end_ad):
    """Add a new fiscal year — BS and AD start/end are both required rather
    than derived from one or the other, since the approximate BS<->AD
    converter elsewhere in this app is explicitly non-authoritative and a
    fiscal year's official dates should come from the real calendar, not a
    guess. Recomputes is_current immediately so the new row takes effect
    without a restart if today already falls inside it."""
    with conn() as c:
        cur = c.execute(
            "INSERT INTO fiscal_year (label_bs, start_bs, end_bs, start_ad, end_ad) VALUES (?,?,?,?,?)",
            (label_bs, start_bs, end_bs, start_ad, end_ad))
        fid = cur.lastrowid
        _sync_current_fiscal_year(c)
    return query("SELECT * FROM fiscal_year WHERE id = ?", (fid,), one=True)


def rollover_rates(from_fy, to_fy):
    """Copy every rate_master row scoped to from_fy into to_fy, always as a
    fresh UNVERIFIED placeholder — even a row verified last year needs its
    own confirmation for the new year, since a Finance Act can change a
    figure that happened to stay put the year before. Rows already present
    for to_fy are left untouched (INSERT OR IGNORE against the table's own
    (domain, code, fiscal_year_label, applies_to) uniqueness). Returns the
    number of rows actually inserted."""
    rows = query("SELECT * FROM rate_master WHERE fiscal_year_label = ?", (from_fy,))
    inserted = 0
    with conn() as c:
        for r in rows:
            note = (r["notes"] or "").strip()
            carry_note = f"Carried forward from FY {from_fy} — confirm against the FY {to_fy} Finance Act before relying on it."
            note = (note + " " + carry_note) if note else carry_note
            cur = c.execute(
                "INSERT OR IGNORE INTO rate_master (domain, code, label, value_num, value_text, unit,"
                " applies_to, fiscal_year_label, source, clause, verification_status, notes)"
                " VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
                (r["domain"], r["code"], r["label"], r["value_num"], r["value_text"], r["unit"],
                 r["applies_to"], to_fy, r["source"], r["clause"], "UNVERIFIED", note))
            if cur.rowcount:
                inserted += 1
    return inserted


# ----------------------------------------------------------------- passwords

def hash_password(password, salt=None):
    salt = salt or secrets.token_hex(16)
    h = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), 200_000)
    return h.hex(), salt


def verify_password(password, pwd_hash, salt):
    calc, _ = hash_password(password, salt)
    return secrets.compare_digest(calc, pwd_hash)


# ------------------------------------------------------ credential encryption
# Portal passwords (IRD/OCR/excise logins) have to be *recoverable*, unlike a
# user's own login password, so hashing doesn't apply — this is reversible
# encryption at rest. Standard library only (no cryptography/pycryptodome
# dependency), so the cipher is HMAC-SHA256 used as a keystream generator in
# counter mode: a well-understood way to build a stream cipher from a PRF,
# though it hasn't had the scrutiny a vetted AEAD library has. The key is a
# random 32-byte value generated on first use and kept in its own file on
# disk — deliberately NOT in the `setting` table, since that table is read
# wholesale into GET /api/bootstrap and would otherwise ship the key to every
# logged-in browser.

_CRED_KEY_PATH = os.path.join(DATA_DIR, "cred_enc.key")


def _cred_key():
    os.makedirs(DATA_DIR, exist_ok=True)
    if not os.path.exists(_CRED_KEY_PATH):
        key = secrets.token_bytes(32)
        fd = os.open(_CRED_KEY_PATH, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        with os.fdopen(fd, "wb") as f:
            f.write(key)
        return key
    with open(_CRED_KEY_PATH, "rb") as f:
        return f.read()


def _cred_keystream(key, nonce, length):
    out, counter = b"", 0
    while len(out) < length:
        out += hmac.new(key, nonce + counter.to_bytes(4, "big"), hashlib.sha256).digest()
        counter += 1
    return out[:length]


def encrypt_secret(plaintext):
    if not plaintext:
        return ""
    key = _cred_key()
    nonce = secrets.token_bytes(16)
    pt = plaintext.encode("utf-8")
    ct = bytes(a ^ b for a, b in zip(pt, _cred_keystream(key, nonce, len(pt))))
    return base64.b64encode(nonce + ct).decode("ascii")


def decrypt_secret(token):
    if not token:
        return ""
    key = _cred_key()
    raw = base64.b64decode(token)
    nonce, ct = raw[:16], raw[16:]
    pt = bytes(a ^ b for a, b in zip(ct, _cred_keystream(key, nonce, len(ct))))
    return pt.decode("utf-8")


def create_user(firm_id, full_name, email, password, role="staff", membership_no=None):
    h, s = hash_password(password)
    return execute(
        "INSERT INTO app_user (firm_id, full_name, email, role, membership_no, pwd_hash, pwd_salt)"
        " VALUES (?,?,?,?,?,?,?)",
        (firm_id, full_name, email.lower().strip(), role, membership_no, h, s),
    )


def issue_session(user_id, device_id=None, days=30):
    token = secrets.token_urlsafe(32)
    expires = (datetime.now(timezone.utc) + timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S")
    execute(
        "INSERT INTO auth_session (token, user_id, device_id, expires_at) VALUES (?,?,?,?)",
        (token, user_id, device_id, expires),
    )
    return token


def user_for_token(token):
    if not token:
        return None
    return query(
        "SELECT u.* FROM auth_session s JOIN app_user u ON u.id = s.user_id"
        " WHERE s.token = ? AND s.revoked_at IS NULL AND s.expires_at > ? AND u.is_active = 1",
        (token, now_iso()), one=True,
    )


# ------------------------------------------------------------- access control

def granted_modules(user_id):
    """Module keys this user has been explicitly granted and which have not
    expired. Meaningless for a super admin — the caller checks is_super_admin
    first and never calls this for one, since a super admin's access isn't a
    finite list of rows. A grant with expires_at in the past is skipped
    entirely rather than deleted, so the Admin screen can still show it as a
    lapsed grant instead of it just silently vanishing.

    Also intersected with the user's own firm's entitlements
    (firm_module_grant) — a platform admin revoking a module from a firm's
    plan takes effect immediately for every one of that firm's staff, not
    just for new grants going forward; admin_create_user/admin_update_user
    already prevent granting outside this ceiling, this is the read-side
    half of the same rule."""
    rows = query(
        "SELECT g.module_key FROM user_module_grant g JOIN app_user u ON u.id = g.user_id"
        " WHERE g.user_id = ? AND (g.expires_at IS NULL OR g.expires_at >= date('now'))"
        " AND g.module_key IN (SELECT module_key FROM firm_module_grant WHERE firm_id = u.firm_id)",
        (user_id,))
    return [r["module_key"] for r in rows]


def module_grants_detail(user_id):
    """Every grant row for this user, expired or not, with its expiry — for
    the Admin screen, which needs to show a lapsed grant rather than hide it."""
    rows = query("SELECT module_key, expires_at FROM user_module_grant WHERE user_id = ?", (user_id,))
    return {r["module_key"]: r["expires_at"] for r in rows}


def visible_client_ids(user):
    """Every client this user may see: their own (client.assigned_to) plus
    anything they've been added to as a collaborator, scoped to their firm
    like every other query in this app. A super admin doesn't call this — the
    caller checks is_super_admin first and skips filtering entirely, since
    'every client in the firm' isn't something this should have to enumerate."""
    rows = query(
        "SELECT id FROM client WHERE firm_id = ? AND assigned_to = ? "
        "UNION SELECT c.id FROM client c JOIN client_collaborator cc ON cc.client_id = c.id"
        " WHERE c.firm_id = ? AND cc.user_id = ?",
        (user["firm_id"], user["id"], user["firm_id"], user["id"]))
    return {r["id"] for r in rows}


def create_invite(user_id, days=7):
    token = secrets.token_urlsafe(32)
    expires = (datetime.now(timezone.utc) + timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S")
    execute("INSERT INTO user_invite (token, user_id, expires_at) VALUES (?,?,?)", (token, user_id, expires))
    return token


def consume_invite(token, password):
    """Validate an unexpired, unused invite token and set the account's real
    password. Returns the activated user row, or None if the token is bad —
    the caller decides how to phrase that to the browser."""
    row = query("SELECT * FROM user_invite WHERE token = ? AND used_at IS NULL AND expires_at > ?",
                (token, now_iso()), one=True)
    if not row:
        return None
    h, s = hash_password(password)
    with conn() as c:
        c.execute("UPDATE app_user SET pwd_hash = ?, pwd_salt = ? WHERE id = ?", (h, s, row["user_id"]))
        c.execute("UPDATE user_invite SET used_at = ? WHERE token = ?", (now_iso(), token))
    return query("SELECT * FROM app_user WHERE id = ?", (row["user_id"],), one=True)


def create_password_reset(user_id, days=1):
    token = secrets.token_urlsafe(32)
    expires = (datetime.now(timezone.utc) + timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S")
    execute("INSERT INTO password_reset (token, user_id, expires_at) VALUES (?,?,?)", (token, user_id, expires))
    return token


def consume_password_reset(token, password):
    row = query("SELECT * FROM password_reset WHERE token = ? AND used_at IS NULL AND expires_at > ?",
                (token, now_iso()), one=True)
    if not row:
        return None
    h, s = hash_password(password)
    with conn() as c:
        c.execute("UPDATE app_user SET pwd_hash = ?, pwd_salt = ? WHERE id = ?", (h, s, row["user_id"]))
        c.execute("UPDATE password_reset SET used_at = ? WHERE token = ?", (now_iso(), token))
    return query("SELECT * FROM app_user WHERE id = ?", (row["user_id"],), one=True)


# --------------------------------------------------------------- client portal
# A second, separate credential space — see schema.sql's "CLIENT PORTAL"
# section for why this isn't just app_user with a flag. Every function here
# is a line-for-line mirror of the staff equivalent above, deliberately: the
# two must never drift into different security postures by accident, and
# keeping them textually parallel makes that easy to check at a glance.

def portal_create_user(client_id, name, email, password):
    h, s = hash_password(password)
    return execute(
        "INSERT INTO client_portal_user (client_id, name, email, pwd_hash, pwd_salt) VALUES (?,?,?,?,?)",
        (client_id, name, email.lower().strip(), h, s))


def portal_issue_session(portal_user_id, days=30):
    token = secrets.token_urlsafe(32)
    expires = (datetime.now(timezone.utc) + timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S")
    execute("INSERT INTO client_portal_session (token, portal_user_id, expires_at) VALUES (?,?,?)",
            (token, portal_user_id, expires))
    return token


def portal_user_for_token(token):
    if not token:
        return None
    return query(
        "SELECT pu.*, c.name client_name, c.firm_id FROM client_portal_session s"
        " JOIN client_portal_user pu ON pu.id = s.portal_user_id JOIN client c ON c.id = pu.client_id"
        " WHERE s.token = ? AND s.revoked_at IS NULL AND s.expires_at > ? AND pu.is_active = 1",
        (token, now_iso()), one=True)


def portal_create_invite(portal_user_id, days=7):
    token = secrets.token_urlsafe(32)
    expires = (datetime.now(timezone.utc) + timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S")
    execute("INSERT INTO client_portal_invite (token, portal_user_id, expires_at) VALUES (?,?,?)",
            (token, portal_user_id, expires))
    return token


def portal_consume_invite(token, password):
    row = query("SELECT * FROM client_portal_invite WHERE token = ? AND used_at IS NULL AND expires_at > ?",
               (token, now_iso()), one=True)
    if not row:
        return None
    h, s = hash_password(password)
    with conn() as c:
        c.execute("UPDATE client_portal_user SET pwd_hash = ?, pwd_salt = ? WHERE id = ?", (h, s, row["portal_user_id"]))
        c.execute("UPDATE client_portal_invite SET used_at = ? WHERE token = ?", (now_iso(), token))
    return query("SELECT * FROM client_portal_user WHERE id = ?", (row["portal_user_id"],), one=True)


def portal_create_password_reset(portal_user_id, days=1):
    token = secrets.token_urlsafe(32)
    expires = (datetime.now(timezone.utc) + timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S")
    execute("INSERT INTO client_portal_password_reset (token, portal_user_id, expires_at) VALUES (?,?,?)",
            (token, portal_user_id, expires))
    return token


def portal_consume_password_reset(token, password):
    row = query("SELECT * FROM client_portal_password_reset WHERE token = ? AND used_at IS NULL AND expires_at > ?",
               (token, now_iso()), one=True)
    if not row:
        return None
    h, s = hash_password(password)
    with conn() as c:
        c.execute("UPDATE client_portal_user SET pwd_hash = ?, pwd_salt = ? WHERE id = ?", (h, s, row["portal_user_id"]))
        c.execute("UPDATE client_portal_password_reset SET used_at = ? WHERE token = ?", (now_iso(), token))
    return query("SELECT * FROM client_portal_user WHERE id = ?", (row["portal_user_id"],), one=True)


# --------------------------------------------------------------- audit trail

def audit(actor_id, action, entity, entity_id=None, summary=None, ip=None):
    execute(
        "INSERT INTO audit_log (actor_id, action, entity, entity_id, summary, ip) VALUES (?,?,?,?,?,?)",
        (actor_id, action, entity, str(entity_id) if entity_id is not None else None, summary, ip),
    )


# ---------------------------------------------------------------- rate master

# The engines (engines.py and friends) call rates()/rate_dict() positionally,
# by domain only — no engine function takes a country_code parameter, and
# that isn't changing (see country_service.use_country's docstring for why).
# This contextvar is how a request supplies "which country" without every
# one of the ~15 call sites across engines.py needing a new parameter:
# app.py wraps a request in country_service.use_firm_country(...), and any
# rates()/rate_dict() call made anywhere underneath it — however deep —
# picks up that country automatically. Unset (the CLI, the worker, tests,
# any code that never enters that context) resolves to 'NP', identical to
# this function's behaviour before country_code existed at all.
_current_country = contextvars.ContextVar("current_country", default="NP")


def set_current_country(code):
    """Returns a token; pass it to reset_current_country to restore the
    previous value. See country_service.use_country — call sites should go
    through that context manager rather than calling this directly."""
    return _current_country.set(code)


def reset_current_country(token):
    _current_country.reset(token)


def get_current_country():
    """Read-only peek at the country set by country_service.use_country/
    use_firm_country — for callers that need to branch on jurisdiction
    (e.g. whether to attach a Bikram Sambat date to engine output), not
    just filter rate_master."""
    return _current_country.get()


def rates(domain=None, fy=None, applies_to=None, country_code=None):
    if country_code is None:
        country_code = _current_country.get()
    sql = "SELECT * FROM rate_master WHERE country_code = ?"
    args = [country_code]
    if domain:
        sql += " AND domain = ?"; args.append(domain)
    if fy:
        sql += " AND (fiscal_year_label = ? OR fiscal_year_label IS NULL)"; args.append(fy)
    if applies_to:
        sql += " AND (applies_to = ? OR applies_to = 'all')"; args.append(applies_to)
    return query(sql + " ORDER BY domain, applies_to, id", args)


def rate_dict(domain, fy, applies_to=None, country_code=None):
    """Return {code: row} plus an 'unverified' list, for the calculation engines."""
    rows = rates(domain, fy, applies_to, country_code=country_code)
    out, unverified = {}, []
    for r in rows:
        out[r["code"]] = r
        if r["verification_status"] == "UNVERIFIED":
            unverified.append(f"{r['domain']}.{r['code']}")
    return out, unverified


def unverified_count(country_code=None):
    """country_code=None (the default) counts across every country — the
    right behaviour for a context with no firm to scope by (run.py's CLI
    output, the public /api/health check) — unlike rates()/rate_dict(),
    which fall back to the ambient use_country()/use_firm_country() context
    because their callers (engines) are always already inside one. A caller
    that DOES have a firm (bootstrap(), GET /api/rates) should pass
    country_service.firm_country_code(firm_id) explicitly."""
    if country_code:
        r = query("SELECT COUNT(*) n FROM rate_master WHERE verification_status = 'UNVERIFIED' AND country_code = ?",
                  (country_code,), one=True)
    else:
        r = query("SELECT COUNT(*) n FROM rate_master WHERE verification_status = 'UNVERIFIED'", one=True)
    return r["n"] if r else 0


# ------------------------------------------------------------ industry defaults

def industry_defaults():
    """Planning defaults for the Projection module, JSON columns parsed to lists."""
    rows = query("SELECT * FROM industry_default ORDER BY category, name")
    out = []
    for r in rows:
        d = dict(r)
        for col in ("util_ramp", "cost_pct", "wc_days", "capex_split"):
            d[col] = json.loads(d[col])
        out.append(d)
    return out


def industry_default(name):
    for row in industry_defaults():
        if row["name"] == name:
            return row
    return None


# ------------------------------------------------------------------ FDI sectors

def fdi_sectors():
    return query("SELECT * FROM fdi_sector ORDER BY name")


def fdi_sector(name):
    for row in fdi_sectors():
        if row["name"] == name:
            return row
    return None


# ------------------------------------------------------------- audit catalogs

def audit_entity_types():
    return query("SELECT * FROM audit_entity_type ORDER BY sort_order, name")


def audit_criteria(entity_type=None):
    """All criteria applicable to entity_type: those with applies_to='all',
    plus any whose applies_to JSON array names this entity type. Returns
    every criterion (applies_to='all' only) when entity_type is None."""
    rows = query("SELECT * FROM audit_criterion ORDER BY sort_order, area")
    out = []
    for r in rows:
        d = dict(r)
        if d["applies_to"] == "all":
            d["applies_to"] = "all"
            out.append(d)
        elif entity_type:
            try:
                names = json.loads(d["applies_to"])
            except (TypeError, ValueError):
                names = []
            if entity_type in names:
                d["applies_to"] = names
                out.append(d)
    return out


# ------------------------------------------------------ workfiles and autosave

def get_state(workfile_id):
    return query("SELECT * FROM workfile_state WHERE workfile_id = ?", (workfile_id,), one=True)


def save_state(workfile_id, payload, user_id=None, device_id=None,
               origin="auto", base_version=None, client_seq=None):
    """
    Upsert the authoritative state and record an autosave beat.

    Optimistic locking: if base_version is supplied and no longer matches, the
    incoming payload is parked in sync_conflict instead of overwriting, and the
    caller gets conflict=True so the browser can show the merge screen.
    """
    body = json.dumps(payload, ensure_ascii=False)
    checksum = hashlib.sha256(body.encode()).hexdigest()[:16]
    with conn() as c:
        cur = c.execute("SELECT version, checksum FROM workfile_state WHERE workfile_id = ?", (workfile_id,))
        row = cur.fetchone()

        if row and base_version is not None and row["version"] != base_version:
            server_body = c.execute(
                "SELECT payload FROM workfile_state WHERE workfile_id = ?", (workfile_id,)
            ).fetchone()["payload"]
            c.execute(
                "INSERT INTO sync_conflict (workfile_id, incoming, server_state, base_version, device_id)"
                " VALUES (?,?,?,?,?)",
                (workfile_id, body, server_body, base_version, device_id),
            )
            return {"conflict": True, "server_version": row["version"]}

        # Unchanged payload: don't burn a snapshot row on an idle heartbeat.
        if row and row["checksum"] == checksum:
            return {"conflict": False, "version": row["version"], "unchanged": True}

        version = (row["version"] + 1) if row else 1
        if row:
            c.execute(
                "UPDATE workfile_state SET payload=?, version=?, updated_at=?, updated_by=?,"
                " device_id=?, checksum=? WHERE workfile_id=?",
                (body, version, now_iso(), user_id, device_id, checksum, workfile_id),
            )
        else:
            c.execute(
                "INSERT INTO workfile_state (workfile_id, payload, version, updated_at, updated_by,"
                " device_id, checksum) VALUES (?,?,?,?,?,?,?)",
                (workfile_id, body, version, now_iso(), user_id, device_id, checksum),
            )
        c.execute(
            "INSERT INTO autosave_snapshot (workfile_id, payload, origin, client_seq, device_id, byte_size)"
            " VALUES (?,?,?,?,?,?)",
            (workfile_id, body, origin, client_seq, device_id, len(body)),
        )
    return {"conflict": False, "version": version, "unchanged": False}


def prune_snapshots(retention_days=120, keep_per_file=200):
    """Housekeeping run by the night worker."""
    cutoff = (datetime.now(timezone.utc) - timedelta(days=retention_days)).strftime("%Y-%m-%d %H:%M:%S")
    with conn() as c:
        c.execute("DELETE FROM autosave_snapshot WHERE saved_at < ? AND origin = 'auto'", (cutoff,))
        c.execute(
            "DELETE FROM autosave_snapshot WHERE id IN ("
            "  SELECT id FROM ("
            "    SELECT id, ROW_NUMBER() OVER (PARTITION BY workfile_id ORDER BY saved_at DESC) rn"
            "    FROM autosave_snapshot) ranked WHERE rn > ?)",
            (keep_per_file,),
        )
