#!/usr/bin/env python3
"""Private native appraisal mirror using only Python's standard library.

Download signal_trade_stock_sync.py alongside this file. Run normally for change
sync and a daily full reconciliation on invocation; no scheduler is installed.
--reconcile replaces from current records. --rebind accepts a new key/scope for
the same dealership. --status reads local counts with no key or API call.
Seller/inspection facts are private. Protect this database; no files are fetched.
"""
from __future__ import annotations

import argparse
import json
import os
import sqlite3
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit

from signal_trade_stock_sync import get_trade_api_account, load_all_native_appraisals, sync_native_appraisal_records, create_native_appraisal_sync_transport

VERSION = "1"
SCOPES = {"appraisal:workflow", "appraisal:workflow_evidence"}
BINDING_NAMES = ("mirror_version", "origin", "business_id", "key_id", "token_prefix", "scope")
TABLE_NAMES = {"signal_native_appraisals", "signal_native_appraisal_sync_state"}


def normalise_origin(value: str) -> str:
    parsed = urlsplit(value)
    host, scheme = (parsed.hostname or "").lower(), parsed.scheme.lower()
    if (not host or scheme not in {"https", "http"} or scheme == "http" and host not in {"localhost", "127.0.0.1", "::1"}
            or parsed.username or parsed.password or parsed.query or parsed.fragment):
        raise ValueError("The API address must use HTTPS except on loopback, with no credentials, query or fragment")
    port = parsed.port
    host_text = f"[{host}]" if ":" in host else host
    suffix = f":{port}" if port is not None and port != (443 if scheme == "https" else 80) else ""
    return f"{scheme}://{host_text}{suffix}"


def encode(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), allow_nan=False)


def timestamp(value: Any) -> datetime | None:
    if not isinstance(value, str):
        return None
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
        if parsed.tzinfo is None or parsed.astimezone(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") != value:
            return None
        return parsed
    except (ValueError, OverflowError):
        return None


def valid_checkpoint(value: Any) -> bool:
    return isinstance(value, str) and 0 < len(value) <= 1024


def read_state(db: sqlite3.Connection) -> dict[str, str]:
    return dict(db.execute("SELECT name,value FROM signal_native_appraisal_sync_state"))


def binding(account: dict[str, Any], origin: str) -> dict[str, str]:
    if account["apiKey"]["scope"] not in SCOPES:
        raise RuntimeError("Use a native appraisal workflow key; figure-read/import keys cannot read private inspections")
    return {"mirror_version": VERSION, "origin": origin, "business_id": account["business"]["id"],
            "key_id": account["apiKey"]["id"], "token_prefix": account["apiKey"]["tokenPrefix"], "scope": account["apiKey"]["scope"]}


def local_status(db: sqlite3.Connection, path: Path) -> dict[str, Any]:
    state = read_state(db)
    return {"database": str(path.resolve()), "bound": all(state.get(name) for name in BINDING_NAMES),
            "ready": state.get("native_initial_sync_complete") == "1",
            "binding": {name: state.get(name) for name in BINDING_NAMES}, "checkpoint": state.get("native_checkpoint"),
            "lastReconciledAt": state.get("native_last_reconciled_at"),
            "appraisals": db.execute("SELECT count(*) FROM signal_native_appraisals").fetchone()[0]}


def status(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {"database": str(path.resolve()), "bound": False, "ready": False, "appraisals": 0}
    db = sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)
    try:
        existing = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table'")}.intersection(TABLE_NAMES)
        if not existing:
            return {"database": str(path.resolve()), "bound": False, "ready": False, "appraisals": 0}
        if existing != TABLE_NAMES:
            raise RuntimeError("This native mirror is incomplete. Review the database and run --rebind")
        return local_status(db, path)
    finally:
        db.close()


def run_sync(path: Path, api_key: str, base_url: str, *, reconcile: bool = False, rebind: bool = False, detail_batch_size: int = 10, request_spacing_ms: int = 600) -> dict[str, Any]:
    if type(detail_batch_size) is not int or not 1 <= detail_batch_size <= 10:
        raise ValueError("detail_batch_size must be a whole number from one to ten")
    origin = normalise_origin(base_url)
    if (type(request_spacing_ms) is not int or not 0 <= request_spacing_ms <= 60000
            or request_spacing_ms < 600 and urlsplit(origin).hostname not in {"localhost", "127.0.0.1", "::1"}):
        raise ValueError("Native request spacing must be 600..60000 ms; lower values are allowed only for loopback fixtures")
    transport = create_native_appraisal_sync_transport(api_key, origin, request_spacing_ms=request_spacing_ms)
    path.parent.mkdir(parents=True, exist_ok=True)
    db = sqlite3.connect(path, timeout=0, isolation_level=None)
    try:
        db.execute("PRAGMA busy_timeout=0")
        db.execute("PRAGMA journal_mode=WAL")
        db.execute("PRAGMA synchronous=FULL")
        # Hold one writer lock through all reads; callbacks never commit.
        db.execute("BEGIN IMMEDIATE")
        existing = bool({row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table'")}.intersection(TABLE_NAMES))
        db.execute("CREATE TABLE IF NOT EXISTS signal_native_appraisals(id TEXT PRIMARY KEY,payload_json TEXT NOT NULL)")
        db.execute("CREATE TABLE IF NOT EXISTS signal_native_appraisal_sync_state(name TEXT PRIMARY KEY,value TEXT NOT NULL)")
        saved = read_state(db)
        if saved.get("origin") and saved["origin"] != origin:
            raise RuntimeError("This mirror belongs to another API address. Use a separate database")
        if saved.get("mirror_version") not in {None, VERSION}:
            raise RuntimeError("This mirror storage version is unsupported. Use a separate database")
        account = get_trade_api_account(api_key, origin, transport=transport)
        current = binding(account, origin)
        if saved.get("business_id") and saved["business_id"] != current["business_id"]:
            raise RuntimeError("This mirror belongs to another dealership. Use a separate database")
        bound = all(saved.get(name) for name in BINDING_NAMES)
        if existing and not bound and not rebind:
            raise RuntimeError("This existing native mirror is unbound. Review the dealership and run --rebind for a full replacement")
        if bound and any(saved[name] != value for name, value in current.items()) and not rebind:
            raise RuntimeError("The connection key or scope changed. Run --rebind to fully replace this dealership mirror")
        server_time, reconciled_time = timestamp(account["generatedAt"]), timestamp(saved.get("native_last_reconciled_at"))
        if server_time is None:
            raise RuntimeError("Signal Trade returned a noncanonical server clock")
        full = (reconcile or rebind or saved.get("native_initial_sync_complete") != "1" or reconciled_time is None
                or server_time < reconciled_time or (server_time - reconciled_time).total_seconds() >= 86400)
        if not full and not valid_checkpoint(saved.get("native_checkpoint")):
            raise RuntimeError("The local checkpoint is invalid. Run --reconcile to recover it")

        def apply(detail: dict[str, Any], _event: dict[str, Any] | None = None) -> None:
            row = detail.get("appraisal")
            if not isinstance(row, dict) or not isinstance(row.get("id"), str) or timestamp(row.get("updatedAt")) is None:
                raise RuntimeError("A native appraisal is missing its current identity or revision")
            db.execute("INSERT INTO signal_native_appraisals(id,payload_json) VALUES(?,?) ON CONFLICT(id) DO UPDATE SET payload_json=excluded.payload_json",
                       (row["id"], encode(detail)))

        start, loaded = saved.get("native_checkpoint"), 0
        if full:
            db.execute("DELETE FROM signal_native_appraisals")
            discovery = load_all_native_appraisals(api_key, apply, origin, detail_batch_size=detail_batch_size, transport=transport)
            start, loaded = discovery["syncCheckpoint"], discovery["recordsUpdated"]
        confirmed: list[str] = []
        result = sync_native_appraisal_records(api_key, start, apply,
                    lambda record_id, _event: db.execute("DELETE FROM signal_native_appraisals WHERE id=?", (record_id,)),
                    lambda value: confirmed.append(value), origin, detail_batch_size=detail_batch_size, transport=transport)
        if len(confirmed) != 1 or not valid_checkpoint(confirmed[0]):
            raise RuntimeError("The native checkpoint was not confirmed")

        def write(name: str, value: str) -> None:
            db.execute("INSERT INTO signal_native_appraisal_sync_state(name,value) VALUES(?,?) ON CONFLICT(name) DO UPDATE SET value=excluded.value", (name, value))

        for name, value in current.items():
            write(name, value)
        write("native_checkpoint", confirmed[0])
        write("native_initial_sync_complete", "1")
        if full:
            write("native_last_reconciled_at", account["generatedAt"])
        current_status = local_status(db, path)
        # No remote or local work may follow this check except comparison/commit.
        final = get_trade_api_account(api_key, origin, transport=transport)
        if binding(final, origin) != current:
            raise RuntimeError("The dealership, key or permission changed. The previous mirror was kept")
        db.commit()
        return {**current_status, "mode": "reconciled" if full else "incremental", "recordsLoaded": loaded, **result}
    except BaseException:
        if db.in_transaction:
            db.rollback()
        raise
    finally:
        db.close()


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    actions = parser.add_mutually_exclusive_group()
    actions.add_argument("--status", action="store_true", help="Local counts only; no key or API request")
    actions.add_argument("--reconcile", action="store_true", help="Replace all records, then catch up before committing")
    actions.add_argument("--rebind", action="store_true", help="Accept a changed key/scope for the same dealership, with full replacement")
    args = parser.parse_args(argv)
    path, key = Path(os.getenv("SIGNAL_TRADE_SQLITE_PATH", "signal-native-appraisals.db")), os.getenv("SIGNAL_TRADE_API_KEY", "")
    try:
        if args.status:
            result = status(path)
        else:
            if not key:
                raise ValueError("Set SIGNAL_TRADE_API_KEY to a server-side native appraisal workflow key")
            batch_text = os.getenv("SIGNAL_TRADE_NATIVE_BATCH_SIZE") or "10"
            if batch_text not in {str(value) for value in range(1, 11)}:
                raise ValueError("SIGNAL_TRADE_NATIVE_BATCH_SIZE must be a whole number from one to ten")
            spacing_text = os.getenv("SIGNAL_TRADE_NATIVE_REQUEST_SPACING_MS", "600")
            if not spacing_text.isascii() or not spacing_text.isdecimal():
                raise ValueError("Native request spacing must be a whole number of milliseconds")
            result = run_sync(path, key, os.getenv("SIGNAL_TRADE_BASE_URL", "https://signalautomotive.uk"), reconcile=args.reconcile, rebind=args.rebind, detail_batch_size=int(batch_text), request_spacing_ms=int(spacing_text))
        print(encode(result))
        return 0
    except (RuntimeError, ValueError, sqlite3.Error, OSError) as error:
        text = str(error).replace(key, "[redacted]") if key else str(error)
        print(f"Native appraisal sync stopped: {text}", file=sys.stderr)
        return 1


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