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
- func Migrate(ctx context.Context, databaseURL string) error
- func MigrateFrom(ctx context.Context, databaseURL string, sources []fs.FS) error
- func Migrations() fs.FS
- type Store
- func (s *Store) Close()
- func (s *Store) CountWorkloadsByStatus(ctx context.Context) ([]readmodel.StatusCount, error)
- func (s *Store) CreateAddressGroup(ctx context.Context, g *policy.AddressGroup) error
- func (s *Store) CreateOperatorToken(ctx context.Context, t operator.Token) error
- func (s *Store) CreateRuleset(ctx context.Context, rs *policy.Ruleset) error
- func (s *Store) CreateService(ctx context.Context, svc *policy.Service) error
- func (s *Store) CreateSession(ctx context.Context, sess operator.Session) error
- func (s *Store) CreateToken(ctx context.Context, t enroll.Token) error
- func (s *Store) CreateWorkload(ctx context.Context, w enroll.Workload, now time.Time) error
- func (s *Store) DeleteAddressGroup(ctx context.Context, id uuid.UUID, expect string) error
- func (s *Store) DeleteExpiredSessions(ctx context.Context, now time.Time) (int64, error)
- func (s *Store) DeleteRuleset(ctx context.Context, id uuid.UUID, expect string) error
- func (s *Store) DeleteService(ctx context.Context, id uuid.UUID, expect string) error
- func (s *Store) DeleteSession(ctx context.Context, idHash []byte) error
- func (s *Store) DirectEnd(ctx context.Context, id identity.WorkloadID) error
- func (s *Store) DirectReconnect(ctx context.Context, id identity.WorkloadID) error
- func (s *Store) FindOperatorTokenByHash(ctx context.Context, hash []byte) (*operator.Token, error)
- func (s *Store) FindTokenByHash(ctx context.Context, hash []byte) (*enroll.Token, error)
- func (s *Store) Flows() *flowstore.Postgres
- func (s *Store) GetAddressGroup(ctx context.Context, id uuid.UUID) (*policy.AddressGroup, error)
- func (s *Store) GetModeChange(ctx context.Context, id uuid.UUID) (*fleet.ModeChangeRecord, error)
- func (s *Store) GetOperator(ctx context.Context) (*operator.Operator, error)
- func (s *Store) GetRuleset(ctx context.Context, id uuid.UUID) (*policy.Ruleset, error)
- func (s *Store) GetService(ctx context.Context, id uuid.UUID) (*policy.Service, error)
- func (s *Store) GetSession(ctx context.Context, idHash []byte) (*operator.Session, error)
- func (s *Store) GetWorkload(ctx context.Context, id identity.WorkloadID) (*enroll.Workload, error)
- func (s *Store) GetWorkloadPolicy(ctx context.Context, id identity.WorkloadID) (*innerwallv1.WorkloadPolicy, error)
- func (s *Store) GetWorkloadRecord(ctx context.Context, id identity.WorkloadID) (*readmodel.WorkloadRecord, error)
- func (s *Store) ListAddressGroups(ctx context.Context) ([]policy.AddressGroup, error)
- func (s *Store) ListListeningServices(ctx context.Context, id identity.WorkloadID) ([]registry.ListeningService, error)
- func (s *Store) ListOperatorTokens(ctx context.Context) ([]operator.Token, error)
- func (s *Store) ListRulesets(ctx context.Context) ([]policy.Ruleset, error)
- func (s *Store) ListServices(ctx context.Context) ([]policy.Service, error)
- func (s *Store) ListTokens(ctx context.Context) ([]enroll.Token, error)
- func (s *Store) ListWorkloadLabelIndex(ctx context.Context) (map[identity.WorkloadID]map[string]string, error)
- func (s *Store) ListWorkloadPage(ctx context.Context, q readmodel.WorkloadPageQuery) ([]readmodel.WorkloadRecord, error)
- func (s *Store) ListWorkloads(ctx context.Context) ([]registry.Workload, error)
- func (s *Store) ListenPolicyChanges(ctx context.Context, log *slog.Logger, onReady func(), ...) error
- func (s *Store) LoadRenderState(ctx context.Context) (*compiler.Inputs, map[identity.WorkloadID]*innerwallv1.WorkloadPolicy, error)
- func (s *Store) LookupWorkload(ctx context.Context, id identity.WorkloadID) (*registry.Workload, error)
- func (s *Store) ModeChangeTx(ctx context.Context, fn func(ctx context.Context, tx fleet.ModeChangeTx) error) error
- func (s *Store) RecordAgent(ctx context.Context, id identity.WorkloadID, agent registry.AgentInfo, ...) error
- func (s *Store) RecordApplied(ctx context.Context, id identity.WorkloadID, version uint64, ...) error
- func (s *Store) RecordApplyFailed(ctx context.Context, id identity.WorkloadID, detail string, now time.Time) error
- func (s *Store) RecordClockReading(ctx context.Context, id identity.WorkloadID, offset time.Duration, skewed bool, ...) error
- func (s *Store) RecordFacts(ctx context.Context, id identity.WorkloadID, facts *innerwallv1.HostFacts, ...) (bool, error)
- func (s *Store) RecordFlowReport(ctx context.Context, id identity.WorkloadID, receivedAt time.Time) (prev *time.Time, proc *registry.AgentProcess, advanced bool, err error)
- func (s *Store) RecordHeartbeat(ctx context.Context, id identity.WorkloadID, hb registry.Heartbeat, ...) error
- func (s *Store) RecordListeningServices(ctx context.Context, id identity.WorkloadID, ...) error
- func (s *Store) RecordOperatorTokenUse(ctx context.Context, id uuid.UUID, now time.Time) error
- func (s *Store) RecordRenewal(ctx context.Context, id identity.WorkloadID, serial string, ...) error
- func (s *Store) RecordSnapshotSent(ctx context.Context, id identity.WorkloadID, at time.Time) error
- func (s *Store) RenderTx(ctx context.Context, fn func(ctx context.Context, tx compiler.Tx) error) error
- func (s *Store) ReplaceWorkloadLabels(ctx context.Context, id identity.WorkloadID, labels []registry.Label, ...) error
- func (s *Store) RetireTx(ctx context.Context, fn func(ctx context.Context, tx fleet.RetireTx) error) error
- func (s *Store) RevokeOperatorToken(ctx context.Context, id uuid.UUID, now time.Time) error
- func (s *Store) RevokeToken(ctx context.Context, id uuid.UUID, now time.Time) error
- func (s *Store) SetOperator(ctx context.Context, op operator.Operator) error
- func (s *Store) SetSyncState(ctx context.Context, id identity.WorkloadID, state innerwallv1.SyncState, ...) error
- func (s *Store) SetWorkloadLabels(ctx context.Context, id identity.WorkloadID, labels []registry.Label) error
- func (s *Store) SetWorkloadMode(ctx context.Context, id identity.WorkloadID, mode innerwallv1.EnforcementMode) error
- func (s *Store) UpdateAddressGroup(ctx context.Context, g *policy.AddressGroup, expect string) error
- func (s *Store) UpdateRuleset(ctx context.Context, rs *policy.Ruleset, expect string) error
- func (s *Store) UpdateService(ctx context.Context, svc *policy.Service, expect string) error
Constants ¶
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 ¶
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.
func MigrateFrom ¶ added in v0.5.0
MigrateFrom applies every pending migration found in sources, read as one set: each source holds goose files at its root, and the set is their union by file name, applied in version order as Migrate applies one. A file name present in more than one source is refused rather than shadowed, since which copy ran would otherwise depend on argument order. Migrate is MigrateFrom over the embedded set alone.
func Migrations ¶ added in v0.5.0
Migrations is the embedded migration set, its files at the root.
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 (*Store) CountWorkloadsByStatus ¶ added in v0.2.0
CountWorkloadsByStatus implements readmodel.Store in one statement.
func (*Store) CreateAddressGroup ¶
CreateAddressGroup implements policy.Store.
func (*Store) CreateOperatorToken ¶
CreateOperatorToken implements operator.Store.
func (*Store) CreateRuleset ¶
CreateRuleset implements policy.Store.
func (*Store) CreateService ¶
CreateService implements policy.Store.
func (*Store) CreateSession ¶
CreateSession implements operator.Store.
func (*Store) CreateToken ¶
CreateToken implements enroll.Store.
func (*Store) CreateWorkload ¶
CreateWorkload implements enroll.Store.
func (*Store) DeleteAddressGroup ¶
DeleteAddressGroup implements policy.Store.
func (*Store) DeleteExpiredSessions ¶
DeleteExpiredSessions implements operator.Store.
func (*Store) DeleteRuleset ¶
DeleteRuleset implements policy.Store.
func (*Store) DeleteService ¶
DeleteService implements policy.Store.
func (*Store) DeleteSession ¶
DeleteSession implements operator.Store.
func (*Store) DirectEnd ¶ added in v0.5.0
DirectEnd asks the replica holding the stream of id, if any replica does, to end it: the workload was retired (ADR-0023). Called inside the retirement's transaction, which Postgres delivers notifications of only on commit, so no replica ends a stream for a retirement that rolled back. Best-effort like every directive; the stream's next heartbeat write is the bound that does not depend on it.
func (*Store) DirectReconnect ¶
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 ¶
FindOperatorTokenByHash implements operator.Store.
func (*Store) FindTokenByHash ¶
FindTokenByHash implements enroll.Store.
func (*Store) Flows ¶
Flows returns the FlowStore over the same database. It is the only path to flow data (ADR-0009).
func (*Store) GetAddressGroup ¶
GetAddressGroup implements policy.Store.
func (*Store) GetModeChange ¶
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 ¶
GetOperator implements operator.Store.
func (*Store) GetRuleset ¶
GetRuleset implements policy.Store.
func (*Store) GetService ¶
GetService implements policy.Store.
func (*Store) GetSession ¶
GetSession implements operator.Store.
func (*Store) GetWorkload ¶
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 ¶
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 ¶
ListOperatorTokens implements operator.Store.
func (*Store) ListRulesets ¶
ListRulesets implements policy.Store.
func (*Store) ListServices ¶
ListServices implements policy.Store.
func (*Store) ListTokens ¶
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 ¶
ListWorkloads implements registry.Store.
func (*Store) ListenPolicyChanges ¶
func (s *Store) ListenPolicyChanges(ctx context.Context, log *slog.Logger, onReady func(), fn func(compiler.Announcement), directed func(compiler.Directive)) error
ListenPolicyChanges delivers every render announcement to fn and every directive to directed until ctx ends; a nil directed 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, presented *registry.PresentedCredential, 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) (prev *time.Time, proc *registry.AgentProcess, advanced bool, err error)
RecordFlowReport implements ingest.ReportRecorder: the receipt of one flow report, never moved backwards. It returns the receipt it replaced, nil for the first ever, with the agent process the latest heartbeat proved running (nil before any heartbeat recorded one), and advanced false when this one was stale and replaced nothing. The statement locks the workload's row, so of two concurrent reports one sees the other's receipt as prev, never the one before: a silence between consecutive receipts is derived exactly once, against the heartbeat pair as it stood when the lock was taken.
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 ¶
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 ¶
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) RetireTx ¶ added in v0.5.0
func (s *Store) RetireTx(ctx context.Context, fn func(ctx context.Context, tx fleet.RetireTx) error) error
RetireTx implements fleet.Store: one transaction holding the render lock, in which a retirement locks the workload's row, stamps it, writes the silence it closes through the flow store bound to the same transaction, renders, and publishes the directive that ends the workload's stream, delivered on commit (ADR-0023). The render lock is taken before the row lock, and nothing takes them in the other order.
func (*Store) RevokeOperatorToken ¶
RevokeOperatorToken implements operator.Store.
func (*Store) RevokeToken ¶
RevokeToken implements enroll.Store.
func (*Store) SetOperator ¶
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 ¶
UpdateRuleset implements policy.Store.
Source Files
Directories