postgresql

README

postgresql

A samsara-compatible PostgreSQL component backed by pgx/v5 connection pooling.

go get github.com/sunkek/samsara-components/postgresql

---

Usage

Register with a supervisor

db := postgresql.New(postgresql.Config{
    Host: "localhost",
    Port: 5432,
    Name: "mydb",
    User: "myuser",
    Pass: "secret",
})
sup.Add(db,
    samsara.WithTier(samsara.TierCritical),
    samsara.WithRestartPolicy(samsara.ExponentialBackoff(5, time.Second)),
)

Or supply a full DSN:

db := postgresql.New(postgresql.Config{
    URI: "postgres://user:pass@host:5432/db?sslmode=require",
})

Use in domain adapters

Depend on the DB interface, not *Component, so adapters stay testable without a real database:

type UserRepo struct {
    db postgresql.DB
}

func (r *UserRepo) FindByID(ctx context.Context, id uuid.UUID) (*User, error) {
    var u User
    err := r.db.Get(ctx, &u, `SELECT * FROM users WHERE id = $1`, id)
    if errors.Is(err, postgresql.ErrNoRows) {
        return nil, ErrNotFound
    }
    return &u, err
}

Transactions

func (r *UserRepo) Transfer(ctx context.Context, from, to uuid.UUID, amount int) (err error) {
    tx, err := r.db.BeginTx(ctx, pgx.TxOptions{})
    if err != nil {
        return err
    }
    defer func() { err = r.db.CommitTx(ctx, tx, err) }()

    if _, err = tx.Exec(ctx, `UPDATE accounts SET balance = balance - $1 WHERE id = $2`, amount, from); err != nil {
        return err
    }
    if _, err = tx.Exec(ctx, `UPDATE accounts SET balance = balance + $1 WHERE id = $2`, amount, to); err != nil {
        return err
    }
    return nil
}

CommitTx commits when err is nil and rolls back when it is non-nil, preserving both errors in the chain so callers can use errors.Is.

---

Configuration

postgresql.Config{
    // Individual fields (all have sensible defaults)
    Host    string        // default: "localhost"
    Port    int           // default: 5432
    Name    string        // default: "postgres"
    User    string        // default: "postgres"
    Pass    string
    SSLMode string        // default: "disable"

    // DSN override — takes precedence over individual fields when non-empty
    URI string

    // Timeouts and pool sizing
    ConnectTimeout time.Duration // default: 10s — startup ping deadline only
    MaxConns       int32         // default: pgx default (min(4, GOMAXPROCS))
    MinConns       int32         // default: 0
}

Options

postgresql.WithLogger(slog.Default())           // attach a structured logger
postgresql.WithName("postgres-replica")         // override component name

---

API reference

DB interface

Method	Description
Select(ctx, &slice, sql, args...)	Scan all rows into a slice of structs
Get(ctx, &struct, sql, args...)	Scan one row; returns ErrNoRows if empty
Exec(ctx, sql, args...)	Execute without scanning; returns command tag
BeginTx(ctx, pgx.TxOptions{})	Start a transaction
CommitTx(ctx, tx, err)	Commit or roll back; use in a defer

*Component satisfies DB. Struct fields are mapped by the db tag.

Sentinel errors

errors.Is(err, postgresql.ErrNoRows) // no rows matched the query

---

Health checking

*Component implements samsara.HealthChecker — the supervisor polls Health(ctx) every health interval and flips /readyz accordingly. No configuration required.

---

Multiple instances (primary + replica)

primary := postgresql.New(cfg.Primary, postgresql.WithName("postgres-primary"))
replica := postgresql.New(cfg.Replica, postgresql.WithName("postgres-replica"))

sup.Add(primary, samsara.WithTier(samsara.TierCritical))
sup.Add(replica, samsara.WithTier(samsara.TierSignificant))

---

Testing adapters without a database

type mockDB struct{ postgresql.DB }

func (m *mockDB) Get(_ context.Context, dst any, _ string, _ ...any) error {
    *dst.(*User) = User{ID: uuid.New(), Name: "test"}
    return nil
}

Or use the TxFinaliser interface to stub transactions:

type fakeTx struct{}
func (f *fakeTx) Commit(_ context.Context) error   { return nil }
func (f *fakeTx) Rollback(_ context.Context) error { return nil }

Documentation

Overview

Package postgresql provides a github.com/sunkek/samsara-compatible PostgreSQL component backed by pgx/v5 connection pooling.

Usage

comp := postgresql.New(postgresql.Config{
    Host: "localhost",
    Port: 5432,
    Name: "mydb",
    User: "myuser",
    Pass: "secret",
})
sup.Add(comp, samsara.WithTier(samsara.TierCritical))

Domain adapters receive *Component (or the DB interface) and use its Select, Get, Exec, and transaction helpers — they never import pgx or pgxscan directly.

Index

• Variables
• type Component
  • func New(cfg Config, opts ...Option) *Component
  • func (c *Component) BeginTx(ctx context.Context, opts pgx.TxOptions) (pgx.Tx, error)
  • func (c *Component) CommitTx(ctx context.Context, tx TxFinaliser, inErr error) error
  • func (c *Component) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
  • func (c *Component) Get(ctx context.Context, dst any, sql string, args ...any) error
  • func (c *Component) Health(ctx context.Context) error
  • func (c *Component) Name() string
  • func (c *Component) Select(ctx context.Context, dst any, sql string, args ...any) error
  • func (c *Component) Start(ctx context.Context, ready func()) error
  • func (c *Component) Stop(ctx context.Context) error
• type Config
• type DB
• type Logger
• type Option
  • func WithLogger(l Logger) Option
  • func WithName(name string) Option
• type TxFinaliser

Variables

var ErrNoRows = pgx.ErrNoRows

ErrNoRows is returned by [Select] and [Get] when no rows match the query. Use errors.Is(err, postgresql.ErrNoRows) to check.

Types

type Component

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

Component is a samsara-compatible PostgreSQL component. Obtain one with New; register it with a samsara supervisor.

Domain adapters should accept DB rather than *Component to keep their tests independent of a real database.

func New

func New(cfg Config, opts ...Option) *Component

New creates a Component from the supplied config. The component is not connected until Component.Start is called.

func (*Component) BeginTx

func (c *Component) BeginTx(ctx context.Context, opts pgx.TxOptions) (pgx.Tx, error)

BeginTx starts a new transaction.

func (*Component) CommitTx

func (c *Component) CommitTx(ctx context.Context, tx TxFinaliser, inErr error) error

CommitTx commits tx when inErr is nil, and rolls back when inErr is non-nil. Both errors are preserved in the returned chain so callers can use errors.Is.

func (*Component) Exec

func (c *Component) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)

Exec executes sql without scanning rows. Useful for INSERT, UPDATE, DELETE, DDL.

func (*Component) Get

func (c *Component) Get(ctx context.Context, dst any, sql string, args ...any) error

Get executes sql and scans the first result row into dst. dst must be a pointer to a struct or scalar type.

func (*Component) Health

func (c *Component) Health(ctx context.Context) error

Health implements samsara.HealthChecker. Returns a non-nil error if the pool cannot reach the database.

func (*Component) Name

func (c *Component) Name() string

Name implements samsara.Component.

func (*Component) Select

func (c *Component) Select(ctx context.Context, dst any, sql string, args ...any) error

Select executes sql and scans all result rows into dst. dst must be a pointer to a slice of structs or scalars.

func (*Component) Start

func (c *Component) Start(ctx context.Context, ready func()) error

Start creates the connection pool, pings the server to confirm reachability, calls ready() to unblock the supervisor, then blocks until Stop or ctx cancellation.

Start is safe to call multiple times across restarts; each call allocates a fresh stopCh so the previous Stop signal does not bleed into the new run.

func (*Component) Stop

func (c *Component) Stop(ctx context.Context) error

Stop signals Start to return and closes the connection pool. It is idempotent and concurrency-safe: multiple concurrent calls are safe, and calling Stop before Start has been called is safe.

type Config

type Config struct {
	// Host is the database server hostname or IP. Defaults to "localhost".
	Host string
	// Port is the database server port. Defaults to 5432.
	Port int
	// Name is the database name. Defaults to "postgres".
	Name string
	// User is the database user. Defaults to "postgres".
	User string
	// Pass is the database password. Special characters are safely encoded.
	Pass string

	// SSLMode controls the SSL negotiation mode (disable, require, verify-ca,
	// verify-full). Defaults to "disable".
	SSLMode string

	// URI overrides all individual fields when non-empty.
	// Must be a valid libpq connection string or pgx DSN.
	URI string

	// ConnectTimeout is the deadline for the initial connection + ping during
	// Start. Defaults to 10 s.
	ConnectTimeout time.Duration

	// MaxConns caps the pool size. 0 means pgx default (min(4, GOMAXPROCS)).
	MaxConns int32
	// MinConns keeps this many connections alive even when idle. 0 means none.
	MinConns int32
}

Config holds all connection parameters for the PostgreSQL component. Supply individual fields or a preformatted [Config.URI] to override them all.

type DB

type DB interface {
	// Select executes sql and scans all result rows into dst (a pointer to a
	// slice). Returns [ErrNoRows] if the result set is empty.
	Select(ctx context.Context, dst any, sql string, args ...any) error

	// Get executes sql and scans the first result row into dst (a pointer to a
	// struct or scalar). Returns [ErrNoRows] if no row was found.
	Get(ctx context.Context, dst any, sql string, args ...any) error

	// Exec executes sql and returns the command tag. Use for INSERT/UPDATE/DELETE
	// where you don't need to scan rows.
	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)

	// BeginTx starts a transaction with the given options.
	BeginTx(ctx context.Context, opts pgx.TxOptions) (pgx.Tx, error)

	// CommitTx commits tx if inErr is nil, and rolls back if inErr is non-nil.
	// Use in a defer to guarantee finalisation:
	//
	//	tx, err := db.BeginTx(ctx, pgx.TxOptions{})
	//	if err != nil { return err }
	//	defer func() { err = db.CommitTx(ctx, tx, err) }()
	CommitTx(ctx context.Context, tx TxFinaliser, inErr error) error
}

DB is the interface that domain adapters should depend on. *Component satisfies it; depend on DB rather than *Component to keep adapters testable.

type UserRepo struct { db postgresql.DB }

type Logger

type Logger interface {
	Info(msg string, args ...any)
	Error(msg string, args ...any)
}

Logger is satisfied by log/slog.Logger and most structured loggers.

type Option

type Option func(*Component)

Option configures a Component.

func WithLogger

func WithLogger(l Logger) Option

WithLogger attaches a structured logger to the component. log/slog.Logger satisfies Logger directly.

func WithName

func WithName(name string) Option

WithName overrides the component name returned by Component.Name. Useful when registering multiple PostgreSQL components with the same supervisor (e.g. a primary and a read replica).

type TxFinaliser

type TxFinaliser interface {
	Commit(ctx context.Context) error
	Rollback(ctx context.Context) error
}

TxFinaliser is the minimal transaction interface required by [CommitTx]. pgx.Tx satisfies it. Define a local stub in tests to avoid a real database.