"""One backend-wide reader of Docker's event stream, for server-side caches.

The browser follows Docker on its own (``/ws/events``) and refetches whatever
an event touched. Two of the answers it refetched were cached here, though,
and did not know anything had happened:

* the dashboard's fleet rollup, held for 25 seconds — a stack stopped in a
  terminal showed as running in the dashboard tiles for up to half a minute,
  and the event that should have refreshed them had come and gone;
* the image update status, whose local digests were only re-read after a
  pull *through StackPilot* — after ``docker compose pull && up -d`` in a
  terminal the "update available" badge stayed until the next registry check,
  an hour by default.

So: :func:`generation` changes with every container or image event, and a
cache that remembers the generation it was computed at knows when it is stale.
Image events also re-read the local digests (no registry calls), debounced so
a ``pull`` of ten layers is one pass.
"""
from __future__ import annotations

import logging
import threading

logger = logging.getLogger("stackpilot.docker_watch")

_EVENT_TYPES = ["container", "image"]
#: Container actions that change nothing a cache holds (see routers/ws.py).
_IGNORED = {
    "exec_create", "exec_start", "exec_die", "exec_detach",
    "attach", "top", "resize", "archive-path", "extract-to-dir",
}
#: Wait this long after the last image event before re-reading digests.
DIGEST_DEBOUNCE_S = 2.0
RECONNECT_MAX_S = 30.0

_generation = 0
_lock = threading.Lock()
_stop = threading.Event()
_thread: threading.Thread | None = None
_digest_timer: threading.Timer | None = None


def generation() -> int:
    """Changes whenever Docker reports a container or image change."""
    return _generation


def bump() -> None:
    global _generation
    with _lock:
        _generation += 1


def _refresh_digests() -> None:
    from services import update_service

    try:
        changed = update_service.refresh_local_digests()
        if changed:
            logger.info("Re-read local digests after image events: %d changed", changed)
    except Exception as exc:  # noqa: BLE001 - a cache refresh must never kill the watcher
        logger.debug("Local digest refresh failed: %s", exc)


def _schedule_digest_refresh() -> None:
    global _digest_timer
    with _lock:
        if _digest_timer is not None:
            _digest_timer.cancel()
        _digest_timer = threading.Timer(DIGEST_DEBOUNCE_S, _refresh_digests)
        _digest_timer.daemon = True
        _digest_timer.start()


def handle(event: dict) -> None:
    """Apply one Docker event to the caches."""
    kind = event.get("Type")
    action = (event.get("Action") or "").split(":")[0]
    if kind == "container" and action in _IGNORED:
        return
    bump()
    if kind == "image":
        _schedule_digest_refresh()


def _run() -> None:
    from docker_client import get_client

    delay = 1.0
    while not _stop.is_set():
        try:
            stream = get_client().events(decode=True, filters={"type": _EVENT_TYPES})
            delay = 1.0
            # Anything may have happened while we were not listening.
            bump()
            for event in stream:
                if _stop.is_set():
                    break
                handle(event)
        except Exception as exc:  # noqa: BLE001 - daemon restarts, socket hiccups
            logger.debug("Docker event stream ended: %s", exc)
        if _stop.wait(delay):
            break
        delay = min(delay * 2, RECONNECT_MAX_S)


def start() -> None:
    global _thread
    if _thread is not None and _thread.is_alive():
        return
    _stop.clear()
    _thread = threading.Thread(target=_run, name="docker-watch", daemon=True)
    _thread.start()


def stop() -> None:
    _stop.set()