import json
import asyncio
import os
import sqlite3
import time
from datetime import datetime, timezone
from types import SimpleNamespace

import httpx
import pytest
import requests

from src import dashboard_auth, gitea_proxy, main
from src.push_notifications import (
    PushConfiguration,
    dispatch_following_changes,
    dispatch_unread_updates,
    send_web_push,
)
from src import push_subscription_store as push_store_module
from src.push_subscription_store import PushSubscriptionStore
from src.push_endpoint_policy import UnsafePushEndpoint


@pytest.fixture(autouse=True)
def resolve_test_push_endpoints_publicly(monkeypatch):
    async def accept(endpoint):
        return endpoint

    monkeypatch.setattr(main, "validate_public_push_endpoint", accept)


def test_dispatch_lease_is_exclusive_recoverable_and_owner_fenced(tmp_path):
    path = tmp_path / "push.sqlite3"
    first = PushSubscriptionStore(path)
    second = PushSubscriptionStore(path)

    assert first.acquire_dispatch_lease("worker-a", now=100, lease_seconds=30) is True
    assert second.acquire_dispatch_lease("worker-b", now=100, lease_seconds=30) is False
    assert second.acquire_dispatch_lease("worker-b", now=131, lease_seconds=30) is True

    assert first.release_dispatch_lease("worker-a") is False
    assert second.release_dispatch_lease("worker-b") is True


def test_following_alert_preferences_are_opt_in_and_checkpoint_each_device(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for session_id in ("session-a", "session-b"):
        store.upsert(session_id, {
            "endpoint": f"https://push.example/{session_id}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })

    assert store.following_preferences("session-a") == {"enabled": False}
    store.set_following_preferences("session-a", enabled=True)

    devices = store.following_notification_devices()
    assert [(device.session_id, device.delivered_fingerprint) for device in devices] == [
        ("session-a", None)
    ]
    store.mark_following_delivered("session-a", "revision-fingerprint")
    assert store.following_notification_devices()[0].delivered_fingerprint == "revision-fingerprint"
    assert store.following_preferences("session-b") == {"enabled": False}


@pytest.mark.anyio
async def test_human_gate_dispatch_is_opt_in_private_deduplicated_and_session_bound(tmp_path):
    dispatch = getattr(__import__("src.push_notifications", fromlist=["dispatch_human_gate_changes"]), "dispatch_human_gate_changes", None)
    assert callable(dispatch), "Human Gate push dispatcher is missing"
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for session_id in ("active", "revoked", "disabled"):
        store.upsert(session_id, {
            "endpoint": f"https://push.example/{session_id}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })
    store.set_human_gate_preferences("active", enabled=True)
    store.set_human_gate_preferences("revoked", enabled=True)
    sent = []

    pending_count = 2

    async def pending_gates():
        return {"complete": True, "count": pending_count, "items": [{
            "title": "Secret launch", "artifact_url": "https://secret.example/token",
            "candidate_hash": "private-hash",
        }]}

    async def send(subscription, payload):
        sent.append((subscription["endpoint"], json.loads(payload)))

    async def statuses(session_ids):
        return {item: ("active" if item == "active" else "revoked") for item in session_ids}

    configuration = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch(
        store, configuration, pending_gates, send, session_statuses=statuses
    ) == 1
    assert sent == [("https://push.example/active", {
        "title": "2 release decisions are waiting",
        "body": "Open Human Gates to review them.",
        "route": "#/my-work/human-gates",
        "tag": "stackchain-human-gates-2",
        "human_gate_count": 2,
    })]
    assert "secret" not in json.dumps(sent).lower()
    assert "private-hash" not in json.dumps(sent).lower()
    assert await dispatch(
        store, configuration, pending_gates, send, session_statuses=statuses
    ) == 0
    pending_count = 0
    assert await dispatch(
        store, configuration, pending_gates, send, session_statuses=statuses
    ) == 0
    assert store.human_gate_notification_devices()[0].delivered_count == 0
    assert len(sent) == 1
    assert store.subscription_for_session("revoked") is None


def test_quiet_hours_hold_unread_revisions_then_mark_one_catch_up_delivery(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/session-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    store.set_quiet_hours(
        "session-a", enabled=True, start="22:00", end="07:00", timezone="UTC"
    )

    inside = datetime(2026, 8, 25, 23, 0, tzinfo=timezone.utc).timestamp()
    after = datetime(2026, 8, 26, 7, 1, tzinfo=timezone.utc).timestamp()

    assert store.quiet_hours("session-a") == {
        "enabled": True, "start": "22:00", "end": "07:00", "timezone": "UTC"
    }
    assert store.claim_unseen({42: "r1"}, now=inside) == []
    delivery = store.claim_unseen({42: "r1"}, now=after)
    assert len(delivery) == 1
    assert delivery[0].thread_revisions == ((42, "r1"),)
    assert delivery[0].catch_up is True
    store.mark_delivered("session-a", delivery[0].thread_revisions)
    assert store.claim_unseen({42: "r1"}, now=after) == []


@pytest.mark.anyio
async def test_unread_dispatch_sends_one_catch_up_digest_after_quiet_hours(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/session-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    store.set_quiet_hours(
        "session-a", enabled=True, start="22:00", end="07:00", timezone="UTC"
    )
    inside = datetime(2026, 8, 25, 23, 0, tzinfo=timezone.utc).timestamp()
    after = datetime(2026, 8, 26, 7, 1, tzinfo=timezone.utc).timestamp()
    assert store.claim_unseen({41: "r1", 42: "r1"}, now=inside) == []
    sent = []

    async def unread():
        return {"items": [
            {"id": 41, "updated_at": "r1"},
            {"id": 42, "updated_at": "r1"},
        ]}

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

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_unread_updates(store, config, unread, send, now=after) == 1
    assert sent == [{
        "title": "2 updates while alerts were paused",
        "body": "Open Updates to catch up in Stackchain.",
        "route": "#/my-work/updates",
        "tag": "stackchain-update-catch-up",
        "update_count": 2,
        "unread_count": 2,
    }]


@pytest.mark.anyio
async def test_following_dispatch_is_private_deduplicated_and_session_bound(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for session_id in ("active", "revoked", "disabled"):
        store.upsert(session_id, {
            "endpoint": f"https://push.example/{session_id}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })
    store.set_following_preferences("active", enabled=True)
    store.set_following_preferences("revoked", enabled=True)
    sent = []

    async def following():
        return {"items": [{
            "repository": "secret/project", "kind": "issue", "number": 42,
            "title": "Sensitive acquisition", "updated_at": "2026-08-23T17:00:00Z",
            "has_unseen_change": True,
        }]}

    async def send(subscription, payload):
        sent.append((subscription["endpoint"], json.loads(payload)))

    async def statuses(session_ids):
        return {session_id: ("active" if session_id == "active" else "revoked") for session_id in session_ids}

    configuration = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_following_changes(
        store, configuration, following, send, session_statuses=statuses
    ) == 1
    assert sent[0][0].endswith("/active")
    assert sent[0][1] == {
        "title": "1 watched item changed",
        "body": "Open Following to review the latest activity.",
        "route": "#/my-work/following",
        "tag": sent[0][1]["tag"],
        "following_count": 1,
    }
    assert sent[0][1]["tag"].startswith("stackchain-following-")
    assert "secret" not in json.dumps(sent[0][1]).lower()
    assert "acquisition" not in json.dumps(sent[0][1]).lower()
    assert await dispatch_following_changes(
        store, configuration, following, send, session_statuses=statuses
    ) == 0
    assert len(sent) == 1
    assert store.subscription_for_session("revoked") is None


@pytest.mark.anyio
async def test_following_changes_wait_for_quiet_hours_and_resume_as_catch_up(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/session-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    store.set_following_preferences("session-a", enabled=True)
    store.set_quiet_hours(
        "session-a", enabled=True, start="22:00", end="07:00", timezone="UTC"
    )
    inside = datetime(2026, 8, 25, 23, 0, tzinfo=timezone.utc).timestamp()
    after = datetime(2026, 8, 26, 7, 1, tzinfo=timezone.utc).timestamp()
    sent = []

    async def following():
        return {"items": [{
            "repository": "private/project", "kind": "issue", "number": 42,
            "title": "Sensitive", "updated_at": "2026-08-25T23:00:00Z",
            "has_unseen_change": True,
        }]}

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

    configuration = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_following_changes(
        store, configuration, following, send, now=inside
    ) == 0
    assert sent == []
    assert await dispatch_following_changes(
        store, configuration, following, send, now=after
    ) == 1
    assert sent[0]["title"] == "1 watched update while alerts were paused"
    assert sent[0]["route"] == "#/my-work/following"


@pytest.mark.anyio
async def test_authenticated_device_controls_following_alerts_independently(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/session-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    monkeypatch.setattr(main, "_push_subscription_store", store)

    async def management_id(_session):
        return "session-a"

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))

    result = await main.update_following_notifications(
        main.FollowingNotificationPayload(enabled=True), request
    )

    assert result == {"following_enabled": True}
    assert (await main.push_status(request))["following_enabled"] is True
    assert store.deadline_preferences("session-a")["enabled"] is False
    assert store.start_day_preferences("session-a")["enabled"] is False


@pytest.mark.anyio
async def test_authenticated_device_controls_human_gate_alerts_independently(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/session-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    monkeypatch.setattr(main, "_push_subscription_store", store)

    async def management_id(_session):
        return "session-a"

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))
    payload_type = getattr(main, "HumanGateNotificationPayload", None)
    endpoint = getattr(main, "update_human_gate_notifications", None)
    assert payload_type is not None and callable(endpoint), "Human Gate push preference API is missing"

    result = await endpoint(payload_type(enabled=True), request)

    assert result == {"human_gates_enabled": True}
    assert (await main.push_status(request))["human_gates_enabled"] is True
    assert store.following_preferences("session-a")["enabled"] is False


@pytest.mark.anyio
async def test_authenticated_device_persists_validated_quiet_hours(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/session-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    monkeypatch.setattr(main, "_push_subscription_store", store)

    async def management_id(_session):
        return "session-a"

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))
    payload = main.QuietHoursPayload(
        enabled=True, start="22:30", end="06:45", timezone="America/New_York"
    )

    result = await main.update_quiet_hours(payload, request)

    assert result == {
        "quiet_hours_enabled": True,
        "quiet_hours_start": "22:30",
        "quiet_hours_end": "06:45",
        "quiet_hours_timezone": "America/New_York",
    }
    status = await main.push_status(request)
    assert {key: status[key] for key in result} == result


def test_quiet_hours_reject_equal_boundaries_and_invalid_timezone():
    with pytest.raises(ValueError):
        main.QuietHoursPayload(enabled=True, start="22:00", end="22:00", timezone="UTC")
    with pytest.raises(ValueError):
        main.QuietHoursPayload(enabled=True, start="22:00", end="07:00", timezone="Moon/Base")


def test_subscription_store_uses_private_filesystem_permissions(tmp_path):
    state_dir = tmp_path / "push-state"
    previous_umask = os.umask(0)
    try:
        database = state_dir / "push.sqlite3"
        PushSubscriptionStore(database)
    finally:
        os.umask(previous_umask)

    assert state_dir.stat().st_mode & 0o777 == 0o700
    assert database.stat().st_mode & 0o777 == 0o600


def test_subscription_store_encrypts_device_credentials_and_reopens_with_same_key(tmp_path):
    database = tmp_path / "push.sqlite3"
    key = b"p" * 32
    subscription = {
        "endpoint": "https://push.example/private-device-canary",
        "keys": {"p256dh": "public-key-canary", "auth": "auth-secret-canary"},
    }

    PushSubscriptionStore(database, encryption_key=key).upsert("session-a", subscription)

    persisted = database.read_bytes()
    assert b"private-device-canary" not in persisted
    assert b"public-key-canary" not in persisted
    assert b"auth-secret-canary" not in persisted
    reopened = PushSubscriptionStore(database, encryption_key=key)
    assert reopened.claim_unseen({42: "r1"})[0].subscription == subscription


def test_subscription_store_migrates_legacy_plaintext_without_resetting_checkpoints(tmp_path):
    database = tmp_path / "push.sqlite3"
    subscription = {
        "endpoint": "https://push.example/legacy-device-canary",
        "keys": {"p256dh": "legacy-public-canary", "auth": "legacy-auth-canary"},
    }
    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_deliveries (
                session_id TEXT NOT NULL,
                thread_id INTEGER NOT NULL,
                revision TEXT NOT NULL DEFAULT '',
                PRIMARY KEY (session_id, thread_id)
            );
            """
        )
        connection.execute(
            "INSERT INTO push_subscriptions VALUES (?, ?, ?)",
            ("session-a", subscription["endpoint"], json.dumps(subscription)),
        )
        connection.execute(
            "INSERT INTO push_deliveries VALUES (?, ?, ?)",
            ("session-a", 42, "r1"),
        )

    store = PushSubscriptionStore(database, encryption_key=b"m" * 32)

    assert store.claim_unseen({42: "r1"}) == []
    updated = store.claim_unseen({42: "r2"})
    assert updated[0].subscription == subscription
    persisted = database.read_bytes()
    assert b"legacy-device-canary" not in persisted
    assert b"legacy-public-canary" not in persisted
    assert b"legacy-auth-canary" not in persisted


def test_push_store_factory_only_requires_its_key_when_push_is_enabled(
    tmp_path, monkeypatch
):
    monkeypatch.delenv("STACKCHAIN_PUSH_STATE_ENCRYPTION_KEY", raising=False)
    factory = getattr(push_store_module, "build_push_subscription_store", None)
    assert callable(factory), "push store factory is missing"

    disabled = factory(tmp_path / "disabled.sqlite3", push_enabled=False)
    assert disabled.is_subscribed("session-a") is False
    assert disabled.deadline_preferences("session-a")["enabled"] is False
    assert disabled.claim_unseen({42: "r1"}, now=0) == []
    assert not (tmp_path / "disabled.sqlite3").exists()
    with pytest.raises(RuntimeError, match="encryption key"):
        factory(tmp_path / "enabled.sqlite3", push_enabled=True)


def test_subscription_store_fails_closed_for_wrong_key_or_tampered_payload(tmp_path):
    database = tmp_path / "push.sqlite3"
    subscription = {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    }
    PushSubscriptionStore(database, encryption_key=b"a" * 32).upsert(
        "session-a", subscription
    )

    with pytest.raises(RuntimeError, match="could not be decrypted"):
        PushSubscriptionStore(database, encryption_key=b"b" * 32)

    with sqlite3.connect(database) as connection:
        payload = connection.execute(
            "SELECT subscription_json FROM push_subscriptions WHERE session_id = ?",
            ("session-a",),
        ).fetchone()[0]
        replacement = "A" if payload[-1] != "A" else "B"
        connection.execute(
            "UPDATE push_subscriptions SET subscription_json = ? WHERE session_id = ?",
            (payload[:-1] + replacement, "session-a"),
        )
    with pytest.raises(RuntimeError, match="could not be decrypted"):
        PushSubscriptionStore(database, encryption_key=b"a" * 32)


def test_subscription_ciphertext_cannot_be_substituted_between_sessions(tmp_path):
    database = tmp_path / "push.sqlite3"
    key = b"s" * 32
    store = PushSubscriptionStore(database, encryption_key=key)
    for session_id in ("session-a", "session-b"):
        store.upsert(
            session_id,
            {
                "endpoint": f"https://push.example/{session_id}",
                "keys": {"p256dh": f"key-{session_id}", "auth": f"auth-{session_id}"},
            },
        )
    with sqlite3.connect(database) as connection:
        payloads = connection.execute(
            "SELECT session_id, subscription_json FROM push_subscriptions ORDER BY session_id"
        ).fetchall()
        connection.executemany(
            "UPDATE push_subscriptions SET subscription_json = ? WHERE session_id = ?",
            ((payloads[1][1], payloads[0][0]), (payloads[0][1], payloads[1][0])),
        )

    with pytest.raises(RuntimeError, match="could not be decrypted"):
        PushSubscriptionStore(database, encryption_key=key)


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

    store.upsert("session-a", subscription)

    first = store.claim_unseen({41, 42})
    store.mark_delivered("session-a", first[0].thread_ids)
    second = store.claim_unseen({41, 42})

    assert [(item.session_id, item.thread_ids) for item in first] == [
        ("session-a", (41, 42))
    ]
    assert second == []


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

    store.mark_delivery_failed("session-a", "unread", "timeout", now=100)
    store.mark_delivery_failed("session-a", "unread", "provider", now=110)
    store.mark_delivery_failed("session-a", "deadline", "provider", now=120)

    assert store.delivery_health("session-a") == {
        "unread": {
            "state": "degraded",
            "consecutive_failures": 2,
            "last_attempted_at": 110,
            "last_succeeded_at": None,
            "reason": "provider",
        },
        "deadline": {
            "state": "degraded",
            "consecutive_failures": 1,
            "last_attempted_at": 120,
            "last_succeeded_at": None,
            "reason": "provider",
        },
    }

    store.mark_delivery_succeeded("session-a", "unread", now=130)

    assert store.delivery_health("session-a")["unread"] == {
        "state": "healthy",
        "consecutive_failures": 0,
        "last_attempted_at": 130,
        "last_succeeded_at": 130,
        "reason": None,
    }
    assert store.delivery_health("session-a")["deadline"]["state"] == "degraded"


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

    first = store.claim_unseen({42: "2026-08-11T12:00:00Z"})
    store.mark_delivered("session-a", first[0].thread_revisions)
    unchanged = store.claim_unseen({42: "2026-08-11T12:00:00Z"})
    updated = store.claim_unseen({42: "2026-08-11T12:05:00Z"})

    assert unchanged == []
    assert updated[0].thread_revisions == ((42, "2026-08-11T12:05:00Z"),)


def test_subscription_store_does_not_realert_for_an_older_snapshot_revision(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    store.mark_delivered("session-a", {42: "2026-08-11T12:05:00Z"})

    assert store.claim_unseen({42: "2026-08-11T12:00:00Z"}) == []


def test_complete_unread_snapshot_reconciles_delivery_and_digest_checkpoints(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    revisions = {1: "r1", 2: "r2", 3: "r3"}
    store.mark_delivered("session-a", revisions)
    store.mark_digest_pending("session-a", {3: "r3"})

    store.reconcile_unread({2})
    delivery = store.claim_unseen(revisions)[0]

    assert delivery.thread_ids == (1, 3)
    assert delivery.digest_ids == ()


@pytest.mark.anyio
async def test_dispatch_reconciles_stale_checkpoints_after_a_complete_snapshot(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    store.mark_delivered("session-a", {1: "r1"})

    async def unread():
        return {"items": [{"id": 2, "updated_at": "r2"}], "complete": True}

    async def send(_subscription, _payload):
        return None

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_unread_updates(store, config, unread, send) == 1

    remaining = store.claim_unseen({1: "r1", 2: "r2"})
    assert remaining[0].thread_ids == (1,)


@pytest.mark.anyio
async def test_dispatch_sends_nothing_and_preserves_checkpoints_for_an_incomplete_snapshot(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    store.mark_delivered("session-a", {1: "r1"})
    sent = []

    async def unread():
        return {"items": [{"id": 2, "updated_at": "r2"}], "complete": False}

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

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_unread_updates(store, config, unread, send) == 0
    assert sent == []
    assert store.claim_unseen({1: "r1"}) == []


def test_subscription_store_migrates_legacy_delivery_without_replaying_it(tmp_path):
    path = tmp_path / "push.sqlite3"
    subscription = {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    }
    with sqlite3.connect(path) 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_deliveries (
                session_id TEXT NOT NULL,
                thread_id INTEGER NOT NULL,
                PRIMARY KEY (session_id, thread_id)
            );
        """)
        connection.execute(
            "INSERT INTO push_subscriptions VALUES (?, ?, ?)",
            ("session-a", subscription["endpoint"], json.dumps(subscription)),
        )
        connection.execute("INSERT INTO push_deliveries VALUES (?, ?)", ("session-a", 42))

    store = PushSubscriptionStore(path)

    assert store.claim_unseen({42: "2026-08-11T12:00:00Z"}) == []
    updated = store.claim_unseen({42: "2026-08-11T12:05:00Z"})
    assert updated[0].thread_revisions == ((42, "2026-08-11T12:05:00Z"),)


@pytest.mark.anyio
async def test_managed_session_active_applies_configured_idle_deadline(monkeypatch):
    calls = []

    class Store:
        def managed_status(self, management_id, *, idle_timeout_seconds):
            calls.append((management_id, idle_timeout_seconds))
            return "active" if management_id == "active-device" else "idle"

    monkeypatch.setenv("STACKCHAIN_DASHBOARD_IDLE_TIMEOUT_SECONDS", "321")
    monkeypatch.setattr(dashboard_auth, "_session_store", lambda: Store())

    assert await dashboard_auth.managed_session_active("active-device") is True
    assert await dashboard_auth.managed_session_active("idle-device") is False
    assert calls == [("active-device", 321), ("idle-device", 321)]


@pytest.mark.anyio
async def test_managed_session_statuses_apply_configured_idle_deadline_in_one_call(monkeypatch):
    calls = []

    class Store:
        def managed_statuses(self, management_ids, *, idle_timeout_seconds):
            calls.append((list(management_ids), idle_timeout_seconds))
            return {management_id: "active" for management_id in set(management_ids)}

    monkeypatch.setenv("STACKCHAIN_DASHBOARD_IDLE_TIMEOUT_SECONDS", "321")
    monkeypatch.setattr(dashboard_auth, "_session_store", lambda: Store())

    assert await dashboard_auth.managed_session_statuses(
        ["device-b", "device-a", "device-b"]
    ) == {"device-a": "active", "device-b": "active"}
    assert calls == [(["device-b", "device-a", "device-b"], 321)]


@pytest.mark.anyio
async def test_managed_session_statuses_bind_batch_to_expected_principal(monkeypatch):
    calls = []

    class Store:
        def managed_statuses(
            self, management_ids, *, idle_timeout_seconds, expected_principal_id
        ):
            calls.append(
                (list(management_ids), idle_timeout_seconds, expected_principal_id)
            )
            return {management_id: "active" for management_id in set(management_ids)}

    monkeypatch.setenv("STACKCHAIN_DASHBOARD_IDLE_TIMEOUT_SECONDS", "321")
    monkeypatch.setattr(dashboard_auth, "_session_store", lambda: Store())

    assert await dashboard_auth.managed_session_statuses(
        ["device-b", "device-a"], expected_principal_id=42
    ) == {"device-a": "active", "device-b": "active"}
    assert calls == [(["device-b", "device-a"], 321, 42)]


@pytest.mark.anyio
async def test_push_session_statuses_use_current_upstream_identity(monkeypatch):
    calls = []

    async def current_user():
        return {"id": 42, "login": "timmy"}

    async def statuses(management_ids, *, expected_principal_id):
        calls.append((list(management_ids), expected_principal_id))
        return {management_id: "active" for management_id in management_ids}

    monkeypatch.setattr(main.gitea_proxy, "current_user", current_user)
    monkeypatch.setattr(main.dashboard_auth, "managed_session_statuses", statuses)

    assert await main._identity_bound_push_session_statuses(["phone-device"]) == {
        "phone-device": "active"
    }
    assert calls == [(["phone-device"], 42)]


@pytest.mark.anyio
async def test_push_session_statuses_fail_closed_when_identity_is_invalid(monkeypatch):
    async def current_user():
        return {"id": "not-an-integer", "login": "timmy"}

    monkeypatch.setattr(main.gitea_proxy, "current_user", current_user)

    with pytest.raises(ValueError, match="identity"):
        await main._identity_bound_push_session_statuses(["phone-device"])


@pytest.mark.anyio
async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch):
    captured = {}
    hold_other_channels = asyncio.Event()

    async def no_wait(_seconds):
        return None

    async def stop_after_capture(*args, **kwargs):
        captured["unread"] = args[2]
        captured.update(kwargs)
        raise asyncio.CancelledError

    async def hold_dispatch(*_args, **_kwargs):
        await hold_other_channels.wait()

    monkeypatch.setenv("STACKCHAIN_PUSH_MAX_CONCURRENCY", "3")
    monkeypatch.setenv("STACKCHAIN_PUSH_MAX_INDIVIDUAL_NOTIFICATIONS", "4")
    monkeypatch.setattr(main.asyncio, "sleep", no_wait)
    monkeypatch.setattr(main, "dispatch_unread_updates", stop_after_capture)
    monkeypatch.setattr(main, "dispatch_following_changes", hold_dispatch)
    monkeypatch.setattr(main, "dispatch_human_gate_changes", hold_dispatch)
    monkeypatch.setattr(main, "dispatch_deadline_reminders", hold_dispatch)
    monkeypatch.setattr(main, "dispatch_start_day_reminders", hold_dispatch)

    with pytest.raises(asyncio.CancelledError):
        await main._push_poll_loop()

    assert captured["unread"] is main.gitea_proxy.unread_notification_snapshot
    assert captured["session_statuses"] is main._identity_bound_push_session_statuses
    assert captured["max_concurrency"] == 3
    assert captured["max_individual_notifications"] == 4


@pytest.mark.anyio
async def test_push_poll_still_dispatches_deadlines_when_unread_dispatch_fails(monkeypatch):
    calls = []

    sleeps = 0
    first_tick = asyncio.Event()

    async def no_wait(_seconds):
        nonlocal sleeps
        sleeps += 1
        if sleeps <= 5:
            if sleeps == 5:
                first_tick.set()
            await first_tick.wait()
        else:
            raise asyncio.CancelledError

    async def fail_unread(*_args, **_kwargs):
        calls.append("unread")
        raise RuntimeError("unread unavailable")

    async def dispatch_deadlines(*_args, **_kwargs):
        calls.append("deadline")

    async def dispatch_start_day(*_args, **_kwargs):
        calls.append("start-day")

    async def dispatch_following(*_args, **_kwargs):
        calls.append("following")

    async def dispatch_human_gates(*_args, **_kwargs):
        calls.append("human-gates")

    monkeypatch.setattr(main.asyncio, "sleep", no_wait)
    monkeypatch.setattr(main, "dispatch_unread_updates", fail_unread)
    monkeypatch.setattr(main, "dispatch_following_changes", dispatch_following)
    monkeypatch.setattr(main, "dispatch_human_gate_changes", dispatch_human_gates)
    monkeypatch.setattr(main, "dispatch_deadline_reminders", dispatch_deadlines)
    monkeypatch.setattr(main, "dispatch_start_day_reminders", dispatch_start_day)

    with pytest.raises(asyncio.CancelledError):
        await main._push_poll_loop()

    assert sorted(calls) == ["deadline", "following", "human-gates", "start-day", "unread"]


@pytest.mark.anyio
async def test_push_poll_deadlines_continue_while_unread_dispatch_is_blocked(monkeypatch):
    unread_started = asyncio.Event()
    release_unread = asyncio.Event()
    deadline_started = asyncio.Event()

    async def no_wait(_seconds):
        return None

    async def blocked_unread(*_args, **_kwargs):
        unread_started.set()
        await release_unread.wait()

    async def dispatch_deadlines(*_args, **_kwargs):
        deadline_started.set()
        await release_unread.wait()

    async def dispatch_start_day(*_args, **_kwargs):
        await release_unread.wait()

    monkeypatch.setattr(main.asyncio, "sleep", no_wait)
    monkeypatch.setattr(main, "dispatch_unread_updates", blocked_unread)
    monkeypatch.setattr(main, "dispatch_following_changes", dispatch_start_day)
    monkeypatch.setattr(main, "dispatch_human_gate_changes", dispatch_start_day)
    monkeypatch.setattr(main, "dispatch_deadline_reminders", dispatch_deadlines)
    monkeypatch.setattr(main, "dispatch_start_day_reminders", dispatch_start_day)

    poll = asyncio.create_task(main._push_poll_loop())
    await asyncio.wait_for(unread_started.wait(), timeout=0.5)
    try:
        await asyncio.wait_for(deadline_started.wait(), timeout=0.5)
    finally:
        poll.cancel()
        with pytest.raises(asyncio.CancelledError):
            await poll


@pytest.mark.anyio
async def test_push_poll_uses_a_lower_independent_deadline_cadence(monkeypatch):
    intervals = []

    async def capture_channel(_dispatch, *, interval):
        intervals.append(interval)

    monkeypatch.setenv("STACKCHAIN_PUSH_POLL_SECONDS", "30")
    monkeypatch.setenv("STACKCHAIN_DEADLINE_POLL_SECONDS", "600")
    monkeypatch.setattr(main, "_push_channel_loop", capture_channel)

    await main._push_poll_loop()

    assert sorted(intervals) == [30.0, 30.0, 30.0, 600.0, 600.0]


@pytest.mark.anyio
async def test_push_channel_loop_skips_missed_ticks_without_overlapping(monkeypatch):
    starts = []
    now = 0.0

    class Loop:
        def time(self):
            return now

    async def record_dispatch():
        nonlocal now
        starts.append("dispatch")
        now = 16.0 if len(starts) == 1 else now
        if len(starts) == 2:
            raise asyncio.CancelledError

    sleeps = []

    async def capture_sleep(seconds):
        nonlocal now
        sleeps.append(seconds)
        now += seconds

    monkeypatch.setattr(main.asyncio, "get_running_loop", lambda: Loop())
    monkeypatch.setattr(main.asyncio, "sleep", capture_sleep)

    with pytest.raises(asyncio.CancelledError):
        await main._push_channel_loop(record_dispatch, interval=5.0)

    assert starts == ["dispatch", "dispatch"]
    assert sleeps == [5.0, 4.0]


def test_dispatch_leases_are_isolated_by_channel(tmp_path):
    path = tmp_path / "push.sqlite3"
    first = PushSubscriptionStore(path)
    second = PushSubscriptionStore(path)

    assert first.acquire_dispatch_lease(
        "unread-owner", channel="unread", now=100, lease_seconds=30
    ) is True
    assert second.acquire_dispatch_lease(
        "deadline-owner", channel="deadline", now=100, lease_seconds=30
    ) is True
    assert second.acquire_dispatch_lease(
        "other-unread-owner", channel="unread", now=100, lease_seconds=30
    ) is False


def test_replacing_subscription_resets_delivery_cursor_and_revocation_removes_device(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    original = {
        "endpoint": "https://push.example/original",
        "keys": {"p256dh": "old-key", "auth": "old-auth"},
    }
    replacement = {
        "endpoint": "https://push.example/replacement",
        "keys": {"p256dh": "new-key", "auth": "new-auth"},
    }
    store.upsert("session-a", original)
    store.claim_unseen({9})

    store.upsert("session-a", replacement)
    delivery = store.claim_unseen({9})
    store.delete_session("session-a")

    assert delivery[0].subscription == replacement
    assert delivery[0].thread_ids == (9,)
    assert store.claim_unseen({10}) == []


@pytest.mark.anyio
async def test_dispatch_sends_one_privacy_safe_deep_link_per_new_thread(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    sent = []

    async def unread():
        return {"items": [
            {"id": 42, "repository": "private/repo", "title": "Secret title"},
        ]}

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

    config = PushConfiguration("public-vapid", "private-vapid", "mailto:ops@example.com")
    assert await dispatch_unread_updates(store, config, unread, send) == 1
    assert await dispatch_unread_updates(store, config, unread, send) == 0
    assert sent[0][1] == {
        "title": "New work update",
        "body": "Tap to review it in Stackchain.",
        "route": "#/my-work/update/42",
        "tag": "stackchain-update-42",
        "notification_id": 42,
        "unread_count": 1,
    }
    assert "private/repo" not in json.dumps(sent)
    assert "Secret title" not in json.dumps(sent)


@pytest.mark.anyio
async def test_dispatch_realerts_when_the_same_thread_has_a_newer_revision(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    revision = "2026-08-11T12:00:00Z"
    sent = []

    async def unread():
        return {"items": [{
            "id": 42,
            "updated_at": revision,
            "repository": "private/repo",
            "title": "Secret follow-up",
        }]}

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

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_unread_updates(store, config, unread, send) == 1
    assert await dispatch_unread_updates(store, config, unread, send) == 0
    revision = "2026-08-11T12:05:00Z"
    assert await dispatch_unread_updates(store, config, unread, send) == 1
    assert [payload["notification_id"] for payload in sent] == [42, 42]
    assert "updated_at" not in json.dumps(sent)
    assert "Secret follow-up" not in json.dumps(sent)


@pytest.mark.anyio
async def test_update_burst_sends_bounded_individual_pushes_and_one_private_digest(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    sent = []

    async def unread():
        return {"items": [
            {
                "id": notification_id,
                "repository": "private/repo",
                "title": f"Secret update {notification_id}",
            }
            for notification_id in range(41, 46)
        ]}

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

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    delivered = await dispatch_unread_updates(
        store,
        config,
        unread,
        send,
        max_individual_notifications=2,
    )

    assert delivered == 3
    assert [payload["tag"] for payload in sent] == [
        "stackchain-update-41",
        "stackchain-update-42",
        "stackchain-update-digest",
    ]
    assert sent[-1] == {
        "title": "3 new work updates",
        "body": "Tap to review them in Stackchain.",
        "route": "#/my-work/updates",
        "tag": "stackchain-update-digest",
        "update_count": 3,
        "unread_count": 5,
    }
    assert [payload["unread_count"] for payload in sent] == [5, 5, 5]
    assert "private/repo" not in json.dumps(sent)
    assert "Secret update" not in json.dumps(sent)
    assert await dispatch_unread_updates(
        store,
        config,
        unread,
        send,
        max_individual_notifications=2,
    ) == 0


@pytest.mark.anyio
async def test_failed_digest_retries_only_overflow_after_individual_checkpoints(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    attempts = []

    async def unread():
        return {"items": [{"id": notification_id} for notification_id in range(51, 56)]}

    async def fail_digest(_subscription, payload):
        decoded = json.loads(payload)
        attempts.append(decoded["tag"])
        if decoded["tag"] == "stackchain-update-digest":
            raise RuntimeError("push provider unavailable")

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_unread_updates(
        store,
        config,
        unread,
        fail_digest,
        max_individual_notifications=2,
    ) == 2
    assert store.claim_unseen(range(51, 56))[0].thread_ids == (53, 54, 55)

    async def succeed(_subscription, payload):
        attempts.append(json.loads(payload)["tag"])

    assert await dispatch_unread_updates(
        store,
        config,
        unread,
        succeed,
        max_individual_notifications=2,
    ) == 1
    assert attempts == [
        "stackchain-update-51",
        "stackchain-update-52",
        "stackchain-update-digest",
        "stackchain-update-digest",
    ]
    assert store.claim_unseen(range(51, 56)) == []


@pytest.mark.anyio
async def test_production_notification_page_dispatches_one_unread_push(tmp_path):
    def upstream(_request):
        return httpx.Response(200, json=[{
            "id": 42,
            "unread": True,
            "repository": {"full_name": "private/repo"},
            "subject": {
                "title": "Secret title",
                "type": "Issue",
                "html_url": "https://forge.example/private/repo/issues/7",
            },
        }])

    gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
    try:
        page = await gitea_proxy.notification_page(1)
    finally:
        await gitea_proxy.stop_client()

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

    async def unread():
        return page

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

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_unread_updates(store, config, unread, send) == 1
    assert sent == [{
        "title": "New work update",
        "body": "Tap to review it in Stackchain.",
        "route": "#/my-work/update/42",
        "tag": "stackchain-update-42",
        "notification_id": 42,
        "unread_count": 1,
    }]
    assert "private/repo" not in json.dumps(sent)
    assert "Secret title" not in json.dumps(sent)


@pytest.mark.anyio
async def test_concurrent_workers_do_not_dispatch_the_same_update(tmp_path):
    path = tmp_path / "push.sqlite3"
    first = PushSubscriptionStore(path)
    second = PushSubscriptionStore(path)
    first.upsert("session-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    sending = asyncio.Event()
    finish = asyncio.Event()
    sent = []

    async def unread():
        return {"items": [{"id": 42}]}

    async def send(_subscription, payload):
        sent.append(json.loads(payload)["tag"])
        sending.set()
        await finish.wait()

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    active = asyncio.create_task(dispatch_unread_updates(first, config, unread, send))
    await sending.wait()
    competing = await asyncio.wait_for(
        dispatch_unread_updates(second, config, unread, send), timeout=0.1
    )
    finish.set()

    assert competing == 0
    assert await active == 1
    assert sent == ["stackchain-update-42"]


@pytest.mark.anyio
async def test_dispatch_removes_an_expired_push_endpoint(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/expired",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })

    class Expired(Exception):
        response = SimpleNamespace(status_code=410)

    async def unread():
        return {"items": [{"id": 8}]}

    async def send(_subscription, _payload):
        raise Expired()

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_unread_updates(store, config, unread, send) == 0
    assert store.is_subscribed("session-a") is False


@pytest.mark.anyio
async def test_dispatch_removes_an_endpoint_that_rebinds_private_without_contacting_it(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("unsafe-device", {
        "endpoint": "https://push.example/rebound",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    contacted = []

    async def unread():
        return {"items": [{"id": 8}]}

    async def reject_rebound(_endpoint):
        raise UnsafePushEndpoint("Endpoint must resolve to a public Web Push service")

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

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_unread_updates(
        store, config, unread, send, endpoint_validator=reject_rebound
    ) == 0
    assert contacted == []
    assert store.is_subscribed("unsafe-device") is False


@pytest.mark.anyio
async def test_web_push_transport_dials_the_address_validated_for_the_original_tls_host():
    endpoint = "https://push.example/device-a"
    resolved = SimpleNamespace(
        endpoint=endpoint,
        hostname="push.example",
        port=443,
        addresses=("93.184.216.34",),
    )
    resolutions = []
    connection = {}

    async def resolve_once(value):
        resolutions.append(value)
        return resolved

    def inspect_transport(**kwargs):
        session = kwargs["requests_session"]
        assert session.trust_env is False
        request = session.prepare_request(requests.Request("POST", endpoint))
        adapter = session.get_adapter(endpoint)
        adapter.add_headers(request)
        host, tls = adapter.build_connection_pool_key_attributes(request, True)
        connection.update(host=host, tls=tls, headers=dict(request.headers))

    await send_web_push(
        {"endpoint": endpoint, "keys": {"p256dh": "key", "auth": "secret"}},
        "{}",
        PushConfiguration("public", "private", "mailto:ops@example.com"),
        endpoint_resolver=resolve_once,
        webpush_sender=inspect_transport,
    )

    assert resolutions == [endpoint]
    assert connection["host"] == {
        "scheme": "https",
        "host": "93.184.216.34",
        "port": 443,
    }
    assert connection["tls"]["assert_hostname"] == "push.example"
    assert connection["tls"]["server_hostname"] == "push.example"
    assert connection["headers"]["Host"] == "push.example"


@pytest.mark.anyio
async def test_connect_time_rebinding_rejection_removes_only_the_unsafe_device(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for session_id in ("unsafe-device", "healthy-device"):
        store.upsert(session_id, {
            "endpoint": f"https://push.example/{session_id}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })
    sent = []

    async def unread():
        return {"items": [{"id": 8}]}

    async def send(subscription, _payload):
        if subscription["endpoint"].endswith("unsafe-device"):
            raise UnsafePushEndpoint("Endpoint rebound before connection")
        sent.append(subscription["endpoint"])

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_unread_updates(store, config, unread, send) == 1
    assert store.is_subscribed("unsafe-device") is False
    assert store.is_subscribed("healthy-device") is True
    assert sent == ["https://push.example/healthy-device"]


@pytest.mark.anyio
async def test_transient_endpoint_failure_does_not_block_healthy_devices(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for session_id in ("session-a", "session-b"):
        store.upsert(session_id, {
            "endpoint": f"https://push.example/{session_id}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })
    sent = []

    async def unread():
        return {"items": [{"id": 8}]}

    async def send(subscription, _payload):
        if subscription["endpoint"].endswith("session-a"):
            raise RuntimeError("provider unavailable")
        sent.append(subscription["endpoint"])

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_unread_updates(store, config, unread, send) == 1
    assert sent == ["https://push.example/session-b"]
    assert store.claim_unseen({8})[0].session_id == "session-a"


@pytest.mark.anyio
async def test_unread_dispatch_records_degraded_health_then_recovers(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    revision = 0

    async def unread():
        nonlocal revision
        revision += 1
        return {"items": [{"id": 42, "updated_at": f"r{revision}"}], "complete": True}

    async def timeout_send(_subscription, _payload):
        raise asyncio.TimeoutError

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    for _ in range(3):
        assert await dispatch_unread_updates(store, config, unread, timeout_send) == 0

    health = store.delivery_health("session-a")["unread"]
    assert health["state"] == "degraded"
    assert health["consecutive_failures"] == 3
    assert health["reason"] == "timeout"

    async def succeed(_subscription, _payload):
        return None

    assert await dispatch_unread_updates(store, config, unread, succeed) == 1
    recovered = store.delivery_health("session-a")["unread"]
    assert recovered["state"] == "healthy"
    assert recovered["consecutive_failures"] == 0
    assert recovered["last_succeeded_at"] is not None


@pytest.mark.anyio
async def test_unexpected_device_failure_waits_for_siblings_and_retries_only_that_device(tmp_path):
    path = tmp_path / "push.sqlite3"
    store = PushSubscriptionStore(path)
    competitor = PushSubscriptionStore(path)
    for session_id in ("failing-device", "healthy-device"):
        store.upsert(session_id, {
            "endpoint": f"https://push.example/{session_id}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })
    healthy_started = asyncio.Event()
    release_healthy = asyncio.Event()
    sent = []

    async def unread():
        return {"items": [{"id": 8, "updated_at": "r1"}], "complete": True}

    async def validate(endpoint):
        if endpoint.endswith("failing-device"):
            raise TimeoutError("endpoint validation unavailable")
        return endpoint

    async def send(subscription, _payload):
        sent.append(subscription["endpoint"])
        healthy_started.set()
        await release_healthy.wait()

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    dispatch = asyncio.create_task(dispatch_unread_updates(
        store,
        config,
        unread,
        send,
        endpoint_validator=validate,
    ))
    await asyncio.wait_for(healthy_started.wait(), timeout=0.2)
    await asyncio.sleep(0)

    assert dispatch.done() is False
    assert competitor.acquire_dispatch_lease(
        "competing-worker", now=time.time(), lease_seconds=60
    ) is False

    release_healthy.set()
    assert await dispatch == 1
    remaining = store.claim_unseen({8: "r1"})
    assert [(item.session_id, item.thread_ids) for item in remaining] == [
        ("failing-device", (8,))
    ]
    assert store.delivery_health("failing-device")["unread"]["reason"] == "timeout"

    retried = []

    async def validate_recovered(endpoint):
        return endpoint

    async def send_retry(subscription, _payload):
        retried.append(subscription["endpoint"])

    assert await dispatch_unread_updates(
        store,
        config,
        unread,
        send_retry,
        endpoint_validator=validate_recovered,
    ) == 1
    assert retried == ["https://push.example/failing-device"]
    assert sent == ["https://push.example/healthy-device"]


@pytest.mark.anyio
async def test_transient_failure_stops_that_device_until_the_next_poll(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for session_id in ("failing-device", "healthy-device"):
        store.upsert(session_id, {
            "endpoint": f"https://push.example/{session_id}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })
    attempts = []

    async def unread():
        return {"items": [
            {"id": 8},
            {"id": 9},
        ]}

    async def send(subscription, payload):
        endpoint = subscription["endpoint"]
        thread_id = int(json.loads(payload)["route"].rsplit("/", 1)[1])
        attempts.append((endpoint, thread_id))
        if endpoint.endswith("failing-device"):
            raise RuntimeError("provider unavailable")

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_unread_updates(store, config, unread, send) == 2
    assert [item for item in attempts if item[0].endswith("failing-device")] == [
        ("https://push.example/failing-device", 8)
    ]
    remaining = {item.session_id: item.thread_ids for item in store.claim_unseen({8, 9})}
    assert remaining == {"failing-device": (8, 9)}


@pytest.mark.anyio
async def test_timed_out_endpoint_does_not_stall_push_fanout(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for session_id in ("session-a", "session-b"):
        store.upsert(session_id, {
            "endpoint": f"https://push.example/{session_id}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })
    sent = []

    async def unread():
        return {"items": [{"id": 13}]}

    async def send(subscription, _payload):
        if subscription["endpoint"].endswith("session-a"):
            await asyncio.Event().wait()
        sent.append(subscription["endpoint"])

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    delivered = await asyncio.wait_for(
        dispatch_unread_updates(
            store, config, unread, send, send_timeout_seconds=0.01
        ),
        timeout=0.2,
    )

    assert delivered == 1
    assert sent == ["https://push.example/session-b"]


@pytest.mark.anyio
async def test_dispatches_devices_concurrently_with_a_strict_bound(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for index in range(4):
        store.upsert(f"session-{index}", {
            "endpoint": f"https://push.example/session-{index}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })
    active = 0
    peak = 0
    bound_reached = asyncio.Event()
    release = asyncio.Event()

    async def unread():
        return {"items": [{"id": 13}]}

    async def send(_subscription, _payload):
        nonlocal active, peak
        active += 1
        peak = max(peak, active)
        if active == 2:
            bound_reached.set()
        await release.wait()
        active -= 1

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    dispatch = asyncio.create_task(dispatch_unread_updates(
        store, config, unread, send, max_concurrency=2
    ))
    await asyncio.wait_for(bound_reached.wait(), timeout=0.2)

    assert peak == 2
    assert dispatch.done() is False

    release.set()
    assert await dispatch == 4
    assert peak == 2


@pytest.mark.anyio
async def test_dispatch_removes_inactive_sessions_without_blocking_active_devices(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for session_id in ("expired-device", "active-device"):
        store.upsert(session_id, {
            "endpoint": f"https://push.example/{session_id}",
            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
        })
    sent = []

    async def unread():
        return {"items": [{"id": 21}]}

    authorization_calls = []

    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"])

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_unread_updates(
        store, config, unread, send, session_statuses=session_statuses
    ) == 1
    assert store.is_subscribed("expired-device") is False
    assert store.is_subscribed("active-device") is True
    assert sent == ["https://push.example/active-device"]
    assert authorization_calls == [["active-device", "expired-device"]]


@pytest.mark.anyio
async def test_dispatch_fails_closed_and_retains_subscriptions_when_session_registry_fails(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("active-device", {
        "endpoint": "https://push.example/active-device",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    sent = []

    async def unread():
        return {"items": [{"id": 22}]}

    async def registry_unavailable(_management_ids):
        raise RuntimeError("session registry unavailable")

    async def send(subscription, _payload):
        sent.append(subscription)

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_unread_updates(
        store, config, unread, send, session_statuses=registry_unavailable
    ) == 0
    assert store.is_subscribed("active-device") is True
    assert sent == []


@pytest.mark.anyio
async def test_authenticated_device_can_subscribe_report_status_and_unsubscribe(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    monkeypatch.setattr(main, "_push_subscription_store", store)
    monkeypatch.setattr(
        main,
        "_push_configuration",
        lambda: PushConfiguration("public-vapid", "private-vapid", "mailto:ops@example.com"),
    )
    async def management_id(_session):
        return "session-a"

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)

    async def no_unread(**_kwargs):
        return {"items": [], "complete": True}

    monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", no_unread)
    request = SimpleNamespace(state=SimpleNamespace(
        dashboard_session=SimpleNamespace(session_id="session-a")
    ))
    payload = main.PushSubscriptionPayload(
        endpoint="https://push.example/device-a",
        keys={"p256dh": "public-key", "auth": "auth-secret"},
    )

    assert await main.push_status(request) == {
        "available": True,
        "subscribed": False,
        "public_key": "public-vapid",
        "deadline_enabled": False,
        "timezone": "UTC",
        "reminder_hour": 9,
        "reminder_days": 2,
        "snoozed_until": None,
        "start_day_enabled": False,
        "start_day_timezone": "UTC",
        "start_day_reminder_hour": 9,
        "following_enabled": False,
        "human_gates_enabled": False,
        "quiet_hours_enabled": False,
        "quiet_hours_start": "22:00",
        "quiet_hours_end": "07:00",
        "quiet_hours_timezone": "UTC",
        "delivery_health": {},
    }
    assert await main.subscribe_push(payload, request) == {"subscribed": True}
    assert (await main.push_status(request))["subscribed"] is True
    assert await main.unsubscribe_push(request) == {"subscribed": False}
    assert (await main.push_status(request))["subscribed"] is False


@pytest.mark.anyio
async def test_authenticated_device_can_send_a_real_test_notification(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    subscription = {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    }
    store.upsert("session-a", subscription)
    monkeypatch.setattr(main, "_push_subscription_store", store)
    configuration = PushConfiguration("public", "private", "mailto:ops@example.com")
    monkeypatch.setattr(main, "_push_configuration", lambda: configuration)

    async def management_id(_session):
        return "session-a"

    sent = []

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

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
    monkeypatch.setattr(main, "send_web_push", send, raising=False)
    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))

    assert await main.test_push_notification(request) == {
        "delivery_state": "healthy",
        "delivered": True,
    }
    assert sent == [(
        subscription,
        {
            "title": "Stackchain notifications are working",
            "body": "This device can receive private work alerts.",
            "route": "#/device-setup",
            "tag": "stackchain-push-test",
        },
        configuration,
    )]
    assert store.delivery_health("session-a")["unread"]["state"] == "healthy"


@pytest.mark.anyio
async def test_test_notification_reports_provider_failure_as_degraded(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    monkeypatch.setattr(main, "_push_subscription_store", store)
    monkeypatch.setattr(
        main, "_push_configuration",
        lambda: PushConfiguration("public", "private", "mailto:ops@example.com"),
    )

    async def management_id(_session):
        return "session-a"

    async def fail(*_args):
        raise asyncio.TimeoutError

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
    monkeypatch.setattr(main, "send_web_push", fail)
    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))

    with pytest.raises(main.HTTPException) as raised:
        await main.test_push_notification(request)

    assert raised.value.status_code == 503
    assert raised.value.detail == "Test notification delivery failed"
    assert store.is_subscribed("session-a") is True
    assert store.delivery_health("session-a")["unread"]["reason"] == "timeout"


@pytest.mark.anyio
async def test_authenticated_device_can_enable_deadline_reminders(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    monkeypatch.setattr(main, "_push_subscription_store", store)

    async def management_id(_session):
        return "session-a"

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))
    payload = main.DeadlineReminderPayload(
        enabled=True, timezone="America/New_York", reminder_hour=9, reminder_days=7
    )

    assert await main.update_deadline_reminders(payload, request) == {
        "deadline_enabled": True,
        "timezone": "America/New_York",
        "reminder_hour": 9,
        "reminder_days": 7,
    }
    assert store.deadline_preferences("session-a")["enabled"] is True
    assert store.deadline_preferences("session-a")["reminder_days"] == 7


@pytest.mark.anyio
async def test_authenticated_device_can_snooze_its_enabled_deadline_reminder(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for session_id in ("session-a", "session-b"):
        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
        )
    monkeypatch.setattr(main, "_push_subscription_store", store)
    monkeypatch.setattr(main.time, "time", lambda: 1_765_000_000)

    async def management_id(_session):
        return "session-a"

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))

    assert await main.snooze_deadline_reminder(request) == {
        "snoozed": True,
        "snoozed_until": 1_765_003_600,
    }
    devices = {device.session_id: device for device in store.deadline_reminder_devices()}
    assert devices["session-a"].snoozed_until == 1_765_003_600
    assert devices["session-b"].snoozed_until is None


@pytest.mark.anyio
async def test_push_status_exposes_only_the_current_devices_active_deadline_snooze(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for session_id in ("session-a", "session-b"):
        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
        )
    store.snooze_deadline_reminder("session-a", now=1_765_000_000)
    store.snooze_deadline_reminder("session-b", now=1_765_001_000)
    monkeypatch.setattr(main, "_push_subscription_store", store)
    monkeypatch.setattr(main.time, "time", lambda: 1_765_000_100)

    async def management_id(_session):
        return "session-a"

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))

    status = await main.push_status(request)

    assert status["snoozed_until"] == 1_765_003_600
    assert store.deadline_preferences("session-a", now=1_765_004_000)["snoozed_until"] is None


@pytest.mark.anyio
async def test_authenticated_device_can_resume_only_its_own_deadline_reminders(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for session_id in ("session-a", "session-b"):
        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
        )
        store.snooze_deadline_reminder(session_id, now=1_765_000_000)
    monkeypatch.setattr(main, "_push_subscription_store", store)

    async def management_id(_session):
        return "session-a"

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))

    assert await main.resume_deadline_reminders(request) == {"snoozed": False}
    devices = {device.session_id: device for device in store.deadline_reminder_devices()}
    assert devices["session-a"].snoozed_until is None
    assert devices["session-b"].snoozed_until == 1_765_003_600


@pytest.mark.anyio
async def test_subscription_rejects_an_unsafe_endpoint_before_persistence(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    monkeypatch.setattr(main, "_push_subscription_store", store)
    monkeypatch.setattr(
        main, "_push_configuration",
        lambda: PushConfiguration("public", "private", "mailto:ops@example.com"),
    )

    async def management_id(_session):
        return "unsafe-device"

    async def reject_private(_endpoint):
        raise UnsafePushEndpoint("Endpoint must resolve to a public Web Push service")

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
    monkeypatch.setattr(main, "validate_public_push_endpoint", reject_private)
    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))
    payload = main.PushSubscriptionPayload(
        endpoint="https://push.example/device-a",
        keys={"p256dh": "public-key", "auth": "auth-secret"},
    )

    with pytest.raises(main.HTTPException) as rejected:
        await main.subscribe_push(payload, request)

    assert rejected.value.status_code == 422
    assert rejected.value.detail == "Endpoint must resolve to a public Web Push service"
    assert store.is_subscribed("unsafe-device") is False


@pytest.mark.anyio
async def test_subscription_baselines_every_revision_from_the_complete_snapshot(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    monkeypatch.setattr(main, "_push_subscription_store", store)
    monkeypatch.setattr(
        main,
        "_push_configuration",
        lambda: PushConfiguration("public", "private", "mailto:ops@example.com"),
    )

    async def management_id(_session):
        return "device-a"

    revisions = {
        thread_id: f"2026-08-11T12:{thread_id % 60:02d}:00Z"
        for thread_id in range(1, 126)
    }

    async def complete_snapshot(**_kwargs):
        return {
            "items": [
                {"id": thread_id, "updated_at": revision}
                for thread_id, revision in revisions.items()
            ],
            "total": 125,
            "complete": True,
        }

    async def first_page_must_not_be_used():
        raise AssertionError("subscription used the first-page inbox endpoint")

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
    monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", complete_snapshot)
    monkeypatch.setattr(main, "notifications", first_page_must_not_be_used)
    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))
    payload = main.PushSubscriptionPayload(
        endpoint="https://push.example/device-a",
        keys={"p256dh": "public-key", "auth": "auth-secret"},
    )

    assert await main.subscribe_push(payload, request) == {"subscribed": True}
    assert store.claim_unseen(revisions) == []


@pytest.mark.anyio
async def test_subscription_baselines_production_notification_ids(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    monkeypatch.setattr(main, "_push_subscription_store", store)
    monkeypatch.setattr(
        main,
        "_push_configuration",
        lambda: PushConfiguration("public", "private", "mailto:ops@example.com"),
    )

    async def management_id(_session):
        return "device-a"

    async def current_unread(**_kwargs):
        return {"items": [{"id": 42}], "complete": True}

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
    monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", current_unread)
    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))
    payload = main.PushSubscriptionPayload(
        endpoint="https://push.example/device-a",
        keys={"p256dh": "public-key", "auth": "auth-secret"},
    )

    assert await main.subscribe_push(payload, request) == {"subscribed": True}
    assert store.claim_unseen({42}) == []
    assert store.claim_unseen({42, 43})[0].thread_ids == (43,)


@pytest.mark.anyio
async def test_subscription_fails_closed_when_existing_unread_baseline_is_unavailable(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    monkeypatch.setattr(main, "_push_subscription_store", store)
    monkeypatch.setattr(
        main, "_push_configuration",
        lambda: PushConfiguration("public", "private", "mailto:ops@example.com"),
    )

    async def management_id(_session):
        return "device-a"

    async def unavailable(**_kwargs):
        raise RuntimeError("Gitea unavailable")

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
    monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", unavailable)
    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))
    payload = main.PushSubscriptionPayload(
        endpoint="https://push.example/device-a",
        keys={"p256dh": "public-key", "auth": "auth-secret"},
    )

    with pytest.raises(main.HTTPException) as raised:
        await main.subscribe_push(payload, request)

    assert raised.value.status_code == 503
    assert store.is_subscribed("device-a") is False


def test_delete_sessions_removes_only_the_requested_device_push_state(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    for session_id in ("management-42-phone", "management-42-laptop", "management-84-phone"):
        store.upsert(
            session_id,
            {
                "endpoint": f"https://push.example/{session_id}",
                "keys": {"p256dh": "public-key", "auth": "auth-secret"},
            },
        )

    store.delete_sessions(["management-42-phone", "management-42-laptop"])

    assert store.is_subscribed("management-42-phone") is False
    assert store.is_subscribed("management-42-laptop") is False
    assert store.is_subscribed("management-84-phone") is True