sql

package
v0.12.0 Latest Latest
Warning

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

Go to latest
Published: Jul 29, 2026 License: MIT Imports: 11 Imported by: 0

Documentation

Overview

Package sql provides protocol-agnostic SQL adapter bindings for the ports package.

All adapters implement the ports.SourceAdapter, ports.SinkAdapter, and ports.IOAdapter interfaces and are wired to pipelines via ports.SourcePort.Bind, ports.SinkPort.Bind, and ports.IOPort.Bind.

Sources (use with ports.SourcePort):

  • QueryAdapter — polls a SQL query at interval, emitting each validated row

Intermediate (use with ports.IOPort):

Sinks (use with ports.SinkPort):

Package sql brings go-codex's codec-based validation to SQL databases by combining three single-purpose tools — each doing exactly what it is designed for:

  • github.com/pressly/goose/v3 manages schema migrations (versioned .sql files, migration history table). Wrapped by Migrator.
  • [sqlc](https://sqlc.dev) generates typed Go structs and query methods from the migrated schema and hand-written SQL query files. This is a developer-time step; no runtime dependency.
  • go-codex Codec[T] + Refine applies business-rule validation on top of the generated structs — rules SQL itself cannot express, centralized and testable in pure Go. Executed by Validate.

Toolchain split

goose owns schema shape (column existence, types, indexes). sqlc catches structural query mistakes at compile time. go-codex validates business rules the application cares about. None of the three tools duplicates the others' work.

Validate

Validate runs a struct through its codex.Codec's encode→decode round trip, applying every Refine and RefineFunc constraint. Use it in two modes:

  • Pre-query validation: reject invalid data before it is written to the database. Invalid structs never reach queries.InsertUser (or equivalent).
  • Post-query validation: validate a row returned by a sqlc query method as defence in depth against data written by other clients that bypassed the codec.

The returned T is the normalized value — the same round-trip semantics used by [format.JSON.Read] throughout the library.

params := db.InsertUserParams{ID: uuid.NewString(), Name: name, Email: email}
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 reaches the DB
}
err = queries.InsertUser(ctx, validated)

Migrator

Migrator wraps goose's migration runner and emits structured observer events per applied or rolled-back migration file. Construct it with NewMigrator, then call Migrator.Up at startup:

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

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

Structured errors

All error types implement slog.LogValuer for zero-effort structured logging and [Unwrap] for errors.As traversal:

Observer integration

Pass any stats.Observer to ValidateOptions.Observer and MigrateOptions.Observer. If the implementation also satisfies stats.SQLObserver, the adapter calls:

Per-field validation failures are always reported via stats.Observer.RecordValidationError with location "sql_row", regardless of whether stats.SQLObserver is implemented.

Index

Examples

Constants

This section is empty.

Variables

This section is empty.

Functions

func DecorateInput added in v0.12.0

func DecorateInput[Req any](
	fn func(context.Context, Req) error,
	codec codex.Codec[Req],
	opts ValidateOptions,
) func(context.Context, Req) error

DecorateInput wraps an exec-style sqlc-generated function (a `:exec` query, shaped `func(context.Context, Req) error`) with pre-query validation: arg is validated via codec BEFORE fn is called — fn is never invoked on invalid input. Returns a function with the identical signature, callable in place of the sqlc method everywhere:

insertUser := sqladapter.DecorateInput(queries.InsertUser, insertParamsCodec,
    sqladapter.ValidateOptions{Table: "users", Op: "insert_user"})
err := insertUser(ctx, params) // codec-validated automatically

This is the declare-once counterpart to calling Validate by hand before every sqlc call (sql.md's "pre-query validation" usage mode) — it bundles the codec, Table, and Op into one reusable value the same way ports.Cache/ports.NewCache bundle a cache's key/format/TTL, so callers don't repeat them at every call site.

Unlike bare Validate (which has no ctx parameter), the returned closure resolves stats.ObserverFromContext when opts.Observer is nil, since it wraps a ctx-taking function.

Returns RowValidationError without calling fn when validation fails.

Example

ExampleDecorateInput shows wrapping an exec-style sqlc function once, getting back a drop-in replacement that validates before every call.

// Stand-in for an sqlc-generated ":exec" method.
insertUserSQL := func(ctx context.Context, u testUser) error {
	fmt.Println("INSERT user:", u.Name)
	return nil
}

insertUser := sqladapter.DecorateInput(insertUserSQL, userCodec,
	sqladapter.ValidateOptions{Table: "users", Op: "insert_user"})

// Call in place of the sqlc method everywhere — validated automatically.
valid := testUser{ID: "f47ac10b-58cc-4372-a567-0e02b2c3d479", Name: "Ada", Email: "ada@example.com", Role: "user"}
if err := insertUser(context.Background(), valid); err != nil {
	fmt.Println("unexpected error:", err)
}

// Invalid input never reaches the sqlc method.
invalid := testUser{ID: "not-a-uuid", Name: "Ada", Email: "ada@example.com", Role: "user"}
if err := insertUser(context.Background(), invalid); err != nil {
	fmt.Println("rejected before insert:", err != nil)
}
Output:
INSERT user: Ada
rejected before insert: true

func DecorateOutput added in v0.12.0

func DecorateOutput[Req, Resp any](
	fn func(context.Context, Req) (Resp, error),
	codec codex.Codec[Resp],
	opts ValidateOptions,
) func(context.Context, Req) (Resp, error)

DecorateOutput wraps a query-style sqlc-generated function (a `:one`/ `:many` query, shaped `func(context.Context, Req) (Resp, error)`) with post-query validation: fn's returned Resp is validated via codec AFTER fn returns — defense in depth against rows written by other clients that bypassed the codec. Returns a function with the identical signature, callable in place of the sqlc method everywhere:

getUser := sqladapter.DecorateOutput(queries.GetUser, userCodec,
    sqladapter.ValidateOptions{Table: "users", Op: "get_user"})
u, err := getUser(ctx, id) // codec-validated automatically

This is the declare-once counterpart to calling Validate by hand after every sqlc call (sql.md's "post-query validation" usage mode).

Unlike bare Validate (which has no ctx parameter), the returned closure resolves stats.ObserverFromContext when opts.Observer is nil, since it wraps a ctx-taking function.

fn's own error (e.g. sql.ErrNoRows) is returned unchanged — validation only runs on a successful call. Returns RowValidationError when fn succeeds but the returned value fails validation.

Example

ExampleDecorateOutput shows wrapping a query-style sqlc function once, getting back a drop-in replacement that validates every returned row.

// Stand-in for an sqlc-generated ":one" method.
getUserSQL := func(ctx context.Context, id string) (testUser, error) {
	return testUser{ID: id, Name: "Ada", Email: "ada@example.com", Role: "user"}, nil
}

getUser := sqladapter.DecorateOutput(getUserSQL, userCodec,
	sqladapter.ValidateOptions{Table: "users", Op: "get_user"})

u, err := getUser(context.Background(), "f47ac10b-58cc-4372-a567-0e02b2c3d479")
if err != nil {
	fmt.Println("unexpected error:", err)
	return
}
fmt.Println("validated:", u.Name)
Output:
validated: Ada

func DrainInsertAdapter

func DrainInsertAdapter[T any](
	codec codex.Codec[T],
	insertFn func(context.Context, T) error,
	opts DrainInsertOptions,
) ports.SinkAdapter[T]

DrainInsertAdapter returns a ports.SinkAdapter that validates and inserts each item via insertFn. Use with ports.SinkPort.Bind:

domain.Readings.Bind(ctx, sql.DrainInsertAdapter(readingCodec,
    func(ctx context.Context, r Reading) error { return db.Insert(ctx, r) },
    sql.DrainInsertOptions{}))

func QueryAdapter

func QueryAdapter[T any](
	codec codex.Codec[T],
	queryFn func(context.Context) ([]T, error),
	interval time.Duration,
	opts QueryStreamOptions,
) ports.SourceAdapter[T]

QueryAdapter returns a ports.SourceAdapter that polls a SQL query at interval, emitting each validated row. Use with ports.SourcePort.Bind:

domain.Configs.Bind(ctx, sql.QueryAdapter(configCodec,
    func(ctx context.Context) ([]Config, error) { return db.ListConfigs(ctx) },
    5*time.Minute, sql.QueryStreamOptions{}))

func QueryEachAdapter

func QueryEachAdapter[In, T any](
	codec codex.Codec[T],
	queryFn func(context.Context, In) ([]T, error),
	opts QueryEachStreamOptions,
) ports.IOAdapter[In, T]

QueryEachAdapter returns a ports.IOAdapter that performs a parameterized SQL query for each In item, emitting each result row as a T item (1:N). Use with ports.IOPort.Bind:

domain.Calibration.Bind(ctx, sql.QueryEachAdapter(thresholdCodec,
    func(ctx context.Context, s SensorReading) ([]Threshold, error) {
        return db.GetThresholdBySensor(ctx, s.SensorID)
    }, sql.QueryEachStreamOptions{Table: "thresholds", Op: "get_by_sensor"}))

func Validate

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

Validate runs v through c's encode→decode round trip, applying every Refine and RefineFunc constraint declared on c.

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

Use Validate to validate a struct returned by a sqlc query method (post-query defence in depth against rows written by other clients) or a struct about to be passed into a sqlc insert or update method (pre-query, so invalid data never reaches the database).

Codec failures are wrapped in RowValidationError. Per-field constraint failures are additionally reported via stats.Observer.RecordValidationError with location "sql_row". If opts.Observer also implements stats.SQLObserver, RecordValidation is called after every Validate call, success or failure.

Example
package main

import (
	"fmt"

	sqladapter "github.com/DaniDeer/go-codex/adapters/sql"
	"github.com/DaniDeer/go-codex/codex"
	"github.com/DaniDeer/go-codex/validate"
)

func main() {
	type Item struct {
		ID   string
		Name string
	}
	itemCodec := codex.Struct(
		codex.RequiredField("id",
			codex.String(),
			func(i Item) string { return i.ID },
			func(i *Item, v string) { i.ID = v }),
		codex.RequiredField("name",
			codex.String().Refine(validate.NonEmptyString),
			func(i Item) string { return i.Name },
			func(i *Item, v string) { i.Name = v }),
	)

	// Pre-query: validate before writing to the database.
	params := Item{ID: "1", Name: "Widget"}
	validated, err := sqladapter.Validate(itemCodec, params, sqladapter.ValidateOptions{
		Table: "items", Op: "insert_item",
	})
	if err != nil {
		fmt.Println("invalid:", err)
		return
	}
	fmt.Println("validated:", validated.Name)
}
Output:
validated: Widget

Types

type DrainInsertOptions

type DrainInsertOptions struct {
	// Table and Op provide error/observability context. When empty, they
	// default from the bound port's [ports.SQLPattern] declaration.
	Table    string
	Op       string
	OnError  func(error)
	Observer stats.Observer
}

DrainInsertOptions configures DrainInsertAdapter.

type InsertStreamError

type InsertStreamError struct {
	// Table names the table being written. Matches [DrainInsertOptions.Table].
	Table string
	// Op names the insert operation. Matches [DrainInsertOptions.Op].
	Op string
	// Err is the underlying database or application error.
	Err error
}

InsertStreamError is passed to DrainInsertOptions.OnError by [DrainInsert] when the user's insertFn returns a database or application-level error after successful codec validation. Distinct from RowValidationError.

func (InsertStreamError) Error

func (e InsertStreamError) Error() string

func (InsertStreamError) LogValue

func (e InsertStreamError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (InsertStreamError) Unwrap

func (e InsertStreamError) Unwrap() error

Unwrap allows errors.Is and errors.As to traverse the underlying error.

type MigrateOptions

type MigrateOptions struct {
	// Observer, when non-nil and implementing [stats.SQLObserver], receives
	// [stats.SQLObserver.RecordMigration] once per applied or rolled-back
	// migration file. Defaults to [stats.NoopObserver] when nil.
	Observer stats.Observer
}

MigrateOptions configures a single Migrator.Up or Migrator.Down call.

type MigrationError

type MigrationError struct {
	// Op is the migration operation: "up", "down", or "status".
	Op string

	// Version is the migration version number for Up/Down, or 0 for Status.
	Version int64

	// Err is the underlying goose error.
	Err error
}

MigrationError is returned by Migrator.Up, Migrator.Down, and Migrator.Status when goose fails. It wraps the original goose error and carries the operation name and version (when applicable) for structured logging and error handling.

MigrationError implements slog.LogValuer:

slog.Error("migration failed", "error", me)
// → {op:"up", version:3, err:"..."}

func (MigrationError) Error

func (e MigrationError) Error() string

func (MigrationError) LogValue

func (e MigrationError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (MigrationError) Unwrap

func (e MigrationError) Unwrap() error

Unwrap allows errors.Is and errors.As to traverse the wrapped goose error.

type MigrationStatus

type MigrationStatus struct {
	// Version is the numeric prefix of the migration file (e.g. 1 for
	// "00001_create_users.sql").
	Version int64

	// Name is the migration file path.
	Name string

	// AppliedAt is when this migration was applied. The zero value means the
	// migration is pending.
	AppliedAt time.Time
}

MigrationStatus describes the applied or pending state of one migration file.

type Migrator

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

Migrator wraps pressly/goose with go-codex structured errors and observer integration. Construct one with NewMigrator.

Migrator never inspects or validates row data — it operates only on schema DDL text. Schema evolution and codec-level row validation are separate concerns by design.

func NewMigrator

func NewMigrator(db *sql.DB, migrations fs.FS, dir string, dialect string) (*Migrator, error)

NewMigrator constructs a Migrator for the given database connection and embedded migrations filesystem.

  • db is the open database connection. The caller owns its lifecycle.
  • migrations is the fs.FS containing migration files, typically an embed.FS declared with //go:embed.
  • dir is the sub-path within migrations where the .sql files live (e.g. "migrations").
  • dialect is one of goose's supported dialect constants as a string: "postgres", "mysql", "sqlite3", "mssql", "redshift", "tidb", "clickhouse", "vertica", "ydb", "spanner", or "turso".
Example
package main

import (
	"context"
	"database/sql"
	"embed"
	"fmt"

	sqladapter "github.com/DaniDeer/go-codex/adapters/sql"
	_ "modernc.org/sqlite"
)

//go:embed testdata/migrations/*.sql
var testMigrationsFS embed.FS

func main() {
	db, err := sql.Open("sqlite", "file::memory:?cache=shared&_example=1")
	if err != nil {
		fmt.Println("open:", err)
		return
	}
	defer db.Close()

	migrator, err := sqladapter.NewMigrator(db, testMigrationsFS, "testdata/migrations", "sqlite3")
	if err != nil {
		fmt.Println("migrator:", err)
		return
	}

	if err := migrator.Up(context.Background(), sqladapter.MigrateOptions{}); err != nil {
		fmt.Println("up:", err)
		return
	}
	fmt.Println("migrations applied")
}
Output:
migrations applied

func (*Migrator) Down

func (m *Migrator) Down(ctx context.Context, opts MigrateOptions) error

Down rolls back the most recently applied migration. Triggers stats.SQLObserver.RecordMigration on opts.Observer when implemented.

Returns a MigrationError with Op "down" on failure.

func (*Migrator) Status

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

Status returns the applied or pending state of every migration file. Returns a MigrationError with Op "status" on failure.

func (*Migrator) Up

func (m *Migrator) Up(ctx context.Context, opts MigrateOptions) error

Up applies all pending migrations. Each successfully applied migration file triggers stats.SQLObserver.RecordMigration on opts.Observer (when the observer implements that interface).

Returns a MigrationError with Op "up" on failure.

type QueryEachStreamOptions

type QueryEachStreamOptions struct {
	// Table and Op provide error/observability context. When empty, they
	// default from the bound port's [ports.SQLPattern] declaration.
	Table    string
	Op       string
	Observer stats.Observer
	Buffer   int
}

QueryEachStreamOptions configures QueryEachAdapter.

type QueryStreamError

type QueryStreamError struct {
	// Table names the table being queried. Matches [QueryStreamOptions.Table].
	Table string
	// Op names the query operation. Matches [QueryStreamOptions.Op].
	Op string
	// Err is the underlying database or application error.
	Err error
}

QueryStreamError is sent to [Stream.Errors] by [QueryStream] when the user's queryFn returns a database or application-level error. Distinct from RowValidationError (codec validation failure).

func (QueryStreamError) Error

func (e QueryStreamError) Error() string

func (QueryStreamError) LogValue

func (e QueryStreamError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (QueryStreamError) Unwrap

func (e QueryStreamError) Unwrap() error

Unwrap allows errors.Is and errors.As to traverse the underlying error.

type QueryStreamOptions

type QueryStreamOptions struct {
	// Table names the table being queried. Used in [QueryStreamError] context.
	// When empty, defaults from the bound port's [ports.SQLPattern] declaration.
	Table string
	// Op names the query operation. Used in [QueryStreamError] context.
	// When empty, defaults from the bound port's [ports.SQLPattern] declaration.
	Op string
	// Observer receives per-row lifecycle events.
	Observer stats.Observer
	// Buffer is the output stream channel buffer size. Default 0.
	Buffer int
}

QueryStreamOptions configures QueryAdapter.

type RowValidationError

type RowValidationError struct {
	// Table names the sqlc-generated table, matching ValidateOptions.Table.
	Table string

	// Op names the sqlc operation, matching ValidateOptions.Op.
	Op string

	// Err is the underlying codec error. Use errors.As to reach
	// *codex.ValidationErrors for per-field detail.
	Err error
}

RowValidationError is returned by Validate when the codec's Refine or RefineFunc constraints reject the value. It wraps the underlying codec error (typically codex.ValidationErrors) so callers can use errors.As to inspect individual field failures.

RowValidationError implements slog.LogValuer for zero-effort structured logging:

slog.Error("row invalid", "error", rve)
// → {table:"users", op:"insert_user", err:{email:"invalid email format"}}

func (RowValidationError) Error

func (e RowValidationError) Error() string

func (RowValidationError) LogValue

func (e RowValidationError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (RowValidationError) Unwrap

func (e RowValidationError) Unwrap() error

Unwrap allows errors.Is and errors.As to traverse the wrapped codec error.

type ValidateOptions

type ValidateOptions struct {
	// Table names the sqlc-generated table or struct for error and observer
	// context (e.g. "users"). Purely descriptive — Validate does not touch
	// the database itself.
	Table string

	// Op names the sqlc operation being wrapped, e.g. "get_user" or
	// "insert_user". Matches sqlc's query name for easy correlation between
	// generated code and validation logs or metrics.
	Op string

	// Observer, when non-nil, receives per-validation lifecycle events.
	// If it also implements [stats.SQLObserver], RecordValidation is called
	// after every Validate call. Per-field failures are always reported via
	// [stats.Observer.RecordValidationError] with location "sql_row",
	// regardless of whether SQLObserver is implemented.
	// Defaults to [stats.NoopObserver] when nil.
	Observer stats.Observer
}

ValidateOptions configures observer and error context for a single Validate call.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL