db

package
v0.60.0 Latest Latest
Warning

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

 Go to latest Published: Jul 19, 2026 License: AGPL-3.0 Imports: 34 Imported by: 0

Documentation

Index

Constants

View Source 
const (

	// DefaultRiverSoftStopTimeout is the fallback drain budget for the self-contained
	// scheduler client (default/test path). Production injects a value parsed from
	// STELLA_RIVER_SOFT_STOP_TIMEOUT via the composition root; this package reads no
	// env itself.
	DefaultRiverSoftStopTimeout = 120 * time.Second
)

Variables

View Source 
var MigrationsFS embed.FS

MigrationsFS holds the goose migration files. Embed the entire directory so it works even when no migrations exist yet.

Functions

func AdvisoryXactLock added in v0.50.0 

func AdvisoryXactLock(ctx context.Context, tx pgx.Tx, key string) error

AdvisoryXactLock acquires a PostgreSQL transaction-scoped advisory lock keyed by an arbitrary string. Concurrent transactions that request the same key run serially; transactions with different keys are unaffected. Under SQLite this serialization was implicit (single writer); under PostgreSQL writers run in parallel, so call this to guard a read-modify-write that must not interleave for a given entity. The lock releases automatically when tx commits or rolls back — there is no unlock to forget. Prefix keys by domain (e.g. "mem:", "group:") so unrelated entities don't share a slot in the 64-bit lock space.

func NewWorkingRiverClient added in v0.50.0 

func NewWorkingRiverClient(pool *pgxpool.Pool, queues map[string]river.QueueConfig, workers *river.Workers, logger *slog.Logger, softStopTimeout time.Duration) (*river.Client[pgx.Tx], error)

NewWorkingRiverClient builds a WORKING (electable) River client that works the given queues with the given workers. River elects a single leader per database to run leader-only maintenance — including the periodic-job enqueuer, which only enqueues the periodic jobs registered on the LEADER client. A second electable client could win leadership and silently starve another client's periodic jobs (e.g. the scheduler's cron).

INVARIANT: exactly ONE electable client per database. That is a process+database scoped property and cannot be enforced here — the test suite legitimately builds many across separate databases in one process. The composition root (cmd/stellad buildSharedRiverClient) is the single production enforcement point: it assembles one client from every subsystem's queues + workers and injects it back via BindRiverClient, so no subsystem constructs its own electable client. A new electable construction site is a bug; route subsystems through the composition root instead. An insert-only client (no queues, never Started) must use river.NewClient directly, not this constructor — which is what the guard below enforces: a working client must actually have queues and workers to work.

softStopTimeout bounds how long Stop waits for in-flight jobs to drain before River cancels their work contexts (escalating to a hard stop). It is injected rather than read from the environment here so this package stays env-free; the composition root parses STELLA_RIVER_SOFT_STOP_TIMEOUT and passes it in.

func OpenDB 

func OpenDB(dsn string, opts ...Option) (*pgxpool.Pool, error)

OpenDB opens the PostgreSQL database at dsn, sizes the connection pool, ensures required PostgreSQL extensions and the schema are present, and returns a pool safe for concurrent use. dsn is a libpq/pgx connection string (e.g. "postgres://user:pass@host:5432/db?sslmode=disable").

The server must be PostgreSQL 18 or newer: the schema baseline defaults ids with the uuidv7() built-in, so migrate() fails with "function uuidv7() does not exist" against an older server.

Types

type AuthStore 

type AuthStore struct {
	*OIDCStore
	// contains filtered or unexported fields
}

AuthStore implements auth.AuthStore using sqlc queries backed by PostgreSQL. It embeds OIDCStore to satisfy all new auth store interfaces so that gateway.go can pass a single *AuthStore to all wiring points without knowing about the split between sqlc-backed and raw-SQL-backed stores.

func NewAuthStore 

func NewAuthStore(db *pgxpool.Pool) *AuthStore

NewAuthStore creates a new AuthStore wrapping the given database connection.

func (*AuthStore) AssignAgent 

func (s *AuthStore) AssignAgent(ctx context.Context, userID string, agentID string) error

func (*AuthStore) CreateUserToken 

func (s *AuthStore) CreateUserToken(ctx context.Context, token auth.UserToken) (auth.UserToken, error)

func (*AuthStore) GetActiveAutoUserToken 

func (s *AuthStore) GetActiveAutoUserToken(ctx context.Context, userID string) (auth.UserToken, error)

func (*AuthStore) GetActiveUserTokenByHash 

func (s *AuthStore) GetActiveUserTokenByHash(ctx context.Context, tokenHash string) (auth.UserToken, error)

func (*AuthStore) GetUserTokenByHash 

func (s *AuthStore) GetUserTokenByHash(ctx context.Context, tokenHash string) (auth.UserToken, error)

func (*AuthStore) ListAgentUserIDs 

func (s *AuthStore) ListAgentUserIDs(ctx context.Context, agentID string) ([]string, error)

func (*AuthStore) ListUserAgentIDs 

func (s *AuthStore) ListUserAgentIDs(ctx context.Context, userID string) ([]string, error)

func (*AuthStore) RemoveAgent 

func (s *AuthStore) RemoveAgent(ctx context.Context, userID string, agentID string) error

func (*AuthStore) RevokeUserToken 

func (s *AuthStore) RevokeUserToken(ctx context.Context, id string) (int64, error)

func (*AuthStore) RotateUserToken 

func (s *AuthStore) RotateUserToken(ctx context.Context, id string) (int64, error)

func (*AuthStore) UpdateUserTokenLastUsed 

func (s *AuthStore) UpdateUserTokenLastUsed(ctx context.Context, id string) (int64, error)

type Embedded added in v0.50.0 

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

Embedded is a managed PostgreSQL server that stellad runs as a child process. It lets the server and its tests run against real PostgreSQL with nothing to install or operate: the server binary is downloaded once and cached under the user home, then started on demand. Start it, open the DSN with OpenDB, and Stop it on shutdown.

func StartEmbedded added in v0.50.0 

func StartEmbedded(dataDir string, port uint32) (*Embedded, error)

StartEmbedded boots a PostgreSQL server on the given TCP port and returns a ready-to-use handle. A port of 0 picks a free ephemeral port — use it for tests so parallel test binaries never collide. dataDir is where the cluster's files live: a stable path persists data across restarts (the server runtime), while an empty string uses a throwaway temp dir that is removed on Stop (tests).

func (*Embedded) DSN added in v0.50.0 

func (e *Embedded) DSN() string

DSN returns the libpq connection string for the server's default "stella" database.

func (*Embedded) DSNFor added in v0.50.0 

func (e *Embedded) DSNFor(database string) string

DSNFor returns the connection string for a named database on the server. Tests use it to reach the maintenance "postgres" database (to create and drop per-test databases) and the isolated databases they clone.

func (*Embedded) Stop added in v0.50.0 

func (e *Embedded) Stop() error

Stop shuts the server down. It is safe to call once; a stopped server cannot be restarted (create a new one with StartEmbedded).

type ExtensionRequirement added in v0.50.0 

type ExtensionRequirement struct {
	Name            string
	Required        bool
	MinVersion      string
	RequiresPreload bool
}

type OIDCStore added in v0.38.0 

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

OIDCStore implements auth store interfaces using raw SQL against the OIDC tables (auth_user, auth_identity, auth_session, channel_identity, auth_credential).

func NewOIDCStore added in v0.38.0 

func NewOIDCStore(db *pgxpool.Pool) *OIDCStore

NewOIDCStore creates an OIDCStore backed by the given database connection.

func (*OIDCStore) AssignAgent added in v0.38.0 

func (s *OIDCStore) AssignAgent(ctx context.Context, userID, agentID string) error

func (*OIDCStore) BeginAuthTx added in v0.38.0 

func (s *OIDCStore) BeginAuthTx(ctx context.Context) (auth.AuthStores, func() error, func(), error)

BeginAuthTx starts a database transaction and returns tx-scoped copies of all auth stores. Implements auth.Transactioner so AuthService.ProcessOIDCLogin can run the entire login flow atomically.

func (*OIDCStore) CountUsers added in v0.38.0 

func (s *OIDCStore) CountUsers(ctx context.Context) (int64, error)

func (*OIDCStore) CreateChannelIdentity added in v0.38.0 

func (s *OIDCStore) CreateChannelIdentity(ctx context.Context, i auth.ChannelIdentity) (auth.ChannelIdentity, error)

func (*OIDCStore) CreateCredential added in v0.38.0 

func (s *OIDCStore) CreateCredential(ctx context.Context, c auth.Credential) (auth.Credential, error)

func (*OIDCStore) CreateLoginIdentity added in v0.38.0 

func (s *OIDCStore) CreateLoginIdentity(ctx context.Context, i auth.LoginIdentity) (auth.LoginIdentity, error)

func (*OIDCStore) CreateSession added in v0.38.0 

func (s *OIDCStore) CreateSession(ctx context.Context, sess auth.Session) (auth.Session, error)

func (*OIDCStore) CreateUser added in v0.38.0 

func (s *OIDCStore) CreateUser(ctx context.Context, u auth.User) (auth.User, error)

func (*OIDCStore) DeleteChannelIdentity added in v0.38.0 

func (s *OIDCStore) DeleteChannelIdentity(ctx context.Context, id string) error

func (*OIDCStore) DeleteCredential added in v0.38.0 

func (s *OIDCStore) DeleteCredential(ctx context.Context, userID string) error

func (*OIDCStore) DeleteExpiredSessions added in v0.38.0 

func (s *OIDCStore) DeleteExpiredSessions(ctx context.Context) error

func (*OIDCStore) DeleteSession added in v0.38.0 

func (s *OIDCStore) DeleteSession(ctx context.Context, id string) error

func (*OIDCStore) DeleteUser added in v0.38.0 

func (s *OIDCStore) DeleteUser(ctx context.Context, id string) error

func (*OIDCStore) DeleteUserSessions added in v0.38.0 

func (s *OIDCStore) DeleteUserSessions(ctx context.Context, userID string) error

func (*OIDCStore) GetChannelIdentity added in v0.38.0 

func (s *OIDCStore) GetChannelIdentity(ctx context.Context, id string) (auth.ChannelIdentity, error)

func (*OIDCStore) GetChannelIdentityByPlatform added in v0.38.0 

func (s *OIDCStore) GetChannelIdentityByPlatform(ctx context.Context, platform, externalID string) (auth.ChannelIdentity, error)

func (*OIDCStore) GetCredentialByUserID added in v0.38.0 

func (s *OIDCStore) GetCredentialByUserID(ctx context.Context, userID string) (auth.Credential, error)

func (*OIDCStore) GetLoginIdentityByProvider added in v0.38.0 

func (s *OIDCStore) GetLoginIdentityByProvider(ctx context.Context, provider, providerSubject string) (auth.LoginIdentity, error)

func (*OIDCStore) GetSession added in v0.38.0 

func (s *OIDCStore) GetSession(ctx context.Context, id string) (auth.Session, error)

func (*OIDCStore) GetSessionByTokenHash added in v0.38.0 

func (s *OIDCStore) GetSessionByTokenHash(ctx context.Context, tokenHash string) (auth.Session, error)

func (*OIDCStore) GetUser added in v0.38.0 

func (s *OIDCStore) GetUser(ctx context.Context, id string) (auth.User, error)

func (*OIDCStore) GetUserByEmail added in v0.38.0 

func (s *OIDCStore) GetUserByEmail(ctx context.Context, email string) (auth.User, error)

func (*OIDCStore) GetVaultUser added in v0.38.0 

func (s *OIDCStore) GetVaultUser(ctx context.Context, id string) (sqlc.VaultUser, error)

GetVaultUser returns the age key fields for a user. Satisfies vault.DB.

func (*OIDCStore) ListActiveUserIDs added in v0.38.0 

func (s *OIDCStore) ListActiveUserIDs(ctx context.Context) ([]string, error)

func (*OIDCStore) ListAgentUserIDs added in v0.38.0 

func (s *OIDCStore) ListAgentUserIDs(ctx context.Context, agentID string) ([]string, error)

func (*OIDCStore) ListChannelIdentitiesByUser added in v0.38.0 

func (s *OIDCStore) ListChannelIdentitiesByUser(ctx context.Context, userID string) ([]auth.ChannelIdentity, error)

func (*OIDCStore) ListLoginIdentitiesByUser added in v0.38.0 

func (s *OIDCStore) ListLoginIdentitiesByUser(ctx context.Context, userID string) ([]auth.LoginIdentity, error)

func (*OIDCStore) ListSessionsByUser added in v0.38.0 

func (s *OIDCStore) ListSessionsByUser(ctx context.Context, userID string) ([]auth.Session, error)

func (*OIDCStore) ListUserAgentIDs added in v0.38.0 

func (s *OIDCStore) ListUserAgentIDs(ctx context.Context, userID string) ([]string, error)

func (*OIDCStore) ListUsers added in v0.38.0 

func (s *OIDCStore) ListUsers(ctx context.Context) ([]auth.User, error)

func (*OIDCStore) ListUsersPaged added in v0.38.0 

func (s *OIDCStore) ListUsersPaged(ctx context.Context, limit, offset int64) ([]auth.User, error)

func (*OIDCStore) RemoveAgent added in v0.38.0 

func (s *OIDCStore) RemoveAgent(ctx context.Context, userID, agentID string) error

func (*OIDCStore) UpdateChannelIdentityExternalID added in v0.38.0 

func (s *OIDCStore) UpdateChannelIdentityExternalID(ctx context.Context, id, externalID string) error

func (*OIDCStore) UpdateCredentialHash added in v0.38.0 

func (s *OIDCStore) UpdateCredentialHash(ctx context.Context, userID, passwordHash string) error

func (*OIDCStore) UpdateLoginIdentity added in v0.38.0 

func (s *OIDCStore) UpdateLoginIdentity(ctx context.Context, i auth.LoginIdentity) error

func (*OIDCStore) UpdateSessionExpiry added in v0.38.0 

func (s *OIDCStore) UpdateSessionExpiry(ctx context.Context, id string, expiresAt time.Time) error

func (*OIDCStore) UpdateUser added in v0.38.0 

func (s *OIDCStore) UpdateUser(ctx context.Context, u auth.User) error

func (*OIDCStore) UpdateUserActive added in v0.38.0 

func (s *OIDCStore) UpdateUserActive(ctx context.Context, userID string, isActive bool) error

func (*OIDCStore) UpdateUserAgeKeys added in v0.38.0 

func (s *OIDCStore) UpdateUserAgeKeys(ctx context.Context, userID, publicKey, privateKey string) error

func (*OIDCStore) UpdateUserDefaultAgent added in v0.38.0 

func (s *OIDCStore) UpdateUserDefaultAgent(ctx context.Context, userID, agentID string) error

func (*OIDCStore) UpdateUserNotifyIdentity added in v0.38.0 

func (s *OIDCStore) UpdateUserNotifyIdentity(ctx context.Context, userID string, identityID *string) error

func (*OIDCStore) UpdateUserRole added in v0.38.0 

func (s *OIDCStore) UpdateUserRole(ctx context.Context, userID string, role string) error

type Option added in v0.50.0 

type Option func(*pgxpool.Config)

Option overrides pool configuration after OpenDB has applied its defaults.

func WithMaxConns added in v0.50.0 

func WithMaxConns(n int32) Option

WithMaxConns caps the pool size. Production uses the built-in default; tests pass a small value so many parallel test databases stay well under the embedded server's max_connections.

Source Files

View all Source files

Directories

Path Synopsis
Package dbtest gives tests an isolated, fully-migrated PostgreSQL database with no external server to run.
 Package dbtest gives tests an isolated, fully-migrated PostgreSQL database with no external server to run.

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.