siftlode/backend/app/logs.py

"""Structured logging for every Siftlode process — the ONE place logging is configured.

Before this module a single deployment printed FOUR different log formats: uvicorn's
`log_config.json` for the API, a hand-rolled StreamHandler in `main.py` for everything that did not
start under uvicorn (tests, scripts), `logging.basicConfig()` in the worker and in the download
child (no timestamp at all), and alembic's `fileConfig`. None of them carried a correlation id, and
uvicorn's access log printed the query string of every request (`?account=1`, `?source=feed`) — so
the logs were both unjoinable and leaky. `configure_logging()` replaces all four.

Event shape — the OpenTelemetry log data model, without the SDK; one JSON object per line:
    ts · level · sev · source · logger · event · msg · attrs
plus whatever correlation the context carries (`request_id`, `user_id`, `debug_session_id`,
`job_id`) and the app version. That shape is what E22.6's admin console filters on and what a Gitea
issue attachment will carry, so it is deliberately flat and stable.

Call sites stay ordinary stdlib logging:

    log = logging.getLogger("siftlode.downloads")
    log.info("job finished", extra=logs.event("download.job.done", job_id=7, bytes=n))

`event()` is optional — a bare `log.info("…")` still comes out well-formed — but the dotted code is
what makes a line queryable later, so add one wherever the line matters.

REDACTION IS APPLIED AT THE HANDLER, not at the call site: every message, every attr value and
every traceback passes `redact()` on the way out. That is the whole point — a secret cannot be
leaked by forgetting to redact it at one of ~125 call sites, and a new call site is safe by
default. Never logged, at any level: passwords, tokens (session, reset/verify, setup, share,
OAuth), cookie blobs, secret config values, and email addresses.
"""

import contextlib
import contextvars
import json
import logging
import logging.config
import re
import sys
import threading
import time
import uuid
from datetime import datetime, timezone
from typing import Any

from app.config import settings
from app.ratelimit import RateLimiter, client_key

# One level below DEBUG for the "Teljes / TRACE" diagnostics level (search strings, redacted request
# bodies). Registered on import so `logging.getLogger(...).log(logs.TRACE, …)` and a `LOG_LEVEL=TRACE`
# both work in every process, including one that never calls configure_logging (a test).
TRACE = 5
logging.addLevelName(TRACE, "TRACE")

# OpenTelemetry severity numbers for our level names. Kept beside the level because the number is
# what a log backend sorts and filters on; the name is for humans reading a terminal.
SEVERITY: dict[str, int] = {
    "TRACE": 1,
    "DEBUG": 5,
    "INFO": 9,
    "WARNING": 13,
    "ERROR": 17,
    "CRITICAL": 21,
}

# Which process this is. Set by configure_logging; every line carries it so one Loki/DB query can
# separate the API's view of a job from the worker's.
_source = "api"


# --- correlation context -------------------------------------------------------------------
# contextvars, like quota's actor attribution (app/quota.py): set at an entry point, read by the
# formatter, never threaded through call signatures. A child task copies the context at creation,
# so anything set before `await call_next(...)` is visible to the whole request.
_request_id: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "log_request_id", default=None
)
_user_id: contextvars.ContextVar[int | None] = contextvars.ContextVar("log_user_id", default=None)
_debug_session: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "log_debug_session", default=None
)
_job_id: contextvars.ContextVar[int | None] = contextvars.ContextVar("log_job_id", default=None)

_CONTEXT_VARS = {
    "request_id": _request_id,
    "user_id": _user_id,
    "debug_session_id": _debug_session,
    "job_id": _job_id,
}


class _RequestScope:
    """What the auth dependency learns about a request, for every line logged after it.

    A MUTABLE object behind a contextvar, not plain contextvars, because of how a request actually
    runs: FastAPI resolves a sync dependency (`auth.current_user`) in a threadpool on a COPY of the
    context, so a `ContextVar.set()` there is gone by the time the handler runs — measured: that is
    how the account never reached a handler's own lines. `log_requests` installs one scope per
    request; every copied context shares the same object, so what the dependency writes is seen by
    the handler and by the request line itself."""

    __slots__ = ("user_id", "level")

    def __init__(self) -> None:
        self.user_id: int | None = None
        # The user's consented diagnostics level as a logging number; None at the base level.
        self.level: int | None = None


_scope: contextvars.ContextVar[_RequestScope | None] = contextvars.ContextVar(
    "log_request_scope", default=None
)


def set_user(user_id: int | None, level: str = "INFO", until: datetime | None = None) -> None:
    """Attribute the rest of this request's log lines to `user_id`, and log them at the user's
    consented diagnostics `level` until `until` (E22 S5; "INFO" = no elevation). Called from the
    auth dependency — the one place the session is actually resolved into an account — rather than
    from the request middleware, which runs OUTSIDE the session middleware."""
    _user_id.set(user_id)
    scope = _scope.get()
    if scope is None:  # outside log_requests (a script, a unit test): this context only
        scope = _RequestScope()
        _scope.set(scope)
    elevated = level != "INFO" and until is not None and until > datetime.now(timezone.utc)
    scope.user_id = user_id
    scope.level = resolve_level(level) if elevated else None
    if user_id is not None:
        _track_elevation(user_id, scope.level, until)


def set_job(job_id: int | None) -> None:
    """Attribute this process's log lines to a download job — for the download CHILD, whose whole
    life is one job (the worker uses `bind()` per job instead, since it juggles several)."""
    _job_id.set(job_id)


@contextlib.contextmanager
def bind(**fields: Any):
    """Bind correlation fields for the duration of a block (a worker job, a scheduler run):

        with logs.bind(job_id=job.id):
            ...

    Unknown field names are ignored rather than raising — a caller binding a field this module does
    not know about is a typo, not a reason to take down a download."""
    tokens = [(_CONTEXT_VARS[k].set(v)) for k, v in fields.items() if k in _CONTEXT_VARS]
    keys = [k for k in fields if k in _CONTEXT_VARS]
    try:
        yield
    finally:
        for key, token in zip(keys, tokens):
            _CONTEXT_VARS[key].reset(token)


def event(code: str, **attrs: Any) -> dict[str, Any]:
    """`extra=` payload for a log call: a stable dotted event code plus structured attributes.

    Returns a plain dict so it composes with stdlib logging and costs nothing when the level is off
    (the dict is built either way; keep the attrs cheap — no DB reads inside the call)."""
    return {"event": code, "attrs": attrs}


# --- redaction -----------------------------------------------------------------------------
# Substring matching, not exact keys: the same secret travels as `password`, `current_password`,
# `new_password`, `smtp_password`, `plex_server_token`, `access_token`, … Matching on the PART is
# what makes a key we have never seen safe by default.
_SECRET_KEY_PARTS = (
    "password",
    "passphrase",
    "token",
    "secret",
    "cookie",
    "authorization",
    "auth_header",
    "ciphertext",
    "private_key",
    "api_key",
    "apikey",
    "credential",
    "session_key",
)

# Query/fragment parameters that carry a credential in a URL we log (OAuth `code`/`state`, our own
# `?token=` setup and share links, `#verify=` / `#reset=` fragments).
_SECRET_PARAM_RE = re.compile(
    r"\b((?:access_token|refresh_token|id_token|token|code|state|key|secret|password|pwd|sig|"
    r"signature|verify|reset|setup)=)([^&\s\"']+)",
    re.IGNORECASE,
)

# Deliberately loose: over-matching redacts a non-address that looks like one (harmless), while
# under-matching puts a real address in a log we attach to an issue (not harmless).
_EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")

REDACTED = "[redacted]"
# `extra={PERSIST: False}` keeps a line out of the `log_events` table while stdout still gets it —
# how `log_requests` caps anonymous traffic per client (app/logsink.py reads it).
PERSIST = "persist"
REDACTED_EMAIL = "[email]"


def redact_text(text: str) -> str:
    """Scrub a free-text string: credential-bearing URL params and email addresses."""
    scrubbed = _SECRET_PARAM_RE.sub(lambda m: f"{m.group(1)}{REDACTED}", text)
    return _EMAIL_RE.sub(REDACTED_EMAIL, scrubbed)


def redact(value: Any, *, key: str = "") -> Any:
    """Recursively scrub a structure destined for the log.

    A value whose KEY names a secret is dropped whole (we never log a prefix of a token — a prefix
    is still a leak and still useless). Everything else is scrubbed as text. Depth is bounded by
    the structures we actually log (attrs dicts), and non-JSON types fall through to `str`."""
    lowered = key.lower()
    if key and any(part in lowered for part in _SECRET_KEY_PARTS):
        return REDACTED
    if isinstance(value, dict):
        return {k: redact(v, key=str(k)) for k, v in value.items()}
    if isinstance(value, (list, tuple)):
        return [redact(v, key=key) for v in value]
    if isinstance(value, str):
        return redact_text(value)
    return value


# --- formatters ----------------------------------------------------------------------------
# Attributes every LogRecord carries; anything else a caller passed via `extra=` is a structured
# attribute and is merged into `attrs` (so `extra={"job_id": 7}` works as well as `event(...)`).
_RECORD_STANDARD = frozenset(
    vars(logging.LogRecord("", 0, "", 0, "", None, None)).keys()
    | {"message", "asctime", "taskName", "event", "attrs", PERSIST, "elevated"}
)


class ContextFilter(logging.Filter):
    """Copies the correlation contextvars onto the record. A Filter (not the formatter) so the
    values are captured at emit time, before a QueueHandler can move the record to another
    thread — the DB sink in S3 depends on that."""

    def filter(self, record: logging.LogRecord) -> bool:
        for name, var in _CONTEXT_VARS.items():
            value = var.get()
            if value is not None and not hasattr(record, name):
                setattr(record, name, value)
        scope = _scope.get()
        if scope is not None and scope.user_id is not None and not hasattr(record, "user_id"):
            record.user_id = scope.user_id
        return True


class LevelGate(logging.Filter):
    """The configured level, enforced at the HANDLER, with one exception: the lines of a request
    whose user raised their diagnostics level (E22 S5).

    The root logger's level cannot do this alone: it is one number for the whole process, and the
    point is to log ONE user's requests in detail while everyone else's stay at the base level. So
    the root is lowered to the lowest level any active elevation needs (`_track_elevation`), and
    this gate drops what only an elevation let through unless the line belongs to that elevated
    request. It also marks such a line `elevated`, which is what makes the `log_events` sink keep
    every line of that request, not just its warnings."""

    def filter(self, record: logging.LogRecord) -> bool:
        scope = _scope.get()
        if scope is not None and scope.level is not None:
            record.elevated = True
            return record.levelno >= min(scope.level, _base_level)
        return record.levelno >= _base_level


def _record_attrs(record: logging.LogRecord) -> dict[str, Any]:
    extra = {k: v for k, v in vars(record).items() if k not in _RECORD_STANDARD}
    attrs = dict(getattr(record, "attrs", None) or {})
    attrs.update({k: v for k, v in extra.items() if k not in _CONTEXT_VARS})
    return attrs


def _safe_message(record: logging.LogRecord) -> str:
    """`record.getMessage()` raises if the args don't match the format string. A broken log call
    must not take down the caller, so the failure is reported IN the line instead."""
    try:
        return record.getMessage()
    except Exception as exc:  # noqa: BLE001 — any formatting error, reported not raised
        return f"{record.msg!r} (log formatting failed: {exc})"


def record_fields(record: logging.LogRecord) -> dict[str, Any]:
    """The event as structured fields, REDACTED — the one definition both the stdout line and the
    `log_events` row are built from, so the two sinks can never disagree about what is safe to keep.
    Absent optional fields are left out; `ts` is an aware datetime."""
    fields: dict[str, Any] = {
        "ts": datetime.fromtimestamp(record.created, timezone.utc),
        "level": record.levelname,
        "sev": SEVERITY.get(record.levelname, 9),
        "source": _source,
        "logger": record.name,
        "msg": redact_text(_safe_message(record)),
    }
    code = getattr(record, "event", None)
    if code:
        fields["event"] = code
    for name in _CONTEXT_VARS:
        value = getattr(record, name, None)
        if value is not None:
            fields[name] = value
    attrs = _record_attrs(record)
    if attrs:
        fields["attrs"] = redact(attrs)
    if record.exc_info:
        fields["exc"] = redact_text(logging.Formatter().formatException(record.exc_info))
    elif record.exc_text:
        # A QueueHandler-style hand-off formats the traceback into exc_text and drops exc_info.
        fields["exc"] = redact_text(record.exc_text)
    fields["version"] = settings.app_version
    return fields


class JsonFormatter(logging.Formatter):
    """One JSON object per line — the shape Loki and the `log_events` table both ingest."""

    def format(self, record: logging.LogRecord) -> str:
        payload = record_fields(record)
        payload["ts"] = payload["ts"].isoformat(timespec="milliseconds")
        return json.dumps(payload, ensure_ascii=False, default=str)


class TextFormatter(logging.Formatter):
    """Human-readable single line for a terminal (`LOG_FORMAT=text`): the same content, ordered for
    reading. Redaction applies here too — a dev log is still a log someone pastes into an issue."""

    def format(self, record: logging.LogRecord) -> str:
        stamp = datetime.fromtimestamp(record.created, timezone.utc).strftime("%H:%M:%S.%f")[:-3]
        head = f"{stamp} {record.levelname:<5} [{record.name}]"
        rid = getattr(record, "request_id", None)
        if rid:
            head += f" {rid}"
        code = getattr(record, "event", None)
        line = f"{head} {code + ' — ' if code else ''}{redact_text(_safe_message(record))}"
        attrs = _record_attrs(record)
        if attrs:
            line += f" {json.dumps(redact(attrs), ensure_ascii=False, default=str)}"
        if record.exc_info:
            line += "\n" + redact_text(self.formatException(record.exc_info))
        return line


# --- configuration -------------------------------------------------------------------------
# Third-party loggers that are noisy at our own level. Pinned explicitly so the diagnostics knob can
# lower the ROOT level to DEBUG/TRACE without drowning the output in SQL statements and HTTP chatter
# — raising THESE is a deliberate, separate act (E22.6's console).
_THIRD_PARTY_LEVELS = {
    "sqlalchemy": "WARNING",
    "sqlalchemy.engine": "WARNING",
    "apscheduler": "WARNING",
    "urllib3": "WARNING",
    "httpx": "WARNING",
    "httpcore": "WARNING",
    "asyncio": "WARNING",
    "multipart": "WARNING",
    "PIL": "WARNING",
    "alembic": "INFO",
    # uvicorn's own startup/shutdown/WebSocket lines are worth keeping; its ACCESS log is not —
    # it is one line per request with the query string in it, replaced by `log_requests` below.
    "uvicorn": "INFO",
    "uvicorn.error": "INFO",
    "uvicorn.access": "CRITICAL",
}


def resolve_level(name: str) -> int:
    """Level name → number. Both ways in (the `LOG_LEVEL` env and the `log_level` admin knob) are
    already held to the `LogLevel` choices, so an unknown name here is a programming error."""
    value = logging.getLevelName(name)
    if not isinstance(value, int):
        raise ValueError(f"unknown log level {name!r}")
    return value


def configure_logging(source: str) -> None:
    """Configure logging for THIS process. Call once, first thing, in every entry point:
    `main.py` (api), `worker.main`, `downloads/runner.main`, `alembic/env.py`.

    `source` names the process in every line. The level starts at `LOG_LEVEL` (the bootstrap default);
    the runtime admin knob goes through `set_level()` instead, so changing it needs no restart.

    Our loggers deliberately keep `propagate=True` and own no handler: everything reaches the single
    root StreamHandler. That is what keeps `caplog` working in tests, and it means a new logger name
    is configured the moment it is created."""
    global _source
    _source = source
    formatter = settings.log_format
    logging.config.dictConfig(
        {
            "version": 1,
            # Never disable loggers created at import time (module-level `getLogger` calls run
            # before this function does).
            "disable_existing_loggers": False,
            "formatters": {
                "json": {"()": f"{__name__}.JsonFormatter"},
                "text": {"()": f"{__name__}.TextFormatter"},
            },
            "filters": {
                "context": {"()": f"{__name__}.ContextFilter"},
                "gate": {"()": f"{__name__}.LevelGate"},
            },
            "handlers": {
                "stdout": {
                    "class": "logging.StreamHandler",
                    "stream": sys.stdout,
                    "formatter": formatter,
                    "filters": ["context", "gate"],
                }
            },
            "root": {"level": resolve_level(settings.log_level), "handlers": ["stdout"]},
            "loggers": {name: {"level": lvl} for name, lvl in _THIRD_PARTY_LEVELS.items()},
        }
    )
    set_level(settings.log_level)


def set_level(name: str) -> None:
    """Change the running process's level (the `log_level` admin knob). Third-party loggers keep
    their pinned levels, so turning on DEBUG shows OUR detail, not SQLAlchemy's."""
    global _base_level
    _base_level = resolve_level(name)
    _apply_root_level(force=True)


def current_level() -> str:
    """The configured level — not the root logger's, which an active elevation may have lowered."""
    return logging.getLevelName(_base_level or logging.getLogger().getEffectiveLevel())


# --- per-user elevation (E22 S5) -----------------------------------------------------------
# The base level (`LOG_LEVEL`, then the admin knob). NOTSET until configure_logging runs — a unit
# test, a script — so the gate lets everything through there, as it did before the gate existed.
_base_level = logging.NOTSET
# user id → (level number, expiry) for every elevation this process has seen and not yet expired.
_elevations: dict[int, tuple[int, datetime]] = {}
_elevations_lock = threading.Lock()
_root_applied: int | None = None


def _track_elevation(user_id: int, level: int | None, until: datetime | None) -> None:
    with _elevations_lock:
        if level is None or until is None:
            _elevations.pop(user_id, None)
        else:
            _elevations[user_id] = (level, until)
        active = bool(_elevations)
    if active or _root_applied != _base_level:
        _apply_root_level()


def _apply_root_level(force: bool = False) -> None:
    """Root level = the lowest of the base and every unexpired elevation (expired ones are pruned
    here). `setLevel` clears every logger's level cache, so it runs only when the number changes."""
    global _root_applied
    if _base_level == logging.NOTSET:
        return  # not configured: the root keeps whatever the process set up
    now = datetime.now(timezone.utc)
    with _elevations_lock:
        for user_id in [u for u, (_, until) in _elevations.items() if until <= now]:
            del _elevations[user_id]
        lowest = min([_base_level, *(lvl for lvl, _ in _elevations.values())])
    if force or lowest != _root_applied:
        logging.getLogger().setLevel(lowest)
        _root_applied = lowest


def current_source() -> str:
    """The process name `configure_logging` was given (api, worker, runner, alembic)."""
    return _source


# --- request logging -----------------------------------------------------------------------
# A client may supply its own request id so a browser log line and a server line join up (E22 S4).
# Accepted only in this shape: it ends up in a log field, and an unbounded header would be a way to
# write arbitrary text into the log.
_REQUEST_ID_RE = re.compile(r"^[A-Za-z0-9._-]{8,64}$")

# Paths whose success is logged at DEBUG rather than INFO: the container healthcheck runs every 15s
# (5.7k lines/day saying nothing) and hashed static assets are served by the hundred per page load.
# A FAILURE on either still logs at WARNING/ERROR — this lowers the floor, it does not hide.
_QUIET_PREFIXES = ("/healthz", "/assets/", "/static/")

_request_log = logging.getLogger("siftlode.http")


async def log_requests(request, call_next):
    """The single access-log line, replacing uvicorn's.

    What changes versus `uvicorn.access`: the ROUTE TEMPLATE instead of the concrete path (so
    `/api/videos/{video_id}` groups), never the query string (it carried `?account=`, and at TRACE
    the redacted query belongs in attrs, not in the URL), a request id echoed back in
    `X-Request-Id` (with the template in `X-Route`), the duration, and the account — read from `request.state`, which the auth
    dependency fills, because this middleware sits outside both the session middleware and the
    downstream task's context."""
    incoming = request.headers.get("x-request-id", "")
    rid = incoming if _REQUEST_ID_RE.match(incoming) else uuid.uuid4().hex[:16]
    token = _request_id.set(rid)
    scope = _RequestScope()
    scope_token = _scope.set(scope)
    request.state.request_id = rid
    started = time.perf_counter()
    try:
        response = await call_next(request)
    except Exception:
        # The unhandled-exception path: FastAPI's ServerErrorMiddleware re-raises after its own
        # 500, so without this the ONE request that most needs a log line is the one with no
        # access-log entry at all.
        _request_log.exception(
            "request failed",
            extra={
                **event(
                    "http.request",
                    method=request.method,
                    path=_route_template(request),
                    duration_ms=round((time.perf_counter() - started) * 1000, 1),
                ),
                **_request_owner(request, logging.ERROR),
            },
        )
        _scope.reset(scope_token)
        _request_id.reset(token)
        raise
    duration_ms = round((time.perf_counter() - started) * 1000, 1)
    status = response.status_code
    quiet = request.url.path.startswith(_QUIET_PREFIXES)
    if status >= 500:
        level = logging.ERROR
    elif status >= 400:
        level = logging.WARNING
    else:
        level = logging.DEBUG if quiet else logging.INFO
    attrs: dict[str, Any] = {
        "method": request.method,
        "path": _route_template(request),
        "status": status,
        "duration_ms": duration_ms,
    }
    # Teljes (TRACE, the user's or the whole process's): the query string too — redacted, and in
    # attrs only, never in the message.
    if request.url.query and _traced(scope):
        attrs["query"] = redact_text(request.url.query)
    _request_log.log(
        level,
        "%s %s %s",
        request.method,
        _route_template(request),
        status,
        extra={**event("http.request", **attrs), **_request_owner(request, level)},
    )
    response.headers["X-Request-Id"] = rid
    # The template, so the browser's copy of this line groups the same way (frontend lib/http.ts).
    # Only a MATCHED route's: a template never carries a path parameter's value (a share token stays
    # `{token}`), while the raw-path fallback of an unmatched request could — the browser then
    # templates the path itself.
    matched = _matched_route(request)
    if matched:
        response.headers["X-Route"] = matched
    _scope.reset(scope_token)
    _request_id.reset(token)
    return response


def _traced(scope: _RequestScope) -> bool:
    """Whether this request is logged at TRACE — its user's consent, or the process's base level."""
    if scope.level is not None and scope.level <= TRACE:
        return True
    return _base_level != logging.NOTSET and _base_level <= TRACE


# Anonymous request lines kept per client per minute. Signed-in traffic is not capped: it is a known
# account and exactly what an admin needs to see. Anonymous traffic is where the volume can come from
# nobody in particular — a scanner walking `/wp-login.php`, a bot on a share link — and each of its
# 404s is a WARNING that would otherwise land in `log_events` without limit. The client key is an
# in-memory counter key only; no address is ever written to a log line or a row.
ANONYMOUS_LINES_PER_MINUTE = 60
_anonymous_lines = RateLimiter(ANONYMOUS_LINES_PER_MINUTE, 60.0)


def _request_owner(request, level: int) -> dict[str, Any]:
    """`extra=` fields naming who the request belongs to, and whether its line may be stored. The
    account comes from `request.state` (filled by the auth dependency), so it lands in the line's
    top-level `user_id` — the correlation field — rather than in `attrs`."""
    user_id = getattr(request.state, "user_id", None)
    if user_id is not None:
        return {"user_id": user_id}
    if level < logging.INFO:
        return {}  # a quiet line is never stored anyway — don't spend the client's allowance on it
    return {} if _anonymous_lines.allow(client_key(request)) else {PERSIST: False}


def _matched_route(request) -> str | None:
    """The matched route's template (`/api/videos/{video_id}`), or None when nothing matched."""
    return getattr(request.scope.get("route"), "path", None)


def _route_template(request) -> str:
    """The matched route's template, else the raw path for a 404 or a static file. Never the query
    string."""
    return _matched_route(request) or request.url.path