siftlode/backend/app/debugsession.py

"""Debug sessions (E16): Start → reproduce → Stop, and the evidence a bug report is built from.

While a session is open its user's browser names it in `X-Debug-Session` on every request. The auth
dependency accepts the header only for the caller's OWN open session (anything else is ignored, never
an error — a stale tab must not break the app) and then logs that request at the session's level with
every line tagged by its id; the browser tags its own events the same way. The issue is later built
from exactly the rows carrying the tag, so what the review list shows is what gets sent.

Starting a session IS the consent for its level, for its duration: the start dialog carries the
consent text. It needs no change to the user's own diagnostics level, and none is made — when the
session ends the level in force is simply the user's own again.

A session is capped at `MAX_MINUTES`, set by the server at Start. An open one past its cap is over
whether or not anything has closed it yet: it reads as over at once, a new Start closes it, and the
hourly log job sweeps the rest. Sessions go with the logs they point at, after `log_retention_days`.

Core layer: imports `models`, `diagnostics`.
"""

import secrets
from datetime import datetime, timedelta, timezone
from enum import StrEnum
from typing import Literal, get_args

from sqlalchemy import and_, delete, func, not_, select, update
from sqlalchemy.orm import Session, defer

from app import diagnostics
from app.models import DebugScreenshot, DebugSession, LogEvent, User

HEADER = "x-debug-session"
MAX_MINUTES = diagnostics.SESSION_MINUTES
SCREENSHOT_MAX_BYTES = 10 * 1024 * 1024
SCREENSHOTS_MAX = 8
# What one user may keep in screenshots across ALL their sessions (until retention takes them): the
# per-session cap alone would let Start → 8 images → Stop repeat without end.
SCREENSHOT_USER_MAX_BYTES = 100 * 1024 * 1024
# The screenshot types kept, and the extension each is exported and sent under — one table, so an
# accepted type always has a name.
IMAGE_TYPES = {"image/png": "png", "image/jpeg": "jpg", "image/webp": "webp"}
# Rows an issue can carry. A 60-minute session at Teljes is a few thousand rows; past this the
# bundle says how many it left out rather than growing without bound.
BUNDLE_MAX = 5_000

SessionLevel = Literal["DEBUG", "TRACE"]
SESSION_LEVELS: tuple[str, ...] = get_args(SessionLevel)
# Detail order of the diagnostics levels: a session raises the level in force, never lowers it.
_DETAIL = {"INFO": 0, "DEBUG": 1, "TRACE": 2}
_ID_LEN = 16


class EndReason(StrEnum):
    STOPPED = "stopped"
    EXPIRED = "expired"


def is_live(session: DebugSession, now: datetime | None = None) -> bool:
    """Open and within its cap — THE liveness rule: an open session past its cap is over."""
    return session.ended_at is None and session.until > (now or datetime.now(timezone.utc))


def open_session(db: Session, user: User) -> DebugSession | None:
    """The user's live session, or None. Read-only: one past its cap is over and reads as None;
    `start` and the hourly sweep are what close it."""
    session = _unclosed(db, user)
    if session is None or not is_live(session, datetime.now(timezone.utc)):
        return None
    return session


def _unclosed(db: Session, user: User) -> DebugSession | None:
    return db.scalar(
        select(DebugSession).where(DebugSession.user_id == user.id, DebugSession.ended_at.is_(None))
    )


def start(db: Session, user: User, level: SessionLevel) -> DebugSession:
    """Open a session, or return the live one — a second tab pressing Start joins the recording
    rather than ending it. One past its cap is closed first: the one-open-per-user index would
    refuse the new row beside it. Flushes; the caller commits. Two concurrent Starts can still both
    get here: the index refuses the second, and the caller joins the one that won."""
    now = datetime.now(timezone.utc)
    current = _unclosed(db, user)
    if current is not None:
        if is_live(current, now):
            return current
        end(current, EndReason.EXPIRED)
        db.flush()
    session = DebugSession(
        id=secrets.token_hex(_ID_LEN // 2),
        user_id=user.id,
        level=level,
        started_at=now,
        until=now + timedelta(minutes=MAX_MINUTES),
    )
    db.add(session)
    db.flush()
    return session


def stop(session: DebugSession) -> None:
    """The user's Stop. Past its cap the session had already ended on its own, and a late Stop must
    not rename that."""
    end(session, EndReason.STOPPED if is_live(session) else EndReason.EXPIRED)


def end(session: DebugSession, reason: EndReason) -> None:
    """Close a session. An expired one is dated at its cap, not at whenever it was noticed."""
    if session.ended_at is not None:
        return
    session.ended_at = (
        session.until
        if reason is EndReason.EXPIRED
        else min(datetime.now(timezone.utc), session.until)
    )
    session.ended_reason = reason


def from_header(db: Session, user: User, value: str | None) -> DebugSession | None:
    """The session a request names, if it is the caller's own and still live; else None."""
    if not value or len(value) != _ID_LEN or not all(c in "0123456789abcdef" for c in value):
        return None
    session = db.get(DebugSession, value)
    if (
        session is None
        or session.user_id != user.id
        or not is_live(session, datetime.now(timezone.utc))
    ):
        return None
    return session


def level_for(user: User, session: DebugSession | None) -> tuple[str, datetime | None]:
    """The diagnostics level (and its expiry) a request is logged at: the user's own, raised to the
    session's while one is named — whichever records more.

    A named session also caps the expiry at its own: this pair is what the process tracks under the
    session (app/logs.py), and an expiry outliving the recording would hold the root logger down
    after it ended."""
    own = diagnostics.effective_level(user)
    until = user.diagnostics_until if own != diagnostics.BASE else None
    if session is None:
        return own, until
    if _DETAIL[session.level] <= _DETAIL[own]:
        return own, min(until, session.until) if until else session.until
    return session.level, session.until


def lock(db: Session, session: DebugSession) -> None:
    """Re-read the session under a row lock, held until the caller commits. Everything that changes
    what an issue is made of — the draft, the screenshots, the send itself — takes it first: a send
    in flight sees none of them change under it, and each of them sees a send that committed."""
    db.refresh(session, with_for_update=True)


def save_draft(session: DebugSession, draft: dict) -> None:
    """Keep the composer's draft (the caller validated, locked and commits). Shape: `title`, `body`
    (the rendered markdown an export or an issue carries), `excluded` (log row ids taken out of the
    evidence) and `form` (the composer's own fields, to reopen it as it was). A delivery's progress
    is not the composer's to replace, and is kept."""
    progress = (session.draft or {}).get("delivery")
    session.draft = {**draft, "delivery": progress} if progress else draft


# --- a delivery's progress -----------------------------------------------------------------
# Written by app/issues/gitea.py into the draft; read HERE, so the send's decision and what the
# composer shows are one reading of it.
def delivery(session: DebugSession) -> dict:
    return dict((session.draft or {}).get("delivery") or {})


def not_uploaded(session: DebugSession) -> list[str]:
    """The files a send that stopped half-way has still to upload — by name, as the issue carries
    them. What the composer names to the user."""
    return list(delivery(session).get("pending") or [])


def images_pending(session: DebugSession) -> bool:
    """Whether the issue's body still has to be edited to show the screenshots it already carries
    (every upload went, the edit that shows them did not)."""
    state = delivery(session)
    shown = state.get("images_shown")
    return not shown and any(n.startswith("screenshot-") for n in state.get("links") or {})


def unfinished(session: DebugSession) -> bool:
    """Whether a send owes the issue anything — the retry's question."""
    return bool(not_uploaded(session)) or images_pending(session)


# A send claims the session while it talks to the tracker, instead of holding the row's lock across
# those calls: a Stop, an autosave or a screenshot would queue behind a lock for the whole send.
# They see the claim and are refused at once instead. A claim older than this was left by a request
# that died mid-send (the process restarted), and the next send takes it over.
SEND_CLAIM_SECONDS = 600


def sending(session: DebugSession) -> bool:
    """Whether a send is under way right now (a live claim)."""
    since = delivery(session).get("sending")
    if not since:
        return False
    started = datetime.fromisoformat(since)
    return (datetime.now(timezone.utc) - started).total_seconds() < SEND_CLAIM_SECONDS


def claim_send(session: DebugSession) -> None:
    """Mark the session as being sent. The caller holds the lock and commits, which is what makes
    the claim visible to everyone else."""
    set_delivery(session, {**delivery(session), "sending": datetime.now(timezone.utc).isoformat()})


def release_send(session: DebugSession) -> None:
    state = {k: v for k, v in delivery(session).items() if k != "sending"}
    set_delivery(session, state)


def set_delivery(session: DebugSession, state: dict) -> None:
    """Write a delivery's progress (app/issues/gitea.py keeps it; this module reads it)."""
    # A new dict, so the JSONB column sees the change.
    session.draft = {**(session.draft or {}), "delivery": state}


def serialize(session: DebugSession | None, screenshots: int = 0) -> dict | None:
    if session is None:
        return None
    return {
        "id": session.id,
        "level": session.level,
        "started_at": session.started_at.isoformat(),
        "until": session.until.isoformat(),
        "ended_at": session.ended_at.isoformat() if session.ended_at else None,
        "ended_reason": session.ended_reason,
        "issue_url": session.issue_url,
        # A send that stopped half-way: the files it has still to upload, and whether the issue's
        # body has still to be edited to show the screenshots it already carries.
        "pending_files": not_uploaded(session),
        "images_pending": images_pending(session),
        "screenshots": screenshots,
        "draft": session.draft,
    }


def view(db: Session, session: DebugSession) -> dict:
    """The session as the API serves it — one shape wherever it is served."""
    return serialize(session, screenshot_count(db, session)) or {}


def screenshot_count(db: Session, session: DebugSession) -> int:
    return (
        db.scalar(
            select(func.count())
            .select_from(DebugScreenshot)
            .where(DebugScreenshot.session_id == session.id)
        )
        or 0
    )


# --- screenshots ---------------------------------------------------------------------------
# By content, not by the declared type: what is stored is served back to a browser as an image.
_SIGNATURES = (
    (b"\x89PNG\r\n\x1a\n", "image/png"),
    (b"\xff\xd8\xff", "image/jpeg"),
)


def sniff(data: bytes) -> str | None:
    """The image type `data` really is — one of IMAGE_TYPES — or None."""
    for magic, mime in _SIGNATURES:
        if data.startswith(magic):
            return mime
    if len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP":
        return "image/webp"
    return None


def stored_screenshot_bytes(db: Session, user_id: int) -> int:
    return (
        db.scalar(
            select(func.coalesce(func.sum(DebugScreenshot.size), 0))
            .join(DebugSession, DebugSession.id == DebugScreenshot.session_id)
            .where(DebugSession.user_id == user_id)
        )
        or 0
    )


class ScreenshotRefused(Exception):
    """A screenshot that cannot be kept; `code` is the refusal's machine word."""

    def __init__(self, code: str) -> None:
        super().__init__(code)
        self.code = code


def add_screenshot(db: Session, session: DebugSession, data: bytes) -> DebugScreenshot:
    """Keep one screenshot. Locks the session and flushes; the caller commits. Refused once the
    issue was sent (it can no longer carry it), past the per-session count or the user's storage,
    or when it is not an image we can show."""
    if not data:
        raise ScreenshotRefused("empty")
    mime = sniff(data)
    if mime is None:
        raise ScreenshotRefused("not_an_image")
    # Under the lock: two uploads at once cannot both pass the count.
    lock(db, session)
    if sending(session):
        raise ScreenshotRefused("sending")
    if session.issue_url is not None:
        raise ScreenshotRefused("already_sent")
    if screenshot_count(db, session) >= SCREENSHOTS_MAX:
        raise ScreenshotRefused("too_many")
    if stored_screenshot_bytes(db, session.user_id) + len(data) > SCREENSHOT_USER_MAX_BYTES:
        raise ScreenshotRefused("storage_full")
    shot = DebugScreenshot(
        session_id=session.id,
        created_at=datetime.now(timezone.utc),
        mime=mime,
        size=len(data),
        data=data,
    )
    db.add(shot)
    db.flush()
    return shot


def remove_screenshot(db: Session, session: DebugSession, shot: DebugScreenshot) -> None:
    """Delete one screenshot, under the lock. Refused while a send is under way or unfinished: a
    delivery names the screenshots by their order, and removing one would renumber what it still
    has to send. Once the issue has everything, this only drops OUR copy — the issue keeps its
    own. The caller commits."""
    lock(db, session)
    if sending(session):
        raise ScreenshotRefused("sending")
    if unfinished(session):
        raise ScreenshotRefused("already_sent")
    db.delete(shot)
    db.flush()


def screenshots(db: Session, session: DebugSession) -> list[DebugScreenshot]:
    """The session's screenshots, oldest first, WITHOUT their bytes loaded."""
    return list(
        db.scalars(
            select(DebugScreenshot)
            .where(DebugScreenshot.session_id == session.id)
            .options(defer(DebugScreenshot.data))
            .order_by(DebugScreenshot.id)
        )
    )


# --- the evidence --------------------------------------------------------------------------
# The browser's own log upload (`POST` here, every few seconds while recording) is how the evidence
# arrives, not part of it: its server lines would fill the bundle with the recording's plumbing. The
# browser leaves its copies out for the same reason (frontend lib/diagnostics `isOwnRequest`).
INGEST_PATH = "/api/diagnostics/events"


def _evidence(session: DebugSession):
    upload_line = and_(
        LogEvent.event == "http.request", LogEvent.attrs["path"].astext == INGEST_PATH
    )
    # `IS TRUE`, not the bare condition: for a row with no event or no attrs it is NULL, and
    # `NOT NULL` would drop exactly the rows that are evidence.
    return (LogEvent.debug_session_id == session.id, not_(upload_line.is_(True)))


def bundle(
    db: Session, session: DebugSession, *, drop_excluded: bool = False
) -> tuple[list[LogEvent], int]:
    """The session's log rows, browser and server interleaved in time, capped at BUNDLE_MAX — and
    how many rows it has in all, so a cut bundle can say so. `drop_excluded` leaves out the rows
    the user took out in the composer: what an export or an issue carries."""
    where = list(_evidence(session))
    excluded = (session.draft or {}).get("excluded") or [] if drop_excluded else []
    if excluded:
        where.append(LogEvent.id.not_in(excluded))
    rows = list(
        db.scalars(
            select(LogEvent).where(*where).order_by(LogEvent.ts, LogEvent.id).limit(BUNDLE_MAX)
        )
    )
    total = len(rows)
    if total == BUNDLE_MAX:
        total = db.scalar(select(func.count()).select_from(LogEvent).where(*where)) or total
    return rows, total


# --- housekeeping (the hourly log job) -----------------------------------------------------
def sweep(db: Session) -> int:
    """Stage closing every open session past its cap; returns how many."""
    now = datetime.now(timezone.utc)
    result = db.execute(
        update(DebugSession)
        .where(DebugSession.ended_at.is_(None), DebugSession.until <= now)
        .values(ended_at=DebugSession.until, ended_reason=EndReason.EXPIRED.value)
        .execution_options(synchronize_session=False)
    )
    return result.rowcount or 0


def gc(db: Session, retention_days: int) -> int:
    """Stage deleting sessions (and by cascade their screenshots) started before the log retention
    window — the rows they would bundle are gone by then too. Returns how many."""
    cutoff = datetime.now(timezone.utc) - timedelta(days=max(1, retention_days))
    result = db.execute(delete(DebugSession).where(DebugSession.started_at < cutoff))
    return result.rowcount or 0