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
- type Store
- func (s *Store) Close()
- 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) 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) RecordFacts(ctx context.Context, id identity.WorkloadID, facts *innerwallv1.HostFacts, ...) (bool, error)
- func (s *Store) RecordHeartbeat(ctx context.Context, id identity.WorkloadID, dropped, overruns uint64, ...) 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) 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.
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) 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) 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), 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) 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) RecordHeartbeat ¶
func (s *Store) RecordHeartbeat(ctx context.Context, id identity.WorkloadID, dropped, overruns uint64, renewalError string, 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) 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) 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