store

package
v0.2.0 Latest Latest
Warning

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

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

Documentation

Overview

Package store is the relational data layer: hand-written SQL in queries/, goose migrations in migrations/, and sqlc-generated Go in db/. There is no ORM, no query builder, and no runtime SQL generation; every production query is in the repository and diffable (ADR-0006). region_id is reserved in the schema from the first migration (ADR-0012).

Index

Constants

View Source 
const PolicyChannel = "innerwall_policy"

PolicyChannel is the notification channel a render announces changed workload policies on. The payload is "<workload-id> <version>". A directive for one workload's stream rides the same channel as "<workload-id> <directive>" (ADR-0018 as amended); a listener that does not know a payload's form skips it.

Variables

This section is empty.

Functions

func Migrate 

func Migrate(ctx context.Context, databaseURL string) error

Migrate applies every pending migration in internal/store/migrations. It is idempotent: a fully migrated database is a no-op. Migrations run through the same goose engine the CLI uses, against the same embedded files that sqlc compiled the queries against. A session-level advisory lock serializes concurrent callers (two replicas starting with --migrate, or test packages sharing one database), so exactly one applies each file.

Types

type Store 

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

Store is the Postgres implementation of the domain persistence interfaces. It is safe for concurrent use.

func Open 

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

Open connects to Postgres and verifies the connection.

func (*Store) Close 

func (s *Store) Close()

Close releases the connection pool.

func (*Store) CountWorkloadsByStatus added in v0.2.0 

func (s *Store) CountWorkloadsByStatus(ctx context.Context) ([]readmodel.StatusCount, error)

CountWorkloadsByStatus implements readmodel.Store in one statement.

func (*Store) CreateAddressGroup 

func (s *Store) CreateAddressGroup(ctx context.Context, g *policy.AddressGroup) error

CreateAddressGroup implements policy.Store.

func (*Store) CreateOperatorToken 

func (s *Store) CreateOperatorToken(ctx context.Context, t operator.Token) error

CreateOperatorToken implements operator.Store.

func (*Store) CreateRuleset 

func (s *Store) CreateRuleset(ctx context.Context, rs *policy.Ruleset) error

CreateRuleset implements policy.Store.

func (*Store) CreateService 

func (s *Store) CreateService(ctx context.Context, svc *policy.Service) error

CreateService implements policy.Store.

func (*Store) CreateSession 

func (s *Store) CreateSession(ctx context.Context, sess operator.Session) error

CreateSession implements operator.Store.

func (*Store) CreateToken 

func (s *Store) CreateToken(ctx context.Context, t enroll.Token) error

CreateToken implements enroll.Store.

func (*Store) CreateWorkload 

func (s *Store) CreateWorkload(ctx context.Context, w enroll.Workload, now time.Time) error

CreateWorkload implements enroll.Store.

func (*Store) DeleteAddressGroup 

func (s *Store) DeleteAddressGroup(ctx context.Context, id uuid.UUID, expect string) error

DeleteAddressGroup implements policy.Store.

func (*Store) DeleteExpiredSessions 

func (s *Store) DeleteExpiredSessions(ctx context.Context, now time.Time) (int64, error)

DeleteExpiredSessions implements operator.Store.

func (*Store) DeleteRuleset 

func (s *Store) DeleteRuleset(ctx context.Context, id uuid.UUID, expect string) error

DeleteRuleset implements policy.Store.

func (*Store) DeleteService 

func (s *Store) DeleteService(ctx context.Context, id uuid.UUID, expect string) error

DeleteService implements policy.Store.

func (*Store) DeleteSession 

func (s *Store) DeleteSession(ctx context.Context, idHash []byte) error

DeleteSession implements operator.Store.

func (*Store) DirectReconnect 

func (s *Store) DirectReconnect(ctx context.Context, id identity.WorkloadID) error

DirectReconnect asks the replica holding the stream of id, if any replica does, to send it a Reconnect directive. It is published on the channel render announcements use and delivered at once. Nothing records it and nothing confirms it: no replica knows which replica holds a stream, so delivery is best-effort by design, and the outcome is observed as the workload's last snapshot instant advancing.

func (*Store) FindOperatorTokenByHash 

func (s *Store) FindOperatorTokenByHash(ctx context.Context, hash []byte) (*operator.Token, error)

FindOperatorTokenByHash implements operator.Store.

func (*Store) FindTokenByHash 

func (s *Store) FindTokenByHash(ctx context.Context, hash []byte) (*enroll.Token, error)

FindTokenByHash implements enroll.Store.

func (*Store) Flows 

func (s *Store) Flows() *flowstore.Postgres

Flows returns the FlowStore over the same database. It is the only path to flow data (ADR-0009).

func (*Store) GetAddressGroup 

func (s *Store) GetAddressGroup(ctx context.Context, id uuid.UUID) (*policy.AddressGroup, error)

GetAddressGroup implements policy.Store.

func (*Store) GetModeChange 

func (s *Store) GetModeChange(ctx context.Context, id uuid.UUID) (*fleet.ModeChangeRecord, error)

GetModeChange returns a recorded mode change with its resolved set, for audit and for tests; there is no operator read of it yet.

func (*Store) GetOperator 

func (s *Store) GetOperator(ctx context.Context) (*operator.Operator, error)

GetOperator implements operator.Store.

func (*Store) GetRuleset 

func (s *Store) GetRuleset(ctx context.Context, id uuid.UUID) (*policy.Ruleset, error)

GetRuleset implements policy.Store.

func (*Store) GetService 

func (s *Store) GetService(ctx context.Context, id uuid.UUID) (*policy.Service, error)

GetService implements policy.Store.

func (*Store) GetSession 

func (s *Store) GetSession(ctx context.Context, idHash []byte) (*operator.Session, error)

GetSession implements operator.Store.

func (*Store) GetWorkload 

func (s *Store) GetWorkload(ctx context.Context, id identity.WorkloadID) (*enroll.Workload, error)

GetWorkload implements enroll.Store.

func (*Store) GetWorkloadPolicy 

func (s *Store) GetWorkloadPolicy(ctx context.Context, id identity.WorkloadID) (*innerwallv1.WorkloadPolicy, error)

GetWorkloadPolicy returns the persisted policy of one workload with its version, or nil when none has been rendered yet.

func (*Store) GetWorkloadRecord 

func (s *Store) GetWorkloadRecord(ctx context.Context, id identity.WorkloadID) (*readmodel.WorkloadRecord, error)

GetWorkloadRecord implements readmodel.Store.

func (*Store) ListAddressGroups 

func (s *Store) ListAddressGroups(ctx context.Context) ([]policy.AddressGroup, error)

ListAddressGroups implements policy.Store.

func (*Store) ListListeningServices 

func (s *Store) ListListeningServices(ctx context.Context, id identity.WorkloadID) ([]registry.ListeningService, error)

ListListeningServices implements registry.Store.

func (*Store) ListOperatorTokens 

func (s *Store) ListOperatorTokens(ctx context.Context) ([]operator.Token, error)

ListOperatorTokens implements operator.Store.

func (*Store) ListRulesets 

func (s *Store) ListRulesets(ctx context.Context) ([]policy.Ruleset, error)

ListRulesets implements policy.Store.

func (*Store) ListServices 

func (s *Store) ListServices(ctx context.Context) ([]policy.Service, error)

ListServices implements policy.Store.

func (*Store) ListTokens 

func (s *Store) ListTokens(ctx context.Context) ([]enroll.Token, error)

ListTokens implements enroll.Store.

func (*Store) ListWorkloadLabelIndex 

func (s *Store) ListWorkloadLabelIndex(ctx context.Context) (map[identity.WorkloadID]map[string]string, error)

ListWorkloadLabelIndex implements readmodel.Store.

func (*Store) ListWorkloadPage 

func (s *Store) ListWorkloadPage(ctx context.Context, q readmodel.WorkloadPageQuery) ([]readmodel.WorkloadRecord, error)

ListWorkloadPage implements readmodel.Store. The page is one statement in fleet order; the page's labels, addresses, and listening services are three more, each over the page's ids.

func (*Store) ListWorkloads 

func (s *Store) ListWorkloads(ctx context.Context) ([]registry.Workload, error)

ListWorkloads implements registry.Store.

func (*Store) ListenPolicyChanges 

func (s *Store) ListenPolicyChanges(ctx context.Context, log *slog.Logger, onReady func(), fn func(compiler.Announcement), reconnect func(identity.WorkloadID)) error

ListenPolicyChanges delivers every render announcement to fn and every reconnect directive to reconnect until ctx ends; a nil reconnect skips directives. It holds one dedicated connection outside the pool; if that connection fails it reconnects with jittered backoff and calls onReady again, because announcements made while it was down are gone and the caller must reconcile against persisted state. A directive made while it was down is gone too, and nothing reconciles it: a directive is a request, not state. onReady is also called once the first LISTEN is in place, so a caller can register streams only after it can hear about their changes.

func (*Store) LoadRenderState 

func (s *Store) LoadRenderState(ctx context.Context) (*compiler.Inputs, map[identity.WorkloadID]*innerwallv1.WorkloadPolicy, error)

LoadRenderState implements fleet.Store: the render inputs and the persisted policies read together in one read-only, repeatable-read transaction, so a dry run sees one consistent state and takes no lock.

func (*Store) LookupWorkload 

func (s *Store) LookupWorkload(ctx context.Context, id identity.WorkloadID) (*registry.Workload, error)

LookupWorkload implements registry.Store.

func (*Store) ModeChangeTx 

func (s *Store) ModeChangeTx(ctx context.Context, fn func(ctx context.Context, tx fleet.ModeChangeTx) error) error

ModeChangeTx implements fleet.Store: one transaction holding the render lock, in which the change is resolved, recorded, applied, and rendered.

func (*Store) RecordAgent 

func (s *Store) RecordAgent(ctx context.Context, id identity.WorkloadID, agent registry.AgentInfo, appliedVersion uint64, now time.Time) error

RecordAgent implements registry.Store.

func (*Store) RecordApplied 

func (s *Store) RecordApplied(ctx context.Context, id identity.WorkloadID, version uint64, state innerwallv1.SyncState, now time.Time) error

RecordApplied implements registry.Store.

func (*Store) RecordApplyFailed 

func (s *Store) RecordApplyFailed(ctx context.Context, id identity.WorkloadID, detail string, now time.Time) error

RecordApplyFailed implements registry.Store.

func (*Store) RecordClockReading added in v0.2.0 

func (s *Store) RecordClockReading(ctx context.Context, id identity.WorkloadID, offset time.Duration, skewed bool, receivedAt time.Time) error

RecordClockReading implements ingest.ClockRecorder: a skewed reading's offset and instant, or the end of a current skew.

func (*Store) RecordFacts 

func (s *Store) RecordFacts(ctx context.Context, id identity.WorkloadID, facts *innerwallv1.HostFacts, now time.Time) (bool, error)

RecordFacts implements registry.Store.

func (*Store) RecordFlowReport added in v0.2.0 

func (s *Store) RecordFlowReport(ctx context.Context, id identity.WorkloadID, receivedAt time.Time) error

RecordFlowReport implements ingest.ReportRecorder: the receipt of one flow report, never moved backwards.

func (*Store) RecordHeartbeat 

func (s *Store) RecordHeartbeat(ctx context.Context, id identity.WorkloadID, hb registry.Heartbeat, now time.Time) error

RecordHeartbeat implements registry.Store.

func (*Store) RecordListeningServices 

func (s *Store) RecordListeningServices(ctx context.Context, id identity.WorkloadID, services []registry.ListeningService, now time.Time) error

RecordListeningServices implements registry.Store.

func (*Store) RecordOperatorTokenUse 

func (s *Store) RecordOperatorTokenUse(ctx context.Context, id uuid.UUID, now time.Time) error

RecordOperatorTokenUse implements operator.Store.

func (*Store) RecordRenewal 

func (s *Store) RecordRenewal(ctx context.Context, id identity.WorkloadID, serial string, expiresAt, now time.Time, late *enroll.LateRenewal) error

RecordRenewal implements enroll.Store.

func (*Store) RecordSnapshotSent 

func (s *Store) RecordSnapshotSent(ctx context.Context, id identity.WorkloadID, at time.Time) error

RecordSnapshotSent implements registry.Store.

func (*Store) RenderTx 

func (s *Store) RenderTx(ctx context.Context, fn func(ctx context.Context, tx compiler.Tx) error) error

RenderTx implements compiler.Store.

func (*Store) ReplaceWorkloadLabels 

func (s *Store) ReplaceWorkloadLabels(ctx context.Context, id identity.WorkloadID, labels []registry.Label, expect string) error

ReplaceWorkloadLabels implements fleet.Store: the labels are replaced only when they still hold the version the caller read (empty for an unconditional write), judged inside the transaction on the rows as they stand.

func (*Store) RevokeOperatorToken 

func (s *Store) RevokeOperatorToken(ctx context.Context, id uuid.UUID, now time.Time) error

RevokeOperatorToken implements operator.Store.

func (*Store) RevokeToken 

func (s *Store) RevokeToken(ctx context.Context, id uuid.UUID, now time.Time) error

RevokeToken implements enroll.Store.

func (*Store) SetOperator 

func (s *Store) SetOperator(ctx context.Context, op operator.Operator) error

SetOperator implements operator.Store. The table holds one row; a second call replaces the credential and display name in place.

func (*Store) SetSyncState 

func (s *Store) SetSyncState(ctx context.Context, id identity.WorkloadID, state innerwallv1.SyncState, detail string, now time.Time) error

SetSyncState implements registry.Store.

func (*Store) SetWorkloadLabels 

func (s *Store) SetWorkloadLabels(ctx context.Context, id identity.WorkloadID, labels []registry.Label) error

SetWorkloadLabels implements registry.Store.

func (*Store) SetWorkloadMode 

func (s *Store) SetWorkloadMode(ctx context.Context, id identity.WorkloadID, mode innerwallv1.EnforcementMode) error

SetWorkloadMode implements registry.Store.

func (*Store) UpdateAddressGroup 

func (s *Store) UpdateAddressGroup(ctx context.Context, g *policy.AddressGroup, expect string) error

UpdateAddressGroup implements policy.Store.

func (*Store) UpdateRuleset 

func (s *Store) UpdateRuleset(ctx context.Context, rs *policy.Ruleset, expect string) error

UpdateRuleset implements policy.Store.

func (*Store) UpdateService 

func (s *Store) UpdateService(ctx context.Context, svc *policy.Service, expect string) error

UpdateService implements policy.Store.

Source Files

View all Source files

Directories

Path Synopsis

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.