src/services/metrics_service.py

"""
metrics_service.py — LibBee v3.1

Fixes applied:
1. All incr / incr_bucket calls are now pure in-memory (no disk I/O per request).
   Counters accumulate in _counters / _buckets dicts.
2. flush() merges in-memory state to disk atomically using a temp-file rename.
   Called by a background task in app.py every 30 seconds and on shutdown.
3. threading.Lock replaced with asyncio.Lock — consistent with FastAPI's async model.
4. _read() logs a warning (with raw content) on JSON parse failure instead of
   silently returning zeros, so corrupt files are visible in logs.
5. _write() uses os.replace (atomic rename) to prevent corrupt state on crash.
"""
import asyncio
import json
import logging
import os
import time
from pathlib import Path
from typing import Any, Dict

logger = logging.getLogger(__name__)

_DEFAULT_STATE: Dict[str, Any] = {
    "agent_requests": 0,
    "search_requests": 0,
    "feedback_total": 0,
    "intents": {},
    "errors": {},
    "summary_fallbacks": 0,
    "follow_up_hits": 0,
    "search_handoffs": 0,
    "last_updated": 0.0,
}

_MAX_QUERY_LOG = 200  # keep last 200 queries in memory


class MetricsService:
    def __init__(self, path: Path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._lock = asyncio.Lock()

        # In-memory accumulators — flushed to disk periodically
        self._counters: Dict[str, int] = {}
        self._buckets: Dict[str, Dict[str, int]] = {}
        self._dirty = False

        # In-memory query log (last N queries, never flushed to disk)
        self._query_log: list = []

        # Seed disk file if it doesn't exist
        if not self.path.exists():
            self._write_to_disk(dict(_DEFAULT_STATE))

    # ── Public API (sync, no I/O) ──────────────────────────────────────────────

    def log_query(
        self,
        question: str,
        intent: str = "",
        tool: str = "",
        model: str = "",
        response_time: float = 0.0,
        result_count: int = 0,
        error: str = "",
    ) -> None:
        """Append a query record to the in-memory log (capped at _MAX_QUERY_LOG)."""
        import datetime
        record = {
            "timestamp": datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S"),
            "question": (question or "")[:120],
            "intent": intent or "",
            "tool": tool or "",
            "model": model or "",
            "response_time": round(response_time, 2),
            "result_count": result_count,
            "error": (error or "")[:80],
        }
        self._query_log.append(record)
        if len(self._query_log) > _MAX_QUERY_LOG:
            self._query_log = self._query_log[-_MAX_QUERY_LOG:]

    def recent_queries(self, limit: int = 50) -> list:
        """Return the most recent `limit` query records, newest first."""
        return list(reversed(self._query_log[-limit:]))

    # ── Public API (sync, no I/O) ──────────────────────────────────────────────

    def incr(self, key: str, amount: int = 1) -> None:
        """Increment a top-level counter in memory."""
        self._counters[key] = self._counters.get(key, 0) + amount
        self._dirty = True

    def incr_bucket(self, bucket: str, key: str, amount: int = 1) -> None:
        """Increment a nested bucket counter in memory."""
        if bucket not in self._buckets:
            self._buckets[bucket] = {}
        self._buckets[bucket][key] = self._buckets[bucket].get(key, 0) + amount
        self._dirty = True

    def snapshot(self) -> Dict[str, Any]:
        """Return current on-disk state merged with in-memory pending counters."""
        data = self._read_from_disk()
        # Merge pending in-memory counters without flushing
        for k, v in self._counters.items():
            data[k] = int(data.get(k, 0)) + v
        for bucket, keys in self._buckets.items():
            b = data.setdefault(bucket, {})
            for k, v in keys.items():
                b[k] = int(b.get(k, 0)) + v
        return data

    # ── Async flush (called by background task + shutdown) ─────────────────────

    async def flush(self) -> None:
        """Merge in-memory counters to disk. No-op if nothing changed."""
        if not self._dirty:
            return
        async with self._lock:
            if not self._dirty:
                return  # double-check after acquiring lock
            data = self._read_from_disk()
            for k, v in self._counters.items():
                data[k] = int(data.get(k, 0)) + v
            for bucket, keys in self._buckets.items():
                b = data.setdefault(bucket, {})
                for k, v in keys.items():
                    b[k] = int(b.get(k, 0)) + v
            self._write_to_disk(data)
            self._counters.clear()
            self._buckets.clear()
            self._dirty = False

    # ── Internal disk helpers ──────────────────────────────────────────────────

    def _read_from_disk(self) -> Dict[str, Any]:
        if not self.path.exists():
            return dict(_DEFAULT_STATE)
        try:
            return json.loads(self.path.read_text(encoding="utf-8"))
        except Exception as exc:
            raw = ""
            try:
                raw = self.path.read_text(encoding="utf-8", errors="replace")[:200]
            except Exception:
                pass
            logger.warning(
                "MetricsService: failed to parse metrics file %s: %s | raw: %r",
                self.path, exc, raw,
            )
            return dict(_DEFAULT_STATE)

    def _write_to_disk(self, data: Dict[str, Any]) -> None:
        data["last_updated"] = time.time()
        tmp = self.path.with_suffix(".tmp")
        try:
            tmp.write_text(
                json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
            )
            os.replace(tmp, self.path)  # atomic on POSIX / Windows
        except Exception as exc:
            logger.error("MetricsService: failed to write metrics: %s", exc)
            try:
                tmp.unlink(missing_ok=True)
            except Exception:
                pass