Andy Dixon
6374e174c6
- 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
90 lines
2.7 KiB
Go
90 lines
2.7 KiB
Go
// 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
|
|
}
|