sqlcommon

package
v0.9.1 Latest Latest
Warning

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

 Go to latest Published: Sep 14, 2026 License: MIT Imports: 19 Imported by: 0

Documentation

Overview

Package sqlcommon implements storage.Store over database/sql using a caller-selected dialect.

Design

Shared queries and transaction shapes cover enrollment, command queues, certificate history, push certificates, user-authentication state and migration. Dialects supply fixed SQL syntax and migrations; data values use parameters. Selected secret columns are sealed through storage/crypt, and Rewrap rotates them in guarded batches.

The driver packages open connections. Satellite domain stores reuse migration/dialect support while owning their own migration sets. Operations across different domain stores are not a distributed transaction.

References

Index

Constants

View Source 
const ClearBatchSize = 5000

ClearBatchSize bounds each Clear statement to limit lock duration. The PostgreSQL performance test measures clearing 100,000 rows; timing depends on the database and host.

View Source 
const DefaultMigrationsTable = "schema_migrations"

DefaultMigrationsTable records the storage schema's applied versions.

View Source 
const RewrapBatchSize = 500

RewrapBatchSize bounds the rows read per SELECT during Rewrap.

Variables

View Source 
var ErrMigration = errors.New("sqlcommon: migration")

ErrMigration is returned for malformed or failing migrations.

View Source 
var ErrTransaction = errors.New("sqlcommon: transaction coordination failed")

Functions

func AfterCommit 

func AfterCommit(ctx context.Context, fn func(context.Context))

AfterCommit schedules an in-process notification after commit. It runs immediately outside a transaction. Its context has no transaction or request cancellation, so the receiver must impose its own bounded lifetime.

func AfterCompletion 

func AfterCompletion(ctx context.Context, fn func(context.Context, bool))

AfterCompletion schedules work outside the transaction on either outcome. Security denials can use this to record an occurrence even after rollback.

func BlobAAD 

func BlobAAD(purpose string, keys ...string) []byte

BlobAAD binds a value to a purpose and unambiguous composite primary key.

func CurrentTransaction 

func CurrentTransaction(ctx context.Context, db *sql.DB) (*sql.Tx, bool)

CurrentTransaction reports the transaction for db in this context.

func Fail 

func Fail(ctx context.Context, err error) error

Fail prevents a transaction from committing, even when an intermediate caller ignores the returned error. Outside a transaction it returns err unchanged.

func InsertIgnoreDuplicateKey 

func InsertIgnoreDuplicateKey(table string, cols, key []string) string

InsertIgnoreDuplicateKey renders INSERT ... AS new ON DUPLICATE KEY UPDATE k = new.k, a no-op update on conflict (MySQL 8.0.19 and later).

func InsertIgnoreOnConflict 

func InsertIgnoreOnConflict(table string, cols, key []string) string

InsertIgnoreOnConflict renders INSERT ... ON CONFLICT (key) DO NOTHING (PostgreSQL and SQLite).

func Migrate 

func Migrate(ctx context.Context, db *sql.DB, d Dialect) ([]int, error)

Migrate applies every pending migration of the dialect's own set in order, each in its own transaction, and returns the versions applied.

func MigrateSet 

func MigrateSet(ctx context.Context, db *sql.DB, d Dialect, set MigrationSet) ([]int, error)

MigrateSet is Migrate for an arbitrary migration set and version table.

func MustSub 

func MustSub(fsys fs.FS, dir string) fs.FS

MustSub returns the sub-tree dir of fsys, panicking when it does not exist. Backends use it to expose their embedded migration directory as a package-level Dialect value.

func OpenBlob 

func OpenBlob(k *crypt.Keyring, purpose string, b []byte, keys ...string) ([]byte, error)

OpenBlob authenticates encrypted values and enforces strict plaintext policy.

func PutPushCertTx 

func PutPushCertTx(ctx context.Context, q interface {
	ExecContext(context.Context, string, ...any) (sql.Result, error)
	QueryRowContext(context.Context, string, ...any) *sql.Row
}, d Dialect, keys *crypt.Keyring, topic string, certPEM, keyPEM []byte, at time.Time) (storage.PushCert, error)

PutPushCertTx publishes an identity in the caller's transaction. It allows certificate workflow state and the runtime push record to commit together.

func RewrapBlobs 

func RewrapBlobs(
	ctx context.Context,
	db *sql.DB,
	d Dialect,
	k *crypt.Keyring,
	cols []BlobColumn,
) (int, error)

RewrapBlobs rotates bounded pages with compare-and-swap writes. Call until zero before removing retired keys; concurrent deletions can move page boundaries.

func Rollback 

func Rollback(ctx context.Context, db *sql.DB, d Dialect, target int) ([]int, error)

Rollback reverts applied migrations of the dialect's own set newer than target (0 reverts all), newest first, each in its own transaction.

func RollbackSet 

func RollbackSet(ctx context.Context, db *sql.DB, d Dialect, set MigrationSet, target int) ([]int, error)

RollbackSet is Rollback for an arbitrary migration set and version table.

func Savepoint 

func Savepoint(ctx context.Context, db *sql.DB, fn func(context.Context, *sql.Tx) error) (err error)

Savepoint preserves a participating store's Update contract inside an outer transaction. An ordinary callback error rolls back only that store operation; Fail still poisons the entire transaction. Callbacks must use the supplied context, and nested SQL operations must run sequentially.

func SealBlob 

func SealBlob(k *crypt.Keyring, purpose string, b []byte, keys ...string) ([]byte, error)

SealBlob encrypts a retained value when a keyring is configured.

func UpsertOnConflict 

func UpsertOnConflict(table string, cols, key []string) string

UpsertOnConflict renders INSERT ... ON CONFLICT (key) DO UPDATE SET col = excluded.col (PostgreSQL and SQLite).

func UpsertOnDuplicateKey 

func UpsertOnDuplicateKey(table string, cols, _ []string) string

UpsertOnDuplicateKey renders INSERT ... AS new ON DUPLICATE KEY UPDATE col = new.col (MySQL 8.0.19 and later).

func Version 

func Version(ctx context.Context, db *sql.DB, d Dialect) (int, error)

Version returns the highest applied version of the default set (0 when none).

func VersionOf 

func VersionOf(ctx context.Context, db *sql.DB, d Dialect, table string) (int, error)

VersionOf returns the highest applied version recorded in table.

Types

type BlobColumn 

type BlobColumn struct {
	Table, Column string
	Keys          []string
}

BlobColumn identifies a retained blob and its ordered primary-key columns. Identifiers must be compile-time schema constants, never request input.

type Dialect 

type Dialect struct {
	// Name is used in errors and in the migration table.
	Name string
	// Dollar selects $1-style placeholders (PostgreSQL); otherwise "?".
	Dollar bool
	// ForUpdate is appended to row-locking SELECTs inside transactions
	// ("FOR UPDATE"), or empty when the engine serialises writers (SQLite).
	ForUpdate string
	// Upsert renders an INSERT that updates the non-key columns on a key
	// conflict. See UpsertOnConflict and UpsertOnDuplicateKey.
	Upsert func(table string, cols, key []string) string
	// Migrations holds NNNN_name.sql files with "-- +up" and "-- +down"
	// sections.
	Migrations fs.FS
	// InsertIgnore renders an INSERT that does nothing when the key already
	// exists. See InsertIgnoreOnConflict and InsertIgnoreDuplicateKey.
	InsertIgnore func(table string, cols, key []string) string
	// IsUniqueViolation reports whether err is the engine's unique-constraint
	// error, so the store returns storage.ErrConflict instead of a driver
	// error when two writers race for the same key.
	IsUniqueViolation func(error) bool
}

Dialect describes what differs between SQL engines.

func (Dialect) Rebind 

func (d Dialect) Rebind(query string) string

Rebind converts "?" placeholders to the dialect's form. A "?" inside a single-quoted literal is left alone.

type Migration 

type Migration struct {
	Version int
	Name    string
	Up      []string
	Down    []string
}

Migration is one parsed migration file.

func LoadMigrations 

func LoadMigrations(fsys fs.FS) ([]Migration, error)

LoadMigrations parses every NNNN_name.sql file in fsys, sorted by version.

type MigrationSet 

type MigrationSet struct {
	// Table is the version table, for example "schema_migrations" or
	// "ddm_schema_migrations". Lower-case identifiers only.
	Table string
	FS    fs.FS
}

MigrationSet is one package's versioned migrations and the table that records them, so several packages (storage, ddm) can share a database without sharing a version sequence.

type Option 

type Option func(*Store)

Option configures New.

func WithKeyring 

func WithKeyring(k *crypt.Keyring) Option

WithKeyring seals the secret columns (unlock tokens, bootstrap tokens, push private keys, user auth tokens) with the keyring. Without one the columns are stored in plaintext, which is the pre-0013 behaviour.

type Pool 

type Pool struct {
	MaxOpenConns    int
	MaxIdleConns    int
	ConnMaxLifetime time.Duration
	ConnMaxIdleTime time.Duration
}

Pool tunes the database/sql connection pool. Zero values keep the driver defaults.

func (Pool) Apply 

func (p Pool) Apply(db *sql.DB)

Apply sets the pool limits on db.

type Queryer 

type Queryer interface {
	ExecContext(context.Context, string, ...any) (sql.Result, error)
	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

Queryer is a pool or the transaction shared by stores participating in Run.

func Query 

func Query(ctx context.Context, db *sql.DB) Queryer

Query returns the transaction for db, or db when no matching transaction exists.

type Store 

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

Store implements storage.Store over a *sql.DB.

func New 

func New(db *sql.DB, d Dialect, opts ...Option) *Store

New wraps an opened database. Call Migrate first.

func (*Store) AssociateCert 

func (s *Store) AssociateCert(
	ctx context.Context,
	id mdm.EnrollmentID,
	hash string,
	at time.Time,
) error

AssociateCert implements storage.CertAuthStore. The unique index on cert_hash is the arbiter: a race between two enrollments for one hash ends with one winner and ErrConflict for the rest. The pin and its history row are written in one transaction (decision record 0014).

func (*Store) AuthenticateEnrollment 

func (s *Store) AuthenticateEnrollment(
	ctx context.Context,
	id mdm.EnrollmentID,
	c storage.AuthenticateChange,
) error

AuthenticateEnrollment implements storage.EnrollmentStore.

func (*Store) BootstrapToken 

func (s *Store) BootstrapToken(ctx context.Context, id mdm.EnrollmentID) ([]byte, error)

BootstrapToken implements storage.BootstrapTokenStore.

func (*Store) CertHash 

func (s *Store) CertHash(ctx context.Context, id mdm.EnrollmentID) (string, error)

CertHash implements storage.CertAuthStore.

func (*Store) CertHashHistory 

func (s *Store) CertHashHistory(ctx context.Context, hash string) ([]storage.CertAssociation, error)

CertHashHistory implements storage.CertAuthStore.

func (*Store) CertHistory 

func (s *Store) CertHistory(
	ctx context.Context,
	id mdm.EnrollmentID,
) ([]storage.CertAssociation, error)

CertHistory implements storage.CertAuthStore.

func (*Store) Clear 

func (s *Store) Clear(
	ctx context.Context,
	id mdm.EnrollmentID,
	f storage.ClearFilter,
) (int64, error)

Clear implements storage.CommandQueue in indexed batches of ClearBatchSize rows. Each batch is its own statement, so a failure part way through returns the count applied so far; callers may simply retry.

func (*Store) ClearCommand 

func (s *Store) ClearCommand(ctx context.Context, id mdm.EnrollmentID, uuid string) (int64, error)

ClearCommand implements storage.CommandClearer.

func (*Store) ClearUserAuth 

func (s *Store) ClearUserAuth(ctx context.Context, id mdm.EnrollmentID) error

ClearUserAuth implements storage.UserAuthStore.

func (*Store) Close 

func (s *Store) Close() error

Close closes the pool.

func (*Store) Commands 

func (s *Store) Commands(
	ctx context.Context,
	id mdm.EnrollmentID,
	q storage.CommandQuery,
	p paging.Page,
) (paging.Result[storage.QueuedCommand], error)

Commands implements storage.CommandQueue with a keyset cursor on the sequence number, newest first.

func (*Store) DB 

func (s *Store) DB() *sql.DB

DB exposes the connection pool for backups, health checks, and tests.

func (*Store) Disable 

func (s *Store) Disable(ctx context.Context, id mdm.EnrollmentID, at time.Time) error

Disable implements storage.EnrollmentStore. Disabling a device channel also disables its user channels, in one transaction.

func (*Store) Enqueue 

func (s *Store) Enqueue(
	ctx context.Context,
	ids []mdm.EnrollmentID,
	cmd *mdm.Command,
	o storage.EnqueueOptions,
) (storage.EnqueueResult, error)

Enqueue implements storage.CommandQueue.

func (*Store) EnrollmentByCertHash 

func (s *Store) EnrollmentByCertHash(ctx context.Context, hash string) (mdm.EnrollmentID, error)

EnrollmentByCertHash implements storage.CertAuthStore.

func (*Store) EnrollmentByID 

func (s *Store) EnrollmentByID(ctx context.Context, id string) (*storage.Enrollment, error)

EnrollmentByID implements storage.EnrollmentStore.

func (*Store) Export 

func (s *Store) Export(
	ctx context.Context,
	p paging.Page,
) (paging.Result[storage.EnrollmentExport], error)

Export implements storage.MigrationStore (decision record 0017): device channels come before the user channels that belong to them because the order is (parent_id, id) and device rows have an empty parent. Each device row costs two extra reads (bootstrap token and history); this is an administrative operation, not a hot path.

func (*Store) Get 

func (s *Store) Get(ctx context.Context, id mdm.EnrollmentID) (*storage.Enrollment, error)

Get implements storage.EnrollmentStore.

func (*Store) Import 

func (s *Store) Import(ctx context.Context, rec storage.EnrollmentExport) error

Import implements storage.MigrationStore.

func (*Store) List 

func (s *Store) List(
	ctx context.Context,
	q storage.EnrollmentQuery,
	p paging.Page,
) (paging.Result[storage.Enrollment], error)

List implements storage.EnrollmentStore with a keyset cursor on id.

func (*Store) Next 

func (s *Store) Next(
	ctx context.Context,
	id mdm.EnrollmentID,
	skipNotNow bool,
	now time.Time,
) (*mdm.Command, error)

Next implements storage.CommandQueue.

func (*Store) Ping 

func (s *Store) Ping(ctx context.Context) error

Ping checks connectivity.

func (*Store) PushCert 

func (s *Store) PushCert(ctx context.Context, topic string) (*storage.PushCert, error)

PushCert implements storage.PushCertStore.

func (*Store) PushCertVersion 

func (s *Store) PushCertVersion(ctx context.Context, topic string) (int64, error)

PushCertVersion implements storage.PushCertStore.

func (*Store) PushCerts 

func (s *Store) PushCerts(ctx context.Context) ([]storage.PushCert, error)

PushCerts implements storage.PushCertStore.

func (*Store) PushInfo 

func (s *Store) PushInfo(
	ctx context.Context,
	ids []mdm.EnrollmentID,
) (map[mdm.EnrollmentID]mdm.Push, error)

PushInfo implements storage.PushStore.

func (*Store) Rewrap 

func (s *Store) Rewrap(ctx context.Context) (int, error)

Rewrap re-seals every value that is unsealed or sealed under a retired key so it uses the active key, and returns how many rows it rewrote. Rows changed by another writer between read and write are skipped, so callers loop until it returns 0.

func (*Store) StoreBootstrapToken 

func (s *Store) StoreBootstrapToken(
	ctx context.Context,
	id mdm.EnrollmentID,
	token []byte,
	at time.Time,
) error

StoreBootstrapToken implements storage.BootstrapTokenStore.

func (*Store) StorePushCert 

func (s *Store) StorePushCert(ctx context.Context, topic string, certPEM, keyPEM []byte, at time.Time) (storage.PushCert, error)

StorePushCert implements storage.PushCertStore (decision record 0015). The private key is sealed when a keyring is configured.

func (*Store) StoreResult 

func (s *Store) StoreResult(
	ctx context.Context,
	id mdm.EnrollmentID,
	resp *mdm.Response,
	now time.Time,
) error

StoreResult implements storage.CommandQueue.

func (*Store) StoreTokenUpdate 

func (s *Store) StoreTokenUpdate(
	ctx context.Context,
	id mdm.EnrollmentID,
	push mdm.Push,
	msg *checkin.TokenUpdate,
	raw []byte,
	at time.Time,
) error

StoreTokenUpdate implements storage.EnrollmentStore.

func (*Store) StoreUserAuthChallenge 

func (s *Store) StoreUserAuthChallenge(
	ctx context.Context,
	id mdm.EnrollmentID,
	challenge string,
	raw []byte,
	at time.Time,
) error

StoreUserAuthChallenge implements storage.UserAuthStore.

func (*Store) StoreUserAuthToken 

func (s *Store) StoreUserAuthToken(
	ctx context.Context,
	id mdm.EnrollmentID,
	token string,
	raw []byte,
	at time.Time,
) error

StoreUserAuthToken implements storage.UserAuthStore.

func (*Store) TouchLastSeen 

func (s *Store) TouchLastSeen(ctx context.Context, id mdm.EnrollmentID, at time.Time) error

TouchLastSeen implements storage.EnrollmentStore; it never moves the timestamp backwards.

func (*Store) TransitionReplacement 

func (s *Store) TransitionReplacement(
	ctx context.Context,
	id mdm.EnrollmentID,
	change storage.ReplacementChange,
) (*storage.Replacement, error)

func (*Store) UpsertAuthenticate 

func (s *Store) UpsertAuthenticate(
	ctx context.Context,
	id mdm.EnrollmentID,
	msg *checkin.Authenticate,
	raw []byte,
	at time.Time,
) error

UpsertAuthenticate implements storage.EnrollmentStore.

func (*Store) UserAuth 

func (s *Store) UserAuth(ctx context.Context, id mdm.EnrollmentID) (*storage.UserAuthState, error)

UserAuth implements storage.UserAuthStore.

type UnitOfWork 

type UnitOfWork struct {
	DB      *sql.DB
	Dialect Dialect
}

UnitOfWork coordinates stores using the same pool. Configure every store with that pool and use Query or CurrentTransaction for each operation. A callback must finish all database work before returning; it must not retain its context for background work or perform a remote side effect inside the transaction.

func (UnitOfWork) Run 

func (w UnitOfWork) Run(ctx context.Context, fn func(context.Context) error) (err error)

Run commits local mutations together. A nested error poisons the enclosing transaction. Different pools cannot be combined into an atomic operation.

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.