from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass, field
from typing import Callable, Coroutine

logger = logging.getLogger(__name__)


@dataclass
class ScheduledTask:
    name: str
    coro_factory: Callable[[], Coroutine]
    interval_seconds: int
    run_on_startup: bool = False


@dataclass
class _DueTracker:
    """Decides which scheduled tasks are due, with a self-overlap guard.

    Pure (no asyncio, no clock of its own — `now` is passed in) so the
    scheduling policy is unit-testable without timing races. A task whose prior
    run is still in flight is NOT re-fired: overlapping poll runs stack up open
    connections/subprocesses and amplify any per-poll resource use — the same
    failure mode behind the fd-leak lockups. The skipped task is retried on the
    next tick once it completes (its last_run isn't advanced while skipped).
    """
    last_run: dict[str, float] = field(default_factory=dict)
    in_flight: set[str] = field(default_factory=set)

    def due(self, tasks: list[ScheduledTask], now: float) -> list[ScheduledTask]:
        ready: list[ScheduledTask] = []
        for task in tasks:
            if now - self.last_run.get(task.name, 0) < task.interval_seconds:
                continue
            if task.name in self.in_flight:
                logger.warning(
                    "Scheduled task %r still running — skipping this tick",
                    task.name)
                continue
            ready.append(task)
        return ready

    def mark_started(self, task: ScheduledTask, now: float) -> None:
        self.in_flight.add(task.name)
        self.last_run[task.name] = now

    def mark_done(self, name: str) -> None:
        self.in_flight.discard(name)


async def start_scheduler(tasks: list[ScheduledTask]) -> None:
    """Run scheduled tasks in a loop. Call with asyncio.create_task()."""
    tracker = _DueTracker()

    def _spawn(task: ScheduledTask, now: float) -> None:
        tracker.mark_started(task, now)
        asyncio.create_task(_run_task(task, tracker))

    for task in tasks:
        if task.run_on_startup:
            logger.info(f"Startup task: {task.name}")
            _spawn(task, asyncio.get_event_loop().time())

    while True:
        now = asyncio.get_event_loop().time()
        for task in tracker.due(tasks, now):
            _spawn(task, now)
        await asyncio.sleep(1)


async def _run_task(task: ScheduledTask, tracker: _DueTracker) -> None:
    try:
        await task.coro_factory()
    except Exception:
        logger.exception(f"Scheduled task {task.name!r} raised an exception")
    finally:
        tracker.mark_done(task.name)