#!/usr/bin/env python3
# /// script
# dependencies = ["pymysql"]
# ///
"""Apply the Credits permission-matrix seed directly to TiDB (#3880, Model-3 pilot).

Companion to ``apply_credits_approval_seed.py``; same idempotent direct-SQL shape,
adapted for permissions. The seed owns three tables — ``PermissionSet`` (the 5
Credits permission SETS, never profiles), ``ObjectPermissions`` and
``FieldPermissions``. It **never** touches ``PermissionSetAssignment``: which user
holds which set is runtime data the org owns (the definitions-vs-assignments
boundary the security owner chose in #3924). Profiles (``IsOwnedByProfile=1``) are
out of scope.

Adoption by PRIMARY key (aligned with the 20260531/007 patch bundle and the #3887
approval seed): every seed row carries a deterministic id — EXISTING rows carry
their exact prod ids, NEW rows (Topup FLS mirror, Regional_Ops FLS) carry minted
``*_pilot`` ids — and each row is written with ``INSERT ... ON DUPLICATE KEY UPDATE
... IsDeleted=0``. So an existing row is adopted by its Id (index-independent — no
reliance on a composite natural-key index that the canonical schema does not
guarantee), a soft-deleted row is reactivated, and a fresh env inserts. Children
reference the permset by its prod id, which is exactly the id the permset upsert
writes, so no id remapping is needed.

Additive / upsert-only: the applier **never deletes** a permission row — the
lowest-blast-radius choice for a first permission pilot. It does not converge
drift by removing undeclared rows; that can be layered on later once the pilot is
proven. Run via ``uv run`` (PEP-723 header pulls pymysql) or ``python3`` where
pymysql is already installed.
"""
from __future__ import annotations

import argparse
import json
import os
import ssl
from collections import Counter
from pathlib import Path
from typing import Any


def _dupes(values: list[str]) -> list[str]:
    """Sorted list of values that appear more than once (empties ignored)."""
    counts = Counter(v for v in values if v)
    return sorted(v for v, n in counts.items() if n > 1)

# The only objects this seed's permission rows may target.
CREDITS_OBJECTS = {"Credits__c", "Credits_Topup__c"}

# Master-detail children (sharingModel ControlledByParent) have NO authorable
# ObjectPermissions row: object CRUD is projected from the parent, and the
# platform seeder suppresses the child OP row (permission_seeder.go
# BuildPermissionsForProfileAndSchema, isParentControlled). Seeding one via
# direct SQL would create a rogue second CRUD path, so the applier refuses
# object-perm rows for these objects. FLS rows are still allowed (field-level
# security applies independently per child field). (Codex P2, PR #3965.)
PARENT_CONTROLLED_OBJECTS = {"Credits_Topup__c"}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Apply the Credits permission-matrix seed directly to TiDB."
    )
    parser.add_argument("--seed-file", required=True, type=Path)
    parser.add_argument("--validate-only", action="store_true")
    return parser.parse_args()


def tls_enabled(raw: str) -> bool:
    return str(raw).strip().lower() in {"", "1", "true", "yes", "on"}


def required_env(name: str) -> str:
    value = os.environ.get(name, "").strip()
    if not value:
        raise RuntimeError(f"{name} is required")
    return value


def connect():
    import pymysql

    kwargs: dict[str, Any] = {
        "host": required_env("TIDB_HOST"),
        "port": int(required_env("TIDB_PORT")),
        "user": required_env("TIDB_USER"),
        "password": os.environ.get("TIDB_PASSWORD", ""),
        "database": required_env("TIDB_DATABASE"),
        "charset": "utf8mb4",
        "autocommit": False,
        "connect_timeout": 10,
        "read_timeout": 60,
        "write_timeout": 60,
        "cursorclass": pymysql.cursors.DictCursor,
    }
    if tls_enabled(os.environ.get("TIDB_TLS", "true")):
        kwargs["ssl"] = ssl.create_default_context()
    return pymysql.connect(**kwargs)


def bool_int(value: Any) -> int:
    return 1 if bool(value) else 0


def nullable_string(value: Any) -> str | None:
    if value is None:
        return None
    text = str(value)
    return text if text else None


def _perm_flags(row: dict[str, Any], keys: tuple[str, ...]) -> None:
    for key in keys:
        value = row.get(key)
        if value not in (0, 1, True, False):
            raise RuntimeError(
                f"permission flag {key!r} must be 0/1, got {value!r} in {row.get('id')!r}"
            )


def validate_seed(seed: dict[str, Any]) -> None:
    permsets = seed.get("permissionSets") or []
    objperms = seed.get("objectPermissions") or []
    fieldperms = seed.get("fieldPermissions") or []
    errors: list[str] = []

    # ---- permission sets ----
    names = [str(p.get("name") or "").strip() for p in permsets]
    ids = [str(p.get("id") or "").strip() for p in permsets]
    if any(not n for n in names):
        errors.append("permissionSet row(s) missing a name")
    if any(not i for i in ids):
        errors.append("permissionSet row(s) missing an id")
    for dupset, label in ((names, "name"), (ids, "id")):
        dupes = _dupes(dupset)
        if dupes:
            errors.append(f"duplicate permissionSet {label}(s): {dupes}")
    seed_permset_ids = set(ids)
    # A profile masquerading as a permset would let the seed mutate a base profile
    # (out of scope). The seed never carries isOwnedByProfile; reject it if present
    # and truthy so a copy-paste from a profile row cannot slip in.
    for p in permsets:
        if bool_int(p.get("isOwnedByProfile")):
            errors.append(f"permissionSet {p.get('id')!r} is profile-owned (out of pilot scope)")
        # Name/Label are varchar(255) under STRICT_TRANS_TABLES — an over-long value
        # passes JSON but fails at INSERT; catch it in validation.
        for col in ("name", "label", "id"):
            if len(str(p.get(col) or "")) > 255:
                errors.append(f"permissionSet {p.get('id')!r} {col} exceeds 255 chars")
        # isActive/isCustom feed bool_int, which coerces any truthy value; reject a
        # non-boolean so a stray string can't silently flip a flag.
        for col in ("isActive", "isCustom"):
            if p.get(col) not in (0, 1, True, False, None):
                errors.append(f"permissionSet {p.get('id')!r} {col} must be 0/1, got {p.get(col)!r}")

    # ---- object / field permission rows ----
    def check_children(rows: list[dict[str, Any]], label: str, natural_keys: tuple[str, ...],
                       flags: tuple[str, ...], *, allow_all_zero: bool) -> None:
        seen: set[tuple] = set()
        for row in rows:
            rid = str(row.get("id") or "").strip() or "<missing-id>"
            parent = str(row.get("parentId") or "").strip()
            sobj = str(row.get("sobjectType") or "").strip()
            if not str(row.get("id") or "").strip():
                errors.append(f"{label} row missing an id")
            if parent not in seed_permset_ids:
                errors.append(
                    f"{label} {rid} references parentId {parent!r} not in this seed's permissionSets"
                )
            if sobj not in CREDITS_OBJECTS:
                errors.append(
                    f"{label} {rid} targets {sobj!r} (only {sorted(CREDITS_OBJECTS)} allowed)"
                )
            try:
                _perm_flags(row, flags)
            except RuntimeError as exc:
                errors.append(str(exc))
            # An all-zero FieldPermissions row is a valid explicit DENY (an absent
            # row would fall back to object-CRUD read, so the deny must persist).
            # An all-zero ObjectPermissions row, by contrast, grants nothing and
            # has no deny semantics — reject it as noise.
            if not allow_all_zero and not any(bool_int(row.get(f)) for f in flags):
                errors.append(f"{label} {rid} grants nothing (all flags 0) — drop it from the seed")
            key = tuple(str(row.get(k) or "").strip() for k in natural_keys)
            if key in seen:
                errors.append(f"duplicate {label} natural key {key}")
            seen.add(key)
            # id uniqueness
        dupes = _dupes([str(r.get("id") or "").strip() for r in rows])
        if dupes:
            errors.append(f"duplicate {label} id(s): {dupes}")

    check_children(objperms, "objectPermission", ("parentId", "sobjectType"),
                   ("read", "create", "edit", "delete", "viewAll", "modifyAll"),
                   allow_all_zero=False)
    for row in objperms:
        if str(row.get("sobjectType") or "").strip() in PARENT_CONTROLLED_OBJECTS:
            errors.append(
                f"objectPermission {row.get('id')!r} targets parent-controlled "
                f"{row.get('sobjectType')!r}; child object CRUD is projected from the "
                f"parent and has no authorable ObjectPermissions row (FLS is fine)"
            )
        # ModifyAll must be cascade-consistent (Salesforce / contract 730_AC2:
        # Modify All auto-enables and locks Read/Edit/Delete/ViewAll — 'Modify All
        # but no delete' is not a valid state). Fail closed on a non-cascaded row so
        # prod's hand-patched ModifyAll=1/Delete=0 anomaly cannot be re-seeded.
        if bool_int(row.get("modifyAll")) and not all(
            bool_int(row.get(f)) for f in ("viewAll", "edit", "delete", "read")
        ):
            errors.append(
                f"objectPermission {row.get('id')!r} sets modifyAll=1 but not all of "
                f"read/edit/delete/viewAll=1 (ModifyAll must cascade to them)"
            )
    check_children(fieldperms, "fieldPermission", ("parentId", "sobjectType", "field"),
                   ("read", "edit"), allow_all_zero=True)

    # field rows must name a fully-qualified Object.Field on their own object
    for row in fieldperms:
        field = str(row.get("field") or "").strip()
        sobj = str(row.get("sobjectType") or "").strip()
        if "." not in field or field.split(".", 1)[0] != sobj:
            errors.append(
                f"fieldPermission {row.get('id')!r} field {field!r} must be '{sobj}.<Field__c>'"
            )

    # The seed owns definitions only. A key that would drive assignment is a
    # category error — fail closed rather than silently ignore it.
    if seed.get("permissionSetAssignments") is not None:
        errors.append("seed carries permissionSetAssignments; the pilot owns definitions only")

    if errors:
        raise RuntimeError("seed validation failed:\n  - " + "\n  - ".join(errors))


def load_seed(path: Path) -> dict[str, Any]:
    seed = json.loads(path.read_text())
    if seed.get("version") != 1:
        raise RuntimeError(f"unsupported seed version in {path}: {seed.get('version')}")
    validate_seed(seed)
    return seed


def upsert_permission_sets(cur, permsets: list[dict[str, Any]]) -> None:
    # Upsert by Id (the seed carries prod ids). ON DUPLICATE KEY UPDATE adopts an
    # existing row and reactivates a soft-deleted one (IsDeleted=0). Never writes a
    # profile: the seed ids are Credits-namespaced and IsOwnedByProfile is pinned 0.
    for permset in permsets:
        # Fail closed if a seed id collides with a profile-owned permission set:
        # the unconditional IsOwnedByProfile=0/ProfileId=NULL below would silently
        # detach that profile's implicit permset. A human must reconcile.
        cur.execute(
            "SELECT IsOwnedByProfile, ProfileId FROM PermissionSet WHERE Id=%s",
            (permset["id"],),
        )
        existing = cur.fetchone()
        if existing is not None and (bool_int(existing["IsOwnedByProfile"]) or existing["ProfileId"] is not None):
            raise RuntimeError(
                f"refusing to adopt PermissionSet {permset['id']!r}: existing row is "
                f"profile-owned (IsOwnedByProfile/ProfileId set) — reconcile manually"
            )
        cur.execute(
            """
            INSERT INTO PermissionSet
              (Id, Name, Label, Description, IsActive, IsCustom,
               IsOwnedByProfile, ProfileId, IsDeleted, CreatedDate, LastModifiedDate)
            VALUES
              (%s, %s, %s, %s, %s, %s, 0, NULL, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
            ON DUPLICATE KEY UPDATE
              Name=VALUES(Name), Label=VALUES(Label), Description=VALUES(Description),
              IsActive=VALUES(IsActive), IsCustom=VALUES(IsCustom),
              IsOwnedByProfile=0, IsDeleted=0, LastModifiedDate=CURRENT_TIMESTAMP
            """,
            (
                permset["id"],
                permset["name"],
                permset["label"],
                nullable_string(permset.get("description")),
                bool_int(permset.get("isActive", True)),
                bool_int(permset.get("isCustom", True)),
            ),
        )


def upsert_object_permissions(cur, rows: list[dict[str, Any]]) -> None:
    # Upsert by Id (prod id adopts by PRIMARY, index-independent); IsDeleted=0
    # reactivates. parentId is the permset prod id, exactly what upsert_permission
    # _sets wrote, so no remapping is needed.
    for row in rows:
        cur.execute(
            """
            INSERT INTO ObjectPermissions
              (Id, ParentId, SobjectType, PermissionsRead, PermissionsCreate,
               PermissionsEdit, PermissionsDelete, PermissionsViewAllRecords,
               PermissionsModifyAllRecords, IsDeleted, CreatedDate, LastModifiedDate)
            VALUES
              (%s, %s, %s, %s, %s, %s, %s, %s, %s, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
            ON DUPLICATE KEY UPDATE
              ParentId=VALUES(ParentId), SobjectType=VALUES(SobjectType),
              PermissionsRead=VALUES(PermissionsRead),
              PermissionsCreate=VALUES(PermissionsCreate),
              PermissionsEdit=VALUES(PermissionsEdit),
              PermissionsDelete=VALUES(PermissionsDelete),
              PermissionsViewAllRecords=VALUES(PermissionsViewAllRecords),
              PermissionsModifyAllRecords=VALUES(PermissionsModifyAllRecords),
              IsDeleted=0, LastModifiedDate=CURRENT_TIMESTAMP
            """,
            (
                row["id"],
                row["parentId"],
                row["sobjectType"],
                bool_int(row.get("read")),
                bool_int(row.get("create")),
                bool_int(row.get("edit")),
                bool_int(row.get("delete")),
                bool_int(row.get("viewAll")),
                bool_int(row.get("modifyAll")),
            ),
        )


def upsert_field_permissions(cur, rows: list[dict[str, Any]]) -> None:
    for row in rows:
        cur.execute(
            """
            INSERT INTO FieldPermissions
              (Id, ParentId, SobjectType, Field, PermissionsRead, PermissionsEdit,
               IsDeleted, CreatedDate, LastModifiedDate)
            VALUES
              (%s, %s, %s, %s, %s, %s, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
            ON DUPLICATE KEY UPDATE
              ParentId=VALUES(ParentId), SobjectType=VALUES(SobjectType), Field=VALUES(Field),
              PermissionsRead=VALUES(PermissionsRead),
              PermissionsEdit=VALUES(PermissionsEdit),
              IsDeleted=0, LastModifiedDate=CURRENT_TIMESTAMP
            """,
            (
                row["id"],
                row["parentId"],
                row["sobjectType"],
                row["field"],
                bool_int(row.get("read")),
                bool_int(row.get("edit")),
            ),
        )


CHILD_TABLES = (
    ("objectPermissions", "ObjectPermissions", ("ParentId", "SobjectType"),
     ("parentId", "sobjectType")),
    ("fieldPermissions", "FieldPermissions", ("ParentId", "SobjectType", "Field"),
     ("parentId", "sobjectType", "field")),
)


def preflight_children(cur, seed: dict[str, Any]) -> None:
    """Fail closed BEFORE writing if a child row's natural key is already held at a
    different id.

    ObjectPermissions and FieldPermissions each carry a UNIQUE natural key
    (``idx_ObjectPermissions_ParentId_SobjectType`` /
    ``idx_FieldPermissions_ParentId_SobjectType_Field``) besides ``PRIMARY(Id)``.
    An ``INSERT ... ON DUPLICATE KEY UPDATE`` carrying the seed's id would collide
    on that natural key instead and update the *existing* row, so the seed's id is
    never written — adoption silently stops being by PRIMARY id. The grants may
    still land correctly, but the deterministic-id contract is broken and the
    divergence is invisible. A human must reconcile."""
    for seed_key, table, db_cols, seed_cols in CHILD_TABLES:
        for row in seed[seed_key]:
            where = " AND ".join(f"{c} = %s" for c in db_cols)
            cur.execute(
                f"SELECT Id FROM {table} WHERE {where} "
                f"AND (IsDeleted = 0 OR IsDeleted IS NULL)",
                tuple(row[c] for c in seed_cols),
            )
            found = cur.fetchone()
            if found is not None and str(found["Id"]) != row["id"]:
                natural = ", ".join(
                    f"{c}={row[s]!r}" for c, s in zip(db_cols, seed_cols)
                )
                raise RuntimeError(
                    f"preflight failed: {table} natural key ({natural}) is already "
                    f"held by id {str(found['Id'])!r}, but the seed declares "
                    f"{row['id']!r}. Upserting would silently adopt the existing row "
                    f"via the UNIQUE natural key and never write the seed id — "
                    f"reconcile manually."
                )


def verify_children_applied(cur, seed: dict[str, Any]) -> None:
    """Fail closed if any child row is not live at its seed id with the seed's
    flags. Verifies the effective grant, not just row presence, so a no-op or a
    partially-applied flag set cannot commit."""
    checks = (
        ("objectPermissions", "ObjectPermissions", {
            "PermissionsRead": "read", "PermissionsCreate": "create",
            "PermissionsEdit": "edit", "PermissionsDelete": "delete",
            "PermissionsViewAllRecords": "viewAll",
            "PermissionsModifyAllRecords": "modifyAll",
        }),
        ("fieldPermissions", "FieldPermissions", {
            "PermissionsRead": "read", "PermissionsEdit": "edit",
        }),
    )
    for seed_key, table, flag_map in checks:
        for row in seed[seed_key]:
            cols = ", ".join(flag_map)
            cur.execute(
                f"SELECT {cols} FROM {table} WHERE Id = %s "
                f"AND (IsDeleted = 0 OR IsDeleted IS NULL)",
                (row["id"],),
            )
            live = cur.fetchone()
            if live is None:
                raise RuntimeError(
                    f"post-apply verification failed: {table} row {row['id']!r} is "
                    f"not live after apply"
                )
            for db_col, seed_col in flag_map.items():
                if bool_int(live[db_col]) != bool_int(row.get(seed_col)):
                    raise RuntimeError(
                        f"post-apply verification failed: {table} row {row['id']!r} "
                        f"has {db_col}={live[db_col]}, seed declares "
                        f"{bool_int(row.get(seed_col))}"
                    )


def verify_applied(cur, seed: dict[str, Any]) -> None:
    """Fail closed if adoption did not land where the seed expects. PermissionSet
    carries a separate ``UNIQUE(Name)`` besides ``PRIMARY(Id)``, so a same-Name row
    at a DIFFERENT id would make the permset upsert converge that other row (via the
    Name conflict) and never insert the seed's id — silently orphaning every child
    (which references the parent by id). Assert every seed permset id is a live row
    (validate_seed already guarantees each child ParentId is one of these ids, so a
    present permset id implies no orphaned child)."""
    permset_ids = [p["id"] for p in seed["permissionSets"]]
    ph = ",".join(["%s"] * len(permset_ids))
    cur.execute(
        f"SELECT Id FROM PermissionSet WHERE Id IN ({ph}) "
        f"AND (IsDeleted = 0 OR IsDeleted IS NULL)",
        tuple(permset_ids),
    )
    present = {str(r["Id"]) for r in cur.fetchall()}
    missing = [i for i in permset_ids if i not in present]
    if missing:
        raise RuntimeError(
            "post-apply verification failed: seed PermissionSet id(s) not live after "
            "apply — a same-Name permset at a different id likely absorbed the write "
            f"via UNIQUE(Name): {missing}"
        )


def apply_seed(seed: dict[str, Any]) -> dict[str, int]:
    conn = connect()
    try:
        with conn.cursor() as cur:
            # Fail closed before mutating anything if a child natural key is
            # already held at a different id (UNIQUE natural key would absorb
            # the upsert and the seed id would never be written).
            preflight_children(cur, seed)
            # Additive / upsert-only, in order (permsets before their children).
            upsert_permission_sets(cur, seed["permissionSets"])
            upsert_object_permissions(cur, seed["objectPermissions"])
            upsert_field_permissions(cur, seed["fieldPermissions"])
            # Fail closed before commit: parents live, children live at their
            # seed ids with the seed's exact flags.
            verify_applied(cur, seed)
            verify_children_applied(cur, seed)
        conn.commit()
        return {
            "permission_sets": len(seed["permissionSets"]),
            "object_permissions": len(seed["objectPermissions"]),
            "field_permissions": len(seed["fieldPermissions"]),
        }
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def main() -> int:
    args = parse_args()
    seed = load_seed(args.seed_file)
    if args.validate_only:
        print(
            "credits permission seed valid: "
            f"{len(seed['permissionSets'])} permission sets, "
            f"{len(seed['objectPermissions'])} object permissions, "
            f"{len(seed['fieldPermissions'])} field permissions"
        )
        return 0
    counts = apply_seed(seed)
    print(
        "applied credits permission seed: "
        f"{counts['permission_sets']} permission sets, "
        f"{counts['object_permissions']} object permissions, "
        f"{counts['field_permissions']} field permissions"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())