import pytest
import aiosqlite
from datetime import datetime, timezone, timedelta
from db import (
    init_db, record_start, record_subscription, get_verified_users,
    mark_blocked, get_stats, get_unsubscribed_users, get_all_users,
    get_total_users, record_material_event, record_broadcast,
    get_material_click_stats, get_gate_stats, get_started_since,
    get_recent_broadcasts,
)

DB_PATH = ":memory:"


@pytest.fixture
async def conn():
    async with aiosqlite.connect(DB_PATH) as db:
        await init_db(db)
        yield db


@pytest.mark.asyncio
async def test_record_start_creates_user(conn):
    await record_start(conn, user_id=111, conference_id="conf-1")
    async with conn.execute("SELECT user_id FROM users WHERE user_id=111") as cur:
        row = await cur.fetchone()
    assert row is not None


@pytest.mark.asyncio
async def test_record_start_is_idempotent(conn):
    await record_start(conn, user_id=111, conference_id="conf-1")
    await record_start(conn, user_id=111, conference_id="conf-1")
    async with conn.execute("SELECT COUNT(*) FROM users WHERE user_id=111") as cur:
        row = await cur.fetchone()
    assert row[0] == 1


@pytest.mark.asyncio
async def test_record_subscription_sets_timestamp(conn):
    await record_start(conn, user_id=222, conference_id="conf-1")
    await record_subscription(conn, user_id=222, conference_id="conf-1")
    async with conn.execute("SELECT subscribed_at FROM users WHERE user_id=222") as cur:
        row = await cur.fetchone()
    assert row[0] is not None


@pytest.mark.asyncio
async def test_get_verified_users_excludes_unverified(conn):
    await record_start(conn, user_id=333, conference_id="conf-1")
    users = await get_verified_users(conn, "conf-1")
    assert 333 not in users


@pytest.mark.asyncio
async def test_get_verified_users_excludes_blocked(conn):
    await record_start(conn, user_id=444, conference_id="conf-1")
    await record_subscription(conn, user_id=444, conference_id="conf-1")
    await mark_blocked(conn, user_id=444)
    users = await get_verified_users(conn, "conf-1")
    assert 444 not in users


@pytest.mark.asyncio
async def test_get_verified_users_returns_subscribed(conn):
    await record_start(conn, user_id=555, conference_id="conf-1")
    await record_subscription(conn, user_id=555, conference_id="conf-1")
    users = await get_verified_users(conn, "conf-1")
    assert 555 in users


@pytest.mark.asyncio
async def test_get_stats_counts_correctly(conn):
    await record_start(conn, user_id=601, conference_id="conf-1")
    await record_start(conn, user_id=602, conference_id="conf-1")
    await record_subscription(conn, user_id=601, conference_id="conf-1")
    await mark_blocked(conn, user_id=603)
    stats = await get_stats(conn, "conf-1")
    assert stats["started"] == 2
    assert stats["subscribed"] == 1
    assert stats["blocked"] == 1


@pytest.mark.asyncio
async def test_record_subscription_raises_for_missing_user(conn):
    with pytest.raises(ValueError, match="no row"):
        await record_subscription(conn, user_id=999, conference_id="conf-x")


@pytest.mark.asyncio
async def test_get_stats_does_not_leak_across_conferences(conn):
    await record_start(conn, user_id=701, conference_id="conf-A")
    await record_subscription(conn, user_id=701, conference_id="conf-A")
    await record_start(conn, user_id=702, conference_id="conf-B")
    stats_a = await get_stats(conn, "conf-A")
    assert stats_a["started"] == 1
    assert stats_a["subscribed"] == 1


@pytest.mark.asyncio
async def test_get_unsubscribed_returns_unsubscribed(conn):
    await record_start(conn, user_id=801, conference_id="conf-1")
    users = await get_unsubscribed_users(conn, "conf-1")
    assert 801 in users


@pytest.mark.asyncio
async def test_get_unsubscribed_excludes_subscribed(conn):
    await record_start(conn, user_id=802, conference_id="conf-1")
    await record_subscription(conn, user_id=802, conference_id="conf-1")
    users = await get_unsubscribed_users(conn, "conf-1")
    assert 802 not in users


@pytest.mark.asyncio
async def test_get_unsubscribed_excludes_blocked(conn):
    await record_start(conn, user_id=803, conference_id="conf-1")
    await mark_blocked(conn, user_id=803)
    users = await get_unsubscribed_users(conn, "conf-1")
    assert 803 not in users


@pytest.mark.asyncio
async def test_get_unsubscribed_empty_when_all_subscribed(conn):
    await record_start(conn, user_id=804, conference_id="conf-1")
    await record_subscription(conn, user_id=804, conference_id="conf-1")
    users = await get_unsubscribed_users(conn, "conf-1")
    assert users == []


@pytest.mark.asyncio
async def test_get_all_users_returns_distinct_across_events(conn):
    await record_start(conn, user_id=101, conference_id="conf-1")
    await record_start(conn, user_id=101, conference_id="webinar-1")
    await record_start(conn, user_id=102, conference_id="conf-1")
    users = await get_all_users(conn, exclude_user_ids=set())
    assert sorted(users) == [101, 102]


@pytest.mark.asyncio
async def test_get_all_users_excludes_blocked(conn):
    await record_start(conn, user_id=201, conference_id="conf-1")
    await record_start(conn, user_id=202, conference_id="conf-1")
    await mark_blocked(conn, user_id=202)
    users = await get_all_users(conn, exclude_user_ids=set())
    assert 201 in users
    assert 202 not in users


@pytest.mark.asyncio
async def test_get_all_users_excludes_passed_ids(conn):
    await record_start(conn, user_id=301, conference_id="conf-1")
    await record_start(conn, user_id=302, conference_id="conf-1")
    users = await get_all_users(conn, exclude_user_ids={302})
    assert users == [301]


@pytest.mark.asyncio
async def test_get_all_users_empty_exclude_does_not_break(conn):
    await record_start(conn, user_id=401, conference_id="conf-1")
    users = await get_all_users(conn, exclude_user_ids=set())
    assert users == [401]


@pytest.mark.asyncio
async def test_get_all_users_returns_empty_when_no_users(conn):
    users = await get_all_users(conn, exclude_user_ids={999})
    assert users == []


@pytest.mark.asyncio
async def test_get_total_users_counts_distinct_across_events(conn):
    await record_start(conn, user_id=501, conference_id="conf-1")
    await record_start(conn, user_id=501, conference_id="webinar-1")
    await record_start(conn, user_id=502, conference_id="conf-1")
    total = await get_total_users(conn)
    assert total == 2


@pytest.mark.asyncio
async def test_get_total_users_excludes_blocked(conn):
    await record_start(conn, user_id=601, conference_id="conf-1")
    await record_start(conn, user_id=602, conference_id="conf-1")
    await mark_blocked(conn, user_id=602)
    total = await get_total_users(conn)
    assert total == 1


@pytest.mark.asyncio
async def test_get_total_users_zero_when_no_users(conn):
    total = await get_total_users(conn)
    assert total == 0


@pytest.mark.asyncio
async def test_record_start_clears_blocked(conn):
    await record_start(conn, user_id=900, conference_id="c1")
    await mark_blocked(conn, user_id=900)
    # returning user restarts → block flag lifted
    await record_start(conn, user_id=900, conference_id="c2")
    async with conn.execute("SELECT COUNT(*) FROM blocked_users WHERE user_id=900") as cur:
        assert (await cur.fetchone())[0] == 0


@pytest.mark.asyncio
async def test_material_click_stats_counts_and_uniques(conn):
    await record_material_event(conn, 1, "c1", "mat-a", "click")
    await record_material_event(conn, 1, "c1", "mat-a", "click")  # same user twice
    await record_material_event(conn, 2, "c1", "mat-a", "click")
    await record_material_event(conn, 3, "c1", "mat-b", "click")
    rows = await get_material_click_stats(conn, "c1")
    assert rows[0] == {"material_id": "mat-a", "clicks": 3, "uniques": 2}
    assert rows[1] == {"material_id": "mat-b", "clicks": 1, "uniques": 1}


@pytest.mark.asyncio
async def test_material_click_stats_ignores_non_click(conn):
    await record_material_event(conn, 1, "c1", "mat-a", "delivered")
    rows = await get_material_click_stats(conn, "c1")
    assert rows == []


@pytest.mark.asyncio
async def test_material_click_stats_scoped_to_conf(conn):
    await record_material_event(conn, 1, "c1", "mat-a", "click")
    await record_material_event(conn, 2, "c2", "mat-a", "click")
    rows = await get_material_click_stats(conn, "c1")
    assert len(rows) == 1 and rows[0]["clicks"] == 1


@pytest.mark.asyncio
async def test_gate_stats_delivered_and_stuck(conn):
    # user 1: delivered (subscribed). user 2: gate_shown then delivered → delivered, not stuck.
    # user 3: only gate_shown → stuck.
    await record_material_event(conn, 1, "c1", "m", "delivered")
    await record_material_event(conn, 2, "c1", "m", "gate_shown")
    await record_material_event(conn, 2, "c1", "m", "delivered")
    await record_material_event(conn, 3, "c1", "m", "gate_shown")
    stats = await get_gate_stats(conn, "c1")
    assert stats == {"delivered": 2, "stuck": 1}


@pytest.mark.asyncio
async def test_gate_stats_empty(conn):
    assert await get_gate_stats(conn, "c1") == {"delivered": 0, "stuck": 0}


@pytest.mark.asyncio
async def test_started_since_window(conn):
    await record_start(conn, user_id=1, conference_id="c1")
    future = (datetime.now(timezone.utc) + timedelta(days=1)).isoformat()
    past = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
    assert await get_started_since(conn, "c1", past) == 1
    assert await get_started_since(conn, "c1", future) == 0


@pytest.mark.asyncio
async def test_started_since_scoped_to_conf(conn):
    await record_start(conn, user_id=1, conference_id="c1")
    await record_start(conn, user_id=2, conference_id="c2")
    past = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
    assert await get_started_since(conn, "c1", past) == 1


@pytest.mark.asyncio
async def test_record_and_get_recent_broadcasts(conn):
    await record_broadcast(conn, "c1", "всем", delivered=10, skipped=2, errors=1)
    await record_broadcast(conn, "c1", "zoom-ссылка", delivered=5, skipped=0, errors=0)
    rows = await get_recent_broadcasts(conn, limit=5)
    assert len(rows) == 2
    # newest first
    assert rows[0]["kind"] == "zoom-ссылка"
    assert rows[0]["delivered"] == 5
    assert rows[1]["kind"] == "всем"


@pytest.mark.asyncio
async def test_get_recent_broadcasts_respects_limit(conn):
    for i in range(4):
        await record_broadcast(conn, "c1", f"b{i}", delivered=i, skipped=0, errors=0)
    rows = await get_recent_broadcasts(conn, limit=2)
    assert len(rows) == 2