diff --git a/accounts/api_accounts.py b/accounts/api_accounts.py index 9a5cf2f..96f3276 100644 --- a/accounts/api_accounts.py +++ b/accounts/api_accounts.py @@ -12,6 +12,7 @@ from fastapi.concurrency import run_in_threadpool from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession +from core import audit from data import climate from data import grid from notifications import push @@ -110,6 +111,8 @@ async def create_subscription( session.add(sub) await session.commit() await session.refresh(sub) + audit.log_activity("sub.create", {"user_id": str(user.id), "sub_id": sub.id, + "cell_id": cell_id, "kind": body.kind}) return sub @@ -121,10 +124,13 @@ async def update_subscription( session: AsyncSession = Depends(get_async_session), ): sub = await _owned_subscription(session, sub_id, user) - for key, value in body.model_dump(exclude_unset=True).items(): + changed = body.model_dump(exclude_unset=True) + for key, value in changed.items(): setattr(sub, key, value) await session.commit() await session.refresh(sub) + audit.log_activity("sub.update", {"user_id": str(user.id), "sub_id": sub.id, + "changed": list(changed.keys())}) return sub @@ -135,8 +141,10 @@ async def delete_subscription( session: AsyncSession = Depends(get_async_session), ): sub = await _owned_subscription(session, sub_id, user) + kind = sub.kind await session.delete(sub) await session.commit() + audit.log_activity("sub.delete", {"user_id": str(user.id), "sub_id": sub_id, "kind": kind}) # --- notifications ----------------------------------------------------------- @@ -254,6 +262,8 @@ async def push_subscribe( ) ) await session.commit() + audit.log_activity("push.subscribe", {"user_id": str(user.id), + "refreshed": existing is not None}) return {"ok": True} @@ -275,6 +285,7 @@ async def push_unsubscribe( if row is not None: await session.delete(row) await session.commit() + audit.log_activity("push.unsubscribe", {"user_id": str(user.id)}) @router.post("/push/test", status_code=202) diff --git a/accounts/users.py b/accounts/users.py index a08fb46..8967126 100644 --- a/accounts/users.py +++ b/accounts/users.py @@ -19,6 +19,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from accounts.db import get_async_session from accounts.models import AccessToken, User +from core import audit from notifications import mailer BASE = "/" + os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/") @@ -77,11 +78,19 @@ class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]): raise InvalidPasswordException(reason="Password is too long.") async def on_after_register(self, user: User, request: "Request | None" = None) -> None: + audit.log_activity("auth.register", {"user_id": str(user.id)}) # Kick off the same token-generation path a manual "resend verification" # request would use, so there is exactly one place that builds the email # (on_after_request_verify below) rather than two copies drifting apart. await self.request_verify(user, request) + async def on_after_login(self, user: User, request: "Request | None" = None, + response=None) -> None: + audit.log_activity("auth.login", {"user_id": str(user.id), "verified": user.is_verified}) + + async def on_after_verify(self, user: User, request: "Request | None" = None) -> None: + audit.log_activity("auth.verify", {"user_id": str(user.id)}) + async def on_after_request_verify( self, user: User, token: str, request: "Request | None" = None ) -> None: diff --git a/core/audit.py b/core/audit.py index 8af0502..24035a4 100644 --- a/core/audit.py +++ b/core/audit.py @@ -30,6 +30,7 @@ AUDIT_DIR = os.path.join(LOG_ROOT, "audit") ERROR_DIR = os.path.join(LOG_ROOT, "errors") ACCESS_DIR = os.path.join(LOG_ROOT, "access") HEARTBEAT_DIR = os.path.join(LOG_ROOT, "heartbeat") +ACTIVITY_DIR = os.path.join(LOG_ROOT, "activity") _lock = threading.Lock() @@ -69,6 +70,17 @@ def log_event(tag: str, record: dict) -> dict: return rec +def log_activity(tag: str, record: dict) -> dict: + """Record a non-error application event — auth, subscriptions, notification + delivery, app lifecycle, slow requests — to a dedicated *activity* stream, + kept separate from the errors folder so these never pollute error dashboards. + ``tag`` names the event (e.g. ``"auth.login"``, ``"notify.pass"``). Carries no + PII: user ids are UUIDs, and no emails/request bodies/push endpoints. Best-effort.""" + rec = {"ts": _now_iso(), "tag": tag, "run_id": current_run_id.get(), **record} + _append(os.path.join(ACTIVITY_DIR, f"activity-{_today()}.jsonl"), rec) + return rec + + def log_access(record: dict) -> None: """Append one request line (client IP, method, path, status) to the daily access log, so request/IP data is retained for later analysis. Best-effort; never raises diff --git a/notifications/mailer.py b/notifications/mailer.py index 2194262..dcbb52d 100644 --- a/notifications/mailer.py +++ b/notifications/mailer.py @@ -27,6 +27,8 @@ import ssl from email.message import EmailMessage from email.utils import formataddr, formatdate, make_msgid, parseaddr +from core import audit + log = logging.getLogger("thermograph.mail") # console -> log the message, send nothing (default; dev + tests) @@ -91,6 +93,7 @@ def send(to: str, subject: str, text: str, html: "str | None" = None) -> bool: if BACKEND != "smtp": # console (the default): prove the flow end-to-end without sending. log.info("mail[console] to=%s subject=%s\n%s", to, subject, text) + audit.log_activity("mail.send", {"ok": True, "backend": "console"}) return True try: @@ -100,7 +103,9 @@ def send(to: str, subject: str, text: str, html: "str | None" = None) -> bool: if SMTP_USER: smtp.login(SMTP_USER, SMTP_PASSWORD) smtp.send_message(msg) + audit.log_activity("mail.send", {"ok": True, "backend": "smtp"}) return True except Exception: # noqa: BLE001 - deliverability is never worth a 500 log.exception("mail: send failed to=%s host=%s:%s", to, SMTP_HOST, SMTP_PORT) + audit.log_activity("mail.send", {"ok": False, "backend": "smtp"}) return False diff --git a/notifications/notify.py b/notifications/notify.py index f0f9d76..509e691 100644 --- a/notifications/notify.py +++ b/notifications/notify.py @@ -161,7 +161,8 @@ def _send_discord_job(job: tuple[str, str, str, str]) -> None: """Best-effort Discord DM of one gathered job. Never raises (send_dm itself doesn't). Runs off the notifier thread's send pool — no DB session here.""" discord_id, title, body, link = job - discord.send_dm(discord_id, title, body, link) + ok = discord.send_dm(discord_id, title, body, link) + audit.log_activity("notif.discord", {"ok": bool(ok)}) # How many push/Discord sends the notifier fans out at once. These are pure @@ -302,6 +303,9 @@ def _process_cell(session, cell_id, subs, today, now, archive_budget): continue sub.last_notified_at = now created += 1 + audit.log_activity("notif.emit", {"user_id": str(sub.user_id), "sub_id": sub.id, + "metric": metric, "direction": direction, + "kind": sub.kind, "percentile": graded["percentile"]}) # Gather who to reach over Web Push and Discord DM (the in-app row above # is the record of truth for the bell; these are side-channels that must # never break it) — the actual sends happen after commit, see below. @@ -345,9 +349,12 @@ def _cleanup_sessions(session) -> None: def run_pass() -> int: """One full evaluation sweep over all active subscriptions. Returns the number of notifications created.""" + t0 = time.perf_counter() now = time.time() today = datetime.date.today() created = 0 + subs: list = [] + by_cell: dict[str, list] = {} archive_budget = [MAX_ARCHIVE_FETCHES_PER_PASS] # missing archives to fetch this pass with sync_session_maker() as session: subs = session.execute( @@ -365,6 +372,10 @@ def run_pass() -> int: _cleanup_sessions(session) # opportunistic expired-token sweep except Exception: # noqa: BLE001 session.rollback() + audit.log_activity("notify.pass", { + "subs_evaluated": len(subs), "cells": len(by_cell), "created": created, + "archives_fetched": MAX_ARCHIVE_FETCHES_PER_PASS - archive_budget[0], + "duration_ms": round((time.perf_counter() - t0) * 1000.0, 1)}) return created diff --git a/tests/conftest.py b/tests/conftest.py index f615a65..fb76f4f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -48,6 +48,8 @@ audit.ERROR_DIR = os.path.join(_TMP, "logs", "errors") # The request-logging middleware runs under TestClient too; keep its access log # out of the repo's logs/ so self-test traffic never pollutes the live monitor. audit.ACCESS_DIR = os.path.join(_TMP, "logs", "access") +audit.ACTIVITY_DIR = os.path.join(_TMP, "logs", "activity") +audit.HEARTBEAT_DIR = os.path.join(_TMP, "logs", "heartbeat") def _years_before(d: datetime.date, years: int) -> datetime.date: diff --git a/tests/notifications/test_discord_dm.py b/tests/notifications/test_discord_dm.py index 555fd8b..f3e97ba 100644 --- a/tests/notifications/test_discord_dm.py +++ b/tests/notifications/test_discord_dm.py @@ -85,7 +85,7 @@ def _fake_session(user): def _sub(): - return types.SimpleNamespace(user_id="u1", lat=47.6, lon=-122.3) + return types.SimpleNamespace(id="s1", user_id="u1", lat=47.6, lon=-122.3) def test_discord_job_gathers_linked_opted_in_user(monkeypatch): diff --git a/web/app.py b/web/app.py index 29c1062..0de9cf8 100644 --- a/web/app.py +++ b/web/app.py @@ -15,7 +15,7 @@ from fastapi import APIRouter, FastAPI, HTTPException, Query, Request, Response from fastapi.concurrency import run_in_threadpool from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.gzip import GZipMiddleware -from fastapi.responses import RedirectResponse +from fastapi.responses import JSONResponse, RedirectResponse from accounts import api_accounts from api import content_routes @@ -167,7 +167,11 @@ async def _lifespan(app): # Same leader gate as the notifier above — the recurring worker jobs # (city warming, IndexNow) must likewise run in exactly one process. scheduler.start() + # Deploy/restart marker: one line correlates a metric shift with a boot. + audit.log_activity("app.start", {"version": app.version, "role": ROLE, "base": BASE, + "notifier": _should_run_notifier(), "pid": os.getpid()}) yield + audit.log_activity("app.stop", {"role": ROLE, "pid": os.getpid()}) notify.stop() scheduler.stop() await _frontend_client.aclose() @@ -202,6 +206,25 @@ if _cors_origins: ) +# Log a structured line for every unhandled 500. Without an exception handler the +# route exception propagates out past the request middleware (call_next re-raises), +# so a 500 is otherwise invisible in both the access log and metrics — this makes +# it visible and countable (tag="http.error" in the errors stream). +SLOW_REQUEST_MS = 2000.0 # http.slow: log any request slower than this + + +@app.exception_handler(Exception) +async def _log_unhandled_exception(request: Request, exc: Exception) -> Response: + try: + audit.log_event("http.error", { + "phase": "http", "path": request.url.path, "method": request.method, + "cat": metrics.classify_inbound(request.url.path, BASE), + "exc_type": type(exc).__name__, "error": repr(exc)[:500]}) + except Exception: # noqa: BLE001 - logging must never mask the original error + pass + return JSONResponse({"detail": "Internal server error."}, status_code=500) + + @app.get("/healthz") def healthz(): """Liveness probe: the process is up and serving. Deliberately does no I/O (no @@ -246,7 +269,9 @@ async def revalidate_static(request, call_next): request answered with an empty 304 instead of a full re-download — page-to-page navigation stops re-transferring the JS/CSS/HTML while still picking up every deploy immediately (no "my change isn't showing" bugs).""" + t0 = time.perf_counter() response = await call_next(request) + dur_ms = (time.perf_counter() - t0) * 1000.0 path = request.url.path try: cat = metrics.classify_inbound(path, BASE) @@ -260,6 +285,14 @@ async def revalidate_static(request, call_next): await run_in_threadpool(audit.log_access, { "ip": _client_ip(request), "method": request.method, "path": path, "status": response.status_code, "cat": cat}) + # Threshold-gated slow-request line: RunAudit only times the 7 graded + # endpoints, so this is the only latency signal for auth/account/content + # routes. Kept rare (>2s) so it never becomes per-request hot-path noise. + # Also a blocking file write, so keep it off the event loop. + if dur_ms > SLOW_REQUEST_MS and cat != "static": + await run_in_threadpool(audit.log_activity, "http.slow", { + "path": path, "method": request.method, "cat": cat, + "status": response.status_code, "duration_ms": round(dur_ms, 1)}) except Exception: # noqa: BLE001 - never let instrumentation break a response pass pages = (BASE, f"{BASE}/", f"{BASE}/calendar", f"{BASE}/day", f"{BASE}/score",