From eeeef897b924d691cfb6eefef8110f15806ae9a6 Mon Sep 17 00:00:00 2001 From: heaven Date: Tue, 4 Aug 2026 23:55:30 +0300 Subject: [PATCH] Land platform P0 skeleton: module, HTTP surface with security headers, server-side sessions with CSRF, read-model migrations, NDJSON ingest interface, zone battery and stack pins --- platform/.golangci.yml | 64 ++++++ platform/BACKLOG.md | 4 + platform/Makefile | 50 +++++ platform/README.md | 17 +- platform/cmd/tmplatformd/main.go | 103 +++++++++ platform/docs/STACK_DECISIONS.md | 64 ++++++ platform/docs/platform-PROGRESS.md | 194 +++++++++++++++- platform/go.mod | 16 ++ platform/go.sum | 54 +++++ platform/internal/auth/csrf.go | 61 +++++ platform/internal/auth/csrf_test.go | 71 ++++++ platform/internal/auth/middleware.go | 74 ++++++ platform/internal/auth/middleware_test.go | 183 +++++++++++++++ platform/internal/auth/principal.go | 32 +++ platform/internal/auth/session.go | 60 +++++ platform/internal/config/config.go | 81 +++++++ platform/internal/config/config_test.go | 48 ++++ platform/internal/httpapi/middleware.go | 106 +++++++++ platform/internal/httpapi/problem.go | 35 +++ platform/internal/httpapi/server.go | 88 ++++++++ platform/internal/httpapi/server_test.go | 165 ++++++++++++++ platform/internal/ingest/decoder.go | 103 +++++++++ platform/internal/ingest/decoder_test.go | 105 +++++++++ platform/internal/ingest/events.go | 141 ++++++++++++ platform/internal/ingest/resync.go | 56 +++++ platform/internal/ingest/sink.go | 53 +++++ platform/internal/ingest/supervisor.go | 118 ++++++++++ platform/internal/ingest/supervisor_test.go | 93 ++++++++ platform/internal/pgstore/migrate.go | 54 +++++ .../pgstore/migrations/00001_identity.sql | 40 ++++ .../pgstore/migrations/00002_readmodel.sql | 212 ++++++++++++++++++ .../pgstore/migrations/00003_usage.sql | 24 ++ platform/internal/pgstore/migrations_test.go | 46 ++++ platform/internal/pgstore/pg_test.go | 206 +++++++++++++++++ platform/internal/pgstore/sessions.go | 85 +++++++ platform/internal/pgstore/store.go | 37 +++ 36 files changed, 2934 insertions(+), 9 deletions(-) create mode 100644 platform/.golangci.yml create mode 100644 platform/Makefile create mode 100644 platform/cmd/tmplatformd/main.go create mode 100644 platform/docs/STACK_DECISIONS.md create mode 100644 platform/go.sum create mode 100644 platform/internal/auth/csrf.go create mode 100644 platform/internal/auth/csrf_test.go create mode 100644 platform/internal/auth/middleware.go create mode 100644 platform/internal/auth/middleware_test.go create mode 100644 platform/internal/auth/principal.go create mode 100644 platform/internal/auth/session.go create mode 100644 platform/internal/config/config.go create mode 100644 platform/internal/config/config_test.go create mode 100644 platform/internal/httpapi/middleware.go create mode 100644 platform/internal/httpapi/problem.go create mode 100644 platform/internal/httpapi/server.go create mode 100644 platform/internal/httpapi/server_test.go create mode 100644 platform/internal/ingest/decoder.go create mode 100644 platform/internal/ingest/decoder_test.go create mode 100644 platform/internal/ingest/events.go create mode 100644 platform/internal/ingest/resync.go create mode 100644 platform/internal/ingest/sink.go create mode 100644 platform/internal/ingest/supervisor.go create mode 100644 platform/internal/ingest/supervisor_test.go create mode 100644 platform/internal/pgstore/migrate.go create mode 100644 platform/internal/pgstore/migrations/00001_identity.sql create mode 100644 platform/internal/pgstore/migrations/00002_readmodel.sql create mode 100644 platform/internal/pgstore/migrations/00003_usage.sql create mode 100644 platform/internal/pgstore/migrations_test.go create mode 100644 platform/internal/pgstore/pg_test.go create mode 100644 platform/internal/pgstore/sessions.go create mode 100644 platform/internal/pgstore/store.go diff --git a/platform/.golangci.yml b/platform/.golangci.yml new file mode 100644 index 00000000..e467960e --- /dev/null +++ b/platform/.golangci.yml @@ -0,0 +1,64 @@ +# golangci-lint configuration for the platform module. Pinned to 2.12.2 (Makefile enforces it): +# findings are version-dependent, so an unpinned linter is not a gate. +# +# The enable list is the backend's minus the linters whose finding classes this module does not +# have yet, plus the SQL ones, which it does. Exclusions are written out rather than taken from a +# preset: a preset also silences things we want to hear about. +version: "2" + +linters: + default: none + enable: + - errcheck + - govet + - ineffassign + - staticcheck + - unused + # zero-cost bug classes for an HTTP + Postgres service + - bodyclose + - copyloopvar + - durationcheck + - errorlint + - makezero + - misspell + - nilerr + - noctx # a request without a context is a request that cannot be cancelled + - rowserrcheck + - sqlclosecheck + - nolintlint # no silent suppressions + + settings: + staticcheck: + checks: + - all + - -ST1000 # default-off in golangci-lint: package comment form + - -ST1003 # default-off: naming conventions + - -ST1016 # default-off: receiver name consistency + - -ST1020 # default-off: comment form on exported methods + - -ST1021 # default-off: comment form on exported types + - -ST1022 # default-off: comment form on exported vars + nolintlint: + require-explanation: true + require-specific: true + allow-unused: false + + exclusions: + rules: + # Close on a reader or a pool handle: nothing actionable comes back. This module has no + # flush-on-Close writer, so it cannot hide a lost write. + - linters: [errcheck] + text: 'Error return value of `[^`]*\.Close` is not checked' + # noctx exists to catch an uncancellable OUTBOUND call. httptest.NewRequest builds a request + # that is served in-process and never leaves it, so the rule fires 11 times and means nothing + # here; production code stays covered. + - linters: [noctx] + path: _test\.go + text: 'httptest\.NewRequest must not be called' + +formatters: + enable: + - gofmt + +issues: + max-issues-per-linter: 0 # never truncate: a hidden tail reads as "clean" + max-same-issues: 0 diff --git a/platform/BACKLOG.md b/platform/BACKLOG.md index 9ea6e17e..a4bc09cb 100644 --- a/platform/BACKLOG.md +++ b/platform/BACKLOG.md @@ -2,6 +2,10 @@ > Ведёт зона `platform/` (решение владельца 02.08, D39.84: фронт и платформа держат СВОИ бэклоги; единый бэклог `docs/PROGRESS.md` остаётся трекером движка/полигона/доков и фронт/платформа-строк не принимает). Нормы те же: ID стабилен навсегда, каждая петля получает диспозицию. Запросы к ДВИЖКУ сюда не пишутся — они заходят строками единого бэклога через оркестратора (пример: строки 99–102). Засеян оркестратором при лендинге D39.84 — дальше правит платформа-сессия. +> **Диспозиции после P0 (04.08)** — в журнале зоны, раздел «Диспозиции бэклога зоны» +> (`docs/platform-PROGRESS.md`): П-1 начата (каркас), П-2/П-3 не трогали, П-4 черновая схема, +> П-5 форма предложена. Дублировать их здесь не стали — у строки один источник истины. + | ID | Хвост | Вес | Источник | |---|---|---|---| | П-1 | **HTTP/SSE-слой и сервисная обвязка** (экс-строка 96 единого бэклога): read-API поверх готовых `OpenReadOnly`-путей движка; SSE-события ПУШИТ воркер, фронт read-model не опрашивает (каждый read-вызов движка — дорогой ре-ингест, до строки 100 единого); аутентификация ратифицирована D39.84: одна серверная сессия в Postgres — `__Host`-кука браузеру · `Authorization: Bearer` десктопу/CLI · principal создаётся ТОЛЬКО в middleware, CSRF только на cookie-пути; ⚠ порядок деплоя: read-путь движка схему НЕ мигрирует (`store.go:135-143`, «schema vN … expects vM») — после апгрейда бинарника по каждой книге первой идёт write-команда; **форма потока ратифицирована D39.85 (`docs/research/23`):** воркер-обёртка платформы супервайзит процесс tmctl, ингестит NDJSON-события идемпотентным апсертом (run_id, seq) в Postgres (Reporting Database, не полный CQRS), SSE — из Postgres; ре-синк на обрыве — `tmctl status --json`; живой SQLite движка НЕ читать (анти-паттерн, аргументы в research/23 §4); ревью-гард: путь Go-модуля платформы никогда не вкладывать под `textmachine/backend/*` | Ф3, после контракта API (строка 95 единого) | D39.81, D39.84, D39.85, STACK_DECISIONS §5 | diff --git a/platform/Makefile b/platform/Makefile new file mode 100644 index 00000000..18cc13f8 --- /dev/null +++ b/platform/Makefile @@ -0,0 +1,50 @@ +# Platform battery as one command: `make check` is what CI calls and what a session runs before +# handing the tree over. Toolchain and linter are PINNED (never "latest"): a gate that changes +# under you on someone else's machine is not a gate. Shape mirrors backend/Makefile deliberately. + +GO ?= go +# go.mod's floor is 1.26.4 (the engine's), but the BUILD toolchain floor here is 1.26.5: it carries +# the crypto/tls and os security fixes, and this module is the one exposed to the network. +GO_MIN_VERSION := 1.26.5 +GOLANGCI_LINT ?= golangci-lint +GOLANGCI_VERSION := 2.12.2 + +.PHONY: build vet fmt lint test check tools-check vuln + +build: tools-check + $(GO) build ./... + +vet: + $(GO) vet ./... + +# `gofmt -l` exits 0 even when it names files, so the emptiness of its output is the assertion. +fmt: + @test -z "$$(gofmt -l .)" || { echo "gofmt: not formatted:"; gofmt -l .; exit 1; } + +tools-check: + @$(GO) version | grep -qE 'go1\.26\.([5-9]|[0-9]{2,})|go1\.(2[7-9]|[3-9][0-9])' || { \ + echo "Go $(GO_MIN_VERSION)+ required (security fixes in a network-facing module); got: $$($(GO) version)"; exit 1; } + @$(GOLANGCI_LINT) --version 2>/dev/null | grep -q " $(GOLANGCI_VERSION) " || { \ + echo "golangci-lint $(GOLANGCI_VERSION) required (findings are version-dependent)."; \ + echo "install: https://github.com/golangci/golangci-lint/releases/tag/v$(GOLANGCI_VERSION)"; exit 1; } + +lint: tools-check + $(GOLANGCI_LINT) run --timeout=10m ./... + +# -race needs cgo. If the C toolchain is missing this fails loudly rather than quietly proving less. +test: + $(GO) test ./... -race -count=1 + +# The battery. It ends by NAMING the tests that did not run: the database-backed ones skip without +# TM_PLATFORM_TEST_DSN, and a silent skip reads as coverage. +check: build vet fmt lint test + @echo "--- did NOT run (no database; set TM_PLATFORM_TEST_DSN) ---" + @$(GO) test ./... -count=1 -v > .skips.log 2>&1 || { echo "the skip-harvest pass FAILED:"; \ + grep -E '^(---|\s+---) FAIL|^FAIL' .skips.log; rm -f .skips.log; exit 1; } + @grep -- '--- SKIP' .skips.log || echo "(none)" + @rm -f .skips.log + +# Not part of `check`: it needs the network (the vulnerability database), and the battery must be +# green on a bare clone offline. CI runs it as its own step (STACK_DECISIONS §5). +vuln: + $(GO) run golang.org/x/vuln/cmd/govulncheck@v1.6.0 ./... diff --git a/platform/README.md b/platform/README.md index ca6d12d9..aab12912 100644 --- a/platform/README.md +++ b/platform/README.md @@ -1,6 +1,12 @@ # platform — control plane (SaaS-слой) -Зона записи сессии «Платформа». Кода ещё нет; **активный промт — `docs/PLATFORM_SESSION_PROMPT.md` (P0, выдан 04.08, D39.100)**, зонный журнал — `docs/platform-PROGRESS.md` (весь прогресс зоны здесь, решение владельца 04.08). +Зона записи сессии «Платформа». P0 собран 04.08 (скелет: HTTP · сессии · схема read-model · +интерфейс ингеста); **активный промт — `docs/PLATFORM_SESSION_PROMPT.md`**, зонный журнал — +`docs/platform-PROGRESS.md` (весь прогресс зоны здесь, решение владельца 04.08), стек — +`docs/STACK_DECISIONS.md`. + +Батарея зоны: `make check` (build · vet · fmt · lint · test -race). Тесты со схемой требуют +`TM_PLATFORM_TEST_DSN`; без него они пропускаются, и `check` называет пропуски вслух. ## ⚠ Git и зона (читать ДО первой строки кода) @@ -42,11 +48,12 @@ ## Стек -Пины и обоснования — [`../frontend/docs/STACK_DECISIONS.md`](../frontend/docs/STACK_DECISIONS.md) §5 -(общий документ решений по обоим новым сервисам; исследование 02.08). +Пины, даты релизов и обоснования — [`docs/STACK_DECISIONS.md`](docs/STACK_DECISIONS.md) (зонный, +live-сверка 04.08); общая записка по обоим новым сервисам — `../frontend/docs/STACK_DECISIONS.md` §5. -Коротко: Go 1.26.4 · стандартный `net/http` + `ServeMux` без роутер-библиотеки · pgx v5.10.0 · -goose v3.27.3 · очередь River v0.42.0 на том же Postgres · `govulncheck` гейтом CI. +Коротко: Go 1.26.4 в `go.mod` (тулчейн сборки ≥1.26.5) · стандартный `net/http` + `ServeMux` без +роутер-библиотеки · PostgreSQL 18 · pgx v5.10.0 · goose v3.27.3 · очередь River v0.42.0 на том же +Postgres (запинена, ещё не подключена — П-3) · `govulncheck` отдельной целью. **Redis не заводим нигде** — зафиксировано как архитектурное «нет». Прогресс наружу — SSE, события **пушит воркер**, а не фронт опрашивает read-model. diff --git a/platform/cmd/tmplatformd/main.go b/platform/cmd/tmplatformd/main.go new file mode 100644 index 00000000..856664ce --- /dev/null +++ b/platform/cmd/tmplatformd/main.go @@ -0,0 +1,103 @@ +// 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/pgstore" +) + +func main() { + // Structured logs on stderr, like the engine's: stdout stays free for anything machine-read. + log := slog.New(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 + } + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + 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() + } + + authn := &auth.Authenticator{ + IdleTTL: cfg.SessionIdleTTL, + Deny: httpapi.ProblemHandler(http.StatusUnauthorized, "Session missing or invalid"), + } + deps := httpapi.Deps{Log: log, Auth: authn, TrustedOrigins: cfg.TrustedOrigins} + if db != nil { + deps.DB = db + authn.Sessions = db + } + handler, err := httpapi.New(deps) + if err != nil { + return err + } + + srv := &http.Server{ + Addr: cfg.Addr, + Handler: handler, + // No WriteTimeout: the SSE stream (P-1) is a long-lived response, and a write deadline set + // here would cut it. Per-request deadlines belong on the handlers that want them. + ReadHeaderTimeout: 10 * time.Second, + IdleTimeout: 2 * time.Minute, + MaxHeaderBytes: 1 << 16, + BaseContext: func(net.Listener) context.Context { return ctx }, + } + + errc := make(chan error, 1) + go func() { + log.Info("listening", "addr", cfg.Addr) + errc <- srv.ListenAndServe() + }() + + select { + case err := <-errc: + if errors.Is(err, http.ErrServerClosed) { + return nil + } + return err + case <-ctx.Done(): + stop() // a second signal now kills instead of waiting + shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + log.Info("shutting down") + return srv.Shutdown(shutdownCtx) + } +} diff --git a/platform/docs/STACK_DECISIONS.md b/platform/docs/STACK_DECISIONS.md new file mode 100644 index 00000000..f2d28edf --- /dev/null +++ b/platform/docs/STACK_DECISIONS.md @@ -0,0 +1,64 @@ +# Стек платформы — пины и обоснования + +> Зонный документ `platform/`. Пины ниже сверены ЖИВЬЁМ 04.08.2026 (Go-прокси `@latest`, +> postgresql.org, go.dev/dl) — версии по памяти не называются. Библиотеки сессия не ратифицирует: +> таблица уходит оркестратору вместе с деревом. +> +> Общая записка по обоим новым сервисам — `frontend/docs/STACK_DECISIONS.md` §5 (02.08). Здесь — +> платформенная часть с датами релизов и сверкой. Что изменилось за два дня: три библиотечных пина +> §5 (pgx · goose · River) на 04.08 всё ещё последние; по Go последним патчем стенда идёт 1.26.5 +> (07.07) — floor `go.mod` оставлен общим с движком, а тулчейн сборки поднят до 1.26.5 (см. ниже). + +## Пины + +| Что | Пин | Релиз пина | Зачем нам | +|---|---|---|---| +| Go (язык, `go.mod`) | **1.26.4** | 02.06.2026 | Тот же floor, что у движка (`backend/go.mod`) — общий стенд собирает оба модуля одним тулчейном | +| Go (тулчейн сборки, `make tools-check`) | **≥1.26.5** | 07.07.2026 | 1.26.5 несёт security-фиксы `crypto/tls` и `os`; сетевой модуль собирается ими, а не «чем-нибудь 1.26» | +| PostgreSQL | **18.x** (проверено на 18.4), floor **16** | 18.4 — май 2026 | 18 — текущая мажорная (19 в бете, в прод не берём); floor 16, потому что River тестируется на трёх последних мажорных | +| HTTP | stdlib `net/http` + `ServeMux` | — | Роутер-библиотека не нужна: `ServeMux` с 1.22 умеет метод+wildcards, а `Request.Pattern` даёт лог по маршруту, не по пути | +| CSRF | stdlib `http.CrossOriginProtection` | Go 1.25 | Ровно тот механизм, что описан в §5 (Sec-Fetch-Site → Origin), теперь в тулчейне — свой велосипед не пишем | +| Postgres-драйвер | `github.com/jackc/pgx/v5` **v5.10.0** | 03.06.2026 | Живой pool, `pgconn.PgError` для проверки констрейнтов, `stdlib` для goose | +| Миграции | `github.com/pressly/goose/v3` **v3.27.3** | 22.07.2026 | Библиотекой + `embed.FS`; `WithSessionLocker` = advisory-лок, две реплики выкатываются по очереди | +| Очередь | `github.com/riverqueue/river` **v0.42.0** | 31.07.2026 | Пин ПОДТВЕРЖДЁН живой сверкой, но **в `go.mod` НЕ добавлен**: П-3 вне скоупа P0, а зависимость без кода — мусор в графе | +| Линтер | `golangci-lint` **2.12.2** | 06.05.2026 | Тот же пин, что у движка: находки версионно-зависимы, разъезд пинов = разные гейты в одном репо | +| Уязвимости | `govulncheck` **v1.6.0** | 09.07.2026 | Отдельная цель `make vuln`, не часть `check`: ей нужна сеть, а батарея обязана быть зелёной на голом клоне офлайн | + +**Redis нет** — архитектурное «нет» из §5 в силе: очередь, лизы и рейт-лимиты живут в том же Postgres. + +## Что решено этой сессией (сверх §5) + +1. **`/healthz` ≠ `/readyz`.** Liveness ничего не трогает (БД лежит — процесс жив), readiness пингует + пул. Пустой `TM_PLATFORM_DSN` — легальный старт: сервис поднимается и честно говорит «не готов». + Иначе супервизор убивает здоровый процесс за то, что база моргнула. +2. **Ops-эндпоинты вне версионного префикса.** `/healthz`, `/readyz` — в корне; контрактная + поверхность целиком под `/v0` (базовый путь спеки платформа ПОДТВЕРЖДАЕТ). +3. **Один mux.** Контрактные маршруты регистрируются с префиксом в паттерне, а не вложенным mux'ом + под `StripPrefix`: вложенный получает КОПИЮ запроса, и `Request.Pattern` наружу не возвращается — + лог пришлось бы писать по сырому пути с id книг. Проверено исполнением. +4. **Гард навешен на поддерево, а не на ручки.** Неизвестный путь под `/v0` отвечает 401 раньше 404: + аноним не должен картографировать поверхность. +5. **Заголовки безопасности — на каждом ответе** (`X-Robots-Tag: noindex`, `Cache-Control: no-store`, + `nosniff`, `no-referrer`): ПТ-34 нельзя оставлять на дисциплину автора следующей ручки. +6. **Батарея зоны — `make check`** (build · vet · fmt · lint · test -race), форма скопирована с + `backend/Makefile` вплоть до именования пропущенных тестов: тихий skip читается как покрытие. +7. **Тесты с БД гейтятся `TM_PLATFORM_TEST_DSN`** и создают СВОЮ базу на прогон (дропают в + `t.Cleanup`). Батарея на голом клоне зелёная и офлайн; с DSN — та же батарея плюс схема. + +## Как поднять локально + +```sh +export TM_PLATFORM_DSN='postgres://user@host:5432/tmplatform?sslmode=disable' +make check # батарея зоны +TM_PLATFORM_MIGRATE=1 go run ./cmd/tmplatformd # миграции + сервер на 127.0.0.1:8080 +curl -s localhost:8080/healthz # ok +curl -s localhost:8080/readyz # ready +``` + +Переменные: `TM_PLATFORM_ADDR` · `TM_PLATFORM_DSN` · `TM_PLATFORM_MIGRATE` · +`TM_PLATFORM_TRUSTED_ORIGINS` (через запятую) · `TM_PLATFORM_SESSION_IDLE` · +`TM_PLATFORM_SESSION_MAX_AGE`. + +⚠ Postgres на стенде отсутствует как системный пакет и sudo нет. Схема и запросы этой сессии +проверены на ЖИВОМ PostgreSQL **18.4**, поднятом без root из бинарников zonky +(`io.zonky.test.postgres`, Maven Central) в скрэтчпаде — вне репозитория и вне зависимостей модуля. diff --git a/platform/docs/platform-PROGRESS.md b/platform/docs/platform-PROGRESS.md index b50a1ead..8961784f 100644 --- a/platform/docs/platform-PROGRESS.md +++ b/platform/docs/platform-PROGRESS.md @@ -6,14 +6,200 @@ ## Текущее состояние -- Кода нет. `go.mod` заведён. Промт P0 выдан 04.08 (`PLATFORM_SESSION_PROMPT.md`, D39.100). -- Контракт API v0 ратифицирован (D39.99, `docs/architecture/14-api-contract/`) — на платформе - дизайн-ответы К-4 (ревизия) · К-7 (пагинация) · К-12 (пуш экспорта) · форма П-5 (лимиты). +- **P0 собран** (сессия 04.08): модуль компилируется, батарея зоны `make check` зелёная, + `/healthz` и `/readyz` проверены живым запуском против живого Postgres 18.4. +- Стек запинен и live-сверен — `STACK_DECISIONS.md` (зонный). +- Дизайн-ответы К-4 · К-7 · К-12 · форма П-5 — ниже, ПРЕДЛОЖЕНИЯМИ на ратификацию. +- Контрактных ручек нет намеренно: они ждут ратификации К-4/К-7 (форма ответов) — это П-1. ## Открытые вопросы к владельцу/оркестратору -_(пусто)_ +1. **⚠ ВЛАДЕЛЬЦУ (П-5).** При сбросе окна лимитов приостановленный (`paused`) перевод + продолжается САМ или ждёт явного «Продолжить»? От ответа зависит, есть ли кнопка на экране и + нужно ли уведомление «продолжили без вас». Технически дёшевы оба варианта. +2. **Оркестратору (контракт).** Поток событий привязан к ПРОГОНУ (`/runs/{runId}/events`), а + статусы `uploading`/`parsing` существуют ДО прогона, и библиотека охватывает книги без прогонов. + Живого канала у них нет вовсе. Нужна либо строка в спеке «до старта прогона состояние + опрашивается», либо пользовательский поток (он же закрыл бы К-12 пушем). Решение — не наше. +3. **Оркестратору (спека).** Третий слой CSRF из STACK §5 требует от браузерного клиента + заголовок `X-TM-Client` на небезопасных запросах cookie-пути. Это требование к ФРОНТУ, и его + место — в описании `sessionCookie` в спеке. Реализовано и проверено тестами. + +## Дизайн-ответы на ратификацию + +### К-4 — ревизия: пер-ресурсная со скоупом КНИГА; чтения её несут + +Обе половины вопроса: + +1. **Чтения ревизию несут — да.** Без неё правило «отбрось чтение старше уже применённого + события» нечем реализовать, а рефетч по возврату фокуса окна включён у `react-query` по + умолчанию — то есть гонка на каждое переключение вкладки, а не редкий случай. +2. **Счётчик — один на КНИГУ.** Все книго-скоупные чтения (карточка, главы, юниты, замечания, + банк, прогон) возвращают ОДНО и то же число — `books.revision`, +1 за материализующую + транзакцию; строки, которых она коснулась, штампуются новым значением. `id` SSE-кадра прогона — + ТО ЖЕ число, поэтому события и чтения книги полностью упорядочены между собой. + +Почему книга, а не сквозной счётчик: + +- сравнивать ревизии осмысленно только внутри скоупа, а книга — минимальный скоуп, в котором + лежит всё, что поток может протухнуть; +- единственный писатель на книгу уже гарантирован (сериализация очереди по `book_id` — П-3, плюс + EXCLUSIVE-лок движка на файл проекта), поэтому счётчику не нужны ни блокировка, ни глобальная + последовательность; +- глобальный счётчик отвергнут по двум причинам: одна горячая последовательность на всех + пользователей и утечка — по разрывам номеров любой клиент оценивает активность всей платформы; +- у библиотеки (`GET /books`) свой скоуп — счётчик на пользователя (`users.library_revision`), + потому что она охватывает книги. + +Просим внести в спеку прозой: (i) `revision` монотонна В ПРЕДЕЛАХ скоупа ресурса и между скоупами +не сравнивается; (ii) кадр потока и книго-скоупные чтения несут ОДИН счётчик; (iii) отбрасывание +устаревшего чтения — обязанность клиента. + +Побочная выгода: тот же штамп даёт докачку потока «строки книги с `revision > X`» БЕЗ журнала +событий — реплей истории остаётся запрещённым (D39.85, контракт §2.11). + +### К-7 — курсор на каждом списке, дефолт «одна страница» + +Замер (сериализация фикстур контрактной формы, случайные значения — не повторяющиеся, иначе gzip +льстит): 2284 главы = **289 КБ JSON / 46 КБ gzip**; 1200 терминов = **229 КБ / 40 КБ**. Одним +ответом влезает — но китайские вебновеллы на 5000+ глав норма, а банк растёт вместе с книгой, +поэтому «всегда одним ответом» — это отложенное молчаливое обрезание. + +Предложение: **keyset-курсор на КАЖДОМ списочном ответе**, параметры `?limit=&cursor=`, поле +`next_cursor: string|null` присутствует ВСЕГДА. Дефолты: главы 5000 (обычная книга = одна +страница), банк 1000, замечания 500; юниты и библиотека курсор тоже несут, хотя практически не +пагинируются. + +- **Keyset, не offset:** материализатор пишет параллельно чтению, а offset на пишущейся таблице + пропускает и дублирует строки; keyset по `(book_id, number)` устойчив к дозаписи. +- **Поле с первого дня у всех списков — намеренно.** Добавить его позже — минорное изменение, + которое у клиента, его не читающего, молча отрезает хвост. +- Курсор непрозрачный, кодирует последний ключ сортировки и `revision`; смена `revision` между + страницами обязывает клиента начать цикл заново, иначе он склеит два состояния. + +### К-12 — опрос; причина структурная, а не вкусовая + +Единственный поток контракта привязан к ПРОГОНУ, а экспорт делают с законченной книги — живого +прогона обычно нет. Пуш завершения потребовал бы второго потока ради одного булева. + +Предложение — индустриальный async request-reply: `POST /books/{id}/exports` → `202` + +`Location`; `GET /books/{id}/exports/{id}` → `200` c `ready:false` и заголовком `Retry-After`, пока +строится, и `ready:true` + `url`, когда готов. Интервал называет СЕРВЕР, клиент не угадывает. +Просим добавить `Retry-After` в спеку. Появится пользовательский поток (вопрос 2 выше) — пуш +поедет им, опрос останется фолбэком. + +### П-5 — форма API лимитов/использования + +`GET /v0/usage` (страница лимитов в настройках): + +```json +{"revision": 42, "state": "ok|approaching|exhausted", "used_percent": 37, + "resets_at": "2026-08-11T00:00:00Z", + "windows": [{"period": "day", "used_percent": 12, "resets_at": "…"}, + {"period": "week", "used_percent": 37, "resets_at": "…"}]} +``` + +- **Сумм нет ни в каком виде.** Процент и время сброса — статус использования, а не деньги + (D39.84 в силе, механика «как Claude Code» — D39.100/ПТ-35). +- **Стоп по потолку:** `BookStatus: paused` + машинная причина. Предлагаем + `Run.paused_reason: "limits_exhausted" | null`: фразу («перевод остановлен: лимиты исчерпаны») + рисует клиент словами владельца (В-3), API несёт состояние. Без поля причины второй повод для + паузы станет ломающим изменением. +- **Источник цифр.** Поток событий денег не несёт и не должен (кадр `ceiling` — только факт), + поэтому платформа метрит из `tmctl status --json` (`committed_usd`) на границах попыток и на + ре-синке; хранит целыми микро-долларами в `usage_windows`. +- **Поднятие потолка — политика платформы, не кнопка на экране.** Платформа сама владеет + `book.yaml`, поднимает `ceilings.book_usd` и перезапускает прогон. **Проверено кодом, что это + безопасно:** `Ceilings` объявлен в `backend/internal/config/book.go:106`, а в канон `BriefHash` + (`:280-297`) НЕ входит — значит поднятие потолка не двигает `brief_hash` → снапшот и не вызывает + ни дрифт, ни ре-билл. Риск «подняли лимит — переплатили книгу заново» снят фактом, не надеждой. + +## Что построено (P0) + +| Кусок | Где | Проверено | +|---|---|---| +| Модуль, layout, батарея | `go.mod` (sibling движка, гард D39.85 соблюдён), `Makefile`, `.golangci.yml` | `make check` зелёный: build · vet · gofmt · lint 0 issues · `go test -race` | +| HTTP-скелет | `internal/httpapi/` | Живой запуск: `/healthz` 200, `/readyz` 200 против живого PG, `/v0/*` 401 problem+json, graceful shutdown по SIGTERM | +| Заголовки ПТ-34 | `internal/httpapi/middleware.go` | Живой ответ несёт `X-Robots-Tag: noindex, nofollow`, `Cache-Control: no-store`, `nosniff`, `no-referrer` | +| Сессии П-1 | `internal/auth/` + `internal/pgstore/sessions.go` | Тесты: обе презентации → principal, Bearer > cookie, истечение/отзыв/свип, токен в БД не попадает (только SHA-256), скольжение окна только во второй половине | +| CSRF | `internal/auth/csrf.go` | stdlib `http.CrossOriginProtection` + обязательный `X-TM-Client` на cookie-пути; 6 кейсов тестом + живой пробой (cookie-POST без заголовка → 403, cross-site → 403) | +| Схема read-model | `internal/pgstore/migrations/` | Миграции применены на ЖИВОМ PostgreSQL 18.4 дважды (идемпотентность), все констрейнты сработали поимённо | +| Интерфейс NDJSON-ингеста | `internal/ingest/` | Тесты: хендшейк обязателен, мажор отвергается, минор и незнакомый тип толерируются, разрыв/повтор seq ловятся; супервизор проверен НАСТОЯЩИМ процессом (exit 3 → `bank_stop`, stderr движка в файл, поток материализован) | +| Ре-синк | `internal/ingest/resync.go` | Тест на фикстуре в форме `pipeline.StatusReport` (имена полей сверены по `backend/internal/pipeline/status.go:37-130`, живого прогона не было): аллоулист берёт своё, деньги/снапшоты игнорируются | + +Не построено намеренно: материализатор `Sink → Postgres` (нужен ратифицированный словарь событий, +иначе перепишется), контрактные ручки и SSE (П-1 после ратификации К-4/К-7), очередь River (П-3), +брокер лимитов (П-2). + +## Находки (грунтованные) + +1. **Ключ идемпотентности `(run_id, seq)` работает только если `run_id` — ДВИЖКОВЫЙ.** Resume + поднимает новый процесс, его `seq` стартует с 1; если ключом взять платформенный run, high-water + mark отбросит весь поток второй попытки. Заведено в схеме: `run_attempts.engine_run_id` (unique) + + `last_seq`. Просьба к строке 103: кадр `hello` обязан нести этот id; идеально — вместе со + строкой 102 (внешний trace-контекст), тогда id назначает платформа и пространство ключей наше. +2. **Дыра контракта — статусы до прогона и библиотека без канала.** См. вопрос 2 выше. +3. **Банк: канала нет — но не полностью.** Подтверждаем находку оркестратора и уточняем состав: + сегодня добываемы (а) ПРЕДЛОЖЕННЫЕ термины стопа — сайдкар/строка 101 и (б) термины, которые + промотировала сама платформа — она же ПИШЕТ mined-delta и сид. Недобываемы `auto`/`ruby`-строки, + материализованные внутри движка. То есть `GET /bank` частично реализуем уже сейчас; полностью — + после артефакта экспорта банка. +4. **Ре-синк не восстанавливает пофазный прогресс:** в `status --json` разбивки нет (строка 99). + После обрыва и до следующего события прогресса клиент увидит агрегат. Записано в коде. +5. **`status --json` — ремонтный путь, не поллинг:** каждый вызов заново ингестит и режет исходник + (1.4–1.5 с CPU на книге 23 МБ — замер фронт-сессии 02.08, не наш; строка 100). +6. **Деньги движка живут в его stderr на уровне INFO.** Поэтому супервизор пишет stderr движка в + ФАЙЛ попытки и не тейлит его в структурный лог платформы — иначе суммы попадут в наш INFO + (запрет D39.84 + норма P0-промта). +7. **Мелочи в чужой зоне (не трогали, лендить оркестратору):** в ратифицированной копии + `docs/architecture/14-api-contract/openapi.yaml` `info.description` всё ещё называет файл + черновиком S3 и ссылается на `../API_CONTRACT_DRAFT.md`; в README той же папки ссылка + «нормативная поверхность → `api-contract/openapi.yaml`» бьёт мимо (файл лежит рядом: + `./openapi.yaml`). Решения D39.100 (`paused`, `eta_seconds`) в YAML ещё не внесены — это работа + S3; схема платформы их уже держит. +8. **Стенд:** Postgres как системного пакета нет и sudo нет, поэтому схема проверена на живом + PostgreSQL 18.4, поднятом БЕЗ root из бинарников zonky в скрэтчпаде (вне репозитория и вне + зависимостей модуля). Тесты с БД гейтятся `TM_PLATFORM_TEST_DSN` и создают свою базу на прогон. + +## Диспозиции бэклога зоны + +| ID | Диспозиция | +|---|---| +| П-1 | **НАЧАТА.** Готово: каркас сессий (схема + мидлварь + CSRF), HTTP-скелет, схема read-model, интерфейс ингеста и ре-синка. Осталось: контрактные ручки, SSE-эндпоинт, материализатор `Sink → Postgres`, воркер. Блокеры: ратификация К-4/К-7 (форма ответов), словарь событий (строка 103) | +| П-2 | Не трогали — гейт «до второго параллельного пользователя» в силе | +| П-3 | Не строили. В схеме заведён гард: частичный уникальный индекс «один живой прогон на книгу» (`runs_one_live_per_book`) — то, что очередь обязана соблюдать, теперь отказывает база. River запинен, но в `go.mod` НЕ добавлен | +| П-4 | Схема `usage_windows` заведена драфтом; источник метрик назван (дельты `committed_usd` из `status --json`). Гейт бюджета ДО старта — вместе с очередью | +| П-5 | Форма предложена выше. Ждёт ответа владельца по авто-продолжению (вопрос 1) | ## Хроника _(записи сессий — сверху новые)_ + +### 04.08.2026 — сессия P0 (платформа №1) + +Прочитано: `CLAUDE.md`, `research/23`, контракт `14-api-contract` (README + openapi.yaml целиком), +`platform/BACKLOG.md`, `frontend/docs/STACK_DECISIONS.md` §5, D39.81/84/85/99/100 по grep. + +Сделано: стек live-сверен (три библиотечных пина §5 — pgx · goose · River — на 04.08 всё ещё +последние; по Go последний патч 1.26.5 от 07.07, floor модуля оставлен общим с движком) → +модуль наполнен → +скелет HTTP + сессии + CSRF → схема read-model тремя миграциями → интерфейс ингеста/супервизии/ +ре-синка → батарея зоны → дизайн-ответы (выше). + +Ревью исполнением: `make check` зелёный; сервер поднят живьём против живого PostgreSQL 18.4 и +опрошен curl'ом (healthz/readyz/401/CSRF-403); миграции применены дважды; констрейнты проверены +поимённо через `pgconn.PgError.ConstraintName`; супервизор проверен настоящим процессом с +контрактными кодами возврата. + +Адверсариальная самопроверка (author≠reviewer) дала четыре правки, каждая внесена: +(а) вложенный mux под `StripPrefix` терял `Request.Pattern`, из-за чего лог писался бы по сырому +пути с id книг — проверено экспериментом, переделано на один mux; (б) отклонённые запросы (401/403) +вообще не логировались, потому что лог висел на маршрутах, а гард стоял снаружи — лог поднят +наружу, добавлен тест «денайл тоже виден»; (в) **дефект, найденный запуском бинарника без БД:** +предъявленный Bearer уходил в nil-хранилище сессий и падал паникой в 500 — теперь отсутствие +хранилища это отказ 401, как и любой другой промах (регрессионный тест на месте); (г) пин тулчейна +поднят до 1.26.5 — в нём security-фиксы `crypto/tls` и `os`, а этот модуль сетевой (в `go.mod` +floor остался 1.26.4, общий с движком). Плюс снят мёртвый код: `crypto/rand.Read` по доке ошибку +не возвращает вовсе (падает), поэтому ветки её обработки убраны, а не оставлены изображать проверку. + +Дерево не коммичено — лендит оркестратор. diff --git a/platform/go.mod b/platform/go.mod index d5b23fea..ce9cfdca 100644 --- a/platform/go.mod +++ b/platform/go.mod @@ -1,3 +1,19 @@ module textmachine/platform go 1.26.4 + +require ( + github.com/jackc/pgx/v5 v5.10.0 + github.com/pressly/goose/v3 v3.27.3 +) + +require ( + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect + github.com/mfridman/interpolate v0.0.2 // indirect + github.com/sethvargo/go-retry v0.4.0 // indirect + go.uber.org/multierr v1.11.0 // indirect + golang.org/x/sync v0.22.0 // indirect + golang.org/x/text v0.40.0 // indirect +) diff --git a/platform/go.sum b/platform/go.sum new file mode 100644 index 00000000..62dfbd71 --- /dev/null +++ b/platform/go.sum @@ -0,0 +1,54 @@ +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= +github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/mattn/go-isatty v0.0.23 h1:cYwCQTQf3HB6xUC+BtyCLZNr7IzbOmoZbmssVNzSyiQ= +github.com/mattn/go-isatty v0.0.23/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A= +github.com/mfridman/interpolate v0.0.2 h1:pnuTK7MQIxxFz1Gr+rjSIx9u7qVjf5VOoM/u6BbAxPY= +github.com/mfridman/interpolate v0.0.2/go.mod h1:p+7uk6oE07mpE/Ik1b8EckO0O4ZXiGAfshKBWLUM9Xg= +github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= +github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/pressly/goose/v3 v3.27.3 h1:pIglVHjw99r4e/hDHHwbl9vfOsDMqUokfkXo6+n/RxA= +github.com/pressly/goose/v3 v3.27.3/go.mod h1:Dag+xpV6o20HR2LFY1j0q6MDwc3f7vPUFDA77R+0yGY= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +github.com/sethvargo/go-retry v0.4.0 h1:9qy1OoIAxBL+gBYnkTnTnWle5wlfsXQlwRzIbbpdqPw= +github.com/sethvargo/go-retry v0.4.0/go.mod h1:tvsjdKG6xfiCx4LSiUZ06kcv38xvdVQwv8R6/VnnVWg= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= +go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs= +golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +modernc.org/libc v1.74.3 h1:a4J+Z8aVaxPyjyxRAdJzw246PqpcFGvVPnfT/AuM5Ws= +modernc.org/libc v1.74.3/go.mod h1:4H7h/MJ8wnjL8RAbp9v3OXgnk22X7MouHIhDbvP3gj4= +modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= +modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= +modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI= +modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw= +modernc.org/sqlite v1.54.0 h1:JCxR4qwkJvOaqAoYcgDoO25Nc+ROg6EJ2LfBVzdrgog= +modernc.org/sqlite v1.54.0/go.mod h1:4ntCLuNmnH8+GNqjka1wNg7KJd5/Hi5FYp8K+XQ7GZw= diff --git a/platform/internal/auth/csrf.go b/platform/internal/auth/csrf.go new file mode 100644 index 00000000..98a78977 --- /dev/null +++ b/platform/internal/auth/csrf.go @@ -0,0 +1,61 @@ +package auth + +import ( + "fmt" + "net/http" +) + +// ClientHeader is the header a browser client must send with every unsafe cookie-authenticated +// request. Any value; its PRESENCE is the assertion. +const ClientHeader = "X-TM-Client" + +// CSRF guards the cookie path and only it: a Bearer token is not ambient authority, so no +// cross-site page can make a browser attach one. +// +// Two layers, both cheap: +// +// - http.CrossOriginProtection (stdlib, Go 1.25+) — Sec-Fetch-Site with an Origin/Host fallback. +// This is the mechanism STACK_DECISIONS §5 describes, and it now ships with the toolchain, so +// we do not hand-roll it. +// - a required custom header on unsafe requests that carry the session cookie. The stdlib check +// ALLOWS a request bearing neither Sec-Fetch-Site nor Origin, on the reasoning that it is not +// a browser. A pre-2023 browser posting a cross-site
is exactly that case, and +// POST /books takes multipart/form-data — a form-reachable content type that triggers no +// preflight. A plain form cannot set a custom header; a fetch() from our own origin can. +// +// trustedOrigins are additional origins allowed to make unsafe requests (a separately deployed +// frontend). Empty means same-origin only. deny writes the 403 body — injected for the same reason +// as Authenticator.Deny: the error shape belongs to the API layer. +func CSRF(trustedOrigins []string, deny http.Handler) (func(http.Handler) http.Handler, error) { + p := http.NewCrossOriginProtection() + p.SetDenyHandler(deny) + for _, o := range trustedOrigins { + if err := p.AddTrustedOrigin(o); err != nil { + return nil, fmt.Errorf("auth: trusted origin %q: %w", o, err) + } + } + return func(next http.Handler) http.Handler { + return p.Handler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if cookieUnsafe(r) && r.Header.Get(ClientHeader) == "" { + deny.ServeHTTP(w, r) + return + } + next.ServeHTTP(w, r) + })) + }, nil +} + +// cookieUnsafe reports a state-changing request presented by cookie. It reads the cookie directly +// rather than the principal: this check runs BEFORE authentication, so that a forged request is +// refused without touching the session table. +func cookieUnsafe(r *http.Request) bool { + switch r.Method { + case http.MethodGet, http.MethodHead, http.MethodOptions: + return false + } + if r.Header.Get("Authorization") != "" { + return false + } + c, err := r.Cookie(CookieName) + return err == nil && c.Value != "" +} diff --git a/platform/internal/auth/csrf_test.go b/platform/internal/auth/csrf_test.go new file mode 100644 index 00000000..5ef1deec --- /dev/null +++ b/platform/internal/auth/csrf_test.go @@ -0,0 +1,71 @@ +package auth + +import ( + "net/http" + "net/http/httptest" + "testing" +) + +func csrfChain(t *testing.T, trusted ...string) (http.Handler, *bool) { + t.Helper() + passed := false + mw, err := CSRF(trusted, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusForbidden) + })) + if err != nil { + t.Fatal(err) + } + return mw(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { passed = true })), &passed +} + +func TestCSRF(t *testing.T) { + cases := []struct { + name string + method string + headers map[string]string + cookie bool + trusted []string + want int + }{ + {name: "same-origin write with the client header", method: http.MethodPost, cookie: true, + headers: map[string]string{"Sec-Fetch-Site": "same-origin", ClientHeader: "web"}, want: http.StatusOK}, + {name: "cross-site write rejected by the stdlib guard", method: http.MethodPost, cookie: true, + headers: map[string]string{"Sec-Fetch-Site": "cross-site", ClientHeader: "web"}, want: http.StatusForbidden}, + // The hole the stdlib guard leaves open on purpose: no Sec-Fetch-Site, no Origin. A + // pre-2023 browser posting a cross-site multipart form looks exactly like this. + {name: "headerless cookie write rejected by the client header", method: http.MethodPost, cookie: true, + want: http.StatusForbidden}, + {name: "bearer write needs no client header", method: http.MethodPost, + headers: map[string]string{"Authorization": "Bearer t"}, want: http.StatusOK}, + {name: "reads are never blocked", method: http.MethodGet, cookie: true, want: http.StatusOK}, + {name: "trusted origin allowed", method: http.MethodPost, cookie: true, trusted: []string{"https://app.example.org"}, + headers: map[string]string{"Sec-Fetch-Site": "cross-site", "Origin": "https://app.example.org", ClientHeader: "web"}, + want: http.StatusOK}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + h, passed := csrfChain(t, tc.trusted...) + r := httptest.NewRequest(tc.method, "/v0/books", nil) + for k, v := range tc.headers { + r.Header.Set(k, v) + } + if tc.cookie { + r.AddCookie(&http.Cookie{Name: CookieName, Value: "t"}) + } + w := httptest.NewRecorder() + h.ServeHTTP(w, r) + if w.Code != tc.want { + t.Fatalf("status = %d, want %d", w.Code, tc.want) + } + if got := *passed; got != (tc.want == http.StatusOK) { + t.Fatalf("handler reached = %v", got) + } + }) + } +} + +func TestCSRFRejectsAMalformedTrustedOrigin(t *testing.T) { + if _, err := CSRF([]string{"app.example.org"}, http.NotFoundHandler()); err == nil { + t.Fatal("an origin without a scheme must be refused at boot, not at request time") + } +} diff --git a/platform/internal/auth/middleware.go b/platform/internal/auth/middleware.go new file mode 100644 index 00000000..4727b644 --- /dev/null +++ b/platform/internal/auth/middleware.go @@ -0,0 +1,74 @@ +package auth + +import ( + "net/http" + "strings" + "time" +) + +// Authenticator turns a presented token into a principal. It is the ONLY place a principal is +// created, which is what keeps the API portable to the desktop client: no endpoint can grow a +// dependency on cookies if no endpoint ever sees one. +type Authenticator struct { + Sessions SessionStore + IdleTTL time.Duration + // Now is injectable so expiry is testable without sleeping. + Now func() time.Time + // Deny writes the 401 body. Injected because the error shape belongs to the API layer + // (problem+json), and auth must not depend on it. + Deny http.Handler +} + +// Require rejects anything that does not carry a live session. +func (a *Authenticator) Require(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + token, via, ok := present(r) + // No store means no session can be proven, which is a denial and not a crash: the service + // is allowed to run without a database (readiness says so), and a caller who presents a + // token must get the same 401 as a caller who presents none. + if !ok || a.Sessions == nil { + a.Deny.ServeHTTP(w, r) + return + } + now := a.now() + digest := Digest(token) + s, err := a.Sessions.Lookup(r.Context(), digest, now) + if err != nil { + // A store failure denies exactly like an unknown token: an authenticated caller is + // what a store failure cannot prove, and a distinguishable answer is an oracle. + a.Deny.ServeHTTP(w, r) + return + } + // Slide the idle window only in its second half. Sliding on every request would turn every + // read into a write, and the session table is on the hot path of every call. + if a.IdleTTL > 0 && s.IdleExpiresAt.Sub(now) < a.IdleTTL/2 { + _ = a.Sessions.Touch(r.Context(), digest, now, a.IdleTTL) + } + ctx := withPrincipal(r.Context(), Principal{UserID: s.UserID, Via: via}) + next.ServeHTTP(w, r.WithContext(ctx)) + }) +} + +func (a *Authenticator) now() time.Time { + if a.Now != nil { + return a.Now() + } + return time.Now() +} + +// present extracts the token. Bearer wins over the cookie when both arrive: an explicit credential +// beats an ambient one, and it keeps a stray cookie from deciding the CSRF path for an API client. +func present(r *http.Request) (token string, via Presentation, ok bool) { + if h := r.Header.Get("Authorization"); h != "" { + scheme, value, found := strings.Cut(h, " ") + if !found || !strings.EqualFold(scheme, "Bearer") || value == "" { + return "", "", false + } + return value, ViaBearer, true + } + c, err := r.Cookie(CookieName) + if err != nil || c.Value == "" { + return "", "", false + } + return c.Value, ViaCookie, true +} diff --git a/platform/internal/auth/middleware_test.go b/platform/internal/auth/middleware_test.go new file mode 100644 index 00000000..18c62e21 --- /dev/null +++ b/platform/internal/auth/middleware_test.go @@ -0,0 +1,183 @@ +package auth + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +type fakeStore struct { + session Session + err error + lookups int + digest []byte + touched int + touchTTL time.Duration +} + +func (f *fakeStore) Lookup(_ context.Context, digest []byte, _ time.Time) (Session, error) { + f.lookups++ + f.digest = digest + return f.session, f.err +} + +func (f *fakeStore) Touch(_ context.Context, _ []byte, _ time.Time, ttl time.Duration) error { + f.touched++ + f.touchTTL = ttl + return nil +} + +func newAuth(store SessionStore, now time.Time) (*Authenticator, *int) { + denied := 0 + return &Authenticator{ + Sessions: store, + IdleTTL: time.Hour, + Now: func() time.Time { return now }, + Deny: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + denied++ + w.WriteHeader(http.StatusUnauthorized) + }), + }, &denied +} + +func TestBothPresentationsYieldAPrincipal(t *testing.T) { + now := time.Now() + token := NewToken() + for name, arm := range map[string]struct { + set func(*http.Request) + via Presentation + }{ + "cookie": {func(r *http.Request) { r.AddCookie(&http.Cookie{Name: CookieName, Value: token}) }, ViaCookie}, + "bearer": {func(r *http.Request) { r.Header.Set("Authorization", "Bearer "+token) }, ViaBearer}, + } { + t.Run(name, func(t *testing.T) { + store := &fakeStore{session: Session{ + UserID: "u1", + IdleExpiresAt: now.Add(time.Hour), + AbsoluteExpiresAt: now.Add(24 * time.Hour), + }} + a, denied := newAuth(store, now) + var seen Principal + h := a.Require(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + seen, _ = FromContext(r.Context()) + })) + r := httptest.NewRequest(http.MethodGet, "/v0/books", nil) + arm.set(r) + h.ServeHTTP(httptest.NewRecorder(), r) + + if *denied != 0 { + t.Fatalf("denied a live session") + } + if seen.UserID != "u1" || seen.Via != arm.via { + t.Fatalf("principal = %+v", seen) + } + // What reaches the store is the digest, never the token itself. + if string(store.digest) == token { + t.Fatal("plaintext token reached the store") + } + }) + } +} + +func TestNoOrBrokenCredentialIsDenied(t *testing.T) { + now := time.Now() + for name, set := range map[string]func(*http.Request){ + "nothing": func(*http.Request) {}, + "empty cookie": func(r *http.Request) { r.AddCookie(&http.Cookie{Name: CookieName, Value: ""}) }, + "wrong scheme": func(r *http.Request) { r.Header.Set("Authorization", "Basic abc") }, + "bearer empty": func(r *http.Request) { r.Header.Set("Authorization", "Bearer ") }, + "other cookie": func(r *http.Request) { r.AddCookie(&http.Cookie{Name: "tm_session", Value: "x"}) }, + "store failure": func(r *http.Request) { r.Header.Set("Authorization", "Bearer t") }, + } { + t.Run(name, func(t *testing.T) { + store := &fakeStore{err: ErrNoSession} + a, denied := newAuth(store, now) + reached := false + h := a.Require(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { reached = true })) + r := httptest.NewRequest(http.MethodGet, "/v0/books", nil) + set(r) + h.ServeHTTP(httptest.NewRecorder(), r) + if reached { + t.Fatal("handler ran without a live session") + } + if *denied != 1 { + t.Fatalf("deny count = %d", *denied) + } + }) + } +} + +// Found by running the binary without a database: a presented token used to reach a nil store and +// panic into a 500. Nothing can be proven without a store, so it must deny like any other miss. +func TestNoStoreDeniesInsteadOfPanicking(t *testing.T) { + a, denied := newAuth(nil, time.Now()) + h := a.Require(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + t.Fatal("handler ran with no session store") + })) + r := httptest.NewRequest(http.MethodGet, "/v0/books", nil) + r.Header.Set("Authorization", "Bearer t") + h.ServeHTTP(httptest.NewRecorder(), r) + if *denied != 1 { + t.Fatalf("deny count = %d", *denied) + } +} + +func TestBearerWinsOverCookie(t *testing.T) { + r := httptest.NewRequest(http.MethodGet, "/v0/books", nil) + r.AddCookie(&http.Cookie{Name: CookieName, Value: "cookie-token"}) + r.Header.Set("Authorization", "Bearer bearer-token") + token, via, ok := present(r) + if !ok || token != "bearer-token" || via != ViaBearer { + t.Fatalf("present() = %q %q %v", token, via, ok) + } +} + +func TestIdleWindowSlidesOnlyInItsSecondHalf(t *testing.T) { + now := time.Now() + for name, tc := range map[string]struct { + remaining time.Duration + want int + }{ + "fresh": {50 * time.Minute, 0}, + "stale": {10 * time.Minute, 1}, + } { + t.Run(name, func(t *testing.T) { + store := &fakeStore{session: Session{ + UserID: "u1", + IdleExpiresAt: now.Add(tc.remaining), + AbsoluteExpiresAt: now.Add(24 * time.Hour), + }} + a, _ := newAuth(store, now) + h := a.Require(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + r := httptest.NewRequest(http.MethodGet, "/v0/books", nil) + r.Header.Set("Authorization", "Bearer t") + h.ServeHTTP(httptest.NewRecorder(), r) + if store.touched != tc.want { + t.Fatalf("touches = %d, want %d", store.touched, tc.want) + } + }) + } +} + +func TestTokensAreUniqueAndDigestIsStable(t *testing.T) { + seen := make(map[string]bool, 64) + for range 64 { + tok := NewToken() + if len(tok) < 40 { // 32 bytes base64url ≈ 43 chars + t.Fatalf("token too short: %q", tok) + } + if seen[tok] { + t.Fatal("token repeated") + } + seen[tok] = true + d := Digest(tok) + if len(d) != 32 { + t.Fatalf("digest is %d bytes", len(d)) + } + if string(d) == tok { + t.Fatal("the stored form is the token itself") + } + } +} diff --git a/platform/internal/auth/principal.go b/platform/internal/auth/principal.go new file mode 100644 index 00000000..00b88cc0 --- /dev/null +++ b/platform/internal/auth/principal.go @@ -0,0 +1,32 @@ +package auth + +import "context" + +// Presentation is HOW the session was presented. It exists for the CSRF layer and for logs, never +// for authorization: a session is a session whichever way it arrived. +type Presentation string + +const ( + ViaCookie Presentation = "cookie" + ViaBearer Presentation = "bearer" +) + +// Principal is the authenticated caller. +type Principal struct { + UserID string + Via Presentation +} + +type principalKey struct{} + +// withPrincipal is deliberately unexported: the principal is created in the middleware of this +// package and nowhere else (D39.84). A handler that could mint one could also invent a user. +func withPrincipal(ctx context.Context, p Principal) context.Context { + return context.WithValue(ctx, principalKey{}, p) +} + +// FromContext returns the principal established by Authenticator. +func FromContext(ctx context.Context) (Principal, bool) { + p, ok := ctx.Value(principalKey{}).(Principal) + return p, ok +} diff --git a/platform/internal/auth/session.go b/platform/internal/auth/session.go new file mode 100644 index 00000000..ece4937a --- /dev/null +++ b/platform/internal/auth/session.go @@ -0,0 +1,60 @@ +// Package auth is the session layer: ONE server-side session (D39.84), presented either as a +// __Host cookie by the browser or as a Bearer token by the desktop and CLI clients. No handler may +// look at either — the principal is created in middleware and read from the request context. +package auth + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "errors" + "time" +) + +// CookieName is the browser presentation. The __Host prefix is not decoration: it forbids a Domain +// attribute and requires Secure + Path=/, which is what stops a sibling subdomain from writing a +// session cookie for the app. +const CookieName = "__Host-tm_session" + +// tokenBytes is 256 bits of entropy (STACK_DECISIONS §5). Opaque: it encodes nothing, so a stolen +// token cannot be read, and a JWT's "verify without the database" property is exactly what we do +// not want — revocation must be immediate. +const tokenBytes = 32 + +// ErrNoSession is returned when nothing live matches the presented token. Callers must not +// distinguish "unknown token" from "expired token" on the wire: the difference is an oracle. +var ErrNoSession = errors.New("auth: no live session") + +// Session is what the store knows about a live session. +type Session struct { + UserID string + IdleExpiresAt time.Time + AbsoluteExpiresAt time.Time +} + +// SessionStore is the persistence the middleware needs. Lookup MUST apply expiry and revocation +// itself, so a store that forgets a clause cannot be papered over here. +type SessionStore interface { + Lookup(ctx context.Context, digest []byte, now time.Time) (Session, error) + Touch(ctx context.Context, digest []byte, now time.Time, idleTTL time.Duration) error +} + +// NewToken mints a credential. The plaintext exists only in this return value and in the client: +// what reaches the database is Digest(token). +// +// No error return: crypto/rand.Read "never returns an error, and always fills b entirely" — it +// crashes the program instead. An error path here would be dead code pretending to be a check. +func NewToken() string { + b := make([]byte, tokenBytes) + rand.Read(b) + return base64.RawURLEncoding.EncodeToString(b) +} + +// Digest is the stored form. Plain SHA-256 with no salt or stretching is correct HERE and only +// here: the input is 256 uniform random bits, so there is no guessable space to slow an attacker +// down in — unlike a password. +func Digest(token string) []byte { + sum := sha256.Sum256([]byte(token)) + return sum[:] +} diff --git a/platform/internal/config/config.go b/platform/internal/config/config.go new file mode 100644 index 00000000..d65f804e --- /dev/null +++ b/platform/internal/config/config.go @@ -0,0 +1,81 @@ +// Package config reads the service's settings from the environment. +package config + +import ( + "fmt" + "os" + "strings" + "time" +) + +// Config is the whole configuration surface. Environment only: a control plane is deployed, not +// hand-run, and a settings file is one more thing to keep in sync with the deployment. +type Config struct { + Addr string + // DSN is the Postgres connection string. Empty is allowed and means "start unready": the + // service answers liveness and reports on /readyz why it cannot serve. + // + // ⚠ It carries a password. It is read from the environment, never logged, and never echoed + // into an error message. + DSN string + // TrustedOrigins are origins besides our own that may make unsafe requests. + TrustedOrigins []string + // SessionIdleTTL is how long a session survives without use; SessionMaxAge is the ceiling no + // amount of use can extend. + SessionIdleTTL time.Duration + SessionMaxAge time.Duration + // Migrate applies pending migrations at boot. Off by default: a rollout should migrate once, + // deliberately, not once per replica. + Migrate bool +} + +// Load reads the environment. +func Load() (Config, error) { + c := Config{ + Addr: env("TM_PLATFORM_ADDR", "127.0.0.1:8080"), + DSN: os.Getenv("TM_PLATFORM_DSN"), + SessionIdleTTL: 14 * 24 * time.Hour, + SessionMaxAge: 90 * 24 * time.Hour, + Migrate: os.Getenv("TM_PLATFORM_MIGRATE") == "1", + } + if raw := os.Getenv("TM_PLATFORM_TRUSTED_ORIGINS"); raw != "" { + for _, o := range strings.Split(raw, ",") { + if o = strings.TrimSpace(o); o != "" { + c.TrustedOrigins = append(c.TrustedOrigins, o) + } + } + } + var err error + if c.SessionIdleTTL, err = duration("TM_PLATFORM_SESSION_IDLE", c.SessionIdleTTL); err != nil { + return Config{}, err + } + if c.SessionMaxAge, err = duration("TM_PLATFORM_SESSION_MAX_AGE", c.SessionMaxAge); err != nil { + return Config{}, err + } + if c.SessionIdleTTL > c.SessionMaxAge { + return Config{}, fmt.Errorf("config: session idle TTL %s exceeds max age %s", c.SessionIdleTTL, c.SessionMaxAge) + } + return c, nil +} + +func env(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} + +func duration(key string, def time.Duration) (time.Duration, error) { + raw := os.Getenv(key) + if raw == "" { + return def, nil + } + d, err := time.ParseDuration(raw) + if err != nil { + return 0, fmt.Errorf("config: %s: %w", key, err) + } + if d <= 0 { + return 0, fmt.Errorf("config: %s must be positive", key) + } + return d, nil +} diff --git a/platform/internal/config/config_test.go b/platform/internal/config/config_test.go new file mode 100644 index 00000000..50334677 --- /dev/null +++ b/platform/internal/config/config_test.go @@ -0,0 +1,48 @@ +package config + +import ( + "testing" + "time" +) + +func TestDefaultsAndOverrides(t *testing.T) { + c, err := Load() + if err != nil { + t.Fatal(err) + } + if c.Addr != "127.0.0.1:8080" { + t.Fatalf("default addr = %q; a control plane must not bind the world by default", c.Addr) + } + if c.Migrate { + t.Fatal("migrations must be opt-in") + } + + t.Setenv("TM_PLATFORM_ADDR", ":9000") + t.Setenv("TM_PLATFORM_SESSION_IDLE", "30m") + t.Setenv("TM_PLATFORM_TRUSTED_ORIGINS", "https://app.example.org, https://desktop.example.org ") + c, err = Load() + if err != nil { + t.Fatal(err) + } + if c.Addr != ":9000" || c.SessionIdleTTL != 30*time.Minute { + t.Fatalf("config = %+v", c) + } + if len(c.TrustedOrigins) != 2 || c.TrustedOrigins[1] != "https://desktop.example.org" { + t.Fatalf("origins = %q", c.TrustedOrigins) + } +} + +func TestImpossibleSessionWindowIsRefusedAtBoot(t *testing.T) { + t.Setenv("TM_PLATFORM_SESSION_IDLE", "100h") + t.Setenv("TM_PLATFORM_SESSION_MAX_AGE", "1h") + if _, err := Load(); err == nil { + t.Fatal("an idle window longer than the absolute one must fail at boot, not at 3am") + } +} + +func TestMalformedDurationIsRefused(t *testing.T) { + t.Setenv("TM_PLATFORM_SESSION_IDLE", "fortnight") + if _, err := Load(); err == nil { + t.Fatal("want a parse error") + } +} diff --git a/platform/internal/httpapi/middleware.go b/platform/internal/httpapi/middleware.go new file mode 100644 index 00000000..caab662a --- /dev/null +++ b/platform/internal/httpapi/middleware.go @@ -0,0 +1,106 @@ +package httpapi + +import ( + "context" + "crypto/rand" + "encoding/base32" + "log/slog" + "net/http" + "time" +) + +type requestIDKey struct{} + +// RequestID stamps every request. The id is ours, never the client's: an id echoed from a header +// lets a caller poison our logs and correlate other users' lines. +func RequestID(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var b [10]byte + // crypto/rand.Read never returns an error; it crashes the program instead. + rand.Read(b[:]) + id := base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(b[:]) + w.Header().Set("X-Request-Id", id) + next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), requestIDKey{}, id))) + }) +} + +// RequestIDOf returns the id stamped by RequestID, or "". +func RequestIDOf(ctx context.Context) string { + id, _ := ctx.Value(requestIDKey{}).(string) + return id +} + +// SecurityHeaders applies the two product invariants to every response. +// +// PT-34: not one byte of a user's translation may reach an indexable URL — noindex is set here, +// once, rather than per handler, because a handler added later would not know the rule. +// Cache-Control: no-store is blanket for the same reason: the contract mandates it for responses +// carrying translated text, and a private API has nothing worth caching in a shared cache. +func SecurityHeaders(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + h := w.Header() + h.Set("X-Robots-Tag", "noindex, nofollow") + h.Set("Cache-Control", "no-store") + h.Set("X-Content-Type-Options", "nosniff") + h.Set("Referrer-Policy", "no-referrer") + next.ServeHTTP(w, r) + }) +} + +// Recover turns a panic into a 500 instead of a dropped connection. +func Recover(log *slog.Logger) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer func() { + if v := recover(); v != nil { + log.ErrorContext(r.Context(), "panic in handler", + "panic", v, "path", r.URL.Path, "request_id", RequestIDOf(r.Context())) + WriteProblem(w, http.StatusInternalServerError, "Internal error", "") + } + }() + next.ServeHTTP(w, r) + }) + } +} + +// AccessLog writes one INFO line per request. Deliberately absent: money (D39.84 and the norm of +// the P0 prompt — costs do not reach INFO), request bodies and any user text. +func AccessLog(log *slog.Logger) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + start := time.Now() + rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK} + next.ServeHTTP(rec, r) + log.InfoContext(r.Context(), "request", + "method", r.Method, + // Pattern, not the raw path: the raw one carries book and run ids, which multiply + // log cardinality and identify a user's library in an operator's index. ServeMux + // fills Pattern in place, so it is readable here even though the mux ran inside. + "route", routeOf(r), + "status", rec.status, + "ms", time.Since(start).Milliseconds(), + "request_id", RequestIDOf(r.Context())) + }) + } +} + +func routeOf(r *http.Request) string { + if p := r.Pattern; p != "" { + return p + } + return "(unmatched)" +} + +// statusRecorder captures the status code. Unwrap keeps http.ResponseController working through +// the wrapper — that is how a later SSE handler will reach Flush. +type statusRecorder struct { + http.ResponseWriter + status int +} + +func (s *statusRecorder) WriteHeader(code int) { + s.status = code + s.ResponseWriter.WriteHeader(code) +} + +func (s *statusRecorder) Unwrap() http.ResponseWriter { return s.ResponseWriter } diff --git a/platform/internal/httpapi/problem.go b/platform/internal/httpapi/problem.go new file mode 100644 index 00000000..e779dd47 --- /dev/null +++ b/platform/internal/httpapi/problem.go @@ -0,0 +1,35 @@ +package httpapi + +import ( + "encoding/json" + "log/slog" + "net/http" +) + +// Problem is an RFC 9457 error body. +// +// Detail NEVER carries engine text. The engine's own detail strings read like "CJK leak in the ru +// output: 第一节" — pipeline vocabulary that must not cross this seam (contract §2.12). What goes +// here is a product phrase or nothing. +type Problem struct { + Type string `json:"type"` + Title string `json:"title"` + Status int `json:"status"` + Detail string `json:"detail,omitempty"` +} + +// WriteProblem renders an error response. +func WriteProblem(w http.ResponseWriter, status int, title, detail string) { + w.Header().Set("Content-Type", "application/problem+json") + w.WriteHeader(status) + if err := json.NewEncoder(w).Encode(Problem{Type: "about:blank", Title: title, Status: status, Detail: detail}); err != nil { + slog.Debug("problem body not delivered", "err", err) // the client went away mid-write + } +} + +// ProblemHandler is a static problem response, for the middleware that must be handed a denier. +func ProblemHandler(status int, title string) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + WriteProblem(w, status, title, "") + }) +} diff --git a/platform/internal/httpapi/server.go b/platform/internal/httpapi/server.go new file mode 100644 index 00000000..9f206470 --- /dev/null +++ b/platform/internal/httpapi/server.go @@ -0,0 +1,88 @@ +package httpapi + +import ( + "context" + "errors" + "log/slog" + "net/http" + + "textmachine/platform/internal/auth" +) + +// Prober is what readiness needs from the database. +type Prober interface { + Ping(ctx context.Context) error +} + +// Deps is everything the HTTP surface is built from. +type Deps struct { + Log *slog.Logger + // DB is nil when no database is configured: the service still starts and still answers + // liveness, and readiness reports why it is not ready. + DB Prober + Auth *auth.Authenticator + // TrustedOrigins are origins besides our own allowed to make unsafe requests. + TrustedOrigins []string + // APIPrefix is the contract's base path ("/v0"). Ops endpoints live outside it: a health check + // is not part of the versioned surface and must not move when the surface does. + APIPrefix string +} + +// New builds the handler. +// +// Contract routes are registered on the SAME mux as the ops ones, with the version prefix written +// into the pattern, and each is wrapped in guard(). One mux is what keeps Request.Pattern +// meaningful all the way out to the access log — a nested mux behind http.StripPrefix hands the +// inner handler a copy, and the pattern the copy learns never comes back. +func New(d Deps) (http.Handler, error) { + if d.APIPrefix == "" { + d.APIPrefix = "/v0" + } + if d.Auth == nil { + return nil, errors.New("httpapi: no authenticator: the API subtree may not be served unguarded") + } + csrf, err := auth.CSRF(d.TrustedOrigins, ProblemHandler(http.StatusForbidden, "Cross-origin request rejected")) + if err != nil { + return nil, err + } + // Every API route goes through this. An anonymous caller therefore gets 401 before 404, which + // is deliberate: the shape of the surface is not public information. + guard := func(h http.Handler) http.Handler { return csrf(d.Auth.Require(h)) } + + mux := http.NewServeMux() + mux.Handle("GET /healthz", http.HandlerFunc(healthz)) + mux.Handle("GET /readyz", readyz(d.DB)) + // The contract's routes land here (P-1), as mux.Handle("GET "+d.APIPrefix+"/books", guard(…)). + // Until then everything under the prefix is a guarded 404 in the shape the contract mandates. + mux.Handle(d.APIPrefix+"/", guard(ProblemHandler(http.StatusNotFound, "Object not found"))) + + // Recover sits INSIDE AccessLog: a panic converted to a 500 still produces a log line, whereas + // a panic unwinding past the logger produces none. + return RequestID(SecurityHeaders(AccessLog(d.Log)(Recover(d.Log)(mux)))), nil +} + +// healthz is liveness: the process is up and serving. It touches nothing, so a database outage +// cannot make a supervisor kill a healthy process. +func healthz(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok\n")) +} + +// readyz is readiness: this instance can serve traffic, which means the database answers. +func readyz(db Prober) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if db == nil { + WriteProblem(w, http.StatusServiceUnavailable, "Not ready", "no database configured") + return + } + if err := db.Ping(r.Context()); err != nil { + // The reason stays in the log; the body says only that we are not ready. + WriteProblem(w, http.StatusServiceUnavailable, "Not ready", "database unreachable") + return + } + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ready\n")) + }) +} diff --git a/platform/internal/httpapi/server_test.go b/platform/internal/httpapi/server_test.go new file mode 100644 index 00000000..4da20a1e --- /dev/null +++ b/platform/internal/httpapi/server_test.go @@ -0,0 +1,165 @@ +package httpapi + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "textmachine/platform/internal/auth" +) + +type prober struct{ err error } + +func (p prober) Ping(context.Context) error { return p.err } + +type liveSessions struct{} + +func (liveSessions) Lookup(context.Context, []byte, time.Time) (auth.Session, error) { + return auth.Session{UserID: "u1", IdleExpiresAt: time.Now().Add(time.Hour), AbsoluteExpiresAt: time.Now().Add(time.Hour)}, nil +} +func (liveSessions) Touch(context.Context, []byte, time.Time, time.Duration) error { return nil } + +type deadSessions struct{} + +func (deadSessions) Lookup(context.Context, []byte, time.Time) (auth.Session, error) { + return auth.Session{}, auth.ErrNoSession +} +func (deadSessions) Touch(context.Context, []byte, time.Time, time.Duration) error { return nil } + +func newServer(t *testing.T, db Prober, sessions auth.SessionStore) (http.Handler, *bytes.Buffer) { + t.Helper() + var logs bytes.Buffer + h, err := New(Deps{ + Log: slog.New(slog.NewJSONHandler(&logs, nil)), + DB: db, + Auth: &auth.Authenticator{ + Sessions: sessions, + IdleTTL: time.Hour, + Deny: ProblemHandler(http.StatusUnauthorized, "Session missing or invalid"), + }, + }) + if err != nil { + t.Fatal(err) + } + return h, &logs +} + +func TestHealthzIsIndependentOfTheDatabase(t *testing.T) { + h, _ := newServer(t, prober{err: errors.New("down")}, deadSessions{}) + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/healthz", nil)) + if w.Code != http.StatusOK { + t.Fatalf("status = %d", w.Code) + } + // PT-34 and the no-store rule are applied to every response, not per handler. + if got := w.Header().Get("X-Robots-Tag"); !strings.Contains(got, "noindex") { + t.Fatalf("X-Robots-Tag = %q", got) + } + if got := w.Header().Get("Cache-Control"); got != "no-store" { + t.Fatalf("Cache-Control = %q", got) + } + if w.Header().Get("X-Request-Id") == "" { + t.Fatal("no request id") + } +} + +func TestReadyzFollowsTheDatabase(t *testing.T) { + for name, tc := range map[string]struct { + db Prober + want int + }{ + "healthy": {prober{}, http.StatusOK}, + "unreachable": {prober{err: errors.New("down")}, http.StatusServiceUnavailable}, + "not configured": {nil, http.StatusServiceUnavailable}, + } { + t.Run(name, func(t *testing.T) { + h, _ := newServer(t, tc.db, deadSessions{}) + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/readyz", nil)) + if w.Code != tc.want { + t.Fatalf("status = %d, want %d", w.Code, tc.want) + } + }) + } +} + +func TestAPISubtreeIsGuardedBeforeItIsRouted(t *testing.T) { + h, logs := newServer(t, prober{}, deadSessions{}) + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/v0/books", nil)) + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401: an anonymous caller must not be able to map the surface", w.Code) + } + assertProblem(t, w, http.StatusUnauthorized) + // A rejected request must still be logged: an invisible 401 is an invisible brute force. + if !strings.Contains(logs.String(), `"status":401`) { + t.Fatalf("denial not logged: %s", logs.String()) + } +} + +func TestAuthenticatedUnknownRouteIs404Problem(t *testing.T) { + h, _ := newServer(t, prober{}, liveSessions{}) + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/v0/nothing-here", nil) + r.Header.Set("Authorization", "Bearer t") + h.ServeHTTP(w, r) + if w.Code != http.StatusNotFound { + t.Fatalf("status = %d", w.Code) + } + assertProblem(t, w, http.StatusNotFound) +} + +func TestAccessLogNamesTheRouteNotThePath(t *testing.T) { + h, logs := newServer(t, prober{}, deadSessions{}) + h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/healthz", nil)) + var line map[string]any + if err := json.Unmarshal(bytes.TrimSpace(logs.Bytes()), &line); err != nil { + t.Fatalf("log line: %v (%q)", err, logs.String()) + } + if line["route"] != "GET /healthz" { + t.Fatalf("route = %v; the pattern is what keeps ids out of the log", line["route"]) + } + if line["status"] != float64(http.StatusOK) { + t.Fatalf("status = %v", line["status"]) + } +} + +func TestPanicBecomesAProblem(t *testing.T) { + var logs bytes.Buffer + inner := http.HandlerFunc(func(http.ResponseWriter, *http.Request) { panic("boom") }) + h := Recover(slog.New(slog.NewJSONHandler(&logs, nil)))(inner) + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/v0/books", nil)) + assertProblem(t, w, http.StatusInternalServerError) +} + +func TestServerRefusesToBuildWithoutAnAuthenticator(t *testing.T) { + if _, err := New(Deps{Log: slog.Default()}); err == nil { + t.Fatal("the API subtree may not be mounted unguarded") + } +} + +func assertProblem(t *testing.T, w *httptest.ResponseRecorder, status int) { + t.Helper() + if got := w.Header().Get("Content-Type"); got != "application/problem+json" { + t.Fatalf("content-type = %q", got) + } + var p Problem + if err := json.Unmarshal(w.Body.Bytes(), &p); err != nil { + t.Fatalf("body: %v (%q)", err, w.Body.String()) + } + if p.Status != status || p.Title == "" { + t.Fatalf("problem = %+v", p) + } + // Engine vocabulary must never reach a client (contract §2.12). + if strings.Contains(p.Detail, "sqlite") || strings.Contains(p.Detail, "pgx") { + t.Fatalf("internals leaked into detail: %q", p.Detail) + } +} diff --git a/platform/internal/ingest/decoder.go b/platform/internal/ingest/decoder.go new file mode 100644 index 00000000..e496c662 --- /dev/null +++ b/platform/internal/ingest/decoder.go @@ -0,0 +1,103 @@ +package ingest + +import ( + "bufio" + "encoding/json" + "errors" + "fmt" + "io" +) + +var ( + // ErrUnsupportedVersion is a major-version refusal: materializing a stream whose meaning + // changed is worse than not materializing it. + ErrUnsupportedVersion = errors.New("ingest: unsupported stream version") + // ErrNoHandshake is a stream whose first line is not the handshake. Ad-hoc unversioned JSON is + // the documented rot path (docker jsonmessage) and is refused at the door. + ErrNoHandshake = errors.New("ingest: stream does not open with hello") + // ErrStreamGap is a seq that skipped or went backwards: lines were lost. Recovery is not + // guesswork — the caller reconciles from `tmctl status --json`, the ratified resync channel. + ErrStreamGap = errors.New("ingest: sequence gap") +) + +// maxLine caps one event. Events carry counters and ids, never text — the translated text travels +// as artifacts — so a megabyte line means the stream is not what we think it is. +const maxLine = 1 << 20 + +// Decoder reads the NDJSON event stream. +type Decoder struct { + sc *bufio.Scanner + hello Hello + greeted bool + lastSeq int64 +} + +func NewDecoder(r io.Reader) *Decoder { + sc := bufio.NewScanner(r) + sc.Buffer(make([]byte, 0, 64<<10), maxLine) + return &Decoder{sc: sc} +} + +// Hello reads and validates the handshake. It must be called before Next. +func (d *Decoder) Hello() (Hello, error) { + ev, err := d.line() + if err != nil { + if errors.Is(err, io.EOF) { + return Hello{}, ErrNoHandshake + } + return Hello{}, err + } + if ev.Type != TypeHello { + return Hello{}, fmt.Errorf("%w: first line is %q", ErrNoHandshake, ev.Type) + } + var h Hello + if err := json.Unmarshal(ev.Data, &h); err != nil { + return Hello{}, fmt.Errorf("ingest: hello payload: %w", err) + } + if err := checkVersion(h.StreamVersion); err != nil { + return Hello{}, err + } + d.hello, d.greeted = h, true + d.lastSeq = ev.Seq + return h, nil +} + +// Next returns the next event, or io.EOF at the end of the stream. An unknown event type is +// returned as-is: tolerating it is the minor-version rule, and dropping it is the sink's decision. +func (d *Decoder) Next() (Envelope, error) { + if !d.greeted { + return Envelope{}, ErrNoHandshake + } + ev, err := d.line() + if err != nil { + return Envelope{}, err + } + switch { + case ev.Seq <= d.lastSeq: + // A duplicate cannot happen inside one pipe, so it is a defect rather than at-least-once + // redelivery — and it is reported, not silently absorbed. + return Envelope{}, fmt.Errorf("%w: seq %d after %d", ErrStreamGap, ev.Seq, d.lastSeq) + case ev.Seq > d.lastSeq+1: + return Envelope{}, fmt.Errorf("%w: seq %d after %d", ErrStreamGap, ev.Seq, d.lastSeq) + } + d.lastSeq = ev.Seq + return ev, nil +} + +func (d *Decoder) line() (Envelope, error) { + for d.sc.Scan() { + raw := d.sc.Bytes() + if len(raw) == 0 { + continue // a blank line is not an event + } + var ev Envelope + if err := json.Unmarshal(raw, &ev); err != nil { + return Envelope{}, fmt.Errorf("ingest: malformed line: %w", err) + } + return ev, nil + } + if err := d.sc.Err(); err != nil { + return Envelope{}, fmt.Errorf("ingest: read stream: %w", err) + } + return Envelope{}, io.EOF +} diff --git a/platform/internal/ingest/decoder_test.go b/platform/internal/ingest/decoder_test.go new file mode 100644 index 00000000..b0fa6c7f --- /dev/null +++ b/platform/internal/ingest/decoder_test.go @@ -0,0 +1,105 @@ +package ingest + +import ( + "context" + "errors" + "io" + "strings" + "testing" +) + +const helloLine = `{"seq":1,"type":"hello","time":"2026-08-04T10:00:00Z","data":{"stream_version":"1.0","engine_run_id":"tr_1","book_id":"gzr"}}` + +func TestHelloIsRequiredFirst(t *testing.T) { + d := NewDecoder(strings.NewReader(`{"seq":1,"type":"progress","data":{}}` + "\n")) + if _, err := d.Hello(); !errors.Is(err, ErrNoHandshake) { + t.Fatalf("want ErrNoHandshake, got %v", err) + } + if _, err := NewDecoder(strings.NewReader("")).Hello(); !errors.Is(err, ErrNoHandshake) { + t.Fatalf("empty stream: want ErrNoHandshake, got %v", err) + } +} + +func TestMajorVersionRefused(t *testing.T) { + line := strings.Replace(helloLine, `"stream_version":"1.0"`, `"stream_version":"2.0"`, 1) + if _, err := NewDecoder(strings.NewReader(line)).Hello(); !errors.Is(err, ErrUnsupportedVersion) { + t.Fatalf("want ErrUnsupportedVersion, got %v", err) + } +} + +func TestMinorVersionAndUnknownFieldsTolerated(t *testing.T) { + line := strings.Replace(helloLine, `"stream_version":"1.0"`, `"stream_version":"1.7","future":42`, 1) + d := NewDecoder(strings.NewReader(line + "\n" + `{"seq":2,"type":"weather","data":{"sky":"grey"}}`)) + if _, err := d.Hello(); err != nil { + t.Fatalf("minor bump must be accepted: %v", err) + } + ev, err := d.Next() + if err != nil { + t.Fatalf("unknown event type must reach the sink: %v", err) + } + if ev.Type != "weather" { + t.Fatalf("type = %q", ev.Type) + } +} + +func TestSequenceGapAndReplayAreReported(t *testing.T) { + for name, second := range map[string]string{ + "gap": `{"seq":5,"type":"progress","data":{}}`, + "replay": `{"seq":1,"type":"progress","data":{}}`, + "reverse": `{"seq":0,"type":"progress","data":{}}`, + } { + t.Run(name, func(t *testing.T) { + d := NewDecoder(strings.NewReader(helloLine + "\n" + second)) + if _, err := d.Hello(); err != nil { + t.Fatal(err) + } + if _, err := d.Next(); !errors.Is(err, ErrStreamGap) { + t.Fatalf("want ErrStreamGap, got %v", err) + } + }) + } +} + +func TestIngestFeedsSinkInOrder(t *testing.T) { + stream := helloLine + "\n" + + `{"seq":2,"type":"progress","data":{"draft":{"done":1,"total":10},"edit":{"done":0,"total":10}}}` + "\n" + + "\n" + // a blank line is not an event + `{"seq":3,"type":"ceiling","data":{"halted":true}}` + "\n" + s := &recordingSink{} + if err := Ingest(context.Background(), strings.NewReader(stream), s); err != nil { + t.Fatalf("ingest: %v", err) + } + if s.hello.EngineRunID != "tr_1" { + t.Fatalf("hello not bound: %+v", s.hello) + } + if got := len(s.applied); got != 2 { + t.Fatalf("applied %d events, want 2", got) + } + if s.applied[0].Type != TypeProgress || s.applied[1].Type != TypeCeiling { + t.Fatalf("order: %v", s.applied) + } +} + +func TestIngestStopsAtMalformedLine(t *testing.T) { + stream := helloLine + "\n" + "{not json\n" + s := &recordingSink{} + err := Ingest(context.Background(), strings.NewReader(stream), s) + if err == nil || errors.Is(err, io.EOF) { + t.Fatalf("want a decode error, got %v", err) + } + if len(s.applied) != 0 { + t.Fatalf("nothing may be applied from a broken stream, got %d", len(s.applied)) + } +} + +type recordingSink struct { + hello Hello + applied []Envelope +} + +func (s *recordingSink) Begin(_ context.Context, h Hello) error { s.hello = h; return nil } + +func (s *recordingSink) Apply(_ context.Context, ev Envelope) error { + s.applied = append(s.applied, ev) + return nil +} diff --git a/platform/internal/ingest/events.go b/platform/internal/ingest/events.go new file mode 100644 index 00000000..b0c752cf --- /dev/null +++ b/platform/internal/ingest/events.go @@ -0,0 +1,141 @@ +// Package ingest is the platform side of the engine seam (D39.85): it supervises a tmctl process, +// reads its NDJSON event stream and hands each event to a sink that materializes it into the +// reporting database. It never opens the engine's SQLite and never parses human output. +// +// ⚠ The emitter does not exist yet — it is row 103 of the engine backlog. The vocabulary below is +// therefore the platform's PROPOSAL, derived from research/23 §7 (which call sites already carry +// the data) and from the API contract §6 (what a reader must be told). It is written as code +// rather than prose so the engine zone can answer it with a diff. +package ingest + +import ( + "encoding/json" + "fmt" + "strconv" + "strings" + "time" +) + +// StreamVersion is the version this decoder implements. The rule is terraform's, ratified by +// D39.85: a MINOR bump adds fields and event types — unknown ones are ignored; a MAJOR bump is +// refused, because a stream whose meaning changed must not be materialized as if it had not. +const StreamVersion = "1.0" + +// Type is the event name. Unknown values are legal under the minor rule and are dropped by the +// sink, not by the decoder. +type Type string + +const ( + // TypeHello is always the first line: the version handshake. + TypeHello Type = "hello" + // TypeProgress carries the per-phase counters. Per phase because a unit is done only once its + // edit resolved, so one end-to-end counter reads zero for the whole draft wave. + TypeProgress Type = "progress" + // TypeUnitDone is one shipped (or withheld) edit unit. + TypeUnitDone Type = "unit_done" + // TypeBankStop is the book-wide signing stop before the edit wave. + TypeBankStop Type = "bank_stop" + // TypeCeiling is the resumable halt on the spend ceiling. It carries NO figures. + TypeCeiling Type = "ceiling" + // TypeFinished is the last line of a clean stream. + TypeFinished Type = "finished" +) + +// Envelope is one line of the stream. +type Envelope struct { + // Seq is per PROCESS and starts at 1. It is the second half of the ratified idempotency key + // (run_id, seq) — where run_id is Hello.EngineRunID, not the platform's run: a resumed run is + // a new process whose seq restarts, so keying on the platform run would drop its whole stream. + Seq int64 `json:"seq"` + Type Type `json:"type"` + Time time.Time `json:"time"` + Data json.RawMessage `json:"data"` +} + +// Hello is the handshake payload. +type Hello struct { + StreamVersion string `json:"stream_version"` + // EngineRunID is the id the engine mints per invocation (today: the trace id, main.go:67-70). + // Once the engine accepts an external trace context (row 102) the platform can supply it, and + // the idempotency namespace becomes ours by construction. + EngineRunID string `json:"engine_run_id"` + BookID string `json:"book_id"` + // ChunkerVersion lets the platform notice that the chapter manifest it persisted was produced + // by a different chunker — the case that silently re-numbers chapters (row 100). + ChunkerVersion string `json:"chunker_version"` +} + +// Counter is a phase counter, in units. +type Counter struct { + Done int `json:"done"` + Total int `json:"total"` +} + +// Progress is the run's counters. ETASeconds mirrors what status already computes +// (pipeline/status.go:112) — the owner asked for a visible ETA (K-5). +type Progress struct { + Draft Counter `json:"draft"` + Edit Counter `json:"edit"` + ETASeconds int `json:"eta_seconds,omitempty"` +} + +// UnitDone is one resolved edit unit. Chapter and Unit are the engine's own numbering; mapping +// them onto the platform's opaque ids is the materializer's job. +// +// Shipped and Flagged are BOTH needed and neither implies the other: a flagged unit legally ships +// with text (sanitizer cleanup, c-lite member drop), and the pair of them is exactly the contract's +// derivation of unit state — shipped → translated, flagged without text → withheld. +type UnitDone struct { + Chapter int `json:"chapter"` + Unit int `json:"unit"` + Wave string `json:"wave"` // draft | edit + Shipped bool `json:"shipped"` + Flagged bool `json:"flagged"` + // Reason is the ENGINE's flag reason (glossary_miss, sanitizer_stripped, …). It is stored and + // never projected: the product phrase is applied at read time from the contract's map, so a + // reason this platform has never heard of still gets a neutral phrase instead of a hole. + Reason string `json:"reason,omitempty"` +} + +// BankStop is the signing stop. The full table travels as an artifact (engine backlog row 101), +// not through the stream: a thousand rows are not an event. +type BankStop struct { + TermsProposed int `json:"terms_proposed"` +} + +// Ceiling is the ceiling halt. It carries the FACT and nothing else: money never reaches the +// platform's wire or its INFO logs (D39.84), and the stop is resumable, so it is not a failure. +type Ceiling struct { + Halted bool `json:"halted"` +} + +// Finished is the terminal line. Outcome mirrors the engine's exit contract so a stream that ends +// cleanly needs no exit-code archaeology: clean | flagged | bank_stop | failed. +type Finished struct { + Outcome string `json:"outcome"` +} + +// checkVersion applies the semver rule to a handshake. +func checkVersion(got string) error { + gotMajor, err := major(got) + if err != nil { + return err + } + wantMajor, err := major(StreamVersion) + if err != nil { + return err + } + if gotMajor != wantMajor { + return fmt.Errorf("%w: stream is %s, this build speaks %s", ErrUnsupportedVersion, got, StreamVersion) + } + return nil +} + +func major(v string) (int, error) { + head, _, _ := strings.Cut(v, ".") + n, err := strconv.Atoi(head) + if err != nil { + return 0, fmt.Errorf("ingest: malformed stream version %q", v) + } + return n, nil +} diff --git a/platform/internal/ingest/resync.go b/platform/internal/ingest/resync.go new file mode 100644 index 00000000..752fd398 --- /dev/null +++ b/platform/internal/ingest/resync.go @@ -0,0 +1,56 @@ +package ingest + +import ( + "encoding/json" + "fmt" +) + +// StatusReport is the ALLOWLISTED subset of `tmctl status --json` (pipeline.StatusReport) that the +// platform materializes. Everything absent here is absent on purpose: snapshot ids, drift, rebill, +// routing, content labels and the operator's flag taxonomy are engine vocabulary that must not +// cross the seam (contract §2.12), and unknown JSON fields are simply ignored by encoding/json. +// +// ⚠ One field is money, and it is here for metering only — see SpendUSD. +// +// ⚠ Limit worth knowing: status has no PHASE split (engine backlog row 99). A resync can therefore +// restore the aggregate counter but not "draft N/M ∥ edit N/M"; the phase split lives only in the +// stream until row 99 lands. A reconciled run shows the aggregate until its next progress event. +type StatusReport struct { + BookID string `json:"book_id"` + TotalUnits int `json:"total_units"` + Done int `json:"done"` + InProgress int `json:"in_progress"` + Flagged int `json:"flagged"` + Pending int `json:"pending"` + // ETASeconds is what the owner asked to show (K-5); the engine already computes it. + ETASeconds float64 `json:"eta_seconds"` + // UnsignedBankTerms backs the signing screen's "N of M decided" while a stop is standing. + UnsignedBankTerms int `json:"unsigned_bank_terms"` + // SpendUSD is the engine's committed spend. The platform meters usage from its DELTA between + // attempts, because the event stream deliberately carries no figures. It is stored in the + // usage tables and NEVER projected into an API response or an INFO log (D39.84). + SpendUSD float64 `json:"committed_usd"` + Chapters []ChapterStatus `json:"chapters"` +} + +// ChapterStatus is the per-chapter passport, allowlisted the same way (no cost, no verdict ranks). +type ChapterStatus struct { + Chapter int `json:"chapter"` + UnitsTotal int `json:"units_total"` + UnitsDone int `json:"units_done"` + UnitsFlagged int `json:"units_flagged"` + UnitsInProgress int `json:"units_in_progress"` + UnitsPending int `json:"units_pending"` + // WorstFlagReason is engine vocabulary: stored, mapped to a product phrase at read time, never + // projected raw. + WorstFlagReason string `json:"worst_flag_reason"` +} + +// DecodeStatus parses a status report. +func DecodeStatus(b []byte) (StatusReport, error) { + var r StatusReport + if err := json.Unmarshal(b, &r); err != nil { + return StatusReport{}, fmt.Errorf("ingest: decode status: %w", err) + } + return r, nil +} diff --git a/platform/internal/ingest/sink.go b/platform/internal/ingest/sink.go new file mode 100644 index 00000000..1fc9f7e7 --- /dev/null +++ b/platform/internal/ingest/sink.go @@ -0,0 +1,53 @@ +package ingest + +import ( + "context" + "errors" + "io" +) + +// Sink materializes a stream into the reporting database. +// +// Idempotency lives HERE and not in the reader, because the effect and the high-water mark +// (run_attempts.last_seq) have to move in ONE transaction: an implementation that applies an event +// and then records it has a crash window that duplicates work. +// +// This is a reporting database, not an event store (research/23 §3): the events are not kept. +type Sink interface { + // Begin binds the stream to an attempt — the engine's run id is the idempotency namespace. + Begin(ctx context.Context, h Hello) error + // Apply materializes one event. It MUST ignore an event whose Seq is not greater than the + // stored high-water mark, and it MUST ignore an unknown Type. + Apply(ctx context.Context, ev Envelope) error +} + +// Ingest reads a stream to its end, feeding a sink. +// +// It stops at the first error and returns it: a gap or a malformed line is not recoverable by +// reading further — the caller reconciles from `tmctl status --json`, which is the whole reason +// that channel is part of the ratified seam. +func Ingest(ctx context.Context, r io.Reader, sink Sink) error { + d := NewDecoder(r) + h, err := d.Hello() + if err != nil { + return err + } + if err := sink.Begin(ctx, h); err != nil { + return err + } + for { + if err := ctx.Err(); err != nil { + return err + } + ev, err := d.Next() + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return err + } + if err := sink.Apply(ctx, ev); err != nil { + return err + } + } +} diff --git a/platform/internal/ingest/supervisor.go b/platform/internal/ingest/supervisor.go new file mode 100644 index 00000000..bd22a673 --- /dev/null +++ b/platform/internal/ingest/supervisor.go @@ -0,0 +1,118 @@ +package ingest + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "os/exec" + "time" +) + +// Outcome is the engine's ratified exit contract (cmd/tmctl/main.go:30-52). It is read from the +// exit code rather than from the log, and a bank stop is deliberately NOT a failure. +type Outcome string + +const ( + OutcomeClean Outcome = "clean" // 0 + OutcomeFailed Outcome = "failed" // 1 — infra failure, and anything unrecognised + OutcomeFlagged Outcome = "flagged" // 2 — completed with flagged units + OutcomeBankStop Outcome = "bank_stop" // 3 — deliberate human-in-the-loop halt +) + +func outcomeOf(code int) Outcome { + switch code { + case 0: + return OutcomeClean + case 2: + return OutcomeFlagged + case 3: + return OutcomeBankStop + default: + return OutcomeFailed + } +} + +// stopGrace is how long the engine has to shut down after the interrupt before it is killed. The +// engine stops gracefully on SIGINT/SIGTERM (signal.NotifyContext in tmctl's main), and it holds an +// EXCLUSIVE lock on its project file — a SIGKILL first would leave that lock behind. +const stopGrace = 30 * time.Second + +// Supervisor runs one tmctl process per attempt. +// +// Stream discipline (research/23 §2): stdout belongs to the event stream and to nothing else; +// the engine's own logs are on stderr and stay there. +type Supervisor struct { + // Bin is the tmctl binary. The platform never links the engine — it spawns it (D39.81). + Bin string + // Workdir is the book's project directory: book.yaml, the source and the engine's private + // SQLite live there. That SQLite is never opened by us (D39.85 §4). + Workdir string + // Env is the child's environment. Provider keys reach the engine through it and must never be + // logged; nil means the parent's environment. + Env []string + // EngineLog receives the child's stderr verbatim. It is a FILE, not our structured logger: the + // engine logs per-call cost estimates at INFO, and money must not enter the platform's INFO + // stream (D39.84). nil discards. + EngineLog io.Writer +} + +// Run spawns the engine and ingests its stream. It returns the outcome even when the stream itself +// failed, because "what did the process do" and "did we materialize all of it" are different +// questions: the second one is answered by reconciling with Status. +func (s *Supervisor) Run(ctx context.Context, sink Sink, args ...string) (Outcome, error) { + cmd := exec.CommandContext(ctx, s.Bin, args...) + cmd.Dir = s.Workdir + cmd.Env = s.Env + cmd.Stderr = s.engineLog() + // CommandContext kills on cancel by default; the engine needs the signal it already handles, + // and WaitDelay is the backstop if it ignores it. + cmd.Cancel = func() error { return cmd.Process.Signal(os.Interrupt) } + cmd.WaitDelay = stopGrace + + stdout, err := cmd.StdoutPipe() + if err != nil { + return OutcomeFailed, fmt.Errorf("ingest: stdout pipe: %w", err) + } + if err := cmd.Start(); err != nil { + return OutcomeFailed, fmt.Errorf("ingest: start %s: %w", s.Bin, err) + } + + ingestErr := Ingest(ctx, stdout, sink) + // Drain whatever is left so the child never blocks on a full pipe while we are waiting for it. + _, _ = io.Copy(io.Discard, stdout) + + waitErr := cmd.Wait() + var exitErr *exec.ExitError + switch { + case waitErr == nil: + return OutcomeClean, ingestErr + case errors.As(waitErr, &exitErr): + return outcomeOf(exitErr.ExitCode()), ingestErr + default: + return OutcomeFailed, errors.Join(waitErr, ingestErr) + } +} + +// Status runs the reconciliation channel: `tmctl status --json` on a stopped or finished run. It +// is read-only and free, but NOT free of CPU — every call re-ingests and re-chunks the source +// (1.4-1.5 s on a 23 MB book, engine backlog row 100), so it is a repair path, not a poll. +func (s *Supervisor) Status(ctx context.Context) (StatusReport, error) { + cmd := exec.CommandContext(ctx, s.Bin, "status", "--json") + cmd.Dir = s.Workdir + cmd.Env = s.Env + cmd.Stderr = s.engineLog() + out, err := cmd.Output() + if err != nil { + return StatusReport{}, fmt.Errorf("ingest: tmctl status: %w", err) + } + return DecodeStatus(out) +} + +func (s *Supervisor) engineLog() io.Writer { + if s.EngineLog == nil { + return io.Discard + } + return s.EngineLog +} diff --git a/platform/internal/ingest/supervisor_test.go b/platform/internal/ingest/supervisor_test.go new file mode 100644 index 00000000..6eb592b1 --- /dev/null +++ b/platform/internal/ingest/supervisor_test.go @@ -0,0 +1,93 @@ +package ingest + +import ( + "bytes" + "context" + "os" + "path/filepath" + "strconv" + "strings" + "testing" +) + +// fakeEngine writes a shell script that behaves like tmctl's contract: NDJSON on stdout, human log +// on stderr, and one of the ratified exit codes. The supervision path is exercised with a real +// process — a mocked one would prove nothing about pipes, signals or exit codes. +func fakeEngine(t *testing.T, stdout, stderr string, code int) string { + t.Helper() + path := filepath.Join(t.TempDir(), "tmctl") + script := "#!/bin/sh\nprintf '%s' " + shellQuote(stdout) + "\nprintf '%s' " + shellQuote(stderr) + " >&2\nexit " + strconv.Itoa(code) + "\n" + if err := os.WriteFile(path, []byte(script), 0o755); err != nil { + t.Fatal(err) + } + return path +} + +func TestRunIngestsStreamAndMapsExitCode(t *testing.T) { + stream := helloLine + "\n" + `{"seq":2,"type":"bank_stop","data":{"terms_proposed":7}}` + "\n" + var engineLog bytes.Buffer + s := &Supervisor{ + Bin: fakeEngine(t, stream, "bank-mining: new terms await owner signature\n", 3), + Workdir: t.TempDir(), + EngineLog: &engineLog, + } + sink := &recordingSink{} + + got, err := s.Run(context.Background(), sink, "translate", "--verify-bank") + if err != nil { + t.Fatalf("run: %v", err) + } + if got != OutcomeBankStop { + t.Fatalf("outcome = %q, want %q (exit 3 is a deliberate halt, not a failure)", got, OutcomeBankStop) + } + if len(sink.applied) != 1 || sink.applied[0].Type != TypeBankStop { + t.Fatalf("stream not materialized: %+v", sink.applied) + } + // The engine's own log must land in the file, not in our structured logger: it carries money. + if !bytes.Contains(engineLog.Bytes(), []byte("bank-mining")) { + t.Fatalf("engine stderr not captured: %q", engineLog.String()) + } +} + +func TestRunReportsOutcomeEvenWhenStreamBreaks(t *testing.T) { + s := &Supervisor{Bin: fakeEngine(t, "not a stream\n", "", 1), Workdir: t.TempDir()} + got, err := s.Run(context.Background(), &recordingSink{}) + if err == nil { + t.Fatal("a broken stream must be reported") + } + if got != OutcomeFailed { + t.Fatalf("outcome = %q, want %q", got, OutcomeFailed) + } +} + +func TestStatusDecodesTheResyncChannel(t *testing.T) { + // A real `tmctl status --json` body carries money and snapshot fields; the allowlist ignores + // them, and this fixture keeps one of each to prove it. + body := `{"book_id":"gzr","snapshot_id":"a1","committed_usd":1.25,"reserved_usd":0.5, + "total_units":10,"done":4,"pending":6,"eta_seconds":900,"unsigned_bank_terms":3, + "chapters":[{"chapter":1,"units_total":2,"units_done":2,"cost_usd":0.4,"verdict":"pass"}]}` + s := &Supervisor{Bin: fakeEngine(t, body, "", 0), Workdir: t.TempDir()} + got, err := s.Status(context.Background()) + if err != nil { + t.Fatalf("status: %v", err) + } + if got.BookID != "gzr" || got.Done != 4 || got.ETASeconds != 900 || got.UnsignedBankTerms != 3 { + t.Fatalf("status = %+v", got) + } + if got.SpendUSD != 1.25 { + t.Fatalf("spend must be metered from status: %v", got.SpendUSD) + } + if len(got.Chapters) != 1 || got.Chapters[0].UnitsDone != 2 { + t.Fatalf("chapters = %+v", got.Chapters) + } +} + +func TestOutcomeOfCoversTheExitContract(t *testing.T) { + for code, want := range map[int]Outcome{0: OutcomeClean, 1: OutcomeFailed, 2: OutcomeFlagged, 3: OutcomeBankStop, 7: OutcomeFailed} { + if got := outcomeOf(code); got != want { + t.Errorf("exit %d = %q, want %q", code, got, want) + } + } +} + +func shellQuote(s string) string { return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'" } diff --git a/platform/internal/pgstore/migrate.go b/platform/internal/pgstore/migrate.go new file mode 100644 index 00000000..2c01709d --- /dev/null +++ b/platform/internal/pgstore/migrate.go @@ -0,0 +1,54 @@ +// Package pgstore is the platform's Postgres access: migrations, the pool, and the queries the +// HTTP layer needs. It is the only package that speaks SQL. +package pgstore + +import ( + "context" + "database/sql" + "embed" + "fmt" + "io/fs" + + _ "github.com/jackc/pgx/v5/stdlib" // database/sql driver "pgx", used by goose only + "github.com/pressly/goose/v3" + "github.com/pressly/goose/v3/lock" +) + +//go:embed migrations/*.sql +var migrationsFS embed.FS + +// Migrations exposes the embedded SQL for tests that check the set without a database. +func Migrations() fs.FS { + sub, err := fs.Sub(migrationsFS, "migrations") + if err != nil { + panic(err) // the path is a compile-time constant of this package + } + return sub +} + +// Migrate applies every pending migration and returns when the schema is current. +// +// It opens its OWN database/sql handle rather than borrowing the pgx pool: goose speaks +// database/sql, and a one-connection handle is what the session locker needs anyway. The lock is a +// Postgres advisory lock, so two instances rolling out at once serialize instead of racing. +func Migrate(ctx context.Context, dsn string) error { + db, err := sql.Open("pgx", dsn) + if err != nil { + return fmt.Errorf("pgstore: open migration handle: %w", err) + } + defer db.Close() + db.SetMaxOpenConns(1) + + locker, err := lock.NewPostgresSessionLocker() + if err != nil { + return fmt.Errorf("pgstore: locker: %w", err) + } + p, err := goose.NewProvider(goose.DialectPostgres, db, Migrations(), goose.WithSessionLocker(locker)) + if err != nil { + return fmt.Errorf("pgstore: goose provider: %w", err) + } + if _, err := p.Up(ctx); err != nil { + return fmt.Errorf("pgstore: migrate: %w", err) + } + return nil +} diff --git a/platform/internal/pgstore/migrations/00001_identity.sql b/platform/internal/pgstore/migrations/00001_identity.sql new file mode 100644 index 00000000..759a33ef --- /dev/null +++ b/platform/internal/pgstore/migrations/00001_identity.sql @@ -0,0 +1,40 @@ +-- +goose Up + +-- ONE server-side session, two presentations: a __Host cookie (browser) or a Bearer token +-- (desktop/CLI). D39.84 — the principal is established in middleware only, so nothing below +-- knows which presentation was used. + +create table users ( + id text primary key, + email text not null, + created_at timestamptz not null default now(), + -- The library read spans books, so it cannot borrow a book's counter: it needs its own + -- monotonic scope. Revisions are compared WITHIN a scope only (K-4 proposal). + library_revision bigint not null default 0 +); + +-- Case-insensitive identity without the citext extension: an extension is a deploy-time +-- privilege we would need on every environment for one column. +create unique index users_email_key on users (lower(email)); + +create table sessions ( + -- The token itself is never stored. A leaked dump must not yield a usable credential, so the + -- lookup key IS the digest: 256 random bits are not brute-forcible, no salt is needed. + token_sha256 bytea primary key, + user_id text not null references users (id) on delete cascade, + created_at timestamptz not null default now(), + last_used_at timestamptz not null default now(), + -- Two clocks: idle expiry slides on use, absolute expiry never does — a stolen token cannot + -- be kept alive forever by touching it. + idle_expires_at timestamptz not null, + absolute_expires_at timestamptz not null, + revoked_at timestamptz +); + +create index sessions_user_idx on sessions (user_id); +-- The expiry sweep deletes by this; expired rows are removed, not archived. +create index sessions_absolute_expiry_idx on sessions (absolute_expires_at); + +-- +goose Down +drop table sessions; +drop table users; diff --git a/platform/internal/pgstore/migrations/00002_readmodel.sql b/platform/internal/pgstore/migrations/00002_readmodel.sql new file mode 100644 index 00000000..12e37d5a --- /dev/null +++ b/platform/internal/pgstore/migrations/00002_readmodel.sql @@ -0,0 +1,212 @@ +-- +goose Up + +-- The reporting database (research/23 §3): state materialized from the engine's event stream and +-- from `tmctl status --json`. It is NOT an event store — no history, no replay. Every table below +-- is a projection the API reads directly. +-- +-- Vocabularies are checked in DDL because the PLATFORM owns them: these values are produced by our +-- own materializer, so an unknown one is a defect, not data. Engine vocabulary (stage names, flag +-- reasons, dispositions) is deliberately absent — it never crosses this seam. + +create table books ( + id text primary key, + owner_id text not null references users (id) on delete cascade, + title text not null, + -- Codes, never names: a language name in a shared layer is pair-specific data (canon §2). + source_lang text not null, + target_lang text not null, + genre text not null default '', + status text not null check (status in ( + 'uploading', 'parsing', 'not_started', 'translating', 'awaiting_bank', 'finalizing', + 'ready', 'stopped', 'rejected', 'failed', 'paused')), + chapter_count integer not null default 0, + character_count bigint not null default 0, + note_count integer not null default 0, + added_at timestamptz not null default now(), + -- ONE monotonic counter per book. Every book-scoped read and every SSE frame id of that book's + -- run carry it, which is what makes "drop a read older than an event already applied" + -- decidable at all (K-4 proposal). Bumped once per materializing transaction; the rows that + -- transaction touched are stamped with the new value, so a delta read is `revision > $1`. + revision bigint not null default 0, + -- The engine project directory: book.yaml, the source file and the engine's private SQLite + -- live here. The platform never opens that SQLite (D39.85 §4) — it only spawns tmctl with this + -- as the working directory. + workdir text not null, + engine_book_id text not null, + source_sha256 bytea, + -- The chunker version the persisted manifest was produced under. A change re-numbers chapters + -- (engine backlog row 100), and chapter ids must survive that: they are ours, not the engine's. + chunker_version text not null default '' +); + +create index books_owner_idx on books (owner_id, added_at desc, id); + +create table runs ( + id text primary key, + book_id text not null references books (id) on delete cascade, + -- The book vocabulary minus the values that describe a book without a run. + status text not null check (status in ( + 'translating', 'awaiting_bank', 'finalizing', 'ready', 'stopped', 'failed', 'paused')), + verify_bank boolean not null, + started_at timestamptz not null default now(), + finished_at timestamptz, + -- Progress is per phase because a unit is done only once its edit resolved: one end-to-end + -- counter reads zero for the whole draft wave (contract §2.5; engine backlog row 99). + draft_done integer not null default 0, + draft_total integer not null default 0, + edit_done integer not null default 0, + edit_total integer not null default 0, + eta_seconds integer, + revision bigint not null default 0 +); + +-- One live run per book. The queue serializes by book_id (P-3) and the engine holds an EXCLUSIVE +-- lock on the project file anyway, so a second live run is a defect the database should refuse +-- rather than a state the API has to explain. +create unique index runs_one_live_per_book on runs (book_id) where finished_at is null; +create index runs_book_idx on runs (book_id, started_at desc); + +-- One engine PROCESS per attempt. The ratified idempotency key is (run_id, seq) — where run_id is +-- the ENGINE's per-invocation id, not the platform run id: resume spawns a new process whose seq +-- restarts at 1, so keying on the platform run would silently drop every event of attempt 2. +create table run_attempts ( + id bigint generated always as identity primary key, + run_id text not null references runs (id) on delete cascade, + attempt_no integer not null, + -- Echoed by the stream's `hello` frame (engine backlog rows 102/103). Null until the handshake + -- arrives: the row is created when the process is spawned, before it says anything. + engine_run_id text, + started_at timestamptz not null default now(), + ended_at timestamptz, + exit_code integer, + -- A high-water mark instead of an event log: the reporting database materializes state, it + -- does not source events (research/23 §3). A duplicate line inside one stream is dropped by + -- seq <= last_seq, and the mark moves in the same transaction as the effect it guards. + last_seq bigint not null default 0, + unique (run_id, attempt_no) +); + +create unique index run_attempts_engine_run_idx on run_attempts (engine_run_id); + +create table chapters ( + id text primary key, + book_id text not null references books (id) on delete cascade, + -- Displayed ordinal, NOT a key: numbering is dense, so editing the source shifts every later + -- chapter (chunk/chunker.go:99-105). The key is the opaque id, which the platform keeps + -- stable across re-chunks. + number integer not null, + -- Empty when the book's data carries no heading: there is no hardwired "Chapter N" form + -- (owner, 04.08 — K-3), and a book without chapter numbers is legal. + heading text not null default '', + units_total integer not null default 0, + -- K-10 is open. Both phase counters exist so that answering it is a projection change rather + -- than a migration; units_done stays the aggregate the resync path can restore today. + units_draft_done integer not null default 0, + units_edit_done integer not null default 0, + units_done integer not null default 0, + note_count integer not null default 0, + revision bigint not null default 0, + unique (book_id, number) +); + +create index chapters_book_order_idx on chapters (book_id, number); +-- Delta reads on reconnect: "everything in this book newer than the revision the client holds". +create index chapters_book_revision_idx on chapters (book_id, revision); + +create table units ( + id text primary key, + chapter_id text not null references chapters (id) on delete cascade, + ordinal integer not null, + source text not null, + target text not null default '', + -- Derived from the PAIR, not from the verdict: a flagged unit legally ships WITH text + -- (contract §2.7). The two checks below ARE that derivation, so a materializer bug that + -- writes "translated with no text" fails here instead of reaching a reader. + state text not null check (state in ('translated', 'withheld', 'pending')), + revision bigint not null default 0, + unique (chapter_id, ordinal), + constraint units_translated_has_text check (state <> 'translated' or target <> ''), + constraint units_unshipped_is_empty check (state = 'translated' or target = '') +); + +create index units_chapter_revision_idx on units (chapter_id, revision); + +create table notes ( + id text primary key, + book_id text not null references books (id) on delete cascade, + chapter_id text references chapters (id) on delete cascade, + unit_id text references units (id) on delete cascade, + -- The ENGINE's flag reason, stored and never projected. The product phrase and the severity + -- step are applied at read time from the contract's map (companion appendix A, K-6), so the + -- owner's wording lands as a data change with no backfill — and an unknown reason still gets + -- a neutral phrase instead of a hole. + reason text not null, + created_at timestamptz not null default now(), + revision bigint not null default 0 +); + +create index notes_book_revision_idx on notes (book_id, revision); +create index notes_unit_idx on notes (unit_id); + +create table bank_terms ( + id text primary key, + book_id text not null references books (id) on delete cascade, + src text not null, + dst text not null default '', + -- Null is legal and means "the engine did not decide": a ruby candidate that is neither a name + -- nor a place carries no type (membank/memseed.go:323-326). Such a row still needs signing. + kind text check (kind in ('name', 'place', 'title', 'term', 'nickname')), + status text not null check (status in ('auto', 'draft', 'approved')), + origin text not null check (origin in ('seed', 'ruby', 'mined')), + sense text not null default '', + since_chapter integer not null default 0, + until_chapter integer not null default 0, + revision bigint not null default 0, + -- The engine's own uniqueness key (store/migrate.go:202). Without the window the same src + -- arrives as several legal rows that look like duplicates. + unique (book_id, src, sense, since_chapter, until_chapter) +); + +create index bank_terms_book_revision_idx on bank_terms (book_id, revision); + +-- Signing is NOT a row edit: the pipeline replaces a book's whole glossary from its deterministic +-- inputs each run (seeding.go:18/110), so a decision is an accumulating instruction that the +-- worker writes into the mined-delta / mined-rejects files before resuming. +create table bank_decisions ( + book_id text not null references books (id) on delete cascade, + term_id text not null references bank_terms (id) on delete cascade, + action text not null check (action in ('promote', 'decline')), + dst text not null default '', + decided_at timestamptz not null default now(), + decided_by text not null references users (id), + primary key (book_id, term_id), + -- The contract's conditional requirement, enforced where it cannot be generated away: an + -- approved term with an empty rendering matches nothing yet reads as an intended one. + constraint bank_decisions_promote_has_dst check (action <> 'promote' or dst <> '') +); + +create table exports ( + id text primary key, + book_id text not null references books (id) on delete cascade, + format text not null, + ready boolean not null default false, + -- Storage key of the artifact. The download URL is minted per request for the owner and is + -- never indexable (PT-34), so it is not a column. + artifact text not null default '', + created_at timestamptz not null default now(), + ready_at timestamptz, + failed_reason text not null default '' +); + +create index exports_book_idx on exports (book_id, created_at desc); + +-- +goose Down +drop table exports; +drop table bank_decisions; +drop table bank_terms; +drop table notes; +drop table units; +drop table chapters; +drop table run_attempts; +drop table runs; +drop table books; diff --git a/platform/internal/pgstore/migrations/00003_usage.sql b/platform/internal/pgstore/migrations/00003_usage.sql new file mode 100644 index 00000000..1ff3d556 --- /dev/null +++ b/platform/internal/pgstore/migrations/00003_usage.sql @@ -0,0 +1,24 @@ +-- +goose Up + +-- P-5 draft (D39.100 / PT-35). Money is METERED here and never leaves: the API projects a state +-- and a percentage, never a sum (D39.84). Micro-USD integers, not floats — the engine's own ledger +-- is a lower bound and we must not add rounding drift on top of it. +-- +-- The platform meters from `tmctl status --json` (committed_usd) at attempt boundaries, because the +-- event stream deliberately carries no figures: the ceiling event says `halted`, nothing more. + +create table usage_windows ( + user_id text not null references users (id) on delete cascade, + period text not null check (period in ('day', 'week')), + started_at timestamptz not null, + ends_at timestamptz not null, + spent_micro_usd bigint not null default 0, + limit_micro_usd bigint not null, + primary key (user_id, period, started_at) +); + +-- The window a user is currently inside is the only one the limits page reads. +create index usage_windows_current_idx on usage_windows (user_id, ends_at desc); + +-- +goose Down +drop table usage_windows; diff --git a/platform/internal/pgstore/migrations_test.go b/platform/internal/pgstore/migrations_test.go new file mode 100644 index 00000000..1ef05ba6 --- /dev/null +++ b/platform/internal/pgstore/migrations_test.go @@ -0,0 +1,46 @@ +package pgstore + +import ( + "io/fs" + "regexp" + "strconv" + "strings" + "testing" +) + +// The migration SET is checkable without a database, and that check is worth having: a file whose +// name does not parse is not "skipped", it silently never runs. +func TestMigrationSetIsWellFormed(t *testing.T) { + names, err := fs.Glob(Migrations(), "*") + if err != nil { + t.Fatal(err) + } + if len(names) == 0 { + t.Fatal("no migrations embedded") + } + nameRe := regexp.MustCompile(`^(\d{5})_[a-z0-9_]+\.sql$`) + prev := 0 + for _, name := range names { + m := nameRe.FindStringSubmatch(name) + if m == nil { + t.Fatalf("%s: goose expects NNNNN_name.sql", name) + } + version, _ := strconv.Atoi(m[1]) + if version <= prev { + t.Fatalf("%s: versions must ascend and never repeat (previous %05d)", name, prev) + } + prev = version + + body, err := fs.ReadFile(Migrations(), name) + if err != nil { + t.Fatal(err) + } + for _, marker := range []string{"-- +goose Up", "-- +goose Down"} { + if !strings.Contains(string(body), marker) { + // A missing Down is not cosmetic: a rollout that cannot be rolled back is a + // one-way door, and goose reports it only when someone tries to walk back. + t.Fatalf("%s: missing %q", name, marker) + } + } + } +} diff --git a/platform/internal/pgstore/pg_test.go b/platform/internal/pgstore/pg_test.go new file mode 100644 index 00000000..e26e17c5 --- /dev/null +++ b/platform/internal/pgstore/pg_test.go @@ -0,0 +1,206 @@ +package pgstore + +import ( + "context" + "crypto/rand" + "encoding/hex" + "errors" + + "net/url" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + + "textmachine/platform/internal/auth" +) + +// The database-backed battery. It runs against TM_PLATFORM_TEST_DSN and skips loudly without it — +// `make check` names the skip, because an invisible skip reads as coverage. +// +// Each run gets its OWN database, created and dropped here: a test that leaves rows behind passes +// once and then lies. +func testDB(t *testing.T) (*Store, context.Context) { + t.Helper() + admin := os.Getenv("TM_PLATFORM_TEST_DSN") + if admin == "" { + t.Skip("TM_PLATFORM_TEST_DSN not set: schema and session tests need a live Postgres") + } + ctx := t.Context() + + var suffix [6]byte + if _, err := rand.Read(suffix[:]); err != nil { + t.Fatal(err) + } + name := "tm_platform_test_" + hex.EncodeToString(suffix[:]) + + adminConn, err := pgx.Connect(ctx, admin) + if err != nil { + t.Fatalf("connect: %v", err) + } + if _, err := adminConn.Exec(ctx, "create database "+pgx.Identifier{name}.Sanitize()); err != nil { + adminConn.Close(ctx) + t.Skipf("cannot create a scratch database (%v): grant CREATEDB or point the DSN at one", err) + } + t.Cleanup(func() { + dropCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + _, _ = adminConn.Exec(dropCtx, "drop database if exists "+pgx.Identifier{name}.Sanitize()+" with (force)") + adminConn.Close(dropCtx) + }) + + dsn := swapDatabase(t, admin, name) + if err := Migrate(ctx, dsn); err != nil { + t.Fatalf("migrate: %v", err) + } + // Twice, because a rollout re-runs it on every replica. + if err := Migrate(ctx, dsn); err != nil { + t.Fatalf("second migrate must be a no-op: %v", err) + } + s, err := Open(ctx, dsn) + if err != nil { + t.Fatal(err) + } + t.Cleanup(s.Close) + if err := s.Ping(ctx); err != nil { + t.Fatal(err) + } + return s, ctx +} + +func swapDatabase(t *testing.T, dsn, name string) string { + t.Helper() + u, err := url.Parse(dsn) + if err != nil { + t.Fatalf("TM_PLATFORM_TEST_DSN must be a URL: %v", err) + } + u.Path = "/" + name + return u.String() +} + +func TestSessionLifecycle(t *testing.T) { + s, ctx := testDB(t) + now := time.Now().UTC().Truncate(time.Millisecond) + seedUser(t, s, ctx, "u1") + + token := auth.NewToken() + digest := auth.Digest(token) + if err := s.CreateSession(ctx, digest, "u1", now, time.Hour, 24*time.Hour); err != nil { + t.Fatal(err) + } + + got, err := s.Lookup(ctx, digest, now) + if err != nil { + t.Fatalf("lookup: %v", err) + } + if got.UserID != "u1" { + t.Fatalf("session = %+v", got) + } + + // Expiry is a clause of the query, not a caller's duty. + if _, err := s.Lookup(ctx, digest, now.Add(2*time.Hour)); !errors.Is(err, auth.ErrNoSession) { + t.Fatalf("expired idle window: %v", err) + } + // Sliding never outlives the absolute deadline. + if err := s.Touch(ctx, digest, now.Add(30*time.Minute), 48*time.Hour); err != nil { + t.Fatal(err) + } + slid, err := s.Lookup(ctx, digest, now.Add(30*time.Minute)) + if err != nil { + t.Fatal(err) + } + if slid.IdleExpiresAt.After(slid.AbsoluteExpiresAt) { + t.Fatalf("idle %s outlived absolute %s", slid.IdleExpiresAt, slid.AbsoluteExpiresAt) + } + + if err := s.RevokeSession(ctx, digest, now); err != nil { + t.Fatal(err) + } + if _, err := s.Lookup(ctx, digest, now); !errors.Is(err, auth.ErrNoSession) { + t.Fatalf("revoked session still resolves: %v", err) + } + + n, err := s.DeleteExpiredSessions(ctx, now.Add(72*time.Hour)) + if err != nil { + t.Fatal(err) + } + if n != 1 { + t.Fatalf("sweep removed %d rows, want 1", n) + } +} + +func TestUnknownTokenIsIndistinguishable(t *testing.T) { + s, ctx := testDB(t) + if _, err := s.Lookup(ctx, auth.Digest("never-issued"), time.Now()); !errors.Is(err, auth.ErrNoSession) { + t.Fatalf("want ErrNoSession, got %v", err) + } +} + +// The read-model's derivation rules are DDL, so a materializer bug fails at the write instead of +// reaching a reader. These assert the constraints actually fire. +func TestReadModelConstraints(t *testing.T) { + s, ctx := testDB(t) + seedUser(t, s, ctx, "u1") + exec(t, s, ctx, `insert into books (id, owner_id, title, source_lang, target_lang, status, workdir, engine_book_id) + values ('bk1','u1','蛊真人','zh','ru','translating','/srv/books/bk1','gzr')`) + exec(t, s, ctx, `insert into chapters (id, book_id, number) values ('ch1','bk1',1)`) + + t.Run("translated needs text", func(t *testing.T) { + assertViolation(t, s, ctx, "units_translated_has_text", + `insert into units (id, chapter_id, ordinal, source, target, state) + values ('un1','ch1',1,'第一节','','translated')`) + }) + t.Run("withheld carries none", func(t *testing.T) { + assertViolation(t, s, ctx, "units_unshipped_is_empty", + `insert into units (id, chapter_id, ordinal, source, target, state) + values ('un2','ch1',2,'第二节','перевод','withheld')`) + }) + t.Run("promote needs a rendering", func(t *testing.T) { + exec(t, s, ctx, `insert into bank_terms (id, book_id, src, status, origin) values ('t1','bk1','方源','draft','mined')`) + assertViolation(t, s, ctx, "bank_decisions_promote_has_dst", + `insert into bank_decisions (book_id, term_id, action, dst, decided_by) values ('bk1','t1','promote','','u1')`) + }) + t.Run("kind may be absent but not invented", func(t *testing.T) { + // A ruby candidate that is neither a name nor a place carries no kind, and the row is still + // signable — the contract forbids inventing one for it. + exec(t, s, ctx, `insert into bank_terms (id, book_id, src, kind, status, origin) values ('t2','bk1','李',null,'draft','ruby')`) + assertViolation(t, s, ctx, "bank_terms_kind_check", + `insert into bank_terms (id, book_id, src, kind, status, origin) values ('t3','bk1','王','org','draft','ruby')`) + }) + t.Run("one live run per book", func(t *testing.T) { + exec(t, s, ctx, `insert into runs (id, book_id, status, verify_bank) values ('r1','bk1','translating',true)`) + assertViolation(t, s, ctx, "runs_one_live_per_book", + `insert into runs (id, book_id, status, verify_bank) values ('r2','bk1','translating',false)`) + }) + t.Run("the paused status exists", func(t *testing.T) { + // D39.100: a ceiling stop is the eleventh book status, and it is not `failed`. + exec(t, s, ctx, `update books set status='paused' where id='bk1'`) + assertViolation(t, s, ctx, "books_status_check", `update books set status='exhausted' where id='bk1'`) + }) +} + +func seedUser(t *testing.T, s *Store, ctx context.Context, id string) { + t.Helper() + exec(t, s, ctx, `insert into users (id, email) values ($1, $1 || '@example.org')`, id) +} + +func exec(t *testing.T, s *Store, ctx context.Context, sql string, args ...any) { + t.Helper() + if _, err := s.pool.Exec(ctx, sql, args...); err != nil { + t.Fatalf("exec %s: %v", sql, err) + } +} + +func assertViolation(t *testing.T, s *Store, ctx context.Context, constraint, sql string) { + t.Helper() + _, err := s.pool.Exec(ctx, sql) + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) { + t.Fatalf("want a constraint violation, got %v", err) + } + if pgErr.ConstraintName != constraint { + t.Fatalf("violated %q, want %q", pgErr.ConstraintName, constraint) + } +} diff --git a/platform/internal/pgstore/sessions.go b/platform/internal/pgstore/sessions.go new file mode 100644 index 00000000..70290d75 --- /dev/null +++ b/platform/internal/pgstore/sessions.go @@ -0,0 +1,85 @@ +package pgstore + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + + "textmachine/platform/internal/auth" +) + +// Lookup resolves a presented token. Expiry and revocation are clauses of THIS query, not checks a +// caller could forget: a row that comes back is live by construction. +func (s *Store) Lookup(ctx context.Context, digest []byte, now time.Time) (auth.Session, error) { + const q = ` + select user_id, idle_expires_at, absolute_expires_at + from sessions + where token_sha256 = $1 + and revoked_at is null + and idle_expires_at > $2 + and absolute_expires_at > $2` + var out auth.Session + err := s.pool.QueryRow(ctx, q, digest, now). + Scan(&out.UserID, &out.IdleExpiresAt, &out.AbsoluteExpiresAt) + if errors.Is(err, pgx.ErrNoRows) { + return auth.Session{}, auth.ErrNoSession + } + if err != nil { + return auth.Session{}, fmt.Errorf("pgstore: lookup session: %w", err) + } + return out, nil +} + +// Touch slides the idle window. It never moves the absolute expiry — that is the point of having +// two clocks — and it is called only in the window's second half, so reads stay reads. +func (s *Store) Touch(ctx context.Context, digest []byte, now time.Time, idleTTL time.Duration) error { + // Deadlines are computed in Go and travel as timestamps: one clock, one place, and no interval + // encoding to reason about. + const q = ` + update sessions + set last_used_at = $2, + idle_expires_at = least($3::timestamptz, absolute_expires_at) + where token_sha256 = $1 + and revoked_at is null + and absolute_expires_at > $2` + if _, err := s.pool.Exec(ctx, q, digest, now, now.Add(idleTTL)); err != nil { + return fmt.Errorf("pgstore: touch session: %w", err) + } + return nil +} + +// CreateSession stores a freshly minted token's digest. The plaintext never reaches this package. +func (s *Store) CreateSession(ctx context.Context, digest []byte, userID string, now time.Time, idleTTL, maxAge time.Duration) error { + const q = ` + insert into sessions (token_sha256, user_id, created_at, last_used_at, idle_expires_at, absolute_expires_at) + values ($1, $2, $3, $3, $4, $5)` + if _, err := s.pool.Exec(ctx, q, digest, userID, now, now.Add(idleTTL), now.Add(maxAge)); err != nil { + return fmt.Errorf("pgstore: create session: %w", err) + } + return nil +} + +// RevokeSession ends one session immediately — the property an opaque server-side session has and +// a self-verifying token does not. +func (s *Store) RevokeSession(ctx context.Context, digest []byte, now time.Time) error { + const q = `update sessions set revoked_at = $2 where token_sha256 = $1 and revoked_at is null` + if _, err := s.pool.Exec(ctx, q, digest, now); err != nil { + return fmt.Errorf("pgstore: revoke session: %w", err) + } + return nil +} + +// DeleteExpiredSessions is the sweep. Expired rows are deleted rather than kept: a session table is +// not an audit log, and "who was logged in last spring" is not a question we want to be able to +// answer from it. +func (s *Store) DeleteExpiredSessions(ctx context.Context, now time.Time) (int64, error) { + const q = `delete from sessions where absolute_expires_at <= $1` + tag, err := s.pool.Exec(ctx, q, now) + if err != nil { + return 0, fmt.Errorf("pgstore: sweep sessions: %w", err) + } + return tag.RowsAffected(), nil +} diff --git a/platform/internal/pgstore/store.go b/platform/internal/pgstore/store.go new file mode 100644 index 00000000..ac165da9 --- /dev/null +++ b/platform/internal/pgstore/store.go @@ -0,0 +1,37 @@ +package pgstore + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// Store is the platform's database handle. +type Store struct { + pool *pgxpool.Pool +} + +// Open builds the pool. It does NOT connect: pgxpool dials lazily, so a database that is down at +// boot makes the service unready rather than dead — readiness is the gate, not the process. +func Open(ctx context.Context, dsn string) (*Store, error) { + cfg, err := pgxpool.ParseConfig(dsn) + if err != nil { + return nil, fmt.Errorf("pgstore: parse dsn: %w", err) + } + pool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + return nil, fmt.Errorf("pgstore: pool: %w", err) + } + return &Store{pool: pool}, nil +} + +// Ping reports whether the database is reachable; it backs /readyz. +func (s *Store) Ping(ctx context.Context) error { + if err := s.pool.Ping(ctx); err != nil { + return fmt.Errorf("pgstore: ping: %w", err) + } + return nil +} + +func (s *Store) Close() { s.pool.Close() }