#!/usr/bin/env python3
"""Recoverable, account-bound SQLite stock mirror using only the standard library.

Run normally for incremental sync and daily full reconciliation. --reconcile
forces a full replacement; --rebind explicitly accepts a new key/scope for the
same dealership, or adopts a legacy unbound database. --status is local only.
The old committed mirror remains readable while a single writer syncs.
"""
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_stock_event_checkpoint, get_stock_sites, get_trade_api_account,
    load_all_stock, sync_stock,
)

MIRROR_VERSION = "1"
STOCK_SCOPES = {"stock:read", "stock:read_write", "stock:manage", "stock:finance_read", "stock:finance_manage"}
TABLES = (
    "CREATE TABLE IF NOT EXISTS signal_stock (id TEXT PRIMARY KEY,payload_json TEXT NOT NULL)",
    "CREATE TABLE IF NOT EXISTS signal_sites (id TEXT PRIMARY KEY,payload_json TEXT NOT NULL)",
    "CREATE TABLE IF NOT EXISTS signal_sync_state (name TEXT PRIMARY KEY,value TEXT NOT NULL)",
)


def normalise_origin(base_url: str) -> str:
    parsed = urlsplit(base_url)
    host = (parsed.hostname or "").lower()
    scheme = 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"}):
        raise ValueError("Signal Trade requests require HTTPS (HTTP is allowed only for loopback examples)")
    if parsed.username or parsed.password or parsed.query or parsed.fragment:
        raise ValueError("The API base URL must not contain credentials, a 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 read_state(db: sqlite3.Connection) -> dict[str, str]:
    return dict(db.execute("SELECT name,value FROM signal_sync_state"))


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


def account_binding(account: dict[str, Any], origin: str) -> dict[str, str]:
    if account["apiKey"]["scope"] not in STOCK_SCOPES:
        raise RuntimeError("This key does not have stock access. Use a stock-scoped key")
    return {"mirror_version": MIRROR_VERSION, "origin": origin, "business_id": account["business"]["id"], "key_id": account["apiKey"]["id"], "scope": account["apiKey"]["scope"]}


def utc_time(value: str) -> datetime:
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError("A mirror timestamp must include its timezone")
    return parsed.astimezone(timezone.utc)


def canonical_time(value: str) -> str:
    parsed = utc_time(value)
    if parsed.isoformat(timespec="milliseconds").replace("+00:00", "Z") != value:
        raise ValueError("Signal Trade must return a canonical UTC timestamp")
    return value


def status(database_path: Path) -> dict[str, Any]:
    if not database_path.exists():
        return {"database": str(database_path), "initialized": False, "vehicles": 0, "sites": 0}
    db = sqlite3.connect(database_path.resolve().as_uri() + "?mode=ro", uri=True)
    try:
        tables = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table'")}
        state = read_state(db) if "signal_sync_state" in tables else {}
        return {"database": str(database_path), "initialized": state.get("stock_initial_sync_complete") == "1",
                "binding": {name: state.get(name) for name in ("mirror_version", "origin", "business_id", "key_id", "scope")},
                "checkpoint": state.get("stock_checkpoint"), "lastReconciledAt": state.get("stock_last_reconciled_at"),
                "vehicles": db.execute("SELECT count(*) FROM signal_stock").fetchone()[0] if "signal_stock" in tables else 0,
                "sites": db.execute("SELECT count(*) FROM signal_sites").fetchone()[0] if "signal_sites" in tables else 0}
    finally:
        db.close()


def run_sync(database_path: Path, api_key: str, base_url: str, *, reconcile: bool = False, rebind: bool = False) -> dict[str, Any]:
    origin = normalise_origin(base_url)
    base_url = origin
    database_path.parent.mkdir(parents=True, exist_ok=True)
    db = sqlite3.connect(database_path, timeout=10, isolation_level=None)
    try:
        db.execute("PRAGMA busy_timeout = 10000")
        db.execute("PRAGMA journal_mode = WAL")
        db.execute("PRAGMA synchronous = FULL")
        # Acquire the only writer before remote reads. No callback commits.
        db.execute("BEGIN IMMEDIATE")
        existing_tables = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table'")}
        existing_mirror = bool(existing_tables.intersection({"signal_stock", "signal_sites", "signal_sync_state"}))
        for table in TABLES:
            db.execute(table)
        state = read_state(db)
        if state.get("origin") and state["origin"] != origin:
            raise RuntimeError("This database belongs to another API origin. Use a separate database")
        if state.get("mirror_version") not in {None, MIRROR_VERSION}:
            raise RuntimeError("This mirror version is unsupported. Preserve the database and use a compatible client")
        account = get_trade_api_account(api_key, base_url)
        binding = account_binding(account, origin)
        if state.get("business_id") and state["business_id"] != binding["business_id"]:
            raise RuntimeError("This database belongs to another dealership. Use a separate database")
        bound = all(state.get(name) for name in binding)
        if not bound and existing_mirror and not rebind:
            raise RuntimeError("This legacy or incomplete mirror is unbound. Review the dealership and run --rebind for a full replacement")
        if bound and any(state[name] != value for name, value in binding.items()) and not rebind:
            raise RuntimeError("The key or permission changed. Run --rebind for a full replacement of this dealership mirror")
        checkpoint = state.get("stock_checkpoint")
        last_reconciled = state.get("stock_last_reconciled_at")
        due = not last_reconciled
        if last_reconciled:
            try:
                due = (utc_time(account["generatedAt"]) - utc_time(canonical_time(last_reconciled))).total_seconds() >= 86400
            except (TypeError, ValueError, OverflowError):
                due = True
        full = reconcile or rebind or not bound or state.get("stock_initial_sync_complete") != "1" or not checkpoint or due
        if not full:
            canonical_time(checkpoint)

        def upsert_vehicle(vehicle: dict[str, Any]) -> None:
            if not isinstance(vehicle.get("updatedAt"), str):
                raise RuntimeError("Signal Trade returned stock without its recorded revision")
            canonical_time(vehicle["updatedAt"])
            db.execute("INSERT INTO signal_stock(id,payload_json) VALUES(?,?) ON CONFLICT(id) DO UPDATE SET payload_json=excluded.payload_json", (vehicle["id"], encode(vehicle)))

        def remove_vehicle(record_id: str) -> None:
            db.execute("DELETE FROM signal_stock WHERE id=?", (record_id,))

        def replace_sites(sites: list[dict[str, Any]]) -> None:
            if not isinstance(sites, list) or any(not isinstance(site, dict) or not isinstance(site.get("id"), str) or not site["id"] or not isinstance(site.get("name"), str) or not site["name"].strip() for site in sites):
                raise RuntimeError("Signal Trade returned an invalid branch list")
            db.execute("DELETE FROM signal_sites")
            db.executemany("INSERT INTO signal_sites(id,payload_json) VALUES(?,?)", [(site["id"], encode(site)) for site in sites])

        loaded = 0
        if full:
            checkpoint = canonical_time(get_stock_event_checkpoint(api_key, base_url))
            # Deletes and page writes stay private until full loading, catch-up
            # and final authorization succeed. Failure restores the old mirror.
            db.execute("DELETE FROM signal_stock")
            db.execute("DELETE FROM signal_sites")
            loaded = load_all_stock(api_key, upsert_vehicle, replace_sites, base_url)
        result = sync_stock(api_key, checkpoint, upsert_vehicle,
                            lambda value: write_state(db, "stock_checkpoint", canonical_time(value)),
                            on_delete=remove_vehicle,
                            on_sites_changed=lambda: replace_sites(get_stock_sites(api_key, base_url)),
                            base_url=base_url)
        canonical_time(result["checkpoint"])
        if utc_time(result["checkpoint"]) < utc_time(checkpoint):
            raise RuntimeError("The stock checkpoint moved backward. The previous mirror was kept")
        for name, value in binding.items():
            write_state(db, name, value)
        write_state(db, "stock_initial_sync_complete", "1")
        if full:
            write_state(db, "stock_last_reconciled_at", result["checkpoint"])
        count = db.execute("SELECT count(*) FROM signal_stock").fetchone()[0]
        # Nothing may follow this remote check except comparison and commit.
        final = get_trade_api_account(api_key, base_url)
        if account_binding(final, origin) != binding:
            raise RuntimeError("The dealership, key or permission changed during sync. The previous mirror was kept")
        db.commit()
        return {"database": str(database_path), "businessId": binding["business_id"], "scope": binding["scope"], "reconciled": full, "vehiclesLoaded": loaded, "vehicles": count, **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("--reconcile", action="store_true", help="Replace the mirror from current stock and replay changes atomically")
    actions.add_argument("--rebind", action="store_true", help="Accept a new key/scope for this dealership and fully replace its mirror")
    actions.add_argument("--status", action="store_true", help="Show local counts, binding and checkpoint without an API request")
    args = parser.parse_args(argv)
    path = Path(os.getenv("SIGNAL_TRADE_SQLITE_PATH", "signal-trade-stock.db"))
    key = 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 stock key")
            result = run_sync(path, key, os.getenv("SIGNAL_TRADE_BASE_URL", "https://signalautomotive.uk"), reconcile=args.reconcile, rebind=args.rebind)
        print(encode(result))
        return 0
    except (RuntimeError, ValueError, sqlite3.Error, OSError) as error:
        # Request failures never include the Authorization header or payloads.
        text = str(error).replace(key, "[redacted]") if key else str(error)
        print(f"Stock sync stopped: {text}", file=sys.stderr)
        return 1


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