pgx

dbx/pg is the kit's default database stack, built on pgx, the most widely used PostgreSQL driver in Go. It provides three things: pool construction from a dbx.Config, a transaction-aware query executor (DB), and a dbx.Transactor implementation that carries a pgx.Tx in the context.

import (
    "github.com/gp-system/gpsystem/dbx"
    "github.com/gp-system/gpsystem/dbx/pg"
)

There is no wrapper API: the package hands back pgxpool.Pool, pgx.Rows and pgconn.CommandTag. Whatever the pgx documentation says holds here too. The kit only centralizes pool construction, instrumentation and the transaction plumbing.

Pool

pool, err := pg.NewPool(ctx, cfg.DB)  // DSN from config, connect, ping
pool := pg.MustNewPool(ctx, cfg.DB)   // same, panics on error, for main()

NewPool parses the pool config from cfg.DB.DSN(), sets MaxConns (DB_MAX_CONNS, default 10), connects, and verifies the connection with a ping, so a wrong password or an unreachable database fails at startup, not on the first request. MustNewPool is the same with a panic; the generated main.go files call it.

The generated entry point ties the pool's shutdown into the lifecycle:

err := fiberx.Run(ctx, cfg.Server, register,
    fiberx.WithCloser("pgxpool", func(context.Context) error {
        pool.Close()
        return nil
    }))

The pool is instrumented with otelpgx: every query emits an OTel span carrying the statement and its timing, nested under the active request/task span. With a no-op tracer provider (dev mode, telemetry off) this is effectively free; with telemetry on the queries show up in the trace waterfall and on Sentry error events.

DBTX: the query surface repositories depend on

type DBTX 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
}

Every repository depends on these three methods. The return types are deliberately raw pgx types (pgconn.CommandTag, pgx.Rows, pgx.Row), not kit wrappers: scanning, CollectRows, RowsAffected() work exactly as the pgx documentation describes.

DBTX is also method-for-method identical to the DBTX interface sqlc generates. The kit's generators scaffold hand-written repositories, but if a project generated its query code with sqlc, those structs accept pg.NewDB(pool) unchanged, and participate in the same context-carried transactions.

DB: the executor that watches the context

db := pg.NewDB(pool) // implements DBTX

pg.DB is the kit's single implementation of DBTX, and it has exactly one trick: every call first looks for a transaction in the context. If WithinTransaction opened one, the call runs on it; if not, on the pool. The repository never knows, and never needs to know, which case it is.

This is where the kit's transaction model comes together: the service calls WithinTransaction, and the repositories join automatically, with unchanged signatures.

A shop repository

The shop sample app's module-level product repository (internal/modules/shop/repository/, shared by every surface). The repository holds a single pg.DBTX field. main.go injects pg.NewDB(pool) through Dependencies:

package repository

type ProductRepository struct {
    db pg.DBTX
}

func NewProductRepository(db pg.DBTX) *ProductRepository {
    return &ProductRepository{db: db}
}

Listing products with cursor pagination (a keyset seek, not OFFSET):

const productSort = "created_at DESC, id DESC" // matches the ORDER BY below

func (r *ProductRepository) ListProducts(
    ctx context.Context, p paginate.CursorParams,
) (paginate.CursorPage[Product], error) {
    args := []any{p.FetchLimit()}
    query := `SELECT id, name, price_cents, stock, created_at FROM products`
    if p.HasCursor() {
        var createdAt time.Time
        var id int64
        if err := p.DecodeKey(&createdAt, &id); err != nil {
            return paginate.CursorPage[Product]{}, err
        }
        query += ` WHERE (created_at, id) < ($2, $3)`
        args = append(args, createdAt, id)
    }
    query += ` ORDER BY ` + productSort + ` LIMIT $1`

    rows, err := r.db.Query(ctx, query, args...)
    if err != nil {
        return paginate.CursorPage[Product]{}, errs.Wrap(err, "products: list")
    }
    products, err := pgx.CollectRows(rows, pgx.RowToStructByName[Product])
    if err != nil {
        return paginate.CursorPage[Product]{}, errs.Wrap(err, "products: scan")
    }
    return paginate.NewCursorPage(products, p, func(last Product) []any {
        return []any{last.CreatedAt, last.ID}
    })
}

And the stock decrement the service calls from the PlaceOrder transaction:

func (r *ProductRepository) DecrementStock(ctx context.Context, productID int64, qty int) error {
    tag, err := r.db.Exec(ctx,
        `UPDATE products SET stock = stock - $2 WHERE id = $1 AND stock >= $2`,
        productID, qty)
    if err != nil {
        return errs.Wrap(err, "products: decrement stock")
    }
    if tag.RowsAffected() == 0 {
        return errs.New("insufficient stock",
            errs.Code("insufficient_stock"), errs.Public("The product is out of stock."))
    }
    return nil
}

Notice what is not in the code: a transaction parameter. ListProducts runs on the pool when called from the public listing endpoint; DecrementStock, called inside the PlaceOrder transaction, joins the same pgx.Tx, both with the same signature. pgx.CollectRows + RowToStructByName is pgx's built-in scanning; the WHERE ... AND stock >= $2 + RowsAffected() pair is the atomic stock check, with no SELECT-then-UPDATE race.

Transactor

tx := pg.NewTransactor(pool) // dbx.Transactor

err := tx.WithinTransaction(ctx, func(ctx context.Context) error {
    if err := s.orders.Insert(ctx, o); err != nil {
        return err
    }
    return s.products.DecrementStock(ctx, o.ProductID, o.Qty) // same tx
})

WithinTransaction opens a transaction on the pool, injects the pgx.Tx into the context, and commits after fn returns. Any error, or panic, which is re-raised after rollback, rolls everything back. When the context already carries a transaction, fn joins it instead of opening a new one: nesting is flat, there are no savepoints, and the outermost call always owns commit/rollback. The full treatment of the pattern, with the shop's PlaceOrder example, is on Transactions.

Escape hatch: the context plumbing

For a custom executor (say, a batch writer or a hand-instrumented query path), the transaction carrying is exported:

ctx = pg.ContextWithTx(ctx, tx)   // put a transaction into the context
tx, ok := pg.TxFromContext(ctx)   // take the transaction out of the context

Application code normally never calls these: pg.DB and the Transactor cover it together. They matter when you need a richer pgx surface than DBTX (e.g. CopyFrom, SendBatch) inside a transaction.

Patterns used

• Narrow, sqlc-compatible repository interface (DBTX): Design patterns.
• Unit of Work / transaction in context (WithinTransaction, txKey{}): Design patterns.