SQL Adapter — adapters/sql¶

See also: adapters/sql on pkg.go.dev · SQL Examples guide · Observer Pattern · Error Handling

Runnable demo: examples/sensor-service — sensor readings service combining adapters/nethttp, adapters/mqtt, and adapters/sql with a shared stats.NewFanout observer; shows sqlc-generated structs, goose migrations, Validate[T] pre-insert and post-read, and field factory functions reusing Refine rules across db.Reading and db.InsertReadingParams.

adapters/sql brings go-codex's codec-based validation to SQL databases by pairing three single-purpose tools, each doing exactly the job it is designed for:

The adapter adds no query builder, no ORM, no row scanner. Its sole job is the T → validate → T boundary: ensuring a struct passes all its codec constraints before it reaches the database or after it comes back from one.

Validate¶

Validate[T] is the core function. It runs a value through its codec's encode→decode round trip, applying every Refine and RefineFunc constraint.

func Validate[T any](c codex.Codec[T], v T, opts ValidateOptions) (T, error)

The returned T is the normalized value — the result of Decode after Encode. This may differ from v when a Refine step normalizes data (e.g. trimming whitespace). This is the same round-trip semantics used by format.JSON.Read throughout go-codex.

Two usage modes¶

Pre-query validation — reject invalid data before it reaches the database:

validated, err := sqladapter.Validate(insertParamsCodec, params,
    sqladapter.ValidateOptions{Table: "users", Op: "insert_user", Observer: obs})
if err != nil {
    return fmt.Errorf("invalid input: %w", err) // never calls queries.InsertUser
}
_ = queries.InsertUser(ctx, validated)

Post-query validation — defence in depth against rows written by other clients that bypassed the codec:

u, err := queries.GetUser(ctx, id)
if err != nil { return nil, err } // sql.ErrNoRows etc — not go-codex's concern

u, err = sqladapter.Validate(userCodec, u,
    sqladapter.ValidateOptions{Table: "users", Op: "get_user", Observer: obs})

ValidateOptions¶

type ValidateOptions struct {
    Table    string         // sqlc table name — for error context and observer
    Op       string         // sqlc query name — e.g. "get_user", "insert_user"
    Observer stats.Observer // nil → stats.NoopObserver
}

Declare once — DecorateInput/DecorateOutput¶

Calling Validate by hand around every sqlc call (above) means repeating Table/Op — and remembering to call it at all — at every call site. DecorateInput/DecorateOutput wrap an sqlc-generated function once and return a drop-in replacement with the identical signature, callable in place of the sqlc method everywhere:

// Pre-query validation, declared once — mirrors the manual pattern above.
insertUser := sqladapter.DecorateInput(queries.InsertUser, insertParamsCodec,
    sqladapter.ValidateOptions{Table: "users", Op: "insert_user"})

err := insertUser(ctx, params) // validated automatically; sqlc never called on invalid input

// Post-query validation, declared once.
getUser := sqladapter.DecorateOutput(queries.GetUser, userCodec,
    sqladapter.ValidateOptions{Table: "users", Op: "get_user"})

u, err := getUser(ctx, id) // validated automatically before the caller sees it

Both reuse ValidateOptions (no new option type) and call Validate internally (no duplicated logic). Unlike bare Validate — which has no ctx parameter — the returned closures DO have ctx in scope (they wrap func(ctx, ...)), so they resolve stats.ObserverFromContext(ctx) when Observer is left nil. DecorateOutput passes fn's own error (e.g. sql.ErrNoRows) through unchanged — validation only runs on success.

This is the SQL equivalent of declaring a ports.Cache/ports.File once and reusing it — see Design pattern: declarative descriptor + plain function for the full picture across file/sql/rest/events/cache. SQL has no templated key/path to attach a per-var codec to (sqlc's typed parameters already fill that role, validated the same way as the row itself) — the decorator's reusable value is the codec + Table/Op bundle, not a key template.

Migrator¶

Migrator wraps pressly/goose/v3 with go-codex structured errors and observer hooks. It never inspects row data — schema evolution and row validation are separate concerns by design.

// NewMigrator validates the dialect and migrations directory at construction time.
func NewMigrator(db *sql.DB, migrations fs.FS, dir string, dialect string) (*Migrator, error)

func (m *Migrator) Up(ctx context.Context, opts MigrateOptions) error
func (m *Migrator) Down(ctx context.Context, opts MigrateOptions) error
func (m *Migrator) Status(ctx context.Context) ([]MigrationStatus, error)

Supported dialect strings: "postgres", "mysql", "sqlite3", "mssql", "redshift", "tidb", "clickhouse", "vertica", "ydb", "spanner", "turso".

Embedding migrations¶

//go:embed migrations/*.sql
var migrationsFS embed.FS

migrator, err := sqladapter.NewMigrator(db, migrationsFS, "migrations", "postgres")
if err != nil { log.Fatal(err) }
if err := migrator.Up(ctx, sqladapter.MigrateOptions{Observer: obs}); err != nil {
    log.Fatal(err)
}

MigrationStatus¶

type MigrationStatus struct {
    Version   int64      // numeric prefix of the migration file
    Name      string     // file path
    AppliedAt time.Time  // zero value = pending
}

MigrateOptions¶

type MigrateOptions struct {
    Observer stats.Observer // nil → stats.NoopObserver
}

Error types¶

All error types implement error, Unwrap(), and slog.LogValuer.

RowValidationError.Err is typically codex.ValidationErrors — use errors.As to reach per-field detail:

var rve sqladapter.RowValidationError
if errors.As(err, &rve) {
    slog.Error("row invalid", "error", rve) // structured: {table, op, err}

    var ve codex.ValidationErrors
    if errors.As(err, &ve) {
        for _, fe := range ve {
            fmt.Printf("field %q: %v\n", fe.Field, fe.Err)
        }
    }
}

sql.ErrNoRows and driver errors from sqlc query methods are not wrapped by go-codex — they pass through unchanged so errors.Is(err, sql.ErrNoRows) continues to work.

Error-path ergonomics — handle/log by default¶

SQL is an internal store boundary with no caller to respond to. DrainInsertAdapter's OnError func(error) IS the handle action (fully owns the error); leaving it nil is the log default (observer-only). For a respond-equivalent — publishing a typed error payload — compose OnError with a declared events.ErrorChannel:

sql.DrainInsertAdapter(db, "orders", format.JSON(orderCodec), sql.DrainInsertOptions{
    OnError: func(err error) {
        if resp, matched, mapErr := errHandle.ErrorResponseFor(err); matched && mapErr == nil &&
            resp.Action == events.ErrorRespond {
            _ = mqttClient.Publish(ctx, &paho.Publish{Topic: resp.Topic, Payload: resp.Body})
            return
        }
        slog.Warn("insert failed", "error", err)
    },
})

See Error handling guide — store/IO boundaries for the full pattern and rationale.

Observer integration — stats.SQLObserver¶

stats.SQLObserver is an optional extension to stats.Observer. The adapter type-asserts the configured observer to SQLObserver before calling SQL-specific hooks — existing observer implementations need not change.

type SQLObserver interface {
    // Called after every Validate call — err is nil on success.
    RecordValidation(table, op string, dur time.Duration, err error)

    // Called once per applied or rolled-back migration file during Up/Down.
    RecordMigration(op, name string, version int64, dur time.Duration, err error)
}

Per-field validation failures are always reported via stats.Observer.RecordValidationError("sql_row", constraint, field), regardless of whether SQLObserver is implemented — consistent with "file" (file formats), "payload" (MQTT), and "body" (HTTP).

stats.NoopObserver, stats.LoggingObserver, and stats.NewFanout all implement SQLObserver.

Implementing SQLObserver¶

type myMetrics struct {
    // embed NoopObserver so the base Observer interface is satisfied
    stats.NoopObserver
}

func (m *myMetrics) RecordValidation(table, op string, dur time.Duration, err error) {
    status := "ok"
    if err != nil { status = "error" }
    sqlValidations.With("table", table, "op", op, "status", status).
        Observe(dur.Seconds())
}

func (m *myMetrics) RecordMigration(op, name string, version int64, dur time.Duration, err error) {
    slog.Info("migration", "op", op, "name", name, "version", version,
        "ms", dur.Milliseconds(), "err", err)
}

Sharing a Codec[T] across SQL and REST/MQTT¶

A codec declared against a sqlc-generated struct is an ordinary Codec[T] and can be reused as the response codec for a REST route or MQTT channel — one source of truth for what a valid value looks like, regardless of transport:

// REST response uses the same codec:
userRoute := rest.NewRoute[GetUserReq, db.User](
    rest.WithResponseCodec(userCodec), ...)

// SQL validation uses the same codec:
u, err = sqladapter.Validate(userCodec, u, sqladapter.ValidateOptions{...})

Updating a Refine rule in userCodec propagates automatically to both the HTTP response validation and the SQL row validation.

What adapters/sql does NOT do¶

• No query builder — callers write SQL text directly in sqlc query files.
• No row scanning — sqlc's generated methods already handle this per-driver.
• No ORM — Validate is the only boundary between go-codex and the DB row.
• No spec generation — SQL tables are not API surfaces; the same Codec[T] contributes its schema when registered with a REST route or MQTT channel.
• No reflection — codecs are declared by hand with explicit get/set closures, consistent with the no-reflection design principle throughout go-codex.