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 uuid import uuid4

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"}
# What a screenshot is called in the zip and on the issue. `issues/bundle.py` builds the names from
# this and `IMAGE_TYPES`; this module and `issues/gitea.py` read them back to tell a screenshot's
# link from a log's. One definition, because three copies of a name is how the next rename breaks
# `images_pending` into silence.
SCREENSHOT_PREFIX = "screenshot-"
# 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_PREFIX) 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 not renewed for this long was left by
# a request that died mid-send (the process restarted), and the next send takes it over — which is
# why a live send RENEWS it every time it reports progress (`set_delivery`). A claim is a lease, and
# a send slower than the lease must not lose it while it is still running.
SEND_CLAIM_SECONDS = 600


def _claim(session: DebugSession) -> dict:
    claim = delivery(session).get("sending")
    return claim if isinstance(claim, dict) else {}


def _stamp(token: str) -> dict:
    return {"at": datetime.now(timezone.utc).isoformat(), "by": token}


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


def claim_send(session: DebugSession) -> str:
    """Mark the session as being sent, and return the token that OWNS this claim. The caller holds
    the lock and commits, which is what makes the claim visible to everyone else."""
    token = uuid4().hex
    _write_delivery(session, {**delivery(session), "sending": _stamp(token)})
    return token


def release_send(session: DebugSession, token: str) -> None:
    """Give back our OWN claim, and only ours: one that aged out and was taken over belongs to the
    send holding it now, and handing that one back would let a third send in beside it."""
    if _claim(session).get("by") != token:
        return
    _write_delivery(session, {k: v for k, v in delivery(session).items() if k != "sending"})


class SendClaimLost(Exception):
    """This send's claim is no longer on the row: it aged out and another send took it over. The
    send must stop — what it uploads from here it can no longer record, so a retry would send it
    again — and it must not write, because the row now holds the other send's progress."""


def claim_owner(db: Session, session: DebugSession) -> str | None:
    """Who holds the row's claim RIGHT NOW, read from the ROW and not from the object.

    `expire_on_commit` is off (app/db.py), so a checkpoint does not re-read: a send that stalled
    past its lease still carries its own dead token in memory long after another send took over.
    A targeted scalar, not `lock`/`db.refresh` — a refresh here would discard the `issue_url` that
    `gitea.deliver` assigns one line before its first progress report, and that link is the one
    thing the send must not lose."""
    return db.scalar(
        select(DebugSession.draft["delivery"]["sending"]["by"].as_string()).where(
            DebugSession.id == session.id
        )
    )


def set_delivery(db: Session, session: DebugSession, state: dict, token: str) -> None:
    """Write a delivery's progress (app/issues/gitea.py keeps it; this module reads it), and renew
    the claim while doing it — a send still reporting progress is still alive, so one slower than
    the lease cannot expire while it is still running.

    The claim is not the caller's to carry: `gitea.deliver` threads ONE snapshot of this dict
    through a whole send, so a progress report that carried the claim wrote a stale one back at
    every step. It is stamped from `token` here instead, and only after the ROW confirms the claim
    is still ours. Without that check the write is a full-column overwrite with no WHERE: a send
    that lost its lease came back and stomped the take-over send's committed links, then put its
    own dead token back on an idle session (deep-review round 4). `release_send` has always made
    the same check; this is the other half of it."""
    if claim_owner(db, session) != token:
        raise SendClaimLost(session.id)
    state = {k: v for k, v in state.items() if k != "sending"}
    state["sending"] = _stamp(token)
    _write_delivery(session, state)


def _write_delivery(session: DebugSession, state: dict) -> None:
    # 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. Both are true of a
        # send that is RUNNING too — every step commits its progress — so `sending` goes with them:
        # without it a reload or a second tab renders a live send as a failed one, and offers a
        # retry that is certain to be refused.
        "pending_files": not_uploaded(session),
        "images_pending": images_pending(session),
        "sending": sending(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 lock_user(db: Session, user_id: int) -> None:
    """Hold this user's own lock for the rest of the transaction. A session's row lock bounds what
    THAT session may keep; the byte cap below is the user's across all of them, and a lock on one
    session cannot serialize a check that reads them all — two uploads to two sessions both read the
    total before either commits, and both pass."""
    db.execute(select(func.pg_advisory_xact_lock(user_id)))


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 user and the session, and flushes; the caller commits.
    Refused while a send is under way or 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 both locks: two uploads at once cannot both pass the count (the session's), nor both
    # pass the user's byte cap from two different sessions (the user's). Always in this order.
    lock_user(db, session.user_id)
    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