from __future__ import annotations

import json
import re
import sqlite3
import sys
from pathlib import Path

import pytest

SHARED_DIR = Path(__file__).resolve().parent
if str(SHARED_DIR.parent) not in sys.path:
    sys.path.insert(0, str(SHARED_DIR.parent))

from Shared.storage.sort_storage import (
    DEFAULT_STORAGE_CODE,
    SCHEMA_VERSION,
    SOURCE_MODES,
    connect_database,
    create_download_batch,
    create_download_job,
    default_database_path,
    initialize_database,
    record_media_saved,
    record_downloaded_file,
    upsert_observation,
    list_accounts,
    list_account_posts,
    get_post_stats,
    soft_delete_account,
    soft_delete_post,
    list_deleted_accounts,
)

REQUIRED_TABLES = {
    "schema_migrations",
    "collector_profile",
    "device_profile",
    "storage_roots",
    "social_accounts",
    "collection_profiles",
    "account_collection_rules",
    "posts",
    "post_stat_latest",
    "post_stat_history",
    "media_assets",
    "transcripts",
    "hook_segments",
    "categories",
    "post_category_assignments",
    "tags",
    "post_tag_assignments",
    "post_notes",
    "review_templates",
    "review_items",
    "post_reviews",
    "review_scores",
    "post_comments",
    "aggregate_definitions",
    "aggregate_accounts",
    "p2p_inventory",
    "sync_queue",
    "point_ledger_cache",
    "download_batches",
    "download_jobs",
    "media_saved_events",
    "post_stat_10m",
    "deletion_events",
}


def make_db(tmp_path: Path) -> tuple[Path, Path]:
    st_root = tmp_path / "bundle" / "_St"
    st_root.mkdir(parents=True)
    db_path = st_root / "Shared" / "data" / "sort_local.sqlite3"
    initialize_database(db_path, st_root)
    return st_root, db_path


def insert_post(db_path: Path, suffix: str = "A") -> tuple[int, int]:
    conn = connect_database(db_path)
    try:
        account_id = conn.execute(
            "INSERT INTO social_accounts(username) VALUES (?)",
            (f"account_{suffix}",),
        ).lastrowid
        post_id = conn.execute(
            "INSERT INTO posts(account_id, shortcode, post_url) VALUES (?, ?, ?)",
            (
                account_id,
                f"SHORT{suffix}",
                f"https://www.instagram.com/reel/SHORT{suffix}/",
            ),
        ).lastrowid
        conn.commit()
        return int(account_id), int(post_id)
    finally:
        conn.close()


def test_sqlite_initialization_and_fixed_stdown(tmp_path: Path) -> None:
    st_root, db_path = make_db(tmp_path)
    result = initialize_database(db_path, st_root)
    expected = tmp_path / "bundle" / "_StDown" / "instagram"
    assert expected.is_dir() and result["storage_root"] == str(expected.resolve())
    assert result["schema_version"] == SCHEMA_VERSION == 3
    assert result["journal_mode"] == "wal" and result["foreign_keys"] == 1
    conn = connect_database(db_path)
    try:
        tables = {
            row[0]
            for row in conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
            )
        }
        assert REQUIRED_TABLES.issubset(tables)
        root = conn.execute(
            "SELECT root_code, root_path, is_default FROM storage_roots"
        ).fetchone()
        assert root["root_code"] == DEFAULT_STORAGE_CODE
        assert root["root_path"] == str(expected.resolve()) and root["is_default"] == 1
        versions = {
            row[0] for row in conn.execute("SELECT version FROM schema_migrations")
        }
        assert {1, 2, 3}.issubset(versions)
    finally:
        conn.close()



def test_migrate_existing_v1_database_to_v3(tmp_path: Path) -> None:
    st_root = tmp_path / "bundle" / "_St"
    st_root.mkdir(parents=True)
    db_path = st_root / "Shared" / "data" / "sort_local.sqlite3"
    db_path.parent.mkdir(parents=True)
    conn = sqlite3.connect(db_path)
    try:
        schema_v1 = (SHARED_DIR / "storage" / "schema_sqlite_v1.sql").read_text(encoding="utf-8")
        conn.executescript(schema_v1)
        conn.commit()
        assert conn.execute("PRAGMA user_version").fetchone()[0] == 1
    finally:
        conn.close()

    result = initialize_database(db_path, st_root)
    assert result["schema_version"] == 3
    conn = connect_database(db_path)
    try:
        tables = {
            row[0]
            for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
        }
        assert {"download_batches", "download_jobs", "media_saved_events", "post_stat_10m", "deletion_events"}.issubset(tables)
    finally:
        conn.close()

def test_sqlite_idempotent_and_post_unique(tmp_path: Path) -> None:
    st_root, db_path = make_db(tmp_path)
    initialize_database(db_path, st_root)
    conn = connect_database(db_path)
    try:
        assert conn.execute("SELECT COUNT(*) FROM storage_roots").fetchone()[0] == 1
        conn.execute(
            "INSERT INTO posts(platform, shortcode, post_url) VALUES('instagram','ABC123','https://www.instagram.com/p/ABC123/')"
        )
        with pytest.raises(sqlite3.IntegrityError):
            conn.execute(
                "INSERT INTO posts(platform, shortcode, post_url) VALUES('instagram','ABC123','https://www.instagram.com/reel/ABC123/')"
            )
    finally:
        conn.close()


@pytest.mark.parametrize("source_mode", sorted(SOURCE_MODES))
def test_unified_media_saved_flow_for_every_source_mode(
    tmp_path: Path, source_mode: str
) -> None:
    _, db_path = make_db(tmp_path)
    account_id, post_id = insert_post(db_path, source_mode[:3])
    batch_id = create_download_batch(
        db_path,
        source_mode=source_mode,
        account_id=account_id,
        sort_by="views" if source_mode == "TOP_SORTED" else "",
        filter_days=7 if source_mode in {"TOP_SORTED", "BULK_FILTERED"} else None,
        requested_count=1,
    )
    job_id = create_download_job(
        db_path,
        batch_id=batch_id,
        post_id=post_id,
        source_mode=source_mode,
        asset_kind="VIDEO",
        rank_no=1 if source_mode == "TOP_SORTED" else None,
        source_url="https://cdn.example/video.mp4",
        expected_file_name="video.mp4",
    )
    event_uuid = f"event-{source_mode}"
    result = record_media_saved(
        db_path,
        post_id=post_id,
        source_mode=source_mode,
        asset_kind="VIDEO",
        relative_path=f"account_{source_mode[:3]}/video.mp4",
        file_name="video.mp4",
        file_size=12345,
        sha256="a" * 64,
        source_url="https://cdn.example/video.mp4",
        download_job_id=job_id,
        event_uuid=event_uuid,
        rank_no=1 if source_mode == "TOP_SORTED" else None,
    )
    duplicate = record_media_saved(
        db_path,
        post_id=post_id,
        source_mode=source_mode,
        asset_kind="VIDEO",
        relative_path=f"account_{source_mode[:3]}/video.mp4",
        file_name="video.mp4",
        file_size=12345,
        sha256="a" * 64,
        event_uuid=event_uuid,
    )
    assert result["idempotent"] is False
    assert duplicate["idempotent"] is True and duplicate["event_id"] == result["event_id"]

    conn = connect_database(db_path)
    try:
        event = conn.execute(
            "SELECT * FROM media_saved_events WHERE id = ?", (result["event_id"],)
        ).fetchone()
        asset = conn.execute(
            "SELECT * FROM media_assets WHERE id = ?", (result["asset_id"],)
        ).fetchone()
        job = conn.execute("SELECT * FROM download_jobs WHERE id = ?", (job_id,)).fetchone()
        post = conn.execute("SELECT media_state FROM posts WHERE id = ?", (post_id,)).fetchone()
        sync = conn.execute(
            "SELECT * FROM sync_queue WHERE entity_type = 'MEDIA_SAVED'"
        ).fetchall()
        assert event["source_mode"] == source_mode
        assert asset["local_status"] == "AVAILABLE" and asset["sha256"] == "a" * 64
        batch = conn.execute("SELECT * FROM download_batches WHERE id = ?", (batch_id,)).fetchone()
        assert job["job_status"] == "COMPLETE"
        assert batch["batch_status"] == "COMPLETE"
        assert asset["storage_root_id"] is not None
        assert post["media_state"] == "MEDIA_AVAILABLE"
        assert len(sync) == 1
        sync_payload = json.loads(sync[0]["payload_json"])
        assert sync_payload["source_mode"] == source_mode
        assert "relative_path" not in sync_payload and "local_path" not in sync_payload
    finally:
        conn.close()


def test_media_saved_rejects_path_escape(tmp_path: Path) -> None:
    _, db_path = make_db(tmp_path)
    _, post_id = insert_post(db_path, "PATH")
    with pytest.raises(ValueError, match="storage root"):
        record_media_saved(
            db_path,
            post_id=post_id,
            source_mode="MANUAL_SINGLE",
            asset_kind="IMAGE",
            relative_path="../outside.jpg",
            file_name="outside.jpg",
            file_size=1,
        )


def test_mysql_additive_schema() -> None:
    v45 = (
        SHARED_DIR / "storage" / "mysql" / "migration_v45_from_yellow_v44.sql"
    ).read_text(encoding="utf-8")
    v46 = (
        SHARED_DIR / "storage" / "mysql" / "migration_v46_media_saved_flow.sql"
    ).read_text(encoding="utf-8")
    combined = v45 + "\n" + v46
    upper = combined.upper()
    assert "DROP TABLE" not in upper and "TRUNCATE TABLE" not in upper
    assert "ALTER TABLE `IG_MEDIA`" not in upper
    assert "ALTER TABLE `IG_MEDIA_USER_DOWNLOAD`" not in upper
    created = {
        match.group(1)
        for match in re.finditer(
            r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS\s+`([^`]+)`", combined, re.I
        )
    }
    assert {
        "collector_members",
        "collector_devices",
        "collector_groups",
        "collector_posts",
        "collector_post_stat_latest",
        "collector_asset_inventory",
        "collector_transcripts",
        "collector_hook_segments",
        "collector_categories",
        "collector_reviews",
        "collector_comments",
        "collector_aggregates",
        "collector_tasks",
        "collector_sync_events",
        "collector_point_ledger",
        "collector_download_batches",
        "collector_download_jobs",
        "collector_media_saved_events",
    }.issubset(created)
    assert "`collector_code` varchar(8) NOT NULL" in v45
    assert "`token_hash` char(64)" in v45
    assert "local_path" not in v46.lower()
    assert "relative_path" not in v46.lower()
    assert "source_mode" in v46 and "client_event_id" in v46

def test_record_downloaded_file_creates_context_for_actual_download_root(tmp_path: Path) -> None:
    st_root = tmp_path / "_St"
    st_root.mkdir()
    media_dir = tmp_path / "BrowserDownloads" / "campaign" / "account_a"
    media_dir.mkdir(parents=True)
    media_file = media_dir / "account_a_upload_unknown_No없음_CODE123.mp4"
    media_file.write_bytes(b"video-data")

    result = record_downloaded_file(
        st_root=st_root,
        file_path=media_file,
        source_mode="BULK_ALL",
        shortcode="CODE123",
        owner_username="account_a",
        asset_kind="VIDEO",
        source_url="https://cdn.example/video.mp4",
        post_url="https://www.instagram.com/reel/CODE123/",
        md5="m" * 32,
        sha256="s" * 64,
        file_size=media_file.stat().st_size,
        account_path="account_a",
        batch_key="batch-all-1",
        requested_count=1,
        event_uuid="event-all-1",
        stats={"views": 100, "likes": 10, "comments": 2, "reposts": 1, "captured_at": "2026-07-19T00:00:00"},
    )

    db_path = default_database_path(st_root)
    with connect_database(db_path) as conn:
        event = conn.execute("SELECT * FROM media_saved_events WHERE id = ?", (result["event_id"],)).fetchone()
        post = conn.execute("SELECT * FROM posts WHERE id = ?", (result["post_id"],)).fetchone()
        batch = conn.execute("SELECT * FROM download_batches WHERE id = ?", (result["batch_id"],)).fetchone()
        job = conn.execute("SELECT * FROM download_jobs WHERE id = ?", (result["job_id"],)).fetchone()
        stat = conn.execute("SELECT * FROM post_stat_latest WHERE post_id = ?", (result["post_id"],)).fetchone()
        root = conn.execute("SELECT * FROM storage_roots WHERE id = ?", (result["storage_root_id"],)).fetchone()
        assert event["relative_path"] == "account_a/" + media_file.name
        assert Path(root["root_path"]) == media_dir.parent
        assert post["media_state"] == "MEDIA_AVAILABLE"
        assert batch["source_mode"] == "BULK_ALL"
        assert batch["batch_status"] == "COMPLETE"
        assert job["job_status"] == "COMPLETE"
        assert stat["views"] == 100
        payload = json.loads(conn.execute("SELECT payload_json FROM sync_queue WHERE entity_type = 'MEDIA_SAVED'").fetchone()[0])
        assert "relative_path" not in payload
        assert "storage_root_path" not in payload


def test_record_downloaded_file_is_idempotent_across_ws_retry(tmp_path: Path) -> None:
    st_root = tmp_path / "_St"
    st_root.mkdir()
    media_dir = tmp_path / "downloads" / "top" / "account_b"
    media_dir.mkdir(parents=True)
    media_file = media_dir / "account_b_upload_unknown_No001_CODE999.jpg"
    media_file.write_bytes(b"image-data")
    kwargs = dict(
        st_root=st_root,
        file_path=media_file,
        source_mode="TOP_SORTED",
        shortcode="CODE999",
        owner_username="account_b",
        asset_kind="IMAGE",
        rank_no=1,
        account_path="account_b",
        batch_key="top-batch-1",
        event_uuid="top-event-1",
        file_size=media_file.stat().st_size,
    )
    first = record_downloaded_file(**kwargs)
    second = record_downloaded_file(**kwargs)
    assert first["event_id"] == second["event_id"]
    assert second["idempotent"] is True
    with connect_database(default_database_path(st_root)) as conn:
        assert conn.execute("SELECT COUNT(*) FROM media_saved_events").fetchone()[0] == 1
        assert conn.execute("SELECT COUNT(*) FROM sync_queue WHERE entity_type = 'MEDIA_SAVED'").fetchone()[0] == 1


def test_download_batch_completes_only_after_requested_files(tmp_path: Path) -> None:
    st_root = tmp_path / "_St"
    st_root.mkdir()
    media_dir = tmp_path / "downloads" / "bulk" / "account_c"
    media_dir.mkdir(parents=True)
    first_file = media_dir / "account_c_upload_unknown_No없음_AAA.jpg"
    second_file = media_dir / "account_c_upload_unknown_No없음_BBB.jpg"
    first_file.write_bytes(b"one")
    second_file.write_bytes(b"two")

    first = record_downloaded_file(
        st_root=st_root,
        file_path=first_file,
        source_mode="BULK_FILTERED",
        shortcode="AAA",
        owner_username="account_c",
        asset_kind="IMAGE",
        account_path="account_c",
        batch_key="filtered-batch-two",
        requested_count=2,
        event_uuid="filtered-event-one",
        file_size=3,
    )
    db_path = default_database_path(st_root)
    with connect_database(db_path) as conn:
        status = conn.execute("SELECT batch_status FROM download_batches WHERE id = ?", (first["batch_id"],)).fetchone()[0]
        assert status == "RUNNING"

    second = record_downloaded_file(
        st_root=st_root,
        file_path=second_file,
        source_mode="BULK_FILTERED",
        shortcode="BBB",
        owner_username="account_c",
        asset_kind="IMAGE",
        account_path="account_c",
        batch_key="filtered-batch-two",
        requested_count=2,
        event_uuid="filtered-event-two",
        file_size=3,
    )
    assert second["batch_id"] == first["batch_id"]
    with connect_database(db_path) as conn:
        status = conn.execute("SELECT batch_status FROM download_batches WHERE id = ?", (first["batch_id"],)).fetchone()[0]
        assert status == "COMPLETE"


def test_observation_10m_repository_and_soft_delete(tmp_path: Path) -> None:
    _, db_path = make_db(tmp_path)
    first = upsert_observation(
        db_path, username="account_a", shortcode="POST1", views=100, likes=10, comments=2,
        observed_at="2026-07-19T14:02:37Z",
    )
    same_bucket = upsert_observation(
        db_path, username="account_a", shortcode="POST1", views=130, likes=12, comments=3,
        observed_at="2026-07-19T14:09:59Z",
    )
    next_bucket = upsert_observation(
        db_path, username="account_a", shortcode="POST1", views=190, likes=18, comments=5,
        observed_at="2026-07-19T14:12:01Z",
    )
    assert first["bucket_at"] == same_bucket["bucket_at"] == "2026-07-19T14:00:00Z"
    assert next_bucket["bucket_at"] == "2026-07-19T14:10:00Z"
    stats = get_post_stats(db_path, first["post_id"])
    assert len(stats) == 2
    assert stats[0]["views"] == 190 and stats[1]["views"] == 130
    assert stats[1]["sample_count"] == 2

    accounts = list_accounts(db_path)
    assert accounts["total"] == 1 and accounts["items"][0]["username"] == "account_a"
    posts = list_account_posts(db_path, account_id=first["account_id"], sort_by="views_delta")
    assert posts["total"] == 1
    assert posts["items"][0]["views_delta"] == 60
    assert posts["items"][0]["views_per_hour"] == 360

    deleted_post = soft_delete_post(db_path, first["post_id"])
    assert deleted_post["deleted"] is True
    assert list_account_posts(db_path, account_id=first["account_id"])["total"] == 0

    upsert_observation(
        db_path, username="account_b", shortcode="POST2", views=20,
        observed_at="2026-07-19T15:01:00Z",
    )
    account_b = next(item for item in list_accounts(db_path)["items"] if item["username"] == "account_b")
    deleted_account = soft_delete_account(db_path, account_b["id"])
    assert deleted_account["username"] == "account_b"
    ignored = upsert_observation(
        db_path, username="account_b", shortcode="POST3", views=999,
        observed_at="2026-07-19T15:12:00Z",
    )
    assert ignored["ignored_deleted"] is True and ignored["deleted_entity"] == "ACCOUNT"
    assert all(item["username"] != "account_b" for item in list_accounts(db_path)["items"])
    assert any(item["username"] == "account_b" for item in list_deleted_accounts(db_path))
    conn = connect_database(db_path)
    try:
        rule = conn.execute("SELECT active, deleted_at FROM social_accounts WHERE id=?", (account_b["id"],)).fetchone()
        assert rule["active"] == 0 and rule["deleted_at"]
        sync_ops = conn.execute("SELECT entity_type, operation FROM sync_queue WHERE operation='DELETE'").fetchall()
        assert {(row["entity_type"], row["operation"]) for row in sync_ops} >= {("POST", "DELETE"), ("ACCOUNT", "DELETE")}
    finally:
        conn.close()
