postgres

package
v0.86.0 Latest Latest
Warning

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

 Go to latest Published: Aug 7, 2026 License: EUPL-1.2 Imports: 25 Imported by: 0

Documentation

Overview

Package postgres implements the observed-plane stores on PostgreSQL via pgx. Schema is managed by embedded, versioned SQL migrations applied at startup inside an advisory lock, so concurrent replicas never race.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Migrate 

func Migrate(ctx context.Context, pool *pgxpool.Pool) error

Migrate applies all pending embedded migrations in filename order. The schema_migrations table records what ran.

Types

type Convergence 

type Convergence struct {

	// Tags resolves a group to its device tags from the config snapshot.
	Tags func(group string) []string
	// Now is injectable for tests; nil uses time.Now.
	Now func() time.Time
	// contains filtered or unexported fields
}

Convergence builds a ports.ConvergenceSource for one tenant. tags lists the devices of a ring's group (from the config plane); the aggregate runs in SQL - application code never iterates devices.

func (*Convergence) RingStatus 

func (c *Convergence) RingStatus(ctx context.Context, groups []string, target string) (rollout.RingStatus, error)

RingStatus implements ports.ConvergenceSource with one aggregate query.

func (*Convergence) RingStragglers added in v0.64.1 

func (c *Convergence) RingStragglers(ctx context.Context, groups []string, target string) ([]rollout.Straggler, error)

RingStragglers lists the devices holding a ring under 100%: never seen, off-target, offline on target, or erroring - with a short reason each. Capped: the list informs an operator, it is not an inventory export.

type DeviceSecretStore 

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

DeviceSecretStore implements ports.DeviceSecretStore.

func (*DeviceSecretStore) Get 

func (d *DeviceSecretStore) Get(ctx context.Context, tenant, tag string, kind secret.Kind) ([]byte, secret.Meta, bool, error)

Get returns the sealed ciphertext and metadata for one device+kind.

func (*DeviceSecretStore) List 

func (d *DeviceSecretStore) List(ctx context.Context, tenant, tag string) ([]secret.Meta, error)

List returns a device's secret metadata, never the ciphertext.

func (*DeviceSecretStore) MarkRevealed 

func (d *DeviceSecretStore) MarkRevealed(ctx context.Context, tenant, tag string, kind secret.Kind, revealedBy string, now time.Time) error

MarkRevealed records who read a secret and when.

func (*DeviceSecretStore) Put 

func (d *DeviceSecretStore) Put(ctx context.Context, tenant, tag string, kind secret.Kind, ciphertext []byte, createdBy string, now time.Time) error

Put stores or replaces the sealed secret for one device+kind, clearing the revealed marker so a freshly stored value reads as never-revealed.

type DiagnosticsStore added in v0.79.0 

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

DiagnosticsStore implements ports.DiagnosticsStore.

func (*DiagnosticsStore) Delete added in v0.79.0 

func (d *DiagnosticsStore) Delete(ctx context.Context, tenant, tag string) error

Delete removes the device's bundle.

func (*DiagnosticsStore) Get added in v0.79.0 

func (d *DiagnosticsStore) Get(ctx context.Context, tenant, tag string) ([]byte, ports.DiagnosticsMeta, bool, error)

Get returns the sealed bundle and metadata.

func (*DiagnosticsStore) Meta added in v0.79.0 

func (d *DiagnosticsStore) Meta(ctx context.Context, tenant, tag string) (ports.DiagnosticsMeta, bool, error)

Meta returns only the metadata (the page render never moves the bundle).

func (*DiagnosticsStore) Put added in v0.79.0 

func (d *DiagnosticsStore) Put(ctx context.Context, tenant, tag string, ciphertext []byte, now time.Time) error

Put stores or replaces the device's sealed bundle.

type DiscoveredStore 

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

DiscoveredStore implements ports.DiscoveredStore.

func (*DiscoveredStore) List 

func (d *DiscoveredStore) List(ctx context.Context, tenant, station string) ([]discovery.Discovered, error)

List returns a station's current discovered set, MAC-sorted.

func (*DiscoveredStore) Remove 

func (d *DiscoveredStore) Remove(ctx context.Context, tenant, station, mac string) error

Remove drops one MAC once it has been enrolled.

func (*DiscoveredStore) Report 

func (d *DiscoveredStore) Report(ctx context.Context, tenant, station string, devices []discovery.Discovered, now time.Time) error

Report replaces the station's whole discovered set in one transaction, so a concurrent List never sees a half-applied report and vanished leases are gone atomically.

The DELETE and every INSERT are queued into one pgx.Batch and sent in a single round-trip (SendBatch), rather than one tx.Exec per device. The domain caps a report at up to discovery.MaxBatch (4096) devices; on a WAN station (~50ms RTT) 4096 sequential round-trips would hold this transaction open for minutes, bloating WAL and locks for no benefit - a single flush is exactly as correct and orders of magnitude cheaper.

type ElevationStore added in v0.79.0 

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

ElevationStore implements ports.ElevationStore.

func (*ElevationStore) Get added in v0.79.0 

func (e *ElevationStore) Get(ctx context.Context, tenant, id string) (elevation.Request, bool, error)

Get returns one request by id.

func (*ElevationStore) Pending added in v0.79.0 

func (e *ElevationStore) Pending(ctx context.Context, tenant string) ([]elevation.Request, error)

Pending returns the requests nobody has answered, oldest first: somebody is standing in front of each of them, and the one who has waited longest is closest to giving up.

Undecided, not unexpired. Expiry is the caller's to apply - the row does not change when the window passes, so a store that tried to filter on time here would be guessing at a clock it does not own.

func (*ElevationStore) Put added in v0.79.0 

func (e *ElevationStore) Put(ctx context.Context, tenant string, r elevation.Request) error

Put writes a request or replaces it with its decided form.

type ImageJobStore 

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

ImageJobStore implements ports.ImageJobStore.

func (*ImageJobStore) Delete 

func (j *ImageJobStore) Delete(ctx context.Context, tenant, station, mac string) error

Delete removes a job.

func (*ImageJobStore) Get 

func (j *ImageJobStore) Get(ctx context.Context, tenant, station, mac string) (imaging.Job, bool, error)

Get returns one job by MAC, or false.

func (*ImageJobStore) GetActiveByTag 

func (j *ImageJobStore) GetActiveByTag(ctx context.Context, tenant, tag string) (imaging.Job, bool, error)

GetActiveByTag returns the newest non-terminal job for a device tag.

func (*ImageJobStore) ListByStation 

func (j *ImageJobStore) ListByStation(ctx context.Context, tenant, station string) ([]imaging.Job, error)

ListByStation returns every job for a station, newest first.

func (*ImageJobStore) ListPending 

func (j *ImageJobStore) ListPending(ctx context.Context, tenant, station string) ([]imaging.Job, error)

ListPending returns jobs a station still has work for: not yet terminal.

func (*ImageJobStore) TransitionStatus 

func (j *ImageJobStore) TransitionStatus(ctx context.Context, tenant, station, mac string, from, to imaging.Status, message string, now time.Time) (bool, error)

TransitionStatus atomically moves a job from `from` to `to`: the WHERE clause pins the row's current status, so of two concurrent reports for the same job only the one that still finds `from` in the database matches a row - the loser's UPDATE affects zero rows instead of clobbering the winner's write. This closes the check-then-act race a separate Get + CanTransition + unconditional-write sequence would have at the application layer (there used to be such an unconditional UpdateStatus on this store; it was removed - every write to an existing job's status now goes through this guarded, conditional path). A status change starts a new step, so progress/step reset - otherwise a terminal record (installed/failed) would keep showing the in-progress percentage/label from whatever step it was last ticking.

func (*ImageJobStore) UpdateProgress 

func (j *ImageJobStore) UpdateProgress(ctx context.Context, tenant, station, mac string, progress int, step string, now time.Time) error

UpdateProgress records how far the current step is (0..100) and its label, without changing status. It is the frequent, unguarded display-only tick.

func (*ImageJobStore) Upsert 

func (j *ImageJobStore) Upsert(ctx context.Context, tenant string, job imaging.Job, now time.Time) error

Upsert creates or replaces a job. A fresh dispatch resets status to the job's status (pending) and stamps created; a re-dispatch of the same MAC keeps the original created time.

type Store 

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

Store implements the observed-plane ports on one pgx pool.

func Open 

func Open(ctx context.Context, dsn string) (*Store, error)

Open connects, migrates and returns the store. The pool is sized by the DSN (pool_max_conns) or pgx defaults.

func (*Store) Add 

func (s *Store) Add(ctx context.Context, n notify.Notification) error

Add stores one notification (idempotent on id).

func (*Store) Close 

func (s *Store) Close()

Close releases the pool.

func (*Store) CountPersonalData added in v0.86.0 

func (s *Store) CountPersonalData(ctx context.Context, tenant, subject, username string) (ports.PersonalDataCounts, error)

CountPersonalData reports what is held without removing it. This is the preview an operator sees before confirming, so it must count exactly what ErasePersonalData would remove - the two queries are deliberately written against the same predicates.

func (*Store) DeleteDeviceStatusBefore added in v0.86.0 

func (s *Store) DeleteDeviceStatusBefore(ctx context.Context, tenant string, cutoff time.Time, known map[string]bool) (int, error)

DeleteDeviceStatusBefore removes check-ins for tags the fleet no longer has AND that have been silent since cutoff.

The empty-known guard is not defensive noise: with an empty set every tag looks forgotten, so a fleet document that failed to load would erase the observed plane. The caller refuses that case too; this refuses it again, because the cost of being wrong here is the whole table.

func (*Store) DeleteElevationBefore added in v0.86.0 

func (s *Store) DeleteElevationBefore(ctx context.Context, tenant string, cutoff time.Time) (int, error)

DeleteElevationBefore removes elevation requests created before cutoff.

func (*Store) DeleteForgeIdentity added in v0.85.0 

func (s *Store) DeleteForgeIdentity(ctx context.Context, tenant string) error

DeleteForgeIdentity removes the tenant's forge identity, which returns the deployment to whatever credential is mounted.

func (*Store) DeleteMailConfig 

func (s *Store) DeleteMailConfig(ctx context.Context, tenant string) error

DeleteMailConfig removes the tenant's SMTP config.

func (*Store) DeleteNotificationsBefore added in v0.86.0 

func (s *Store) DeleteNotificationsBefore(ctx context.Context, tenant string, cutoff time.Time) (int, error)

DeleteNotificationsBefore removes notifications older than cutoff, and the read markers that point at them - an orphaned marker is a row nobody can ever reach.

func (*Store) DeleteSeenUsersBefore added in v0.86.0 

func (s *Store) DeleteSeenUsersBefore(ctx context.Context, tenant string, cutoff time.Time) (int, error)

DeleteSeenUsersBefore removes cached operator identities not seen since cutoff. The cache is a convenience - the directory remains the source of truth - so losing a row costs one lookup, not a fact.

func (*Store) DeviceSecrets 

func (s *Store) DeviceSecrets() *DeviceSecretStore

DeviceSecrets exposes the per-device secret store on the pool.

func (*Store) Diagnostics added in v0.79.0 

func (s *Store) Diagnostics() *DiagnosticsStore

Diagnostics exposes the bundle store on the pool.

func (*Store) Discovered 

func (s *Store) Discovered() *DiscoveredStore

Discovered exposes the pre-enrollment store on the pool.

func (*Store) Elevation added in v0.79.0 

func (s *Store) Elevation() *ElevationStore

Elevation exposes the request store on the pool.

func (*Store) EmailForSubject 

func (s *Store) EmailForSubject(ctx context.Context, tenant, subject string) (string, bool, error)

EmailForSubject returns a user's e-mail. A stored empty e-mail counts as "no address", so the caller does not try to mail an empty string.

func (*Store) EmailsForAudience 

func (s *Store) EmailsForAudience(ctx context.Context, tenant, group string) ([]string, error)

EmailsForAudience returns the e-mails of every seen user in a group.

func (*Store) ErasePersonalData added in v0.86.0 

func (s *Store) ErasePersonalData(ctx context.Context, tenant, subject, username string) (ports.PersonalDataCounts, error)

ErasePersonalData removes the rows CountPersonalData counted.

Requests this person DECIDED are deliberately left: they are somebody else's record of who approved their access, and erasing them on this person's request would destroy another data subject's evidence.

func (*Store) Get 

func (s *Store) Get(ctx context.Context, tenant, tag string) (observed.DeviceStatus, bool, error)

Get implements ports.StatusStore.

func (*Store) GetFacts 

func (s *Store) GetFacts(ctx context.Context, tenant, tag string) ([]byte, time.Time, bool, error)

GetFacts implements ports.InventoryStore.

func (*Store) GetForgeIdentity added in v0.85.0 

func (s *Store) GetForgeIdentity(ctx context.Context, tenant string) (forge.Identity, bool, error)

GetForgeIdentity returns the tenant's forge identity, if any.

func (*Store) GetMailConfig 

func (s *Store) GetMailConfig(ctx context.Context, tenant string) (mail.Config, bool, error)

GetMailConfig returns the tenant's SMTP config, if any.

func (*Store) GetPrefs 

func (s *Store) GetPrefs(ctx context.Context, tenant, subject string) (identity.Preferences, bool, error)

GetPrefs returns a user's stored preferences.

func (*Store) ImageJobs 

func (s *Store) ImageJobs() *ImageJobStore

ImageJobs exposes the imaging-execution store on the pool.

func (*Store) LeaderLoop added in v0.79.0 

func (s *Store) LeaderLoop(ctx context.Context, key int64, log *slog.Logger, run func(ctx context.Context))

LeaderLoop runs `run` on whichever replica currently holds the advisory lock keyed by `key`, restarting the election if leadership is lost. It blocks until ctx is cancelled. Stateless workers (reads) do NOT belong here - they run on every replica; only committing workers are gated.

func (*Store) List 

func (s *Store) List(ctx context.Context, tenant string) ([]observed.DeviceStatus, error)

List implements ports.StatusStore.

func (*Store) ListFor 

func (s *Store) ListFor(ctx context.Context, tenant, subject string, memberships []string, limit int) ([]notify.Notification, error)

ListFor returns the newest notifications a reader should see, each carrying its per-reader Read flag.

func (*Store) MarkAllRead 

func (s *Store) MarkAllRead(ctx context.Context, tenant, subject string, memberships []string) error

MarkAllRead marks every notification the reader can see as read.

func (*Store) MarkRead 

func (s *Store) MarkRead(ctx context.Context, tenant, subject, id string) error

MarkRead records that this reader read one notification.

func (*Store) NewConvergence 

func (s *Store) NewConvergence(tenant string, tags func(group string) []string) *Convergence

NewConvergence wires the convergence source.

func (*Store) Ping 

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

Ping implements ports.StatusStore (deep readiness).

func (*Store) PutFacts 

func (s *Store) PutFacts(ctx context.Context, tenant, tag string, facts []byte, now time.Time) error

PutFacts implements ports.InventoryStore.

func (*Store) PutForgeIdentity added in v0.85.0 

func (s *Store) PutForgeIdentity(ctx context.Context, tenant string, id forge.Identity) error

PutForgeIdentity upserts the tenant's forge identity.

func (*Store) PutMailConfig 

func (s *Store) PutMailConfig(ctx context.Context, tenant string, c mail.Config) error

PutMailConfig upserts the tenant's SMTP config.

func (*Store) PutPrefs 

func (s *Store) PutPrefs(ctx context.Context, tenant, subject string, p identity.Preferences, now time.Time) error

PutPrefs upserts a user's preferences.

func (*Store) RecordUser 

func (s *Store) RecordUser(ctx context.Context, tenant, subject, email, name string, groups []string) error

RecordUser upserts what a login revealed. Groups replace the prior set so a membership change is reflected on the next login.

func (*Store) Tokens 

func (s *Store) Tokens() *TokenStore

Tokens exposes the token store on the pool.

func (*Store) UnreadCount 

func (s *Store) UnreadCount(ctx context.Context, tenant, subject string, memberships []string) (int, error)

UnreadCount counts the reader's notifications with no read record.

func (*Store) Upsert 

func (s *Store) Upsert(ctx context.Context, tenant string, c observed.CheckIn, now time.Time) (bool, error)

Upsert implements ports.StatusStore: one write per check-in, keyed (tenant, tag). Empty phase/revision in a check-in keeps the stored value (a light heartbeat never erases richer state).

ackChanged is computed inside the same statement (a CTE reading the pre-write ack, compared against the post-write ack in RETURNING) rather than via a separate SELECT beforehand: a read-then-write here would let two concurrent check-ins for the same tag both observe the same prior ack and both report a change, duplicating a wipe-outcome notification for a security-relevant event. The CTE runs against the pre-statement snapshot (standard Postgres semantics for data-modifying CTEs), so this is race-free.

type TokenStore 

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

TokenStore implements ports.TokenStore.

func (*TokenStore) Delete 

func (t *TokenStore) Delete(ctx context.Context, id string) error

Delete implements ports.TokenStore.

func (*TokenStore) Get 

func (t *TokenStore) Get(ctx context.Context, id string) (token.Token, bool, error)

Get implements ports.TokenStore.

func (*TokenStore) ListByKind 

func (t *TokenStore) ListByKind(ctx context.Context, kind token.Kind) ([]token.Token, error)

ListByKind implements ports.TokenStore.

func (*TokenStore) ListBySubject 

func (t *TokenStore) ListBySubject(ctx context.Context, subject string) ([]token.Token, error)

ListBySubject implements ports.TokenStore.

func (*TokenStore) Put 

func (t *TokenStore) Put(ctx context.Context, tok token.Token) error

Put implements ports.TokenStore.

func (*TokenStore) TouchLastUsed 

func (t *TokenStore) TouchLastUsed(ctx context.Context, id string, at time.Time) error

TouchLastUsed implements ports.TokenStore.

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.