backend/app/interactive/interaction/cooldown.py

"""
Per-session action cooldown tracking.

Lives in-memory in a ``CooldownTracker`` keyed by (session_id,
action_id). The tracker is optional — cooldown decisions can also
be made directly from the ix_session_events table (query last use
of this action) — but the in-memory path is faster on the hot
path.

``check_cooldown`` and ``mark_used`` are pure helpers; the router
module decides when to consult / update them.
"""
from __future__ import annotations

import time
from dataclasses import dataclass, field
from typing import Dict, Optional, Tuple


@dataclass
class CooldownTracker:
    """In-process per-session cooldown map."""

    entries: Dict[Tuple[str, str], float] = field(default_factory=dict)

    def clear(self, session_id: str) -> None:
        self.entries = {k: v for k, v in self.entries.items() if k[0] != session_id}


# Module-level singleton (fine for single-process; swap for Redis later).
_TRACKER = CooldownTracker()


def _now_ms() -> int:
    return int(time.time() * 1000)


def check_cooldown(
    session_id: str, action_id: str, cooldown_sec: int, *, now_ms: Optional[int] = None,
) -> float:
    """Return remaining cooldown seconds (0.0 if ready).

    If ``cooldown_sec`` is 0 or negative the action is always
    ready — returns 0.0. Likewise, an action that has never been
    used in this session is always ready.
    """
    if cooldown_sec <= 0:
        return 0.0
    key = (session_id, action_id)
    if key not in _TRACKER.entries:
        return 0.0  # never used → ready
    now = now_ms if now_ms is not None else _now_ms()
    last = _TRACKER.entries[key]
    elapsed_s = (now - last) / 1000.0
    if elapsed_s >= cooldown_sec:
        return 0.0
    return max(0.0, float(cooldown_sec) - elapsed_s)


def mark_used(
    session_id: str, action_id: str, *, now_ms: Optional[int] = None,
) -> None:
    """Record that an action fired just now."""
    now = now_ms if now_ms is not None else _now_ms()
    _TRACKER.entries[(session_id, action_id)] = now


def reset_session(session_id: str) -> None:
    """Clear all cooldown entries for a session (on end / restart)."""
    _TRACKER.clear(session_id)