import asyncio
import json
import sqlite3

import httpx
import pytest

from src import main
from src.live_snapshot_store import LiveSnapshotStore


@pytest.fixture(autouse=True)
def reset_live_snapshot_task(tmp_path, monkeypatch):
    monkeypatch.setattr(
        main,
        "_live_snapshot_store",
        LiveSnapshotStore(tmp_path / "live.sqlite3", clock=lambda: main._live_snapshot_clock()),
    )
    main._live_snapshot_task = None
    main._live_snapshot_value = None
    main._live_snapshot_created_at = None
    main._live_section_created_at = {
        section: None for section in main.LIVE_SNAPSHOT_SECTIONS
    }
    main._live_section_failure_count = {
        section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
    }
    main._live_section_retry_at = {
        section: None for section in main.LIVE_SNAPSHOT_SECTIONS
    }
    main._live_snapshot_refreshing_sections = set()
    main._live_section_revisions = {
        section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
    }
    yield
    main._live_snapshot_task = None
    main._live_snapshot_value = None
    main._live_snapshot_created_at = None
    main._live_section_created_at = {
        section: None for section in main.LIVE_SNAPSHOT_SECTIONS
    }
    main._live_section_failure_count = {
        section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
    }
    main._live_section_retry_at = {
        section: None for section in main.LIVE_SNAPSHOT_SECTIONS
    }
    main._live_snapshot_refreshing_sections = set()
    main._live_section_revisions = {
        section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
    }


def payload(response):
    return json.loads(response.body)


@pytest.mark.anyio
async def test_live_snapshot_fetches_user_once_and_updates_work_and_activity(monkeypatch):
    calls = {"user": 0}

    async def user():
        calls["user"] += 1
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    async def events(authenticated_user):
        assert authenticated_user["login"] == "timmy"
        return [{"type": "push"}]

    async def updates():
        return {
            "items": [{"id": 42, "title": "Mentioned you"}],
            "page": 1,
            "total": 125,
            "has_more": True,
        }

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", events)
    monkeypatch.setattr(main, "notifications", updates)

    response = await main.live_snapshot()
    result = payload(response)

    assert calls["user"] == 1
    assert result["context"]["user"]["login"] == "timmy"
    assert result["events"] == [{"type": "push"}]
    assert result["notifications"] == [{"id": 42, "title": "Mentioned you"}]
    assert result["notification_pagination"] == {
        "page": 1, "total": 125, "has_more": True
    }
    assert result["sections"] == {
        "context": "fresh", "events": "fresh", "notifications": "fresh"
    }
    assert set(result["revisions"]) == {"context", "events", "notifications"}
    assert len(set(result["revisions"].values())) == 1
    assert result["revisions"]["context"].endswith(".1")


@pytest.mark.anyio
async def test_live_snapshot_omits_section_bodies_at_known_revisions(monkeypatch):
    async def user():
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    async def events(_authenticated_user):
        return [{"type": "push"}]

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", events)
    monkeypatch.setattr(main, "notifications", empty)

    first = payload(await main.live_snapshot())
    second = payload(await main.live_snapshot(
        context_revision=first["revisions"]["context"],
        events_revision=first["revisions"]["events"],
        notifications_revision=first["revisions"]["notifications"],
    ))

    assert "context" not in second
    assert "events" not in second
    assert "notifications" not in second
    assert "notification_pagination" not in second
    assert second["sections"] == first["sections"]
    assert second["revisions"] == first["revisions"]
    assert "freshness" in second


@pytest.mark.anyio
async def test_warm_unchanged_polls_only_load_shared_metadata(monkeypatch):
    async def user():
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", lambda _user: empty())
    monkeypatch.setattr(main, "notifications", empty)
    first = payload(await main.live_snapshot())
    full_load = main._live_snapshot_store.load
    metadata_load = main._live_snapshot_store.load_metadata
    calls = {"full": 0, "metadata": 0}

    def counted_full_load():
        calls["full"] += 1
        return full_load()

    def counted_metadata_load():
        calls["metadata"] += 1
        return metadata_load()

    monkeypatch.setattr(main._live_snapshot_store, "load", counted_full_load)
    monkeypatch.setattr(main._live_snapshot_store, "load_metadata", counted_metadata_load)

    for _ in range(10):
        response = await main.live_snapshot(
            context_revision=first["revisions"]["context"],
            events_revision=first["revisions"]["events"],
            notifications_revision=first["revisions"]["notifications"],
        )
        assert response.status_code == 200
        assert "context" not in payload(response)

    assert calls == {"full": 0, "metadata": 10}


@pytest.mark.anyio
async def test_live_snapshot_token_from_another_worker_never_suppresses_content(monkeypatch, tmp_path):
    async def user():
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", lambda _user: empty())
    monkeypatch.setattr(main, "notifications", empty)
    worker_a = payload(await main.live_snapshot())
    worker_a_token = worker_a["revisions"]["context"]

    main._live_snapshot_value = None
    main._live_snapshot_created_at = None
    main._live_section_created_at = {
        section: None for section in main.LIVE_SNAPSHOT_SECTIONS
    }
    main._live_section_revisions = {
        section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
    }
    main._live_snapshot_store = LiveSnapshotStore(
        tmp_path / "other-worker.sqlite3", clock=lambda: main._live_snapshot_clock()
    )

    worker_b = payload(await main.live_snapshot(context_revision=worker_a_token))

    assert worker_a_token.endswith(".1")
    assert worker_b["revisions"]["context"].endswith(".1")
    assert worker_b["revisions"]["context"] != worker_a_token
    assert worker_b["context"]["user"]["login"] == "timmy"


@pytest.mark.anyio
async def test_live_snapshot_rejects_malformed_revision_before_upstream_work(monkeypatch):
    async def unexpected_user():
        raise AssertionError("malformed revision reached upstream work")

    monkeypatch.setattr(main, "current_user", unexpected_user)
    transport = httpx.ASGITransport(app=main.app)

    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.get("/api/v1/live?context_revision=not-a-token")

    assert response.status_code == 422


@pytest.mark.anyio
async def test_live_snapshot_keeps_fresh_context_when_activity_fails(monkeypatch):
    async def user():
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    async def failing_events(authenticated_user):
        raise ConnectionError("secret upstream detail")

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", failing_events)
    monkeypatch.setattr(main, "notifications", empty)

    result = payload(await main.live_snapshot())

    assert result["context"]["user"]["login"] == "timmy"
    assert result["events"] is None
    assert result["sections"] == {
        "context": "fresh",
        "events": "temporarily unavailable",
        "notifications": "fresh",
    }
    assert "secret" not in json.dumps(result)


@pytest.mark.anyio
async def test_live_snapshot_keeps_work_and_activity_when_notifications_fail(monkeypatch):
    async def user():
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    async def events(_authenticated_user):
        return [{"type": "push"}]

    async def failing_updates():
        raise ConnectionError("private notification failure")

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", events)
    monkeypatch.setattr(main, "notifications", failing_updates)

    result = payload(await main.live_snapshot())

    assert result["context"]["user"]["login"] == "timmy"
    assert result["events"] == [{"type": "push"}]
    assert result["notifications"] is None
    assert result["sections"]["notifications"] == "temporarily unavailable"
    assert "private" not in json.dumps(result)


@pytest.mark.anyio
async def test_live_snapshot_keeps_fresh_activity_when_work_fails(monkeypatch):
    async def user():
        return {"id": 1, "login": "timmy"}

    async def failing_repos():
        raise ConnectionError("work unavailable")

    async def empty():
        return []

    async def events(authenticated_user):
        return [{"type": "push"}]

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", failing_repos)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", events)
    monkeypatch.setattr(main, "notifications", empty)

    result = payload(await main.live_snapshot())

    assert result["context"] is None
    assert result["events"] == [{"type": "push"}]
    assert result["sections"] == {
        "context": "temporarily unavailable",
        "events": "fresh",
        "notifications": "fresh",
    }


@pytest.mark.anyio
async def test_live_snapshot_reuses_completed_snapshot_inside_freshness_window(monkeypatch):
    user_calls = 0
    user_started = asyncio.Event()
    release = asyncio.Event()

    async def user():
        nonlocal user_calls
        user_calls += 1
        user_started.set()
        return {"id": 1, "login": "timmy"}

    async def blocked_repos():
        await release.wait()
        return []

    async def empty():
        return []

    async def events(authenticated_user):
        return []

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", blocked_repos)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", events)
    monkeypatch.setattr(main, "notifications", empty)

    first = asyncio.create_task(main.live_snapshot())
    await user_started.wait()
    second = asyncio.create_task(main.live_snapshot())

    release.set()
    await asyncio.gather(first, second)
    await main.live_snapshot()

    assert user_calls == 1


@pytest.mark.anyio
async def test_stale_snapshot_returns_immediately_while_one_refresh_revalidates(monkeypatch):
    now = 100.0
    refresh_started = asyncio.Event()
    release_refresh = asyncio.Event()
    builds = 0

    async def snapshot(_sections=None):
        nonlocal builds
        builds += 1
        if builds == 2:
            refresh_started.set()
            await release_refresh.wait()
        return {
            "context": {"generation": builds},
            "events": [],
            "notifications": [],
            "sections": {},
        }

    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
    monkeypatch.setattr(main, "_build_live_snapshot", snapshot)

    first = payload(await main.live_snapshot())
    now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
    stale_request = asyncio.create_task(main.live_snapshot())
    await refresh_started.wait()
    await asyncio.sleep(0)
    returned_immediately = stale_request.done()
    release_refresh.set()
    stale = payload(await stale_request)

    assert returned_immediately is True
    assert first["context"]["generation"] == 1
    assert stale["context"]["generation"] == 1
    assert stale["freshness"]["stale"] is True
    assert stale["freshness"]["revalidating"] is True
    assert builds == 2


@pytest.mark.anyio
async def test_failed_revalidation_enters_cooldown_and_keeps_last_snapshot(monkeypatch):
    now = 100.0
    builds = 0

    async def snapshot(_sections=None):
        nonlocal builds
        builds += 1
        if builds > 1:
            raise ConnectionError("private upstream failure")
        return {
            "context": {"generation": 1},
            "events": [{"id": 1}],
            "notifications": [],
            "sections": {
                "context": "fresh",
                "events": "fresh",
                "notifications": "fresh",
            },
        }

    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
    monkeypatch.setattr(main, "_build_live_snapshot", snapshot)

    await main.live_snapshot()
    now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
    stale = payload(await main.live_snapshot())
    assert main._live_snapshot_task is not None
    await asyncio.gather(main._live_snapshot_task, return_exceptions=True)
    degraded = payload(await main.live_snapshot())

    assert stale["freshness"]["revalidating"] is True
    assert degraded["context"] == {"generation": 1}
    assert degraded["freshness"]["stale"] is True
    assert degraded["freshness"]["degraded"] is True
    assert degraded["freshness"]["last_refresh_failed"] is True
    assert degraded["freshness"]["revalidating"] is False
    assert degraded["freshness"]["retry_in_seconds"] > 0
    assert builds == 2
    assert "private" not in json.dumps(degraded)


@pytest.mark.anyio
async def test_partial_refresh_updates_fresh_sections_and_retains_failed_sections(monkeypatch):
    now = 100.0
    builds = 0

    async def snapshot(_sections=None):
        nonlocal builds
        builds += 1
        if builds == 1:
            return {
                "context": {"generation": 1},
                "events": [{"id": "last-known"}],
                "notifications": [{"id": 7}],
                "notification_pagination": {"page": 1, "total": 1, "has_more": False},
                "sections": {"context": "fresh", "events": "fresh", "notifications": "fresh"},
            }
        return {
            "context": {"generation": 2},
            "events": None,
            "notifications": [{"id": 8}],
            "notification_pagination": {"page": 1, "total": 1, "has_more": False},
            "sections": {
                "context": "fresh",
                "events": "temporarily unavailable",
                "notifications": "fresh",
            },
        }

    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
    monkeypatch.setattr(main, "_build_live_snapshot", snapshot)

    await main.live_snapshot()
    now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
    await main.live_snapshot()
    assert main._live_snapshot_task is not None
    await main._live_snapshot_task
    result = payload(await main.live_snapshot())

    assert result["context"] == {"generation": 2}
    assert result["events"] == [{"id": "last-known"}]
    assert result["notifications"] == [{"id": 8}]
    assert result["sections"] == {
        "context": "fresh",
        "events": "stale",
        "notifications": "fresh",
    }
    assert result["freshness"]["degraded"] is True
    assert result["freshness"]["revalidating"] is False


@pytest.mark.anyio
async def test_live_snapshot_reports_per_bucket_upstream_latency(monkeypatch):
    delays = {"context": 0.03, "events": 0.02, "notifications": 0.06}

    async def user():
        await asyncio.sleep(delays["context"])
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    async def events(_authenticated_user):
        await asyncio.sleep(delays["events"])
        return [{"type": "push"}]

    async def updates():
        await asyncio.sleep(delays["notifications"])
        return {"items": [], "page": 1, "total": 0, "has_more": False}

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", events)
    monkeypatch.setattr(main, "notifications", updates)

    result = payload(await main.live_snapshot())

    latency = result["freshness"]["latency_ms"]
    assert set(latency) == {"context", "events", "notifications"}
    for section, seconds in delays.items():
        assert isinstance(latency[section], int)
        assert latency[section] >= seconds * 1000


@pytest.mark.anyio
async def test_bucket_latency_measures_each_feed_individually(monkeypatch):
    """A fast feed must not inherit a slow sibling's batch wall time."""
    async def user():
        await asyncio.sleep(0.03)
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    async def fast_events(_authenticated_user):
        return [{"type": "push"}]

    async def slow_updates():
        await asyncio.sleep(0.12)
        return {"items": [], "page": 1, "total": 0, "has_more": False}

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", fast_events)
    monkeypatch.setattr(main, "notifications", slow_updates)

    result = payload(await main.live_snapshot())

    latency = result["freshness"]["latency_ms"]
    # The fast events feed finished immediately; its displayed latency must
    # reflect that feed alone, not the whole gather batch (~120 ms).
    assert latency["events"] < latency["notifications"]
    assert latency["events"] < 60
    assert latency["notifications"] >= 120


@pytest.mark.anyio
async def test_bucket_latency_survives_partial_refresh_and_failure(monkeypatch):
    """A failed bucket keeps its last known latency instead of vanishing."""
    now = 100.0
    builds = 0

    async def snapshot(_sections=None):
        nonlocal builds
        builds += 1
        if builds == 1:
            return {
                "context": {"generation": 1},
                "events": [],
                "notifications": [],
                "latency_ms": {"context": 11, "events": 12, "notifications": 13},
                "sections": {
                    "context": "fresh", "events": "fresh", "notifications": "fresh",
                },
            }
        return {
            "context": {"generation": 2},
            "events": None,
            "notifications": [],
            "latency_ms": {"context": 21, "notifications": 23},
            "sections": {
                "context": "fresh",
                "events": "temporarily unavailable",
                "notifications": "fresh",
            },
        }

    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
    monkeypatch.setattr(main, "_build_live_snapshot", snapshot)

    first = payload(await main.live_snapshot())
    assert first["freshness"]["latency_ms"] == {
        "context": 11, "events": 12, "notifications": 13,
    }

    now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
    await main.live_snapshot()
    assert main._live_snapshot_task is not None
    await main._live_snapshot_task
    result = payload(await main.live_snapshot())

    assert result["freshness"]["latency_ms"] == {
        "context": 21, "events": 12, "notifications": 23,
    }


def test_live_payload_without_latency_data_reports_no_buckets():
    value = {
        "context": {},
        "events": None,
        "notifications": None,
        "sections": {"context": "fresh"},
    }
    result = main._live_snapshot_payload(value, stale=False, revalidating=False)

    assert result["freshness"]["latency_ms"] == {}
    assert "latency_ms" not in result


@pytest.mark.anyio
async def test_notification_cooldown_does_not_stop_due_work_and_activity_refreshes(monkeypatch):
    now = 100.0
    calls = {"context": 0, "events": 0, "notifications": 0}
    event_refresh_times = []
    notification_outage = False

    async def user():
        return {"id": 1, "login": "timmy"}

    async def work():
        calls["context"] += 1
        return []

    async def empty_work():
        return []

    async def events(_authenticated_user):
        calls["events"] += 1
        event_refresh_times.append(now)
        return [{"generation": calls["events"]}]

    async def updates():
        calls["notifications"] += 1
        if notification_outage:
            raise ConnectionError("notifications unavailable")
        return []

    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", work)
    monkeypatch.setattr(main, "issues", empty_work)
    monkeypatch.setattr(main, "pull_requests", empty_work)
    monkeypatch.setattr(main, "activity_events", events)
    monkeypatch.setattr(main, "notifications", updates)

    await main.live_snapshot()
    notification_outage = True

    for elapsed in (
        main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1,
        main.LIVE_SNAPSHOT_RETRY_BASE_SECONDS,
    ):
        now += elapsed
        await main.live_snapshot()
        assert main._live_snapshot_task is not None
        await main._live_snapshot_task

    # Notification backoff is now 10 seconds. The healthy sections become due
    # one second before that cooldown expires and must refresh independently.
    now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
    response = payload(await main.live_snapshot())
    assert main._live_snapshot_task is not None
    await main._live_snapshot_task
    refreshed = payload(await main.live_snapshot())

    assert calls == {"context": 3, "events": 3, "notifications": 3}
    assert event_refresh_times == [100.0, 109.0, 123.0]
    assert response["freshness"]["sections"]["context"]["revalidating"] is True
    assert refreshed["events"] == [{"generation": 3}]
    assert refreshed["sections"]["notifications"] == "stale"
    assert refreshed["freshness"]["sections"]["notifications"]["retry_in_seconds"] == 1


@pytest.mark.anyio
async def test_retry_window_starts_one_refresh_and_success_clears_degraded_state(monkeypatch):
    now = 100.0
    builds = 0
    retry_started = asyncio.Event()
    release_retry = asyncio.Event()

    async def snapshot(_sections=None):
        nonlocal builds
        builds += 1
        if builds == 1:
            return {
                "context": {"generation": 1}, "events": [], "notifications": [],
                "sections": {"context": "fresh", "events": "fresh", "notifications": "fresh"},
            }
        if builds == 2:
            return {
                "context": None, "events": [], "notifications": [],
                "sections": {
                    "context": "temporarily unavailable",
                    "events": "fresh",
                    "notifications": "fresh",
                },
            }
        retry_started.set()
        await release_retry.wait()
        return {
            "context": {"generation": 3}, "events": [], "notifications": [],
            "sections": {"context": "fresh", "events": "fresh", "notifications": "fresh"},
        }

    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
    monkeypatch.setattr(main, "_build_live_snapshot", snapshot)

    await main.live_snapshot()
    now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
    await main.live_snapshot()
    assert main._live_snapshot_task is not None
    await main._live_snapshot_task
    now += main.LIVE_SNAPSHOT_RETRY_BASE_SECONDS

    first = asyncio.create_task(main.live_snapshot())
    second = asyncio.create_task(main.live_snapshot())
    await asyncio.wait_for(retry_started.wait(), timeout=1.0)
    assert builds == 3
    assert payload(await first)["freshness"]["revalidating"] is True
    assert payload(await second)["freshness"]["revalidating"] is True

    release_retry.set()
    refresh_task = main._live_snapshot_task
    assert refresh_task is not None
    await refresh_task
    recovered = payload(await main.live_snapshot())

    assert recovered["context"] == {"generation": 3}
    assert recovered["freshness"]["degraded"] is False
    assert recovered["freshness"]["last_refresh_failed"] is False
    assert recovered["freshness"]["retry_in_seconds"] == 0


@pytest.mark.anyio
async def test_cancelling_one_waiter_does_not_cancel_the_shared_snapshot(monkeypatch):
    started = asyncio.Event()
    release = asyncio.Event()
    snapshot_calls = 0

    async def blocked_snapshot(_sections=None):
        nonlocal snapshot_calls
        snapshot_calls += 1
        started.set()
        await release.wait()
        return {
            "context": {},
            "events": [],
            "notifications": [],
            "sections": {},
        }

    monkeypatch.setattr(main, "_build_live_snapshot", blocked_snapshot)

    disconnected = asyncio.create_task(main.live_snapshot())
    await started.wait()
    survivor = asyncio.create_task(main.live_snapshot())
    await asyncio.sleep(0)

    disconnected.cancel()
    with pytest.raises(asyncio.CancelledError):
        await disconnected

    assert main._live_snapshot_task is not None
    assert not main._live_snapshot_task.done()

    release.set()
    response = await survivor

    assert response.status_code == 200
    assert snapshot_calls == 1


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

    async def snapshot_that_exceeds_deadline(_sections=None):
        started.set()
        try:
            await asyncio.Event().wait()
        finally:
            cancelled.set()

    monkeypatch.setattr(main, "_build_live_snapshot", snapshot_that_exceeds_deadline)
    monkeypatch.setattr(main, "CONTEXT_TIMEOUT_SECONDS", 0.01)

    first = asyncio.create_task(main.live_snapshot())
    await started.wait()
    second = asyncio.create_task(main.live_snapshot())

    try:
        responses = await asyncio.gather(first, second)
        await asyncio.wait_for(cancelled.wait(), timeout=0.1)

        assert [response.status_code for response in responses] == [503, 503]
        assert [payload(response)["error"] for response in responses] == [
            "Gitea live snapshot timed out after 0.01s",
            "Gitea live snapshot timed out after 0.01s",
        ]
    finally:
        task = main._live_snapshot_task
        if task is not None and not task.done():
            task.cancel()
            with pytest.raises(asyncio.CancelledError):
                await task


@pytest.mark.anyio
async def test_cold_refresh_cooldown_remains_a_retryable_503(monkeypatch):
    now = 100.0

    async def unavailable(_sections=None):
        raise ConnectionError("private upstream failure")

    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
    monkeypatch.setattr(main, "_build_live_snapshot", unavailable)

    first = await main.live_snapshot()
    second = await main.live_snapshot()

    assert first.status_code == 503
    assert second.status_code == 503
    assert payload(second) == {
        "error": "Gitea live snapshot is temporarily unavailable"
    }


@pytest.mark.anyio
async def test_cancelled_refresh_releases_shared_lease(monkeypatch):
    started = asyncio.Event()

    async def blocked(_sections=None):
        started.set()
        await asyncio.Event().wait()

    monkeypatch.setattr(main, "_build_live_snapshot", blocked)
    request = asyncio.create_task(main.live_snapshot())
    await started.wait()
    assert main._live_snapshot_task is not None

    main._live_snapshot_task.cancel()
    await asyncio.gather(request, main._live_snapshot_task, return_exceptions=True)

    replacement = main._live_snapshot_store.try_acquire_refresh(
        {"context"}, lease_seconds=1
    )
    assert replacement is not None


@pytest.mark.anyio
async def test_shared_store_outage_returns_retryable_503(monkeypatch):
    def unavailable():
        raise sqlite3.OperationalError("private database detail")

    monkeypatch.setattr(main._live_snapshot_store, "load", unavailable)

    response = await main.live_snapshot()

    assert response.status_code == 503
    assert response.headers["retry-after"] == "1"
    assert payload(response) == {
        "error": "Gitea live snapshot state is temporarily unavailable"
    }


@pytest.mark.anyio
async def test_independent_workers_reuse_shared_live_snapshot(monkeypatch, tmp_path):
    calls = 0

    async def user():
        nonlocal calls
        calls += 1
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", lambda _user: empty())
    monkeypatch.setattr(main, "notifications", empty)
    path = tmp_path / "worker-shared.sqlite3"
    monkeypatch.setattr(main, "_live_snapshot_store", LiveSnapshotStore(path, clock=lambda: 100.0))

    first = payload(await main.live_snapshot())
    main._live_snapshot_value = None
    main._live_snapshot_created_at = None
    main._live_section_created_at = {section: None for section in main.LIVE_SNAPSHOT_SECTIONS}
    main._live_snapshot_store = LiveSnapshotStore(path, clock=lambda: 100.0)
    second = payload(await main.live_snapshot())

    assert calls == 1
    assert second["context"] == first["context"]
    assert second["revisions"] == first["revisions"]


@pytest.mark.anyio
async def test_live_snapshot_persists_reboot_stable_wall_timestamps(monkeypatch, tmp_path):
    epoch = 1_700_000_000.0

    async def user():
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: epoch, raising=False)
    monkeypatch.setattr(
        main,
        "_live_snapshot_store",
        LiveSnapshotStore(tmp_path / "wall.sqlite3", clock=lambda: main._live_snapshot_clock()),
    )
    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", lambda _user: empty())
    monkeypatch.setattr(main, "notifications", empty)

    response = await main.live_snapshot()

    assert response.status_code == 200
    assert set(main._live_snapshot_store.load().created_at.values()) == {epoch}


@pytest.mark.anyio
async def test_failed_upstream_attempt_never_overwrites_prior_successful_latency(monkeypatch, tmp_path):
    """A failed upstream-attempt duration must never overwrite the previous
    successful latency, and is never mislabeled as 'last known'. With no prior
    success the section shows 'not measured'."""
    epoch = 1000.0
    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: epoch)
    monkeypatch.setattr(
        main,
        "_live_snapshot_store",
        LiveSnapshotStore(tmp_path / "lat.sqlite3", clock=lambda: main._live_snapshot_clock()),
    )

    async def user():
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    async def failing_notifications():
        raise ConnectionError("Gitea notifications endpoint refused connection")

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", lambda _user: empty())

    async def healthy_notifications():
        return {"items": [], "page": 1, "total": 0, "has_more": False}

    monkeypatch.setattr(main, "notifications", healthy_notifications)

    # First call: all sections succeed, each records a real latency.
    result = payload(await main.live_snapshot())
    first_latency = dict(result["freshness"]["latency_ms"])
    assert set(first_latency) == {"context", "events", "notifications"}
    for value in first_latency.values():
        assert isinstance(value, int) and value >= 0

    # Second call: notifications fails upstream. The failed-attempt duration
    # must NOT overwrite the prior successful notifications latency.
    monkeypatch.setattr(main, "notifications", failing_notifications)

    # Advance the clock past freshness so a real re-fetch occurs.
    epoch += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
    await main.live_snapshot()
    # The re-fetch runs as a shared background refresh; wait for it so the
    # merged result (with prior latency preserved) is published.
    assert main._live_snapshot_task is not None
    await main._live_snapshot_task
    result2 = payload(await main.live_snapshot())
    second_latency = result2["freshness"]["latency_ms"]

    # context/events remain measured (they succeeded again); notifications
    # retains its PRIOR successful latency, not the failed-attempt duration.
    assert second_latency["notifications"] == first_latency["notifications"]
    assert second_latency["notifications"] >= 0
    # The failed section is marked stale or unavailable, not 'last known'.
    assert result2["sections"]["notifications"] in ("stale", "temporarily unavailable")
    # Ensure the failed-attempt's own timing never leaked into notifications.
    assert second_latency["events"] >= first_latency["events"]


@pytest.mark.anyio
async def test_submillisecond_upstream_measures_zero_no_fabricated_floor(monkeypatch):
    """Sub-millisecond operations measure as 0, never a fabricated 1 ms floor."""
    async def fast_user():
        return {"id": 1, "login": "timmy"}

    async def fast_empty():
        return []

    monkeypatch.setattr(main, "current_user", fast_user)
    monkeypatch.setattr(main, "repos", fast_empty)
    monkeypatch.setattr(main, "issues", fast_empty)
    monkeypatch.setattr(main, "pull_requests", fast_empty)
    monkeypatch.setattr(main, "activity_events", lambda _user: fast_empty())
    monkeypatch.setattr(main, "notifications", fast_empty)

    result = payload(await main.live_snapshot())
    latency = result["freshness"]["latency_ms"]

    for section in ("context", "events", "notifications"):
        value = latency[section]
        assert isinstance(value, int) and not isinstance(value, bool)
        assert value >= 0
        # Must never be a fabricated positive floor from sub-ms work.
        assert value != 1 or value >= 1000  # 1 ms is only legitimate at >=1ms


@pytest.mark.anyio
async def test_notifications_fetch_starts_before_auth_completes(monkeypatch):
    """Notifications is auth-independent: its fetch starts immediately on the
    same tick as the shared current_user() fetch. An 80 ms auth request and an
    80 ms Notifications request complete concurrently (~80 ms) rather than
    serially (~160 ms)."""

    import time

    notifications_started = asyncio.Event()
    auth_started = asyncio.Event()
    concurrency = {"notifications_started_before_auth_slept": False}

    async def slow_notifications():
        notifications_started.set()
        await asyncio.sleep(0.08)
        return {"items": [], "page": 1, "total": 0, "has_more": False}

    async def slow_user():
        auth_started.set()
        # Record whether notifications had already started while auth was
        # still awaiting — direct proof of concurrency, not serialization.
        if notifications_started.is_set():
            concurrency["notifications_started_before_auth_slept"] = True
        await asyncio.sleep(0.08)
        if notifications_started.is_set():
            concurrency["notifications_started_before_auth_slept"] = True
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    monkeypatch.setattr(main, "current_user", slow_user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", lambda _user: empty())
    monkeypatch.setattr(main, "notifications", slow_notifications)

    # Drive the snapshot builder directly so the measurement is deterministic
    # and not inflated by store/SQLite round-trips in the HTTP layer.
    wall_start = time.perf_counter()
    result = await main._build_live_snapshot()
    total = time.perf_counter() - wall_start

    # Both fetches were attempted.
    assert auth_started.is_set()
    assert notifications_started.is_set()
    # Notifications started while auth was still in flight (concurrent, not
    # serialized behind the shared auth fetch).
    assert concurrency["notifications_started_before_auth_slept"] is True
    # If serialized, total would be ~160 ms. Concurrent means ~80 ms.
    # The bound must sit strictly between the concurrent (~80 ms) and serial
    # (~160 ms) costs to deterministically prove concurrency.
    assert total < 0.11


def test_strict_latency_validation_backend_rejects_malformed():
    """Backend strict-type gate for latency telemetry rejects everything that
    is not a finite nonnegative bounded integer and never coerces/clips."""
    for value in [None, True, False, "120", "fast", [120], 120.9, -5, float("inf"), 99999999]:
        assert main._valid_latency_ms(value) is False, value
    assert main._valid_latency_ms(0) is True
    assert main._valid_latency_ms(1) is True
    assert main._valid_latency_ms(3600000) is True
    assert main._valid_latency_ms(3600001) is False
    assert main._measured_latency_ms(0.0) == 0
    assert main._measured_latency_ms(0.0004) == 0  # sub-ms rounds to 0, no 1ms floor
    assert main._measured_latency_ms(-1.0) is None
    assert main._measured_latency_ms(float("inf")) is None