#!/usr/bin/env python3
"""Runnable SQLite mirror for Signal Trade CRM records.

Stores the latest Signal customer, opportunity, task and activity payloads as
JSON keyed by their stable Signal IDs. The SQLite checkpoint is committed in
the same transaction as records delivered by each change-feed poll.
"""

from __future__ import annotations

import json
import os
import sqlite3
from pathlib import Path
from typing import Any

from signal_trade_stock_sync import (
    get_trade_event_checkpoint,
    load_all_crm,
    sync_trade_records,
)


KINDS = ("customers", "opportunities", "tasks", "activities")
EVENT_KIND = {
    "customers": "customer",
    "opportunities": "opportunity",
    "tasks": "task",
    "activities": "activity",
}

api_key = os.environ["SIGNAL_TRADE_API_KEY"]
base_url = os.getenv("SIGNAL_TRADE_BASE_URL", "https://signalautomotive.uk")
database_path = Path(os.getenv("SIGNAL_TRADE_CRM_SQLITE_PATH", "signal-trade-crm.db"))
database_path.parent.mkdir(parents=True, exist_ok=True)
db = sqlite3.connect(database_path, timeout=10)
db.execute("PRAGMA busy_timeout = 10000")
db.execute("PRAGMA journal_mode = WAL")
db.execute("PRAGMA synchronous = FULL")

db.executescript(
    """
    CREATE TABLE IF NOT EXISTS signal_crm_records (
        kind TEXT NOT NULL,
        id TEXT NOT NULL,
        payload_json TEXT NOT NULL,
        PRIMARY KEY(kind, id)
    );
    CREATE TABLE IF NOT EXISTS signal_sync_state (
        name TEXT PRIMARY KEY,
        value TEXT NOT NULL
    );
    """
)
db.commit()


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


def upsert(kind: str, record: dict[str, Any]) -> None:
    record_id = record.get("id")
    if not isinstance(record_id, str) or not record_id:
        raise RuntimeError(f"Signal Trade returned a {kind} record without an ID")
    db.execute(
        "INSERT INTO signal_crm_records(kind,id,payload_json) VALUES(?,?,?) "
        "ON CONFLICT(kind,id) DO UPDATE SET payload_json=excluded.payload_json",
        (kind, record_id, encode(record)),
    )


def read_state(name: str) -> str | None:
    row = db.execute("SELECT value FROM signal_sync_state WHERE name=?", (name,)).fetchone()
    return row[0] if row else None


def write_state(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),
    )
    # When called by sync_trade_records, record upserts above and this
    # checkpoint share one transaction. A crash replays the old window.
    db.commit()


try:
    checkpoint = read_state("crm_checkpoint")
    if read_state("crm_initial_sync_complete") != "1":
        # Take the checkpoint first so changes made during the full load are
        # picked up by the following event poll.
        checkpoint = get_trade_event_checkpoint(api_key, "crm", base_url)
        for kind in KINDS:
            count = load_all_crm(
                api_key,
                kind,
                on_record=lambda record, record_kind=kind: upsert(record_kind, record),
                base_url=base_url,
            )
            print(f"Loaded {count} {kind}")
        with db:
            write_state("crm_checkpoint", checkpoint)
            write_state("crm_initial_sync_complete", "1")

    result = sync_trade_records(
        api_key,
        "crm",
        checkpoint,
        on_record=lambda record, event: upsert(
            next(kind for kind, event_kind in EVENT_KIND.items() if event_kind == event["type"]),
            record,
        ),
        save_checkpoint=lambda value: write_state("crm_checkpoint", value),
        base_url=base_url,
    )
    print(json.dumps({"database": str(database_path), **result}, ensure_ascii=False))
finally:
    db.close()