Help us shape the future of Go! Take the 2026 Go Developer Survey

postgres

package
v6.0.1 Latest Latest
Warning

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

Go to latest
Published: Oct 28, 2026 License: MIT Imports: 20 Imported by: 0

Documentation

Overview

Package postgres provides command/desired-state transactions and tenant audit history for explicit control-plane composition. NewRuntime borrows a go-postgres/v3 pool. The deploying application applies the hand-maintained embedded MigrationSource with its own runner before use. Constructors do not install/verify schema or own pool shutdown. Driver diagnostics may be wrapped in Go errors; the administrative HTTP layer selects public codes.

Index

Constants

View Source
const (
	// MaxCommandPageSize bounds one command-history database response.
	MaxCommandPageSize uint32 = 1_000
	// MaxCommandCursorBytes bounds opaque command-history pagination state.
	MaxCommandCursorBytes = 1_024
	// MaxCommandRetentionBatch bounds one terminal-command cleanup transaction.
	MaxCommandRetentionBatch uint32 = 1_000
)
View Source
const (
	// MaxAuditBatch bounds verification and retention database work.
	MaxAuditBatch uint32 = 1_000
)

Variables

View Source
var (
	// ErrInvalidAuditRequest reports an unscoped or unbounded audit operation.
	ErrInvalidAuditRequest = errors.New("postgres: invalid audit request")
	// ErrAuditAnchorNotFound reports a tenant without initialized audit state.
	ErrAuditAnchorNotFound = errors.New("postgres: audit anchor not found")
	// ErrInvalidAuditState reports malformed persisted audit metadata.
	ErrInvalidAuditState = errors.New("postgres: invalid audit state")
)
View Source
var (
	// ErrInvalidCommandRequest reports an unbounded command lookup scope.
	ErrInvalidCommandRequest = errors.New("postgres: invalid command request")
	// ErrInvalidCommandState reports a malformed persisted command result.
	ErrInvalidCommandState = errors.New("postgres: invalid command state")
	// ErrInvalidCommandRetentionRequest reports an unsafe cleanup scope.
	ErrInvalidCommandRetentionRequest = errors.New("postgres: invalid command retention request")
	// ErrInvalidCommandRetentionState reports an impossible deletion count.
	ErrInvalidCommandRetentionState = errors.New("postgres: invalid command retention state")
)
View Source
var (
	// ErrNilQueryer reports a missing PostgreSQL read boundary.
	ErrNilQueryer = errors.New("postgres: queryer is nil")
	// ErrInvalidDesiredLookup reports an unscoped or unsupported target.
	ErrInvalidDesiredLookup = errors.New("postgres: invalid desired-state lookup")
	// ErrDesiredStateNotFound reports a target without durable desired state.
	ErrDesiredStateNotFound = errors.New("postgres: desired state not found")
)
View Source
var (
	// ErrNilBeginner reports a missing PostgreSQL transaction source.
	ErrNilBeginner = errors.New("postgres: transaction beginner is nil")
	// ErrIdempotencyConflict reports reuse of a key for another command.
	ErrIdempotencyConflict = errors.New("postgres: idempotency key conflicts with stored command")
	// ErrCompletionConflict reports an attempt to replace a terminal result.
	ErrCompletionConflict = errors.New("postgres: terminal command result conflicts with stored result")
	// ErrCommandNotFound reports completion of an unknown tenant-scoped key.
	ErrCommandNotFound = errors.New("postgres: command not found")
	// ErrAuditSequenceExhausted reports an audit chain beyond bigint capacity.
	ErrAuditSequenceExhausted = errors.New("postgres: audit sequence exhausted")
	// ErrAuditHashInvalid reports malformed persisted audit-chain state.
	ErrAuditHashInvalid = errors.New("postgres: invalid audit hash")
	// ErrDesiredStateConflict reports a concurrent desired-state replacement.
	ErrDesiredStateConflict = errors.New("postgres: desired state update conflict")
)
View Source
var ErrInvalidRuntimePool = errors.New("postgres: invalid runtime pool")

ErrInvalidRuntimePool reports a missing or uninitialized PostgreSQL pool.

View Source
var ErrResultNotTerminal = errors.New("postgres: command result is not terminal")

ErrResultNotTerminal rejects accepted results at the completion boundary.

Functions

func MigrationSource

func MigrationSource() (migrations.Source, error)

MigrationSource returns the immutable embedded control-plane schema history.

Types

type AuditPage

type AuditPage struct {
	// Entries is a newly materialized sequence-ordered page, verified relative
	// to its first stored predecessor.
	Entries []history.Entry
	// NextSequence is the exclusive cursor for another page; zero means no
	// further page was detected.
	NextSequence uint64
}

AuditPage is one verified bounded tenant audit-history page.

type AuditStore

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

AuditStore verifies and retains tenant audit history transactionally.

func NewAuditStore

func NewAuditStore(beginner gopostgres.Beginner) (*AuditStore, error)

NewAuditStore creates a transaction-backed audit operations repository.

func (*AuditStore) AuditSensitiveAccess

func (s *AuditStore) AuditSensitiveAccess(
	ctx context.Context,
	access controlplane.SensitiveAccess,
) error

AuditSensitiveAccess durably records one authorized privileged record read.

func (*AuditStore) ListTenant

func (s *AuditStore) ListTenant(
	ctx context.Context,
	tenant string,
	after uint64,
	limit uint32,
) (AuditPage, error)

ListTenant returns one internally linked page after the exclusive sequence cursor. It checks contiguity/digests relative to the first stored predecessor, not linkage back to the tenant retention anchor; use VerifyTenant for that traversal. Neither operation authenticates database state against an external trusted root.

func (*AuditStore) RetainBefore

func (s *AuditStore) RetainBefore(
	ctx context.Context,
	tenant string,
	cutoff time.Time,
	batchSize uint32,
) (RetentionResult, error)

RetainBefore deletes one contiguous old prefix and advances its anchor.

func (*AuditStore) VerifyTenant

func (s *AuditStore) VerifyTenant(
	ctx context.Context,
	tenant string,
	pageSize uint32,
) (VerificationReport, error)

VerifyTenant traverses retained tenant history from its database anchor in a repeatable-read snapshot. pageSize bounds each fetched batch, not total work; callers own the overall context/deadline and external anchor trust.

type CommandPage

type CommandPage struct {
	// Records is a newly materialized newest-first page ordered by requested
	// time and idempotency key.
	Records []CommandRecord
	// NextCursor is opaque bounded pagination state; empty means no further
	// page was detected.
	NextCursor string
}

CommandPage is one bounded newest-first tenant command-history page.

type CommandRecord

type CommandRecord struct {
	// Command is the stored envelope, including independently materialized
	// option pointers.
	Command controlplane.Command
	// Result is the stored outcome with materialized optional acknowledgement
	// fields.
	Result controlplane.CommandResult
}

CommandRecord joins one immutable command envelope to its durable outcome.

type CommandRetentionResult

type CommandRetentionResult struct {
	// Deleted counts terminal commands removed after audit/desired-state
	// reference refusals.
	Deleted uint32
}

CommandRetentionResult reports one bounded terminal-command cleanup batch.

type CommandStore

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

CommandStore reads tenant-scoped durable command outcomes.

func NewCommandStore

func NewCommandStore(beginner gopostgres.Beginner) (*CommandStore, error)

NewCommandStore creates a transaction-backed command result reader.

func (*CommandStore) Get

func (s *CommandStore) Get(
	ctx context.Context,
	tenant string,
	key string,
) (controlplane.CommandResult, error)

Get returns one durable result by tenant and idempotency key.

func (*CommandStore) ListTenant

func (s *CommandStore) ListTenant(
	ctx context.Context,
	tenant string,
	cursor string,
	limit uint32,
) (CommandPage, error)

ListTenant returns one newest-first tenant command-history page.

func (*CommandStore) RetainCommandsBefore

func (s *CommandStore) RetainCommandsBefore(
	ctx context.Context,
	tenant string,
	cutoff time.Time,
	batchSize uint32,
) (CommandRetentionResult, error)

RetainCommandsBefore deletes old terminal commands that no retained audit event or current desired state references.

type DesiredStore

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

DesiredStore reads tenant-scoped operational state for worker convergence.

func NewDesiredStore

func NewDesiredStore(queryer rowQueryer) (*DesiredStore, error)

NewDesiredStore creates a desired-state read repository.

func (*DesiredStore) Get

func (s *DesiredStore) Get(
	ctx context.Context,
	tenant string,
	target controlplane.Target,
) (control.DesiredRecord, error)

Get returns the latest durable desired state for one target.

type Journal

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

Journal persists command outcomes and their audit records in one transaction.

func NewJournal

func NewJournal(beginner gopostgres.Beginner) (*Journal, error)

NewJournal creates a journal backed by postgres transaction semantics.

func (*Journal) Accept

Accept atomically records a new command, applicable desired-state revision and initial audit event. Exact duplicates return stored results without a new event or desired revision; changed content conflicts. CommandID alone does not select a different envelope, and persisted times normalize to UTC microseconds.

func (*Journal) Complete

func (j *Journal) Complete(
	ctx context.Context,
	result controlplane.CommandResult,
) error

Complete atomically records a terminal command result and audit event. Core transaction work receives at most five seconds, or the earlier deadline of the context passed here; values and cancellation are preserved. Service intentionally detaches pre-dispatch cancellation before calling Complete. PostgreSQL rollback has its separate five-second cleanup budget; cooperative driver I/O means this is not a forced total end-to-end completion limit.

func (*Journal) MarkAcknowledged

func (j *Journal) MarkAcknowledged(ctx context.Context, result controlplane.CommandResult) error

MarkAcknowledged atomically persists a valid data-plane acknowledgement and audit before terminal completion. Acknowledgement is distinct from business success: a reported failed/partial outcome may still have been acknowledged.

func (*Journal) MarkDispatched

func (j *Journal) MarkDispatched(ctx context.Context, result controlplane.CommandResult) error

MarkDispatched atomically persists the pre-dispatch recovery boundary and its audit event.

type PoolReadiness

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

PoolReadiness adapts postgres bounded health checks to the API contract.

func (*PoolReadiness) Ready

func (readiness *PoolReadiness) Ready(ctx context.Context) error

Ready performs a bounded PostgreSQL ping through postgres.

type RetentionResult

type RetentionResult struct {
	// Deleted counts entries deleted in this bounded transaction.
	Deleted uint32
	// AnchorSequence is the last retained-away sequence after prefix deletion.
	AnchorSequence uint64
	// AnchorHash preserves the digest needed to link later retained history.
	AnchorHash history.Hash
	// RetainedThrough is the monotonic UTC retention cutoff represented by the
	// removed prefix.
	RetainedThrough time.Time
}

RetentionResult describes one bounded retained-prefix cleanup batch.

type Runtime

type Runtime struct {
	// Journal coordinates command, desired-state and audit writes against the
	// borrowed pool.
	Journal *Journal
	// Audit provides tenant audit reads, verification and retention.
	Audit *AuditStore
	// Commands provides scoped command lookup, pagination and retention.
	Commands *CommandStore
	// Desired reads latest durable desired state; it does not drive worker
	// convergence.
	Desired *DesiredStore
	// Readiness pings the borrowed pool through its configured health-check
	// budget.
	Readiness *PoolReadiness
}

Runtime is the complete PostgreSQL-backed control-plane service bundle. The caller retains ownership of the underlying pool.

func NewRuntime

func NewRuntime(pool *gopostgres.Pool) (*Runtime, error)

NewRuntime wires persistence services to a go-postgres/v3 pool. It refuses nil or uninitialized pools. The caller retains pool ownership; transactions use the native pgxpool, and readiness uses the pool's bounded Ping.

type VerificationReport

type VerificationReport struct {
	// Events counts retained events traversed, excluding an already retained-away prefix.
	Events uint64
	// HeadSequence is the last verified retained sequence or anchor sequence
	// for an empty chain.
	HeadSequence uint64
	// HeadHash is the corresponding database-backed chain digest, not external
	// authentication.
	HeadHash history.Hash
}

VerificationReport identifies the verified tenant audit head.

Jump to

Keyboard shortcuts

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