"""
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,
		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[None] = asyncio.create_task(asyncio.sleep(timeout))
				t_sd: asyncio.Task[None] = 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[None] = asyncio.create_task(event.wait())
				t_shutdown: asyncio.Task[None] = 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 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()