import json
import logging
import os
import sqlite3
import stat
import threading

import pytest

from jira_hub import backup, db, paths, projects, queue, settings


def test_connect_creates_schema(open_db):
    conn = open_db()
    names = {r["name"] for r in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'")}
    assert {"setting", "issue", "issue_link", "issue_comment", "issue_note", "batch", "command", "sync_run",
            "agent_status", "project", "issue_change"} <= names
    assert "created_at" in {r["name"] for r in conn.execute("PRAGMA table_info(issue_comment)")}
    assert "history_at" in {r["name"] for r in conn.execute("PRAGMA table_info(sync_run)")}
    assert "project" in {r["name"] for r in conn.execute("PRAGMA table_info(batch)")}
    assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == "5"
    assert conn.execute("SELECT COUNT(*) AS n FROM agent_status").fetchone()["n"] == 1
    cols = {r["name"] for r in conn.execute("PRAGMA table_info(command)")}
    assert "deliveries" in cols


def _drop_v5(conn):
    """Снять с базы текущей схемы то, что добавила v5: так строятся базы прежних версий."""
    conn.execute("DROP TABLE issue_change")
    conn.execute("ALTER TABLE issue_comment DROP COLUMN created_at")
    conn.execute("ALTER TABLE sync_run DROP COLUMN history_at")


def _v1_database(path):
    """База в состоянии схемы v1: без command.deliveries, schema_version = 1, с живыми данными."""
    conn = db.connect(path)
    conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('x', 'claude', '2026-10-13T00:00:00+00:00')")
    for status in ("claimed", "done", "queued"):
        conn.execute("INSERT INTO command(batch_id, seq, type, params, status, claimed_at) VALUES (1, 1, 'refresh', '{}', ?, ?)",
                     (status, "2026-10-13T00:00:00+00:00" if status == "claimed" else None))
    conn.execute("INSERT INTO issue_note(key, fact_status) VALUES ('A-1', 'В работе')")
    _drop_v5(conn)
    conn.execute("ALTER TABLE command DROP COLUMN deliveries")
    conn.execute("ALTER TABLE batch DROP COLUMN project")
    conn.execute("DROP TABLE project")
    conn.execute("UPDATE setting SET value = '1' WHERE key = 'schema_version'")
    conn.close()


def test_migrates_v1_with_backup(tmp_path):
    path = tmp_path / "hub.sqlite"
    _v1_database(path)
    conn = db.connect(path)
    assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == "5"
    rows = conn.execute("SELECT status, deliveries FROM command ORDER BY id").fetchall()
    assert [(r["status"], r["deliveries"]) for r in rows] == [("claimed", 1), ("done", 0), ("queued", 0)]
    assert conn.execute("SELECT fact_status FROM issue_note WHERE key = 'A-1'").fetchone()["fact_status"] == "В работе"
    copies = list((tmp_path / "backups").glob("*-schema-v1.sqlite"))
    assert len(copies) == 1
    old = sqlite3.connect(copies[0])                       # не db.connect — он бы мигрировал копию
    assert "deliveries" not in {r[1] for r in old.execute("PRAGMA table_info(command)")}
    assert old.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()[0] == "1"
    old.close()
    conn.close()
    db.connect(path).close()                               # повторный connect не мигрирует и не копирует
    assert len(list((tmp_path / "backups").glob("*.sqlite"))) == 1


def test_new_database_makes_no_backup(tmp_path):
    db.connect(tmp_path / "hub.sqlite").close()
    assert not (tmp_path / "backups").exists()


def test_connect_is_idempotent(tmp_path):
    path = tmp_path / "hub.sqlite"
    db.connect(path).close()
    conn = db.connect(path)
    assert conn.execute("SELECT COUNT(*) AS n FROM agent_status").fetchone()["n"] == 1
    conn.close()


def test_transaction_rolls_back_on_error(tmp_path, open_db):
    conn = open_db(tmp_path / "hub.sqlite")
    try:
        with db.transaction(conn):
            conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('x', 'y', 'z')")
            raise RuntimeError("boom")
    except RuntimeError:
        pass
    assert conn.execute("SELECT COUNT(*) AS n FROM batch").fetchone()["n"] == 0


def test_nested_transaction_joins_outer(tmp_path, open_db):
    conn = open_db(tmp_path / "hub.sqlite")
    try:
        with db.transaction(conn):
            with db.transaction(conn):
                conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('x', 'y', 'z')")
            assert conn.in_transaction                                  # вложенный блок не закоммитил сам
            assert conn.execute("SELECT COUNT(*) AS n FROM batch").fetchone()["n"] == 1
            raise RuntimeError("boom")
    except RuntimeError:
        pass
    assert conn.execute("SELECT COUNT(*) AS n FROM batch").fetchone()["n"] == 0   # откат внешней откатил и вложенную


def test_nested_transaction_error_propagates_to_outer(tmp_path, open_db):
    conn = open_db(tmp_path / "hub.sqlite")
    try:
        with db.transaction(conn):
            conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('a', 'y', 'z')")
            with db.transaction(conn):
                raise RuntimeError("inner")
    except RuntimeError:
        pass
    assert conn.execute("SELECT COUNT(*) AS n FROM batch").fetchone()["n"] == 0


def test_parse_and_utcnow():
    assert db.parse("2020-01-01T00:00:00").tzinfo is not None
    assert db.parse(db.now()) <= db.utcnow()


def test_nested_transaction_rollback_is_local(tmp_path, open_db):
    conn = open_db(tmp_path / "hub.sqlite")
    with db.transaction(conn):
        conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('outer', 'y', 'z')")
        try:
            with db.transaction(conn):
                conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('inner', 'y', 'z')")
                raise RuntimeError("inner")
        except RuntimeError:
            pass
    assert [r["title"] for r in conn.execute("SELECT title FROM batch")] == ["outer"]


def _mode(p):
    return stat.S_IMODE(p.stat().st_mode)


def test_data_dir_is_private(tmp_path):
    home = tmp_path / "data"
    home.mkdir()
    home.chmod(0o755)
    db.connect(home / "hub.sqlite").close()
    assert _mode(home) == 0o700


def test_existing_backups_dir_made_private_on_start(tmp_path):
    from jira_hub import api
    backups = tmp_path / "backups"
    backups.mkdir()
    backups.chmod(0o755)
    api.Hub(tmp_path / "hub.sqlite", backups_dir=backups)
    assert _mode(backups) == 0o700


def test_connect_without_directory_does_not_chmod_cwd(tmp_path, monkeypatch):
    monkeypatch.chdir(tmp_path)
    tmp_path.chmod(0o755)
    db.connect("hub.sqlite").close()
    assert _mode(tmp_path) == 0o755


def test_backups_dir_is_private(tmp_path):
    conn = db.connect(tmp_path / "hub.sqlite")
    backup.make_backup(conn, tmp_path / "backups", "batch-1")
    conn.close()
    assert _mode(tmp_path / "backups") == 0o700


def test_secure_dir_failure_only_warns(tmp_path, monkeypatch, caplog):
    def deny(*a, **k):
        raise PermissionError("нет прав")
    monkeypatch.setattr(os, "chmod", deny)
    with caplog.at_level(logging.WARNING):
        paths.secure_dir(tmp_path / "x")
    assert (tmp_path / "x").is_dir() and "0700" in caplog.text


def test_parallel_connect_makes_one_backup(tmp_path):
    path = tmp_path / "hub.sqlite"
    _v1_database(path)
    barrier, errors = threading.Barrier(4), []

    def go():
        try:
            barrier.wait()
            db.connect(path).close()
        except Exception as e:  # noqa: BLE001
            errors.append(e)

    threads = [threading.Thread(target=go) for _ in range(4)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(30)
    assert errors == []
    assert len(list((tmp_path / "backups").glob("*-schema-v1.sqlite"))) == 1
    conn = db.connect(path)
    assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == "5"
    conn.close()


def test_migrates_v2_with_backup(tmp_path):
    """База v2 (без issue_comment) с задачей и заметкой → v3: копия schema-v2, таблица комментариев пустая, данные на месте."""
    path = tmp_path / "hub.sqlite"
    conn = db.connect(path)
    conn.execute("INSERT INTO issue(key, summary, fetched_at) VALUES ('A-1', 'Т', '2026-10-15T00:00:00+00:00')")
    conn.execute("INSERT INTO issue_note(key, fact_status) VALUES ('A-1', 'В работе')")
    _drop_v5(conn)
    conn.execute("DROP TABLE issue_comment")
    conn.execute("ALTER TABLE batch DROP COLUMN project")
    conn.execute("DROP TABLE project")
    conn.execute("UPDATE setting SET value = '2' WHERE key = 'schema_version'")
    conn.close()
    conn = db.connect(path)
    assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == "5"
    assert conn.execute("SELECT COUNT(*) AS n FROM issue_comment").fetchone()["n"] == 0
    assert conn.execute("SELECT summary FROM issue WHERE key = 'A-1'").fetchone()["summary"] == "Т"
    assert conn.execute("SELECT fact_status FROM issue_note WHERE key = 'A-1'").fetchone()["fact_status"] == "В работе"
    copies = list((tmp_path / "backups").glob("*-schema-v2.sqlite"))
    assert len(copies) == 1
    conn.close()


def test_migrated_claimed_redelivered_end_to_end(tmp_path):
    """База v1 с командой в claimed -> миграция (deliveries = 1) -> claim довыдаёт её с deliveries = 2 (M-10 #12)."""
    path = tmp_path / "hub.sqlite"
    _v1_database(path)
    conn = db.connect(path)
    settings.ensure_defaults(conn)
    cmd = queue.claim(conn)
    assert cmd["id"] == 1
    assert queue.get_command(conn, 1)["deliveries"] == 2
    conn.close()


def test_sqlite_without_unix_vfs_deadlock():
    # O-14 (#29): в SQLite 3.51.0–3.51.1 взаимная блокировка в unix-VFS (исправлена в 3.51.2) вешала хаб под
    # параллельными запросами — потоки ждали друг друга в sqlite3.connect/close. Python хаба — от uv (pyproject, tool.uv)
    assert sqlite3.sqlite_version_info[:2] != (3, 51) or sqlite3.sqlite_version_info >= (3, 51, 2), sqlite3.sqlite_version


def _v3_database(path, base_jql="project = AURORA AND labels = AWPCA", board_id=4216):
    """База в состоянии схемы v3: без project и batch.project, base_jql/board_id в setting, очередь и журнал на старой JQL."""
    conn = db.connect(path)
    settings.ensure_defaults(conn)
    for key, value in (("base_jql", base_jql), ("board_id", board_id)):
        conn.execute("INSERT OR REPLACE INTO setting(key, value) VALUES (?, ?)", (key, json.dumps(value)))
    ts = "2026-10-13T00:00:00+00:00"
    for status, jql in (("done", base_jql), ("queued", base_jql), ("claimed", base_jql), ("queued", "key = A-1")):
        conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('x', 'claude', ?)", (ts,))
        batch = conn.execute("SELECT MAX(id) AS id FROM batch").fetchone()["id"]
        conn.execute("INSERT INTO command(batch_id, seq, type, params, status) VALUES (?, 1, 'refresh', ?, ?)",
                     (batch, json.dumps({"jql": jql}), status))
        if status == "done" and jql == base_jql:
            cid = conn.execute("SELECT MAX(id) AS id FROM command").fetchone()["id"]
            conn.execute("INSERT INTO sync_run(started_at, finished_at, kind, query, command_id) VALUES (?, ?, 'jql', ?, ?)",
                         (ts, ts, base_jql, cid))
    conn.execute("INSERT INTO issue(key, summary, fetched_at) VALUES ('A-1', 'Т', ?)", (ts,))
    conn.execute("INSERT INTO issue_note(key, fact_status) VALUES ('A-1', 'В работе')")
    _drop_v5(conn)
    conn.execute("ALTER TABLE batch DROP COLUMN project")
    conn.execute("DROP TABLE project")
    conn.execute("UPDATE setting SET value = '3' WHERE key = 'schema_version'")
    conn.close()


def test_migrates_v3_creates_awpca(tmp_path):
    path = tmp_path / "hub.sqlite"
    _v3_database(path)
    conn = db.connect(path)
    new = 'project = AURORA AND labels in ("AWPCA")'
    assert projects.list_projects(conn)[0] | {"created_at": None, "issues_count": None} == {
        "label": "AWPCA", "name": "АРМ КА", "board_id": 4216, "board_type": None, "created_at": None, "issues_count": None}
    assert settings.get(conn, "base_jql") == "project = AURORA AND labels = AWPCA"     # ключи на месте
    assert settings.get(conn, "board_id") == 4216
    assert conn.execute("SELECT query FROM sync_run").fetchone()["query"] == new
    jqls = {r["status"]: json.loads(r["params"]).get("jql") for r in conn.execute("SELECT status, params FROM command WHERE type = 'refresh'")}
    assert jqls["done"] == "project = AURORA AND labels = AWPCA"                        # завершённые не трогаем
    assert jqls["claimed"] == new
    queued = [json.loads(r["params"])["jql"] for r in conn.execute("SELECT params FROM command WHERE status = 'queued'")]
    assert sorted(queued) == sorted([new, "key = A-1"])
    assert conn.execute("SELECT project FROM batch").fetchall()[0]["project"] is None
    assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == "5"
    assert len(list((tmp_path / "backups").glob("*-schema-v3.sqlite"))) == 1
    conn.close()


def test_migrates_v3_two_labels_board_to_nobody(tmp_path, caplog):
    _v3_database(tmp_path / "hub.sqlite", base_jql="project = AURORA AND labels in (AWPCA, STOCKS_DRRK)")
    conn = db.connect(tmp_path / "hub.sqlite")
    got = {p["label"]: (p["name"], p["board_id"]) for p in projects.list_projects(conn)}
    assert got == {"AWPCA": ("АРМ КА", None), "STOCKS_DRRK": ("STOCKS_DRRK", None)}
    assert "4216" in caplog.text
    conn.close()


def test_migrates_v3_unparsed_jql_keeps_running(tmp_path, caplog):
    _v3_database(tmp_path / "hub.sqlite", base_jql="project = AURORA AND labels = AWPCA ORDER BY key")
    conn = db.connect(tmp_path / "hub.sqlite")
    assert projects.list_projects(conn) == []
    assert "ORDER BY key" in caplog.text
    assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == "5"
    conn.close()


@pytest.mark.parametrize("raw, want", [
    ("2026-10-22T13:05:11.000+0300", "2026-10-22T10:05:11+00:00"),
    ("2026-10-22T13:05:11+03:00", "2026-10-22T10:05:11+00:00"),
    ("2026-10-22T10:05:11Z", "2026-10-22T10:05:11+00:00"),
    ("2026-10-22T10:05:11.123Z", "2026-10-22T10:05:11+00:00"),
    ("2026-10-22T10:05:11.123", "2026-10-22T10:05:11+00:00"),
    ("2026-10-22T02:05:11-0500", "2026-10-22T07:05:11+00:00"),
    ("вчера", None), ("", None), (None, None), (42, None),
    ("9999-12-31T23:00:00-0300", None), ("0001-01-01T00:00:00+0300", None),      # M-1: выход за диапазон datetime
])
def test_jira_utc(raw, want):
    assert db.jira_utc(raw) == want


def _v4_database(path):
    """База v4: комментарий со временем Jira, полный прогон в журнале синхронизаций."""
    conn = db.connect(path)
    conn.execute("INSERT INTO issue_comment(key, comment_id, created, body) VALUES ('X', '1', '2026-10-22T13:05:11.000+0300', 'a')")
    conn.execute("INSERT INTO issue_comment(key, comment_id, created, body) VALUES ('X', '2', 'мусор', 'b')")
    conn.execute("INSERT INTO sync_run(started_at, finished_at, kind, query) VALUES "
                 "('2026-10-22T09:00:00+00:00', '2026-10-22T09:01:00+00:00', 'jql', 'q')")
    _drop_v5(conn)
    conn.execute("UPDATE setting SET value = '4' WHERE key = 'schema_version'")
    conn.close()


def test_migrates_v4(tmp_path):
    path = tmp_path / "hub.sqlite"
    _v4_database(path)
    conn = db.connect(path)
    assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == "5"
    rows = {r["comment_id"]: (r["created"], r["created_at"]) for r in conn.execute("SELECT * FROM issue_comment")}
    assert rows == {"1": ("2026-10-22T13:05:11.000+0300", "2026-10-22T10:05:11+00:00"), "2": ("мусор", None)}
    assert conn.execute("SELECT history_at FROM sync_run").fetchone()["history_at"] is None
    assert conn.execute("SELECT COUNT(*) AS n FROM issue_change").fetchone()["n"] == 0
    assert len(list((tmp_path / "backups").glob("*-schema-v4.sqlite"))) == 1
    conn.close()