postgres

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Jun 7, 2026 License: MIT Imports: 19 Imported by: 0

Documentation

Overview

Package postgres provides a PostgreSQL implementation of configcore ports. It does NOT import adapters/postgres — it defines its own DBTX interface to match pgx.Tx / pgxpool.Pool, keeping the cell decoupled from the adapter layer.

Package postgres provides a PostgreSQL implementation of configcore ports.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type ConfigRepoOption 

type ConfigRepoOption func(*ConfigRepository)

ConfigRepoOption configures optional behavior on ConfigRepository.

func WithOnStaleCipher 

func WithOnStaleCipher(fn func(key, storedKeyID, currentKeyID string)) ConfigRepoOption

WithOnStaleCipher sets a callback invoked when a stale-key value is detected during a read. Callers may wire this to a prometheus counter: 

WithOnStaleCipher(func(key, storedKeyID, currentKeyID string) {
    staleCipherTotal.Inc()
})

The callback receives (key, storedKeyID, currentKeyID). When nil, it is skipped; slog.Warn is always emitted regardless.

type ConfigRepository 

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

ConfigRepository implements ports.ConfigRepository using PostgreSQL.

func NewConfigRepository 

func NewConfigRepository(
	s *Session, tr kcrypto.ValueTransformer, logger *slog.Logger,
	clk clock.Clock, opts ...ConfigRepoOption,
) *ConfigRepository

NewConfigRepository creates a ConfigRepository that resolves the ambient pgx.Tx from the context on each call, enabling transactional participation via persistence.TxCtxKey. Session is the sole production entry point; use the unexported newConfigRepositoryFromDBTX in tests.

clk must be non-nil; pass clock.Real() in production and clockmock.New() in tests. If logger is nil, slog.Default() is used.

Requires migrations 001–010 to be applied first (see adapters/postgres/migrations/).

func (*ConfigRepository) Create 

func (r *ConfigRepository) Create(ctx context.Context, t tenant.TenantID, entry *domain.ConfigEntry) error

Create inserts a new config entry. For sensitive=true: encrypts value and writes cipher columns; value column is set to "". For sensitive=false: writes plaintext value; cipher columns are NULL.

func (*ConfigRepository) Delete 

func (r *ConfigRepository) Delete(ctx context.Context, t tenant.TenantID, key string, expectedVersion int) (*domain.ConfigEntry, error)

Delete atomically removes a config entry by key if expectedVersion matches the stored version (CAS guard). Returns the deleted row via DELETE...RETURNING, enabling callers to publish a tombstone event without a separate pre-read. Returns ErrConfigRepoNotFound if the key does not exist, or ErrVersionConflict if expectedVersion does not match.

func (*ConfigRepository) GetByKey 

func (r *ConfigRepository) GetByKey(ctx context.Context, t tenant.TenantID, key string) (*domain.ConfigEntry, error)

GetByKey retrieves a config entry by key with transparent decryption for sensitive entries. Sets entry.Stale=true when keyID != current active key.

func (*ConfigRepository) GetVersion 

func (r *ConfigRepository) GetVersion(ctx context.Context, t tenant.TenantID, configID string, version int) (*domain.ConfigVersion, error)

GetVersion retrieves a specific config version with transparent decryption.

func (*ConfigRepository) List 

func (r *ConfigRepository) List(ctx context.Context, t tenant.TenantID, params query.ListParams) ([]*domain.ConfigEntry, error)

List retrieves config entries with keyset cursor pagination, scoped to the tenant via the WHERE tenant_id = $1 predicate.

Performance note: keyset pagination on `(tenant_id, key, id)` scans well in practice because (a) the primary-key unique index supports the `id` tie- breaker and (b) the `key` column is typically low-cardinality for admin browsing within a tenant. A dedicated `(tenant_id ASC, key ASC, id ASC)` composite index can be added in a future migration if sort-heavy list traffic warrants it; it is intentionally not shipped in migration 051 (the tenant rebuild) to keep that migration's scope minimal.

Sensitive entries: List does NOT decrypt values. Instead, the Value field is set to "***" (sentinel) and KeyID / Stale are preserved from the cipher columns. Callers must use GetByKey to retrieve the decrypted plaintext for a specific entry.

This design avoids bulk decryption on list operations and prevents accidental exposure of sensitive values in list responses.

func (*ConfigRepository) PublishVersion 

func (r *ConfigRepository) PublishVersion(ctx context.Context, t tenant.TenantID, version *domain.ConfigVersion) error

PublishVersion inserts a config version record. For sensitive=true: encrypts value and writes cipher columns.

func (*ConfigRepository) RepoReady 

func (r *ConfigRepository) RepoReady(ctx context.Context) error

RepoReady implements healthz.RepoProber. It issues three cheap non-transactional representative Exec probes — SELECT 1 FROM config_entries WHERE false, SELECT 1 FROM feature_flags WHERE false, and SELECT 1 FROM config_versions WHERE false — so that missing tables, dropped columns, or revoked table-level permissions are detected independently of the pool-level postgres_ready probe. WHERE false short-circuits the scan so there is no result-iteration overhead, and Exec (matching PGSessionStore.RepoReady / LedgerStore.RepoReady) collapses each table probe to a single failure branch — no transaction is opened. config_versions is included because it is load-bearing for tenant-scoped versioning (PR-2b #1479).

func (*ConfigRepository) Update 

func (r *ConfigRepository) Update(
	ctx context.Context, t tenant.TenantID, key string, expectedVersion int, value string,
) (*domain.ConfigEntry, error)

Update atomically sets value and increments version if expectedVersion matches the current version (CAS guard). Preserves the existing sensitive flag — reads it internally via SELECT...FOR UPDATE to eliminate any TOCTOU race on the sensitive flag. Returns ErrConfigRepoNotFound if the key does not exist, or ErrVersionConflict if expectedVersion does not match the stored version.

func (*ConfigRepository) UpdateForRollback 

func (r *ConfigRepository) UpdateForRollback(
	ctx context.Context, t tenant.TenantID, key string, expectedVersion int, value string, sensitive bool,
) (*domain.ConfigEntry, error)

UpdateForRollback atomically sets value AND sensitive, increments version if expectedVersion matches the current version (CAS guard). Used exclusively by configpublish.Rollback to restore a snapshot's sensitivity alongside its value. Returns ErrConfigRepoNotFound if the key does not exist, or ErrVersionConflict if expectedVersion does not match.

type DBTX 

type DBTX interface {
	Exec(ctx context.Context, sql string, args ...any) (int64, error)
	Query(ctx context.Context, sql string, args ...any) (Rows, error)
	QueryRow(ctx context.Context, sql string, args ...any) RowScanner
}

DBTX abstracts the database operations needed by ConfigRepository. Both pgxpool.Pool and pgx.Tx satisfy this interface.

type FlagRepository 

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

FlagRepository implements ports.FlagRepository using PostgreSQL.

Write paths (Create/Update/Delete/Toggle) require an ambient pgx.Tx in ctx via persistence.TxCtxKey — enforced by resolveWriteDB. Read paths (GetByKey/List) fall back to the pool when no tx is present.

ref: Unleash feature-environment-store.ts — Toggle is a dedicated method that does NOT overwrite rollout_percentage or description, preventing concurrent-write data loss (Unleash lesson: "write + event must be atomic").

func NewFlagRepository 

func NewFlagRepository(s *Session, clk clock.Clock) *FlagRepository

NewFlagRepository creates a FlagRepository that resolves the ambient pgx.Tx from the context on each write call.

clk must be non-nil; pass clock.Real() in production and clockmock.New() in tests. Requires migrations 008 (table) and 009 (concurrent index) to be applied. The adapterpg schema guard (VerifyExpectedVersion) enforces the actual current expected version at startup; this comment is documentation-only and deliberately does not duplicate that check.

func (*FlagRepository) Create 

func (r *FlagRepository) Create(ctx context.Context, t tenant.TenantID, flag *domain.FeatureFlag) error

Create inserts a new feature flag. All 8 columns are written plus tenant_id.

func (*FlagRepository) Delete 

func (r *FlagRepository) Delete(ctx context.Context, t tenant.TenantID, key string, expectedVersion int) (*domain.FeatureFlag, error)

Delete removes a feature flag by key if expectedVersion matches the stored version (CAS guard). Returns the deleted entity via DELETE...RETURNING. Returns ErrFlagNotFound if the key does not exist, or ErrVersionConflict if expectedVersion does not match.

CAS flow: rowsAffected==0 → probe GetByKey:

  • exists → ErrVersionConflict (409)
  • not found → ErrFlagNotFound (404)

func (*FlagRepository) GetByKey 

func (r *FlagRepository) GetByKey(ctx context.Context, t tenant.TenantID, key string) (*domain.FeatureFlag, error)

GetByKey retrieves a feature flag by key.

func (*FlagRepository) List 

func (r *FlagRepository) List(ctx context.Context, t tenant.TenantID, params query.ListParams) ([]*domain.FeatureFlag, error)

List retrieves feature flags with keyset cursor pagination, scoped to the tenant via the WHERE tenant_id = $1 predicate. Requires composite index: CREATE INDEX idx_feature_flags_tenant_key_id ON feature_flags (tenant_id ASC, key ASC, id ASC).

func (*FlagRepository) Toggle 

func (r *FlagRepository) Toggle(
	ctx context.Context, t tenant.TenantID, key string, expectedVersion int, enabled bool,
) (*domain.FeatureFlag, error)

Toggle atomically sets the enabled state and increments version by 1 if expectedVersion matches the stored version (CAS guard). It does NOT overwrite rollout_percentage or description. Returns the updated flag via RETURNING clause. Returns ErrFlagNotFound if the key does not exist, or ErrVersionConflict if expectedVersion does not match.

ref: Unleash feature-environment-store.ts toggleEnvironment — dedicated toggle method prevents concurrent overwrites on unrelated fields.

CAS flow: rowsAffected==0 → probe GetByKey:

  • exists → ErrVersionConflict (409)
  • not found → ErrFlagNotFound (404)

func (*FlagRepository) Update 

func (r *FlagRepository) Update(
	ctx context.Context, t tenant.TenantID, key string, expectedVersion int, enabled bool, rolloutPercentage int, description string,
) (*domain.FeatureFlag, error)

Update atomically sets enabled, rollout_percentage, description, and increments version by 1 via UPDATE...SET version=version+1 RETURNING. Returns the updated flag. Returns ErrFlagNotFound if key does not exist, or ErrVersionConflict if expectedVersion does not match.

CAS flow: rowsAffected==0 → probe GetByKey:

  • exists → ErrVersionConflict (409)
  • not found → ErrFlagNotFound (404)

type PlaintextMigrationConfig 

type PlaintextMigrationConfig struct {
	// BatchSize is the number of rows to encrypt per DB round-trip.
	// Defaults to 50 when zero.
	BatchSize int

	// RateLimitDelay is an optional sleep between batches to reduce DB load.
	// Zero means no delay.
	RateLimitDelay time.Duration
}

PlaintextMigrationConfig controls the batch-encrypt migration behavior.

type PlaintextMigrationResult 

type PlaintextMigrationResult struct {
	// Processed is the number of rows that were encrypted during this run.
	Processed int
	// Skipped is the number of rows that were already encrypted (idempotent).
	Skipped int
}

PlaintextMigrationResult summarizes a completed migration run.

type RowScanner 

type RowScanner interface {
	Scan(dest ...any) error
}

RowScanner is a local copy of adapters/postgres.RowScanner; cells/ cannot import adapters/ — intentional per layering rules.

type Rows 

type Rows interface {
	Next() bool
	Scan(dest ...any) error
	Close()
	Err() error
}

Rows abstracts a query result set.

type Session 

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

Session resolves the ambient pgx.Tx from ctx (placed there by adapters/postgres.TxManager via persistence.TxCtxKey) or falls back to the pool for non-transactional reads.

Design: cells/ cannot import adapters/postgres — the layering rule forbids it. persistence.TxCtxKey is kernel-owned so both layers can share the key. Session wraps the concrete pgx types inside a dbtxAdapter that implements the cell-local DBTX interface (int64 Exec return), keeping the test mocks unchanged.

ref: go-zero TransactCtx — session injected via context; downstream participants retrieve from ctx without knowing the adapter. Adopted pattern.

func NewSession 

func NewSession(pool *pgxpool.Pool) *Session

NewSession creates a Session backed by the given pool.

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.