"""The video state machine and episode numbering."""

from __future__ import annotations

from datetime import date

import pytest

from ytstream import videos

from conftest import add_channel, add_video


def test_insert_and_get(conn, channel):
    row = add_video(conn, channel["id"], "vid00000001")
    assert row["state"] == videos.LISTED
    assert videos.exists(conn, "vid00000001")
    assert not videos.exists(conn, "nosuchvideo")


def test_insert_is_idempotent(conn, channel):
    add_video(conn, channel["id"], "vid00000001", title="First")
    add_video(conn, channel["id"], "vid00000001", title="Second")
    assert conn.execute("SELECT COUNT(*) FROM video").fetchone()[0] == 1
    assert videos.get(conn, "vid00000001")["title"] == "First"


def test_aged_out_is_terminal_and_never_revived():
    assert videos.AGED_OUT in videos.TERMINAL
    assert videos.AGED_OUT in videos.NEVER_REVIVE
    # skipped_old is terminal but IS revivable by rescan; conflating the two
    # would resurrect months of deleted episodes as new.
    assert videos.SKIPPED_OLD in videos.TERMINAL
    assert videos.SKIPPED_OLD not in videos.NEVER_REVIVE


def test_no_download_era_states_survive():
    """Materialising a text file cannot fail, so there is no retry ladder."""
    for gone in ("pending", "downloading", "downloaded", "failed", "deferred"):
        assert gone not in (
            videos.LISTED, videos.MATERIALISED, videos.SKIPPED_SHORT,
            videos.SKIPPED_LIVE, videos.SKIPPED_OLD, videos.AGED_OUT,
        )


def test_mark_aged_out_clears_rel_path_but_keeps_the_row(conn, channel):
    add_video(conn, channel["id"], "vid00000001", state=videos.MATERIALISED)
    with conn:
        conn.execute("UPDATE video SET rel_path = 'a/b.strm' WHERE video_id = ?",
                     ("vid00000001",))

    videos.mark_aged_out(conn, "vid00000001")

    row = videos.get(conn, "vid00000001")
    assert row is not None
    assert row["state"] == videos.AGED_OUT
    assert row["rel_path"] is None
    assert row["deleted_at"]


def test_mark_materialised_records_everything(conn, channel):
    add_video(conn, channel["id"], "vid00000001")
    videos.mark_materialised(conn, "vid00000001", rel_path="a/b.strm", season=2026,
                             episode=8120, upload_date="2026-08-12", duration=99,
                             title="T")
    row = videos.get(conn, "vid00000001")
    assert (row["state"], row["season"], row["episode"], row["duration"]) == (
        videos.MATERIALISED, 2026, 8120, 99
    )


def test_set_duration_alone(conn, channel):
    add_video(conn, channel["id"], "vid00000001", duration=None)
    videos.set_duration(conn, "vid00000001", 1234)
    assert videos.get(conn, "vid00000001")["duration"] == 1234


# ------------------------------------------------------------ episode numbers


def test_first_episode_of_a_day(conn, channel):
    add_video(conn, channel["id"], "vid00000001")
    assert videos.next_episode(
        conn, channel["id"], date(2026, 8, 12), "vid00000001"
    ) == (2026, 8120)


def test_ordinal_counts_existing_rows_for_that_day(conn, channel):
    add_video(conn, channel["id"], "vid00000001")
    with conn:
        conn.execute("UPDATE video SET season = 2026, episode = 8120 "
                     "WHERE video_id = ?", ("vid00000001",))
    add_video(conn, channel["id"], "vid00000002")

    assert videos.next_episode(
        conn, channel["id"], date(2026, 8, 12), "vid00000002"
    ) == (2026, 8121)


def test_ordinal_ignores_other_days(conn, channel):
    add_video(conn, channel["id"], "vid00000001")
    with conn:
        conn.execute("UPDATE video SET season = 2026, episode = 8110 "
                     "WHERE video_id = ?", ("vid00000001",))
    add_video(conn, channel["id"], "vid00000002")

    assert videos.next_episode(
        conn, channel["id"], date(2026, 8, 12), "vid00000002"
    ) == (2026, 8120)


def test_ordinal_ignores_other_channels(conn, channel):
    other = add_channel(conn, "UC" + "z" * 22, "Other", "Other")
    add_video(conn, other["id"], "vid00000001")
    with conn:
        conn.execute("UPDATE video SET season = 2026, episode = 8120 "
                     "WHERE video_id = ?", ("vid00000001",))
    add_video(conn, channel["id"], "vid00000002")

    assert videos.next_episode(
        conn, channel["id"], date(2026, 8, 12), "vid00000002"
    ) == (2026, 8120)


def test_ordinal_clamps_at_ten_uploads_a_day(conn, channel):
    for index in range(10):
        video_id = f"vid{index:08d}"
        add_video(conn, channel["id"], video_id)
        with conn:
            conn.execute("UPDATE video SET season = 2026, episode = ? "
                         "WHERE video_id = ?", (8120 + index, video_id))
    add_video(conn, channel["id"], "vid00000099")

    season, episode = videos.next_episode(
        conn, channel["id"], date(2026, 8, 12), "vid00000099"
    )

    assert (season, episode) == (2026, 8129)


def test_a_videos_own_row_does_not_bump_its_ordinal(conn, channel):
    """Re-materialising must produce the same number, not the next one."""
    add_video(conn, channel["id"], "vid00000001")
    with conn:
        conn.execute("UPDATE video SET season = 2026, episode = 8120 "
                     "WHERE video_id = ?", ("vid00000001",))

    assert videos.next_episode(
        conn, channel["id"], date(2026, 8, 12), "vid00000001"
    ) == (2026, 8120)


# --------------------------------------------------------------------- queues


def test_claim_listed_returns_only_listed_rows(conn, channel):
    add_video(conn, channel["id"], "listed00001", state=videos.LISTED)
    add_video(conn, channel["id"], "done0000001", state=videos.MATERIALISED)
    add_video(conn, channel["id"], "short000001", state=videos.SKIPPED_SHORT)
    add_video(conn, channel["id"], "aged0000001", state=videos.AGED_OUT)

    rows = videos.claim_listed(conn)

    assert [row["video_id"] for row in rows] == ["listed00001"]


def test_claim_listed_is_oldest_upload_first(conn, channel):
    add_video(conn, channel["id"], "newer000001", upload_date="2026-08-10")
    add_video(conn, channel["id"], "older000001", upload_date="2026-01-01")

    rows = videos.claim_listed(conn)

    assert [row["video_id"] for row in rows] == ["older000001", "newer000001"]


def test_claim_listed_honours_the_limit(conn, channel):
    for index in range(5):
        add_video(conn, channel["id"], f"vid{index:08d}")
    assert len(videos.claim_listed(conn, limit=2)) == 2


def test_materialised_for_channel_is_newest_first(conn, channel):
    """The retention sweep counts down from the newest to honour min_keep_videos."""
    add_video(conn, channel["id"], "older000001", upload_date="2026-01-01",
              state=videos.MATERIALISED)
    add_video(conn, channel["id"], "newer000001", upload_date="2026-08-10",
              state=videos.MATERIALISED)

    rows = videos.materialised_for_channel(conn, channel["id"])

    assert [row["video_id"] for row in rows] == ["newer000001", "older000001"]


def test_queue_depth_and_counts(conn, channel):
    add_video(conn, channel["id"], "listed00001", state=videos.LISTED)
    add_video(conn, channel["id"], "listed00002", state=videos.LISTED)
    add_video(conn, channel["id"], "done0000001", state=videos.MATERIALISED)

    assert videos.queue_depth(conn) == 2
    counts = videos.counts_by_state(conn)
    assert counts[videos.LISTED] == 2
    assert counts[videos.MATERIALISED] == 1


def test_deleting_a_channel_cascades_to_its_videos(conn, channel):
    add_video(conn, channel["id"], "vid00000001")
    with conn:
        conn.execute("DELETE FROM channel WHERE id = ?", (channel["id"],))
    assert conn.execute("SELECT COUNT(*) FROM video").fetchone()[0] == 0


# --------------------------------------------------------------- migrations


def test_a_v1_database_migrates_to_v2(tmp_path):
    """The live database on susan was created at v1. A fresh install must end up
    identical to a migrated one, which is why the description column is a v2
    migration rather than an edit to the v1 script."""
    import sqlite3 as sq

    from ytstream import db

    path = tmp_path / "v1.db"
    raw = sq.connect(path)
    raw.executescript(db._SCHEMA_V1)
    raw.execute("PRAGMA user_version = 1")
    raw.commit()
    raw.close()

    conn = db.connect(path)
    try:
        assert conn.execute("PRAGMA user_version").fetchone()[0] == db.SCHEMA_VERSION
        columns = {row[1] for row in conn.execute("PRAGMA table_info(video)")}
        assert "description" in columns
    finally:
        conn.close()


def test_migration_is_idempotent(tmp_path):
    from ytstream import db

    path = tmp_path / "twice.db"
    for _ in range(3):
        conn = db.connect(path)
        conn.close()
    conn = db.connect(path)
    try:
        columns = [row[1] for row in conn.execute("PRAGMA table_info(video)")]
        assert columns.count("description") == 1
    finally:
        conn.close()


def test_fresh_and_migrated_schemas_match(tmp_path):
    """A fresh v2 install and a v1 database brought forward must agree, or the two
    populations diverge silently."""
    import sqlite3 as sq

    from ytstream import db

    fresh = db.connect(tmp_path / "fresh.db")
    fresh_cols = [tuple(row)[1:3] for row in fresh.execute("PRAGMA table_info(video)")]
    fresh.close()

    old = tmp_path / "old.db"
    raw = sq.connect(old)
    raw.executescript(db._SCHEMA_V1)
    raw.execute("PRAGMA user_version = 1")
    raw.commit()
    raw.close()
    migrated = db.connect(old)
    migrated_cols = [tuple(row)[1:3]
                     for row in migrated.execute("PRAGMA table_info(video)")]
    migrated.close()

    assert fresh_cols == migrated_cols


def test_description_survives_a_round_trip(conn, channel):
    add_video(conn, channel["id"], "vid00000001", description="A synopsis")
    assert videos.get(conn, "vid00000001")["description"] == "A synopsis"