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

postgres

package
v0.31.0 Latest Latest
Warning

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

 Go to latest Published: Oct 28, 2026 License: Apache-2.0 Imports: 27 Imported by: 0

Documentation

Overview

Package postgres implements filament.DataStore, filament.ScheduleStore and a pipeline store on PostgreSQL via pgx. Queries are generated by sqlc from datastore/postgres/queries/*.sql

Index

Constants

View Source 
const DefaultTenantID = string(filament.DefaultTenantID)

DefaultTenantID is the tenant seeded by migrations and used by POC commands.

Variables

This section is empty.

Functions

func Migrate 

func Migrate(db *sql.DB) error

Migrate applies all pending migrations using the given *sql.DB (a "pgx" stdlib connection — see NewSQLDB). Safe to call on every process start; goose tracks applied versions in its own goose_db_version table.

func NewPool 

func NewPool(ctx context.Context, dsn string) (*pgxpool.Pool, error)

NewPool opens a pgx connection pool for runtime queries.

func NewSQLDB 

func NewSQLDB(dsn string) (*sql.DB, error)

NewSQLDB opens a database/sql handle over the "pgx" driver, for running migrations (goose needs database/sql, not a pgxpool.Pool).

Types

type Store 

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

Store provides ordinary and continuous-stream persistence on PostgreSQL.

func New 

func New(pool *pgxpool.Pool) *Store

New wraps an already-connected pool. Run Migrate before first use.

func (*Store) ActivateStream added in v0.6.0 

func (s *Store) ActivateStream(ctx context.Context, req filament.StreamActivation) (filament.StreamState, error)

ActivateStream snapshots one run's execution inputs and initializes enabled intent. It neither admits a worker nor changes an existing activation's spec.

func (*Store) ClaimDue 

func (s *Store) ClaimDue(ctx context.Context, now time.Time, limit int) ([]filament.ScheduleState, error)

ClaimDue leases up to limit, enabled schedules not currently under an unexpired lease, and stamps them claimed_at = now() in the same transaction as the SELECT ... FOR UPDATE SKIP LOCKED so a second scheduler replica racing this call cannot pick up the same row: it will either block-and-skip (SKIP LOCKED) or see claimed_at within filament.ScheduleLeaseTTL and filter it out. The lease is released by the next SaveSchedule call or expires after the TTL if the scheduler that claimed it crashes before calling SaveSchedule.

func (*Store) ClaimStreamAttempt added in v0.6.0 

func (s *Store) ClaimStreamAttempt(ctx context.Context, lease filament.LeaseToken) error

ClaimStreamAttempt is a durable, one-time claim. Duplicate dispatches cannot run an admitted attempt twice, even after the first worker exits.

func (*Store) Close 

func (s *Store) Close() error

Close releases the underlying connection pool.

func (*Store) CommitEpoch added in v0.6.0 

func (s *Store) CommitEpoch(ctx context.Context, request filament.EpochCommit) (filament.CommittedEpoch, error)

CommitEpoch atomically stores immutable certificate bytes and domain progress. Identical retries return historical success before checking current authority; they never renew a lease. New commits require a bound pipeline completion and a live owner, current membership, sequential epoch, and comparable progress.

func (*Store) ConfigureStreamCodecs added in v0.6.0 

func (s *Store) ConfigureStreamCodecs(codecs filament.CodecResolver)

ConfigureStreamCodecs supplies the codecs used to certify and compare stream progress. Call during boot, before using the store. The resolver and its codecs must remain fixed, pure, deterministic, and concurrency-safe for the store's lifetime. Configuration does not start workers or enable continuous execution.

func (*Store) CreateConnection 

func (s *Store) CreateConnection(ctx context.Context, c filament.Connection) (filament.Connection, error)

CreateConnection stores a new connection at version 1, rejecting a duplicate ID.

func (*Store) CreateNotifier added in v0.3.1 

func (s *Store) CreateNotifier(ctx context.Context, n *ingestionv1.Notifier) (*ingestionv1.Notifier, error)

CreateNotifier adds a rule.

func (*Store) CreatePipeline 

func (s *Store) CreatePipeline(ctx context.Context, p *ingestionv1.Pipeline) (*ingestionv1.Pipeline, error)

CreatePipeline stores a new pipeline.

func (*Store) CreatePipelineVersion 

func (s *Store) CreatePipelineVersion(ctx context.Context, tenant filament.TenantID, pipelineID string, v *ingestionv1.PipelineVersion) (*ingestionv1.PipelineVersion, error)

CreatePipelineVersion appends an immutable graph version to a pipeline.

func (*Store) CreatePipelineWithSchedule 

func (s *Store) CreatePipelineWithSchedule(ctx context.Context, p *ingestionv1.Pipeline, schedule *filament.ScheduleState) (*ingestionv1.Pipeline, error)

CreatePipelineWithSchedule stores a pipeline and optional schedule atomically.

func (*Store) CreateRun 

func (s *Store) CreateRun(ctx context.Context, r filament.RunState) error

CreateRun inserts the run or promotes a pre-created RunScheduled row; a row that has progressed past RunScheduled is left untouched.

func (*Store) CreateRunWithReplicationStream added in v0.0.27 

func (s *Store) CreateRunWithReplicationStream(ctx context.Context, r filament.RunState, desired filament.ReplicationStream) error

CreateRunWithReplicationStream atomically resolves the desired stream and admits the run. A route-fence conflict rolls the stream change back.

func (*Store) DedupSeen 

func (s *Store) DedupSeen(ctx context.Context, tenant string, run filament.RunID, seq uint64) (bool, error)

DedupSeen reports whether seq has already been applied for (tenant, run), advancing the run's high-water mark on the first call to see it. This relies on the tracker processing each run's facts through one sequential pump, so seq only ever increases per run

func (*Store) DeleteConnection 

func (s *Store) DeleteConnection(ctx context.Context, tenant filament.TenantID, id string) error

DeleteConnection soft-deletes the connection with the given ID; deleting a missing ID is a no-op. The name is freed for reuse by the partial unique index.

func (*Store) DeleteNotifier added in v0.3.1 

func (s *Store) DeleteNotifier(ctx context.Context, tenant filament.TenantID, pipelineID, id string) (*ingestionv1.Notifier, error)

DeleteNotifier marks a live rule deleted and returns it for secret cleanup.

func (*Store) DeletePipeline 

func (s *Store) DeletePipeline(ctx context.Context, tenant filament.TenantID, id string) error

DeletePipeline soft-deletes a pipeline and its notifiers, and removes schedules and pending runs. Versions and run history are kept.

func (*Store) DeleteResourceCheckpoint 

func (s *Store) DeleteResourceCheckpoint(ctx context.Context, tenant filament.TenantID, key filament.ResourceCheckpointKey) error

DeleteResourceCheckpoint resets durable progress for one route/resource.

func (*Store) DeleteRun 

func (s *Store) DeleteRun(ctx context.Context, tenant filament.TenantID, id filament.RunID) error

DeleteRun removes the run; resources, checkpoints, and dedup rows cascade. Missing is a no-op.

func (*Store) DeleteSchedule 

func (s *Store) DeleteSchedule(ctx context.Context, tenant filament.TenantID, id filament.ScheduleID) error

DeleteSchedule removes a schedule by id along with its pre-created scheduled runs, so nothing lingers as upcoming work. The runs reap goes first: deleting the schedules row SET-NULLs runs.schedule_id, after which the rows are unreachable by schedule id even inside this transaction.

func (*Store) EndAttempt added in v0.6.0 

func (s *Store) EndAttempt(ctx context.Context, req filament.EndAttemptRequest) error

EndAttempt conditionally records termination for the latest route attempt. Clean means the owner has already joined its work and proven native closure; reaped/unproven declarations never authorize takeover. A retired generation's still-current owner may record closure before its lease expires. Reaped and unproven observations may also be recorded after expiry; neither grants takeover.

func (*Store) EnsureTenant 

func (s *Store) EnsureTenant(ctx context.Context, id filament.TenantID, name string) error

EnsureTenant creates or updates a tenant row.

func (*Store) EnsureUser added in v0.0.27 

func (s *Store) EnsureUser(ctx context.Context, tenant filament.TenantID, externalID string) (filament.UserID, error)

EnsureUser creates or touches the user row for a provider subject within a tenant and returns filament's id for it.

func (*Store) GetEpoch added in v0.6.0 

func (s *Store) GetEpoch(ctx context.Context, req filament.EpochLookup) (filament.CommittedEpoch, error)

GetEpoch resolves a historical commit by tenant and generation without checking live lease authority. A recovered certificate is not permission to acknowledge.

func (*Store) ListConnections 

func (s *Store) ListConnections(ctx context.Context, f filament.ConnectionFilter) ([]filament.Connection, int, error)

ListConnections returns one filtered, sorted page and its pre-page total.

func (*Store) ListNotifiers added in v0.3.1 

func (s *Store) ListNotifiers(ctx context.Context, tenant filament.TenantID, pipelineID string, includeDeleted bool) ([]*ingestionv1.Notifier, error)

ListNotifiers returns one pipeline's rules ordered by ID.

func (*Store) ListPipelineVersions 

func (s *Store) ListPipelineVersions(ctx context.Context, f filament.PipelineVersionFilter) ([]*ingestionv1.PipelineVersion, int, error)

ListPipelineVersions returns all of a pipeline's graph versions, newest first. An unknown pipeline yields an empty slice.

func (*Store) ListPipelines 

func (s *Store) ListPipelines(ctx context.Context, f filament.PipelineFilter) ([]*ingestionv1.Pipeline, int, error)

ListPipelines returns pipelines matching the filter.

func (*Store) ListReconcileCandidates added in v0.6.0 

func (s *Store) ListReconcileCandidates(ctx context.Context, req filament.ReconcileQuery) (filament.ReconcilePage, error)

ListReconcileCandidates returns a bounded tenant-scoped keyset page. Each state is coherent; the page is not a transaction spanning all routes. Include paused and stopped streams so a supervisor can reconcile draining owners.

func (*Store) ListReplicationStreamResources added in v0.0.27 

func (s *Store) ListReplicationStreamResources(ctx context.Context, streamID string) ([]filament.ReplicationStreamResource, error)

ListReplicationStreamResources returns current and historical resource membership for one stream, sorted by resource name.

func (*Store) ListResourceCheckpoints 

func (s *Store) ListResourceCheckpoints(ctx context.Context, tenant filament.TenantID, route filament.ResourceCheckpointRoute) ([]filament.ResourceCheckpointState, error)

ListResourceCheckpoints returns every durable cursor under one route, ordered by resource.

func (*Store) ListResources 

func (s *Store) ListResources(ctx context.Context, tenant filament.TenantID, id filament.RunID) ([]filament.ResourceState, error)

ListResources returns a run's resource states.

func (*Store) ListRetiredReplicationStreams added in v0.0.27 

func (s *Store) ListRetiredReplicationStreams(ctx context.Context, pipelineID, route string) ([]filament.ReplicationStream, error)

ListRetiredReplicationStreams returns predecessor generations still eligible for connector-side cleanup.

func (*Store) ListRuns 

func (s *Store) ListRuns(ctx context.Context, f filament.RunFilter) ([]filament.RunState, int, error)

ListRuns returns runs matching the filter, newest effective time first, each with resource states attached. Effective time is StartedAt, falling back to RequestedAt, ScheduledAt, then CreatedAt: a pending scheduled run sorts by its future fire time (pinning it above finished work) and slides into chronological place once promoted.

func (*Store) ListSchedules 

func (s *Store) ListSchedules(ctx context.Context, f filament.ScheduleFilter) ([]filament.ScheduleState, error)

ListSchedules returns schedules matching the filter.

func (*Store) ListTenants 

func (s *Store) ListTenants(ctx context.Context) ([]Tenant, error)

ListTenants returns all tenant rows ordered by id.

func (*Store) LoadCheckpoint 

func (s *Store) LoadCheckpoint(ctx context.Context, tenant filament.TenantID, id filament.RunID, resource string) (filament.Checkpoint, error)

LoadCheckpoint returns a resource's saved cursor for the run.

func (*Store) LoadConnection 

func (s *Store) LoadConnection(ctx context.Context, tenant filament.TenantID, id string) (filament.Connection, error)

LoadConnection returns the connection with the given ID, including soft-deleted ones so callers can still read a deleted connection's metadata. DeletedAt tells them apart.

func (*Store) LoadNotifier added in v0.3.1 

func (s *Store) LoadNotifier(ctx context.Context, tenant filament.TenantID, pipelineID, id string) (*ingestionv1.Notifier, error)

LoadNotifier returns a rule, including deleted rules for secret cleanup.

func (*Store) LoadPipeline 

func (s *Store) LoadPipeline(ctx context.Context, tenant filament.TenantID, id string) (*ingestionv1.Pipeline, error)

LoadPipeline returns a pipeline by ID, including soft-deleted ones so callers can still read a deleted pipeline's metadata. DeletedAt tells them apart.

func (*Store) LoadPipelineSchedule 

func (s *Store) LoadPipelineSchedule(ctx context.Context, tenant filament.TenantID, pipelineID string) (filament.ScheduleState, error)

LoadPipelineSchedule returns the schedule attached to pipelineID.

func (*Store) LoadPipelineVersion 

func (s *Store) LoadPipelineVersion(ctx context.Context, tenant filament.TenantID, pipelineID string, version int64) (*ingestionv1.PipelineVersion, error)

LoadPipelineVersion returns a specific immutable pipeline graph version.

func (*Store) LoadReplicationStream added in v0.0.27 

func (s *Store) LoadReplicationStream(ctx context.Context, id string) (filament.ReplicationStream, error)

LoadReplicationStream returns one stream generation by ID.

func (*Store) LoadResourceCheckpoint 

func (s *Store) LoadResourceCheckpoint(ctx context.Context, tenant filament.TenantID, key filament.ResourceCheckpointKey) (filament.ResourceCheckpointState, error)

LoadResourceCheckpoint returns durable cross-run progress for one route and resource.

func (*Store) LoadRun 

func (s *Store) LoadRun(ctx context.Context, tenant filament.TenantID, id filament.RunID) (filament.RunState, error)

LoadRun returns the run with its current resource states reattached.

func (*Store) LoadSchedule 

func (s *Store) LoadSchedule(ctx context.Context, tenant filament.TenantID, id filament.ScheduleID) (filament.ScheduleState, error)

LoadSchedule returns one schedule by id.

func (*Store) LoadStreamState added in v0.6.0 

func (s *Store) LoadStreamState(ctx context.Context, req filament.StreamStateRequest) (filament.StreamState, error)

LoadStreamState reads a coherent snapshot under the route and membership locks.

func (*Store) LoadTenant 

func (s *Store) LoadTenant(ctx context.Context, id filament.TenantID) (Tenant, error)

LoadTenant returns a tenant row by id.

func (*Store) MarkReplicationStreamCleaned added in v0.0.27 

func (s *Store) MarkReplicationStreamCleaned(ctx context.Context, id string) error

MarkReplicationStreamCleaned removes a retired generation from subsequent connector cleanup scans while preserving its audit row.

func (*Store) Name 

func (s *Store) Name() string

Name identifies this store implementation.

func (*Store) PendingStreamRuns added in v0.6.0 

func (s *Store) PendingStreamRuns(ctx context.Context, after string, limit int) ([]filament.RunState, error)

PendingStreamRuns is a bounded cross-tenant scan used only by the supervisor.

func (*Store) Ping 

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

Ping verifies database connectivity.

func (*Store) Pool 

func (s *Store) Pool() *pgxpool.Pool

Pool exposes the underlying connection pool for components backed by the same database, e.g. the postgres secrets provider.

func (*Store) ReconcileReplicationStreamResources added in v0.0.27 

func (s *Store) ReconcileReplicationStreamResources(ctx context.Context, streamID string, tenant filament.TenantID, resources []string, bootstrapMode string) ([]filament.ReplicationStreamResource, error)

ReconcileReplicationStreamResources records the complete desired membership of a stream. New and re-added names are pending; omitted names are retired.

func (*Store) ReleaseScheduleClaim 

func (s *Store) ReleaseScheduleClaim(ctx context.Context, tenant filament.TenantID, id filament.ScheduleID) error

ReleaseScheduleClaim makes a claimed occurrence eligible for retry.

func (*Store) RenewLease added in v0.6.0 

func (s *Store) RenewLease(ctx context.Context, lease filament.LeaseToken, ttl time.Duration) error

RenewLease extends a currently live lease without resurrecting expired owners.

func (*Store) ResolveReplicationStream added in v0.0.27 

func (s *Store) ResolveReplicationStream(ctx context.Context, desired filament.ReplicationStream) (filament.ReplicationStream, error)

ResolveReplicationStream reuses the active route generation when its continuity fingerprint matches. Otherwise it retires that generation and creates the caller-supplied stream as the next generation atomically.

func (*Store) ResolveTenant added in v0.0.27 

func (s *Store) ResolveTenant(ctx context.Context, externalID, name string) (filament.TenantID, error)

ResolveTenant maps a provider organization id onto filament's tenant id, minting the row on first sight and refreshing the display name.

func (*Store) RetireUnclaimedAttempt added in v0.6.0 

func (s *Store) RetireUnclaimedAttempt(ctx context.Context, lease filament.LeaseToken) error

RetireUnclaimedAttempt ends an unclaimed dispatch after expiry or disabled intent, then synchronizes the logical run without retiring a claimed worker.

func (*Store) SaveCheckpoint 

func (s *Store) SaveCheckpoint(ctx context.Context, tenant filament.TenantID, id filament.RunID, cp filament.Checkpoint) error

SaveCheckpoint upserts a resource's cursor for the run.

func (*Store) SaveResourceCheckpoint 

func (s *Store) SaveResourceCheckpoint(ctx context.Context, tenant filament.TenantID, state filament.ResourceCheckpointState) error

SaveResourceCheckpoint upserts durable cross-run progress for one immutable pipeline route and resource.

func (*Store) SaveRun 

func (s *Store) SaveRun(ctx context.Context, r filament.RunState) error

SaveRun upserts the run row and reattaches any carried Resources into run_resource_states, mirroring datastore/sqlite's SaveRun contract.

func (*Store) SaveSchedule 

func (s *Store) SaveSchedule(ctx context.Context, st filament.ScheduleState) error

SaveSchedule upserts the schedule row, releasing any ClaimDue lease.

func (*Store) SetDesiredState added in v0.6.0 

func (s *Store) SetDesiredState(ctx context.Context, change filament.DesiredStateChange) error

SetDesiredState persists intent under optimistic revision control. Existing owners may drain after pause/stop, but no new attempt may start until enabled.

func (*Store) SetStreamResourceError added in v0.6.0 

func (s *Store) SetStreamResourceError(ctx context.Context, lease filament.LeaseToken, resource, message string) error

SetStreamResourceError reports startup health without changing stream membership or certified counters. Serialize with epoch commits and attempt replacement.

func (*Store) StartAttempt added in v0.6.0 

func (s *Store) StartAttempt(ctx context.Context, req filament.StartAttemptRequest) (filament.Attempt, error)

StartAttempt reserves immutable execution without dispatching a Job. A retry of an admitted execution ID returns the original attempt without lease renewal. Until destination fencing exists, every predecessor must have ended cleanly.

func (*Store) StreamProgress added in v0.6.0 

func (s *Store) StreamProgress(ctx context.Context, tenant filament.TenantID, run filament.RunID) (int64, int64, time.Time, error)

StreamProgress reads totals maintained atomically by the epoch-insert query. Idempotent CommitEpoch retries never insert again and cannot double-count.

func (*Store) TransitionRun 

func (s *Store) TransitionRun(
	ctx context.Context,
	tenant filament.TenantID,
	id filament.RunID,
	from []filament.RunStatus,
	to filament.RunStatus,
	opts filament.RunTransitionOptions,
) (filament.RunState, error)

TransitionRun serializes lifecycle commands on the run row and updates it only when the current status belongs to from.

func (*Store) UpdateConnection 

func (s *Store) UpdateConnection(ctx context.Context, c filament.Connection) (filament.Connection, error)

UpdateConnection replaces a stored connection, enforcing optimistic version matching.

func (*Store) UpdateNotifier added in v0.3.1 

func (s *Store) UpdateNotifier(ctx context.Context, n *ingestionv1.Notifier) (*ingestionv1.Notifier, error)

UpdateNotifier replaces a live rule's settings.

func (*Store) UpdatePipeline 

func (s *Store) UpdatePipeline(ctx context.Context, p *ingestionv1.Pipeline) (*ingestionv1.Pipeline, error)

UpdatePipeline updates a pipeline's mutable metadata.

func (*Store) UpsertResource 

func (s *Store) UpsertResource(ctx context.Context, rs filament.ResourceState) error

UpsertResource inserts or updates one resource state row.

type Tenant 

type Tenant struct {
	ID        filament.TenantID
	Name      string
	CreatedAt time.Time
	UpdatedAt time.Time
}

Tenant is a control-plane tenant row.

Source Files

View all Source files

Directories

Path Synopsis
Package metrics implements filament.MetricsStore over the same Postgres runs table datastore/postgres.Store persists to.
 Package metrics implements filament.MetricsStore over the same Postgres runs table datastore/postgres.Store persists to.

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.