#!/usr/bin/env python3
"""Render and apply the pack-owned reference-data manifest with mysql CLI.

The production runner deliberately has no PyMySQL installation contract. This
applier therefore uses the same checked-in source and dependency semantics as
the shared bulk seed path, but keeps the production write backend to mysql CLI.
"""

from __future__ import annotations

import argparse
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import time
from typing import Any


AUDIT_DATETIME_COLUMNS = ("CreatedDate", "LastModifiedDate", "SystemModstamp")
AUDIT_USER_COLUMNS = ("CreatedById", "LastModifiedById", "OwnerId")
# IsDeleted is here so "seeding never revives a deleted row" is a property of the applier
# rather than of the current contents of the reference-data sources. Without it, adding
# `"IsDeleted": false` to any source's `fields` block puts the column straight back into the
# ON DUPLICATE KEY UPDATE clause and silently restores resurrection.
NEVER_UPDATE_COLUMNS = {"Id", "CreatedDate", "CreatedById", "IsDeleted"}
TEMPLATE_RE = re.compile(r"\{\{\s*([^}]+?)\s*\}\}")


class PackReferenceDataError(RuntimeError):
    pass


class MissingReference(PackReferenceDataError):
    def __init__(self, key: str, expression: str):
        super().__init__(f"missing record reference {key!r} in {expression!r}")
        self.key = key
        self.expression = expression


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Apply pack-owned reference data from a manifest.")
    parser.add_argument("--manifest-file", required=True, type=Path)
    parser.add_argument("--state-file", required=True, type=Path)
    parser.add_argument("--summary-file", required=True, type=Path)
    parser.add_argument("--objects", required=True, help="Comma-separated manifest object allowlist.")
    parser.add_argument("--dry-run", action="store_true", help="Plan only; do not write TiDB rows.")
    parser.add_argument("--db-host", default=os.environ.get("TIDB_HOST", ""))
    parser.add_argument("--db-port", type=int, default=int(os.environ.get("TIDB_PORT", "4000")))
    parser.add_argument("--db-user", default=os.environ.get("TIDB_USER", ""))
    parser.add_argument("--db-password", default=os.environ.get("TIDB_PASSWORD", ""))
    parser.add_argument("--db-database", default=os.environ.get("TIDB_DATABASE", ""))
    parser.add_argument("--db-tls", default=os.environ.get("TIDB_TLS", "true"))
    return parser.parse_args()


def load_json(path: Path, *, label: str) -> dict[str, Any]:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except OSError as exc:
        raise PackReferenceDataError(f"unable to read {label}: {path}") from exc
    except json.JSONDecodeError as exc:
        raise PackReferenceDataError(f"invalid JSON {label}: {path}") from exc
    if not isinstance(payload, dict):
        raise PackReferenceDataError(f"{label} must be a JSON object: {path}")
    return payload


def parse_object_list(raw: str) -> list[str]:
    objects = [item.strip() for item in raw.split(",") if item.strip()]
    if not objects:
        raise PackReferenceDataError("--objects must name at least one manifest object")
    if len(objects) != len(set(objects)):
        raise PackReferenceDataError("--objects must not contain duplicates")
    return objects


def manifest_object_specs(manifest: dict[str, Any]) -> dict[str, dict[str, Any]]:
    if manifest.get("version") != 1:
        raise PackReferenceDataError(
            f"unsupported reference-data manifest version: {manifest.get('version')!r}"
        )
    raw_objects = manifest.get("objects")
    if not isinstance(raw_objects, dict) or not raw_objects:
        raise PackReferenceDataError("reference-data manifest objects must be a non-empty object")

    specs: dict[str, dict[str, Any]] = {}
    for object_api_name, raw_spec in raw_objects.items():
        if not isinstance(object_api_name, str) or not object_api_name.strip():
            raise PackReferenceDataError("reference-data manifest contains an empty object name")
        if not isinstance(raw_spec, dict):
            raise PackReferenceDataError(f"manifest object spec must be an object: {object_api_name}")
        source = raw_spec.get("source")
        if isinstance(source, str) and source.strip():
            sources = [source]
        elif isinstance(source, list) and source and all(
            isinstance(item, str) and item.strip() for item in source
        ):
            sources = list(source)
        else:
            raise PackReferenceDataError(
                f"manifest object {object_api_name} must name a source file "
                "or a non-empty list of source files"
            )
        depends_on = raw_spec.get("depends_on", [])
        if not isinstance(depends_on, list) or not all(
            isinstance(item, str) and item.strip() for item in depends_on
        ):
            raise PackReferenceDataError(
                f"manifest object {object_api_name} depends_on must be a list of object names"
            )
        specs[object_api_name] = {"sources": sources, "depends_on": list(depends_on)}

    for object_api_name, spec in specs.items():
        missing = [item for item in spec["depends_on"] if item not in specs]
        if missing:
            raise PackReferenceDataError(
                f"manifest object {object_api_name} depends on unknown objects: {', '.join(missing)}"
            )
    return specs


def resolve_source_path(manifest_file: Path, raw_source: str) -> Path:
    root = manifest_file.parent.resolve()
    candidate = (root / raw_source).resolve()
    try:
        candidate.relative_to(root)
    except ValueError as exc:
        raise PackReferenceDataError(f"reference-data source escapes manifest directory: {raw_source}") from exc
    return candidate


def validate_dependency_closure(selected: list[str], specs: dict[str, dict[str, Any]]) -> None:
    selected_set = set(selected)
    missing: dict[str, list[str]] = {}
    for object_api_name in selected:
        dependencies = [item for item in specs[object_api_name]["depends_on"] if item not in selected_set]
        if dependencies:
            missing[object_api_name] = dependencies
    if missing:
        detail = "; ".join(
            f"{object_api_name} requires {', '.join(dependencies)}"
            for object_api_name, dependencies in missing.items()
        )
        raise PackReferenceDataError(f"reference-data object dependency is missing: {detail}")


def validate_record(record: Any, source_file: Path) -> dict[str, Any]:
    if not isinstance(record, dict):
        raise PackReferenceDataError(f"reference-data record must be an object: {source_file}")
    key = record.get("key")
    object_api_name = record.get("object")
    match = record.get("match")
    fields = record.get("fields")
    if not isinstance(key, str) or not key.strip():
        raise PackReferenceDataError(f"reference-data record has no key: {source_file}")
    if not isinstance(object_api_name, str) or not object_api_name.strip():
        raise PackReferenceDataError(f"reference-data record {key} has no object: {source_file}")
    if not isinstance(match, dict) or not match:
        raise PackReferenceDataError(f"reference-data record {key} has an empty match: {source_file}")
    if not isinstance(fields, dict):
        raise PackReferenceDataError(f"reference-data record {key} fields must be an object: {source_file}")
    return record


def build_seed(manifest_file: Path, selected: list[str], specs: dict[str, dict[str, Any]]) -> dict[str, Any]:
    declared_by_source: dict[Path, set[str]] = {}
    for object_api_name, spec in specs.items():
        for raw_source in spec["sources"]:
            source_file = resolve_source_path(manifest_file, raw_source)
            declared_by_source.setdefault(source_file, set()).add(object_api_name)

    selected_set = set(selected)
    records: list[dict[str, Any]] = []
    record_keys: set[str] = set()
    found_objects: set[str] = set()
    for source_file, declared_objects in declared_by_source.items():
        payload = load_json(source_file, label="reference-data source")
        if payload.get("version") != 1:
            raise PackReferenceDataError(
                f"unsupported reference-data source version in {source_file}: {payload.get('version')!r}"
            )
        raw_records = payload.get("records")
        if not isinstance(raw_records, list):
            raise PackReferenceDataError(f"reference-data source records must be a list: {source_file}")
        for raw_record in raw_records:
            record = validate_record(raw_record, source_file)
            object_api_name = str(record["object"])
            if object_api_name not in declared_objects:
                raise PackReferenceDataError(
                    f"reference-data source {source_file.name} declares unexpected object {object_api_name}; "
                    f"allowed={','.join(sorted(declared_objects))}"
                )
            key = str(record["key"])
            if key in record_keys:
                raise PackReferenceDataError(f"duplicate reference-data record key: {key}")
            record_keys.add(key)
            if object_api_name in selected_set:
                records.append(record)
                found_objects.add(object_api_name)

    missing_records = [object_api_name for object_api_name in selected if object_api_name not in found_objects]
    if missing_records:
        raise PackReferenceDataError(
            "reference-data manifest has no records for requested objects: " + ", ".join(missing_records)
        )
    return {"baseline_records": records}


def stable_uuid(namespace: str, key: str) -> str:
    digest = hashlib.sha1(f"{namespace}:{key}".encode("utf-8")).hexdigest()[:32]
    chars = list(digest)
    chars[12] = "5"
    chars[16] = "8"
    return "-".join(
        [
            "".join(chars[0:8]),
            "".join(chars[8:12]),
            "".join(chars[12:16]),
            "".join(chars[16:20]),
            "".join(chars[20:32]),
        ]
    )


def resolve_path(source: Any, path_expr: str) -> Any:
    current = source
    normalized = re.sub(r"\[(\d+)\]", r".\1", path_expr)
    for part in [part for part in normalized.split(".") if part]:
        if current is None:
            return None
        if isinstance(current, dict):
            current = current.get(part)
        elif isinstance(current, list) and part.isdigit():
            index = int(part)
            current = current[index] if index < len(current) else None
        else:
            return None
    return current


def resolve_expression(expression: str, context: dict[str, Any]) -> Any:
    expression = expression.strip()
    if expression.startswith("record.") or expression.startswith("records."):
        path_expr = re.sub(r"^records?\.", "", expression)
        record_key = path_expr.split(".", 1)[0]
        value = resolve_path(context["records"], path_expr)
        if value is None:
            raise MissingReference(record_key, expression)
        return value
    return None


def materialize_value(value: Any, context: dict[str, Any]) -> Any:
    if isinstance(value, str):
        full_match = TEMPLATE_RE.fullmatch(value.strip())
        if full_match:
            resolved = resolve_expression(full_match.group(1), context)
            return value if resolved is None else resolved

        def replace(match: re.Match[str]) -> str:
            resolved = resolve_expression(match.group(1), context)
            return match.group(0) if resolved is None else str(resolved)

        return TEMPLATE_RE.sub(replace, value)
    if isinstance(value, list):
        return [materialize_value(item, context) for item in value]
    if isinstance(value, dict):
        return {key: materialize_value(nested, context) for key, nested in value.items()}
    return value


def stable_object(value: Any) -> Any:
    if isinstance(value, list):
        return [stable_object(item) for item in value]
    if isinstance(value, dict):
        return {key: stable_object(value[key]) for key in sorted(value)}
    return value


def match_signature(object_api_name: str, match_fields: dict[str, Any]) -> str:
    return json.dumps(
        {"object": object_api_name, "match": stable_object(match_fields)},
        sort_keys=True,
        separators=(",", ":"),
    )


def state_for_row(row: dict[str, Any]) -> dict[str, Any]:
    return {"id": row["id"], "object": row["object"], **row["match"], **row["fields"]}


def plan_one_object(
    object_api_name: str,
    object_records: list[dict[str, Any]],
    context: dict[str, Any],
) -> tuple[list[dict[str, Any]], dict[str, dict[str, Any]], int]:
    rows_by_signature: dict[str, dict[str, Any]] = {}
    ordered_signatures: list[str] = []
    local_records = dict(context["records"])

    for record in object_records:
        key = str(record["key"])
        match = materialize_value(record["match"], {**context, "records": local_records})
        fields = materialize_value(record["fields"], {**context, "records": local_records})
        signature = match_signature(object_api_name, match)
        row = rows_by_signature.get(signature)
        if row is None:
            row = {
                "key": key,
                "object": object_api_name,
                "id": stable_uuid(object_api_name, key),
                "match": match,
                "fields": fields,
                "alias_keys": [key],
            }
            rows_by_signature[signature] = row
            ordered_signatures.append(signature)
        else:
            row["fields"] = fields
            row["alias_keys"].append(key)
        state = state_for_row(row)
        for alias_key in row["alias_keys"]:
            local_records[alias_key] = state

    rows = [rows_by_signature[signature] for signature in ordered_signatures]
    state: dict[str, dict[str, Any]] = {}
    for row in rows:
        row_state = state_for_row(row)
        for alias_key in row["alias_keys"]:
            state[alias_key] = row_state
    collapsed = sum(max(0, len(row["alias_keys"]) - 1) for row in rows)
    return rows, state, collapsed


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


def sql_value(value: Any) -> str:
    if value is None:
        return "NULL"
    if isinstance(value, bool):
        return "1" if value else "0"
    if isinstance(value, (int, float)):
        return str(value)
    return sql_string(str(value))


def quote_identifier(identifier: str) -> str:
    return f"`{identifier.replace('`', '``')}`"


def mysql_base_args(args: argparse.Namespace) -> list[str]:
    missing = [
        name
        for name, value in [
            ("TIDB_HOST", args.db_host),
            ("TIDB_USER", args.db_user),
            ("TIDB_DATABASE", args.db_database),
        ]
        if not str(value or "").strip()
    ]
    if missing:
        raise PackReferenceDataError(f"missing database settings: {', '.join(missing)}")
    client = os.environ.get("PACK_REFERENCE_DATA_MYSQL_CLIENT", "").strip() or "mysql"
    command = [
        client,
        "--batch",
        "--raw",
        "--skip-column-names",
        "--connect-timeout=10",
        "-h",
        args.db_host,
        "-P",
        str(args.db_port),
        "-u",
        args.db_user,
        args.db_database,
    ]
    if str(args.db_tls).strip().lower() in {"", "1", "true", "yes", "on"}:
        command.insert(1, "--ssl")
    return command


def mysql_query(args: argparse.Namespace, sql: str) -> str:
    env = os.environ.copy()
    env["MYSQL_PWD"] = args.db_password
    try:
        result = subprocess.run(
            [*mysql_base_args(args), "-e", sql],
            check=True,
            capture_output=True,
            env=env,
            text=True,
        )
    except subprocess.CalledProcessError as exc:
        detail = (exc.stderr or exc.stdout or "mysql CLI command failed").strip()
        raise PackReferenceDataError(detail) from exc
    return result.stdout


def resolve_table_name(args: argparse.Namespace, object_api_name: str) -> str:
    sql = (
        "SELECT table_name FROM information_schema.tables "
        "WHERE table_schema = DATABASE() "
        f"AND LOWER(table_name) = LOWER({sql_string(object_api_name)}) "
        f"ORDER BY CASE WHEN table_name = {sql_string(object_api_name)} THEN 0 ELSE 1 END, table_name LIMIT 1"
    )
    table = mysql_query(args, sql).strip()
    if not table:
        raise PackReferenceDataError(f"missing target table for {object_api_name}")
    return table.splitlines()[0]


def fetch_columns(args: argparse.Namespace, table: str) -> dict[str, str]:
    sql = (
        "SELECT column_name FROM information_schema.columns "
        "WHERE table_schema = DATABASE() "
        f"AND table_name = {sql_string(table)} ORDER BY ordinal_position"
    )
    columns = [line.strip() for line in mysql_query(args, sql).splitlines() if line.strip()]
    return {column.lower(): column for column in columns}


def fetch_admin_user_id(args: argparse.Namespace, columns: dict[str, str]) -> str:
    if not any(column.lower() in columns for column in AUDIT_USER_COLUMNS):
        return ""
    username = os.environ.get("VERIFY_USERNAME", "admin@example.invalid").strip() or "admin@example.invalid"
    rows = mysql_query(args, f"SELECT Id FROM `User` WHERE Username = {sql_value(username)} LIMIT 1").strip().splitlines()
    if not rows:
        raise PackReferenceDataError(f"{username} is required for reference-data audit columns")
    return rows[0]


def preflight_database(args: argparse.Namespace, seed: dict[str, Any], selected: list[str]) -> None:
    """Fail before any write when a manifest record cannot target the runtime schema."""
    records_by_object: dict[str, list[dict[str, Any]]] = {object_api_name: [] for object_api_name in selected}
    for record in seed["baseline_records"]:
        records_by_object[str(record["object"])].append(record)

    for object_api_name in selected:
        table = resolve_table_name(args, object_api_name)
        columns = fetch_columns(args, table)
        required = {"Id"}
        for record in records_by_object[object_api_name]:
            required.update(record["match"])
            required.update(record["fields"])
        missing = sorted(field for field in required if field.lower() not in columns)
        if missing:
            raise PackReferenceDataError(
                f"{object_api_name} missing target columns: {', '.join(missing)}"
            )
        collided = soft_deleted_collisions(
            args, object_api_name, table, columns, records_by_object[object_api_name]
        )
        if collided:
            shown = collided[:5]
            listed = ", ".join(shown) + (", ..." if len(collided) > len(shown) else "")
            raise PackReferenceDataError(
                f"{object_api_name} has {len(collided)} soft-deleted rows the seed would land on "
                f"({listed}); the upsert would write into rows nothing can read -- restore them, "
                f"or hard-delete them if the retirement was intended, then re-run"
            )
        fetch_admin_user_id(args, columns)


def soft_deleted_collisions(
    args: argparse.Namespace,
    object_api_name: str,
    table: str,
    columns: dict[str, str],
    records: list[dict[str, Any]],
) -> list[str]:
    """Ids of soft-deleted rows this seed would land on, by either route.

    This has to run before any write, not after. `apply_rows` sends its batches through the
    mysql CLI with no enclosing transaction, so by the time a post-upsert count notices the
    shortfall the write has already committed -- and earlier objects in the layer have
    committed too. Checking here fails the whole phase with nothing applied.

    Both routes have to be searched, because neither contains the other:

    * by **match key** -- a row that pre-dates the seed holds a native id, so the insert
      collides on a natural unique key rather than the primary key and the update lands on the
      deleted row, while no row carrying the seed id exists at all.
    * by **seed id** -- the ids are derived from `key`, not from `match`. An operator who
      edits an updateable match field and then deletes the row leaves it carrying the seed id
      while its match no longer matches, so the insert collides on the primary key instead.

    The id set is taken from every record key, which is a superset of the ids actually written
    when two keys collapse onto one match signature. A soft-deleted row carrying any id in
    this seed's derived namespace is worth stopping for, and erring wide is the safe direction
    for a gate.
    """
    deleted_column = columns.get("isdeleted")
    if not deleted_column or not records:
        return []
    where_parts: list[str] = []
    for record in records:
        conditions = []
        for field, value in record["match"].items():
            if isinstance(value, str) and TEMPLATE_RE.search(value):
                # Match values are materialized after preflight. None carry templates today,
                # and a probe that silently compared the literal `{{ ... }}` would under-report
                # rather than fail, so refuse instead of guessing.
                raise PackReferenceDataError(
                    f"{object_api_name}.{field} uses a template in its match; the soft-delete "
                    f"preflight cannot resolve it before the write"
                )
            conditions.append(
                f"{quote_identifier(columns[field.lower()])} <=> {sql_value(value)}"
            )
        where_parts.append("(" + " AND ".join(conditions) + ")")
    seed_ids = sorted({stable_uuid(object_api_name, str(record["key"])) for record in records})
    where_parts.append(
        f"{quote_identifier(columns['id'])} IN "
        f"({', '.join(sql_value(seed_id) for seed_id in seed_ids)})"
    )
    sql = (
        f"SELECT {quote_identifier(columns['id'])} FROM {quote_identifier(table)} "
        f"WHERE ({' OR '.join(where_parts)}) "
        f"AND COALESCE({quote_identifier(deleted_column)}, 0) = 1"
    )
    # One column is selected, so each line is the id.
    return sorted(line.strip() for line in mysql_query(args, sql).splitlines() if line.strip())


def normalize_match_value(value: Any) -> str:
    if value is None:
        return "\\N"
    if isinstance(value, bool):
        return "1" if value else "0"
    return str(value)


def row_match_key(match: dict[str, Any], fields: tuple[str, ...]) -> tuple[str, ...]:
    return tuple(normalize_match_value(match.get(field)) for field in fields)


def pre_resolve_existing_ids(
    args: argparse.Namespace,
    object_api_name: str,
    rows: list[dict[str, Any]],
) -> tuple[str, dict[str, str]]:
    table = resolve_table_name(args, object_api_name)
    columns = fetch_columns(args, table)
    if "id" not in columns:
        raise PackReferenceDataError(f"{object_api_name} missing Id column")
    groups: dict[tuple[str, ...], list[dict[str, Any]]] = {}
    for row in rows:
        fields = tuple(row["match"].keys())
        for field in fields:
            if field.lower() not in columns:
                raise PackReferenceDataError(f"{object_api_name}.{field} is missing from target table")
        groups.setdefault(fields, []).append(row)

    for fields, group_rows in groups.items():
        where_parts = []
        for row in group_rows:
            conditions = [
                f"{quote_identifier(columns[field.lower()])} <=> {sql_value(row['match'][field])}"
                for field in fields
            ]
            where_parts.append("(" + " AND ".join(conditions) + ")")
        select_columns = [quote_identifier(columns["id"])] + [
            quote_identifier(columns[field.lower()]) for field in fields
        ]
        # Soft-deleted rows must not participate in match resolution. Correcting a period on
        # an object whose key fields are immutable means deleting and re-creating it, which
        # leaves the old row behind; counting it here would either raise "duplicate existing
        # match" against its live replacement or adopt an invisible row's id.
        deleted_column = columns.get("isdeleted")
        # COALESCE, not a bare `= 0`: reading NULL as deleted would hide an existing row from
        # match resolution and produce a duplicate insert or a natural-key failure. Both
        # spellings are common in this repo -- the record read path emits a bare `= 0`
        # (`Builder.ExcludeDeleted`) while the hand-written SQL layer this script belongs to
        # uses COALESCE. No seeded table can hold a NULL today (`IsDeleted` is NOT NULL
        # DEFAULT 0), so this is the safe side of a choice that is currently unobservable.
        live_predicate = (
            f" AND COALESCE({quote_identifier(deleted_column)}, 0) = 0" if deleted_column else ""
        )
        sql = (
            f"SELECT {', '.join(select_columns)} FROM {quote_identifier(table)} "
            f"WHERE ({' OR '.join(where_parts)}){live_predicate}"
        )
        existing: dict[tuple[str, ...], str] = {}
        for line in mysql_query(args, sql).splitlines():
            parts = line.split("\t")
            if len(parts) != len(fields) + 1:
                continue
            key = tuple(parts[1:])
            if key in existing:
                raise PackReferenceDataError(
                    f"duplicate existing {object_api_name} match for {key}"
                )
            existing[key] = parts[0]
        for row in group_rows:
            existing_id = existing.get(row_match_key(row["match"], fields))
            if existing_id:
                row["id"] = existing_id
    return table, columns


def normalize_row(
    object_api_name: str,
    row: dict[str, Any],
    columns: dict[str, str],
    admin_user_id: str,
) -> dict[str, Any]:
    raw = {"Id": row["id"], **row["match"], **row["fields"]}
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
    for column in AUDIT_DATETIME_COLUMNS:
        if column.lower() in columns and column not in raw:
            raw[column] = now
    if "isdeleted" in columns and "IsDeleted" not in raw:
        raw["IsDeleted"] = False
    if admin_user_id:
        for column in AUDIT_USER_COLUMNS:
            if column.lower() in columns and column not in raw:
                raw[column] = admin_user_id

    normalized: dict[str, Any] = {}
    for field, value in raw.items():
        actual = columns.get(field.lower())
        if actual is None:
            raise PackReferenceDataError(f"{object_api_name}.{field} is missing from target table")
        normalized[actual] = value
    return normalized


def update_columns(row: dict[str, Any], columns: dict[str, str]) -> set[str]:
    result: set[str] = set()
    for field in {**row["match"], **row["fields"]}:
        actual = columns.get(field.lower())
        if actual is None:
            raise PackReferenceDataError(f"{row['object']}.{field} is missing from target table")
        if actual not in NEVER_UPDATE_COLUMNS:
            result.add(actual)
    for field in ("LastModifiedDate", "SystemModstamp"):
        actual = columns.get(field.lower())
        if actual:
            result.add(actual)
    return result


def apply_rows(
    args: argparse.Namespace,
    object_api_name: str,
    rows: list[dict[str, Any]],
    table: str,
    columns: dict[str, str],
) -> dict[str, Any]:
    started = time.perf_counter()
    admin_user_id = fetch_admin_user_id(args, columns)
    normalized_rows = [normalize_row(object_api_name, row, columns, admin_user_id) for row in rows]
    grouped: dict[tuple[tuple[str, ...], tuple[str, ...]], list[dict[str, Any]]] = {}
    for row, values in zip(rows, normalized_rows):
        insert_columns = tuple(column for column in columns.values() if column in values)
        if columns["id"] not in insert_columns:
            raise PackReferenceDataError(f"{object_api_name} insert rows do not include Id")
        insert_columns = (columns["id"],) + tuple(
            column for column in insert_columns if column != columns["id"]
        )
        update_names = tuple(sorted(update_columns(row, columns) - {columns["id"]}))
        grouped.setdefault((insert_columns, update_names), []).append(values)

    statements: list[str] = []
    for (insert_columns, update_names), batch_rows in grouped.items():
        update_clause = ", ".join(
            f"{quote_identifier(column)} = VALUES({quote_identifier(column)})"
            for column in update_names
        ) or f"{quote_identifier(columns['id'])} = {quote_identifier(columns['id'])}"
        value_rows = [
            "(" + ", ".join(sql_value(row[column]) for column in insert_columns) + ")"
            for row in batch_rows
        ]
        statements.append(
            f"INSERT INTO {quote_identifier(table)} "
            f"({', '.join(quote_identifier(column) for column in insert_columns)}) "
            f"VALUES {', '.join(value_rows)} ON DUPLICATE KEY UPDATE {update_clause}"
        )
    mysql_query(args, ";\n".join(statements))
    ids = ", ".join(sql_value(row["id"]) for row in rows)
    # A row that exists but is soft-deleted is not a seeded row -- nothing can read it. Counting
    # it here is what let an upsert land on an invisible row and still report success, which is
    # the failure this verification exists to catch.
    deleted_column = columns.get("isdeleted")
    live_predicate = (
        f" AND COALESCE({quote_identifier(deleted_column)}, 0) = 0" if deleted_column else ""
    )
    found = int(
        mysql_query(
            args,
            f"SELECT COUNT(*) FROM {quote_identifier(table)} "
            f"WHERE {quote_identifier(columns['id'])} IN ({ids}){live_predicate}",
        ).strip()
        or "0"
    )
    if found != len(rows):
        # Backstop only: preflight already refused to start when a seeded match key sat on
        # a soft-deleted row. This catches a row deleted between that check and this write.
        raise PackReferenceDataError(
            f"{object_api_name} verification found {found}/{len(rows)} live rows"
        )
    return {
        "object": object_api_name,
        "rows": len(rows),
        "aliases": sum(len(row["alias_keys"]) for row in rows),
        "batches": len(statements),
        "duration_ms": round((time.perf_counter() - started) * 1000),
    }


def plan_and_apply(
    args: argparse.Namespace,
    seed: dict[str, Any],
    selected: list[str],
) -> tuple[dict[str, dict[str, Any]], list[list[str]], list[dict[str, Any]], int]:
    grouped: dict[str, list[dict[str, Any]]] = {object_api_name: [] for object_api_name in selected}
    for record in seed["baseline_records"]:
        grouped[str(record["object"])].append(record)
    pending = {object_api_name: records for object_api_name, records in grouped.items() if records}
    context: dict[str, Any] = {"records": {}}
    layers: list[list[str]] = []
    results: list[dict[str, Any]] = []
    collapsed = 0

    while pending:
        successes: list[tuple[str, list[dict[str, Any]], dict[str, dict[str, Any]], int]] = []
        unresolved: dict[str, str] = {}
        layer_context = {**context, "records": dict(context["records"])}
        for object_api_name, records in list(pending.items()):
            try:
                rows, state, object_collapsed = plan_one_object(object_api_name, records, layer_context)
            except MissingReference as exc:
                unresolved[object_api_name] = str(exc)
                continue
            successes.append((object_api_name, rows, state, object_collapsed))

        if not successes:
            detail = "; ".join(f"{name}: {reason}" for name, reason in unresolved.items())
            raise PackReferenceDataError(f"unresolved reference-data dependencies: {detail}")

        layer: list[str] = []
        for object_api_name, rows, _state, object_collapsed in successes:
            if args.dry_run:
                table = ""
                columns: dict[str, str] = {}
            else:
                table, columns = pre_resolve_existing_ids(args, object_api_name, rows)
            state = {}
            for row in rows:
                row_state = state_for_row(row)
                for alias_key in row["alias_keys"]:
                    state[alias_key] = row_state
            context["records"].update(state)
            if not args.dry_run:
                results.append(apply_rows(args, object_api_name, rows, table, columns))
            collapsed += object_collapsed
            pending.pop(object_api_name)
            layer.append(object_api_name)
        layers.append(layer)

    return context["records"], layers, results, collapsed


def write_json(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")


def main() -> int:
    args = parse_args()
    started = time.perf_counter()
    manifest_file = args.manifest_file.resolve()
    manifest = load_json(manifest_file, label="reference-data manifest")
    specs = manifest_object_specs(manifest)
    selected = parse_object_list(args.objects)
    unknown = [item for item in selected if item not in specs]
    if unknown:
        raise PackReferenceDataError(
            f"reference-data object is not declared in manifest: {', '.join(unknown)}; "
            f"allowed={','.join(specs)}"
        )
    validate_dependency_closure(selected, specs)
    seed = build_seed(manifest_file, selected, specs)
    if not args.dry_run:
        preflight_database(args, seed, selected)
    records, layers, apply_results, collapsed = plan_and_apply(args, seed, selected)
    summary = {
        "dry_run": args.dry_run,
        "apply_engine": "mysql-cli",
        "reference_data_objects": selected,
        "reference_data_record_count": len({state["id"] for state in records.values()}),
        "reference_data_alias_count": len(records),
        "collapsed_duplicate_records": collapsed,
        "layers": layers,
        "unsupported": [],
        "coverage_complete": True,
        "apply_results": apply_results,
        "duration_ms": round((time.perf_counter() - started) * 1000),
    }
    write_json(args.state_file, {"records": records})
    write_json(args.summary_file, summary)
    print(json.dumps(summary, indent=2, sort_keys=True))
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except PackReferenceDataError as exc:
        print(f"pack reference data error: {exc}", file=sys.stderr)
        raise SystemExit(1)