"""Migration tests. Run with: python test_db.py

The v1->v2 migration rebuilds the media table, so it drops the original.
These cases cover the paths that rebuild touches before it ever runs against
a real bot_db/media.db.
"""
import asyncio
import importlib
import os
import sqlite3
import sys
import tempfile

V1_SCHEMA = """
    CREATE TABLE media (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        owner_id INTEGER NOT NULL,
        file_id TEXT NOT NULL,
        file_unique_id TEXT NOT NULL UNIQUE,
        media_type TEXT NOT NULL,
        keywords TEXT NOT NULL,
        added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    );
"""


def user_version(path: str) -> int:
    with sqlite3.connect(path) as c:
        return c.execute("PRAGMA user_version").fetchone()[0]


def load_db(path: str):
    os.environ["DB_PATH"] = path
    import db
    return importlib.reload(db)


async def test_legacy_upgrade(tmp: str):
    path = os.path.join(tmp, "legacy.db")
    with sqlite3.connect(path) as c:
        c.executescript(V1_SCHEMA)
        c.execute("INSERT INTO media (id,owner_id,file_id,file_unique_id,media_type,keywords)"
                  " VALUES (7,111,'FID_A','UNI_A','photo','cat,funny')")
        c.execute("INSERT INTO media (id,owner_id,file_id,file_unique_id,media_type,keywords)"
                  " VALUES (9,111,'FID_B','UNI_B','gif','dog')")

    db = load_db(path)
    await db.init_db()

    # Pinned to the module so adding a migration does not require editing this.
    assert user_version(path) == db.SCHEMA_VERSION
    rows = sorted((r["id"], r["keywords"]) for r in await db.search_media(""))
    assert rows == [(7, "cat,funny"), (9, "dog")], f"rows not preserved: {rows}"
    with sqlite3.connect(path) as c:
        statuses = {r[0] for r in c.execute("SELECT status FROM media")}
        tables = {r[0] for r in c.execute("SELECT name FROM sqlite_master WHERE type='table'")}
    assert statuses == {"approved"}, f"upgraded rows must stay live: {statuses}"
    assert "vault_items" in tables, "upgrading should create the vault table too"
    with sqlite3.connect(path) as c:
        visibilities = {r[0] for r in c.execute("SELECT visibility FROM media")}
    assert visibilities == {"private"}, \
        f"upgrading must not opt an existing library into sharing: {visibilities}"
    print("  legacy v1 database upgrades with ids, keywords and visibility intact")


async def test_per_owner_uniqueness(tmp: str):
    path = os.path.join(tmp, "owners.db")
    db = load_db(path)
    await db.init_db()

    await db.add_media(111, "FID_A", "UNI_A", "photo", ["cat"])
    await db.add_media(222, "FID_A2", "UNI_A", "photo", ["cat"])
    assert len(await db.search_media("cat")) == 2, "two owners must each keep their own row"
    assert len(await db.search_media("", owner_id=222)) == 1, "owner scoping failed"
    print("  two owners can save the same file; owner scoping filters correctly")


async def test_readd_keeps_id(tmp: str):
    path = os.path.join(tmp, "readd.db")
    db = load_db(path)
    await db.init_db()

    await db.add_media(111, "FID_A", "UNI_A", "photo", ["cat"])
    original = (await db.search_media("cat"))[0]["id"]
    await db.add_media(111, "FID_NEW", "UNI_A", "photo", ["kitten"])
    row = await db.get_media_by_id(original)

    assert row is not None, "re-adding the same file orphaned its id"
    assert row["keywords"] == "kitten" and row["file_id"] == "FID_NEW"
    print("  re-adding a file updates in place and keeps its id")


async def test_fresh_and_idempotent(tmp: str):
    path = os.path.join(tmp, "fresh.db")
    db = load_db(path)
    await db.init_db()

    with sqlite3.connect(path) as c:
        sql = c.execute("SELECT sql FROM sqlite_master WHERE name='media'").fetchone()[0]
    assert "UNIQUE(owner_id, file_unique_id)" in sql
    assert user_version(path) == db.SCHEMA_VERSION

    await db.init_db()
    await db.init_db()
    print("  fresh install lands on the current version; init_db is idempotent")


async def test_pending_is_hidden(tmp: str):
    path = os.path.join(tmp, "pending.db")
    db = load_db(path)
    await db.init_db()

    await db.add_media(111, "F1", "U1", "photo", ["live"])
    await db.add_media(222, "F2", "U2", "photo", ["waiting"], status="pending")

    assert len(await db.search_media("")) == 1, "pending items must not be searchable"
    assert await db.count_media() == 1
    assert len(await db.list_media(0, 30)) == 1, "pending items must not appear in /list"
    assert len(await db.search_media("", status=None)) == 2, "status=None should see everything"
    assert await db.count_pending() == 1
    print("  pending items stay out of search, /list and counts until approved")


async def test_queue_is_oldest_first(tmp: str):
    path = os.path.join(tmp, "queue.db")
    db = load_db(path)
    await db.init_db()

    for i in range(3):
        await db.add_media(100 + i, f"F{i}", f"U{i}", "photo", [f"kw{i}"], status="pending")

    first = await db.next_pending()
    assert first["keywords"] == "kw0", f"queue must be oldest-first, got {first['keywords']}"

    # Approving removes it from the queue and the next one steps up.
    assert await db.set_status(first["id"], "approved", 999) is True
    second = await db.next_pending()
    assert second["keywords"] == "kw1"
    assert await db.count_pending() == 2
    print("  queue serves oldest first and advances as items are handled")


async def test_two_moderators_cannot_double_review(tmp: str):
    path = os.path.join(tmp, "race.db")
    db = load_db(path)
    await db.init_db()

    await db.add_media(111, "F1", "U1", "photo", ["kw"], status="pending")
    item = await db.next_pending()

    assert await db.set_status(item["id"], "approved", 111) is True
    assert await db.set_status(item["id"], "rejected", 222) is False, \
        "a second moderator must not overwrite a completed review"

    row = await db.get_media_by_id(item["id"])
    assert row["status"] == "approved", "first review must win"
    print("  a second moderator cannot overturn an already-reviewed item")


async def test_rejected_stays_hidden(tmp: str):
    path = os.path.join(tmp, "rejected.db")
    db = load_db(path)
    await db.init_db()

    await db.add_media(111, "F1", "U1", "photo", ["nope"], status="pending")
    item = await db.next_pending()
    await db.set_status(item["id"], "rejected", 999)

    assert await db.search_media("nope") == [], "rejected items must never surface"
    assert await db.count_pending() == 0
    assert await db.next_pending() is None
    print("  rejected items leave the queue and never surface in search")


async def test_new_media_is_private(tmp: str):
    path = os.path.join(tmp, "visibility.db")
    db = load_db(path)
    await db.init_db()

    await db.add_media(111, "F1", "U1", "photo", ["kw"])
    assert (await db.search_media("kw"))[0]["visibility"] == "private"
    assert await db.list_syncable() == [], "nothing should be shared by default"
    print("  media is private unless explicitly shared")


async def test_only_public_and_approved_can_sync(tmp: str):
    path = os.path.join(tmp, "syncable.db")
    db = load_db(path)
    await db.init_db()

    await db.add_media(111, "F1", "U1", "photo", ["private_approved"])
    await db.add_media(111, "F2", "U2", "photo", ["public_approved"])
    await db.add_media(111, "F3", "U3", "photo", ["public_pending"], status="pending")
    await db.add_media(111, "F4", "U4", "photo", ["public_rejected"], status="rejected")

    # Force the states set_visibility would refuse, to prove the query itself
    # filters rather than relying on the writer being careful.
    with sqlite3.connect(path) as c:
        c.execute("UPDATE media SET visibility='public' WHERE file_unique_id IN ('U2','U3','U4')")

    syncable = [r["keywords"] for r in await db.list_syncable()]
    assert syncable == ["public_approved"], f"leaked: {syncable}"
    assert await db.count_public() == 1
    print("  only approved AND public media is ever syncable")


async def test_sharing_requires_approval(tmp: str):
    path = os.path.join(tmp, "share_guard.db")
    db = load_db(path)
    await db.init_db()

    await db.add_media(111, "F1", "U1", "photo", ["waiting"], status="pending")
    pending = (await db.search_media("", status="pending"))[0]
    assert await db.set_visibility(pending["id"], "public") is False, \
        "unreviewed media must not be shareable"
    assert await db.list_syncable() == []
    print("  media awaiting moderation cannot be marked shared")


async def test_unsharing_stops_future_syncs(tmp: str):
    path = os.path.join(tmp, "unshare.db")
    db = load_db(path)
    await db.init_db()

    await db.add_media(111, "F1", "U1", "photo", ["kw"])
    item = (await db.search_media("kw"))[0]
    assert await db.set_visibility(item["id"], "public") is True
    assert len(await db.list_syncable()) == 1

    assert await db.set_visibility(item["id"], "private") is True
    assert await db.list_syncable() == [], "unsharing should remove it from the sync set"
    print("  unsharing takes an item back out of the sync set")


async def test_vault_never_syncs(tmp: str):
    path = os.path.join(tmp, "vault_sync.db")
    db = load_db(path)
    await db.init_db()

    await db.add_vault_item(111, 500, [1, 2], "a private post", ["personal"])
    # The vault lives in its own table, so the sync query cannot reach it at all.
    assert await db.list_syncable() == []
    assert await db.count_public() == 0
    print("  vault posts have no path into the sync set")


async def main():
    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    with tempfile.TemporaryDirectory() as tmp:
        for test in (test_legacy_upgrade, test_per_owner_uniqueness,
                     test_readd_keeps_id, test_fresh_and_idempotent,
                     test_pending_is_hidden, test_queue_is_oldest_first,
                     test_two_moderators_cannot_double_review,
                     test_rejected_stays_hidden, test_new_media_is_private,
                     test_only_public_and_approved_can_sync,
                     test_sharing_requires_approval,
                     test_unsharing_stops_future_syncs, test_vault_never_syncs):
            await test(tmp)
    print("\nAll database tests passed.")


if __name__ == "__main__":
    asyncio.run(main())