pg

package
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 16, 2026 License: Apache-2.0 Imports: 27 Imported by: 0

Documentation

Overview

Package pg provides peng's PostgreSQL dialect. Its Provider is backed by a pgxpool.Pool; seed functions receive pgx.Tx; the SessionLocker uses pg_advisory_lock for cross-process serialization.

This is one of several dialect packages under peng. Users pick the one matching their database and use it directly: 

import (
    "github.com/jackc/pgx/v5/pgxpool"
    "github.com/2kims/peng"
    "github.com/2kims/peng/pg"
)

pool, _ := pgxpool.New(ctx, dsn)
p, _ := pg.NewProvider(peng.Config{Dialect: peng.DialectPostgres}, pool)
p.Up(ctx)

Seed registration is per-dialect because the SeedFunc signature depends on the dialect's transaction type. pg.RegisterSeed adds to a pg-specific global registry; the auto-generated registry.go in peng/seeds/ wires those calls up at import time.

Index

Constants

View Source 
const DefaultLockName = "peng"

DefaultLockName is the lock identifier used by NewSessionLocker when no name is given.

Variables

View Source 
var ErrNoSeedRunner = errors.New("peng/pg: no SeedRunner configured for Go seeds (run `peng build` and pass pg.WithSeedRunner)")

ErrNoSeedRunner is returned by ApplySeed / ApplyAllSeeds when a Go seed is encountered but the Provider has no SeedRunner configured.

Functions

func HashName 

func HashName(name string) int64

HashName derives a deterministic int64 lock key from a name.

func SchemaModelFromPool 

func SchemaModelFromPool(ctx context.Context, pool *pgxpool.Pool) (*peng.Schema, error)

SchemaModelFromPool fetches the typed schema model from any pgxpool — used by callers that need a model from a database the Provider doesn't own (e.g. an auto-scratch DB during /schema/preview).

No advisory lock is taken; the caller is presumed to be the only user of the pool. This is fine for scratch databases (transient, single-purpose) but would be wrong for the live DB — that's why the Provider's [SchemaModel] takes the lock and this exported helper doesn't.

func WithAutoScratch 

func WithAutoScratch(ctx context.Context, liveDSN string, fn func(scratchDSN string) error) error

WithAutoScratch creates a temporary database on the same Postgres server as liveDSN, calls fn with the new database's DSN, then drops the database on return — even if fn errors. The "maintenance" connection (used to run CREATE DATABASE and DROP DATABASE) is opened against the conventional "postgres" database on the same server using the same credentials as liveDSN.

Requires the role authenticated by liveDSN to have CREATEDB privilege. Postgres on Docker / testcontainers / homebrew gives this by default; managed services like RDS / Cloud SQL may need an explicit GRANT.

The scratch database is named peng_diff_<unix>_<rand>, easy to identify and clean up manually if a peng run crashes before the defer fires.

liveDSN must be a URL-style DSN (postgres://...). Keyword-style DSNs (host=... dbname=...) are not supported by this helper — pass --scratch-dsn explicitly when using one.

Types

type DBTX 

type DBTX = pgstore.DBTX

DBTX is the subset of pgx types that the postgres Store implementation uses. Defined as a type alias to the sqlc-generated interface so the *postgres.Store adapter satisfies pg.Store without method-signature gymnastics.

Satisfied by *pgxpool.Pool, *pgxpool.Conn, *pgx.Conn, pgx.Tx.

type Option 

type Option func(*Provider) error

Option configures a Provider during construction.

func WithIsolationLevel 

func WithIsolationLevel(level pgx.TxIsoLevel) Option

WithIsolationLevel sets the transaction isolation level used by every per-migration and per-seed transaction.

func WithLocker 

func WithLocker(l SessionLocker) Option

WithLocker injects a SessionLocker, overriding the default.

func WithLogger 

func WithLogger(l *slog.Logger) Option

WithLogger overrides the default discard logger.

func WithOutOfOrder 

func WithOutOfOrder(policy peng.OutOfOrderPolicy) Option

WithOutOfOrder selects what Up does when it discovers a pending migration whose version is older than the current head. The default is peng.OutOfOrderError — Up refuses to apply such migrations until they're rebased. peng.OutOfOrderWarn relaxes that to "apply with a warning" for teams that don't want to rebase on every conflict.

func WithRetry 

func WithRetry(rp RetryPolicy) Option

WithRetry configures retries for migrations and seeds that fail with a transient Postgres error (serialization failure / deadlock).

func WithSeedRunner 

func WithSeedRunner(r SeedRunner) Option

WithSeedRunner injects a SeedRunner for executing Go seeds.

func WithStore 

func WithStore(s Store) Option

WithStore injects a Store implementation, bypassing the default.

func WithoutLocker 

func WithoutLocker() Option

WithoutLocker disables session-level locking.

type PostgresSessionLocker 

type PostgresSessionLocker struct {
	// contains filtered or unexported fields
}

PostgresSessionLocker uses pg_advisory_lock to serialize peng operations against the same database. The lock is held for the lifetime of the *pgxpool.Conn used to acquire it.

func NewSessionLocker 

func NewSessionLocker(name string) *PostgresSessionLocker

NewSessionLocker returns a locker keyed on name.

func (*PostgresSessionLocker) Lock 

func (l *PostgresSessionLocker) Lock(ctx context.Context, conn *pgxpool.Conn) error

func (*PostgresSessionLocker) Unlock 

func (l *PostgresSessionLocker) Unlock(ctx context.Context, conn *pgxpool.Conn) error

type Provider 

type Provider struct {
	// contains filtered or unexported fields
}

Provider is the postgres-specific implementation of peng operations. It satisfies cli.Provider but the CLI is the only place that uses the interface — direct users construct *Provider and call its methods.

func NewProvider 

func NewProvider(cfg peng.Config, pool *pgxpool.Pool, opts ...Option) (*Provider, error)

NewProvider constructs a postgres Provider.

func (*Provider) ApplyAllSeeds 

func (p *Provider) ApplyAllSeeds(ctx context.Context, opts ...peng.ApplySeedOption) ([]peng.SeedRunResult, error)

ApplyAllSeeds applies every discovered seed in version order, stopping at the first failure (the failing seed is left unrecorded; earlier successful seeds remain applied, mirroring migration semantics).

func (*Provider) ApplySeed 

func (p *Provider) ApplySeed(ctx context.Context, name string, opts ...peng.ApplySeedOption) (*peng.SeedRunResult, error)

ApplySeed applies a single seed by name from SeedsDir.

func (*Provider) Close 

func (p *Provider) Close()

Close closes the underlying pgxpool. Implements cli.Provider.

func (*Provider) Config 

func (p *Provider) Config() peng.Config

Config returns a copy of the Provider's configuration.

func (*Provider) Down 

func (p *Provider) Down(ctx context.Context) (*peng.Migration, error)

Down reverts the most recently applied migration (by version).

func (*Provider) Drift 

func (p *Provider) Drift(ctx context.Context) (peng.DriftReport, error)

Drift collects every drift condition: checksum mismatches, orphaned applied rows, and out-of-order pendings.

func (*Provider) Logger 

func (p *Provider) Logger() *slog.Logger

Logger returns the Provider's logger (slog by default).

func (*Provider) MarkApplied 

func (p *Provider) MarkApplied(ctx context.Context, migs []peng.Migration, force bool) error

MarkApplied records migs in peng_migrations as already-applied without executing their SQL. It is the backfill step of `peng adopt`: the migrations have already run against this database under another tool (or are a baseline of the live schema), so peng must only learn that they are done.

Each row is stored with the real checksum hashSQL(m.UpSQL) — the same hash Up records — so adopted migrations are drift-clean: a later `peng drift` compares the file against this checksum and finds no change. (This is deliberately unlike SchemaApply's empty-checksum sentinel, which marks "applied via snapshot, file not authoritative.")

Unless force is set, MarkApplied refuses a database whose peng_migrations table already has rows, mirroring SchemaApply's freshness guard — adopting twice would double-insert.

func (*Provider) Pool 

func (p *Provider) Pool() *pgxpool.Pool

Pool returns the underlying pgxpool.

func (*Provider) RunQuery 

func (p *Provider) RunQuery(ctx context.Context, sql string, args []any) (*peng.QueryResult, error)

RunQuery executes an arbitrary SQL statement against the live pool and returns the results as a QueryResult. Parameters must already be substituted to $N positional form (use QueryDef.BuildSQL first).

func (*Provider) SchemaApply 

func (p *Provider) SchemaApply(ctx context.Context, snap *peng.Snapshot) error

SchemaApply runs a snapshot against a fresh database. Refuses if peng_migrations already has any rows (use `peng up` to advance a non-fresh database). On success, every migration on disk with version <= snapshot.HeadVersion is recorded with an empty checksum — a sentinel meaning "applied via snapshot, not via Up." Drift detection ignores empty checksums, so backfilled rows don't trigger false positives.

func (*Provider) SchemaDiff 

func (p *Provider) SchemaDiff(ctx context.Context, liveDSN, scratchDSN string) (string, error)

SchemaDiff compares the live database to what peng's migrations declare. It applies every migration against scratchDSN (which must point at an empty database peng does NOT create or drop — caller's responsibility), pg_dumps both databases, normalizes pg_dump's noise out, and returns a unified diff.

Returns an empty string when the two schemas match after normalization. Otherwise the returned string is a unified diff where `-` lines are in live but not migrations (suspicious; possible manual edits to the live DB) and `+` lines are in migrations but not live (unapplied migrations or manual deletions from live).

Requires pg_dump on PATH. The peng bookkeeping tables (peng_migrations, peng_seeds) are excluded from both dumps because their differences are noise — they exist on both sides with different contents (apply order, durations).

func (*Provider) SchemaDump 

func (p *Provider) SchemaDump(ctx context.Context, dsn string) (*peng.Snapshot, error)

SchemaDump shells out to pg_dump to produce a schema-only snapshot of the target database. The bookkeeping tables (peng_migrations, peng_seeds) are excluded; the snapshot represents only the user schema.

Requires pg_dump on PATH. dsn must be a Postgres connection string that pg_dump understands (the same one used to build the pgxpool is fine).

func (*Provider) SchemaModel 

func (p *Provider) SchemaModel(ctx context.Context) (*peng.Schema, error)

SchemaModel introspects the live database and returns a structured peng.Schema suitable for visualization. Distinct from SchemaDump (which produces a pg_dump SQL string) and SchemaDiff (which compares two databases) — this method builds a typed model from pg_catalog and information_schema queries.

Per-Index details still aren't populated (the basic INDEX list is available via pg_index but adding it inflates the model with noise for small schemas; deferred until peng-tui has somewhere useful to render them).

func (*Provider) SchemaPreview 

func (p *Provider) SchemaPreview(ctx context.Context, liveDSN, scratchDSN string) (string, error)

SchemaPreview shows what running `peng up` would do to the live schema, without touching live. It mirrors live (schema + peng bookkeeping rows) to scratchDSN, captures that pre-state, applies the pending migrations on scratch, captures the post-state, and returns a unified diff.

Empty return means there are no pending migrations or applying them produces no schema change (rare but possible — e.g. a migration that only inserts data via a seed-like ALTER... DEFAULT).

scratchDSN must point at an empty database peng does NOT create or drop — caller's responsibility.

func (*Provider) SchemaPreviewModels 

func (p *Provider) SchemaPreviewModels(ctx context.Context, liveDSN, scratchDSN string) (diff string, before, after *peng.Schema, err error)

SchemaPreviewModels is [SchemaPreview] plus the typed schema models from both sides. Used by peng-tui's diagram-diff modal so the client can render a colored "after" diagram + per-table delta drawer without re-issuing introspection requests.

Returns the same unified-diff text as [SchemaPreview]; `before` is the live schema model; `after` is the scratch schema model after pending migrations have been applied. Auto-scratch handling stays with the caller (handlers wrap this in WithAutoScratch when no scratch DSN is configured).

func (*Provider) SeedStatus 

func (p *Provider) SeedStatus(ctx context.Context) ([]peng.SeedStatus, error)

SeedStatus returns the state of every discovered seed paired with its applied record (if any).

func (*Provider) Status 

func (p *Provider) Status(ctx context.Context) ([]peng.MigrationStatus, error)

Status returns the state of every migration in the migrations directory paired with its applied record (if any), plus drift flags (ChecksumChanged, OutOfOrder). Orphaned applied rows do not appear here — use Drift to surface them.

func (*Provider) Store 

func (p *Provider) Store() Store

Store returns the bookkeeping store.

func (*Provider) Up 

func (p *Provider) Up(ctx context.Context) ([]peng.Migration, error)

Up applies every pending migration in version order. When an out-of-order pending is detected (version older than the current head), behavior depends on the Provider's OutOfOrderPolicy: OutOfOrderError (default) returns an error before applying any pendings; OutOfOrderWarn logs a warning and applies anyway.

type RetryPolicy 

type RetryPolicy struct {
	MaxAttempts  int
	InitialDelay time.Duration
}

RetryPolicy configures how the Provider retries transactions that fail with a transient Postgres error (serialization failure 40001 or deadlock 40P01). The default zero value means "no retry".

type Seed 

type Seed struct {
	Version string
	Name    string
	Kind    SeedKind
	// Path is the absolute filesystem path. For SeedSQL it points at
	// the .sql file; for SeedGo it points at the package directory.
	Path string
}

Seed is one discovered seed under the Provider's SeedsDir.

func DiscoverSeeds 

func DiscoverSeeds(dir string) ([]Seed, error)

DiscoverSeeds returns the seeds under dir, sorted by version ascending. SQL seeds are top-level <version>_<name>.sql files; Go seeds are <version>_<name>/ subdirectories. Entries that don't match either shape are skipped silently. A non-existent dir returns an empty slice with no error so a freshly-init'd project doesn't trip.

type SeedKind 

type SeedKind int

SeedKind distinguishes how a discovered seed is executed.

SQL seeds are single .sql files at the top of SeedsDir; the Provider runs them inline in a single transaction.

Go seeds are subdirectories of SeedsDir whose contents are compiled into a seed-runner binary by `peng build`. Provider.ApplySeed delegates execution to a SeedRunner configured via WithSeedRunner — without one, applying a Go seed fails with a clear error. 

const (
	SeedSQL SeedKind = iota
	SeedGo
)

func (SeedKind) String 

func (k SeedKind) String() string

type SeedRunner 

type SeedRunner interface {
	Run(ctx context.Context, seed Seed) error
}

SeedRunner executes a Go seed out-of-process. M3 of the v1.0.0 refactor wires the default implementation that exec()s .peng/bin/seed-runner against the Provider's DSN; until then, Go seeds require an injected runner via WithSeedRunner or applying them returns ErrNoSeedRunner.

type SessionLocker 

type SessionLocker interface {
	Lock(ctx context.Context, conn *pgxpool.Conn) error
	Unlock(ctx context.Context, conn *pgxpool.Conn) error
}

SessionLocker holds a lock for the lifetime of a single *pgxpool.Conn. Both Lock and Unlock must be called on the same connection, and the connection must remain checked out from the pool while the lock is held.

type Store 

type Store interface {
	EnsureTables(ctx context.Context, db DBTX) error

	InsertMigration(ctx context.Context, db DBTX, version, name string, durationMs int64, checksum string) error
	DeleteMigration(ctx context.Context, db DBTX, version string) error
	GetMigration(ctx context.Context, db DBTX, version string) (*peng.MigrationRecord, error)
	ListMigrations(ctx context.Context, db DBTX) ([]peng.MigrationRecord, error)

	InsertSeed(ctx context.Context, db DBTX, version, name string, durationMs int64) error
	DeleteSeed(ctx context.Context, db DBTX, version string) error
	GetSeed(ctx context.Context, db DBTX, version string) (*peng.SeedRecord, error)
	ListSeeds(ctx context.Context, db DBTX) ([]peng.SeedRecord, error)
}

Store reads and writes peng's bookkeeping tables for the postgres dialect.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.