257 lines
8.7 KiB
Go
257 lines
8.7 KiB
Go
// Command tmplatformd is the TextMachine control plane: HTTP for the frontend, Postgres for the
|
|
// read model, and (from P-1 onward) a worker that supervises tmctl processes. It contains no
|
|
// translation logic: the engine is spawned, never linked (D39.81).
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"log/slog"
|
|
"net"
|
|
"net/http"
|
|
"os"
|
|
"os/signal"
|
|
"syscall"
|
|
"time"
|
|
|
|
"textmachine/platform/internal/auth"
|
|
"textmachine/platform/internal/config"
|
|
"textmachine/platform/internal/httpapi"
|
|
"textmachine/platform/internal/login"
|
|
"textmachine/platform/internal/metrics"
|
|
"textmachine/platform/internal/pgstore"
|
|
"textmachine/platform/internal/reqid"
|
|
)
|
|
|
|
// sessionSweep is how often expired sessions are deleted. The table is small and the work is a
|
|
// single DELETE, so the interval is about not accumulating rows, not about load.
|
|
const sessionSweep = time.Hour
|
|
|
|
func main() {
|
|
// Structured logs on stderr, like the engine's: stdout stays free for anything machine-read.
|
|
// The handler is wrapped so every *Context call carries its request id without saying so.
|
|
log := slog.New(reqid.WithContext(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo})))
|
|
if err := run(log); err != nil {
|
|
log.Error("fatal", "err", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
func run(log *slog.Logger) error {
|
|
cfg, err := config.Load()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// Before anything else is decided: an environment-only configuration leaves no artifact to read
|
|
// back, so a deployment whose EnvironmentFile= never loaded looks exactly like one whose values
|
|
// were chosen (register row PD-114, ratified 09.08). Secrets and money amounts are named, never
|
|
// printed.
|
|
cfg.LogEffective(log)
|
|
|
|
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
|
defer stop()
|
|
|
|
// One registry for the process, whether or not anything scrapes it: the instruments are wired
|
|
// into the request path and the sweep either way, and a nil registry would make the telemetry a
|
|
// second code path that only runs in production.
|
|
telemetry := metrics.New()
|
|
|
|
var db *pgstore.Store
|
|
if cfg.DSN == "" {
|
|
// Deliberate: the process still serves liveness so a supervisor can start it before the
|
|
// database exists. /readyz is the honest signal, and it says no.
|
|
log.Warn("no TM_PLATFORM_DSN: starting without a database, /readyz will report not ready")
|
|
} else {
|
|
if cfg.Migrate {
|
|
if err := pgstore.Migrate(ctx, cfg.DSN); err != nil {
|
|
return err
|
|
}
|
|
log.Info("migrations applied")
|
|
}
|
|
if db, err = pgstore.Open(ctx, cfg.DSN); err != nil {
|
|
return err
|
|
}
|
|
defer db.Close()
|
|
}
|
|
|
|
cookies := auth.Cookies{Insecure: cfg.InsecureCookies}
|
|
if cfg.InsecureCookies {
|
|
log.Warn("TM_PLATFORM_INSECURE_COOKIES: serving the session cookie without Secure, under a dev name — never in production")
|
|
}
|
|
authn := &auth.Authenticator{
|
|
IdleTTL: cfg.SessionIdleTTL,
|
|
Cookies: cookies,
|
|
Log: log,
|
|
Deny: httpapi.ProblemHandler(http.StatusUnauthorized, "Session missing or invalid"),
|
|
}
|
|
deps := httpapi.Deps{Log: log, Auth: authn, TrustedOrigins: cfg.TrustedOrigins,
|
|
HSTS: !cfg.InsecureCookies, Observe: telemetry.Middleware()}
|
|
if db != nil {
|
|
deps.DB = db
|
|
authn.Sessions = db
|
|
go sweepSessions(ctx, db, log)
|
|
}
|
|
|
|
switch {
|
|
case !cfg.LoginEnabled():
|
|
log.Warn("no TM_PLATFORM_OIDC_ISSUER: sign-in is not mounted")
|
|
case db == nil:
|
|
// A login writes rows. Mounting it without a database would answer every attempt with a 503
|
|
// from deep inside the flow instead of saying so once, here.
|
|
return errors.New("sign-in is configured but TM_PLATFORM_DSN is not: a login needs the database")
|
|
default:
|
|
lg, err := login.New(login.Config{
|
|
Provider: cfg.OIDCProvider,
|
|
Issuer: cfg.OIDCIssuer,
|
|
ClientID: cfg.OIDCClientID,
|
|
ClientSecret: cfg.OIDCClientSecret,
|
|
RedirectURL: cfg.OIDCRedirectURL,
|
|
AfterLogin: cfg.AfterLogin,
|
|
SessionIdleTTL: cfg.SessionIdleTTL,
|
|
SessionMaxAge: cfg.SessionMaxAge,
|
|
SignupGrantMicroUSD: cfg.SignupGrantMicroUSD,
|
|
}, db, cookies, log)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
lg.SetFail(httpapi.WriteProblem)
|
|
deps.Login = lg
|
|
go sweepLogins(ctx, db, log)
|
|
}
|
|
|
|
if db != nil {
|
|
stopRunner, err := startRunner(ctx, cfg, db, log, telemetry, &deps)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer stopRunner()
|
|
}
|
|
|
|
handler, err := httpapi.New(deps)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
stopMetrics := serveMetrics(ctx, cfg.MetricsAddr, telemetry, log)
|
|
defer stopMetrics()
|
|
|
|
// A second signal must kill rather than wait: once the drain starts, the handler is
|
|
// unregistered and the next SIGTERM goes back to being fatal.
|
|
go func() {
|
|
<-ctx.Done()
|
|
stop()
|
|
}()
|
|
|
|
srv := httpapi.NewServer(cfg.Addr, handler, log)
|
|
ln, err := srv.Listen(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
log.Info("listening", "addr", ln.Addr().String())
|
|
return srv.Run(ctx, ln)
|
|
}
|
|
|
|
// serveMetrics exposes the exposition endpoint on a listener of its OWN, and returns the function
|
|
// that stops it.
|
|
//
|
|
// Separate from the API on purpose, and the reason is that this surface has no session behind it.
|
|
// The contract's surface answers 401 before 404 so that an anonymous caller cannot map it (STACK §4),
|
|
// and metrics carry operational shape — queue depth, run counts, the size of this deployment —
|
|
// which is the same class of fact. The two ways out of that are inventing a second authorization
|
|
// model for scrapers or binding the endpoint where only the host can reach it; the second is what
|
|
// every control plane does, and it is one line of deployment rather than one more thing to get
|
|
// right. Default 127.0.0.1; empty turns it off.
|
|
//
|
|
// A listener that cannot bind is a WARN and not a fatal: telemetry is how a service is watched, not
|
|
// how it serves, and taking the control plane down because a port is busy would make the watching
|
|
// more dangerous than the not-watching.
|
|
func serveMetrics(ctx context.Context, addr string, m *metrics.Metrics, log *slog.Logger) func() {
|
|
if addr == "" {
|
|
log.Warn("no TM_PLATFORM_METRICS_ADDR: this instance exposes no metrics")
|
|
return func() {}
|
|
}
|
|
mux := http.NewServeMux()
|
|
mux.Handle("GET /metrics", m.Handler())
|
|
srv := &http.Server{
|
|
Addr: addr,
|
|
Handler: mux,
|
|
// The same deadlines the API carries, and for once a WriteTimeout too: nothing here streams,
|
|
// and a scrape that hangs is a connection held for nothing.
|
|
ReadHeaderTimeout: 5 * time.Second,
|
|
ReadTimeout: 10 * time.Second,
|
|
WriteTimeout: 30 * time.Second,
|
|
IdleTimeout: 60 * time.Second,
|
|
BaseContext: func(net.Listener) context.Context { return context.WithoutCancel(ctx) },
|
|
}
|
|
var lc net.ListenConfig
|
|
ln, err := lc.Listen(ctx, "tcp", addr)
|
|
if err != nil {
|
|
log.Warn("metrics endpoint could not be served", "addr", addr, "err", err)
|
|
return func() {}
|
|
}
|
|
log.Info("serving metrics", "addr", ln.Addr().String())
|
|
go func() {
|
|
if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
|
log.Error("metrics endpoint stopped", "err", err)
|
|
}
|
|
}()
|
|
return func() {
|
|
stop, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
|
defer cancel()
|
|
_ = srv.Shutdown(stop)
|
|
}
|
|
}
|
|
|
|
// sweepSessions deletes rows past their absolute expiry (PD-7). A failed sweep is logged and
|
|
// retried on the next tick: it is housekeeping, and it must never take the service down.
|
|
func sweepSessions(ctx context.Context, db *pgstore.Store, log *slog.Logger) {
|
|
t := time.NewTicker(sessionSweep)
|
|
defer t.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-t.C:
|
|
c, cancel := context.WithTimeout(ctx, 30*time.Second)
|
|
n, err := db.SweepSessions(c, time.Now())
|
|
cancel()
|
|
switch {
|
|
case err != nil:
|
|
log.Error("session sweep failed", "err", err)
|
|
case n > 0:
|
|
log.Info("session sweep", "deleted", n)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// loginJournalRetention is how long a sign-in stays in the journal. Long enough to answer "was
|
|
// that me last month", short enough that an unauthenticated endpoint cannot grow the table without
|
|
// end.
|
|
const loginJournalRetention = 180 * 24 * time.Hour
|
|
|
|
// sweepLogins deletes abandoned authorization requests and journal entries past retention.
|
|
func sweepLogins(ctx context.Context, db *pgstore.Store, log *slog.Logger) {
|
|
t := time.NewTicker(15 * time.Minute)
|
|
defer t.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-t.C:
|
|
c, cancel := context.WithTimeout(ctx, time.Minute)
|
|
states, err := db.DeleteExpiredLoginStates(c, time.Now())
|
|
if err != nil {
|
|
log.Error("login state sweep failed", "err", err)
|
|
}
|
|
events, err := db.DeleteOldLoginEvents(c, time.Now().Add(-loginJournalRetention))
|
|
cancel()
|
|
if err != nil {
|
|
log.Error("login journal sweep failed", "err", err)
|
|
}
|
|
if states > 0 || events > 0 {
|
|
log.Info("login sweep", "states", states, "events", events)
|
|
}
|
|
}
|
|
}
|
|
}
|