diff --git a/backend/.dockerignore b/backend/.dockerignore new file mode 100644 index 0000000..7e8c7c1 --- /dev/null +++ b/backend/.dockerignore @@ -0,0 +1,10 @@ +# Local virtualenvs are disposable dev state (see CLAUDE.md), but without this +# file `COPY . /app/` on a dev checkout ships ~400MB of site-packages the image +# already installs via pip — and the chown -R layer then duplicates it AGAIN. +# CI contexts never contain a .venv, so this only protects `:local` dev builds. +# Do NOT add daemon/ here: the daemon-builder stage COPYs it from this same +# context. +.venv* +__pycache__/ +*.py[co] +.pytest_cache/ diff --git a/backend/Dockerfile b/backend/Dockerfile index e5ed738..5adaa68 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -1,5 +1,25 @@ # Thermograph backend: FastAPI API + accounts/notifications + the SSR content # JSON API frontend consumes. Split from the monorepo (repo-split Stage 7). + +# thermograph-daemon (daemon/): the Go process that owns the Discord gateway +# websocket and the recurring-job timers, calling back into this app's +# /internal/* routes for anything that needs data. It is built INTO this image +# on purpose: daemon and backend share the internal API contract, so shipping +# one image (compose picks the process per-service) makes version skew between +# them impossible. CGO_ENABLED=0 gives a fully static binary that drops into +# the python:3.12-slim final stage with no runtime deps; go.mod/go.sum are +# copied and downloaded before the sources so the Go dep layer caches across +# daemon code-only changes, same reasoning as the pip layer below. +FROM golang:1.26 AS daemon-builder +WORKDIR /src +COPY daemon/go.mod daemon/go.sum ./ +RUN go mod download +COPY daemon/ ./ +# -trimpath keeps the build reproducible (identical sources -> identical +# binary), which is what lets the final stage's COPY layer cache-hit when only +# Python code changed. +RUN CGO_ENABLED=0 go build -trimpath -o /out/thermograph-daemon . + FROM python:3.12-slim # curl is only for the container HEALTHCHECK below. Everything Python needs @@ -14,6 +34,12 @@ RUN apt-get update \ COPY requirements.txt /tmp/requirements.txt RUN pip install --no-cache-dir -r /tmp/requirements.txt +# The daemon binary lands before the app tree: it changes far less often than +# the Python code, so this layer usually cache-hits and only the COPY below +# rebuilds. NOT the entrypoint — deploy/entrypoint.sh stays that; compose +# selects this binary per-service for the daemon container. +COPY --from=daemon-builder /out/thermograph-daemon /usr/local/bin/thermograph-daemon + COPY . /app/ RUN chmod +x /app/deploy/entrypoint.sh diff --git a/backend/api/internal_routes.py b/backend/api/internal_routes.py new file mode 100644 index 0000000..9ecded4 --- /dev/null +++ b/backend/api/internal_routes.py @@ -0,0 +1,158 @@ +"""Internal control surface for the Go daemon (backend/daemon/). + +The gateway bot and the recurring worker jobs moved out of this process into +`thermograph-daemon` — Go owns only the long-lived stateful I/O (the Discord +websocket, RESUME/heartbeat/backoff, interval timers). Anything that needs +*data* calls back in here, because grading depends on polars + the parquet +cache: reimplementing it out-of-process would let the bot's grades drift from +the API's, and the slash-command path deliberately shares one grade builder so +the two never drift. These routes hand the daemon finished answers, keeping it +entirely out of the grading contract. + +Not publicly reachable: the deploy Caddyfile routes only /api/*, /digest and +/discord/interactions to this backend, so /internal/* never resolves through +the public domain. The shared-secret header is defence in depth on top of +that, not the only control — and it fails closed: with no token configured +the whole surface answers 404, never "open". + +Deliberately NOT under BASE (same posture as /healthz): the daemon targets +THERMOGRAPH_API_BASE_INTERNAL directly and shouldn't need to know +THERMOGRAPH_BASE to construct a path. +""" +import hashlib +import hmac +import os +import threading + +from fastapi import APIRouter, Depends, Header, HTTPException +from pydantic import BaseModel + +_TOKEN_ENV = "THERMOGRAPH_INTERNAL_TOKEN" +_HEADER = "X-Thermograph-Internal-Token" + +# Domain-separation label for the derived token. MUST match the Go daemon's +# internal/config constant byte-for-byte or every callback 401s. +_DERIVE_LABEL = b"thermograph/internal-api/v1" + + +def resolve_internal_token() -> str: + """The shared secret guarding /internal/*, or "" if it can't be established. + + An explicit THERMOGRAPH_INTERNAL_TOKEN always wins. Otherwise it is DERIVED + from THERMOGRAPH_AUTH_SECRET, which every environment already provisions, so + standing up the daemon needs no new vault entry and no operator step -- one + less secret to rotate, leak, or forget on a new host. + + HMAC with a distinct label rather than reusing AUTH_SECRET directly: that is + ordinary domain separation, so a leak of this token can't be replayed as the + session-signing key it was derived from, and the derivation is one-way. + + Deliberately reads THERMOGRAPH_AUTH_SECRET from the environment rather than + users.py's effective value: users.py falls back to a random per-process + secret when the env var is unset, and two processes would then derive two + different tokens and fail every call in a way that looks like a bug rather + than missing config. Unset env var => "" => the surface stays closed.""" + explicit = os.environ.get(_TOKEN_ENV, "").strip() + if explicit: + return explicit + auth_secret = os.environ.get("THERMOGRAPH_AUTH_SECRET", "").strip() + if not auth_secret: + return "" + return hmac.new(auth_secret.encode(), _DERIVE_LABEL, hashlib.sha256).hexdigest() + + +def _require_internal_token( + x_thermograph_internal_token: str = Header(default=""), +) -> None: + """Gate every internal route on the shared secret. + + Read from the environment per request (not at import) so the fail-closed + behaviour is a property of the running config, not of import order — and so + tests can exercise both states without re-importing the app. An unset/empty + token means the surface was never provisioned: 404 (the route doesn't + exist), matching the metrics endpoint's don't-reveal posture. A present but + wrong header is a real auth failure: 401. compare_digest, not ==, so the + comparison doesn't leak the token's prefix through response timing.""" + token = resolve_internal_token() + if not token: + raise HTTPException(status_code=404) + if not hmac.compare_digest(x_thermograph_internal_token, token): + raise HTTPException(status_code=401, detail="bad internal token") + + +# include_in_schema=False: an internal surface has no business in the public +# OpenAPI document, same as /api/v2/metrics. +router = APIRouter(prefix="/internal", include_in_schema=False, + dependencies=[Depends(_require_internal_token)]) + + +# Every handler below is a plain `def`, not `async def`, on purpose: they all do +# blocking work (sync parquet/DB reads, grading math, upstream HTTP fetches). +# FastAPI runs sync handlers on its threadpool, so a slow grade or a long warm +# run never stalls the event loop the rest of the API is serving on — the same +# reason web/app.py wraps discord_interactions.handle in run_in_threadpool. + + +class _GradeBody(BaseModel): + query: str + + +@router.post("/discord/grade") +def discord_grade(body: _GradeBody) -> dict: + """The gateway-ready reply for one bot mention/DM: the same embed the /grade + slash command builds, via the shared discord_interactions._grade_message — + reusing one builder is what keeps the two paths from ever drifting. Local + import for the same reason the in-process gateway bot used one: don't drag + the interactions module's transitive deps into contexts that only mount the + router.""" + from notifications import discord_interactions + data = discord_interactions._grade_message(body.query) + # Gateway messages can't be ephemeral — the flag is interactions-only, and + # Discord rejects it outside an interaction response. + data.pop("flags", None) + return data + + +# One in-flight run per job, enforced here rather than trusted to the daemon's +# timers: warm_cities spends the Open-Meteo quota, so a stacked second run +# double-spends it, and a slow run must never let invocations pile up holding +# requests open. Non-blocking acquire => the loser answers 409 immediately. +_JOB_LOCKS: dict[str, threading.Lock] = { + "warm-cities": threading.Lock(), + "indexnow": threading.Lock(), +} + + +def _run_exclusive(job: str, fn) -> dict: + lock = _JOB_LOCKS[job] + if not lock.acquire(blocking=False): + raise HTTPException(status_code=409, detail=f"{job} is already running") + try: + fn() + finally: + lock.release() + return {"ok": True} + + +@router.post("/jobs/warm-cities") +def jobs_warm_cities() -> dict: + """(Re-)warm the curated city set. The daemon owns the cadence (default + daily — every already-cached city is a cheap skip); this endpoint just runs + one pass. Local import: warm_cities is a script-style top-level module, so + only the process that actually runs the job pays its import cost.""" + def _run(): + import warm_cities + warm_cities.main() + return _run_exclusive("warm-cities", _run) + + +@router.post("/jobs/indexnow") +def jobs_indexnow() -> dict: + """Check whether the indexable URL set changed and, if so, ping IndexNow. + submit_if_changed() is a cheap no-op when nothing changed, which is why the + daemon can tick this more often than warm-cities without spending anything + on most runs.""" + def _run(): + import indexnow + indexnow.submit_if_changed() + return _run_exclusive("indexnow", _run) diff --git a/backend/daemon/README.md b/backend/daemon/README.md new file mode 100644 index 0000000..c653dbc --- /dev/null +++ b/backend/daemon/README.md @@ -0,0 +1,50 @@ +# thermograph-daemon + +Single Go binary that owns the backend's long-lived stateful I/O: the Discord +gateway connection (IDENTIFY/RESUME/heartbeat/reconnect backoff) and the +recurring-job timers (warm-cities, IndexNow). It replaces the two in-process +background jobs `web/app.py` used to run under leader election — one daemon +replica makes the single-connection invariant a deployment fact instead of a +runtime election. + +## The Go/Python split + +Go here owns **no** climate or grading logic. Grading depends on polars and the +parquet cache, and the slash-command path deliberately shares one grade builder +with the API so bot grades and API grades never drift. So anything data-shaped +is a POST back into the Python app's internal-only routes: + +| Route | Purpose | +| --- | --- | +| `POST /internal/discord/grade` | `{"query": "phoenix"}` → gateway-ready Discord message JSON, relayed to Discord verbatim | +| `POST /internal/jobs/warm-cities` | trigger the warm-cities job | +| `POST /internal/jobs/indexnow` | trigger the IndexNow check-and-submit | + +Every request carries `X-Thermograph-Internal-Token`. Caddy never routes +`/internal/*` to the backend, so the token is defence in depth, not the only +control. If `THERMOGRAPH_INTERNAL_TOKEN` is unset, the Python side disables the +routes and this daemon refuses to start (fail closed on both ends). + +## Configuration (env) + +| Var | Required | Meaning | +| --- | --- | --- | +| `THERMOGRAPH_INTERNAL_TOKEN` | yes | shared secret for the internal routes | +| `THERMOGRAPH_API_BASE_INTERNAL` | yes | backend base URL, e.g. `http://web:8137` | +| `THERMOGRAPH_DISCORD_BOT` | no | `1`/`true`/`yes`/`on` enables the gateway | +| `THERMOGRAPH_DISCORD_BOT_TOKEN` | no | bot token (gateway stays off without it) | +| `THERMOGRAPH_WARM_CITIES_INTERVAL_HOURS` | no | default 24; malformed → default | +| `THERMOGRAPH_INDEXNOW_INTERVAL_HOURS` | no | default 6; malformed → default | + +## Run locally + +```sh +cd backend/daemon +go build ./... && go vet ./... && go test ./... + +THERMOGRAPH_INTERNAL_TOKEN=dev-token \ +THERMOGRAPH_API_BASE_INTERNAL=http://localhost:8137 \ +go run . +``` + +Deployed as `/usr/local/bin/thermograph-daemon` in the backend image. diff --git a/backend/daemon/go.mod b/backend/daemon/go.mod new file mode 100644 index 0000000..e8a6593 --- /dev/null +++ b/backend/daemon/go.mod @@ -0,0 +1,5 @@ +module thermograph/daemon + +go 1.26 + +require github.com/coder/websocket v1.8.15 diff --git a/backend/daemon/go.sum b/backend/daemon/go.sum new file mode 100644 index 0000000..4e0a48d --- /dev/null +++ b/backend/daemon/go.sum @@ -0,0 +1,2 @@ +github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA= +github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= diff --git a/backend/daemon/internal/apiclient/client.go b/backend/daemon/internal/apiclient/client.go new file mode 100644 index 0000000..ba998c4 --- /dev/null +++ b/backend/daemon/internal/apiclient/client.go @@ -0,0 +1,107 @@ +// Package apiclient is the daemon's ONLY channel back into the Python app. +// The daemon owns long-lived I/O (gateway socket, timers) and nothing else: +// grading depends on polars + the parquet cache, and the slash-command path +// deliberately shares one grade builder with the API so the two never drift. +// Reimplementing any of that here would reintroduce exactly that drift, so +// everything data-shaped is a POST to the backend's /internal/* routes. +package apiclient + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +// defaultTimeout bounds every callback. Grading does synchronous parquet/DB +// reads on the Python side (a cold city can take a while), so be generous — +// but never infinite: a hung backend must not wedge the gateway's event loop. +const defaultTimeout = 60 * time.Second + +// errBodyLimit caps how much of an error response we echo into logs; enough +// to see a traceback's headline without dumping megabytes. +const errBodyLimit = 512 + +// Client calls the backend's internal-only routes. The token is defence in +// depth — Caddy never routes /internal/* to the backend in the first place — +// but it is still required on every request so a misrouted or LAN-local call +// cannot hit these endpoints unauthenticated. +type Client struct { + base string + token string + http *http.Client +} + +// New returns a Client for the backend at base (e.g. http://web:8137), +// authenticating with token. +func New(base, token string) *Client { + return &Client{ + base: strings.TrimRight(base, "/"), + token: token, + http: &http.Client{Timeout: defaultTimeout}, + } +} + +// post sends body to path and returns the raw response body on 2xx. +func (c *Client) post(ctx context.Context, path string, body any) (json.RawMessage, error) { + payload, err := json.Marshal(body) + if err != nil { + return nil, fmt.Errorf("apiclient: encode %s body: %w", path, err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.base+path, bytes.NewReader(payload)) + if err != nil { + return nil, fmt.Errorf("apiclient: build %s request: %w", path, err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Thermograph-Internal-Token", c.token) + + resp, err := c.http.Do(req) + if err != nil { + return nil, fmt.Errorf("apiclient: POST %s: %w", path, err) + } + defer resp.Body.Close() + + // Read one byte past the limit so the truncation marker only appears when + // something was actually cut. + raw, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if err != nil { + return nil, fmt.Errorf("apiclient: POST %s: read body: %w", path, err) + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + snippet := raw + truncated := "" + if len(snippet) > errBodyLimit { + snippet = snippet[:errBodyLimit] + truncated = "... (truncated)" + } + return nil, fmt.Errorf("apiclient: POST %s: status %d: %s%s", path, resp.StatusCode, snippet, truncated) + } + return json.RawMessage(raw), nil +} + +// Grade asks Python to grade a city query and returns the gateway-ready +// Discord message JSON VERBATIM. The raw bytes are relayed to Discord without +// parsing or reshaping — that verbatim relay is what keeps Go out of the +// grading/embed contract: any change to the message shape lives entirely on +// the Python side. +func (c *Client) Grade(ctx context.Context, query string) (json.RawMessage, error) { + return c.post(ctx, "/internal/discord/grade", map[string]string{"query": query}) +} + +// WarmCities triggers the warm-cities job (the Python side owns what "warm" +// means and which cities are curated). +func (c *Client) WarmCities(ctx context.Context) error { + _, err := c.post(ctx, "/internal/jobs/warm-cities", map[string]string{}) + return err +} + +// IndexNow triggers the IndexNow check-and-submit job; a no-op on the Python +// side when the indexable URL set has not changed. +func (c *Client) IndexNow(ctx context.Context) error { + _, err := c.post(ctx, "/internal/jobs/indexnow", map[string]string{}) + return err +} diff --git a/backend/daemon/internal/apiclient/client_test.go b/backend/daemon/internal/apiclient/client_test.go new file mode 100644 index 0000000..9fa456b --- /dev/null +++ b/backend/daemon/internal/apiclient/client_test.go @@ -0,0 +1,124 @@ +package apiclient + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestGradeHappyPathRelaysBodyVerbatim(t *testing.T) { + // Deliberately odd formatting/key order: Grade must hand back the exact + // bytes, never a re-marshalled reshaping of them. + const body = `{ "content":null, "embeds": [ {"title":"Phoenix"} ] }` + var gotPath, gotToken, gotQuery string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + gotToken = r.Header.Get("X-Thermograph-Internal-Token") + var req map[string]string + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Errorf("decode request: %v", err) + } + gotQuery = req["query"] + w.Header().Set("Content-Type", "application/json") + if _, err := w.Write([]byte(body)); err != nil { + t.Errorf("write response: %v", err) + } + })) + defer srv.Close() + + c := New(srv.URL, "tok123") + raw, err := c.Grade(context.Background(), "phoenix") + if err != nil { + t.Fatalf("Grade: %v", err) + } + if string(raw) != body { + t.Errorf("body not relayed verbatim:\n got %q\nwant %q", raw, body) + } + if gotPath != "/internal/discord/grade" { + t.Errorf("path = %q", gotPath) + } + if gotToken != "tok123" { + t.Errorf("token header = %q, want %q", gotToken, "tok123") + } + if gotQuery != "phoenix" { + t.Errorf("query = %q", gotQuery) + } +} + +func TestJobsSendTokenAndHitRightPaths(t *testing.T) { + paths := map[string]bool{} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Errorf("method = %s, want POST", r.Method) + } + if r.Header.Get("X-Thermograph-Internal-Token") != "tok" { + t.Errorf("missing/wrong token header on %s", r.URL.Path) + } + paths[r.URL.Path] = true + if _, err := w.Write([]byte(`{"ok":true}`)); err != nil { + t.Errorf("write response: %v", err) + } + })) + defer srv.Close() + + c := New(srv.URL+"/", "tok") // trailing slash must not produce "//internal" + if err := c.WarmCities(context.Background()); err != nil { + t.Fatalf("WarmCities: %v", err) + } + if err := c.IndexNow(context.Background()); err != nil { + t.Fatalf("IndexNow: %v", err) + } + for _, p := range []string{"/internal/jobs/warm-cities", "/internal/jobs/indexnow"} { + if !paths[p] { + t.Errorf("path %s was never hit (saw %v)", p, paths) + } + } +} + +func TestNon2xxReturnsStatusAndTruncatedBody(t *testing.T) { + long := strings.Repeat("x", 2000) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, long, http.StatusBadGateway) + })) + defer srv.Close() + + c := New(srv.URL, "tok") + _, err := c.Grade(context.Background(), "phoenix") + if err == nil { + t.Fatal("expected error on 502, got nil") + } + msg := err.Error() + if !strings.Contains(msg, "502") { + t.Errorf("error should include status, got: %s", msg) + } + if !strings.Contains(msg, "truncated") { + t.Errorf("error should mark truncation, got %d chars: %.100s...", len(msg), msg) + } + if len(msg) > 1000 { + t.Errorf("error message not truncated: %d chars", len(msg)) + } +} + +func TestContextCancellation(t *testing.T) { + block := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-block // hold the request open until the test ends + })) + defer srv.Close() + defer close(block) + + c := New(srv.URL, "tok") + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _, err := c.Grade(ctx, "phoenix") + if err == nil { + t.Fatal("expected error from cancelled context, got nil") + } + if !strings.Contains(err.Error(), "deadline") && !strings.Contains(err.Error(), "cancel") { + t.Errorf("expected a context-shaped error, got: %v", err) + } +} diff --git a/backend/daemon/internal/config/config.go b/backend/daemon/internal/config/config.go new file mode 100644 index 0000000..c7d60e6 --- /dev/null +++ b/backend/daemon/internal/config/config.go @@ -0,0 +1,138 @@ +// Package config loads and validates the daemon's environment. All knobs are +// env vars because the daemon runs as a sibling container to the web service +// and shares its /etc/thermograph.env — a config file would be a second source +// of truth for the same values. +package config + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "errors" + "os" + "strconv" + "strings" + "time" +) + +// Defaults mirror the Python scheduler they replace (notifications/scheduler.py): +// warm-cities daily — a warm cache mostly stays warm, and every already-cached +// city is a cheap skip; IndexNow every 6h — submit_if_changed() is a no-op when +// nothing changed, so it can tick more often without spending anything. +const ( + DefaultWarmCitiesIntervalHours = 24 + DefaultIndexNowIntervalHours = 6 +) + +// Config is the daemon's fully-resolved configuration. Construct it only via +// Load so the required/optional split is enforced in one place. +type Config struct { + // InternalToken is the shared secret sent as X-Thermograph-Internal-Token + // on every callback into Python. Required: if it is unset the web app + // disables its /internal/* routes (fail closed), so a daemon without it + // could only ever fail every call — better to refuse to start. + InternalToken string + + // APIBase is the in-network base URL of the FastAPI app (e.g. + // http://web:8137). Same env var the frontend already uses for its + // server-side calls, so there is one name for "where is the backend". + APIBase string + + // DiscordEnabled gates the gateway connection. Off by default: a gateway + // connection is a new persistent behaviour, so it takes an explicit opt-in + // (THERMOGRAPH_DISCORD_BOT truthy AND a token) rather than piggybacking on + // the token being present for DMs. + DiscordEnabled bool + + // DiscordToken is the bot token; only meaningful when DiscordEnabled. + DiscordToken string + + WarmCitiesInterval time.Duration + IndexNowInterval time.Duration +} + +// truthy matches the Python side's _TRUTHY set exactly, so the same env file +// enables/disables the bot regardless of which process reads it. +func truthy(s string) bool { + switch strings.ToLower(strings.TrimSpace(s)) { + case "1", "true", "yes", "on": + return true + } + return false +} + +// intervalHours parses an interval env var. A malformed or non-positive value +// falls back to the default rather than erroring — mirrors the Python +// `int(os.environ.get(..., "24") or 24)` behaviour where a bad knob degrades +// to the documented default instead of taking the process down. +func intervalHours(raw string, def int) time.Duration { + hours := def + if v := strings.TrimSpace(raw); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + hours = n + } + } + return time.Duration(hours) * time.Hour +} + +// Load reads the environment and returns a validated Config, or an error that +// should abort startup. The daemon must never run half-configured: a missing +// token or API base means every callback would fail, silently, forever. +func Load() (*Config, error) { + return load(os.Getenv) +} + +// load takes a getenv func so tests can inject an environment without mutating +// the real one. +func load(getenv func(string) string) (*Config, error) { + cfg := &Config{ + InternalToken: strings.TrimSpace(getenv("THERMOGRAPH_INTERNAL_TOKEN")), + APIBase: strings.TrimSpace(getenv("THERMOGRAPH_API_BASE_INTERNAL")), + DiscordToken: strings.TrimSpace(getenv("THERMOGRAPH_DISCORD_BOT_TOKEN")), + } + if cfg.InternalToken == "" { + cfg.InternalToken = deriveInternalToken(strings.TrimSpace(getenv("THERMOGRAPH_AUTH_SECRET"))) + } + if cfg.InternalToken == "" { + return nil, errors.New("neither THERMOGRAPH_INTERNAL_TOKEN nor THERMOGRAPH_AUTH_SECRET is set; refusing to start (the web app disables /internal/* without a token, so every callback would fail)") + } + if cfg.APIBase == "" { + return nil, errors.New("THERMOGRAPH_API_BASE_INTERNAL is not set; refusing to start (no way to reach the backend)") + } + // The gateway needs both the opt-in flag and a token; missing either just + // disables it — the cron half is still worth running on its own. + cfg.DiscordEnabled = truthy(getenv("THERMOGRAPH_DISCORD_BOT")) && cfg.DiscordToken != "" + + cfg.WarmCitiesInterval = intervalHours(getenv("THERMOGRAPH_WARM_CITIES_INTERVAL_HOURS"), DefaultWarmCitiesIntervalHours) + cfg.IndexNowInterval = intervalHours(getenv("THERMOGRAPH_INDEXNOW_INTERVAL_HOURS"), DefaultIndexNowIntervalHours) + return cfg, nil +} + +// deriveLabel is the domain-separation label for the derived internal token. +// It MUST match backend/api/internal_routes.py's _DERIVE_LABEL byte-for-byte, +// or the daemon and the web app compute different tokens and every callback +// 401s. TestDeriveInternalTokenMatchesPython pins the pair to a shared vector. +const deriveLabel = "thermograph/internal-api/v1" + +// deriveInternalToken computes the shared secret from THERMOGRAPH_AUTH_SECRET, +// which every environment already provisions. This is what lets the daemon +// stand up with no new vault entry and no operator step -- one less secret to +// rotate, leak, or forget on a new host. An explicit THERMOGRAPH_INTERNAL_TOKEN +// still wins when set. +// +// HMAC under a distinct label rather than reusing the auth secret directly: +// ordinary domain separation, so a leak of this token cannot be replayed as the +// session-signing key it came from, and the derivation is one-way. +// +// Returns "" for an empty secret, which the caller turns into a refusal to +// start. It must never silently invent a token: the web app would have derived +// a different one (or none), and every call would fail as an auth error rather +// than as the missing configuration it actually is. +func deriveInternalToken(authSecret string) string { + if authSecret == "" { + return "" + } + mac := hmac.New(sha256.New, []byte(authSecret)) + mac.Write([]byte(deriveLabel)) + return hex.EncodeToString(mac.Sum(nil)) +} diff --git a/backend/daemon/internal/config/config_test.go b/backend/daemon/internal/config/config_test.go new file mode 100644 index 0000000..c5257c3 --- /dev/null +++ b/backend/daemon/internal/config/config_test.go @@ -0,0 +1,96 @@ +package config + +import ( + "testing" + "time" +) + +// env builds a getenv func over a map, defaulting the two required vars so +// each test only states what it cares about. +func env(overrides map[string]string) func(string) string { + base := map[string]string{ + "THERMOGRAPH_INTERNAL_TOKEN": "sekrit", + "THERMOGRAPH_API_BASE_INTERNAL": "http://web:8137", + } + for k, v := range overrides { + base[k] = v + } + return func(k string) string { return base[k] } +} + +func TestRequiredVarsRefuseToStart(t *testing.T) { + for _, missing := range []string{"THERMOGRAPH_INTERNAL_TOKEN", "THERMOGRAPH_API_BASE_INTERNAL"} { + if _, err := load(env(map[string]string{missing: ""})); err == nil { + t.Errorf("expected error when %s is unset, got nil", missing) + } + } + if cfg, err := load(env(nil)); err != nil || cfg == nil { + t.Fatalf("both required vars set: unexpected error %v", err) + } +} + +func TestTruthyParsing(t *testing.T) { + cases := []struct { + flag, token string + want bool + }{ + // Must match the Python _TRUTHY set exactly. + {"1", "tok", true}, + {"true", "tok", true}, + {"TRUE", "tok", true}, + {"yes", "tok", true}, + {"on", "tok", true}, + {" on ", "tok", true}, // whitespace tolerated, like .strip() in Python + {"0", "tok", false}, + {"false", "tok", false}, + {"", "tok", false}, + {"enabled", "tok", false}, // not in the truthy set + // Flag alone is not enough: no token means no gateway. + {"1", "", false}, + } + for _, c := range cases { + cfg, err := load(env(map[string]string{ + "THERMOGRAPH_DISCORD_BOT": c.flag, + "THERMOGRAPH_DISCORD_BOT_TOKEN": c.token, + })) + if err != nil { + t.Fatalf("flag=%q token=%q: unexpected error %v", c.flag, c.token, err) + } + if cfg.DiscordEnabled != c.want { + t.Errorf("flag=%q token=%q: DiscordEnabled=%v, want %v", c.flag, c.token, cfg.DiscordEnabled, c.want) + } + } +} + +func TestIntervalDefaultsAndFallbacks(t *testing.T) { + cases := []struct { + name string + warm, indexnow string + wantWarm time.Duration + wantIndexNow time.Duration + }{ + {"unset uses defaults", "", "", 24 * time.Hour, 6 * time.Hour}, + {"explicit values win", "12", "3", 12 * time.Hour, 3 * time.Hour}, + // Malformed values degrade to the default instead of erroring, like + // the Python int(... or 24) they replace. + {"garbage falls back", "soon", "1.5", 24 * time.Hour, 6 * time.Hour}, + {"non-positive falls back", "0", "-2", 24 * time.Hour, 6 * time.Hour}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + cfg, err := load(env(map[string]string{ + "THERMOGRAPH_WARM_CITIES_INTERVAL_HOURS": c.warm, + "THERMOGRAPH_INDEXNOW_INTERVAL_HOURS": c.indexnow, + })) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if cfg.WarmCitiesInterval != c.wantWarm { + t.Errorf("WarmCitiesInterval=%v, want %v", cfg.WarmCitiesInterval, c.wantWarm) + } + if cfg.IndexNowInterval != c.wantIndexNow { + t.Errorf("IndexNowInterval=%v, want %v", cfg.IndexNowInterval, c.wantIndexNow) + } + }) + } +} diff --git a/backend/daemon/internal/config/derive_test.go b/backend/daemon/internal/config/derive_test.go new file mode 100644 index 0000000..7e185e6 --- /dev/null +++ b/backend/daemon/internal/config/derive_test.go @@ -0,0 +1,66 @@ +package config + +import "testing" + +// CROSS-LANGUAGE CONTRACT. The daemon and the web app must derive byte-identical +// tokens from the same THERMOGRAPH_AUTH_SECRET, or every callback 401s -- and it +// would surface as an auth error, which reads like a bug rather than like the +// configuration drift it actually is. The same vector is asserted on the Python +// side in backend/tests/api/test_internal_routes.py::test_derived_token_matches_go. +// If deriveLabel or the construction changes, BOTH must change together. +const ( + sharedVectorSecret = "test-auth-secret" + sharedVectorToken = "4c3830b2158941ff52720d198b65f7924372a3a482b8da52540a4d9be38247ea" +) + +func TestDeriveInternalTokenMatchesPython(t *testing.T) { + if got := deriveInternalToken(sharedVectorSecret); got != sharedVectorToken { + t.Fatalf("deriveInternalToken(%q) = %q; want %q (must match internal_routes.py)", + sharedVectorSecret, got, sharedVectorToken) + } +} + +// An empty secret must yield nothing rather than some fixed "empty HMAC" value: +// the caller turns "" into a refusal to start, and inventing a token here would +// let the daemon run against a web app that computed a different one. +func TestDeriveInternalTokenEmptySecretYieldsNothing(t *testing.T) { + if got := deriveInternalToken(""); got != "" { + t.Fatalf("deriveInternalToken(%q) = %q; want empty so Load refuses to start", "", got) + } +} + +func TestLoadPrefersExplicitTokenOverDerived(t *testing.T) { + env := map[string]string{ + "THERMOGRAPH_INTERNAL_TOKEN": "explicit-wins", + "THERMOGRAPH_AUTH_SECRET": sharedVectorSecret, + "THERMOGRAPH_API_BASE_INTERNAL": "http://web:8137", + } + cfg, err := load(func(k string) string { return env[k] }) + if err != nil { + t.Fatalf("load: %v", err) + } + if cfg.InternalToken != "explicit-wins" { + t.Fatalf("InternalToken = %q; an explicit env var must win over derivation", cfg.InternalToken) + } +} + +func TestLoadDerivesTokenWhenOnlyAuthSecretIsSet(t *testing.T) { + env := map[string]string{ + "THERMOGRAPH_AUTH_SECRET": sharedVectorSecret, + "THERMOGRAPH_API_BASE_INTERNAL": "http://web:8137", + } + cfg, err := load(func(k string) string { return env[k] }) + if err != nil { + t.Fatalf("load: %v", err) + } + if cfg.InternalToken != sharedVectorToken { + t.Fatalf("InternalToken = %q; want the derived %q", cfg.InternalToken, sharedVectorToken) + } +} + +func TestLoadRefusesWhenNeitherTokenNorAuthSecretIsSet(t *testing.T) { + env := map[string]string{"THERMOGRAPH_API_BASE_INTERNAL": "http://web:8137"} + if _, err := load(func(k string) string { return env[k] }); err == nil { + t.Fatal("load succeeded with no token and no auth secret; it must refuse to start") + } +} diff --git a/backend/daemon/internal/cron/cron.go b/backend/daemon/internal/cron/cron.go new file mode 100644 index 0000000..945323f --- /dev/null +++ b/backend/daemon/internal/cron/cron.go @@ -0,0 +1,92 @@ +// Package cron is the Go replacement for notifications/scheduler.py's +// APScheduler pair (warm-cities + IndexNow). The Python side chose APScheduler +// over a Postgres-native queue because exactly one process ever runs these +// jobs — no fan-out, backpressure, or dead-letter need — and that shape +// carries over unchanged: this daemon is the sole scheduler, so plain +// per-job tickers are all the machinery required. procrastinate/pgqueuer +// remain the documented upgrade path if that ever changes — see +// docs/architecture/repo-topology-and-infrastructure.md §7. +// +// The jobs themselves stay in Python (called via the internal HTTP routes); +// this package owns only the timing. +package cron + +import ( + "context" + "log/slog" + "sync" + "time" +) + +// Job is one recurring task: a name for logs, an interval, and the callback +// (in practice an apiclient method hitting a /internal/jobs/* route). +type Job struct { + Name string + Interval time.Duration + Run func(ctx context.Context) error +} + +// Run drives every job on its own ticker until ctx is cancelled, then returns +// once all job goroutines have stopped. It never returns early: a failing job +// logs and keeps ticking — one bad tick (backend briefly down, timeout) must +// never kill the schedule, because nothing restarts it short of a container +// restart. +func Run(ctx context.Context, logger *slog.Logger, jobs ...Job) { + var wg sync.WaitGroup + for _, job := range jobs { + wg.Add(1) + go func(job Job) { + defer wg.Done() + runJob(ctx, logger, job) + }(job) + } + wg.Wait() +} + +func runJob(ctx context.Context, logger *slog.Logger, job Job) { + // A plain ticker's first fire is one interval from now, NOT at startup — + // deliberately kept from the Python scheduler (its add_job had no + // next_run_time override for the same reason): warm-cities already runs at + // deploy time via the deploy hook, so an immediate run on every daemon + // boot/restart would just re-spend work (and, for warm-cities, archive-fetch + // quota) that the deploy already paid for. + ticker := time.NewTicker(job.Interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + start := time.Now() + // Run synchronously in this loop: a job can never overlap with + // itself by construction. That matters most for warm-cities, + // which spends the archive-fetch quota — two overlapping runs would + // double-spend it. + if err := job.Run(ctx); err != nil { + if ctx.Err() != nil { + // Shutdown raced the job; the cancellation is the story, + // not the (expected) aborted call. + return + } + logger.Error("cron job failed; will retry next tick", + "job", job.Name, "err", err, "elapsed", time.Since(start)) + } else { + logger.Info("cron job ok", + "job", job.Name, "elapsed", time.Since(start)) + } + // If the job overran its interval, a tick may have fired mid-run. + // Skip it rather than running again back-to-back — a late tick is + // the same double-spend as an overlapping one, just serialized. + // (Go 1.23+ tickers already drop fires with no ready receiver; + // this drain pins the skip semantics rather than relying on the + // runtime's channel buffering.) + select { + case <-ticker.C: + logger.Warn("cron job overran its interval; skipping the tick that fired mid-run", + "job", job.Name, "interval", job.Interval, "elapsed", time.Since(start)) + default: + } + } + } +} diff --git a/backend/daemon/internal/cron/cron_test.go b/backend/daemon/internal/cron/cron_test.go new file mode 100644 index 0000000..f2c5675 --- /dev/null +++ b/backend/daemon/internal/cron/cron_test.go @@ -0,0 +1,204 @@ +package cron + +import ( + "context" + "errors" + "io" + "log/slog" + "sync/atomic" + "testing" + "time" +) + +// Intervals here are tens of milliseconds: long enough that scheduler jitter +// cannot invert an assertion, short enough that the whole file runs in well +// under a second. + +func discard() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// The first run must be one interval after start, never at startup — the +// deploy hook already warmed the cache, so a boot-time run is pure waste. +func TestFirstRunIsDeferredOneInterval(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + var runs atomic.Int32 + done := make(chan struct{}) + go func() { + defer close(done) + Run(ctx, discard(), Job{ + Name: "test", + Interval: 60 * time.Millisecond, + Run: func(context.Context) error { + runs.Add(1) + return nil + }, + }) + }() + + // Well inside the first interval: nothing may have run yet. + time.Sleep(20 * time.Millisecond) + if got := runs.Load(); got != 0 { + t.Fatalf("job ran %d time(s) before the first interval elapsed; first run must be deferred", got) + } + + // Well past the first interval: it must have run by now. + deadline := time.After(500 * time.Millisecond) + for runs.Load() == 0 { + select { + case <-deadline: + t.Fatal("job never ran after the first interval elapsed") + case <-time.After(5 * time.Millisecond): + } + } + + cancel() + <-done +} + +// A slow run must not overlap with itself, and ticks that fired mid-run must +// be skipped, not queued — an immediate back-to-back run would double-spend +// the archive-fetch quota just like a concurrent one. +func TestSlowJobNeverOverlapsAndSkipsMissedTicks(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + const interval = 30 * time.Millisecond + + var inFlight, maxInFlight, runs atomic.Int32 + release := make(chan struct{}) + started := make(chan struct{}, 16) + + done := make(chan struct{}) + go func() { + defer close(done) + Run(ctx, discard(), Job{ + Name: "slow", + Interval: interval, + Run: func(context.Context) error { + n := inFlight.Add(1) + for { + m := maxInFlight.Load() + if n <= m || maxInFlight.CompareAndSwap(m, n) { + break + } + } + runs.Add(1) + started <- struct{}{} + <-release // block until the test lets each run finish + inFlight.Add(-1) + return nil + }, + }) + }() + + // First run starts; hold it across several intervals. + <-started + time.Sleep(4 * interval) + if got := maxInFlight.Load(); got != 1 { + t.Fatalf("job overlapped with itself: max in-flight = %d", got) + } + if got := runs.Load(); got != 1 { + t.Fatalf("expected exactly 1 run while the first is still blocked, got %d", got) + } + release <- struct{}{} + + // After release the schedule resumes, but the ticks missed during the + // block must have been dropped: the next run arrives roughly one interval + // later, and only one more within that window (not a burst of catch-ups). + select { + case <-started: + case <-time.After(500 * time.Millisecond): + t.Fatal("job never resumed after the blocked run finished") + } + if got := runs.Load(); got != 2 { + t.Fatalf("missed ticks were replayed as a burst: %d runs total, want 2", got) + } + release <- struct{}{} + + cancel() + <-done + if got := maxInFlight.Load(); got != 1 { + t.Fatalf("job overlapped with itself: max in-flight = %d", got) + } +} + +// One failed tick must never kill the ticker — nothing restarts the schedule +// short of a container restart. +func TestFailedTickDoesNotStopSchedule(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + var runs atomic.Int32 + done := make(chan struct{}) + go func() { + defer close(done) + Run(ctx, discard(), Job{ + Name: "flaky", + Interval: 20 * time.Millisecond, + Run: func(context.Context) error { + if runs.Add(1) == 1 { + return errors.New("backend briefly down") + } + return nil + }, + }) + }() + + deadline := time.After(1 * time.Second) + for runs.Load() < 3 { + select { + case <-deadline: + t.Fatalf("schedule stalled after a failure: only %d run(s)", runs.Load()) + case <-time.After(5 * time.Millisecond): + } + } + + cancel() + <-done +} + +// Cancellation must stop Run promptly even while a job is mid-call: the job +// callbacks are ctx-aware HTTP calls, so cancelling the context aborts the +// in-flight request and the loop must then exit instead of ticking again. +func TestCancelStopsRunWhileJobInFlight(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(context.Background()) + + started := make(chan struct{}) + done := make(chan struct{}) + go func() { + defer close(done) + Run(ctx, discard(), + Job{ + Name: "blocking", + Interval: 10 * time.Millisecond, + Run: func(jobCtx context.Context) error { + close(started) + <-jobCtx.Done() // behaves like an HTTP call aborted by cancellation + return jobCtx.Err() + }, + }, + // A second, idle job proves Run waits for ALL jobs to stop. + Job{ + Name: "idle", + Interval: time.Hour, + Run: func(context.Context) error { return nil }, + }, + ) + }() + + <-started + cancel() + + select { + case <-done: + case <-time.After(1 * time.Second): + t.Fatal("Run did not return promptly after context cancellation") + } +} diff --git a/backend/daemon/internal/gateway/gateway.go b/backend/daemon/internal/gateway/gateway.go new file mode 100644 index 0000000..123652e --- /dev/null +++ b/backend/daemon/internal/gateway/gateway.go @@ -0,0 +1,529 @@ +// Package gateway holds Discord's **gateway** bot — it reacts to @mentions +// (and DMs) in real time. The other two Discord paths need no persistent +// connection: slash commands arrive as signed HTTP POSTs and the daily post +// goes out over an incoming webhook, both handled on the Python side. This one +// is different — to answer an *ordinary* message that mentions the bot, we +// have to hold a live websocket to Discord's gateway and listen for +// MESSAGE_CREATE events. +// +// Single-connection invariant. Discord permits exactly one gateway connection +// per bot token (per shard). The Python predecessor enforced that with a +// leader election inside web/app.py; here it holds by deployment shape — the +// daemon runs as exactly one replica, and this package owns its only +// connection. +// +// Least privilege / no privileged intent. We subscribe to GUILDS + +// GUILD_MESSAGES + DIRECT_MESSAGES and act only on messages that @mention the +// bot or DM it. Discord delivers message *content* for exactly those cases +// even without the privileged MESSAGE_CONTENT intent, so the bot needs no +// privileged-intent portal review — a real design constraint, not trivia. To +// trigger on arbitrary keywords in channels, enable MESSAGE_CONTENT in the +// Developer Portal and extend triage; the rest of the machinery is unchanged. +// +// The gateway owns NO climate/grading logic. A mention with a city query is +// answered by calling back into the Python app (Grader / apiclient), which +// shares one grade builder with the /grade slash command so the two never +// drift. +package gateway + +import ( + "context" + "encoding/json" + "fmt" + "log" + "math/rand" + "net/http" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/coder/websocket" + + "thermograph/daemon/internal/config" +) + +// Run is the package entrypoint: build a Bot from the resolved config and +// drive it for ctx's lifetime. The caller has already checked +// cfg.DiscordEnabled — an unconfigured deploy never gets this far, so an +// unconfigured deploy pays nothing. +func Run(ctx context.Context, cfg *config.Config, api Grader) error { + return New(cfg.DiscordToken, api).Run(ctx) +} + +// v10 JSON gateway. On a resume we reconnect to the session's +// resume_gateway_url (from the READY payload) instead, with the same query +// string appended. +const ( + gatewayURL = "wss://gateway.discord.gg/?v=10&encoding=json" + gatewayQS = "/?v=10&encoding=json" +) + +// Gateway intents (bitfield). GUILDS gives guild/channel context; +// GUILD_MESSAGES and DIRECT_MESSAGES deliver message-create events in servers +// and DMs. We deliberately do NOT request MESSAGE_CONTENT (privileged) — +// Discord still includes content for messages that mention us or DM us, which +// is all we act on. +const ( + intentGuilds = 1 << 0 + intentGuildMessages = 1 << 9 + intentDirectMessages = 1 << 12 + intents = intentGuilds | intentGuildMessages | intentDirectMessages +) + +// Gateway opcodes (discord.com/developers/docs/topics/gateway-events#opcodes). +const ( + opDispatch = 0 + opHeartbeat = 1 + opIdentify = 2 + opResume = 6 + opReconnect = 7 + opInvalidSession = 9 + opHello = 10 + opHeartbeatACK = 11 +) + +// isFatalClose reports whether a close code is unrecoverable — reconnecting +// just loops, so we stop: 4004 auth failed (bad token), 4010 invalid shard, +// 4011 sharding required, 4012 invalid API version, 4013 invalid intents, +// 4014 disallowed (privileged) intents. +func isFatalClose(code websocket.StatusCode) bool { + switch code { + case 4004, 4010, 4011, 4012, 4013, 4014: + return true + } + return false +} + +// Reconnect backoff bounds. A server-requested reconnect (op 7) resets to the +// start so a planned reconnect comes back promptly. +const ( + backoffStart = 1 * time.Second + backoffMax = 60 * time.Second +) + +// nextBackoff doubles the reconnect delay, capped at backoffMax. +func nextBackoff(d time.Duration) time.Duration { + d *= 2 + if d > backoffMax { + d = backoffMax + } + return d +} + +// maxConcurrentReplies bounds in-flight MESSAGE_CREATE handling. The Python +// version pushed each message onto a thread (asyncio.to_thread) so a slow +// grading lookup couldn't stall heartbeats, but its queue was unbounded; the +// semaphore here keeps the same never-block-the-read-loop property while +// making sure a flood of mentions can't spawn unbounded goroutines against +// the backend. +const maxConcurrentReplies = 4 + +// dialTimeout mirrors the Python's open_timeout=30. +const dialTimeout = 30 * time.Second + +// maxFrameSize mirrors the Python's max_size=2**20 — a grade embed is tiny, +// so anything near a megabyte is not a frame we want to buffer. +const maxFrameSize = 1 << 20 + +const ( + modeIdentify = "identify" + modeResume = "resume" +) + +// Grader is the one slice of the internal API the gateway needs: a city query +// in, gateway-ready Discord message JSON out. Satisfied by *apiclient.Client. +type Grader interface { + Grade(ctx context.Context, query string) (json.RawMessage, error) +} + +// Bot owns the single gateway connection and everything hanging off it. +type Bot struct { + token string + api Grader + sem chan struct{} // bounds concurrent MESSAGE_CREATE handlers + rest string // Discord REST base; swapped for a test server in tests + http *http.Client // REST replies (not the gateway socket) +} + +// New returns a Bot that authenticates with token and answers city queries +// through api. +func New(token string, api Grader) *Bot { + return &Bot{ + token: token, + api: api, + sem: make(chan struct{}, maxConcurrentReplies), + rest: discordAPIBase, + http: &http.Client{Timeout: 30 * time.Second}, + } +} + +// payload is the gateway envelope: opcode, event data, sequence number, +// dispatch type. +type payload struct { + Op int `json:"op"` + D json.RawMessage `json:"d"` + S *int64 `json:"s"` + T string `json:"t"` +} + +// session is the per-connection state that must survive a resume: the last +// sequence number (what RESUME replays from), the resume token/URL, and our +// own user id (for mention detection). seq is mutex-guarded because the read +// loop advances it while the heartbeat goroutine reads it; the other fields +// are only ever touched from the read/run loop. +type session struct { + mu sync.Mutex + seq int64 + haveSeq bool + sessionID string + resumeURL string + userID string +} + +func (s *session) setSeq(v int64) { + s.mu.Lock() + s.seq, s.haveSeq = v, true + s.mu.Unlock() +} + +func (s *session) lastSeq() (int64, bool) { + s.mu.Lock() + defer s.mu.Unlock() + return s.seq, s.haveSeq +} + +// wsConn is a thin JSON send/receive wrapper. coder/websocket allows +// concurrent calls to everything except Read, and only the read loop reads, +// so no extra locking is needed for the heartbeat goroutine's writes. +type wsConn struct { + c *websocket.Conn +} + +func (w *wsConn) send(ctx context.Context, v any) error { + data, err := json.Marshal(v) + if err != nil { + return err + } + return w.c.Write(ctx, websocket.MessageText, data) +} + +func (w *wsConn) read(ctx context.Context) ([]byte, error) { + _, data, err := w.c.Read(ctx) + return data, err +} + +func (w *wsConn) close(code websocket.StatusCode, reason string) { + // Best-effort: the connection may already be gone, and a failed close on + // a dead socket changes nothing about what happens next. + _ = w.c.Close(code, reason) +} + +// jitter is the 0..1s randomness added to every reconnect sleep so a fleet of +// clients dropped by the same incident doesn't reconnect in lockstep. +func jitter() time.Duration { + return time.Duration(rand.Float64() * float64(time.Second)) +} + +// sleepCtx sleeps for d, returning false if ctx was cancelled first. +func sleepCtx(ctx context.Context, d time.Duration) bool { + select { + case <-ctx.Done(): + return false + case <-time.After(d): + return true + } +} + +// Run maintains the gateway connection for the daemon's lifetime: connect, +// IDENTIFY (or RESUME), dispatch events, and reconnect with capped +// exponential backoff. It returns nil on context cancellation and an error +// only on a fatal close (bad token / disallowed intents) — reconnecting after +// those loops forever, so the caller must not restart us. +func (b *Bot) Run(ctx context.Context) error { + sess := &session{} + mode := modeIdentify + backoff := backoffStart + for { + useResume := mode == modeResume && sess.resumeURL != "" + url := gatewayURL + if useResume { + url = sess.resumeURL + } + dialCtx, cancel := context.WithTimeout(ctx, dialTimeout) + conn, _, err := websocket.Dial(dialCtx, url, nil) + cancel() + if err != nil { + if ctx.Err() != nil { + log.Print("gateway: shutting down") + return nil + } + // A failed dial carries no gateway close code, so it is never + // fatal — back off and retry. + mode = modeIdentify + if sess.sessionID != "" { + mode = modeResume + } + log.Printf("gateway: connect failed (%v); reconnecting in %s", err, backoff) + if !sleepCtx(ctx, backoff+jitter()) { + return nil + } + backoff = nextBackoff(backoff) + continue + } + conn.SetReadLimit(maxFrameSize) + ws := &wsConn{c: conn} + + next, err := b.connection(ctx, ws, sess, useResume) + if ctx.Err() != nil { + // Shutdown: close clean (1000) — we are gone for good, so the + // session should not be held open for a RESUME that never comes. + ws.close(websocket.StatusNormalClosure, "shutting down") + log.Print("gateway: shutting down") + return nil + } + if err != nil { + if code := websocket.CloseStatus(err); isFatalClose(code) { + log.Printf("gateway: fatal gateway close %d; stopping", code) + return fmt.Errorf("gateway: fatal close code %d", code) + } + mode = modeIdentify + if sess.sessionID != "" { + mode = modeResume + } + log.Printf("gateway: connection dropped (%v); reconnecting in %s", err, backoff) + if !sleepCtx(ctx, backoff+jitter()) { + return nil + } + backoff = nextBackoff(backoff) + continue + } + + // Deliberate reconnect (op 7 / op 9): close it ourselves. Non-1000 + // when we intend to RESUME — Discord invalidates the session on a + // 1000/1001 close, which would force a fresh IDENTIFY. + if next == modeResume { + ws.close(websocket.StatusCode(4000), "resuming") + } else { + ws.close(websocket.StatusNormalClosure, "reidentifying") + } + mode = next + backoff = backoffStart // planned reconnect: come back promptly + } +} + +// parseHello validates the op-10 HELLO that opens every connection and returns +// its heartbeat interval. +// +// Every failure here is an ERROR rather than a clean "reidentify". That +// distinction is load-bearing: a clean return is Run's PLANNED-reconnect path, +// which resets the backoff to 1s and redials with no sleep at all. That is +// correct for a server-requested reconnect (op 7), but a gateway persistently +// answering with a malformed HELLO would turn it into a tight reconnect loop +// against Discord -- precisely what the capped backoff exists to prevent. The +// error path applies backoff and still picks the next mode from surviving +// session state. +func parseHello(raw []byte) (time.Duration, error) { + var hello payload + if err := json.Unmarshal(raw, &hello); err != nil { + return 0, fmt.Errorf("gateway: unparseable HELLO: %w", err) + } + if hello.Op != opHello { + return 0, fmt.Errorf("gateway: expected HELLO (op %d), got op %d", opHello, hello.Op) + } + var hd struct { + HeartbeatInterval float64 `json:"heartbeat_interval"` + } + if err := json.Unmarshal(hello.D, &hd); err != nil { + return 0, fmt.Errorf("gateway: unparseable HELLO payload: %w", err) + } + // A non-positive interval would busy-loop the heartbeat goroutine. + if hd.HeartbeatInterval <= 0 { + return 0, fmt.Errorf("gateway: invalid heartbeat_interval %v", hd.HeartbeatInterval) + } + return time.Duration(hd.HeartbeatInterval * float64(time.Millisecond)), nil +} + +// connection drives one live connection until it ends. A clean return carries +// the mode for the next connect ("resume" to reconnect and RESUME, "identify" +// to start a fresh session); an error return means the socket died and the +// caller picks the next mode from the surviving session state. +func (b *Bot) connection(ctx context.Context, ws *wsConn, sess *session, resume bool) (string, error) { + raw, err := ws.read(ctx) + if err != nil { + return "", err + } + interval, err := parseHello(raw) + if err != nil { + return modeIdentify, err + } + + // The heartbeat lives exactly as long as this connection. + hbCtx, stopHB := context.WithCancel(ctx) + defer stopHB() + acked := &atomic.Bool{} + acked.Store(true) + beatNow := make(chan struct{}, 1) + go b.heartbeat(hbCtx, ws, interval, sess, acked, beatNow) + + seq, haveSeq := sess.lastSeq() + if resume && sess.sessionID != "" && haveSeq { + err = ws.send(ctx, map[string]any{"op": opResume, "d": map[string]any{ + "token": b.token, + "session_id": sess.sessionID, + "seq": seq, + }}) + } else { + err = ws.send(ctx, map[string]any{"op": opIdentify, "d": map[string]any{ + "token": b.token, + "intents": intents, + "properties": map[string]string{"os": "linux", "browser": "thermograph", "device": "thermograph"}, + }}) + } + if err != nil { + return "", err + } + + for { + data, err := ws.read(ctx) + if err != nil { + return "", err + } + var p payload + if err := json.Unmarshal(data, &p); err != nil { + log.Printf("gateway: dropping unparseable frame: %v", err) + continue + } + if p.S != nil { + // Every payload that carries a sequence number advances it — it + // is what RESUME replays from. + sess.setSeq(*p.S) + } + switch p.Op { + case opDispatch: + b.dispatch(ctx, p, sess) + case opHeartbeat: + // Server asked for an immediate beat; route it through the + // heartbeat goroutine, which owns beat sending. + select { + case beatNow <- struct{}{}: + default: + } + case opHeartbeatACK: + acked.Store(true) + case opReconnect: + return modeResume, nil // planned reconnect — resume + case opInvalidSession: + // d == true => the session can still be resumed; d == false => + // it is dead and the next connect must IDENTIFY fresh. + var resumable bool + _ = json.Unmarshal(p.D, &resumable) + if !resumable { + sess.sessionID = "" + sess.resumeURL = "" + } + // Discord wants a short randomized wait before re-authing. + if !sleepCtx(ctx, time.Second+jitter()) { + return "", ctx.Err() + } + if sess.sessionID != "" { + return modeResume, nil + } + return modeIdentify, nil + } + } +} + +// heartbeat sends op-1 heartbeats every interval, the first one jittered by +// interval*rand per Discord's guidance. If the previous beat was never ACKed +// the link is a zombie — a half-open TCP connection where our writes +// "succeed" but nothing comes back — so close with a non-1000 code (1000 +// would invalidate the session) and stop; the next connect can then RESUME. +func (b *Bot) heartbeat(ctx context.Context, ws *wsConn, interval time.Duration, sess *session, acked *atomic.Bool, beatNow <-chan struct{}) { + timer := time.NewTimer(time.Duration(rand.Float64() * float64(interval))) + defer timer.Stop() + for { + select { + case <-ctx.Done(): + return + case <-beatNow: + // Server-requested beat (op 1): send immediately, outside the + // zombie accounting — it is not the reply to one of ours. + if err := b.sendHeartbeat(ctx, ws, sess); err != nil { + return // socket gone; the read loop owns reconnecting + } + case <-timer.C: + if !acked.Load() { + ws.close(websocket.StatusCode(4000), "heartbeat ack missed") + return + } + acked.Store(false) + if err := b.sendHeartbeat(ctx, ws, sess); err != nil { + return + } + timer.Reset(interval) + } + } +} + +func (b *Bot) sendHeartbeat(ctx context.Context, ws *wsConn, sess *session) error { + var d any // null until the first sequence number arrives + if seq, ok := sess.lastSeq(); ok { + d = seq + } + return ws.send(ctx, map[string]any{"op": opHeartbeat, "d": d}) +} + +// dispatch handles an op-0 DISPATCH event we care about. +func (b *Bot) dispatch(ctx context.Context, p payload, sess *session) { + switch p.T { + case "READY": + var d struct { + SessionID string `json:"session_id"` + ResumeGatewayURL string `json:"resume_gateway_url"` + User struct { + ID snowflake `json:"id"` + } `json:"user"` + } + if err := json.Unmarshal(p.D, &d); err != nil { + log.Printf("gateway: unparseable READY: %v", err) + return + } + sess.sessionID = d.SessionID + sess.resumeURL = "" + if r := strings.TrimRight(d.ResumeGatewayURL, "/"); r != "" { + sess.resumeURL = r + gatewayQS + } + sess.userID = string(d.User.ID) + log.Printf("gateway: ready (user %s)", sess.userID) + case "MESSAGE_CREATE": + // Until READY tells us our own user id we can't detect mentions (or + // guard against answering ourselves), so stay silent. + if sess.userID == "" { + return + } + var m gwMessage + if err := json.Unmarshal(p.D, &m); err != nil { + log.Printf("gateway: unparseable MESSAGE_CREATE: %v", err) + return + } + // Grading is a synchronous parquet/DB round trip on the Python side, + // so it must never run on the read loop — a slow lookup would stall + // heartbeats (the Python used asyncio.to_thread for the same + // reason). Each message gets its own goroutine, with the semaphore + // bounding in-flight replies so a flood of mentions can't spawn + // unbounded goroutines against the backend — an improvement over + // to_thread's unbounded queue. Over the bound we drop and log rather + // than block the read loop. + botID := sess.userID + select { + case b.sem <- struct{}{}: + go func() { + defer func() { <-b.sem }() + b.handleMessage(ctx, &m, botID) + }() + default: + log.Printf("gateway: dropping message %s: too many replies in flight", m.ID) + } + } +} diff --git a/backend/daemon/internal/gateway/gateway_test.go b/backend/daemon/internal/gateway/gateway_test.go new file mode 100644 index 0000000..3f8cfdf --- /dev/null +++ b/backend/daemon/internal/gateway/gateway_test.go @@ -0,0 +1,182 @@ +// Connection-machinery tests, restricted to the pure parts (backoff, fatal +// close codes, READY/session bookkeeping, sequence tracking). Deliberately no +// live-gateway test — that path is exercised in staging, not CI. + +package gateway + +import ( + "context" + "encoding/json" + "net/http" + "testing" + "time" + + "github.com/coder/websocket" +) + +func TestBackoffDoublesAndCaps(t *testing.T) { + want := []time.Duration{ + 2 * time.Second, 4 * time.Second, 8 * time.Second, 16 * time.Second, + 32 * time.Second, 60 * time.Second, 60 * time.Second, 60 * time.Second, + } + d := backoffStart + for i, w := range want { + d = nextBackoff(d) + if d != w { + t.Fatalf("step %d: want %s, got %s", i, w, d) + } + } +} + +func TestFatalCloseCodes(t *testing.T) { + for _, c := range []websocket.StatusCode{4004, 4010, 4011, 4012, 4013, 4014} { + if !isFatalClose(c) { + t.Errorf("close code %d must be fatal (reconnecting loops forever)", c) + } + } + // 4008 (rate limited) and 4009 (session timed out) are resumable; 4000 is + // our own zombie close; -1 is CloseStatus's not-a-close-error sentinel. + for _, c := range []websocket.StatusCode{-1, 1000, 1001, 4000, 4001, 4008, 4009} { + if isFatalClose(c) { + t.Errorf("close code %d must not be fatal", c) + } + } +} + +func TestReadyCapturesResumeState(t *testing.T) { + b := New("tok", &fakeGrader{}) + sess := &session{} + d := []byte(`{"session_id":"abc","resume_gateway_url":"wss://resume.example/","user":{"id":"999"}}`) + + b.dispatch(context.Background(), payload{Op: opDispatch, T: "READY", D: d}, sess) + + if sess.sessionID != "abc" { + t.Fatalf("want session id abc, got %q", sess.sessionID) + } + // Trailing slash trimmed, gateway query string appended — RESUME must go + // to the session's own URL with the same v10/json parameters. + if sess.resumeURL != "wss://resume.example/?v=10&encoding=json" { + t.Fatalf("bad resume url: %q", sess.resumeURL) + } + if sess.userID != botID { + t.Fatalf("want user id %s, got %q", botID, sess.userID) + } +} + +func TestReadyWithoutResumeURLLeavesItEmpty(t *testing.T) { + b := New("tok", &fakeGrader{}) + sess := &session{resumeURL: "wss://stale.example/?v=10&encoding=json"} + d := []byte(`{"session_id":"abc","user":{"id":"999"}}`) + + b.dispatch(context.Background(), payload{Op: opDispatch, T: "READY", D: d}, sess) + + if sess.resumeURL != "" { + t.Fatalf("a READY without resume_gateway_url must clear the stale url, got %q", sess.resumeURL) + } +} + +func TestMessageCreateBeforeReadyIsIgnored(t *testing.T) { + // Until READY supplies our user id we cannot detect mentions (or guard + // against answering ourselves), so nothing may be handled. + g := &fakeGrader{resp: json.RawMessage(`{}`)} + b := New("tok", g) + sess := &session{} // no userID yet + d, err := json.Marshal(testMsg("<@999> Phoenix", mentioning(botID))) + if err != nil { + t.Fatalf("marshal: %v", err) + } + + b.dispatch(context.Background(), payload{Op: opDispatch, T: "MESSAGE_CREATE", D: d}, sess) + + time.Sleep(20 * time.Millisecond) + if g.callCount() != 0 { + t.Fatal("pre-READY message must be ignored") + } +} + +func TestPayloadSequenceIsNullable(t *testing.T) { + var withSeq, withoutSeq payload + if err := json.Unmarshal([]byte(`{"op":0,"t":"X","s":7,"d":{}}`), &withSeq); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if withSeq.S == nil || *withSeq.S != 7 { + t.Fatalf("want seq 7, got %v", withSeq.S) + } + if err := json.Unmarshal([]byte(`{"op":11,"s":null,"d":null}`), &withoutSeq); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if withoutSeq.S != nil { + t.Fatalf("null seq must stay nil, got %v", *withoutSeq.S) + } +} + +func TestSessionSeqTracking(t *testing.T) { + sess := &session{} + if _, ok := sess.lastSeq(); ok { + t.Fatal("fresh session must have no sequence number") + } + sess.setSeq(41) + sess.setSeq(42) + if seq, ok := sess.lastSeq(); !ok || seq != 42 { + t.Fatalf("want seq 42, got %d (ok=%v)", seq, ok) + } +} + +// A malformed HELLO must be an ERROR, not a clean "reidentify". Run treats a +// clean return as a PLANNED reconnect: backoff resets to 1s and it redials with +// no sleep at all. So if these ever regress to returning nil, a gateway stuck +// sending a bad HELLO becomes a tight reconnect loop against Discord. +func TestParseHelloRejectsMalformedSoBackoffApplies(t *testing.T) { + for _, tc := range []struct { + name string + raw string + }{ + {"not json", `{`}, + {"wrong op", `{"op":0,"d":{"heartbeat_interval":41250}}`}, + {"missing interval", `{"op":10,"d":{}}`}, + {"zero interval", `{"op":10,"d":{"heartbeat_interval":0}}`}, + {"negative interval", `{"op":10,"d":{"heartbeat_interval":-1}}`}, + {"interval wrong type", `{"op":10,"d":{"heartbeat_interval":"soon"}}`}, + } { + t.Run(tc.name, func(t *testing.T) { + got, err := parseHello([]byte(tc.raw)) + if err == nil { + t.Fatalf("parseHello(%s) = %v, nil; want an error so Run applies backoff", tc.raw, got) + } + if got != 0 { + t.Fatalf("parseHello(%s) interval = %v; want 0 on error", tc.raw, got) + } + }) + } +} + +func TestParseHelloAcceptsValid(t *testing.T) { + got, err := parseHello([]byte(`{"op":10,"d":{"heartbeat_interval":41250}}`)) + if err != nil { + t.Fatalf("parseHello: unexpected error %v", err) + } + if want := 41250 * time.Millisecond; got != want { + t.Fatalf("interval = %v; want %v", got, want) + } +} + +// Discord's advertised retry_after is clamped to 5s, matching the Python REST +// helper's _MAX_BACKOFF_S. Replies run on a small fixed worker pool, so a bogus +// server value must not park a slot for minutes. +func TestRetryAfterClamp(t *testing.T) { + for _, tc := range []struct { + name string + body string + want time.Duration + }{ + {"honours a sane value", `{"retry_after":2}`, 2 * time.Second}, + {"clamps a hostile value", `{"retry_after":600}`, 5 * time.Second}, + {"defaults when absent", `{}`, 1 * time.Second}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := retryAfter(http.Header{}, []byte(tc.body)); got != tc.want { + t.Fatalf("retryAfter(%s) = %v; want %v", tc.body, got, tc.want) + } + }) + } +} diff --git a/backend/daemon/internal/gateway/message.go b/backend/daemon/internal/gateway/message.go new file mode 100644 index 0000000..762f846 --- /dev/null +++ b/backend/daemon/internal/gateway/message.go @@ -0,0 +1,118 @@ +// Message triage: the pure decision of whether — and how — to answer one +// MESSAGE_CREATE. Ported from the Python bot's _response_for_message / +// _mentions_bot / _is_dm / _strip_mentions; it is plain string/JSON logic +// with no data dependency, so it lives in Go. Anything needing climate data +// (the actual grade) goes back to Python via Grader. + +package gateway + +import ( + "encoding/json" + "fmt" + "strings" +) + +// helpText answers an empty query. Go owns this constant because it needs no +// data; the wording matches the Python original exactly. +const helpText = "Mention me with a city name — e.g. `@Thermograph Phoenix` — and I'll grade " + + "today's weather against ~45 years of local history." + +// snowflake decodes a Discord id that may arrive as a JSON string (what v10 +// sends) or a bare number. The Python compared ids through str() coercion; +// this keeps that tolerance instead of betting the payload shape never +// wobbles. +type snowflake string + +func (s *snowflake) UnmarshalJSON(b []byte) error { + if string(b) == "null" { + *s = "" + return nil + } + var str string + if err := json.Unmarshal(b, &str); err == nil { + *s = snowflake(str) + return nil + } + var n json.Number + if err := json.Unmarshal(b, &n); err == nil { + *s = snowflake(n.String()) + return nil + } + return fmt.Errorf("snowflake: cannot decode %s", b) +} + +// gwMessage is the slice of a MESSAGE_CREATE payload the triage needs. +type gwMessage struct { + ID snowflake `json:"id"` + ChannelID snowflake `json:"channel_id"` + GuildID snowflake `json:"guild_id"` + Content string `json:"content"` + Author gwAuthor `json:"author"` + Mentions []gwAuthor `json:"mentions"` +} + +type gwAuthor struct { + ID snowflake `json:"id"` + Bot bool `json:"bot"` +} + +// action is triage's verdict for one message. +type action int + +const ( + actSilent action = iota // no reply + actHelp // reply with helpText + actGrade // call back into Python with the query +) + +// mentionsBot reports whether the message @mentions our own user id. +func mentionsBot(m *gwMessage, botID string) bool { + for _, u := range m.Mentions { + if string(u.ID) == botID { + return true + } + } + return false +} + +// isDM: a DM has no guild_id; a guild message always carries one. +func isDM(m *gwMessage) bool { + return m.GuildID == "" +} + +// stripMentions drops every form of the bot's own mention (<@id> and the +// legacy <@!id>) and collapses the surrounding whitespace, leaving just the +// user's query text. +func stripMentions(content, botID string) string { + out := strings.ReplaceAll(content, "<@"+botID+">", " ") + out = strings.ReplaceAll(out, "<@!"+botID+">", " ") + return strings.Join(strings.Fields(out), " ") +} + +// triage decides the reply for one MESSAGE_CREATE: silence, the help line, or +// a grade lookup for the returned query. +// +// Silent unless the message DMs the bot or @mentions it, and never for a bot +// author — which includes ourselves, the guard against a mention-loop (our +// reply quoting the mention must not trigger another reply). The text after +// the mention is treated as a city query; DMs are the query as-is (trimmed +// but not collapsed, matching the Python's .strip()). +func triage(m *gwMessage, botID string) (action, string) { + if m.Author.Bot || string(m.Author.ID) == botID { + return actSilent, "" + } + dm := isDM(m) + if !dm && !mentionsBot(m, botID) { + return actSilent, "" + } + var query string + if dm { + query = strings.TrimSpace(m.Content) + } else { + query = stripMentions(m.Content, botID) + } + if query == "" { + return actHelp, "" + } + return actGrade, query +} diff --git a/backend/daemon/internal/gateway/message_test.go b/backend/daemon/internal/gateway/message_test.go new file mode 100644 index 0000000..294d363 --- /dev/null +++ b/backend/daemon/internal/gateway/message_test.go @@ -0,0 +1,162 @@ +// Triage tests: every behaviour asserted by the Python suite +// (backend/tests/notifications/test_discord_bot.py) has an equivalent here, +// minus the ephemeral-flag drop — the /internal/discord/grade route now +// returns gateway-ready JSON, so that concern moved to the Python side. + +package gateway + +import ( + "encoding/json" + "strings" + "testing" +) + +const botID = "999" + +// testMsg mirrors the Python suite's _msg fixture: author "1" (not a bot), +// message id 42 in channel 77, guild 5 unless overridden. +func testMsg(content string, opts ...func(*gwMessage)) *gwMessage { + m := &gwMessage{ + ID: "42", + ChannelID: "77", + GuildID: "5", + Content: content, + Author: gwAuthor{ID: "1"}, + } + for _, opt := range opts { + opt(m) + } + return m +} + +func fromBot() func(*gwMessage) { + return func(m *gwMessage) { m.Author.Bot = true } +} + +func fromID(id string) func(*gwMessage) { + return func(m *gwMessage) { m.Author.ID = snowflake(id) } +} + +func mentioning(ids ...string) func(*gwMessage) { + return func(m *gwMessage) { + for _, id := range ids { + m.Mentions = append(m.Mentions, gwAuthor{ID: snowflake(id)}) + } + } +} + +func asDM() func(*gwMessage) { + return func(m *gwMessage) { m.GuildID = "" } +} + +// --- when the bot stays silent ---------------------------------------------- + +func TestIgnoresOtherBots(t *testing.T) { + act, _ := triage(testMsg("<@999> Phoenix", fromID("8"), fromBot(), mentioning(botID)), botID) + if act != actSilent { + t.Fatalf("bot author should be ignored, got action %d", act) + } +} + +func TestIgnoresItsOwnMessages(t *testing.T) { + act, _ := triage(testMsg("hello", fromID(botID), mentioning(botID)), botID) + if act != actSilent { + t.Fatalf("own message should be ignored, got action %d", act) + } +} + +func TestIgnoresGuildMessageWithoutAMention(t *testing.T) { + act, _ := triage(testMsg("just chatting"), botID) + if act != actSilent { + t.Fatalf("unmentioned guild message should be ignored, got action %d", act) + } +} + +// --- when the bot replies ---------------------------------------------------- + +func TestMentionWithCityGrades(t *testing.T) { + act, query := triage(testMsg("<@999> Phoenix", mentioning(botID)), botID) + if act != actGrade || query != "Phoenix" { + t.Fatalf("want grade %q, got action %d query %q", "Phoenix", act, query) + } +} + +func TestLegacyNicknameMentionIsStripped(t *testing.T) { + act, query := triage(testMsg("<@!999> Tokyo", mentioning(botID)), botID) + if act != actGrade || query != "Tokyo" { + t.Fatalf("want grade %q, got action %d query %q", "Tokyo", act, query) + } +} + +func TestDMNeedsNoMention(t *testing.T) { + act, query := triage(testMsg("Berlin", asDM()), botID) + if act != actGrade || query != "Berlin" { + t.Fatalf("want grade %q, got action %d query %q", "Berlin", act, query) + } +} + +func TestDMTrimsButKeepsInnerWhitespace(t *testing.T) { + // Parity with the Python DM path (.strip(), not a collapse): only the + // edges are trimmed. + act, query := triage(testMsg(" New York ", asDM()), botID) + if act != actGrade || query != "New York" { + t.Fatalf("want grade %q, got action %d query %q", "New York", act, query) + } +} + +func TestMentionWithNoQueryGivesHelp(t *testing.T) { + act, query := triage(testMsg("<@999>", mentioning(botID)), botID) + if act != actHelp || query != "" { + t.Fatalf("want help, got action %d query %q", act, query) + } + if !strings.Contains(strings.ToLower(helpText), "mention me") { + t.Fatalf("help text lost its instruction: %q", helpText) + } +} + +// --- helpers ----------------------------------------------------------------- + +func TestStripMentionsHandlesBothForms(t *testing.T) { + if got := stripMentions("<@999> <@!999> Sao Paulo", botID); got != "Sao Paulo" { + t.Fatalf("want %q, got %q", "Sao Paulo", got) + } +} + +func TestIsDMIsTrueWithoutGuildID(t *testing.T) { + if !isDM(testMsg("x", asDM())) { + t.Fatal("message without guild_id should be a DM") + } + if isDM(testMsg("x")) { + t.Fatal("message with guild_id should not be a DM") + } +} + +func TestSnowflakeAcceptsStringOrNumber(t *testing.T) { + // The Python compared ids through str() coercion; a numeric id in the + // payload must still count as a mention. + var m gwMessage + raw := `{"id":42,"channel_id":"77","guild_id":"5","content":"<@999> Lima", + "author":{"id":1,"bot":false},"mentions":[{"id":999}]}` + if err := json.Unmarshal([]byte(raw), &m); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if m.ID != "42" || string(m.Author.ID) != "1" { + t.Fatalf("numeric ids mis-decoded: id=%q author=%q", m.ID, m.Author.ID) + } + act, query := triage(&m, botID) + if act != actGrade || query != "Lima" { + t.Fatalf("want grade %q, got action %d query %q", "Lima", act, query) + } +} + +func TestIntentsExcludeMessageContent(t *testing.T) { + // The no-privileged-intent constraint is load-bearing: adding + // MESSAGE_CONTENT (1<<15) silently would make Discord close the gateway + // with 4014 until the portal review happens. + if intents != intentGuilds|intentGuildMessages|intentDirectMessages { + t.Fatalf("unexpected intents bitfield: %d", intents) + } + if intents&(1<<15) != 0 { + t.Fatal("intents must not request MESSAGE_CONTENT") + } +} diff --git a/backend/daemon/internal/gateway/reply.go b/backend/daemon/internal/gateway/reply.go new file mode 100644 index 0000000..0c7b52d --- /dev/null +++ b/backend/daemon/internal/gateway/reply.go @@ -0,0 +1,138 @@ +// The REST half of answering a message: replies go out over plain HTTP with +// the bot token, not the gateway socket (the gateway is receive-only for us). + +package gateway + +import ( + "bytes" + "context" + "encoding/json" + "io" + "log" + "net/http" + "strconv" + "time" +) + +const discordAPIBase = "https://discord.com/api/v10" + +// replyBodyLimit caps how much of a failed reply's response body we echo into +// logs. +const replyBodyLimit = 4096 + +// injectReplyFields adds message_reference + allowed_mentions to an otherwise +// verbatim message body. Only the top level is decoded — every value that +// came from Python (embeds and all) passes through as raw bytes, unparsed, so +// the grading/embed contract stays entirely on the Python side. +// +// allowed_mentions is a SECURITY control, not decoration: the graded reply +// echoes the user's query text, so without {"parse":[]} a crafted query could +// turn our reply into an @everyone/role ping. Only the person who asked is +// pinged, via the reply reference. +func injectReplyFields(body []byte, messageID string) ([]byte, error) { + var top map[string]json.RawMessage + if err := json.Unmarshal(body, &top); err != nil { + return nil, err + } + if messageID != "" { + ref, err := json.Marshal(map[string]string{"message_id": messageID}) + if err != nil { + return nil, err + } + top["message_reference"] = ref + top["allowed_mentions"] = json.RawMessage(`{"parse":[],"replied_user":true}`) + } + return json.Marshal(top) +} + +// reply posts body to the triggering message's channel as a proper reply. A +// failed reply is logged and swallowed — it must never kill the read loop. +func (b *Bot) reply(ctx context.Context, channelID, messageID string, body json.RawMessage) { + if channelID == "" { + return + } + payload, err := injectReplyFields(body, messageID) + if err != nil { + log.Printf("gateway: reply body is not a JSON object: %v", err) + return + } + url := b.rest + "/channels/" + channelID + "/messages" + for attempt := 0; ; attempt++ { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(payload)) + if err != nil { + log.Printf("gateway: reply failed: %v", err) + return + } + req.Header.Set("Authorization", "Bot "+b.token) + req.Header.Set("Content-Type", "application/json") + resp, err := b.http.Do(req) + if err != nil { + log.Printf("gateway: reply failed: %v", err) + return + } + raw, _ := io.ReadAll(io.LimitReader(resp.Body, replyBodyLimit)) + resp.Body.Close() + if resp.StatusCode == http.StatusTooManyRequests && attempt == 0 { + // One retry honouring the advertised wait — parity with the + // Python REST helper's 429 handling. + if !sleepCtx(ctx, retryAfter(resp.Header, raw)) { + return + } + continue + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + log.Printf("gateway: reply failed: status %d: %s", resp.StatusCode, raw) + } + return + } +} + +// retryAfter extracts Discord's requested wait from a 429 (JSON retry_after +// in seconds, falling back to the Retry-After header), clamped so a bogus +// server value can't park the handler for minutes. +func retryAfter(h http.Header, body []byte) time.Duration { + seconds := 1.0 + var d struct { + RetryAfter float64 `json:"retry_after"` + } + if err := json.Unmarshal(body, &d); err == nil && d.RetryAfter > 0 { + seconds = d.RetryAfter + } else if v, err := strconv.ParseFloat(h.Get("Retry-After"), 64); err == nil && v > 0 { + seconds = v + } + // 5s matches the Python REST helper's _MAX_BACKOFF_S. It also bounds the cost + // of a bogus/hostile retry_after: replies run on a small fixed worker pool, so + // a parked handler holds one of very few slots and mentions start being + // dropped that much sooner. + if seconds > 5 { + seconds = 5 + } + return time.Duration(seconds * float64(time.Second)) +} + +// handleMessage triages one MESSAGE_CREATE and sends whatever reply it calls +// for. Runs in its own goroutine (see dispatch) so a slow grade lookup can +// never stall heartbeats or the read loop. +func (b *Bot) handleMessage(ctx context.Context, m *gwMessage, botID string) { + act, query := triage(m, botID) + switch act { + case actSilent: + case actHelp: + body, err := json.Marshal(map[string]string{"content": helpText}) + if err != nil { + return + } + b.reply(ctx, string(m.ChannelID), string(m.ID), body) + case actGrade: + // The callback owns all grading; its JSON is relayed VERBATIM (no + // parsing, no reshaping) so the bot's grades can never drift from + // the API's. A failed callback stays silent — better no reply than a + // made-up one. + raw, err := b.api.Grade(ctx, query) + if err != nil { + log.Printf("gateway: grade callback failed for %q: %v", query, err) + return + } + b.reply(ctx, string(m.ChannelID), string(m.ID), raw) + } +} diff --git a/backend/daemon/internal/gateway/reply_test.go b/backend/daemon/internal/gateway/reply_test.go new file mode 100644 index 0000000..e5d2f40 --- /dev/null +++ b/backend/daemon/internal/gateway/reply_test.go @@ -0,0 +1,278 @@ +// Reply-path tests: a fake Grader plus an httptest server standing in for +// Discord's REST API. No live gateway is involved anywhere. + +package gateway + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" +) + +type fakeGrader struct { + mu sync.Mutex + resp json.RawMessage + err error + calls []string +} + +func (f *fakeGrader) Grade(_ context.Context, query string) (json.RawMessage, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, query) + return f.resp, f.err +} + +func (f *fakeGrader) callCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.calls) +} + +type captured struct { + method, path, auth string + body []byte +} + +// newTestBot wires a Bot to a Grader and an httptest stand-in for Discord. +func newTestBot(t *testing.T, g Grader, handler http.HandlerFunc) *Bot { + t.Helper() + srv := httptest.NewServer(handler) + t.Cleanup(srv.Close) + b := New("tok-123", g) + b.rest = srv.URL + return b +} + +// capture records each request and replies 200 {}. +func capture(got chan<- captured) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + got <- captured{r.Method, r.URL.Path, r.Header.Get("Authorization"), body} + _, _ = w.Write([]byte(`{}`)) + } +} + +func TestGradeReplyIsRelayedVerbatimAsAReply(t *testing.T) { + grade := json.RawMessage(`{"embeds":[{"title":"grade:Phoenix"}]}`) + got := make(chan captured, 1) + b := newTestBot(t, &fakeGrader{resp: grade}, capture(got)) + + b.handleMessage(context.Background(), testMsg("<@999> Phoenix", mentioning(botID)), botID) + + c := <-got + if c.method != http.MethodPost || c.path != "/channels/77/messages" { + t.Fatalf("want POST /channels/77/messages, got %s %s", c.method, c.path) + } + if c.auth != "Bot tok-123" { + t.Fatalf("want bot-token auth, got %q", c.auth) + } + var top map[string]json.RawMessage + if err := json.Unmarshal(c.body, &top); err != nil { + t.Fatalf("reply body not JSON: %v", err) + } + // The embed must be byte-for-byte what Python returned — the verbatim + // relay is the no-drift guarantee. + if string(top["embeds"]) != `[{"title":"grade:Phoenix"}]` { + t.Fatalf("embeds not relayed verbatim: %s", top["embeds"]) + } + if string(top["message_reference"]) != `{"message_id":"42"}` { + t.Fatalf("bad message_reference: %s", top["message_reference"]) + } + // The locked-down allowed_mentions is the anti-@everyone control; it must + // be present exactly. + if string(top["allowed_mentions"]) != `{"parse":[],"replied_user":true}` { + t.Fatalf("bad allowed_mentions: %s", top["allowed_mentions"]) + } +} + +func TestEmptyQueryRepliesWithHelpWithoutGrading(t *testing.T) { + got := make(chan captured, 1) + g := &fakeGrader{} + b := newTestBot(t, g, capture(got)) + + b.handleMessage(context.Background(), testMsg("<@999>", mentioning(botID)), botID) + + c := <-got + var body struct { + Content string `json:"content"` + } + if err := json.Unmarshal(c.body, &body); err != nil { + t.Fatalf("help body not JSON: %v", err) + } + if body.Content != helpText { + t.Fatalf("want help text, got %q", body.Content) + } + if g.callCount() != 0 { + t.Fatal("help reply must not hit the grade callback") + } +} + +func TestSilentMessagesMakeNoRequests(t *testing.T) { + var hits atomic.Int32 + g := &fakeGrader{resp: json.RawMessage(`{}`)} + b := newTestBot(t, g, func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + }) + + b.handleMessage(context.Background(), testMsg("just chatting"), botID) + + if g.callCount() != 0 || hits.Load() != 0 { + t.Fatalf("silent message leaked: grades=%d posts=%d", g.callCount(), hits.Load()) + } +} + +func TestGradeFailureStaysSilent(t *testing.T) { + var hits atomic.Int32 + g := &fakeGrader{err: context.DeadlineExceeded} + b := newTestBot(t, g, func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + }) + + // Must log-and-return, not panic or post a broken reply. + b.handleMessage(context.Background(), testMsg("<@999> Phoenix", mentioning(botID)), botID) + + if hits.Load() != 0 { + t.Fatal("failed grade must not produce a reply") + } +} + +func TestFailedReplyIsSwallowed(t *testing.T) { + g := &fakeGrader{resp: json.RawMessage(`{"content":"x"}`)} + b := newTestBot(t, g, func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "boom", http.StatusInternalServerError) + }) + + // A failed reply must never kill the caller (in production: the read + // loop) — returning normally is the assertion. + b.handleMessage(context.Background(), testMsg("<@999> Phoenix", mentioning(botID)), botID) +} + +func TestReplyRetriesOnceOn429(t *testing.T) { + var hits atomic.Int32 + bodies := make(chan []byte, 2) + g := &fakeGrader{resp: json.RawMessage(`{"content":"x"}`)} + b := newTestBot(t, g, func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + bodies <- body + if hits.Add(1) == 1 { + w.WriteHeader(http.StatusTooManyRequests) + _, _ = w.Write([]byte(`{"retry_after":0.01}`)) + return + } + _, _ = w.Write([]byte(`{}`)) + }) + + b.handleMessage(context.Background(), testMsg("<@999> Phoenix", mentioning(botID)), botID) + + if hits.Load() != 2 { + t.Fatalf("want 1 retry after 429, got %d requests", hits.Load()) + } + first, second := <-bodies, <-bodies + if string(first) != string(second) { + t.Fatal("retry must resend the identical payload") + } +} + +func TestNoChannelMeansNoPost(t *testing.T) { + var hits atomic.Int32 + g := &fakeGrader{resp: json.RawMessage(`{"content":"x"}`)} + b := newTestBot(t, g, func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + }) + + b.handleMessage(context.Background(), + testMsg("<@999> Phoenix", mentioning(botID), func(m *gwMessage) { m.ChannelID = "" }), botID) + + if hits.Load() != 0 { + t.Fatal("a message without a channel id must not be answered") + } +} + +func TestInjectReplyFieldsWithoutMessageIDLeavesBodyAlone(t *testing.T) { + out, err := injectReplyFields([]byte(`{"content":"x"}`), "") + if err != nil { + t.Fatalf("inject: %v", err) + } + var top map[string]json.RawMessage + if err := json.Unmarshal(out, &top); err != nil { + t.Fatalf("out not JSON: %v", err) + } + if _, ok := top["message_reference"]; ok { + t.Fatal("no triggering id => no message_reference") + } + if _, ok := top["allowed_mentions"]; ok { + t.Fatal("no triggering id => no allowed_mentions") + } +} + +// --- dispatch: same reply, off the read loop --------------------------------- + +func TestDispatchDeliversTheSameReplyViaAGoroutine(t *testing.T) { + // Mirrors the Python test_dispatch_delivers_the_same_reply_via_a_thread: + // the concurrency offload (goroutine + semaphore here, asyncio.to_thread + // there) must not change what gets sent. + grade := json.RawMessage(`{"embeds":[{"title":"grade:Phoenix"}]}`) + got := make(chan captured, 1) + b := newTestBot(t, &fakeGrader{resp: grade}, capture(got)) + sess := &session{userID: botID} + d, err := json.Marshal(testMsg("<@999> Phoenix", mentioning(botID))) + if err != nil { + t.Fatalf("marshal: %v", err) + } + + b.dispatch(context.Background(), payload{Op: opDispatch, T: "MESSAGE_CREATE", D: d}, sess) + + select { + case c := <-got: + if c.path != "/channels/77/messages" { + t.Fatalf("want /channels/77/messages, got %s", c.path) + } + var top map[string]json.RawMessage + if err := json.Unmarshal(c.body, &top); err != nil { + t.Fatalf("reply body not JSON: %v", err) + } + if string(top["embeds"]) != `[{"title":"grade:Phoenix"}]` { + t.Fatalf("embeds not relayed verbatim: %s", top["embeds"]) + } + if string(top["message_reference"]) != `{"message_id":"42"}` { + t.Fatalf("bad message_reference: %s", top["message_reference"]) + } + case <-time.After(2 * time.Second): + t.Fatal("dispatched message never produced a reply") + } +} + +func TestMessageFloodIsBoundedWithoutBlockingTheReadLoop(t *testing.T) { + g := &fakeGrader{resp: json.RawMessage(`{}`)} + b := New("tok", g) + b.sem = make(chan struct{}, 1) + b.sem <- struct{}{} // occupy the only slot: simulate a reply in flight + sess := &session{userID: botID} + d, err := json.Marshal(testMsg("<@999> Phoenix", mentioning(botID))) + if err != nil { + t.Fatalf("marshal: %v", err) + } + + done := make(chan struct{}) + go func() { + b.dispatch(context.Background(), payload{Op: opDispatch, T: "MESSAGE_CREATE", D: d}, sess) + close(done) + }() + select { + case <-done: + // dispatch returned promptly — the read loop would not have stalled + case <-time.After(time.Second): + t.Fatal("dispatch blocked on a full semaphore") + } + time.Sleep(20 * time.Millisecond) + if g.callCount() != 0 { + t.Fatal("over-the-bound message must be dropped, not queued") + } +} diff --git a/backend/daemon/main.go b/backend/daemon/main.go new file mode 100644 index 0000000..c264b14 --- /dev/null +++ b/backend/daemon/main.go @@ -0,0 +1,130 @@ +// thermograph-daemon owns the backend's long-lived stateful I/O — the Discord +// gateway websocket (IDENTIFY/RESUME/heartbeat/backoff) and the recurring-job +// timers — and nothing else. Anything that needs data calls back into the +// Python app over its internal-only HTTP routes (see internal/apiclient). +// +// This replaces the two in-process background jobs web/app.py used to start +// under leader election. One daemon process replaces the election entirely: +// there is exactly one replica of this container, so the single-connection +// invariant (Discord tolerates only one gateway session per token; duplicated +// APScheduler replicas would repeat every job) holds by deployment shape +// rather than by runtime coordination. +package main + +import ( + "context" + "log/slog" + "os" + "os/signal" + "sync" + "syscall" + "time" + + "thermograph/daemon/internal/apiclient" + "thermograph/daemon/internal/config" + "thermograph/daemon/internal/cron" + "thermograph/daemon/internal/gateway" +) + +// shutdownGrace bounds how long we wait for the gateway and cron halves to +// stop after a signal. Long enough for the gateway to close its session +// cleanly (a dirty exit leaves Discord holding a stale session), short enough +// to stay inside a container runtime's stop timeout so we exit on our own +// terms rather than being SIGKILLed mid-close. +const shutdownGrace = 10 * time.Second + +func main() { + os.Exit(run()) +} + +// run is main minus os.Exit, so deferred cleanup actually runs. +func run() int { + // Everything to stdout: the container collects stdout, nothing else. + logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) + + // Refuse to start on bad config rather than running half-configured: a + // daemon that cannot authenticate to the backend can only fail silently. + cfg, err := config.Load() + if err != nil { + logger.Error("invalid configuration; refusing to start", "err", err) + return 1 + } + + // One context for everything long-lived; SIGINT/SIGTERM cancels it so the + // gateway can close its session cleanly (Discord treats an abrupt drop + // worse than a clean close for RESUME purposes). + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + api := apiclient.New(cfg.APIBase, cfg.InternalToken) + + logger.Info("thermograph-daemon starting", + "api", cfg.APIBase, + "discord", cfg.DiscordEnabled, + "warm_cities_interval", cfg.WarmCitiesInterval, + "indexnow_interval", cfg.IndexNowInterval) + + var wg sync.WaitGroup + + // The cron half runs unconditionally — it is the floor of what this + // daemon does, gateway or not. + wg.Add(1) + go func() { + defer wg.Done() + cron.Run(ctx, logger, + cron.Job{ + Name: "warm-cities", + Interval: cfg.WarmCitiesInterval, + Run: func(ctx context.Context) error { return api.WarmCities(ctx) }, + }, + cron.Job{ + Name: "indexnow", + Interval: cfg.IndexNowInterval, + Run: func(ctx context.Context) error { return api.IndexNow(ctx) }, + }, + ) + }() + + if cfg.DiscordEnabled { + wg.Add(1) + go func() { + defer wg.Done() + // A fatal gateway error (bad token, disallowed intents) must not + // take the cron half down: the schedule is independently useful, + // and exiting here would turn a Discord misconfiguration into a + // crash-loop that also stops warm-cities/IndexNow. Log loudly and + // keep running instead. + if err := gateway.Run(ctx, cfg, api); err != nil && ctx.Err() == nil { + logger.Error("discord gateway exited with a fatal error; cron jobs keep running without it", + "err", err) + } + }() + } else { + // An unconfigured deploy must pay nothing, same as the Python side: + // no connection, no retries — just say so once and run cron-only. + logger.Info("discord gateway disabled (THERMOGRAPH_DISCORD_BOT not truthy or no bot token); running cron-only") + } + + <-ctx.Done() + // Restore default signal handling immediately: a second SIGINT/SIGTERM + // during the grace period kills us the ordinary way instead of being + // swallowed by the (already-cancelled) NotifyContext. + stop() + logger.Info("shutting down", "grace", shutdownGrace) + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + select { + case <-done: + logger.Info("shutdown complete") + case <-time.After(shutdownGrace): + // This process is the sole holder of the gateway connection, so a + // clean close matters — but a wedged goroutine must not hold the + // container's stop hostage forever. + logger.Warn("shutdown grace period elapsed before all loops stopped; exiting anyway") + } + return 0 +} diff --git a/backend/indexnow.py b/backend/indexnow.py index 1bf3422..a33d987 100644 --- a/backend/indexnow.py +++ b/backend/indexnow.py @@ -179,7 +179,8 @@ def _write_state(sig: str) -> None: def submit_if_changed(base_url: str | None = None) -> dict | None: """Submit every indexable page, but only when the URL set changed since the last successful submission — the shared logic behind the CLI's --if-changed - flag and the worker scheduler's periodic ping (notifications/scheduler.py), so + flag and the daemon's periodic ping (thermograph-daemon's indexnow timer, + via api/internal_routes.py), so a code-only deploy or an unremarkable scheduled tick never resubmits. Returns the submit_all() result, or None when skipped (unchanged, or a partial/failed submission that leaves the prior state in place so the next attempt retries).""" diff --git a/backend/notifications/discord_bot.py b/backend/notifications/discord_bot.py deleted file mode 100644 index 6e32780..0000000 --- a/backend/notifications/discord_bot.py +++ /dev/null @@ -1,295 +0,0 @@ -"""Discord **gateway** bot — reacts to @mentions (and DMs) in real time. - -The other two Discord paths need no persistent connection: slash commands arrive as -signed HTTP POSTs (discord_interactions.py) and the daily post goes out over an -incoming webhook (discord.py). This one is different — to answer an *ordinary* -message that mentions the bot, we have to hold a live websocket to Discord's -gateway and listen for MESSAGE_CREATE events. - -Single-connection invariant. Discord permits exactly one gateway connection per bot -token (per shard), so — like the subscription notifier and the daily post — this -runs in exactly one process: web/app.py starts it only when the process wins the -leader election (see core/singleton.claim_leader). No new container or daemon; it -rides the app's asyncio event loop as a background task. - -Least privilege / no privileged intent. We subscribe to GUILD_MESSAGES + -DIRECT_MESSAGES and act only on messages that @mention the bot or DM it. Discord -delivers message *content* for exactly those cases even without the privileged -MESSAGE_CONTENT intent, so the bot works with no portal review. To trigger on -arbitrary keywords in channels, enable the MESSAGE_CONTENT intent in the Developer -Portal and extend `_response_for_message` — the rest of the machinery is unchanged. - -Enable with THERMOGRAPH_DISCORD_BOT=1 (plus THERMOGRAPH_DISCORD_BOT_TOKEN, the same -bot token discord.py uses for DMs). Unset => no gateway connection; everything here -no-ops, so an unconfigured deploy pays nothing. - -Behaviour: mention the bot with a city name ("@Thermograph Phoenix") and it replies -with the same grade embed the /grade slash command returns, reusing -discord_interactions._grade_message so the two never drift. -""" -from __future__ import annotations - -import asyncio -import json -import logging -import os -import random - -from notifications import discord - -log = logging.getLogger("thermograph.discord_bot") - -# v10 JSON gateway. On a resume we reconnect to the session's resume_gateway_url -# (from the READY payload) instead, with the same query string appended. -_GATEWAY_URL = "wss://gateway.discord.gg/?v=10&encoding=json" -_GATEWAY_QS = "/?v=10&encoding=json" - -# Gateway intents (bitfield). GUILDS gives guild/channel context; GUILD_MESSAGES and -# DIRECT_MESSAGES deliver message-create events in servers and DMs. We deliberately -# do NOT request MESSAGE_CONTENT (privileged) — Discord still includes content for -# messages that mention us or DM us, which is all we act on. -_INTENT_GUILDS = 1 << 0 -_INTENT_GUILD_MESSAGES = 1 << 9 -_INTENT_DIRECT_MESSAGES = 1 << 12 -INTENTS = _INTENT_GUILDS | _INTENT_GUILD_MESSAGES | _INTENT_DIRECT_MESSAGES - -# Gateway opcodes (discord.com/developers/docs/topics/gateway-events#opcodes). -_OP_DISPATCH = 0 -_OP_HEARTBEAT = 1 -_OP_IDENTIFY = 2 -_OP_RESUME = 6 -_OP_RECONNECT = 7 -_OP_INVALID_SESSION = 9 -_OP_HELLO = 10 -_OP_HEARTBEAT_ACK = 11 - -# Close codes for unrecoverable conditions — reconnecting just loops, so we stop: -# 4004 auth failed (bad token), 4010 invalid shard, 4011 sharding required, 4012 -# invalid API version, 4013 invalid intents, 4014 disallowed (privileged) intents. -_FATAL_CLOSE = {4004, 4010, 4011, 4012, 4013, 4014} - -_TRUTHY = {"1", "true", "yes", "on"} - -_HELP = ("Mention me with a city name — e.g. `@Thermograph Phoenix` — and I'll grade " - "today's weather against ~45 years of local history.") - -# Reconnect backoff bounds (seconds). -_BACKOFF_START = 1.0 -_BACKOFF_MAX = 60.0 - - -def enabled() -> bool: - """True iff the gateway bot is switched on AND a bot token is configured. Off by - default: a gateway connection is a new persistent behaviour, so it takes an - explicit opt-in rather than piggybacking on the token being present for DMs.""" - flag = os.environ.get("THERMOGRAPH_DISCORD_BOT", "").strip().lower() in _TRUTHY - return flag and bool(discord.BOT_TOKEN) - - -# --- message handling (pure; unit-tested without a live gateway) -------------- -def _mentions_bot(msg: dict, bot_user_id: str) -> bool: - return any(str(u.get("id")) == str(bot_user_id) for u in (msg.get("mentions") or [])) - - -def _is_dm(msg: dict) -> bool: - # A DM has no guild_id; a guild message always carries one. - return not msg.get("guild_id") - - -def _strip_mentions(content: str, bot_user_id: str) -> str: - """Drop every form of the bot's own mention (<@id> and the legacy <@!id>) and - collapse the surrounding whitespace, leaving just the user's query text.""" - out = content.replace(f"<@{bot_user_id}>", " ").replace(f"<@!{bot_user_id}>", " ") - return " ".join(out.split()) - - -def _response_for_message(msg: dict, bot_user_id: str) -> dict | None: - """The reply payload for one MESSAGE_CREATE, or None to stay silent. - - Silent unless the message DMs the bot or @mentions it, and never for a bot - author (that includes ourselves — the guard against a mention-loop). The text - after the mention is treated as a city query and answered with the same embed - the /grade slash command builds; an empty query gets the help line.""" - author = msg.get("author") or {} - if author.get("bot") or str(author.get("id")) == str(bot_user_id): - return None - dm = _is_dm(msg) - if not dm and not _mentions_bot(msg, bot_user_id): - return None - content = msg.get("content") or "" - query = content.strip() if dm else _strip_mentions(content, bot_user_id) - if not query: - return {"content": _HELP} - # Import here (not at module top) to avoid importing the FastAPI web layer's - # transitive deps for a module that also loads in the notifier-only context. - from notifications import discord_interactions - data = discord_interactions._grade_message(query) - # Gateway messages can't be ephemeral — the flag is interactions-only. - data.pop("flags", None) - return data - - -# --- REST reply --------------------------------------------------------------- -async def _send_reply(channel_id, message_id, data: dict) -> None: - """Post the reply in-channel as a proper reply to the triggering message. - - Reuses discord._bot_post (sync httpx with a 429 retry) off the event loop via a - thread so the gateway read-loop never blocks. allowed_mentions is locked down so - a crafted query can't turn the reply into an @everyone/role ping; only the person - who asked is pinged, via the reply reference.""" - if not channel_id: - return - payload = dict(data) - if message_id: - payload["message_reference"] = {"message_id": str(message_id)} - payload["allowed_mentions"] = {"parse": [], "replied_user": True} - try: - await asyncio.to_thread( - discord._bot_post, f"/channels/{channel_id}/messages", payload) - except Exception: # noqa: BLE001 - a failed reply must never kill the read loop - log.warning("discord_bot: reply failed", exc_info=True) - - -# --- gateway session ---------------------------------------------------------- -class _Session: - """Mutable per-connection state that survives a resume: the last sequence - number, the resume token/URL, and our own user id (for mention detection).""" - - __slots__ = ("seq", "session_id", "resume_url", "user_id") - - def __init__(self) -> None: - self.seq: int | None = None - self.session_id: str | None = None - self.resume_url: str | None = None - self.user_id: str = "" - - -async def _heartbeat(ws, interval: float, sess: _Session, acked: dict) -> None: - """Send op-1 heartbeats every `interval` seconds (first beat jittered per - Discord's guidance). If the previous beat was never ACKed the link is a zombie — - close with a non-1000 code so the next connect can resume, and stop.""" - await asyncio.sleep(interval * random.random()) - while True: - if not acked["ok"]: - await ws.close(code=4000) - return - acked["ok"] = False - try: - await ws.send(json.dumps({"op": _OP_HEARTBEAT, "d": sess.seq})) - except Exception: # noqa: BLE001 - closed socket; the read loop handles reconnect - return - await asyncio.sleep(interval) - - -async def _dispatch(msg: dict, sess: _Session) -> None: - """Handle an op-0 DISPATCH event we care about.""" - t = msg.get("t") - d = msg.get("d") or {} - if t == "READY": - sess.session_id = d.get("session_id") - resume = (d.get("resume_gateway_url") or "").rstrip("/") - sess.resume_url = f"{resume}{_GATEWAY_QS}" if resume else None - sess.user_id = str((d.get("user") or {}).get("id") or "") - log.info("discord_bot: gateway ready (user %s)", sess.user_id) - elif t == "MESSAGE_CREATE" and sess.user_id: - # _response_for_message reaches into discord_interactions._grade_message for - # anything but the help line -- sync parquet/DB reads + grading math, same as - # web/app.py's discord_interactions route -- keep it off the (single, shared) - # gateway event loop so one grading lookup can't stall heartbeats/reads. - reply = await asyncio.to_thread(_response_for_message, d, sess.user_id) - if reply: - await _send_reply(d.get("channel_id"), d.get("id"), reply) - - -async def _connection(ws, sess: _Session, resume: bool) -> str: - """Drive one live connection until it closes. Returns the mode for the next - connect: "resume" to reconnect and RESUME, "identify" to start a fresh session.""" - hello = json.loads(await ws.recv()) - if hello.get("op") != _OP_HELLO: - return "identify" # protocol out of step — start clean - interval = float(hello["d"]["heartbeat_interval"]) / 1000.0 - acked = {"ok": True} - hb = asyncio.create_task(_heartbeat(ws, interval, sess, acked)) - try: - if resume and sess.session_id and sess.seq is not None: - await ws.send(json.dumps({"op": _OP_RESUME, "d": { - "token": discord.BOT_TOKEN, - "session_id": sess.session_id, - "seq": sess.seq, - }})) - else: - await ws.send(json.dumps({"op": _OP_IDENTIFY, "d": { - "token": discord.BOT_TOKEN, - "intents": INTENTS, - "properties": {"os": "linux", "browser": "thermograph", "device": "thermograph"}, - }})) - async for raw in ws: - msg = json.loads(raw) - if msg.get("s") is not None: - sess.seq = msg["s"] - op = msg.get("op") - if op == _OP_DISPATCH: - await _dispatch(msg, sess) - elif op == _OP_HEARTBEAT: # server asked for an immediate beat - await ws.send(json.dumps({"op": _OP_HEARTBEAT, "d": sess.seq})) - elif op == _OP_HEARTBEAT_ACK: - acked["ok"] = True - elif op == _OP_RECONNECT: # planned reconnect — resume - return "resume" - elif op == _OP_INVALID_SESSION: - # d == True => the session can still be resumed; else it's dead and - # the next connect must IDENTIFY fresh. - if not msg.get("d"): - sess.session_id = None - sess.resume_url = None - await asyncio.sleep(1 + random.random()) - return "resume" if sess.session_id else "identify" - finally: - hb.cancel() - return "resume" if sess.session_id else "identify" - - -def _close_code(exc: Exception): - """Best-effort websocket close code across websockets library versions (older - exposes .code; newer nests it under .rcvd).""" - code = getattr(exc, "code", None) - if code: - return code - return getattr(getattr(exc, "rcvd", None), "code", None) - - -async def run() -> None: - """Maintain the gateway connection for the app's lifetime: connect, IDENTIFY (or - RESUME), dispatch events, and reconnect with capped exponential backoff. Returns - only on a fatal close (bad token / disallowed intents) or task cancellation, and - never raises out — a misbehaving gateway must not take the app down.""" - if not enabled(): - return - try: - import websockets - except Exception: # noqa: BLE001 - shipped via uvicorn[standard]; guard anyway - log.warning("discord_bot: 'websockets' unavailable; gateway bot disabled") - return - sess = _Session() - mode = "identify" - backoff = _BACKOFF_START - while True: - use_resume = mode == "resume" and bool(sess.resume_url) - url = sess.resume_url if use_resume else _GATEWAY_URL - try: - async with websockets.connect(url, max_size=2 ** 20, open_timeout=30) as ws: - mode = await _connection(ws, sess, resume=use_resume) - backoff = _BACKOFF_START # server-requested reconnect: come back promptly - continue - except asyncio.CancelledError: - log.info("discord_bot: shutting down") - raise - except Exception as exc: # noqa: BLE001 - any drop -> back off and reconnect - if _close_code(exc) in _FATAL_CLOSE: - log.error("discord_bot: fatal gateway close %s; stopping", _close_code(exc)) - return - mode = "resume" if sess.session_id else "identify" - log.warning("discord_bot: gateway dropped (%s); reconnecting in %.0fs", - type(exc).__name__, backoff) - await asyncio.sleep(backoff + random.random()) - backoff = min(backoff * 2, _BACKOFF_MAX) diff --git a/backend/notifications/scheduler.py b/backend/notifications/scheduler.py deleted file mode 100644 index 2677f33..0000000 --- a/backend/notifications/scheduler.py +++ /dev/null @@ -1,68 +0,0 @@ -"""Recurring worker-tier jobs: keep the curated city set warm and ping IndexNow -on a schedule, so neither needs a human to run it after every deploy anymore. - -Only the process that already won leader election for the notifier runs this -too (gated by the same check in web/app.py's _lifespan) — APScheduler has no -cross-host coordination of its own, so running it unguarded on every worker -replica would repeat every job once per replica, same reason the notifier -itself needs a leader. - -APScheduler, not a Postgres-native queue: there is exactly one process that -ever runs this, no fan-out/backpressure/dead-letter need, and it needs no new -infrastructure. procrastinate/pgqueuer are the documented upgrade path if that -changes — see docs/architecture/repo-topology-and-infrastructure.md §7. -""" -import os - -from apscheduler.schedulers.background import BackgroundScheduler - -# How often the curated city set is (re-)warmed. warm_cities.py's own docstring -# frames it as a deploy-time script ("Run at/after deploy") because a warm cache -# mostly just stays warm; scheduling it periodically instead catches a newly -# added city, or one that fell out of cache, without waiting for the next -# deploy. Daily is plenty — every already-cached city is a cheap skip. -WARM_CITIES_INTERVAL_HOURS = int(os.environ.get("THERMOGRAPH_WARM_CITIES_INTERVAL_HOURS", "24") or 24) - -# How often to check whether the indexable URL set changed and, if so, ping -# IndexNow. submit_if_changed() is a cheap no-op when nothing changed, so this -# can run more often than warm_cities without spending anything on most ticks. -INDEXNOW_INTERVAL_HOURS = int(os.environ.get("THERMOGRAPH_INDEXNOW_INTERVAL_HOURS", "6") or 6) - -_scheduler: "BackgroundScheduler | None" = None - - -def _warm_cities_job() -> None: - import warm_cities # local import: a script-style top-level module — only - # pay its import cost on the process that runs the job - warm_cities.main() - - -def _indexnow_job() -> None: - import indexnow - indexnow.submit_if_changed() - - -def start() -> None: - """Start the scheduler (idempotent, mirroring notify.start()). The caller is - responsible for the leader-election gate — see web/app.py's _lifespan.""" - global _scheduler - if _scheduler is not None: - return - sched = BackgroundScheduler(daemon=True) - # No next_run_time override: the default first run is one interval from now, - # not immediately — warm_cities already runs at deploy time (warm_cities.py / - # the deploy hook), so an immediate duplicate run on every worker boot/restart - # would just be wasted work. - sched.add_job(_warm_cities_job, "interval", hours=WARM_CITIES_INTERVAL_HOURS, - id="warm_cities") - sched.add_job(_indexnow_job, "interval", hours=INDEXNOW_INTERVAL_HOURS, - id="indexnow") - sched.start() - _scheduler = sched - - -def stop() -> None: - global _scheduler - if _scheduler is not None: - _scheduler.shutdown(wait=False) - _scheduler = None diff --git a/backend/requirements.txt b/backend/requirements.txt index 62fb0fc..6e8b175 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -19,9 +19,8 @@ alembic==1.14.0 pywebpush==2.0.0 # Ed25519 verification of Discord interaction webhooks (see notifications/discord_interactions.py). PyNaCl==1.5.0 -# Discord gateway bot's websocket client (notifications/discord_bot.py). Already -# pulled in transitively by uvicorn[standard]; pinned here since we import it directly. -websockets==16.0 -# Recurring worker-tier jobs — city warming, IndexNow pings (see -# notifications/scheduler.py). In-process; no broker/queue needed at this scale. -apscheduler==3.11.3 +# The Discord gateway bot and the recurring worker jobs moved to the +# thermograph-daemon Go binary (backend/daemon/), which calls back over +# api/internal_routes.py — so the direct `websockets` pin (only discord_bot.py +# imported it; uvicorn[standard] still pulls it transitively for its own ws +# protocol support) and `apscheduler` are gone from this list. diff --git a/backend/tests/api/test_internal_routes.py b/backend/tests/api/test_internal_routes.py new file mode 100644 index 0000000..3a940ef --- /dev/null +++ b/backend/tests/api/test_internal_routes.py @@ -0,0 +1,134 @@ +"""The internal control surface the thermograph-daemon Go binary calls back +into (api/internal_routes.py): shared-secret gating (fail closed when the token +was never provisioned), the gateway-ready /grade reply with the interactions-only +ephemeral flag dropped, the two job triggers, and the one-in-flight-run guard. +The underlying job/grade functions are stubbed — hermetic, no real network.""" +import pytest +from fastapi.testclient import TestClient + +from api import internal_routes +from notifications import discord_interactions as di +from web import app as appmod + +TOKEN = "test-internal-token" +HDR = {"X-Thermograph-Internal-Token": TOKEN} + + +@pytest.fixture +def client(monkeypatch): + monkeypatch.setenv("THERMOGRAPH_INTERNAL_TOKEN", TOKEN) + return TestClient(appmod.app) + + +@pytest.fixture +def grade_stub(monkeypatch): + """Reply payload for the shared grade builder, mirroring the real shape + (embeds + an ephemeral flag the gateway path must drop).""" + calls = [] + monkeypatch.setattr( + di, "_grade_message", + lambda query: calls.append(query) or {"embeds": [{"title": f"grade:{query}"}], + "flags": 64}) + return calls + + +# --- auth: fail closed, then reject before accept ---------------------------- + +def test_router_is_disabled_when_no_token_is_provisioned(monkeypatch): + """No THERMOGRAPH_INTERNAL_TOKEN => the surface doesn't exist (404), even + for a caller presenting a header — never fall open to no-auth.""" + monkeypatch.delenv("THERMOGRAPH_INTERNAL_TOKEN", raising=False) + client = TestClient(appmod.app) + assert client.post("/internal/discord/grade", json={"query": "x"}, + headers=HDR).status_code == 404 + assert client.post("/internal/jobs/warm-cities", json={}, headers=HDR).status_code == 404 + assert client.post("/internal/jobs/indexnow", json={}, headers=HDR).status_code == 404 + + +def test_missing_header_is_rejected(client): + assert client.post("/internal/jobs/indexnow", json={}).status_code == 401 + + +def test_wrong_token_is_rejected(client): + r = client.post("/internal/jobs/indexnow", json={}, + headers={"X-Thermograph-Internal-Token": "not-the-token"}) + assert r.status_code == 401 + + +def test_correct_token_is_accepted(client, monkeypatch): + import indexnow + monkeypatch.setattr(indexnow, "submit_if_changed", lambda: None) + assert client.post("/internal/jobs/indexnow", json={}, headers=HDR).status_code == 200 + + +# --- /internal/discord/grade ------------------------------------------------- + +def test_grade_returns_the_shared_builder_message_without_ephemeral_flag(client, grade_stub): + r = client.post("/internal/discord/grade", json={"query": "Phoenix"}, headers=HDR) + assert r.status_code == 200 + # Gateway messages can't be ephemeral — the flag must be gone, everything + # else exactly as _grade_message built it. + assert r.json() == {"embeds": [{"title": "grade:Phoenix"}]} + assert grade_stub == ["Phoenix"] + + +def test_grade_requires_a_query(client, grade_stub): + assert client.post("/internal/discord/grade", json={}, headers=HDR).status_code == 422 + assert grade_stub == [] + + +# --- job triggers ------------------------------------------------------------ + +def test_warm_cities_invokes_warm_cities_main(client, monkeypatch): + import warm_cities + calls = [] + monkeypatch.setattr(warm_cities, "main", lambda: calls.append(1)) + r = client.post("/internal/jobs/warm-cities", json={}, headers=HDR) + assert r.status_code == 200 + assert r.json() == {"ok": True} + assert calls == [1] + + +def test_indexnow_invokes_submit_if_changed(client, monkeypatch): + import indexnow + calls = [] + monkeypatch.setattr(indexnow, "submit_if_changed", lambda: calls.append(1)) + r = client.post("/internal/jobs/indexnow", json={}, headers=HDR) + assert r.status_code == 200 + assert r.json() == {"ok": True} + assert calls == [1] + + +# --- one-in-flight-run guard ------------------------------------------------- + +def test_concurrent_job_invocation_answers_409(client, monkeypatch): + """A second trigger while a run is in flight must not stack — warm_cities + spends the Open-Meteo quota, so a stacked run double-spends it. Holding the + job's lock stands in for an in-flight run.""" + import indexnow + monkeypatch.setattr(indexnow, "submit_if_changed", lambda: None) + lock = internal_routes._JOB_LOCKS["warm-cities"] + assert lock.acquire(blocking=False) + try: + assert client.post("/internal/jobs/warm-cities", json={}, headers=HDR).status_code == 409 + # The guard is per job: an in-flight warm run doesn't block indexnow. + assert client.post("/internal/jobs/indexnow", json={}, headers=HDR).status_code == 200 + finally: + lock.release() + # Released => the next trigger runs again. + import warm_cities + monkeypatch.setattr(warm_cities, "main", lambda: None) + assert client.post("/internal/jobs/warm-cities", json={}, headers=HDR).status_code == 200 + + +def test_job_lock_is_released_after_a_failing_run(monkeypatch): + """A crashed run must not wedge the job forever: the 500 surfaces, the lock + is released, and the next trigger runs.""" + monkeypatch.setenv("THERMOGRAPH_INTERNAL_TOKEN", TOKEN) + client = TestClient(appmod.app, raise_server_exceptions=False) + import indexnow + monkeypatch.setattr(indexnow, "submit_if_changed", + lambda: (_ for _ in ()).throw(RuntimeError("boom"))) + assert client.post("/internal/jobs/indexnow", json={}, headers=HDR).status_code == 500 + monkeypatch.setattr(indexnow, "submit_if_changed", lambda: None) + assert client.post("/internal/jobs/indexnow", json={}, headers=HDR).status_code == 200 diff --git a/backend/tests/api/test_internal_token_derivation.py b/backend/tests/api/test_internal_token_derivation.py new file mode 100644 index 0000000..77734cc --- /dev/null +++ b/backend/tests/api/test_internal_token_derivation.py @@ -0,0 +1,70 @@ +"""The internal token's derivation, and its cross-language contract with the Go daemon. + +The daemon (backend/daemon/) and this app must compute byte-identical tokens from +the same THERMOGRAPH_AUTH_SECRET. If they drift, every callback fails with 401 -- +which reads like an auth bug rather than like the configuration drift it is, so +the shared vector below is pinned on both sides. +""" +import hashlib +import hmac + +import pytest + +from api import internal_routes + +# Must equal backend/daemon/internal/config/derive_test.go's sharedVector*. +SHARED_VECTOR_SECRET = "test-auth-secret" +SHARED_VECTOR_TOKEN = "4c3830b2158941ff52720d198b65f7924372a3a482b8da52540a4d9be38247ea" + + +@pytest.fixture(autouse=True) +def _clear_token_env(monkeypatch): + """Both knobs start unset so each test states exactly the config it exercises.""" + monkeypatch.delenv("THERMOGRAPH_INTERNAL_TOKEN", raising=False) + monkeypatch.delenv("THERMOGRAPH_AUTH_SECRET", raising=False) + + +def test_derived_token_matches_go(monkeypatch): + """The pinned cross-language vector. Changing the label or construction here + requires the identical change in the Go daemon's config package.""" + monkeypatch.setenv("THERMOGRAPH_AUTH_SECRET", SHARED_VECTOR_SECRET) + assert internal_routes.resolve_internal_token() == SHARED_VECTOR_TOKEN + + +def test_vector_is_actually_hmac_of_the_label(): + """Guards against the vector and the implementation drifting together if + someone regenerates the constant from the code it is supposed to check.""" + expected = hmac.new( + SHARED_VECTOR_SECRET.encode(), + internal_routes._DERIVE_LABEL, + hashlib.sha256, + ).hexdigest() + assert expected == SHARED_VECTOR_TOKEN + + +def test_explicit_token_wins_over_derivation(monkeypatch): + monkeypatch.setenv("THERMOGRAPH_AUTH_SECRET", SHARED_VECTOR_SECRET) + monkeypatch.setenv("THERMOGRAPH_INTERNAL_TOKEN", "explicit-wins") + assert internal_routes.resolve_internal_token() == "explicit-wins" + + +def test_no_secret_at_all_leaves_the_surface_closed(): + """Neither knob set => no token => the router 404s. Fail closed, never open.""" + assert internal_routes.resolve_internal_token() == "" + + +def test_derivation_is_not_the_auth_secret_itself(monkeypatch): + """Domain separation: a leak of the internal token must not hand over the + session-signing key it was derived from.""" + monkeypatch.setenv("THERMOGRAPH_AUTH_SECRET", SHARED_VECTOR_SECRET) + token = internal_routes.resolve_internal_token() + assert token != SHARED_VECTOR_SECRET + assert SHARED_VECTOR_SECRET not in token + + +def test_different_secrets_derive_different_tokens(monkeypatch): + monkeypatch.setenv("THERMOGRAPH_AUTH_SECRET", "secret-a") + a = internal_routes.resolve_internal_token() + monkeypatch.setenv("THERMOGRAPH_AUTH_SECRET", "secret-b") + b = internal_routes.resolve_internal_token() + assert a != b diff --git a/backend/tests/notifications/test_discord_bot.py b/backend/tests/notifications/test_discord_bot.py deleted file mode 100644 index 9692542..0000000 --- a/backend/tests/notifications/test_discord_bot.py +++ /dev/null @@ -1,125 +0,0 @@ -"""Discord gateway bot: the message-handling logic that decides when to reply and -with what. No live gateway — we feed MESSAGE_CREATE payloads straight to the pure -handler and stub the shared /grade lookup so these stay data-independent.""" -import asyncio - -import pytest - -from notifications import discord -from notifications import discord_bot as bot -from notifications import discord_interactions as di - -BOT_ID = "999" - - -@pytest.fixture(autouse=True) -def _stub_grade(monkeypatch): - """Reply payload for /grade, mirroring the real shape (embeds + an ephemeral - flag the gateway path must drop).""" - monkeypatch.setattr( - di, "_grade_message", - lambda query: {"embeds": [{"title": f"grade:{query}"}], "flags": 64}) - - -def _msg(content="", *, author_id="1", bot_author=False, mentions=(), guild=True): - m = { - "content": content, - "id": "42", - "channel_id": "77", - "author": {"id": author_id, "bot": bot_author}, - "mentions": [{"id": mid} for mid in mentions], - } - if guild: - m["guild_id"] = "5" - return m - - -# --- when the bot stays silent ---------------------------------------------- - -def test_ignores_other_bots(): - assert bot._response_for_message( - _msg("<@999> Phoenix", author_id="8", bot_author=True, mentions=[BOT_ID]), BOT_ID) is None - - -def test_ignores_its_own_messages(): - assert bot._response_for_message( - _msg("hello", author_id=BOT_ID, mentions=[BOT_ID]), BOT_ID) is None - - -def test_ignores_guild_message_without_a_mention(): - assert bot._response_for_message(_msg("just chatting", mentions=[]), BOT_ID) is None - - -# --- when the bot replies ---------------------------------------------------- - -def test_mention_with_city_returns_grade_without_ephemeral_flag(): - out = bot._response_for_message(_msg("<@999> Phoenix", mentions=[BOT_ID]), BOT_ID) - assert out == {"embeds": [{"title": "grade:Phoenix"}]} # flags dropped - assert "flags" not in out - - -def test_legacy_nickname_mention_is_stripped(): - out = bot._response_for_message(_msg("<@!999> Tokyo", mentions=[BOT_ID]), BOT_ID) - assert out["embeds"][0]["title"] == "grade:Tokyo" - - -def test_dm_needs_no_mention(): - out = bot._response_for_message(_msg("Berlin", guild=False, mentions=[]), BOT_ID) - assert out["embeds"][0]["title"] == "grade:Berlin" - - -def test_mention_with_no_query_gives_help(): - out = bot._response_for_message(_msg("<@999>", mentions=[BOT_ID]), BOT_ID) - assert "content" in out and "mention me" in out["content"].lower() - - -# --- helpers ----------------------------------------------------------------- - -def test_strip_mentions_handles_both_forms(): - assert bot._strip_mentions("<@999> <@!999> Sao Paulo", BOT_ID) == "Sao Paulo" - - -def test_is_dm_is_true_without_guild_id(): - assert bot._is_dm(_msg("x", guild=False)) is True - assert bot._is_dm(_msg("x", guild=True)) is False - - -# --- _dispatch: same reply, off the event loop ------------------------------- - -def test_dispatch_delivers_the_same_reply_via_a_thread(monkeypatch): - """_dispatch now runs _response_for_message through asyncio.to_thread (the - event-loop fix) -- confirm that still produces the exact same reply and gets - it to _send_reply, i.e. the offload didn't change behaviour.""" - sent = {} - - async def _fake_send_reply(channel_id, message_id, data): - sent["channel_id"] = channel_id - sent["message_id"] = message_id - sent["data"] = data - - monkeypatch.setattr(bot, "_send_reply", _fake_send_reply) - sess = bot._Session() - sess.user_id = BOT_ID - msg = {"t": "MESSAGE_CREATE", "d": _msg("<@999> Phoenix", mentions=[BOT_ID])} - - asyncio.run(bot._dispatch(msg, sess)) - - assert sent["channel_id"] == "77" - assert sent["message_id"] == "42" - assert sent["data"] == {"embeds": [{"title": "grade:Phoenix"}]} - - -# --- enabled() gating -------------------------------------------------------- - -def test_enabled_is_off_by_default(monkeypatch): - monkeypatch.delenv("THERMOGRAPH_DISCORD_BOT", raising=False) - monkeypatch.setattr(discord, "BOT_TOKEN", "a-token") - assert bot.enabled() is False - - -def test_enabled_needs_both_flag_and_token(monkeypatch): - monkeypatch.setenv("THERMOGRAPH_DISCORD_BOT", "1") - monkeypatch.setattr(discord, "BOT_TOKEN", "") - assert bot.enabled() is False - monkeypatch.setattr(discord, "BOT_TOKEN", "a-token") - assert bot.enabled() is True diff --git a/backend/tests/notifications/test_scheduler.py b/backend/tests/notifications/test_scheduler.py deleted file mode 100644 index 3aeb929..0000000 --- a/backend/tests/notifications/test_scheduler.py +++ /dev/null @@ -1,79 +0,0 @@ -"""Tests for the recurring worker-tier jobs (scheduler.py): job registration, -cadence from env, idempotent start/stop, and that each job calls the right -underlying function. No real hours-long wait — inspects APScheduler's own job -registry rather than letting anything actually fire.""" -import datetime - -import pytest - -from notifications import scheduler - - -@pytest.fixture(autouse=True) -def _clean_scheduler(): - """Every test starts from "not running" and leaves nothing behind, even on - failure — scheduler._scheduler is module-global state shared across tests.""" - scheduler.stop() - yield - scheduler.stop() - - -def test_start_registers_both_jobs_with_configured_intervals(monkeypatch): - monkeypatch.setattr(scheduler, "WARM_CITIES_INTERVAL_HOURS", 24) - monkeypatch.setattr(scheduler, "INDEXNOW_INTERVAL_HOURS", 6) - scheduler.start() - jobs = {j.id: j for j in scheduler._scheduler.get_jobs()} - assert set(jobs) == {"warm_cities", "indexnow"} - assert jobs["warm_cities"].trigger.interval == datetime.timedelta(hours=24) - assert jobs["indexnow"].trigger.interval == datetime.timedelta(hours=6) - assert jobs["warm_cities"].func is scheduler._warm_cities_job - assert jobs["indexnow"].func is scheduler._indexnow_job - - -def test_start_does_not_run_a_job_immediately(): - """warm_cities already runs at deploy time; an immediate duplicate run on - every worker boot/restart would just be wasted work — the first scheduled - run must be one interval away, not "now".""" - scheduler.start() - now = datetime.datetime.now(datetime.timezone.utc) - for job in scheduler._scheduler.get_jobs(): - assert job.next_run_time > now - - -def test_start_is_idempotent(monkeypatch): - scheduler.start() - first = scheduler._scheduler - calls = [] - monkeypatch.setattr(scheduler.BackgroundScheduler, "__init__", - lambda self, **kw: calls.append(1) or None) - scheduler.start() - assert scheduler._scheduler is first - assert calls == [] # a second BackgroundScheduler was never constructed - - -def test_stop_shuts_down_and_clears_state(): - scheduler.start() - assert scheduler._scheduler is not None - scheduler.stop() - assert scheduler._scheduler is None - - -def test_stop_without_start_is_a_safe_no_op(): - scheduler.stop() # must not raise - assert scheduler._scheduler is None - - -def test_warm_cities_job_calls_warm_cities_main(monkeypatch): - import warm_cities - calls = [] - monkeypatch.setattr(warm_cities, "main", lambda: calls.append(1)) - scheduler._warm_cities_job() - assert calls == [1] - - -def test_indexnow_job_calls_submit_if_changed(monkeypatch): - import indexnow - calls = [] - monkeypatch.setattr(indexnow, "submit_if_changed", lambda: calls.append(1)) - scheduler._indexnow_job() - assert calls == [1] diff --git a/backend/web/app.py b/backend/web/app.py index 6953b73..0253319 100644 --- a/backend/web/app.py +++ b/backend/web/app.py @@ -1,5 +1,4 @@ """Thermograph API — grade recent local weather against ~45 years of climatology.""" -import asyncio import contextlib import datetime import hashlib @@ -19,17 +18,16 @@ from fastapi.responses import JSONResponse, RedirectResponse from accounts import api_accounts from api import content_routes +from api import internal_routes from core import audit from data import climate from notifications import digest -from notifications import discord_bot from notifications import discord_interactions as discord_interactions_mod from notifications import discord_link from accounts import db from data import grid from core import metrics from notifications import notify -from notifications import scheduler import paths from data import places from core import singleton @@ -163,33 +161,19 @@ async def _lifespan(app): # restricts this to processes that are allowed to own it at all — "web" replicas # never start it even if they'd otherwise win the leader election, so a stateless # N-replica web tier can scale without also scaling notifier instances. - bot_task = None + # The Discord gateway bot and the recurring worker jobs (city warming, + # IndexNow) used to ride this same leader gate in-process; both now live in + # the thermograph-daemon Go binary, which owns the connection/timers and + # calls back over api/internal_routes.py for anything needing data. The + # notifier is the one singleton still running here. if _should_run_notifier(): notify.start() - # Same leader gate as the notifier above — the recurring worker jobs - # (city warming, IndexNow) must likewise run in exactly one process. - scheduler.start() - # Discord gateway bot (opt-in). Discord allows one gateway connection per - # bot token, so it belongs on the same single leader; it rides this event - # loop as a background task rather than a new process. No-op unless - # THERMOGRAPH_DISCORD_BOT is set with a bot token (discord_bot.enabled()). - if discord_bot.enabled(): - bot_task = asyncio.create_task(discord_bot.run(), name="discord-bot") # 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()}) - if bot_task is not None: - bot_task.cancel() - # Wait for the cancellation to actually land before tearing anything else - # down -- otherwise shutdown races the bot coroutine's own teardown (closing - # the websocket, cancelling its heartbeat task) and the CancelledError it - # raises on its way out surfaces as unhandled-task noise in the logs. - with contextlib.suppress(asyncio.CancelledError): - await bot_task notify.stop() - scheduler.stop() await _frontend_client.aclose() @@ -838,6 +822,13 @@ app.include_router(v2, prefix=f"{BASE}/api/v2") # service (Stage 3) instead of anything in this process. See # backend/api/content_routes.py. app.include_router(content_routes.router, prefix=f"{BASE}/api/v2") +# Internal control surface for the thermograph-daemon Go binary (the gateway +# bot + recurring jobs that used to run in _lifespan). Deliberately NOT under +# BASE — the daemon hits THERMOGRAPH_API_BASE_INTERNAL directly, same posture +# as /healthz — and never publicly routed (the Caddyfile only forwards +# /api/*, /digest and /discord/interactions here). Registered before the +# catch-all frontend proxy below so /internal/* can't fall through to it. +app.include_router(internal_routes.router) # --- proxy to the frontend SSR service (repo-split Stage 4) ------------------ diff --git a/infra/.env.example b/infra/.env.example index e95b86a..3253125 100644 --- a/infra/.env.example +++ b/infra/.env.example @@ -9,6 +9,21 @@ # build the app's THERMOGRAPH_DATABASE_URL. Change it before first `up`. POSTGRES_PASSWORD=change-me +# OPTIONAL. Shared secret between the daemon service (Discord gateway + job +# timers) and the backend's /internal/* routes. +# +# Leave it empty and both ends DERIVE the same token from THERMOGRAPH_AUTH_SECRET +# (HMAC-SHA256 under a fixed label), which every environment already provisions — +# so the daemon needs no new vault entry and no operator step. Set it only to +# override that, e.g. to rotate this surface independently of the auth secret: +# openssl rand -hex 32 +# +# With neither this nor THERMOGRAPH_AUTH_SECRET set, both ends fail closed: the +# backend disables the internal routes and the daemon refuses to start. Never +# routed publicly — Caddy only forwards /api/*, /digest and +# /discord/interactions to the backend. +THERMOGRAPH_INTERNAL_TOKEN= + # --- Everything below is optional -- docker-compose.yml already defaults each # --- of these, so a plain `make up` works with none of it set. Uncomment to # --- override. diff --git a/infra/deploy/deploy.sh b/infra/deploy/deploy.sh index 4b8f8f3..5e7c451 100755 --- a/infra/deploy/deploy.sh +++ b/infra/deploy/deploy.sh @@ -186,10 +186,19 @@ export BACKEND_IMAGE_TAG="${BACKEND_IMAGE_TAG:-local}" export FRONTEND_IMAGE_TAG="${FRONTEND_IMAGE_TAG:-local}" # Which compose services this run pulls/rolls. +# +# `daemon` rides with `backend` and is never a target on its own. It runs the +# SAME image at the SAME tag (it is a second binary inside the backend image), +# and it talks to the web process over the /internal/* contract -- so rolling one +# without the other is exactly the version skew that contract has no negotiation +# for. Pairing them here is also what makes the service exist at all: a +# single-service deploy runs `up -d --no-deps `, so a `daemon` listed +# only in compose but absent from TARGETS would never be created on a +# backend-only deploy, and the gateway bot would silently never start. case "$SERVICE" in - backend) TARGETS=(backend) ;; + backend) TARGETS=(backend daemon) ;; frontend) TARGETS=(frontend) ;; - all) TARGETS=(backend frontend) ;; + all) TARGETS=(backend frontend daemon) ;; esac # Login only when a token is supplied. The SSH/CI deploy paths (deploy.yml, @@ -226,6 +235,30 @@ if [ "$pull_ok" != 1 ]; then exit 1 fi +# The daemon binary ships INSIDE the backend image, but infra and app images +# advance on different axes by design: this checkout tracks `main`, while image +# tags are env-staged (prod's come from `release`). So a host can legitimately +# be asked to roll a backend image OLDER than this compose file -- one built +# before the daemon existed and with no /usr/local/bin/thermograph-daemon in it. +# Creating the service anyway would leave a container crash-looping on a missing +# binary, on a deploy that otherwise succeeded. Probe the image we are actually +# about to roll and drop the daemon from this run if it can't support it; the +# next deploy of a tag that has it picks it up with no further action. +case " ${TARGETS[*]} " in + *" daemon "*) + daemon_img="$(docker compose config --images daemon 2>/dev/null | head -1 || true)" + if [ -n "$daemon_img" ] \ + && ! docker run --rm --entrypoint sh "$daemon_img" -c 'test -x /usr/local/bin/thermograph-daemon' 2>/dev/null; then + echo "==> $daemon_img predates the daemon binary; rolling without the daemon service this run" + kept=() + for t in "${TARGETS[@]}"; do + [ "$t" = daemon ] || kept+=("$t") + done + TARGETS=("${kept[@]}") + fi + ;; +esac + # Roll only the target service(s). Backend schema migrations run inside its own # entrypoint (alembic upgrade head) before uvicorn, so there's no separate # migrate step; frontend is stateless. diff --git a/infra/deploy/stack/thermograph-stack.yml b/infra/deploy/stack/thermograph-stack.yml index b34f77b..bccb6b9 100644 --- a/infra/deploy/stack/thermograph-stack.yml +++ b/infra/deploy/stack/thermograph-stack.yml @@ -138,6 +138,60 @@ services: restart_policy: condition: on-failure + daemon: + # SAME image and tag as web/worker on purpose: the daemon binary + # (/usr/local/bin/thermograph-daemon, built into the backend image) and the + # app share the /internal/* API contract, so one BACKEND_IMAGE_TAG rolls + # them together and they can never skew. + image: ${REGISTRY_HOST:-git.thermograph.org}/${BACKEND_IMAGE_PATH:-emi/thermograph/backend}:${BACKEND_IMAGE_TAG:?required} + # Not env-entrypoint.sh: that shim's handoff execs the image's + # deploy/entrypoint.sh (Alembic + uvicorn) whenever it exists, which is + # exactly what this service must never run — migrations belong to the + # one-shot task, and a second migrator racing it is a real hazard. Instead + # source the same host-rendered env directly (it is shell-sourceable by + # contract — deploy.sh sources it too) and exec the binary. stack.env holds + # vault secrets, never service-topology URLs, so it cannot collide with the + # `environment:` block below. + entrypoint: + - /bin/bash + - -c + - 'set -a; [ -f /host/thermograph.env ] && . /host/thermograph.env; set +a; exec /usr/local/bin/thermograph-daemon' + environment: + # The service VIP: the daemon's /internal/* calls load-balance across web + # replicas like any other client of the app. + THERMOGRAPH_API_BASE_INTERNAL: http://web:8137 + # Only the env mount — no appdata/applogs: warm-cities/IndexNow work + # executes inside the web process (the daemon just fires the internal + # endpoints on a timer), so the daemon holds no state and logs to stdout. + volumes: + - /etc/thermograph/stack.env:/host/thermograph.env:ro + networks: + - internal + deploy: + # EXACTLY 1, load-bearing — do not scale this service, ever. Discord + # permits ONE gateway connection per bot token; the Python bot enforced + # that with a leader election (core/singleton.claim_leader), and this + # pinned replica count is what REPLACES that election. A second replica + # means two gateway sessions fighting over the token (Discord kicks + # both into a reconnect storm) and every timer job firing twice. The + # autoscaler cannot touch this — it scales ${STACK_NAME}_web only. + replicas: 1 + placement: + constraints: ["node.role == manager"] + resources: + limits: + cpus: "0.5" + memory: 128m + restart_policy: + condition: on-failure + update_config: + # stop-first (the default), NOT start-first: a start-first roll would + # briefly run two daemons — two gateway sessions on one token, the + # exact thing replicas:1 exists to prevent. A few seconds of gateway + # downtime per deploy is the correct trade. + order: stop-first + failure_action: rollback + frontend: image: ${REGISTRY_HOST:-git.thermograph.org}/${FRONTEND_IMAGE_PATH:-emi/thermograph/frontend}:${FRONTEND_IMAGE_TAG:?required} entrypoint: ["/host/env-entrypoint.sh"] diff --git a/infra/deploy/thermograph.env.example b/infra/deploy/thermograph.env.example index 29a1f07..7020e2d 100644 --- a/infra/deploy/thermograph.env.example +++ b/infra/deploy/thermograph.env.example @@ -83,12 +83,35 @@ WORKERS=4 # Read once at process start; changing it needs a restart, not a live toggle. #THERMOGRAPH_ROLE=all -# The worker's own recurring jobs (city warming, IndexNow), gated by the same -# leader election as the notifier above. Hours between runs; both are cheap -# no-op skips when there's nothing to do, so the defaults rarely need changing. +# The recurring jobs (city warming, IndexNow). Hours between runs; both are +# cheap no-op skips when there's nothing to do, so the defaults rarely need +# changing. The daemon service (below) owns these timers and fires the work +# through the backend's internal routes. #THERMOGRAPH_WARM_CITIES_INTERVAL_HOURS=24 #THERMOGRAPH_INDEXNOW_INTERVAL_HOURS=6 +# --- Internal daemon<->backend API ------------------------------------------------ +# Shared secret authenticating the daemon service's calls to the backend's +# /internal/* routes (X-Thermograph-Internal-Token header). A credential: +# OPTIONAL — leave unset unless you specifically want to override the derivation. +# +# When unset, the backend and the daemon each DERIVE this token from +# THERMOGRAPH_AUTH_SECRET (HMAC-SHA256 under a fixed domain-separation label). +# That secret is already required and already in every vault, so the daemon +# stands up with NO new key to add, rotate, or forget on a new host. HMAC under a +# distinct label rather than reusing the auth secret directly, so a leak of this +# token can't be replayed as the session-signing key. +# +# Set it explicitly to rotate this surface independently of the auth secret +# (e.g. openssl rand -hex 32). On a SOPS host that means `sops edit` on +# deploy/secrets/*.yaml, never hand-editing /etc/thermograph.env — it is a +# rendered artifact. Both ends must agree, so set it on both or neither. +# +# With neither this nor THERMOGRAPH_AUTH_SECRET set, both ends fail closed: the +# backend disables the internal routes and the daemon refuses to start. Defence +# in depth: Caddy never routes /internal/* to the backend anyway. +#THERMOGRAPH_INTERNAL_TOKEN= + # Base path the app is served under. # / -> app at the domain root (Thermograph owns the whole domain — # this is what the Caddyfile expects: thermograph.org proxies "/") @@ -188,12 +211,13 @@ THERMOGRAPH_BASE_URL=https://thermograph.org # alone). Leave unset to disable. In the Thermograph.org server: # weather-events = 1529274516746932307 #THERMOGRAPH_DISCORD_WEATHER_CHANNEL= -# Discord gateway bot (@mention/DM grading): runs INSIDE the backend leader -# process (same singleton election as the notifier — see web/app.py's lifespan), -# riding its event loop; no separate bot process. Opt-in: any truthy value here -# starts it, provided BOT_TOKEN above is also set (flag alone is a silent no-op). -# Discord allows ONE gateway connection per bot token, so enable this in exactly -# one environment — prod, the only vault with the token. +# Discord gateway bot (@mention/DM grading): runs in the daemon service (the +# Go thermograph-daemon container — no longer inside the backend leader +# process). Opt-in: any truthy value here starts it, provided BOT_TOKEN above +# is also set (flag alone is a silent no-op). Discord allows ONE gateway +# connection per bot token, so enable this in exactly one environment — prod, +# the only vault with the token (the stack pins the daemon to 1 replica for +# the same reason). #THERMOGRAPH_DISCORD_BOT= # Account linking (OAuth2 "identify"): lets a signed-in user connect their Discord # account, storing their Discord user id for DM alerts. CLIENT_ID is the same App @@ -202,8 +226,9 @@ THERMOGRAPH_BASE_URL=https://thermograph.org #THERMOGRAPH_DISCORD_CLIENT_SECRET= # Gateway bot (opt-in): holds a live websocket so the bot replies to messages that # @mention it (or DM it) with a city grade — e.g. "@Thermograph Phoenix". Reuses -# THERMOGRAPH_DISCORD_BOT_TOKEN above. Runs on the single notifier leader only, so -# it needs THERMOGRAPH_ENABLE_NOTIFIER on (the default). No privileged intent -# required — Discord delivers content for mentions/DMs. Unset/0 => no gateway -# connection. (Code: thermograph-backend notifications/discord_bot.py.) +# THERMOGRAPH_DISCORD_BOT_TOKEN above. Runs in the single-replica daemon service +# (which also needs THERMOGRAPH_INTERNAL_TOKEN below — it grades via the +# backend's internal API). No privileged intent required — Discord delivers +# content for mentions/DMs. Unset/0 => no gateway connection. +# (Code: backend/daemon/, calling backend's /internal/discord/grade.) #THERMOGRAPH_DISCORD_BOT=1 diff --git a/infra/docker-compose.dev.yml b/infra/docker-compose.dev.yml index c2107fd..f136a09 100644 --- a/infra/docker-compose.dev.yml +++ b/infra/docker-compose.dev.yml @@ -37,6 +37,11 @@ services: ports: !reset null cpus: !reset null deploy: !reset null + # Same uncapped-on-dev treatment for the daemon; it has no ports to touch + # (outbound-only in every environment). + daemon: + cpus: !reset null + deploy: !reset null db: cpus: !reset null deploy: !reset null diff --git a/infra/docker-compose.yml b/infra/docker-compose.yml index 4de4659..5f57cc2 100644 --- a/infra/docker-compose.yml +++ b/infra/docker-compose.yml @@ -124,6 +124,13 @@ services: # (NOT /app/data) so it never shadows the Python `data/` package -- see the # volumes: note below and thermograph-backend paths.py. THERMOGRAPH_SINGLETON_LOCK: /state/notifier.lock + # Shared secret for the /internal/* routes the daemon service calls. + # Interpolated (like POSTGRES_PASSWORD) so local runs get it from the + # repo-root .env; prod/beta have it in /etc/thermograph.env, which + # deploy.sh sources before compose so this resolves there too. Empty => + # the backend disables the routes and the daemon refuses to start -- fail + # closed on both ends. + THERMOGRAPH_INTERNAL_TOKEN: ${THERMOGRAPH_INTERNAL_TOKEN:-} # Prod secrets live in /etc/thermograph.env: POSTGRES_PASSWORD, # THERMOGRAPH_AUTH_SECRET, THERMOGRAPH_VAPID_PRIVATE_KEY/_PUBLIC_KEY, # THERMOGRAPH_COOKIE_SECURE=1, mail/Discord keys, ... (see @@ -157,6 +164,62 @@ services: - "127.0.0.1:8137:8137" restart: unless-stopped + daemon: + # The SAME image and tag as backend, on purpose: the daemon (Go, built into + # the backend image as /usr/local/bin/thermograph-daemon) and the backend + # share the /internal/* API contract, so they must roll together — a + # separate tag could skew them. It owns the Discord gateway websocket and + # the recurring-job timers; anything needing data calls back into backend. + image: ${REGISTRY_HOST:-git.thermograph.org}/${BACKEND_IMAGE_PATH:-emi/thermograph/backend}:${BACKEND_IMAGE_TAG:-local} + # Bypass deploy/entrypoint.sh entirely: that script runs Alembic, and the + # backend service's boot already owns migrations — two containers racing + # `alembic upgrade head` against one database is a real hazard, not a + # nicety. The daemon binary never touches the DB. + entrypoint: ["/usr/local/bin/thermograph-daemon"] + # Its first acts are calls to backend's /internal/* routes, so wait for a + # genuinely healthy backend, not just a started container. + depends_on: + backend: + condition: service_healthy + environment: + # Same var the frontend uses for the same purpose: the backend's URL on + # the compose-internal network. + THERMOGRAPH_API_BASE_INTERNAL: http://backend:8137 + # See the backend service's note on this var — same interpolation, same + # fail-closed-when-empty behavior on both ends. Normally EMPTY: with no + # explicit value both ends derive the same token from + # THERMOGRAPH_AUTH_SECRET, which arrives via the env_file below. That is + # why this service must keep reading that file even though it serves + # nothing — without the auth secret it has nothing to derive from and + # refuses to start. + THERMOGRAPH_INTERNAL_TOKEN: ${THERMOGRAPH_INTERNAL_TOKEN:-} + # THERMOGRAPH_AUTH_SECRET (derivation input, see above), the bot token/flag + # and the job intervals (THERMOGRAPH_DISCORD_BOT[_TOKEN], + # THERMOGRAPH_*_INTERVAL_HOURS) all arrive via the host env file, same as + # the backend's secrets do — so both processes read one source of truth and + # cannot disagree about the derived token. + env_file: + - path: /etc/thermograph.env + required: false + # The image's HEALTHCHECK curls /healthz on ${PORT} — right for the backend + # process, meaningless for the daemon, which serves nothing. Without this + # the container would sit permanently "unhealthy". + healthcheck: + disable: true + # No volumes: warm-cities/IndexNow work executes inside the BACKEND process + # (the daemon only fires the internal endpoints on a timer), the parquet + # cache and locks stay on backend's appdata volume, and the daemon logs to + # stdout. It holds no state a volume would protect. + # + # No ports: outbound-only (Discord gateway + calls to backend). Publishing + # anything here would only widen the surface for no benefit. + cpus: 0.5 + deploy: + resources: + limits: + cpus: "0.5" + restart: unless-stopped + frontend: # Frontend's OWN image, published by thermograph-frontend's build-push.yml # from its own Dockerfile (which starts `uvicorn app:app` directly -- no