"""
Per-project async change queue for long-polling subscribers.

Producers (typically server-side git hooks via /_internal/notify) call notify(project)
when a project's refs change. Consumers (action handlers with updates=true) call
wait_for_changes(projects, timeout) to park their request until any matching project
is notified or the timeout elapses.
"""

from __future__ import annotations

import asyncio
from typing import Literal


class ChangeQueue:
    """Per-project notification fan-out using one-shot asyncio.Events per waiter."""

    DEFAULT_TIMEOUT_SECONDS: float = 30.0

    def __init__(self) -> None:
        self._waiters: dict[str, list[asyncio.Event]] = {}
        self._lock: asyncio.Lock = asyncio.Lock()

    async def wait_for_changes(
        self,
        projects: list[str],
        timeout: float = DEFAULT_TIMEOUT_SECONDS,
        shutdown_event: asyncio.Event | None = None,
    ) -> str | None:
        """Park until any of `projects` is notified, `shutdown_event` is set, or `timeout` elapses.

        Returns the notified project name, or None on timeout/shutdown.
        """
        if shutdown_event and shutdown_event.is_set():
            return None

        if not projects:
            if shutdown_event:
                t_sleep: asyncio.Task[object] = asyncio.create_task(asyncio.sleep(timeout))
                t_sd: asyncio.Task[object] = asyncio.create_task(shutdown_event.wait())
                try:
                    done, pending = await asyncio.wait(
                        {t_sleep, t_sd},
                        timeout=timeout,
                        return_when=asyncio.FIRST_COMPLETED,
                    )
                finally:
                    for t in pending:
                        t.cancel()
                    await asyncio.gather(t_sleep, t_sd, return_exceptions=True)
                return None
            await asyncio.sleep(timeout)
            return None

        event: asyncio.Event = asyncio.Event()
        async with self._lock:
            for project in projects:
                self._waiters.setdefault(project, []).append(event)
        try:
            if shutdown_event:
                t_notify: asyncio.Task[Literal[True]] = asyncio.create_task(event.wait())
                t_shutdown: asyncio.Task[Literal[True]] = asyncio.create_task(shutdown_event.wait())
                try:
                    done, pending = await asyncio.wait(
                        {t_notify, t_shutdown},
                        timeout=timeout,
                        return_when=asyncio.FIRST_COMPLETED,
                    )
                finally:
                    for t in pending:
                        t.cancel()
                    await asyncio.gather(t_notify, t_shutdown, return_exceptions=True)
                if not done:
                    return None
                if t_shutdown in done:
                    return None
                return getattr(event, "_change_queue_project", None)

            try:
                await asyncio.wait_for(event.wait(), timeout=timeout)
            except TimeoutError:
                return None
            return getattr(event, "_change_queue_project", None)
        finally:
            async with self._lock:
                for project in projects:
                    bucket: list[asyncio.Event] = self._waiters.get(project, [])
                    if event in bucket:
                        bucket.remove(event)
                    if not bucket:
                        self._waiters.pop(project, None)

    async def notify(self, project: str) -> int:
        """Wake all waiters subscribed to `project`. Returns the number of waiters woken."""
        async with self._lock:
            waiters: list[asyncio.Event] = self._waiters.pop(project, [])
        for event in waiters:
            event._change_queue_project = project  # type: ignore[attr-defined]
            event.set()
        return len(waiters)

    async def notify_all(self) -> int:
        """Wake every current waiter. Returns the number of waiters woken."""
        async with self._lock:
            snapshot: list[tuple[str, asyncio.Event]] = [
                (project, event) for project, bucket in self._waiters.items() for event in bucket
            ]
            self._waiters.clear()
        seen: set[int] = set()
        woken = 0
        for project, event in snapshot:
            eid = id(event)
            if eid in seen:
                continue
            seen.add(eid)
            event._change_queue_project = project  # type: ignore[attr-defined]
            event.set()
            woken += 1
        return woken

    async def waiter_count(self, project: str | None = None) -> int:
        """Diagnostic: count waiters for one project (or total across all projects)."""
        async with self._lock:
            if project is not None:
                return len(self._waiters.get(project, []))
            return sum(len(bucket) for bucket in self._waiters.values())


CHANGE_QUEUE: ChangeQueue = ChangeQueue()