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


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,
	) -> str | None:
		"""Park until any of `projects` is notified, or `timeout` seconds elapse.

		Returns the notified project name, or None on timeout.
		"""
		if not projects:
			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:
			await asyncio.wait_for(event.wait(), timeout=timeout)
		except TimeoutError:
			return 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)
		return getattr(event, "_change_queue_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 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()