app/control/account/runtime.py

"""Account runtime singletons and hot-apply helpers.

These helpers expose the process-local account refresh runtime without making
callers import ``app.main``. Admin handlers use them to reconcile strategy and
scheduler state after hot config updates.
"""

from typing import TYPE_CHECKING, Literal

if TYPE_CHECKING:
    from .refresh import AccountRefreshService
    from .scheduler import AccountRefreshScheduler

_refresh_service: "AccountRefreshService | None" = None
_refresh_scheduler: "AccountRefreshScheduler | None" = None
_refresh_scheduler_leader = False


def set_refresh_service(service: "AccountRefreshService | None") -> None:
    """Register the process-global account refresh service."""
    global _refresh_service
    _refresh_service = service


def get_refresh_service() -> "AccountRefreshService | None":
    """Return the registered account refresh service, if any."""
    return _refresh_service


def set_refresh_scheduler(scheduler: "AccountRefreshScheduler | None") -> None:
    """Register the process-global account refresh scheduler."""
    global _refresh_scheduler
    _refresh_scheduler = scheduler


def get_refresh_scheduler() -> "AccountRefreshScheduler | None":
    """Return the registered account refresh scheduler, if any."""
    return _refresh_scheduler


def set_refresh_scheduler_leader(is_leader: bool) -> None:
    """Record whether this worker currently owns the refresh scheduler lock."""
    global _refresh_scheduler_leader
    _refresh_scheduler_leader = bool(is_leader)


def is_refresh_scheduler_leader() -> bool:
    """Return True when this worker is the active refresh-scheduler leader."""
    return _refresh_scheduler_leader


def reconcile_refresh_runtime(
    enabled: bool | None = None,
) -> Literal["quota", "random"]:
    """Hot-apply refresh strategy and scheduler state for the current worker."""
    from app.dataplane.account.selector import current_strategy, set_strategy
    from app.platform.config.snapshot import config
    from app.platform.logging.logger import logger

    refresh_enabled = (
        config.get_bool("account.refresh.enabled", False)
        if enabled is None
        else bool(enabled)
    )
    target_strategy: Literal["quota", "random"] = (
        "quota" if refresh_enabled else "random"
    )
    previous_strategy = current_strategy()
    if previous_strategy != target_strategy:
        set_strategy(target_strategy)

    scheduler_action = "unchanged"
    scheduler = _refresh_scheduler
    if scheduler is not None and _refresh_scheduler_leader:
        if refresh_enabled:
            if not scheduler.is_running():
                scheduler.start()
                scheduler_action = "started"
        elif scheduler.is_running():
            scheduler.stop()
            scheduler_action = "stopped"

    if previous_strategy != target_strategy or scheduler_action != "unchanged":
        logger.info(
            "account refresh runtime reconciled: previous_strategy={} strategy={} leader={} scheduler_action={}",
            previous_strategy,
            target_strategy,
            _refresh_scheduler_leader,
            scheduler_action,
        )
    return target_strategy


__all__ = [
    "get_refresh_service",
    "set_refresh_service",
    "get_refresh_scheduler",
    "set_refresh_scheduler",
    "is_refresh_scheduler_leader",
    "set_refresh_scheduler_leader",
    "reconcile_refresh_runtime",
]