FabledScribe/src/scribe/services/system_usage.py

"""System usage telemetry — were an area's rulings read once shown?

The twin of `note_usage` and `rule_usage` for Systems (milestone 444). The
rulings arm shows a System's rulings when a command or edit touches the
System's files; a pull is somebody then opening the System (`get_system`). The
ratio says whether delivering rulings by path earns its line.

Fire-and-forget like its siblings: telemetry never adds latency to, or
breaks, the surface it observes, and failures report through the shared
canary rather than vanishing.
"""
from __future__ import annotations

import logging

from sqlalchemy import func, select

from scribe.models import async_session
from scribe.models.base import iso
from scribe.models.system_usage import PULLED, SURFACED, SystemUsageEvent
from scribe.services.background import report_telemetry_failure, spawn

logger = logging.getLogger(__name__)


async def _insert_events(rows: list[dict]) -> None:
    """Persist usage rows. Best-effort: failures degrade, visibly."""
    try:
        async with async_session() as session:
            session.add_all([SystemUsageEvent(**row) for row in rows])
            await session.commit()
    except Exception:
        await report_telemetry_failure("system_usage", "write")


def _rows(user_id, system_ids, event: str, source: str, project_id) -> list[dict]:
    pid = int(project_id or 0) or None
    return [
        {"user_id": user_id, "system_id": int(sid), "event": event,
         "source": source, "project_id": pid}
        for sid in system_ids
    ]


def record_system_surfaced(
    *, user_id: int | None, system_ids: list[int], source: str,
    project_id: int | None = None,
) -> None:
    """Fire-and-forget: these Systems' rulings were shown in full. A repeat
    rendered as a short reference is not a surfacing and is not recorded."""
    try:
        rows = _rows(user_id, system_ids, SURFACED, source, project_id)
    except Exception:
        logger.debug("system usage payload build failed", exc_info=True)
        return
    if rows:
        spawn(_insert_events(rows), site="system_usage_write")


def record_system_pulled(
    *, user_id: int | None, system_id: int, source: str,
    project_id: int | None = None,
) -> None:
    """Fire-and-forget: a System was opened in full."""
    try:
        rows = _rows(user_id, [system_id], PULLED, source, project_id)
    except Exception:
        logger.debug("system usage payload build failed", exc_info=True)
        return
    spawn(_insert_events(rows), site="system_usage_write")


def empty_system_usage() -> dict:
    """The zero readout — a System no event has touched.

    Rendered unconditionally, like `empty_rule_usage`, so a System predating
    the table reads as "never shown, never opened" rather than as a missing
    key. Every System in an install predates it, so for a while this is the
    normal state and must not look like a broken readout.

    The same four keys as the client's `RecordUsage`, but read differently: a
    System's rulings are delivered IN FULL in the injected line, so a
    surfacing with no pull is the arm working, not dead weight. See
    `retrieval_telemetry`'s `system_usage` block for why no ratio is offered.
    """
    return {
        "surfaced_count": 0,
        "pull_count": 0,
        "last_surfaced_at": None,
        "last_pulled_at": None,
    }


async def usage_for_systems(system_ids: list[int]) -> dict[int, dict]:
    """Aggregate usage for a set of Systems: {system_id: {counts + timestamps}}.

    One GROUP BY for the whole list, never a query per row — this decorates a
    list view. Systems with no events come back as `empty_system_usage()`.
    Fails open: a readout that could break the Systems list is worse than a
    zero, but the failure is reported so the zero is not mistaken for silence
    (#2663).
    """
    ids = [int(s) for s in system_ids]
    out: dict[int, dict] = {sid: empty_system_usage() for sid in ids}
    if not ids:
        return out
    try:
        async with async_session() as session:
            rows = (
                await session.execute(
                    select(
                        SystemUsageEvent.system_id,
                        SystemUsageEvent.event,
                        func.count().label("n"),
                        func.max(SystemUsageEvent.created_at).label("last_at"),
                    )
                    .where(SystemUsageEvent.system_id.in_(ids))
                    .group_by(SystemUsageEvent.system_id, SystemUsageEvent.event)
                )
            ).all()
    except Exception:
        await report_telemetry_failure("system_usage", "readout")
        return out
    for system_id, event, n, last_at in rows:
        slot = out.get(int(system_id))
        if slot is None:
            continue
        if event == SURFACED:
            slot["surfaced_count"] = int(n)
            slot["last_surfaced_at"] = iso(last_at)
        elif event == PULLED:
            slot["pull_count"] = int(n)
            slot["last_pulled_at"] = iso(last_at)
    return out