import multiprocessing
import sqlite3

import pytest

from suvi import db, paths


def make_frame(conn, satellite=16, wavelength=171, minute=0):
    t_start = 1715299200 + minute * paths.CADENCE
    name = paths.FrameName(
        satellite=satellite,
        wavelength=wavelength,
        t_start=t_start,
        t_end=t_start + paths.CADENCE,
        version="1-0-2",
    )
    return db.upsert_frame(conn, name, name.relpath(), size_bytes=1774080, mtime=1.0), name


def test_connect_creates_schema(db_path):
    conn = db.connect(db_path)
    tables = {
        row["name"]
        for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
    }
    assert {"frame", "header", "detector_run", "detection", "remote_file", "meta"} <= tables
    assert conn.execute(
        "SELECT value FROM meta WHERE key='schema_version'"
    ).fetchone()["value"] == str(db.SCHEMA_VERSION)
    conn.close()


def test_init_schema_is_idempotent(db_path):
    conn = db.connect(db_path)
    db.init_schema(conn)
    db.init_schema(conn)
    assert conn.execute("SELECT count(*) c FROM meta").fetchone()["c"] == 1
    conn.close()


def test_schema_version_mismatch_is_refused(db_path):
    conn = db.connect(db_path)
    conn.execute("UPDATE meta SET value='99' WHERE key='schema_version'")
    conn.commit()
    with pytest.raises(RuntimeError, match="schema version"):
        db.init_schema(conn)
    conn.close()


def test_readonly_connect_requires_an_existing_file(tmp_path):
    with pytest.raises(FileNotFoundError):
        db.connect(str(tmp_path / "absent.sqlite"), readonly=True)


def test_readonly_connect_cannot_write(db_path):
    db.connect(db_path).close()
    conn = db.connect(db_path, readonly=True)
    with pytest.raises(sqlite3.OperationalError):
        conn.execute("INSERT INTO meta (key, value) VALUES ('x', 'y')")
    conn.close()


def test_upsert_frame_is_stable_and_updates_in_place(db_path):
    conn = db.connect(db_path)
    first, name = make_frame(conn)
    again = db.upsert_frame(conn, name, name.relpath(), size_bytes=999, mtime=2.0)
    assert first == again
    row = conn.execute("SELECT * FROM frame WHERE id=?", (first,)).fetchone()
    assert row["size_bytes"] == 999 and row["mtime"] == 2.0
    assert conn.execute("SELECT count(*) c FROM frame").fetchone()["c"] == 1
    conn.close()


def test_duplicate_slot_under_a_new_version_is_rejected(db_path):
    """Two files claiming the same observation must not silently shadow each other."""
    conn = db.connect(db_path)
    _, name = make_frame(conn)
    other = paths.FrameName(
        name.satellite, name.wavelength, name.t_start, name.t_end, version="1-0-3"
    )
    with pytest.raises(sqlite3.IntegrityError):
        db.upsert_frame(conn, other, other.relpath())
    conn.close()


def test_frame_id_by_slot(db_path):
    conn = db.connect(db_path)
    frame_id, name = make_frame(conn)
    assert db.frame_id_by_slot(conn, *name.slot) == frame_id
    assert db.frame_id_by_slot(conn, 18, 171, name.t_start) is None
    conn.close()


def test_detector_runs_are_ordered_by_recency(db_path):
    conn = db.connect(db_path)
    first = db.create_detector_run(conn, "header_v1", {"threshold": 1})
    second = db.create_detector_run(conn, "header_v1", {"threshold": 2})
    assert db.latest_run_id(conn, "header_v1") == second
    assert first != second
    assert db.latest_run_id(conn, "nope") is None
    conn.close()


def test_record_detections_stores_scores_and_upserts(db_path):
    conn = db.connect(db_path)
    frame_id, _ = make_frame(conn)
    run = db.create_detector_run(conn, "header_v1", {})
    db.record_detections(conn, run, [(frame_id, "bad", "eclipse", {"img_mean": 1e-4}, 12)])
    db.record_detections(conn, run, [(frame_id, "good", None, {"img_mean": 0.3}, 15)])
    conn.commit()
    row = conn.execute("SELECT * FROM detection").fetchone()
    assert row["verdict"] == "good"
    assert row["reason"] is None
    assert row["scores_json"] == '{"img_mean": 0.3}'
    assert conn.execute("SELECT count(*) c FROM detection").fetchone()["c"] == 1
    conn.close()


def test_record_detections_rejects_unknown_verdicts(db_path):
    conn = db.connect(db_path)
    frame_id, _ = make_frame(conn)
    run = db.create_detector_run(conn, "header_v1", {})
    with pytest.raises(ValueError, match="Unknown verdict"):
        db.record_detections(conn, run, [(frame_id, "maybe", None, None, 0)])
    conn.close()


def test_detections_cascade_when_a_run_is_deleted(db_path):
    conn = db.connect(db_path)
    frame_id, _ = make_frame(conn)
    run = db.create_detector_run(conn, "header_v1", {})
    db.record_detections(conn, run, [(frame_id, "good", None, None, 0)])
    conn.execute("DELETE FROM detector_run WHERE id=?", (run,))
    conn.commit()
    assert conn.execute("SELECT count(*) c FROM detection").fetchone()["c"] == 0
    conn.close()


def test_good_slots_filters_by_verdict_window_and_band(db_path):
    conn = db.connect(db_path)
    run = db.create_detector_run(conn, "header_v1", {})
    ids = {}
    for minute in range(4):
        for wavelength in (171, 304):
            frame_id, name = make_frame(conn, wavelength=wavelength, minute=minute)
            ids[(minute, wavelength)] = (frame_id, name)
    # Everything good except one frame, plus a decoy on the other satellite.
    db.record_detections(
        conn,
        run,
        [
            (fid, "bad" if key == (1, 171) else "good", None, None, 0)
            for key, (fid, _) in ids.items()
        ],
    )
    other, _ = make_frame(conn, satellite=18, minute=0)
    db.record_detections(conn, run, [(other, "good", None, None, 0)])
    conn.commit()

    base = ids[(0, 171)][1].t_start
    rows = db.good_slots(conn, run, base, base + 3 * paths.CADENCE, 16, (171,))
    assert [r["t_start"] for r in rows] == [base, base + 2 * paths.CADENCE]
    assert all(r["satellite"] == 16 and r["wavelength"] == 171 for r in rows)

    both = db.good_slots(conn, run, base, base + 4 * paths.CADENCE, 16, (171, 304))
    assert len(both) == 7  # 8 frames minus the one marked bad
    conn.close()


def test_unscanned_frames_lists_only_frames_without_headers(db_path):
    conn = db.connect(db_path)
    first, _ = make_frame(conn, minute=0)
    second, _ = make_frame(conn, minute=1)
    db.record_headers(conn, [(first, {"img_mean": 0.3, "eclipse": 0}, None)])
    conn.commit()
    assert [r["id"] for r in db.unscanned_frames(conn)] == [second]
    assert len(db.unscanned_frames(conn, limit=0)) == 0
    conn.close()


def test_record_headers_round_trips_values_and_errors(db_path):
    conn = db.connect(db_path)
    first, _ = make_frame(conn, minute=0)
    second, _ = make_frame(conn, minute=1)
    db.record_headers(
        conn,
        [
            (first, {"img_mean": 0.31, "degraded": 0, "datasum": "261950686"}, None),
            (second, {}, "truncated: 17 bytes"),
        ],
    )
    conn.commit()
    good = conn.execute("SELECT * FROM header WHERE frame_id=?", (first,)).fetchone()
    assert good["img_mean"] == pytest.approx(0.31)
    assert good["read_ok"] == 1 and good["read_error"] is None
    bad = conn.execute("SELECT * FROM header WHERE frame_id=?", (second,)).fetchone()
    assert bad["read_ok"] == 0 and bad["read_error"].startswith("truncated")
    assert bad["img_mean"] is None
    conn.close()


def test_record_headers_upserts(db_path):
    conn = db.connect(db_path)
    frame_id, _ = make_frame(conn)
    db.record_headers(conn, [(frame_id, {"img_mean": 0.1}, None)])
    db.record_headers(conn, [(frame_id, {"img_mean": 0.9}, None)])
    conn.commit()
    assert conn.execute("SELECT count(*) c FROM header").fetchone()["c"] == 1
    assert conn.execute("SELECT img_mean FROM header").fetchone()["img_mean"] == 0.9
    conn.close()


def test_remote_file_round_trip(db_path):
    conn = db.connect(db_path)
    db.record_remote_files(conn, [("http://x/a.fits", 100.0, 17, "a.fits", 5.0)])
    db.record_remote_files(conn, [("http://x/a.fits", 200.0, 18, "a.fits", 6.0)])
    conn.commit()
    assert db.get_remote_mtime(conn, "http://x/a.fits") == 200.0
    assert db.get_remote_mtime(conn, "http://x/missing.fits") is None
    assert conn.execute("SELECT count(*) c FROM remote_file").fetchone()["c"] == 1
    conn.close()


def test_writer_loop_drains_a_queue_from_another_process(db_path):
    """Workers stay read-only; one writer owns the lock.  This is that contract."""
    conn = db.connect(db_path)
    frame_ids = [make_frame(conn, minute=i)[0] for i in range(20)]
    run = db.create_detector_run(conn, "header_v1", {})
    conn.commit()
    conn.close()

    ctx = multiprocessing.get_context("spawn")
    queue = ctx.Queue()
    writer = ctx.Process(target=db.writer_loop, args=(db_path, queue), daemon=True)
    writer.start()
    for frame_id in frame_ids:
        queue.put(("detection", (run, frame_id, "good", None, {"s": 1}, 3)))
    queue.put(("header", (frame_ids[0], {"img_mean": 0.5}, None)))
    queue.put(("remote_file", ("http://x/a.fits", 1.0, 2, "a", 3.0)))
    queue.put(db.STOP)
    writer.join(timeout=60)
    assert writer.exitcode == 0

    conn = db.connect(db_path, readonly=True)
    assert conn.execute("SELECT count(*) c FROM detection").fetchone()["c"] == 20
    assert conn.execute("SELECT count(*) c FROM header").fetchone()["c"] == 1
    assert conn.execute("SELECT count(*) c FROM remote_file").fetchone()["c"] == 1
    conn.close()


def test_writer_loop_rejects_unknown_message_kinds(db_path):
    ctx = multiprocessing.get_context("spawn")
    queue = ctx.Queue()
    writer = ctx.Process(target=db.writer_loop, args=(db_path, queue), daemon=True)
    writer.start()
    queue.put(("bogus", ()))
    writer.join(timeout=60)
    assert writer.exitcode not in (0, None)


def test_latest_run_is_scoped_by_config(db_path):
    """Runs accumulate one per detector per case, so 'latest' is ambiguous.

    Resolving a detector without its config picks whichever case ran most
    recently -- and scoring one case's verdicts against another case's injections
    gives numbers that are wrong, not merely noisy.
    """
    conn = db.connect(db_path)
    first = db.create_detector_run(conn, "header_v1", {"case": "alpha"})
    second = db.create_detector_run(conn, "header_v1", {"case": "beta"})

    assert db.latest_run_id(conn, "header_v1") == second          # global latest
    assert db.latest_run_id(conn, "header_v1", {"case": "alpha"}) == first
    assert db.latest_run_id(conn, "header_v1", {"case": "beta"}) == second
    assert db.latest_run_id(conn, "header_v1", {"case": "gamma"}) is None
    conn.close()


def test_scoped_lookup_takes_the_newest_matching_run(db_path):
    conn = db.connect(db_path)
    db.create_detector_run(conn, "disc_v1", {"case": "alpha"})
    newer = db.create_detector_run(conn, "disc_v1", {"case": "alpha"})
    assert db.latest_run_id(conn, "disc_v1", {"case": "alpha"}) == newer
    conn.close()