textmachine/platform/internal/pgstore/store.go

package pgstore

import (
	"context"
	"errors"
	"fmt"
	"io/fs"
	"sync"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/pressly/goose/v3"
)

// Store is the platform's database handle.
type Store struct {
	pool *pgxpool.Pool
}

// Pool limits. Bounded explicitly: how many connections a control plane may hold is a property of
// the database's max_connections, not of the machine running the binary.
const (
	defaultMaxConns = 16
	defaultMinConns = 2
)

// 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)
	}
	// Applied only where the operator said nothing, and "said nothing" is asked of pgx's own parser
	// rather than guessed from the resulting value. pgxpool reads pool_* out of RuntimeParams and
	// deletes them (pgxpool/pool.go), so a second parse still has them: that is the one place where
	// "the DSN mentions this key" is answered exactly, for both DSN forms, quoting and service files.
	//
	// Guessing was the previous form: it compared MaxConns with pgxpool's default of max(4, NumCPU)
	// and treated equality as "unset" — indistinguishable from an operator choosing that same number,
	// so a replica sized to its core count had its number silently replaced by ours. pool_min_conns
	// was worse: pgx's default is 0, so an explicit 0 could never survive.
	set, err := pgx.ParseConfig(dsn)
	if err != nil {
		return nil, fmt.Errorf("pgstore: parse dsn: %w", err)
	}
	if _, ok := set.RuntimeParams["pool_max_conns"]; !ok {
		cfg.MaxConns = defaultMaxConns
	}
	if _, ok := set.RuntimeParams["pool_min_conns"]; !ok {
		cfg.MinConns = defaultMinConns
	}
	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.
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
}

// Ready backs /readyz, and it asks a harder question than Ping: not "is a database there" but "is
// the database THIS BUILD was made for". Reachability alone answered yes against a Postgres with no
// tables at all — which is not a corner case but the normal middle of a rollout, because Migrate is
// off by default and the deploy notes make migrating a separate step. An instance in that window
// used to report ready and fail every query it then served.
//
// Only a schema BEHIND this binary is unready. A schema ahead of it is a newer release that has
// already migrated, and refusing to serve then would take the old instance down during the rollout
// it is supposed to survive. ⚠ That case is currently SILENT — nothing logs it, because the caller
// only logs the error branch and this Store has no logger. An operator running an old binary on a
// newer schema gets no signal from here.
func (s *Store) Ready(ctx context.Context) error {
	// No Ping first: the query below needs a connection and a round trip of its own, and it already
	// fails when the database is unreachable. Two round trips per probe, every few seconds, bought
	// nothing.
	want, err := latestMigration()
	if err != nil {
		return err
	}
	// Read directly with the pool rather than through a goose Provider: a Provider needs its own
	// database/sql handle, and this runs every few seconds.
	//
	// ⚠ goose.TableName() is the package-level legacy setting, which is NOT what goose.NewProvider
	// consults — the Provider resolves its own. They agree only because newProvider never passes
	// goose.WithTableName; adding it there without changing this would leave readiness querying a
	// table that does not exist and the service permanently unready.
	var applied int64
	err = s.pool.QueryRow(ctx,
		`select coalesce(max(version_id), 0) from `+pgx.Identifier{goose.TableName()}.Sanitize()+
			` where is_applied`).Scan(&applied)
	if err != nil {
		var pg *pgconn.PgError
		// 42P01: the version table itself is absent, so nothing was ever applied.
		if errors.As(err, &pg) && pg.Code == "42P01" {
			return fmt.Errorf("%w: no migrations have been applied; this build needs version %d", ErrSchemaBehind, want)
		}
		return fmt.Errorf("pgstore: read schema version: %w", err)
	}
	if applied < want {
		return fmt.Errorf("%w: schema is at version %d, this build needs %d", ErrSchemaBehind, applied, want)
	}
	// The queue keeps its own migration ledger (River owns that SQL), so goose's version says nothing
	// about it. Readiness has to cover it for the same reason it covers goose at all: an instance that
	// answers "ready" and then cannot accept a run has told the load balancer a falsehood (PD-68).
	var queue *string
	if err := s.pool.QueryRow(ctx, `select to_regclass('river_job')::text`).Scan(&queue); err != nil {
		return fmt.Errorf("pgstore: read queue schema: %w", err)
	}
	if queue == nil {
		return fmt.Errorf("%w: the queue schema is not installed", ErrSchemaBehind)
	}
	return nil
}

// ErrSchemaBehind is a database that answers but has not been migrated up to this binary.
var ErrSchemaBehind = errors.New("pgstore: schema is behind this build")

// latestMigration is the highest version embedded in this binary — a build-time constant, so it is
// computed once rather than on every readiness probe. The version is read by goose's own
// NumericComponent, so "what counts as the version of this filename" has one answer in the zone.
var latestMigration = sync.OnceValues(func() (int64, error) {
	entries, err := fs.ReadDir(Migrations(), ".")
	if err != nil {
		return 0, fmt.Errorf("pgstore: read embedded migrations: %w", err)
	}
	var latest int64
	for _, e := range entries {
		n, err := goose.NumericComponent(e.Name())
		if err != nil {
			return 0, fmt.Errorf("pgstore: migration %q: %w", e.Name(), err)
		}
		latest = max(latest, n)
	}
	if latest == 0 {
		return 0, errors.New("pgstore: no migrations are embedded in this binary")
	}
	return latest, nil
})

func (s *Store) Close() { s.pool.Close() }

// Pool is the connection pool, for the ONE library that has to speak SQL of its own: the queue owns
// its schema and its statements, and re-implementing them here to preserve a rule about who writes
// SQL would be a second, worse copy of somebody else's migrations.
func (s *Store) Pool() *pgxpool.Pool { return s.pool }