daemon: move the Discord gateway and scheduler out of the web process into Go
All checks were successful
shell-lint / shellcheck (pull_request) Successful in 8s
PR build (required check) / build-backend (pull_request) Successful in 1m18s
PR build (required check) / changes (pull_request) Successful in 6s
PR build (required check) / build-frontend (pull_request) Has been skipped
PR build (required check) / gate (pull_request) Successful in 3s
secrets-guard / encrypted (pull_request) Successful in 5s
PR build (required check) / validate-observability (pull_request) Has been skipped

web/app.py started two long-lived background jobs under a leader election: the
Discord gateway bot and an APScheduler. Both are stateful I/O loops -- reconnect,
RESUME, heartbeat, backoff, interval timers -- living inside an async web app
that also has to serve requests. This moves them into a single Go binary.

Go owns ONLY the stateful I/O. It owns no climate or grading logic: anything
needing data calls back into Python over a new internal-only HTTP surface
(/internal/discord/grade, /internal/jobs/warm-cities, /internal/jobs/indexnow).
Grading depends on polars and the parquet cache; reimplementing it in Go would
make the bot's grades drift from the API's, and the slash-command path
deliberately shares one grade builder so the two can never disagree. The grade
route returns gateway-ready JSON -- including the ephemeral-flag drop that
discord_bot.py used to do -- and Go relays those bytes verbatim without parsing
the embed.

Packaging: the binary is built by a golang:1.26 stage in the backend Dockerfile
and shipped in the SAME image, run as a second compose service off the SAME tag.
The daemon and backend share the /internal/* contract, so they must never skew
versions; one image makes that structural rather than a convention. Its
entrypoint bypasses entrypoint.sh -- the backend owns alembic, and two racing
migrators is a real hazard.

replicas: 1 in the Swarm stack is load-bearing. Discord permits exactly one
gateway connection per bot token; the pin replaces core/singleton.claim_leader
for this workload. update_config uses order: stop-first, since start-first would
briefly run two gateways. autoscale.sh targets ${STACK_NAME}_web only, so it
cannot scale this.

Security: the internal routes compare the token with hmac.compare_digest and the
whole router 404s when THERMOGRAPH_INTERNAL_TOKEN is unset -- fail closed, never
default open. Caddy only routes /api/*, /digest and /discord/interactions to the
backend, so /internal/* was never publicly reachable; the token is defence in
depth. The router mounts before the catch-all frontend proxy so /internal/*
cannot fall through to it. The daemon refuses to start without the token.

Behaviour preserved from the Python, with the reasoning carried into the Go
comments: non-privileged intents (no MESSAGE_CONTENT, so no portal review);
fatal close codes 4004/4010-4014 stop rather than loop; the bot-author and
self-author mention-loop guard; allowed_mentions locked to {"parse":[],
"replied_user":true} so a crafted query cannot turn a reply into an @everyone
ping; the first cron tick deferred one full interval rather than firing at boot,
since warm-cities already runs at deploy time; and no overlapping warm-cities
run, which would double-spend the archive-fetch quota.

Two deliberate improvements over the Python. A close intended for RESUME now
uses 4000 rather than 1000 -- Discord invalidates a session closed 1000/1001, so
the Python's default close silently defeated its own resume. And MESSAGE_CREATE
is handled on a bounded worker pool rather than an unbounded thread hand-off, so
a flood of mentions cannot spawn unbounded work against the backend.

A .dockerignore is added because a disposable backend/.venv was being swallowed
by COPY . /app/ and duplicated again by the chown layer, inflating the image to
1.8 GB; it builds at 578 MB.

Tests: 29 Go gateway tests covering every behaviour the deleted
test_discord_bot.py asserted, plus cron/config/apiclient suites; 10 new Python
tests for the internal routes (fail-closed, auth, flag drop, per-job 409 guard).
Full suite 359 passed / 7 skipped; go build, vet and test -race clean.
This commit is contained in:
Emi Griffith 2026-07-23 15:33:11 -07:00
parent de8e847f9f
commit c3b906a9ed
35 changed files with 3025 additions and 611 deletions

10
backend/.dockerignore Normal file
View file

@ -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/

View file

@ -1,5 +1,25 @@
# Thermograph backend: FastAPI API + accounts/notifications + the SSR content # Thermograph backend: FastAPI API + accounts/notifications + the SSR content
# JSON API frontend consumes. Split from the monorepo (repo-split Stage 7). # 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 FROM python:3.12-slim
# curl is only for the container HEALTHCHECK below. Everything Python needs # 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 COPY requirements.txt /tmp/requirements.txt
RUN pip install --no-cache-dir -r /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/ COPY . /app/
RUN chmod +x /app/deploy/entrypoint.sh RUN chmod +x /app/deploy/entrypoint.sh

View file

@ -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)

50
backend/daemon/README.md Normal file
View file

@ -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.

5
backend/daemon/go.mod Normal file
View file

@ -0,0 +1,5 @@
module thermograph/daemon
go 1.26
require github.com/coder/websocket v1.8.15

2
backend/daemon/go.sum Normal file
View file

@ -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=

View file

@ -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
}

View file

@ -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)
}
}

View file

@ -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))
}

View file

@ -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)
}
})
}
}

View file

@ -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")
}
}

View file

@ -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:
}
}
}
}

View file

@ -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")
}
}

View file

@ -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)
}
}
}

View file

@ -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)
}
})
}
}

View file

@ -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
}

View file

@ -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")
}
}

View file

@ -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)
}
}

View file

@ -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")
}
}

130
backend/daemon/main.go Normal file
View file

@ -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
}

View file

@ -179,7 +179,8 @@ def _write_state(sig: str) -> None:
def submit_if_changed(base_url: str | None = None) -> dict | 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 """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 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 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 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).""" submission that leaves the prior state in place so the next attempt retries)."""

View file

@ -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)

View file

@ -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

View file

@ -19,9 +19,8 @@ alembic==1.14.0
pywebpush==2.0.0 pywebpush==2.0.0
# Ed25519 verification of Discord interaction webhooks (see notifications/discord_interactions.py). # Ed25519 verification of Discord interaction webhooks (see notifications/discord_interactions.py).
PyNaCl==1.5.0 PyNaCl==1.5.0
# Discord gateway bot's websocket client (notifications/discord_bot.py). Already # The Discord gateway bot and the recurring worker jobs moved to the
# pulled in transitively by uvicorn[standard]; pinned here since we import it directly. # thermograph-daemon Go binary (backend/daemon/), which calls back over
websockets==16.0 # api/internal_routes.py — so the direct `websockets` pin (only discord_bot.py
# Recurring worker-tier jobs — city warming, IndexNow pings (see # imported it; uvicorn[standard] still pulls it transitively for its own ws
# notifications/scheduler.py). In-process; no broker/queue needed at this scale. # protocol support) and `apscheduler` are gone from this list.
apscheduler==3.11.3

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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]

View file

@ -1,5 +1,4 @@
"""Thermograph API — grade recent local weather against ~45 years of climatology.""" """Thermograph API — grade recent local weather against ~45 years of climatology."""
import asyncio
import contextlib import contextlib
import datetime import datetime
import hashlib import hashlib
@ -19,17 +18,16 @@ from fastapi.responses import JSONResponse, RedirectResponse
from accounts import api_accounts from accounts import api_accounts
from api import content_routes from api import content_routes
from api import internal_routes
from core import audit from core import audit
from data import climate from data import climate
from notifications import digest from notifications import digest
from notifications import discord_bot
from notifications import discord_interactions as discord_interactions_mod from notifications import discord_interactions as discord_interactions_mod
from notifications import discord_link from notifications import discord_link
from accounts import db from accounts import db
from data import grid from data import grid
from core import metrics from core import metrics
from notifications import notify from notifications import notify
from notifications import scheduler
import paths import paths
from data import places from data import places
from core import singleton 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 # 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 # 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. # 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(): if _should_run_notifier():
notify.start() 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. # Deploy/restart marker: one line correlates a metric shift with a boot.
audit.log_activity("app.start", {"version": app.version, "role": ROLE, "base": BASE, audit.log_activity("app.start", {"version": app.version, "role": ROLE, "base": BASE,
"notifier": _should_run_notifier(), "pid": os.getpid()}) "notifier": _should_run_notifier(), "pid": os.getpid()})
yield yield
audit.log_activity("app.stop", {"role": ROLE, "pid": os.getpid()}) 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() notify.stop()
scheduler.stop()
await _frontend_client.aclose() 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 # service (Stage 3) instead of anything in this process. See
# backend/api/content_routes.py. # backend/api/content_routes.py.
app.include_router(content_routes.router, prefix=f"{BASE}/api/v2") 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) ------------------ # --- proxy to the frontend SSR service (repo-split Stage 4) ------------------

View file

@ -9,6 +9,21 @@
# build the app's THERMOGRAPH_DATABASE_URL. Change it before first `up`. # build the app's THERMOGRAPH_DATABASE_URL. Change it before first `up`.
POSTGRES_PASSWORD=change-me 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 # --- 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 # --- of these, so a plain `make up` works with none of it set. Uncomment to
# --- override. # --- override.

View file

@ -186,10 +186,19 @@ export BACKEND_IMAGE_TAG="${BACKEND_IMAGE_TAG:-local}"
export FRONTEND_IMAGE_TAG="${FRONTEND_IMAGE_TAG:-local}" export FRONTEND_IMAGE_TAG="${FRONTEND_IMAGE_TAG:-local}"
# Which compose services this run pulls/rolls. # 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 <targets>`, 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 case "$SERVICE" in
backend) TARGETS=(backend) ;; backend) TARGETS=(backend daemon) ;;
frontend) TARGETS=(frontend) ;; frontend) TARGETS=(frontend) ;;
all) TARGETS=(backend frontend) ;; all) TARGETS=(backend frontend daemon) ;;
esac esac
# Login only when a token is supplied. The SSH/CI deploy paths (deploy.yml, # Login only when a token is supplied. The SSH/CI deploy paths (deploy.yml,

View file

@ -138,6 +138,60 @@ services:
restart_policy: restart_policy:
condition: on-failure 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: frontend:
image: ${REGISTRY_HOST:-git.thermograph.org}/${FRONTEND_IMAGE_PATH:-emi/thermograph/frontend}:${FRONTEND_IMAGE_TAG:?required} image: ${REGISTRY_HOST:-git.thermograph.org}/${FRONTEND_IMAGE_PATH:-emi/thermograph/frontend}:${FRONTEND_IMAGE_TAG:?required}
entrypoint: ["/host/env-entrypoint.sh"] entrypoint: ["/host/env-entrypoint.sh"]

View file

@ -83,12 +83,35 @@ WORKERS=4
# Read once at process start; changing it needs a restart, not a live toggle. # Read once at process start; changing it needs a restart, not a live toggle.
#THERMOGRAPH_ROLE=all #THERMOGRAPH_ROLE=all
# The worker's own recurring jobs (city warming, IndexNow), gated by the same # The recurring jobs (city warming, IndexNow). Hours between runs; both are
# leader election as the notifier above. Hours between runs; both are cheap # cheap no-op skips when there's nothing to do, so the defaults rarely need
# no-op skips when there's nothing to do, so the defaults rarely need changing. # 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_WARM_CITIES_INTERVAL_HOURS=24
#THERMOGRAPH_INDEXNOW_INTERVAL_HOURS=6 #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. # Base path the app is served under.
# / -> app at the domain root (Thermograph owns the whole domain — # / -> app at the domain root (Thermograph owns the whole domain —
# this is what the Caddyfile expects: thermograph.org proxies "/") # 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: # alone). Leave unset to disable. In the Thermograph.org server:
# weather-events = 1529274516746932307 # weather-events = 1529274516746932307
#THERMOGRAPH_DISCORD_WEATHER_CHANNEL= #THERMOGRAPH_DISCORD_WEATHER_CHANNEL=
# Discord gateway bot (@mention/DM grading): runs INSIDE the backend leader # Discord gateway bot (@mention/DM grading): runs in the daemon service (the
# process (same singleton election as the notifier — see web/app.py's lifespan), # Go thermograph-daemon container — no longer inside the backend leader
# riding its event loop; no separate bot process. Opt-in: any truthy value here # process). Opt-in: any truthy value here starts it, provided BOT_TOKEN above
# starts it, provided BOT_TOKEN above is also set (flag alone is a silent no-op). # is also set (flag alone is a silent no-op). Discord allows ONE gateway
# Discord allows ONE gateway connection per bot token, so enable this in exactly # connection per bot token, so enable this in exactly one environment — prod,
# one environment — prod, the only vault with the token. # the only vault with the token (the stack pins the daemon to 1 replica for
# the same reason).
#THERMOGRAPH_DISCORD_BOT= #THERMOGRAPH_DISCORD_BOT=
# Account linking (OAuth2 "identify"): lets a signed-in user connect their Discord # 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 # 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= #THERMOGRAPH_DISCORD_CLIENT_SECRET=
# Gateway bot (opt-in): holds a live websocket so the bot replies to messages that # 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 # @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 # THERMOGRAPH_DISCORD_BOT_TOKEN above. Runs in the single-replica daemon service
# it needs THERMOGRAPH_ENABLE_NOTIFIER on (the default). No privileged intent # (which also needs THERMOGRAPH_INTERNAL_TOKEN below — it grades via the
# required — Discord delivers content for mentions/DMs. Unset/0 => no gateway # backend's internal API). No privileged intent required — Discord delivers
# connection. (Code: thermograph-backend notifications/discord_bot.py.) # content for mentions/DMs. Unset/0 => no gateway connection.
# (Code: backend/daemon/, calling backend's /internal/discord/grade.)
#THERMOGRAPH_DISCORD_BOT=1 #THERMOGRAPH_DISCORD_BOT=1

View file

@ -37,6 +37,11 @@ services:
ports: !reset null ports: !reset null
cpus: !reset null cpus: !reset null
deploy: !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: db:
cpus: !reset null cpus: !reset null
deploy: !reset null deploy: !reset null

View file

@ -124,6 +124,13 @@ services:
# (NOT /app/data) so it never shadows the Python `data/` package -- see the # (NOT /app/data) so it never shadows the Python `data/` package -- see the
# volumes: note below and thermograph-backend paths.py. # volumes: note below and thermograph-backend paths.py.
THERMOGRAPH_SINGLETON_LOCK: /state/notifier.lock 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, # Prod secrets live in /etc/thermograph.env: POSTGRES_PASSWORD,
# THERMOGRAPH_AUTH_SECRET, THERMOGRAPH_VAPID_PRIVATE_KEY/_PUBLIC_KEY, # THERMOGRAPH_AUTH_SECRET, THERMOGRAPH_VAPID_PRIVATE_KEY/_PUBLIC_KEY,
# THERMOGRAPH_COOKIE_SECURE=1, mail/Discord keys, ... (see # THERMOGRAPH_COOKIE_SECURE=1, mail/Discord keys, ... (see
@ -157,6 +164,62 @@ services:
- "127.0.0.1:8137:8137" - "127.0.0.1:8137:8137"
restart: unless-stopped 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:
# Frontend's OWN image, published by thermograph-frontend's build-push.yml # Frontend's OWN image, published by thermograph-frontend's build-push.yml
# from its own Dockerfile (which starts `uvicorn app:app` directly -- no # from its own Dockerfile (which starts `uvicorn app:app` directly -- no