From 6374e174c67238ab92fc4d79ec09443c579f4fe5 Mon Sep 17 00:00:00 2001 From: Andy Dixon Date: Sun, 14 Jun 2026 23:08:37 +0100 Subject: [PATCH] feat: add configuration, logging, metrics, scheduler and storage foundation - config: defaults -> JSON file -> NOEMA_* env with validation and redaction - observability: slog with secret-redacting handler; Prometheus registry with the metric set from the plan (events, workspace, memory, LLM, tools, queues) - scheduler: supervised Group and bounded Pool so no goroutine is unowned - storage: pgx pool with statement timeout, embedded migration runner (advisory lock, checksums, up/down/status), per-test schema helper - migrations/0001: append-only guard trigger, users, sessions, api tokens, recovery codes, audit log, settings - pkg/ids: UUIDv7 with monotonic counter --- go.mod | 21 +++ go.sum | 50 ++++++ internal/config/config.go | 246 +++++++++++++++++++++++++++ internal/config/config_test.go | 51 ++++++ internal/observability/log.go | 62 +++++++ internal/observability/log_test.go | 17 ++ internal/observability/metrics.go | 63 +++++++ internal/scheduler/scheduler.go | 166 ++++++++++++++++++ internal/scheduler/scheduler_test.go | 63 +++++++ internal/storage/db.go | 89 ++++++++++ internal/storage/migrate.go | 222 ++++++++++++++++++++++++ internal/storage/migrate_test.go | 53 ++++++ internal/storage/testdb.go | 50 ++++++ internal/version/version.go | 5 + migrations/0001_foundation.down.sql | 7 + migrations/0001_foundation.up.sql | 85 +++++++++ migrations/embed.go | 11 ++ pkg/ids/ids.go | 112 ++++++++++++ pkg/ids/ids_test.go | 49 ++++++ 19 files changed, 1422 insertions(+) create mode 100644 go.sum create mode 100644 internal/config/config.go create mode 100644 internal/config/config_test.go create mode 100644 internal/observability/log.go create mode 100644 internal/observability/log_test.go create mode 100644 internal/observability/metrics.go create mode 100644 internal/scheduler/scheduler.go create mode 100644 internal/scheduler/scheduler_test.go create mode 100644 internal/storage/db.go create mode 100644 internal/storage/migrate.go create mode 100644 internal/storage/migrate_test.go create mode 100644 internal/storage/testdb.go create mode 100644 internal/version/version.go create mode 100644 migrations/0001_foundation.down.sql create mode 100644 migrations/0001_foundation.up.sql create mode 100644 migrations/embed.go create mode 100644 pkg/ids/ids.go create mode 100644 pkg/ids/ids_test.go diff --git a/go.mod b/go.mod index ac2a75a..741fc42 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,24 @@ module github.com/andydixon/noema go 1.26.1 + +require ( + github.com/jackc/pgx/v5 v5.11.0 + github.com/prometheus/client_golang v1.24.1 +) + +require ( + github.com/beorn7/perks v1.0.1 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + 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/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.70.1 // indirect + github.com/prometheus/procfs v0.21.1 // indirect + golang.org/x/sync v0.22.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.org/x/text v0.40.0 // indirect + google.golang.org/protobuf v1.36.11 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..c0032fa --- /dev/null +++ b/go.sum @@ -0,0 +1,50 @@ +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +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/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +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.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg= +github.com/jackc/pgx/v5 v5.11.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/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +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/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU= +github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY= +github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc= +github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI= +github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY= +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/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= +go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= +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.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +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= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +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= diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..850c451 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,246 @@ +// Package config loads daemon configuration with a predictable precedence: +// built-in defaults, then an optional JSON file, then NOEMA_* environment +// variables. Runtime tunables that live in the database are handled by the +// settings store, not here. +package config + +import ( + "encoding/json" + "errors" + "fmt" + "net/url" + "os" + "strconv" + "strings" + "time" +) + +// Config is the fully resolved daemon configuration. +type Config struct { + HTTPAddr string `json:"http_addr"` + Env string `json:"env"` // development | production + LogLevel string `json:"log_level"` + LogFormat string `json:"log_format"` // text | json + + DatabaseURL string `json:"database_url"` + MasterKey string `json:"master_key"` // base64, 32 bytes; secret + + BootstrapAdminUser string `json:"bootstrap_admin_user"` + BootstrapAdminPassword string `json:"bootstrap_admin_password"` // secret + + ResearchDir string `json:"research_dir"` + + MaxWorkers int `json:"max_workers"` + MaxConcurrentLLM int `json:"max_concurrent_llm"` + DailyTokenBudget int64 `json:"daily_token_budget"` + + SessionIdleTTL time.Duration `json:"session_idle_ttl"` + SessionAbsoluteTTL time.Duration `json:"session_absolute_ttl"` + MaxRequestBody int64 `json:"max_request_body"` +} + +// Defaults returns the built-in configuration. +func Defaults() Config { + return Config{ + HTTPAddr: "127.0.0.1:8080", + Env: "development", + LogLevel: "info", + LogFormat: "text", + BootstrapAdminUser: "admin", + ResearchDir: "./research", + MaxWorkers: 8, + MaxConcurrentLLM: 4, + DailyTokenBudget: 2_000_000, + SessionIdleTTL: 12 * time.Hour, + SessionAbsoluteTTL: 7 * 24 * time.Hour, + MaxRequestBody: 1 << 20, + } +} + +// Load resolves configuration from defaults, the optional file at path, and +// the environment provided by lookup (normally os.LookupEnv). +func Load(path string, lookup func(string) (string, bool)) (Config, error) { + c := Defaults() + if path != "" { + b, err := os.ReadFile(path) + if err != nil { + return c, fmt.Errorf("config: read %s: %w", path, err) + } + dec := json.NewDecoder(strings.NewReader(string(b))) + dec.DisallowUnknownFields() + var f fileConfig + if err := dec.Decode(&f); err != nil { + return c, fmt.Errorf("config: parse %s: %w", path, err) + } + f.apply(&c) + } + if err := c.applyEnv(lookup); err != nil { + return c, err + } + return c, c.Validate() +} + +// fileConfig mirrors Config but with durations as strings so JSON stays readable. +type fileConfig struct { + HTTPAddr *string `json:"http_addr"` + Env *string `json:"env"` + LogLevel *string `json:"log_level"` + LogFormat *string `json:"log_format"` + DatabaseURL *string `json:"database_url"` + MasterKey *string `json:"master_key"` + BootstrapAdminUser *string `json:"bootstrap_admin_user"` + BootstrapAdminPassword *string `json:"bootstrap_admin_password"` + ResearchDir *string `json:"research_dir"` + MaxWorkers *int `json:"max_workers"` + MaxConcurrentLLM *int `json:"max_concurrent_llm"` + DailyTokenBudget *int64 `json:"daily_token_budget"` + SessionIdleTTL *string `json:"session_idle_ttl"` + SessionAbsoluteTTL *string `json:"session_absolute_ttl"` + MaxRequestBody *int64 `json:"max_request_body"` +} + +func (f fileConfig) apply(c *Config) { + setS := func(dst *string, src *string) { + if src != nil { + *dst = *src + } + } + setS(&c.HTTPAddr, f.HTTPAddr) + setS(&c.Env, f.Env) + setS(&c.LogLevel, f.LogLevel) + setS(&c.LogFormat, f.LogFormat) + setS(&c.DatabaseURL, f.DatabaseURL) + setS(&c.MasterKey, f.MasterKey) + setS(&c.BootstrapAdminUser, f.BootstrapAdminUser) + setS(&c.BootstrapAdminPassword, f.BootstrapAdminPassword) + setS(&c.ResearchDir, f.ResearchDir) + if f.MaxWorkers != nil { + c.MaxWorkers = *f.MaxWorkers + } + if f.MaxConcurrentLLM != nil { + c.MaxConcurrentLLM = *f.MaxConcurrentLLM + } + if f.DailyTokenBudget != nil { + c.DailyTokenBudget = *f.DailyTokenBudget + } + if f.MaxRequestBody != nil { + c.MaxRequestBody = *f.MaxRequestBody + } + if f.SessionIdleTTL != nil { + if d, err := time.ParseDuration(*f.SessionIdleTTL); err == nil { + c.SessionIdleTTL = d + } + } + if f.SessionAbsoluteTTL != nil { + if d, err := time.ParseDuration(*f.SessionAbsoluteTTL); err == nil { + c.SessionAbsoluteTTL = d + } + } +} + +func (c *Config) applyEnv(lookup func(string) (string, bool)) error { + str := func(key string, dst *string) { + if v, ok := lookup("NOEMA_" + key); ok { + *dst = v + } + } + var errs []error + integer := func(key string, dst *int) { + if v, ok := lookup("NOEMA_" + key); ok { + n, err := strconv.Atoi(v) + if err != nil { + errs = append(errs, fmt.Errorf("NOEMA_%s: %w", key, err)) + return + } + *dst = n + } + } + integer64 := func(key string, dst *int64) { + if v, ok := lookup("NOEMA_" + key); ok { + n, err := strconv.ParseInt(v, 10, 64) + if err != nil { + errs = append(errs, fmt.Errorf("NOEMA_%s: %w", key, err)) + return + } + *dst = n + } + } + dur := func(key string, dst *time.Duration) { + if v, ok := lookup("NOEMA_" + key); ok { + d, err := time.ParseDuration(v) + if err != nil { + errs = append(errs, fmt.Errorf("NOEMA_%s: %w", key, err)) + return + } + *dst = d + } + } + str("HTTP_ADDR", &c.HTTPAddr) + str("ENV", &c.Env) + str("LOG_LEVEL", &c.LogLevel) + str("LOG_FORMAT", &c.LogFormat) + str("DATABASE_URL", &c.DatabaseURL) + str("MASTER_KEY", &c.MasterKey) + str("BOOTSTRAP_ADMIN_USER", &c.BootstrapAdminUser) + str("BOOTSTRAP_ADMIN_PASSWORD", &c.BootstrapAdminPassword) + str("RESEARCH_DIR", &c.ResearchDir) + integer("MAX_WORKERS", &c.MaxWorkers) + integer("MAX_CONCURRENT_LLM", &c.MaxConcurrentLLM) + integer64("DAILY_TOKEN_BUDGET", &c.DailyTokenBudget) + integer64("MAX_REQUEST_BODY", &c.MaxRequestBody) + dur("SESSION_IDLE_TTL", &c.SessionIdleTTL) + dur("SESSION_ABSOLUTE_TTL", &c.SessionAbsoluteTTL) + return errors.Join(errs...) +} + +// Validate checks invariants that would otherwise fail later and less clearly. +func (c Config) Validate() error { + var errs []error + if c.Env != "development" && c.Env != "production" { + errs = append(errs, fmt.Errorf("env must be development or production, got %q", c.Env)) + } + switch c.LogLevel { + case "debug", "info", "warn", "error": + default: + errs = append(errs, fmt.Errorf("log_level %q invalid", c.LogLevel)) + } + if c.LogFormat != "text" && c.LogFormat != "json" { + errs = append(errs, fmt.Errorf("log_format %q invalid", c.LogFormat)) + } + if c.DatabaseURL == "" { + errs = append(errs, errors.New("database_url is required")) + } else if _, err := url.Parse(c.DatabaseURL); err != nil { + errs = append(errs, fmt.Errorf("database_url: %w", err)) + } + if c.MaxWorkers < 1 || c.MaxConcurrentLLM < 1 { + errs = append(errs, errors.New("max_workers and max_concurrent_llm must be >= 1")) + } + if c.SessionIdleTTL <= 0 || c.SessionAbsoluteTTL < c.SessionIdleTTL { + errs = append(errs, errors.New("session ttls must be positive and absolute >= idle")) + } + if c.MaxRequestBody < 4096 { + errs = append(errs, errors.New("max_request_body too small")) + } + return errors.Join(errs...) +} + +// Production reports whether secure-by-default production behaviour applies. +func (c Config) Production() bool { return c.Env == "production" } + +// Redacted returns a copy safe for logging: secrets and DSN passwords removed. +func (c Config) Redacted() Config { + r := c + if r.MasterKey != "" { + r.MasterKey = "[set]" + } + if r.BootstrapAdminPassword != "" { + r.BootstrapAdminPassword = "[set]" + } + if u, err := url.Parse(r.DatabaseURL); err == nil && u.User != nil { + if _, has := u.User.Password(); has { + u.User = url.UserPassword(u.User.Username(), "xxxxx") + r.DatabaseURL = u.String() + } + } + return r +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..6724d95 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,51 @@ +package config + +import ( + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func env(m map[string]string) func(string) (string, bool) { + return func(k string) (string, bool) { v, ok := m[k]; return v, ok } +} + +func TestPrecedenceDefaultsFileEnv(t *testing.T) { + dir := t.TempDir() + p := filepath.Join(dir, "c.json") + os.WriteFile(p, []byte(`{"http_addr":"1.2.3.4:1","log_level":"debug","database_url":"postgres://u:p@h/db","session_idle_ttl":"1h","session_absolute_ttl":"2h"}`), 0o600) + c, err := Load(p, env(map[string]string{"NOEMA_LOG_LEVEL": "warn", "NOEMA_MAX_WORKERS": "3"})) + if err != nil { + t.Fatal(err) + } + if c.HTTPAddr != "1.2.3.4:1" || c.LogLevel != "warn" || c.MaxWorkers != 3 || c.SessionIdleTTL != time.Hour || c.Env != "development" { + t.Fatalf("unexpected %+v", c) + } +} + +func TestValidationAndRedaction(t *testing.T) { + if _, err := Load("", env(map[string]string{})); err == nil || !strings.Contains(err.Error(), "database_url") { + t.Fatalf("expected database_url error, got %v", err) + } + if _, err := Load("", env(map[string]string{"NOEMA_DATABASE_URL": "postgres://x", "NOEMA_MAX_WORKERS": "abc"})); err == nil { + t.Fatal("expected parse error") + } + c, err := Load("", env(map[string]string{"NOEMA_DATABASE_URL": "postgres://u:secret@h/db", "NOEMA_MASTER_KEY": "k"})) + if err != nil { + t.Fatal(err) + } + r := c.Redacted() + if strings.Contains(r.DatabaseURL, "secret") || r.MasterKey != "[set]" { + t.Fatalf("redaction failed: %+v", r) + } +} + +func TestUnknownFileKeyRejected(t *testing.T) { + p := filepath.Join(t.TempDir(), "c.json") + os.WriteFile(p, []byte(`{"databse_url":"x"}`), 0o600) + if _, err := Load(p, env(nil)); err == nil { + t.Fatal("expected unknown field error") + } +} diff --git a/internal/observability/log.go b/internal/observability/log.go new file mode 100644 index 0000000..31d1726 --- /dev/null +++ b/internal/observability/log.go @@ -0,0 +1,62 @@ +// Package observability configures structured logging and Prometheus metrics. +package observability + +import ( + "context" + "io" + "log/slog" + "strings" +) + +// secretKeys are attribute names whose values are never written to logs, +// whatever the caller passed. The list is deliberately broad. +var secretKeys = []string{"password", "passwd", "secret", "token", "api_key", "apikey", "authorization", "cookie", "master_key", "recovery_code"} + +// NewLogger builds a slog.Logger at the given level and format ("text" or +// "json") that redacts any attribute whose key looks like a secret. +func NewLogger(w io.Writer, level, format string) *slog.Logger { + var lvl slog.Level + switch level { + case "debug": + lvl = slog.LevelDebug + case "warn": + lvl = slog.LevelWarn + case "error": + lvl = slog.LevelError + default: + lvl = slog.LevelInfo + } + opts := &slog.HandlerOptions{Level: lvl, ReplaceAttr: redact} + var h slog.Handler + if format == "json" { + h = slog.NewJSONHandler(w, opts) + } else { + h = slog.NewTextHandler(w, opts) + } + return slog.New(h) +} + +func redact(_ []string, a slog.Attr) slog.Attr { + k := strings.ToLower(a.Key) + for _, s := range secretKeys { + if strings.Contains(k, s) { + return slog.String(a.Key, "[redacted]") + } + } + return a +} + +type ctxKey struct{} + +// WithLogger stores a request-scoped logger in ctx. +func WithLogger(ctx context.Context, l *slog.Logger) context.Context { + return context.WithValue(ctx, ctxKey{}, l) +} + +// Logger returns the request-scoped logger or the default logger. +func Logger(ctx context.Context) *slog.Logger { + if l, ok := ctx.Value(ctxKey{}).(*slog.Logger); ok { + return l + } + return slog.Default() +} diff --git a/internal/observability/log_test.go b/internal/observability/log_test.go new file mode 100644 index 0000000..a7c5afc --- /dev/null +++ b/internal/observability/log_test.go @@ -0,0 +1,17 @@ +package observability + +import ( + "bytes" + "strings" + "testing" +) + +func TestRedaction(t *testing.T) { + var buf bytes.Buffer + l := NewLogger(&buf, "info", "json") + l.Info("x", "api_key", "sk-123", "Password", "hunter2", "user", "andy") + out := buf.String() + if strings.Contains(out, "sk-123") || strings.Contains(out, "hunter2") || !strings.Contains(out, "andy") { + t.Fatalf("redaction wrong: %s", out) + } +} diff --git a/internal/observability/metrics.go b/internal/observability/metrics.go new file mode 100644 index 0000000..2308a96 --- /dev/null +++ b/internal/observability/metrics.go @@ -0,0 +1,63 @@ +package observability + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/collectors" +) + +// Metrics holds every Prometheus instrument Noema exposes. Instruments are +// created once here so that names are visible in one place and no package +// registers globals. +type Metrics struct { + Registry *prometheus.Registry + + HTTPRequests *prometheus.CounterVec + HTTPDuration *prometheus.HistogramVec + AuthFailures prometheus.Counter + Events *prometheus.CounterVec + Candidates prometheus.Counter + Admissions prometheus.Counter + Retrievals prometheus.Counter + Beliefs prometheus.Gauge + Predictions prometheus.Counter + PredictionErr prometheus.Counter + LLMRequests *prometheus.CounterVec + LLMTokens *prometheus.CounterVec + LLMFailures *prometheus.CounterVec + ToolRequests *prometheus.CounterVec + ToolDenied *prometheus.CounterVec + QueueDepth *prometheus.GaugeVec + Workers *prometheus.GaugeVec +} + +// NewMetrics creates and registers all instruments on a fresh registry. +func NewMetrics() *Metrics { + reg := prometheus.NewRegistry() + m := &Metrics{ + Registry: reg, + HTTPRequests: prometheus.NewCounterVec(prometheus.CounterOpts{Name: "noema_http_requests_total", Help: "HTTP requests by route and status."}, []string{"route", "status"}), + HTTPDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{Name: "noema_http_request_duration_seconds", Help: "HTTP request latency.", Buckets: prometheus.DefBuckets}, []string{"route"}), + AuthFailures: prometheus.NewCounter(prometheus.CounterOpts{Name: "noema_auth_failures_total", Help: "Failed authentication attempts."}), + Events: prometheus.NewCounterVec(prometheus.CounterOpts{Name: "noema_events_total", Help: "Cognitive events appended by type."}, []string{"type"}), + Candidates: prometheus.NewCounter(prometheus.CounterOpts{Name: "noema_workspace_candidates_total", Help: "Thought candidates scored."}), + Admissions: prometheus.NewCounter(prometheus.CounterOpts{Name: "noema_workspace_admissions_total", Help: "Thoughts admitted to the workspace."}), + Retrievals: prometheus.NewCounter(prometheus.CounterOpts{Name: "noema_memory_retrievals_total", Help: "Memory retrievals."}), + Beliefs: prometheus.NewGauge(prometheus.GaugeOpts{Name: "noema_beliefs_total", Help: "Beliefs currently held across minds."}), + Predictions: prometheus.NewCounter(prometheus.CounterOpts{Name: "noema_predictions_total", Help: "Predictions created."}), + PredictionErr: prometheus.NewCounter(prometheus.CounterOpts{Name: "noema_prediction_errors_total", Help: "Prediction error events."}), + LLMRequests: prometheus.NewCounterVec(prometheus.CounterOpts{Name: "noema_llm_requests_total", Help: "LLM requests by provider and role."}, []string{"provider", "role"}), + LLMTokens: prometheus.NewCounterVec(prometheus.CounterOpts{Name: "noema_llm_tokens_total", Help: "LLM tokens by provider and direction."}, []string{"provider", "direction"}), + LLMFailures: prometheus.NewCounterVec(prometheus.CounterOpts{Name: "noema_llm_failures_total", Help: "LLM failures by provider."}, []string{"provider"}), + ToolRequests: prometheus.NewCounterVec(prometheus.CounterOpts{Name: "noema_tool_requests_total", Help: "Capability invocations by capability."}, []string{"capability"}), + ToolDenied: prometheus.NewCounterVec(prometheus.CounterOpts{Name: "noema_tool_denied_total", Help: "Denied capability invocations."}, []string{"capability"}), + QueueDepth: prometheus.NewGaugeVec(prometheus.GaugeOpts{Name: "noema_cognitive_queue_depth", Help: "Pending work per queue."}, []string{"queue"}), + Workers: prometheus.NewGaugeVec(prometheus.GaugeOpts{Name: "noema_worker_count", Help: "Busy workers per pool."}, []string{"pool"}), + } + reg.MustRegister( + collectors.NewGoCollector(), collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}), + m.HTTPRequests, m.HTTPDuration, m.AuthFailures, m.Events, m.Candidates, m.Admissions, m.Retrievals, + m.Beliefs, m.Predictions, m.PredictionErr, m.LLMRequests, m.LLMTokens, m.LLMFailures, m.ToolRequests, + m.ToolDenied, m.QueueDepth, m.Workers, + ) + return m +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go new file mode 100644 index 0000000..3224b1e --- /dev/null +++ b/internal/scheduler/scheduler.go @@ -0,0 +1,166 @@ +// Package scheduler owns every background goroutine in the daemon. Nothing +// outside this package should use the go statement for long-lived work: that +// rule is what makes "no unbounded goroutines" auditable. +package scheduler + +import ( + "context" + "errors" + "log/slog" + "runtime/debug" + "sync" + "sync/atomic" + "time" +) + +// Group runs named goroutines bound to one context and waits for all of them. +// Panics inside a task are recovered, logged and reported as errors so a +// malformed input in one worker cannot take the daemon down. +type Group struct { + ctx context.Context + cancel context.CancelFunc + log *slog.Logger + wg sync.WaitGroup + mu sync.Mutex + errs []error + active atomic.Int64 +} + +// NewGroup creates a group whose tasks are cancelled when ctx ends or Stop is called. +func NewGroup(ctx context.Context, log *slog.Logger) *Group { + ctx, cancel := context.WithCancel(ctx) + return &Group{ctx: ctx, cancel: cancel, log: log} +} + +// Context returns the group's context. +func (g *Group) Context() context.Context { return g.ctx } + +// Go starts fn as a supervised goroutine. +func (g *Group) Go(name string, fn func(ctx context.Context) error) { + g.wg.Add(1) + g.active.Add(1) + go func() { + defer g.wg.Done() + defer g.active.Add(-1) + defer func() { + if r := recover(); r != nil { + g.log.Error("task panicked", "task", name, "panic", r, "stack", string(debug.Stack())) + g.record(errors.New("task " + name + " panicked")) + } + }() + if err := fn(g.ctx); err != nil && !errors.Is(err, context.Canceled) { + g.log.Error("task failed", "task", name, "err", err) + g.record(err) + } + }() +} + +// Every runs fn on a fixed interval until the context ends. Each run is +// recovered individually so a single failure does not stop the ticker. +func (g *Group) Every(name string, interval time.Duration, fn func(ctx context.Context) error) { + g.Go(name, func(ctx context.Context) error { + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return nil + case <-t.C: + g.safeRun(name, ctx, fn) + } + } + }) +} + +func (g *Group) safeRun(name string, ctx context.Context, fn func(context.Context) error) { + defer func() { + if r := recover(); r != nil { + g.log.Error("periodic task panicked", "task", name, "panic", r, "stack", string(debug.Stack())) + } + }() + if err := fn(ctx); err != nil && !errors.Is(err, context.Canceled) { + g.log.Warn("periodic task failed", "task", name, "err", err) + } +} + +func (g *Group) record(err error) { + g.mu.Lock() + g.errs = append(g.errs, err) + g.mu.Unlock() +} + +// Active returns the number of running tasks. +func (g *Group) Active() int64 { return g.active.Load() } + +// Stop cancels the group and waits for all tasks, returning any errors. +func (g *Group) Stop() error { + g.cancel() + g.wg.Wait() + g.mu.Lock() + defer g.mu.Unlock() + return errors.Join(g.errs...) +} + +// Wait blocks until all tasks have returned without cancelling them. +func (g *Group) Wait() error { + g.wg.Wait() + g.mu.Lock() + defer g.mu.Unlock() + return errors.Join(g.errs...) +} + +// Pool bounds concurrent execution of short tasks. Submit blocks when the pool +// is full (backpressure) rather than queueing without limit. +type Pool struct { + sem chan struct{} + busy atomic.Int64 +} + +// NewPool creates a pool allowing size concurrent tasks. +func NewPool(size int) *Pool { + if size < 1 { + size = 1 + } + return &Pool{sem: make(chan struct{}, size)} +} + +// Run executes fn once a slot is free, or returns ctx.Err() while waiting. +// It runs fn on the caller's goroutine; combine with Group.Go for async work. +func (p *Pool) Run(ctx context.Context, fn func(ctx context.Context) error) error { + select { + case p.sem <- struct{}{}: + case <-ctx.Done(): + return ctx.Err() + } + defer func() { <-p.sem }() + p.busy.Add(1) + defer p.busy.Add(-1) + return fn(ctx) +} + +// Busy returns the number of occupied slots. +func (p *Pool) Busy() int64 { return p.busy.Load() } + +// Size returns the pool capacity. +func (p *Pool) Size() int { return cap(p.sem) } + +// Parallel runs each task through the pool concurrently and waits for all of +// them, collecting errors. It never starts more than the pool size at once. +func Parallel(ctx context.Context, p *Pool, tasks []func(ctx context.Context) error) error { + var wg sync.WaitGroup + errs := make([]error, len(tasks)) + for i, task := range tasks { + wg.Add(1) + go func() { + defer wg.Done() + defer func() { + if r := recover(); r != nil { + errs[i] = errors.New("parallel task panicked") + } + }() + errs[i] = p.Run(ctx, task) + }() + } + wg.Wait() + return errors.Join(errs...) +} diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go new file mode 100644 index 0000000..af938aa --- /dev/null +++ b/internal/scheduler/scheduler_test.go @@ -0,0 +1,63 @@ +package scheduler + +import ( + "context" + "log/slog" + "sync/atomic" + "testing" + "time" +) + +func TestPoolBoundsConcurrency(t *testing.T) { + p := NewPool(3) + var cur, peak atomic.Int64 + tasks := make([]func(context.Context) error, 20) + for i := range tasks { + tasks[i] = func(ctx context.Context) error { + n := cur.Add(1) + for { + old := peak.Load() + if n <= old || peak.CompareAndSwap(old, n) { + break + } + } + time.Sleep(5 * time.Millisecond) + cur.Add(-1) + return nil + } + } + if err := Parallel(context.Background(), p, tasks); err != nil { + t.Fatal(err) + } + if peak.Load() > 3 { + t.Fatalf("peak concurrency %d exceeded pool size", peak.Load()) + } +} + +func TestGroupRecoversPanicAndCancels(t *testing.T) { + g := NewGroup(context.Background(), slog.New(slog.DiscardHandler)) + g.Go("boom", func(ctx context.Context) error { panic("x") }) + blocked := make(chan struct{}) + g.Go("waiter", func(ctx context.Context) error { close(blocked); <-ctx.Done(); return ctx.Err() }) + <-blocked + err := g.Stop() + if err == nil { + t.Fatal("expected panic error") + } + if g.Active() != 0 { + t.Fatal("tasks still active") + } +} + +func TestPoolRunHonoursCancellation(t *testing.T) { + p := NewPool(1) + ctx, cancel := context.WithCancel(context.Background()) + release := make(chan struct{}) + go p.Run(context.Background(), func(context.Context) error { <-release; return nil }) + time.Sleep(time.Millisecond) + cancel() + if err := p.Run(ctx, func(context.Context) error { return nil }); err != context.Canceled { + t.Fatalf("expected cancellation, got %v", err) + } + close(release) +} diff --git a/internal/storage/db.go b/internal/storage/db.go new file mode 100644 index 0000000..e225fee --- /dev/null +++ b/internal/storage/db.go @@ -0,0 +1,89 @@ +// Package storage opens the PostgreSQL pool, applies embedded migrations and +// provides small helpers shared by repositories. All SQL lives in the domain +// packages; this package knows nothing about minds or events. +package storage + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" +) + +// Open connects to PostgreSQL and verifies the connection. +func Open(ctx context.Context, url string) (*pgxpool.Pool, error) { + cfg, err := pgxpool.ParseConfig(url) + if err != nil { + return nil, fmt.Errorf("storage: parse url: %w", err) + } + if cfg.MaxConns < 8 { + cfg.MaxConns = 8 + } + cfg.MaxConnLifetime = time.Hour + cfg.HealthCheckPeriod = 30 * time.Second + if cfg.ConnConfig.RuntimeParams == nil { + cfg.ConnConfig.RuntimeParams = map[string]string{} + } + cfg.ConnConfig.RuntimeParams["application_name"] = "noema" + // A statement timeout bounds runaway queries from any subsystem (T18). + if _, set := cfg.ConnConfig.RuntimeParams["statement_timeout"]; !set { + cfg.ConnConfig.RuntimeParams["statement_timeout"] = "30000" + } + pool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + return nil, fmt.Errorf("storage: connect: %w", err) + } + pingCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + if err := pool.Ping(pingCtx); err != nil { + pool.Close() + return nil, fmt.Errorf("storage: ping: %w", err) + } + return pool, nil +} + +// ErrNotFound is returned by repositories when a row does not exist. +var ErrNotFound = errors.New("not found") + +// ErrConflict is returned when a uniqueness constraint is violated. +var ErrConflict = errors.New("conflict") + +// Translate maps driver errors to storage sentinels so callers can use errors.Is. +func Translate(err error) error { + if err == nil { + return nil + } + if errors.Is(err, pgx.ErrNoRows) { + return ErrNotFound + } + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == "23505" { + return fmt.Errorf("%w: %s", ErrConflict, pgErr.ConstraintName) + } + return err +} + +// Tx runs fn in a transaction, committing on nil error and rolling back otherwise. +func Tx(ctx context.Context, pool *pgxpool.Pool, fn func(tx pgx.Tx) error) error { + tx, err := pool.Begin(ctx) + if err != nil { + return fmt.Errorf("begin: %w", err) + } + defer tx.Rollback(ctx) //nolint:errcheck // rollback after commit is a no-op + if err := fn(tx); err != nil { + return err + } + return tx.Commit(ctx) +} + +// Querier is implemented by both *pgxpool.Pool and pgx.Tx so repositories can +// run inside or outside a transaction. +type Querier interface { + Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) + Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) + QueryRow(ctx context.Context, sql string, args ...any) pgx.Row +} diff --git a/internal/storage/migrate.go b/internal/storage/migrate.go new file mode 100644 index 0000000..125e941 --- /dev/null +++ b/internal/storage/migrate.go @@ -0,0 +1,222 @@ +package storage + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io/fs" + "sort" + "strings" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// migrationLock is an arbitrary advisory lock key so concurrent daemons do not +// race to apply the same migration. +const migrationLock = 0x6e6f656d61 // "noema" + +// Migration is one numbered schema step. +type Migration struct { + Version int + Name string + Up, Down string + Checksum string +} + +// MigrationStatus describes one migration and whether it has been applied. +type MigrationStatus struct { + Migration + Applied bool + AppliedAt string + Drift bool // applied checksum differs from the embedded file +} + +// LoadMigrations parses NNNN_name.up.sql / .down.sql pairs from fsys. +func LoadMigrations(fsys fs.FS) ([]Migration, error) { + entries, err := fs.ReadDir(fsys, ".") + if err != nil { + return nil, err + } + byVersion := map[int]*Migration{} + for _, e := range entries { + name := e.Name() + if !strings.HasSuffix(name, ".sql") { + continue + } + var version int + var rest string + if _, err := fmt.Sscanf(name, "%d_%s", &version, &rest); err != nil { + return nil, fmt.Errorf("migration %s: bad name", name) + } + body, err := fs.ReadFile(fsys, name) + if err != nil { + return nil, err + } + m := byVersion[version] + if m == nil { + m = &Migration{Version: version} + byVersion[version] = m + } + switch { + case strings.HasSuffix(name, ".up.sql"): + m.Name = strings.TrimSuffix(rest, ".up.sql") + m.Up = string(body) + sum := sha256.Sum256(body) + m.Checksum = hex.EncodeToString(sum[:]) + case strings.HasSuffix(name, ".down.sql"): + m.Down = string(body) + default: + return nil, fmt.Errorf("migration %s: must end in .up.sql or .down.sql", name) + } + } + out := make([]Migration, 0, len(byVersion)) + for _, m := range byVersion { + if m.Up == "" { + return nil, fmt.Errorf("migration %d has no up file", m.Version) + } + out = append(out, *m) + } + sort.Slice(out, func(i, j int) bool { return out[i].Version < out[j].Version }) + for i, m := range out { + if m.Version != i+1 { + return nil, fmt.Errorf("migrations must be contiguous from 1; got %d at position %d", m.Version, i+1) + } + } + return out, nil +} + +func ensureTable(ctx context.Context, tx pgx.Tx) error { + _, err := tx.Exec(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations ( + version integer PRIMARY KEY, name text NOT NULL, checksum text NOT NULL, applied_at timestamptz NOT NULL DEFAULT now())`) + return err +} + +// MigrateUp applies all pending migrations and returns the names applied. +func MigrateUp(ctx context.Context, pool *pgxpool.Pool, fsys fs.FS) ([]string, error) { + ms, err := LoadMigrations(fsys) + if err != nil { + return nil, err + } + var applied []string + err = withLock(ctx, pool, func(tx pgx.Tx) error { + if err := ensureTable(ctx, tx); err != nil { + return err + } + done, err := appliedVersions(ctx, tx) + if err != nil { + return err + } + for _, m := range ms { + if sum, ok := done[m.Version]; ok { + if sum != m.Checksum { + return fmt.Errorf("migration %04d_%s was modified after being applied", m.Version, m.Name) + } + continue + } + if _, err := tx.Exec(ctx, m.Up); err != nil { + return fmt.Errorf("migration %04d_%s: %w", m.Version, m.Name, err) + } + if _, err := tx.Exec(ctx, `INSERT INTO schema_migrations (version, name, checksum) VALUES ($1,$2,$3)`, m.Version, m.Name, m.Checksum); err != nil { + return err + } + applied = append(applied, fmt.Sprintf("%04d_%s", m.Version, m.Name)) + } + return nil + }) + return applied, err +} + +// MigrateDown rolls back the most recently applied migration. +func MigrateDown(ctx context.Context, pool *pgxpool.Pool, fsys fs.FS) (string, error) { + ms, err := LoadMigrations(fsys) + if err != nil { + return "", err + } + var rolled string + err = withLock(ctx, pool, func(tx pgx.Tx) error { + if err := ensureTable(ctx, tx); err != nil { + return err + } + var version int + err := tx.QueryRow(ctx, `SELECT coalesce(max(version),0) FROM schema_migrations`).Scan(&version) + if err != nil { + return err + } + if version == 0 { + return nil + } + m := ms[version-1] + if m.Down == "" { + return fmt.Errorf("migration %04d_%s has no down file", m.Version, m.Name) + } + if _, err := tx.Exec(ctx, m.Down); err != nil { + return fmt.Errorf("rollback %04d_%s: %w", m.Version, m.Name, err) + } + if _, err := tx.Exec(ctx, `DELETE FROM schema_migrations WHERE version=$1`, version); err != nil { + return err + } + rolled = fmt.Sprintf("%04d_%s", m.Version, m.Name) + return nil + }) + return rolled, err +} + +// Status reports every known migration and its applied state. +func Status(ctx context.Context, pool *pgxpool.Pool, fsys fs.FS) ([]MigrationStatus, error) { + ms, err := LoadMigrations(fsys) + if err != nil { + return nil, err + } + type row struct{ checksum, at string } + done := map[int]row{} + rows, err := pool.Query(ctx, `SELECT version, checksum, applied_at::text FROM schema_migrations`) + if err == nil { + defer rows.Close() + for rows.Next() { + var v int + var r row + if err := rows.Scan(&v, &r.checksum, &r.at); err != nil { + return nil, err + } + done[v] = r + } + } // a missing table simply means nothing is applied + out := make([]MigrationStatus, 0, len(ms)) + for _, m := range ms { + s := MigrationStatus{Migration: m} + if r, ok := done[m.Version]; ok { + s.Applied, s.AppliedAt, s.Drift = true, r.at, r.checksum != m.Checksum + } + out = append(out, s) + } + return out, nil +} + +func appliedVersions(ctx context.Context, tx pgx.Tx) (map[int]string, error) { + rows, err := tx.Query(ctx, `SELECT version, checksum FROM schema_migrations`) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[int]string{} + for rows.Next() { + var v int + var c string + if err := rows.Scan(&v, &c); err != nil { + return nil, err + } + out[v] = c + } + return out, rows.Err() +} + +func withLock(ctx context.Context, pool *pgxpool.Pool, fn func(tx pgx.Tx) error) error { + return Tx(ctx, pool, func(tx pgx.Tx) error { + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, int64(migrationLock)); err != nil { + return fmt.Errorf("migration lock: %w", err) + } + return fn(tx) + }) +} diff --git a/internal/storage/migrate_test.go b/internal/storage/migrate_test.go new file mode 100644 index 0000000..1afbc8e --- /dev/null +++ b/internal/storage/migrate_test.go @@ -0,0 +1,53 @@ +package storage + +import ( + "context" + "testing" + "testing/fstest" + + "github.com/andydixon/noema/migrations" +) + +func TestLoadMigrationsValidation(t *testing.T) { + if _, err := LoadMigrations(migrations.FS); err != nil { + t.Fatalf("embedded migrations invalid: %v", err) + } + bad := fstest.MapFS{"0002_x.up.sql": {Data: []byte("select 1")}} + if _, err := LoadMigrations(bad); err == nil { + t.Fatal("expected contiguity error") + } +} + +func TestMigrateUpDownAndAppendOnly(t *testing.T) { + pool := TestPool(t, migrations.FS) + ctx := context.Background() + st, err := Status(ctx, pool, migrations.FS) + if err != nil { + t.Fatal(err) + } + for _, s := range st { + if !s.Applied || s.Drift { + t.Fatalf("migration %d not cleanly applied: %+v", s.Version, s) + } + } + // Append-only trigger on audit_log. + if _, err := pool.Exec(ctx, `INSERT INTO audit_log (id, actor_kind, action) VALUES (gen_random_uuid(),'system','test')`); err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, `UPDATE audit_log SET action='x'`); err == nil { + t.Fatal("update on audit_log should be rejected") + } + if _, err := pool.Exec(ctx, `DELETE FROM audit_log`); err == nil { + t.Fatal("delete on audit_log should be rejected") + } + // Roll everything back and forward again. + for i := len(st); i > 0; i-- { + if _, err := MigrateDown(ctx, pool, migrations.FS); err != nil { + t.Fatalf("down %d: %v", i, err) + } + } + applied, err := MigrateUp(ctx, pool, migrations.FS) + if err != nil || len(applied) != len(st) { + t.Fatalf("re-apply: %v (%d)", err, len(applied)) + } +} diff --git a/internal/storage/testdb.go b/internal/storage/testdb.go new file mode 100644 index 0000000..5fe6cae --- /dev/null +++ b/internal/storage/testdb.go @@ -0,0 +1,50 @@ +package storage + +import ( + "context" + "crypto/rand" + "encoding/hex" + "io/fs" + "os" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// TestPool returns a pool bound to a fresh schema with all migrations applied. +// It skips the test when NOEMA_TEST_DATABASE_URL is unset. Each test gets its +// own schema so tests may run in parallel and leave nothing behind. +func TestPool(t testing.TB, migrations fs.FS) *pgxpool.Pool { + t.Helper() + url := os.Getenv("NOEMA_TEST_DATABASE_URL") + if url == "" { + t.Skip("NOEMA_TEST_DATABASE_URL not set") + } + ctx := context.Background() + var b [6]byte + rand.Read(b[:]) + schema := "t_" + hex.EncodeToString(b[:]) + + admin, err := Open(ctx, url) + if err != nil { + t.Fatalf("connect: %v", err) + } + if _, err := admin.Exec(ctx, "CREATE SCHEMA "+schema); err != nil { + t.Fatalf("create schema: %v", err) + } + cfg, _ := pgxpool.ParseConfig(url) + cfg.ConnConfig.RuntimeParams = map[string]string{"search_path": schema + ",public"} + pool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + t.Fatalf("connect schema: %v", err) + } + if _, err := MigrateUp(ctx, pool, migrations); err != nil { + t.Fatalf("migrate: %v", err) + } + t.Cleanup(func() { + pool.Close() + admin.Exec(ctx, "DROP SCHEMA "+schema+" CASCADE") //nolint:errcheck + admin.Close() + }) + return pool +} diff --git a/internal/version/version.go b/internal/version/version.go new file mode 100644 index 0000000..d2be884 --- /dev/null +++ b/internal/version/version.go @@ -0,0 +1,5 @@ +// Package version exposes build information injected at link time. +package version + +// Version is set via -ldflags "-X github.com/andydixon/noema/internal/version.Version=...". +var Version = "dev" diff --git a/migrations/0001_foundation.down.sql b/migrations/0001_foundation.down.sql new file mode 100644 index 0000000..0c0071a --- /dev/null +++ b/migrations/0001_foundation.down.sql @@ -0,0 +1,7 @@ +DROP TABLE IF EXISTS settings; +DROP TABLE IF EXISTS audit_log; +DROP TABLE IF EXISTS api_tokens; +DROP TABLE IF EXISTS sessions; +DROP TABLE IF EXISTS recovery_codes; +DROP TABLE IF EXISTS users; +DROP FUNCTION IF EXISTS noema_forbid_change(); diff --git a/migrations/0001_foundation.up.sql b/migrations/0001_foundation.up.sql new file mode 100644 index 0000000..3e63a06 --- /dev/null +++ b/migrations/0001_foundation.up.sql @@ -0,0 +1,85 @@ +-- Foundation: append-only guard, users, sessions, API tokens, audit log, settings. + +-- Raise on any UPDATE/DELETE. Attached to every history table so immutability +-- is enforced by the database, not by application discipline (ADR-0002). +CREATE OR REPLACE FUNCTION noema_forbid_change() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + RAISE EXCEPTION 'table % is append-only', TG_TABLE_NAME USING ERRCODE = 'integrity_constraint_violation'; +END $$; + +CREATE TABLE users ( + id uuid PRIMARY KEY, + username text NOT NULL, + password_hash text NOT NULL, + role text NOT NULL CHECK (role IN ('administrator','operator','observer')), + disabled boolean NOT NULL DEFAULT false, + failed_attempts integer NOT NULL DEFAULT 0, + locked_until timestamptz, + totp_secret_enc text, + totp_enabled boolean NOT NULL DEFAULT false, + password_changed_at timestamptz NOT NULL DEFAULT now(), + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() +); +CREATE UNIQUE INDEX users_username_lower_idx ON users (lower(username)); + +CREATE TABLE recovery_codes ( + id uuid PRIMARY KEY, + user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE, + code_hash bytea NOT NULL, + used_at timestamptz +); +CREATE INDEX recovery_codes_user_idx ON recovery_codes (user_id); + +CREATE TABLE sessions ( + id uuid PRIMARY KEY, + user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE, + token_hash bytea NOT NULL UNIQUE, + csrf_token text NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + last_seen_at timestamptz NOT NULL DEFAULT now(), + expires_at timestamptz NOT NULL, + ip text NOT NULL DEFAULT '', + user_agent text NOT NULL DEFAULT '', + revoked_at timestamptz +); +CREATE INDEX sessions_user_idx ON sessions (user_id) WHERE revoked_at IS NULL; + +CREATE TABLE api_tokens ( + id uuid PRIMARY KEY, + user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE, + name text NOT NULL, + token_hash bytea NOT NULL UNIQUE, + created_at timestamptz NOT NULL DEFAULT now(), + last_used_at timestamptz, + expires_at timestamptz, + revoked_at timestamptz +); +CREATE INDEX api_tokens_user_idx ON api_tokens (user_id); + +-- Administrative actions. Append-only. +CREATE TABLE audit_log ( + id uuid PRIMARY KEY, + seq bigserial NOT NULL, + at timestamptz NOT NULL DEFAULT now(), + actor_kind text NOT NULL CHECK (actor_kind IN ('user','system','mind')), + actor_id text NOT NULL DEFAULT '', + actor_name text NOT NULL DEFAULT '', + action text NOT NULL, + target_kind text NOT NULL DEFAULT '', + target_id text NOT NULL DEFAULT '', + details jsonb NOT NULL DEFAULT '{}'::jsonb, + ip text NOT NULL DEFAULT '' +); +CREATE INDEX audit_log_seq_idx ON audit_log (seq); +CREATE INDEX audit_log_target_idx ON audit_log (target_kind, target_id); +CREATE TRIGGER audit_log_append_only BEFORE UPDATE OR DELETE ON audit_log + FOR EACH ROW EXECUTE FUNCTION noema_forbid_change(); + +-- Runtime-editable settings (tunables only, never secrets). +CREATE TABLE settings ( + key text PRIMARY KEY, + value jsonb NOT NULL, + updated_at timestamptz NOT NULL DEFAULT now(), + updated_by text NOT NULL DEFAULT '' +); diff --git a/migrations/embed.go b/migrations/embed.go new file mode 100644 index 0000000..3d1814f --- /dev/null +++ b/migrations/embed.go @@ -0,0 +1,11 @@ +// Package migrations embeds the SQL schema migrations applied by +// internal/storage. Files are named NNNN_name.up.sql / NNNN_name.down.sql and +// are never edited once applied anywhere. +package migrations + +import "embed" + +// FS contains every migration file. +// +//go:embed *.sql +var FS embed.FS diff --git a/pkg/ids/ids.go b/pkg/ids/ids.go new file mode 100644 index 0000000..39dbed4 --- /dev/null +++ b/pkg/ids/ids.go @@ -0,0 +1,112 @@ +// Package ids generates UUIDv7 identifiers. +// +// UUIDv7 is used for every Noema record because it is globally unique and +// time-ordered, which keeps event indexes dense and makes ids sort by creation +// time without a second column. +package ids + +import ( + "crypto/rand" + "encoding/hex" + "errors" + "sync" + "time" +) + +// ID is a 128-bit UUID in canonical textual form. +type ID string + +// Nil is the zero ID. +const Nil ID = "" + +var ( + mu sync.Mutex + lastMS int64 + lastSeq uint16 + randPool = rand.Reader +) + +// New returns a new UUIDv7. Within the same millisecond ids are made strictly +// increasing by a 12-bit counter so ordering is stable even under bursts. +func New() ID { + var b [16]byte + if _, err := randPool.Read(b[6:]); err != nil { + panic("ids: crypto/rand unavailable: " + err.Error()) // no safe fallback exists + } + ms := time.Now().UnixMilli() + mu.Lock() + if ms <= lastMS { + ms = lastMS + lastSeq++ + if lastSeq >= 1<<12 { // counter overflow: borrow a millisecond + lastMS++ + ms = lastMS + lastSeq = 0 + } + } else { + lastMS = ms + lastSeq = uint16(b[6])<<4 | uint16(b[7])>>4 // random start, per RFC 9562 method 1 + lastSeq &= 0x7ff // leave headroom for increments + } + seq := lastSeq + mu.Unlock() + + b[0] = byte(ms >> 40) + b[1] = byte(ms >> 32) + b[2] = byte(ms >> 24) + b[3] = byte(ms >> 16) + b[4] = byte(ms >> 8) + b[5] = byte(ms) + b[6] = 0x70 | byte(seq>>8) // version 7 + high 4 bits of counter + b[7] = byte(seq) // low 8 bits of counter + b[8] = 0x80 | (b[8] & 0x3f) // RFC 4122 variant + return ID(format(b)) +} + +func format(b [16]byte) string { + var out [36]byte + hex.Encode(out[0:8], b[0:4]) + out[8] = '-' + hex.Encode(out[9:13], b[4:6]) + out[13] = '-' + hex.Encode(out[14:18], b[6:8]) + out[18] = '-' + hex.Encode(out[19:23], b[8:10]) + out[23] = '-' + hex.Encode(out[24:36], b[10:16]) + return string(out[:]) +} + +// Parse validates the canonical textual form and returns it lower-cased. +func Parse(s string) (ID, error) { + if len(s) != 36 || s[8] != '-' || s[13] != '-' || s[18] != '-' || s[23] != '-' { + return Nil, errors.New("ids: invalid uuid format") + } + var b [16]byte + src := s[0:8] + s[9:13] + s[14:18] + s[19:23] + s[24:36] + if _, err := hex.Decode(b[:], []byte(src)); err != nil { + return Nil, errors.New("ids: invalid uuid hex") + } + return ID(format(b)), nil +} + +// Time extracts the millisecond timestamp embedded in a UUIDv7. It returns the +// zero time for ids that are not version 7. +func (id ID) Time() time.Time { + if len(id) != 36 { + return time.Time{} + } + var b [16]byte + src := string(id[0:8] + id[9:13] + id[14:18] + id[19:23] + id[24:36]) + if _, err := hex.Decode(b[:], []byte(src)); err != nil || b[6]>>4 != 7 { + return time.Time{} + } + ms := int64(b[0])<<40 | int64(b[1])<<32 | int64(b[2])<<24 | int64(b[3])<<16 | int64(b[4])<<8 | int64(b[5]) + return time.UnixMilli(ms) +} + +// String implements fmt.Stringer. +func (id ID) String() string { return string(id) } + +// IsZero reports whether the id is unset. +func (id ID) IsZero() bool { return id == Nil } diff --git a/pkg/ids/ids_test.go b/pkg/ids/ids_test.go new file mode 100644 index 0000000..24abd04 --- /dev/null +++ b/pkg/ids/ids_test.go @@ -0,0 +1,49 @@ +package ids + +import ( + "sort" + "testing" + "time" +) + +func TestNewIsOrderedUniqueAndParsable(t *testing.T) { + const n = 5000 + seen := make(map[ID]struct{}, n) + prev := ID("") + for i := 0; i < n; i++ { + id := New() + if _, dup := seen[id]; dup { + t.Fatalf("duplicate id %s", id) + } + seen[id] = struct{}{} + if id <= prev { + t.Fatalf("ids not increasing: %s after %s", id, prev) + } + prev = id + if p, err := Parse(string(id)); err != nil || p != id { + t.Fatalf("parse round trip failed for %s: %v", id, err) + } + if id[14] != '7' { + t.Fatalf("not version 7: %s", id) + } + } + ids := make([]string, 0, n) + for id := range seen { + ids = append(ids, string(id)) + } + sort.Strings(ids) + if d := time.Since(ID(ids[0]).Time()); d < 0 || d > time.Minute { + t.Fatalf("embedded time implausible: %v", d) + } +} + +func TestParseRejectsGarbage(t *testing.T) { + for _, s := range []string{"", "not-a-uuid", "0192b7e4-3c7a-7c3e-8f1e-0123456789zz", "0192b7e43c7a7c3e8f1e0123456789ab"} { + if _, err := Parse(s); err == nil { + t.Errorf("expected error for %q", s) + } + } + if _, err := Parse("0192B7E4-3C7A-7C3E-8F1E-0123456789AB"); err != nil { + t.Errorf("upper case should parse: %v", err) + } +}