postgres

package
v5.0.0 Latest Latest
Warning

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

Go to latest
Published: Oct 22, 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