import asyncio
import json
import secrets
import time
from dataclasses import dataclass
from typing import Awaitable, Callable

from src.push_subscription_store import PushSubscriptionStore


@dataclass(frozen=True)
class PushConfiguration:
    public_key: str
    private_key: str
    subject: str

    @property
    def enabled(self) -> bool:
        return bool(self.public_key and self.private_key and self.subject)


async def send_web_push(
    subscription: dict, payload: str, configuration: PushConfiguration
) -> None:
    from pywebpush import webpush

    await asyncio.to_thread(
        webpush,
        subscription_info=subscription,
        data=payload,
        vapid_private_key=configuration.private_key,
        vapid_claims={"sub": configuration.subject},
        ttl=300,
    )


async def dispatch_unread_updates(
    store: PushSubscriptionStore,
    configuration: PushConfiguration,
    unread: Callable[[], Awaitable[dict]],
    send: Callable[[dict, str], Awaitable[None]] | None = None,
    *,
    session_active: Callable[[str], Awaitable[bool]] | None = None,
    lease_seconds: float = 60.0,
    send_timeout_seconds: float = 10.0,
    max_concurrency: int = 8,
    max_individual_notifications: int = 3,
) -> int:
    if not configuration.enabled:
        return 0
    owner = secrets.token_urlsafe(18)
    acquired = await asyncio.to_thread(
        store.acquire_dispatch_lease,
        owner,
        now=time.time(),
        lease_seconds=lease_seconds,
    )
    if not acquired:
        return 0
    try:
        page = await unread()
        thread_ids = {
            int(item["id"])
            for item in page.get("items", [])
            if isinstance(item, dict) and str(item.get("id", "")).isdigit()
        }
        deliveries = await asyncio.to_thread(store.claim_unseen, thread_ids)
        if session_active is not None:
            try:
                authorized = [await session_active(item.session_id) for item in deliveries]
            except Exception:
                # Authorization state is mandatory for delivery. Preserve subscriptions
                # so a temporary registry failure can be retried safely.
                return 0
            for delivery, active in zip(deliveries, authorized):
                if not active:
                    await asyncio.to_thread(store.delete_session, delivery.session_id)
            deliveries = [
                delivery
                for delivery, active in zip(deliveries, authorized)
                if active
            ]
        semaphore = asyncio.Semaphore(max(1, max_concurrency))
        ownership_lost = asyncio.Event()

        async def dispatch_device(delivery) -> int:
            async with semaphore:
                if ownership_lost.is_set():
                    return 0
                count = 0
                digest_pending = set(delivery.digest_ids)
                new_ids = tuple(
                    thread_id
                    for thread_id in delivery.thread_ids
                    if thread_id not in digest_pending
                )
                individual_ids = new_ids[:max(0, max_individual_notifications)]
                overflow_ids = delivery.digest_ids + new_ids[len(individual_ids):]
                can_send_digest = True
                for thread_id in individual_ids:
                    still_owner = await asyncio.to_thread(
                        store.acquire_dispatch_lease,
                        owner,
                        now=time.time(),
                        lease_seconds=lease_seconds,
                    )
                    if not still_owner:
                        ownership_lost.set()
                        return count
                    payload = json.dumps(
                        {
                            "title": "New work update",
                            "body": "Tap to review it in Stackchain.",
                            "route": f"#/my-work/update/{thread_id}",
                            "tag": f"stackchain-update-{thread_id}",
                            "notification_id": thread_id,
                        },
                        separators=(",", ":"),
                    )
                    try:
                        if send is None:
                            operation = send_web_push(
                                delivery.subscription, payload, configuration
                            )
                        else:
                            operation = send(delivery.subscription, payload)
                        await asyncio.wait_for(operation, timeout=send_timeout_seconds)
                    except Exception as error:
                        status = getattr(
                            getattr(error, "response", None), "status_code", None
                        )
                        if status in {404, 410}:
                            await asyncio.to_thread(
                                store.delete_session, delivery.session_id
                            )
                        can_send_digest = False
                        # Leave this device's transient failures unseen for a later
                        # poll instead of paying the endpoint deadline repeatedly.
                        break
                    await asyncio.to_thread(
                        store.mark_delivered, delivery.session_id, (thread_id,)
                    )
                    count += 1
                if overflow_ids and can_send_digest:
                    still_owner = await asyncio.to_thread(
                        store.acquire_dispatch_lease,
                        owner,
                        now=time.time(),
                        lease_seconds=lease_seconds,
                    )
                    if not still_owner:
                        ownership_lost.set()
                        return count
                    payload = json.dumps(
                        {
                            "title": f"{len(overflow_ids)} new work updates",
                            "body": "Tap to review them in Stackchain.",
                            "route": "#/my-work/updates",
                            "tag": "stackchain-update-digest",
                            "update_count": len(overflow_ids),
                        },
                        separators=(",", ":"),
                    )
                    try:
                        if send is None:
                            operation = send_web_push(
                                delivery.subscription, payload, configuration
                            )
                        else:
                            operation = send(delivery.subscription, payload)
                        await asyncio.wait_for(operation, timeout=send_timeout_seconds)
                    except Exception as error:
                        status = getattr(
                            getattr(error, "response", None), "status_code", None
                        )
                        if status in {404, 410}:
                            await asyncio.to_thread(
                                store.delete_session, delivery.session_id
                            )
                        else:
                            await asyncio.to_thread(
                                store.mark_digest_pending,
                                delivery.session_id,
                                overflow_ids,
                            )
                        return count
                    await asyncio.to_thread(
                        store.mark_delivered, delivery.session_id, overflow_ids
                    )
                    count += 1
                return count

        counts = await asyncio.gather(
            *(dispatch_device(delivery) for delivery in deliveries)
        )
        return sum(counts)
    finally:
        await asyncio.to_thread(store.release_dispatch_lease, owner)