"""SQLite 台账：断点、幂等判断与审计的唯一事实源。"""

from __future__ import annotations

import os
import sqlite3
import threading
import time
from typing import Any

SCHEMA = """
CREATE TABLE IF NOT EXISTS runs (
    run_id      TEXT PRIMARY KEY,
    started_at  TEXT NOT NULL,
    finished_at TEXT,
    sprint      TEXT NOT NULL,
    work_dir    TEXT NOT NULL,
    dry_run     INTEGER NOT NULL DEFAULT 0,
    status      TEXT NOT NULL DEFAULT 'RUNNING'
);

CREATE TABLE IF NOT EXISTS jobs (
    run_id       TEXT NOT NULL,
    project_id   TEXT NOT NULL,
    project_name TEXT NOT NULL,
    system_name  TEXT NOT NULL,
    source_file  TEXT NOT NULL,
    action       TEXT NOT NULL,
    stage        TEXT NOT NULL,
    worksheet_id INTEGER,
    version_id   INTEGER,
    reason       TEXT,
    error        TEXT,
    updated_at   TEXT NOT NULL,
    PRIMARY KEY (run_id, project_id)
);

CREATE TABLE IF NOT EXISTS events (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id     TEXT NOT NULL,
    project_id TEXT,
    stage      TEXT NOT NULL,
    action     TEXT NOT NULL,
    detail     TEXT,
    elapsed_ms INTEGER,
    created_at TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_jobs_project ON jobs (project_id, system_name, updated_at);
CREATE INDEX IF NOT EXISTS idx_events_run ON events (run_id, id);
"""


def now() -> str:
    return time.strftime("%Y-%m-%d %H:%M:%S")


class Store:
    def __init__(self, db_path: str) -> None:
        directory = os.path.dirname(os.path.abspath(db_path))
        os.makedirs(directory, exist_ok=True)
        self._lock = threading.Lock()
        self._conn = sqlite3.connect(db_path, check_same_thread=False)
        self._conn.row_factory = sqlite3.Row
        with self._lock:
            self._conn.executescript(SCHEMA)
            self._conn.commit()

    def close(self) -> None:
        with self._lock:
            self._conn.close()

    # ------------------------------------------------------------------ runs

    def start_run(self, run_id: str, sprint: str, work_dir: str, dry_run: bool) -> None:
        with self._lock:
            self._conn.execute(
                "INSERT OR REPLACE INTO runs (run_id, started_at, sprint, work_dir, dry_run, status) "
                "VALUES (?, ?, ?, ?, ?, 'RUNNING')",
                (run_id, now(), sprint, work_dir, int(dry_run)),
            )
            self._conn.commit()

    def finish_run(self, run_id: str, status: str = "DONE") -> None:
        with self._lock:
            self._conn.execute(
                "UPDATE runs SET finished_at = ?, status = ? WHERE run_id = ?", (now(), status, run_id)
            )
            self._conn.commit()

    # ------------------------------------------------------------------ jobs

    def upsert_job(
        self,
        run_id: str,
        *,
        project_id: str,
        project_name: str,
        system_name: str,
        source_file: str,
        action: str,
        stage: str,
        worksheet_id: int | None = None,
        version_id: int | None = None,
        reason: str = "",
        error: str = "",
    ) -> None:
        with self._lock:
            self._conn.execute(
                """
                INSERT INTO jobs (run_id, project_id, project_name, system_name, source_file, action, stage,
                                  worksheet_id, version_id, reason, error, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(run_id, project_id) DO UPDATE SET
                    action = excluded.action,
                    stage = excluded.stage,
                    worksheet_id = COALESCE(excluded.worksheet_id, jobs.worksheet_id),
                    version_id = COALESCE(excluded.version_id, jobs.version_id),
                    reason = excluded.reason,
                    error = excluded.error,
                    updated_at = excluded.updated_at
                """,
                (
                    run_id,
                    project_id,
                    project_name,
                    system_name,
                    source_file,
                    str(action),
                    stage,
                    worksheet_id,
                    version_id,
                    reason,
                    error,
                    now(),
                ),
            )
            self._conn.commit()

    def latest_stage(self, project_id: str, system_name: str) -> str | None:
        """跨 run 取该项目该 Sprint 最近一次落库的阶段，用于断点续跑。"""
        job = self.latest_job(project_id, system_name)
        return job["stage"] if job else None

    def latest_job(self, project_id: str, system_name: str) -> dict[str, Any] | None:
        """最近一次**真实**执行记录。dry-run 不参与幂等判断，否则会把演练当成已完成。

        `SKIPPED` 行也不参与：它只表示「这一轮不用做」，不带任何进度信息。若把它当成
        最近记录，下次 plan 会把「已推送 TFS」看成「台账停在 SKIPPED」，误判 resume 并
        再推一次 —— 而 TFS 推送接口没有服务端去重，重复推送就是重复工作项。
        """
        with self._lock:
            row = self._conn.execute(
                """
                SELECT j.* FROM jobs j
                JOIN runs r ON r.run_id = j.run_id
                WHERE j.project_id = ? AND j.system_name = ? AND r.dry_run = 0
                  AND j.stage <> 'SKIPPED'
                ORDER BY j.updated_at DESC LIMIT 1
                """,
                (project_id, system_name),
            ).fetchone()
        return dict(row) if row else None

    def jobs(self, run_id: str) -> list[dict[str, Any]]:
        with self._lock:
            rows = self._conn.execute(
                "SELECT * FROM jobs WHERE run_id = ? ORDER BY project_name", (run_id,)
            ).fetchall()
        return [dict(row) for row in rows]

    # ------------------------------------------------------------------ events

    def add_event(
        self,
        run_id: str,
        *,
        stage: str,
        action: str,
        project_id: str = "",
        detail: str = "",
        elapsed_ms: int = 0,
    ) -> None:
        with self._lock:
            self._conn.execute(
                "INSERT INTO events (run_id, project_id, stage, action, detail, elapsed_ms, created_at) "
                "VALUES (?, ?, ?, ?, ?, ?, ?)",
                (run_id, project_id, stage, action, detail[:2000], elapsed_ms, now()),
            )
            self._conn.commit()

    def events(self, run_id: str) -> list[dict[str, Any]]:
        with self._lock:
            rows = self._conn.execute(
                "SELECT * FROM events WHERE run_id = ? ORDER BY id", (run_id,)
            ).fetchall()
        return [dict(row) for row in rows]

    def run_info(self, run_id: str) -> dict[str, Any] | None:
        with self._lock:
            row = self._conn.execute("SELECT * FROM runs WHERE run_id = ?", (run_id,)).fetchone()
        return dict(row) if row else None

    def last_run(self) -> str | None:
        with self._lock:
            row = self._conn.execute("SELECT run_id FROM runs ORDER BY started_at DESC LIMIT 1").fetchone()
        return row["run_id"] if row else None