import asyncio
import json
from datetime import datetime, timezone

import pytest

from src import gitea_proxy
from src.push_notifications import PushConfiguration, dispatch_deadline_reminders
from src.push_subscription_store import PushSubscriptionStore


@pytest.mark.anyio
async def test_deadline_reminder_sends_one_private_local_day_digest_and_deduplicates(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("device-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    store.set_deadline_preferences(
        "device-a", enabled=True, timezone="America/New_York", reminder_hour=9
    )
    sent = []

    async def assigned():
        return {
            "complete": True,
            "items": [
                {
                    "id": 42,
                    "title": "Private launch plan",
                    "repository": {"full_name": "private/repo"},
                    "due_date": "2026-08-14T12:00:00Z",
                },
                {"id": 43, "due_date": "2026-08-20T12:00:00Z"},
            ],
        }

    async def send(_subscription, payload):
        sent.append(json.loads(payload))

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    now = datetime(2026, 8, 13, 13, 5, tzinfo=timezone.utc)

    assert await dispatch_deadline_reminders(store, config, assigned, send, now=now) == 1
    assert await dispatch_deadline_reminders(store, config, assigned, send, now=now) == 0
    assert sent == [{
        "title": "1 deadline needs attention",
        "body": "Open Agenda to review or replan it.",
        "route": "#/my-work/agenda",
        "protect_route": "#/my-work/agenda/protect-today",
        "tag": "stackchain-deadline-digest-2026-08-13",
        "deadline_count": 1,
    }]
    assert "Private launch plan" not in json.dumps(sent)
    assert "private/repo" not in json.dumps(sent)


@pytest.mark.anyio
async def test_deadline_reminder_counts_calendar_days_per_device_timezone(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for device, timezone_name in (("tokyo", "Asia/Tokyo"), ("la", "America/Los_Angeles")):
        store.upsert(device, {
            "endpoint": f"https://push.example/{device}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })
        store.set_deadline_preferences(device, enabled=True, timezone=timezone_name, reminder_hour=0)

    async def assigned():
        return {"complete": True, "items": [
            {"id": 1, "due_date": "2026-08-15T23:59:59Z"},
            {"id": 2, "due_date": "2026-08-16T23:59:59Z"},
        ]}

    sent = {}
    async def send(subscription, payload):
        device = subscription["endpoint"].rsplit("/", 1)[-1]
        sent[device] = json.loads(payload)["deadline_count"]

    delivered = await dispatch_deadline_reminders(
        store, PushConfiguration("public", "private", "mailto:ops@example.com"),
        assigned, send, now=datetime(2026, 8, 13, 23, 30, tzinfo=timezone.utc),
    )

    assert delivered == 2
    assert sent == {"tokyo": 2, "la": 1}


@pytest.mark.anyio
async def test_deadline_reminder_uses_each_devices_confirmed_horizon(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for device, reminder_days in (("today", 0), ("week", 7)):
        store.upsert(device, {
            "endpoint": f"https://push.example/{device}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })
        store.set_deadline_preferences(
            device,
            enabled=True,
            timezone="UTC",
            reminder_hour=9,
            reminder_days=reminder_days,
        )

    async def assigned():
        return {"complete": True, "items": [
            {"id": 1, "due_date": "2026-08-13"},
            {"id": 2, "due_date": "2026-08-15"},
            {"id": 3, "due_date": "2026-08-20"},
            {"id": 4, "due_date": "2026-08-21"},
        ]}

    sent = {}

    async def send(subscription, payload):
        device = subscription["endpoint"].rsplit("/", 1)[-1]
        sent[device] = json.loads(payload)["deadline_count"]

    delivered = await dispatch_deadline_reminders(
        store,
        PushConfiguration("public", "private", "mailto:ops@example.com"),
        assigned,
        send,
        now=datetime(2026, 8, 13, 10, 0, tzinfo=timezone.utc),
    )

    assert delivered == 2
    assert sent == {"today": 1, "week": 3}


@pytest.mark.anyio
async def test_deadline_reminder_fails_closed_for_incomplete_snapshot_and_before_local_hour(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("device-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    store.set_deadline_preferences(
        "device-a", enabled=True, timezone="America/Los_Angeles", reminder_hour=9
    )
    sent = []

    async def incomplete():
        return {"complete": False, "items": [{"id": 42, "due_date": "2026-08-14T12:00:00Z"}]}

    async def complete():
        return {"complete": True, "items": [{"id": 42, "due_date": "2026-08-14T12:00:00Z"}]}

    async def send(_subscription, payload):
        sent.append(payload)

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_deadline_reminders(
        store, config, incomplete, send,
        now=datetime(2026, 8, 13, 18, 0, tzinfo=timezone.utc),
    ) == 0
    assert await dispatch_deadline_reminders(
        store, config, complete, send,
        now=datetime(2026, 8, 13, 15, 0, tzinfo=timezone.utc),
    ) == 0
    assert sent == []


@pytest.mark.anyio
async def test_deadline_reminder_skips_snapshot_until_a_device_reaches_its_local_hour(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("device-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    store.set_deadline_preferences(
        "device-a", enabled=True, timezone="America/Los_Angeles", reminder_hour=9
    )
    snapshot_calls = 0

    async def assigned():
        nonlocal snapshot_calls
        snapshot_calls += 1
        return {"complete": True, "items": []}

    delivered = await dispatch_deadline_reminders(
        store,
        PushConfiguration("public", "private", "mailto:ops@example.com"),
        assigned,
        now=datetime(2026, 8, 13, 15, 0, tzinfo=timezone.utc),
    )

    assert delivered == 0
    assert snapshot_calls == 0


@pytest.mark.anyio
async def test_deadline_reminder_skips_snapshot_after_every_device_was_delivered_today(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for session_id, timezone_name in (
        ("device-a", "UTC"),
        ("device-b", "America/New_York"),
    ):
        store.upsert(session_id, {
            "endpoint": f"https://push.example/{session_id}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })
        store.set_deadline_preferences(
            session_id, enabled=True, timezone=timezone_name, reminder_hour=9
        )
        store.mark_deadline_reminder_delivered(session_id, "2026-08-13")
    snapshot_calls = 0

    async def assigned():
        nonlocal snapshot_calls
        snapshot_calls += 1
        return {"complete": True, "items": []}

    delivered = await dispatch_deadline_reminders(
        store,
        PushConfiguration("public", "private", "mailto:ops@example.com"),
        assigned,
        now=datetime(2026, 8, 13, 15, 0, tzinfo=timezone.utc),
    )

    assert delivered == 0
    assert snapshot_calls == 0


def test_deadline_preferences_persist_on_the_existing_device_subscription(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("device-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })

    store.set_deadline_preferences(
        "device-a", enabled=True, timezone="Europe/London", reminder_hour=8
    )

    assert store.deadline_preferences("device-a") == {
        "enabled": True,
        "timezone": "Europe/London",
        "reminder_hour": 8,
        "reminder_days": 2,
    }


def test_existing_deadline_preferences_migrate_to_two_day_horizon(tmp_path):
    database = tmp_path / "push.sqlite3"
    import sqlite3

    with sqlite3.connect(database) as connection:
        connection.executescript("""
            CREATE TABLE push_subscriptions (
                session_id TEXT PRIMARY KEY,
                endpoint TEXT NOT NULL UNIQUE,
                subscription_json TEXT NOT NULL
            );
            CREATE TABLE push_deadline_preferences (
                session_id TEXT PRIMARY KEY,
                enabled INTEGER NOT NULL DEFAULT 0,
                timezone TEXT NOT NULL DEFAULT 'UTC',
                reminder_hour INTEGER NOT NULL DEFAULT 9,
                delivered_local_day TEXT
            );
            INSERT INTO push_subscriptions VALUES ('device-a', 'https://push.example/a', '{}');
            INSERT INTO push_deadline_preferences VALUES ('device-a', 1, 'UTC', 8, NULL);
        """)

    store = PushSubscriptionStore(database)

    assert store.deadline_preferences("device-a")["reminder_days"] == 2


@pytest.mark.anyio
async def test_assigned_deadline_snapshot_is_pagination_complete(monkeypatch):
    pages = {
        1: {"items": [{"id": 1}], "total": 2, "has_more": True},
        2: {"items": [{"id": 2}], "total": 2, "has_more": False},
    }

    async def work_page(stream, page, limit):
        assert stream == "issue"
        assert limit == 1
        return pages[page]

    monkeypatch.setattr(gitea_proxy, "work_page", work_page)

    assert await gitea_proxy.assigned_issue_snapshot(limit=1) == {
        "items": [{"id": 1}, {"id": 2}],
        "complete": True,
    }


@pytest.mark.anyio
async def test_assigned_deadline_snapshot_fetches_remaining_pages_concurrently(monkeypatch):
    active = 0
    peak = 0
    remaining_started = asyncio.Event()

    async def work_page(stream, page, limit):
        nonlocal active, peak
        assert stream == "issue"
        assert limit == 1
        if page == 1:
            return {"items": [{"id": 1}], "total": 5, "has_more": True}
        active += 1
        peak = max(peak, active)
        if active == 2:
            remaining_started.set()
        await asyncio.wait_for(remaining_started.wait(), timeout=0.2)
        await asyncio.sleep(0)
        active -= 1
        return {"items": [{"id": page}], "total": 5, "has_more": page < 5}

    monkeypatch.setattr(gitea_proxy, "work_page", work_page)

    assert await gitea_proxy.assigned_issue_snapshot(limit=1, max_concurrency=2) == {
        "items": [{"id": 1}, {"id": 2}, {"id": 3}, {"id": 4}, {"id": 5}],
        "complete": True,
    }
    assert peak == 2


@pytest.mark.anyio
@pytest.mark.parametrize(
    "remaining",
    [
        {"items": [{"id": 2}], "total": 3, "has_more": True},
        {"items": [{"id": 1}], "total": 2, "has_more": False},
    ],
)
async def test_assigned_deadline_snapshot_rejects_changed_or_duplicate_pages(
    monkeypatch, remaining
):
    async def work_page(_stream, page, _limit):
        if page == 1:
            return {"items": [{"id": 1}], "total": 2, "has_more": True}
        return remaining

    monkeypatch.setattr(gitea_proxy, "work_page", work_page)

    with pytest.raises(ValueError, match="incomplete|changed"):
        await gitea_proxy.assigned_issue_snapshot(limit=1)


@pytest.mark.anyio
async def test_assigned_deadline_snapshot_enforces_aggregate_deadline(monkeypatch):
    cancelled = asyncio.Event()

    async def work_page(_stream, _page, _limit):
        try:
            await asyncio.sleep(60)
        finally:
            cancelled.set()

    monkeypatch.setattr(gitea_proxy, "work_page", work_page)

    with pytest.raises(TimeoutError):
        await gitea_proxy.assigned_issue_snapshot(deadline_seconds=0.01)
    assert cancelled.is_set()


@pytest.mark.anyio
async def test_competing_workers_send_one_deadline_digest(tmp_path):
    path = tmp_path / "push.sqlite3"
    first = PushSubscriptionStore(path)
    second = PushSubscriptionStore(path)
    first.upsert("device-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    first.set_deadline_preferences(
        "device-a", enabled=True, timezone="UTC", reminder_hour=9
    )
    sending = __import__("asyncio").Event()
    release = __import__("asyncio").Event()
    sent = []

    async def assigned():
        return {"complete": True, "items": [{"id": 1, "due_date": "2026-08-14T00:00:00Z"}]}

    async def send(_subscription, payload):
        sent.append(payload)
        sending.set()
        await release.wait()

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    now = datetime(2026, 8, 13, 10, 0, tzinfo=timezone.utc)
    active = __import__("asyncio").create_task(
        dispatch_deadline_reminders(first, config, assigned, send, now=now)
    )
    await sending.wait()
    competing = await dispatch_deadline_reminders(second, config, assigned, send, now=now)
    release.set()

    assert competing == 0
    assert await active == 1
    assert len(sent) == 1


@pytest.mark.anyio
async def test_deadline_reminders_bound_fanout_and_isolate_failed_devices(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for index in range(4):
        session_id = f"device-{index}"
        store.upsert(session_id, {
            "endpoint": f"https://push.example/{session_id}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })
        store.set_deadline_preferences(
            session_id, enabled=True, timezone="UTC", reminder_hour=9
        )
    active = 0
    peak = 0
    two_active = asyncio.Event()

    async def assigned():
        return {"complete": True, "items": [{"id": 1, "due_date": "2026-08-14T00:00:00Z"}]}

    async def send(subscription, _payload):
        nonlocal active, peak
        active += 1
        peak = max(peak, active)
        if active == 2:
            two_active.set()
        await asyncio.wait_for(two_active.wait(), timeout=0.5)
        active -= 1
        if subscription["endpoint"].endswith("device-1"):
            raise RuntimeError("provider failed")

    delivered = await dispatch_deadline_reminders(
        store,
        PushConfiguration("public", "private", "mailto:ops@example.com"),
        assigned,
        send,
        now=datetime(2026, 8, 13, 10, 0, tzinfo=timezone.utc),
        max_concurrency=2,
    )

    assert delivered == 3
    assert peak == 2


@pytest.mark.anyio
async def test_deadline_reminders_bulk_authorize_due_devices_and_prune_inactive(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for session_id in ("active-device", "revoked-device"):
        store.upsert(session_id, {
            "endpoint": f"https://push.example/{session_id}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })
        store.set_deadline_preferences(
            session_id, enabled=True, timezone="UTC", reminder_hour=9
        )
    authorization_calls = []
    sent = []

    async def assigned():
        return {"complete": True, "items": [{"id": 1, "due_date": "2026-08-14T00:00:00Z"}]}

    async def session_statuses(management_ids):
        authorization_calls.append(list(management_ids))
        return {
            management_id: "active" if management_id == "active-device" else "revoked"
            for management_id in management_ids
        }

    async def send(subscription, _payload):
        sent.append(subscription["endpoint"])

    delivered = await dispatch_deadline_reminders(
        store,
        PushConfiguration("public", "private", "mailto:ops@example.com"),
        assigned,
        send,
        now=datetime(2026, 8, 13, 10, 0, tzinfo=timezone.utc),
        session_statuses=session_statuses,
    )

    assert delivered == 1
    assert authorization_calls == [["active-device", "revoked-device"]]
    assert sent == ["https://push.example/active-device"]
    assert store.is_subscribed("revoked-device") is False