store

package
v0.4.4 Latest Latest
Warning

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

 Go to latest Published: Aug 2, 2026 License: Apache-2.0 Imports: 12 Imported by: 0

Documentation

Overview

Package store provides typed CRUD over the Wardyn schema using pgx/v5. All writes are serialised through pgxpool; callers supply contexts with deadlines. Most operations are methods on PG (see iface.go); InsertAuditEvent stays a free function taking the pool explicitly since it predates a Store value in the audit.Recorder wiring.

Naming conventions:

  • Create* inserts and returns the full hydrated row.
  • Get* fetches by primary key; returns ErrNotFound when absent.
  • List* returns a slice (empty, never nil) without a hard limit unless stated.
  • Update*/Decide* are point mutations with explicit optimistic guards.

Sandbox ref -> substrate routing rows (migration 0021). This is the Postgres implementation of the orchestrator's RefStore seam: the orchestrator write-throughs each created sandbox's ref and owning-substrate NAME here so a control-plane restart can rehydrate lifecycle routing (Exec/Wait/Attach/ Status/Stop/Kill — i.e. the kill switch) in multi-substrate deployments. Kept out of store.go on purpose (it sits at a lint size boundary).

Index

Constants

This section is empty.

Variables

View Source 
var ErrAlreadyDecided = types.ErrApprovalAlreadyDecided

ErrAlreadyDecided is returned when DecideApproval is called on an approval that has already left the PENDING state. Fail closed: never allow a second decision to silently overwrite the first.

View Source 
var ErrDuplicatePending = types.ErrDuplicatePendingApproval

ErrDuplicatePending is returned by CreateApproval when a partial unique index (approvals_pending_credential_uniq / approvals_pending_noncred_uniq) rejects a second open PENDING approval for the same dedup key — i.e. a concurrent raise lost the race. Callers treat it as a dedup signal (re-read the existing PENDING row and return it), NOT a hard failure. approval.RequestApproval errors.Is-es this exact value; both names alias the one sentinel in internal/types.

View Source 
var ErrNotFound = errors.New("store: not found")

ErrNotFound is returned when a Get* call finds no row.

Functions

func InsertAuditEvent 

func InsertAuditEvent(ctx context.Context, pool *pgxpool.Pool, ev types.AuditEvent) error

InsertAuditEvent appends a single audit event. Implements audit.Recorder. The Postgres trigger blocks UPDATE/DELETE; this function only ever INSERTs.

Types

type AuditFilter added in v0.4.4 

type AuditFilter struct {
	Since        time.Time       // events at or after this instant
	Until        time.Time       // events strictly before this instant
	Action       string          // exact action, e.g. "run.kill"
	ActionPrefix string          // action family, e.g. "credential." or "egress."
	ActorType    types.ActorType // human / agent / system
	Outcome      string          // success / failure / warn
}

AuditFilter narrows the audit feed to answer the questions a bigger window never can — "what happened between 14:00 and 16:00", "every secret.write this quarter", "every failure" — where the answer is 40 events inside 200k. The zero value matches everything, so an unfiltered read keeps taking the plain QueryRecentAuditEventsPage / QueryAuditEventsPage path unchanged.

It renders TWO ways from one declaration — SQL (where) for the Pager path and Go (Matches) for the fetch-all fallback — because a filter applied on only one of them returns silently UNFILTERED events on the other, which in a governance product is a quiet wrong answer. Keep the two in step; they are deliberately adjacent.

Note actor is NOT filterable: for an agent event it IS the per-run SPIFFE ID, so ?actor= would merely restate ?run_id=. ActorType (human/agent/system) is the question that actually has more than one answer today.

func (AuditFilter) IsZero added in v0.4.4 

func (f AuditFilter) IsZero() bool

IsZero reports whether the filter narrows nothing.

func (AuditFilter) Keep added in v0.4.4 

func (f AuditFilter) Keep(events []types.AuditEvent) []types.AuditEvent

Keep returns the events matching f, in order. A zero filter returns events unchanged (no copy).

func (AuditFilter) Matches added in v0.4.4 

func (f AuditFilter) Matches(ev types.AuditEvent) bool

Matches reports whether ev satisfies every set predicate.

type PG 

type PG struct {
	Pool *pgxpool.Pool
}

PG is the Postgres-backed Store: its methods (defined in store.go) hold the query bodies directly, so there is exactly one implementation of each query.

func NewPG 

func NewPG(pool *pgxpool.Pool) PG

NewPG returns a PG Store over pool.

func (PG) ClaimWorkspaceActiveRun 

func (s PG) ClaimWorkspaceActiveRun(ctx context.Context, id, runID uuid.UUID, expected *uuid.UUID) (types.Workspace, bool, error)

ClaimWorkspaceActiveRun compare-and-sets active_run_id from expected (possibly nil) to runID — the atomic serial-import-step gate. Two concurrent step launches that both observed the same free slot cannot both win: the loser gets applied=false and must NOT launch. Returns ErrNotFound only when the workspace does not exist.

func (PG) ClearWorkspaceActiveRun 

func (s PG) ClearWorkspaceActiveRun(ctx context.Context, id, runID uuid.UUID) (bool, error)

ClearWorkspaceActiveRun clears active_run_id ONLY while it still points at runID (conditional, single statement) — a terminal run's cleanup can never clobber a step that was concurrently launched and now owns the pointer.

func (PG) CreateApproval 

func (s PG) CreateApproval(ctx context.Context, a types.ApprovalRequest) (types.ApprovalRequest, error)

CreateApproval inserts a new approval request.

func (PG) CreateGrant 

func (s PG) CreateGrant(ctx context.Context, g types.CredentialGrant) (types.CredentialGrant, error)

CreateGrant inserts a credential grant (eligibility record) and returns it.

func (PG) CreatePolicy 

func (s PG) CreatePolicy(ctx context.Context, p types.RunPolicy) (types.RunPolicy, error)

CreatePolicy inserts a policy and returns the persisted row.

func (PG) CreateRun 

func (s PG) CreateRun(ctx context.Context, r types.AgentRun) (types.AgentRun, error)

CreateRun inserts a new run and returns the persisted row.

func (PG) CreateWorkspace 

func (s PG) CreateWorkspace(ctx context.Context, ws types.Workspace) (types.Workspace, error)

CreateWorkspace inserts an onboarded workspace and returns the persisted row. Profile is internal/workspacescan's opaque WorkspaceProfile blob (nil until scanned).

func (PG) DecideApproval 

func (s PG) DecideApproval(ctx context.Context, id uuid.UUID, state types.ApprovalState, decidedBy, reason string) (types.ApprovalRequest, error)

DecideApproval transitions an approval from PENDING to the given state. Returns ErrAlreadyDecided if the approval is not PENDING (fail-closed). Uses a single UPDATE with WHERE state='PENDING' to prevent TOCTOU races.

func (PG) DeletePolicy 

func (s PG) DeletePolicy(ctx context.Context, id uuid.UUID) error

DeletePolicy removes a policy by id. Returns ErrNotFound when no row matched. Note: agent_runs.policy_id has NO foreign key, so a delete always succeeds even while runs still reference the policy — those runs keep a dangling policy_id. The run's authorization envelope survives regardless: dispatch records the fully-widened spec as a run.policy.effective event in the append-only audit log.

func (PG) DeleteRef added in v0.3.1 

func (s PG) DeleteRef(ctx context.Context, ref string) error

DeleteRef removes the row for ref. Idempotent: deleting a missing ref is nil.

func (PG) DeleteWorkspace 

func (s PG) DeleteWorkspace(ctx context.Context, id uuid.UUID) error

DeleteWorkspace removes a workspace by id. Returns ErrNotFound when no row matched.

func (PG) GetApproval 

func (s PG) GetApproval(ctx context.Context, id uuid.UUID) (types.ApprovalRequest, error)

GetApproval returns the approval for id, or ErrNotFound.

func (PG) GetPolicy 

func (s PG) GetPolicy(ctx context.Context, id uuid.UUID) (types.RunPolicy, error)

GetPolicy returns the policy for id, or ErrNotFound.

func (PG) GetRef added in v0.3.1 

func (s PG) GetRef(ctx context.Context, ref string) (string, bool, error)

GetRef returns the substrate name for ref. A missing row is (found=false, nil error) — pre-migration and unknown refs are expected, not errors.

func (PG) GetRun 

func (s PG) GetRun(ctx context.Context, id uuid.UUID) (types.AgentRun, error)

GetRun returns the run for id, or ErrNotFound.

func (PG) GetSiteConfig 

func (s PG) GetSiteConfig(ctx context.Context) (types.SiteConfig, error)

GetSiteConfig returns the operator-wide site config, or a ZERO-VALUE SiteConfig (not an error) when no row has been written yet — first boot has no config, and "unconfigured" is a valid, common state rather than a failure the caller must special-case.

func (PG) GetWorkspace 

func (s PG) GetWorkspace(ctx context.Context, id uuid.UUID) (types.Workspace, error)

GetWorkspace returns the workspace for id, or ErrNotFound.

func (PG) GetWorkspaceBySource 

func (s PG) GetWorkspaceBySource(ctx context.Context, kind types.WorkspaceKind, source string) (types.Workspace, error)

GetWorkspaceBySource returns the workspace with the given kind+source, or ErrNotFound — the read side of the partial-unique (source) WHERE kind='local_dir' index, and the lookup a repo-kind workspace resolves by.

func (PG) LatestAuditEventByAction 

func (s PG) LatestAuditEventByAction(ctx context.Context, action string) (types.AuditEvent, error)

LatestAuditEventByAction returns the most recent audit event whose action equals the given action, or ErrNotFound when none exists. Used by /healthz to find the latest kernel.sensor.heartbeat that drives the eBPF ground-truth health state (so the stream reports healthy only while beats are arriving).

func (PG) ListApprovals 

func (s PG) ListApprovals(ctx context.Context, stateFilter types.ApprovalState) ([]types.ApprovalRequest, error)

ListApprovals returns approvals filtered by state. Pass empty string to list all.

func (PG) ListApprovalsPage added in v0.3.1 

func (s PG) ListApprovalsPage(ctx context.Context, stateFilter types.ApprovalState, p Page) ([]types.ApprovalRequest, error)

ListApprovalsPage returns approvals filtered by state (empty = all) in reverse request order, bounded by p. The all-state feed rides approvals_requested_at_idx (0020); a single-state filter rides approvals_state_requested_at_idx (0023), which serves both the WHERE and the ORDER BY without a sort.

func (PG) ListGrantsByRun 

func (s PG) ListGrantsByRun(ctx context.Context, runID uuid.UUID) ([]types.CredentialGrant, error)

ListGrantsByRun returns all grants for a run.

func (PG) ListPolicies 

func (s PG) ListPolicies(ctx context.Context) ([]types.RunPolicy, error)

ListPolicies returns all policies in reverse creation order. The slice is empty (never nil) when no policies exist.

func (PG) ListPoliciesPage added in v0.3.1 

func (s PG) ListPoliciesPage(ctx context.Context, p Page) ([]types.RunPolicy, error)

ListPoliciesPage returns policies in reverse creation order, bounded by p. run_policies_created_at_idx (0023) covers the ORDER BY.

func (PG) ListRuns 

func (s PG) ListRuns(ctx context.Context) ([]types.AgentRun, error)

ListRuns returns all runs in reverse creation order (unbounded).

func (PG) ListRunsPage added in v0.3.1 

func (s PG) ListRunsPage(ctx context.Context, p Page) ([]types.AgentRun, error)

ListRunsPage returns runs in reverse creation order, bounded by p. The agent_runs_created_at_idx (0020) makes the ORDER BY + LIMIT an index scan.

func (PG) ListWorkspaces 

func (s PG) ListWorkspaces(ctx context.Context) ([]types.Workspace, error)

ListWorkspaces returns all workspaces in reverse creation order. The slice is empty (never nil) when no workspaces exist.

func (PG) ListWorkspacesPage added in v0.3.1 

func (s PG) ListWorkspacesPage(ctx context.Context, p Page) ([]types.Workspace, error)

ListWorkspacesPage returns workspaces in reverse creation order, bounded by p. workspaces_created_at_idx (0023) covers the ORDER BY.

func (PG) PutRef added in v0.3.1 

func (s PG) PutRef(ctx context.Context, ref, substrateName string) error

PutRef upserts the ref -> substrate-name row (ref is the primary key).

func (PG) PutSiteConfig 

func (s PG) PutSiteConfig(ctx context.Context, cfg types.SiteConfig) (types.SiteConfig, error)

PutSiteConfig upserts the single operator-wide site config row and returns the persisted value. The `singleton` primary key (CHECKed true) makes a second row impossible at the schema level; a write always REPLACES the whole document (no partial merge — the API layer decodes and validates the full document before calling this).

func (PG) QueryAuditEvents 

func (s PG) QueryAuditEvents(ctx context.Context, runID uuid.UUID, limit int) ([]types.AuditEvent, error)

QueryAuditEvents returns audit events for a run in time order. limit <= 0 means no explicit limit (returns up to 1000).

func (PG) QueryAuditEventsFilteredPage added in v0.4.4 

func (s PG) QueryAuditEventsFilteredPage(ctx context.Context, runID *uuid.UUID, f AuditFilter, p Page) ([]types.AuditEvent, error)

QueryAuditEventsFilteredPage returns audit events narrowed by f, bounded by p. A non-nil runID scopes to one run and keeps that trail's chronological seq ASC order (the per-run contract in docs/sdk.md); nil is the newest-first global feed. Only a request that actually sets a predicate takes this path — an unfiltered read stays on the two plain queries above, whose plans are unchanged.

No new index: 0001 covers (time) and 0017 (action, seq DESC); a bounded LIMIT query index-scans the pkey backward and filters, which is what the caps are for.

func (PG) QueryAuditEventsPage added in v0.3.1 

func (s PG) QueryAuditEventsPage(ctx context.Context, runID uuid.UUID, p Page) ([]types.AuditEvent, error)

QueryAuditEventsPage returns a run's audit events in seq (chronological) order, bounded by p. audit_events_run_seq_idx (0023) makes WHERE run_id + ORDER BY seq an indexed range scan with no sort; OFFSET pages forward without flipping to DESC, so the per-run trail stays ASC (docs/sdk.md's exit-code contract) and a caller pages to the newest events with ?offset=.

func (PG) QueryRecentAuditEvents 

func (s PG) QueryRecentAuditEvents(ctx context.Context, limit int) ([]types.AuditEvent, error)

QueryRecentAuditEvents returns the most-recent audit events across ALL runs, newest first — the global SIEM-style feed the Audit view renders. Per-run queries (QueryAuditEvents) stay chronological; this global tail is reverse- chronological and bounded by limit.

func (PG) QueryRecentAuditEventsPage added in v0.3.1 

func (s PG) QueryRecentAuditEventsPage(ctx context.Context, p Page) ([]types.AuditEvent, error)

QueryRecentAuditEventsPage returns the newest-first global audit feed, bounded by p. seq is the audit_events PRIMARY KEY, so ORDER BY seq DESC + LIMIT is an index-scan-backward with no added index (see 0020's audit note).

func (PG) SetRunAgentExecID added in v0.3.1 

func (s PG) SetRunAgentExecID(ctx context.Context, id uuid.UUID, execID string) error

SetRunAgentExecID scoped-writes ONLY the agent_exec_id column. Called once right after the driver execs the agent (the exec id exists only after Exec, so this is a scoped update, not a CreateRun column value). The crash reconciler reads it to observe agent liveness across a restart.

func (PG) SetRunImage added in v0.2.0 

func (s PG) SetRunImage(ctx context.Context, id uuid.UUID, image string) error

SetRunImage scoped-writes ONLY the resolved-image provenance column. Called once after image resolution (the image is resolved after the row is inserted, so this is a scoped update, not a CreateRun column).

func (PG) SetSandboxRef 

func (s PG) SetSandboxRef(ctx context.Context, id uuid.UUID, ref string) error

SetSandboxRef records the runner reference (container ID / pod name).

func (PG) SetWorkspaceApprovedEgress 

func (s PG) SetWorkspaceApprovedEgress(ctx context.Context, id uuid.UUID, domains []string) (types.Workspace, error)

SetWorkspaceApprovedEgress replaces ONLY the operator-owned approved-egress column (plus updated_at), returning the updated row. Scoped on purpose: an approval must never clobber a concurrently-persisted scan (an async repo scan's profile/status land via the full-column UpdateWorkspace, and a read-modify-write here would silently revert them).

func (PG) SetWorkspaceBuiltImage 

func (s PG) SetWorkspaceBuiltImage(ctx context.Context, id uuid.UUID, imageRef, builtHash string) (types.Workspace, error)

SetWorkspaceBuiltImage scoped-writes ONLY the image cache columns (the build-once/reuse-many cache) — the anti-clobber discipline the other scoped writers established; the previous full-row cache write could revert every concurrently-persisted async field from a stale snapshot.

func (PG) SetWorkspaceImportState 

func (s PG) SetWorkspaceImportState(ctx context.Context, id uuid.UUID,
	status types.WorkspaceStatus, activeRunID *uuid.UUID, expectedActive *uuid.UUID,
	verifyResult json.RawMessage, verifiedHash string, verifiedAt *time.Time) (types.Workspace, bool, error)

SetWorkspaceImportState is the scoped writer the import orchestrator uses to advance the pipeline without a full-row read-modify-write: it sets status + the in-flight run pointer, and (when a verify reports) the verify result and proven-working markers. Any nil pointer leaves that column unchanged via COALESCE-on-sentinel is avoided by taking explicit values — callers pass the current values for columns they don't mean to change.

FENCED (mirrors SetWorkspaceScanResult): the write is conditional on the import-step slot still holding expectedActive, so a caller that decided what to write from a STALE read cannot land it. Every caller here does check-then- act (read the workspace, decide, write), and this was the only unfenced workspace writer — a finalize/update racing a live verify/record run could overwrite the fresher state the concurrent run had just written, which is exactly the class of race the C001 finalize guard closed at one call site only. Pass expectedActive = the active_run_id observed in the read the decision came from (nil means "expected no in-flight run"); applied=false means the slot moved under the caller, which must then re-read rather than retry blindly. Returns ErrNotFound only when the workspace does not exist.

func (PG) SetWorkspaceLLMCred added in v0.4.0 

func (s PG) SetWorkspaceLLMCred(ctx context.Context, id uuid.UUID, cred *types.WorkspaceLLMCred) (types.Workspace, error)

SetWorkspaceLLMCred replaces ONLY the operator-owned model/harness cred binding column (plus updated_at), returning the updated row. Scoped like SetWorkspaceApprovedEgress so it can never clobber a concurrently-persisted async scan. Pass nil to clear the binding.

func (PG) SetWorkspaceRecordResult 

func (s PG) SetWorkspaceRecordResult(ctx context.Context, id uuid.UUID,
	taskKey string, result json.RawMessage, onlyIfStatus string) (types.Workspace, bool, error)

SetWorkspaceRecordResult atomically upserts ONE task's entry in the Record Mode record_results map (jsonb || merge — never a whole-map read-modify- write, so concurrent writers of DIFFERENT tasks can never lose each other's entries). When onlyIfStatus is non-empty the write applies only while the task's CURRENT stored status equals it (single-statement compare-and-set): a late streaming upload can never revert a completed capture, and a double capture no-ops. Returns applied=false (no error) on a guard miss.

func (PG) SetWorkspaceScanResult added in v0.3.1 

func (s PG) SetWorkspaceScanResult(ctx context.Context, id uuid.UUID, profile json.RawMessage, runID uuid.UUID) (types.Workspace, bool, error)

SetWorkspaceScanResult records a governed scan run's derived profile — a SCOPED, FENCED write mirroring ClaimWorkspaceActiveRun: it sets ONLY profile + status= scanned and RELEASES the import-step slot, conditional on the run STILL owning it (active_run_id=runID). A superseded / lagging upload (a newer run claimed the slot, or a reconcile released it) matches no row → applied=false, so it can neither clobber a fresher profile nor revert a concurrently-persisted column (approved_egress, setup_commands) the way the old full-row UpdateWorkspace did. Returns ErrNotFound only when the workspace does not exist.

func (PG) SetWorkspaceSetupCommands 

func (s PG) SetWorkspaceSetupCommands(ctx context.Context, id uuid.UUID, cmds json.RawMessage) (types.Workspace, error)

SetWorkspaceSetupCommands replaces ONLY the operator-approved setup-commands column (scoped write, same anti-clobber discipline as approved-egress). The blob is opaque []workspacescan.SetupCommand JSON.

func (PG) TouchRun 

func (s PG) TouchRun(ctx context.Context, id uuid.UUID) error

TouchRun bumps a run's updated_at to now() without changing any other field. It is the activity keepalive the interactive-attach handler calls so the idle reaper (which measures idleness by agent_runs.updated_at) does not stop a run that a human is actively attached to. Returns ErrNotFound when no row matched.

func (PG) UpdatePolicy 

func (s PG) UpdatePolicy(ctx context.Context, id uuid.UUID, name string, spec types.RunPolicySpec) (types.RunPolicy, error)

UpdatePolicy replaces a policy's name and spec and bumps updated_at, returning the persisted row. Returns ErrNotFound when no policy has the given id. The caller is responsible for validating the spec before calling (policies are admin-gated config; the API validates every spec before it reaches the store).

func (PG) UpdateRunStateIf 

func (s PG) UpdateRunStateIf(ctx context.Context, id uuid.UUID, fromState, toState types.RunState) (bool, error)

UpdateRunStateIf conditionally transitions a run from fromState to toState in a single UPDATE ... WHERE id=$ AND state=$from, returning whether the update applied. It is the optimistic guard the completion watcher uses: it only transitions a run that is STILL in fromState (e.g. RUNNING), so a concurrent kill/stop that already moved the run to a terminal state is never clobbered (TOCTOU-safe, like DecideApproval). A false return with a nil error means the run existed but was no longer in fromState (or did not exist) — the caller treats this as "someone else won the transition" and does nothing.

func (PG) UpdateRunStateIfIdle 

func (s PG) UpdateRunStateIfIdle(ctx context.Context, id uuid.UUID, fromState, toState types.RunState, notAfter time.Time) (bool, error)

UpdateRunStateIfIdle is UpdateRunStateIf plus an idleness guard: it transitions a run from fromState to toState ONLY when the row is still in fromState AND its updated_at has NOT advanced past notAfter (the snapshot the caller observed). This closes the reaper's idleness TOCTOU: the idle scan reads updated_at in a snapshot, but an active `wardyn attach` TouchRun (which bumps updated_at while leaving state=RUNNING) can land between snapshot and stop. Guarding only on state=RUNNING would then stop the now-active run, defeating the keepalive. Passing the snapshot's updated_at as notAfter makes a run touched after the snapshot no-op the stop (rows-affected 0 => false), so the reaper leaves it be and retries on the next tick. Returns (true, nil) when the transition applied.

func (PG) UpdateWorkspace 

func (s PG) UpdateWorkspace(ctx context.Context, id uuid.UUID, ws types.Workspace) (types.Workspace, error)

UpdateWorkspace replaces a workspace's editable identity fields (name, kind, source, ref, default_target) and bumps updated_at, returning the persisted row. It is a FULL-column write (it also sets profile, image_ref, built_profile_hash, status and the other scan-owned columns), which is why callers must round-trip the fetched row. Returns ErrNotFound when no workspace has the given id.

handleUpdateWorkspace does round-trip, and resets the scan-owned fields + ApprovedEgress itself when source/kind changed — the persisted profile and egress approvals were reviewed against the OLD source.

type Page added in v0.3.1 

type Page struct {
	Limit  int
	Offset int
}

Page bounds a List query to Limit rows after skipping Offset, ordered by the query's own ORDER BY. A zero or negative Limit means UNBOUNDED — the historical List* behaviour the internal callers depend on (ReconcileOnBoot's stranded-run scan, the create-run workspace-collision scan, the approval fan-out) all need the whole table, so they call the plain List* wrappers below. The public read handlers pass an explicit Limit (capped by api.parseListPage) via the *Page methods so an external client can never pull down an unbounded payload.

type Pager added in v0.3.1 

type Pager interface {
	ListRunsPage(ctx context.Context, p Page) ([]types.AgentRun, error)
	ListPoliciesPage(ctx context.Context, p Page) ([]types.RunPolicy, error)
	ListWorkspacesPage(ctx context.Context, p Page) ([]types.Workspace, error)
	ListApprovalsPage(ctx context.Context, stateFilter types.ApprovalState, p Page) ([]types.ApprovalRequest, error)
	QueryAuditEventsPage(ctx context.Context, runID uuid.UUID, p Page) ([]types.AuditEvent, error)
	QueryRecentAuditEventsPage(ctx context.Context, p Page) ([]types.AuditEvent, error)
	// QueryAuditEventsFilteredPage serves a NARROWED audit read (time range,
	// action, outcome, actor type); see AuditFilter in auditfilter.go. An
	// unfiltered read still uses the two methods above.
	QueryAuditEventsFilteredPage(ctx context.Context, runID *uuid.UUID, f AuditFilter, p Page) ([]types.AuditEvent, error)
}

Pager is the paginated read surface. It is deliberately NOT part of the Store interface: the control plane has many test doubles that embed store.Store and override a handful of methods, and widening Store would silently route their list calls to the embedded nil interface. Handlers type-assert s.cfg.Store to Pager and fall back to the unbounded List* + in-Go windowing when a store (a test fake) does not implement it. Production always uses PG, which does.

type Recorder 

type Recorder struct {
	Pool *pgxpool.Pool
}

Recorder wraps a pool and implements audit.Recorder via InsertAuditEvent. This satisfies the assignment: "internal/store implements audit.Recorder (Record == InsertAuditEvent)".

func (Recorder) Record 

func (rec Recorder) Record(ctx context.Context, ev types.AuditEvent) error

Record appends ev to the append-only audit_events table.

type Store 

type Store interface {
	// AgentRun.
	CreateRun(ctx context.Context, r types.AgentRun) (types.AgentRun, error)
	GetRun(ctx context.Context, id uuid.UUID) (types.AgentRun, error)
	ListRuns(ctx context.Context) ([]types.AgentRun, error)
	UpdateRunStateIf(ctx context.Context, id uuid.UUID, fromState, toState types.RunState) (bool, error)
	UpdateRunStateIfIdle(ctx context.Context, id uuid.UUID, fromState, toState types.RunState, notAfter time.Time) (bool, error)
	SetSandboxRef(ctx context.Context, id uuid.UUID, ref string) error
	SetRunImage(ctx context.Context, id uuid.UUID, image string) error
	SetRunAgentExecID(ctx context.Context, id uuid.UUID, execID string) error
	TouchRun(ctx context.Context, id uuid.UUID) error

	// RunPolicy.
	CreatePolicy(ctx context.Context, p types.RunPolicy) (types.RunPolicy, error)
	GetPolicy(ctx context.Context, id uuid.UUID) (types.RunPolicy, error)
	ListPolicies(ctx context.Context) ([]types.RunPolicy, error)
	UpdatePolicy(ctx context.Context, id uuid.UUID, name string, spec types.RunPolicySpec) (types.RunPolicy, error)
	DeletePolicy(ctx context.Context, id uuid.UUID) error

	// Workspace.
	CreateWorkspace(ctx context.Context, ws types.Workspace) (types.Workspace, error)
	GetWorkspace(ctx context.Context, id uuid.UUID) (types.Workspace, error)
	GetWorkspaceBySource(ctx context.Context, kind types.WorkspaceKind, source string) (types.Workspace, error)
	ListWorkspaces(ctx context.Context) ([]types.Workspace, error)
	UpdateWorkspace(ctx context.Context, id uuid.UUID, ws types.Workspace) (types.Workspace, error)
	SetWorkspaceApprovedEgress(ctx context.Context, id uuid.UUID, domains []string) (types.Workspace, error)
	SetWorkspaceLLMCred(ctx context.Context, id uuid.UUID, cred *types.WorkspaceLLMCred) (types.Workspace, error)
	SetWorkspaceSetupCommands(ctx context.Context, id uuid.UUID, cmds json.RawMessage) (types.Workspace, error)
	SetWorkspaceRecordResult(ctx context.Context, id uuid.UUID, taskKey string, result json.RawMessage, onlyIfStatus string) (types.Workspace, bool, error)
	ClaimWorkspaceActiveRun(ctx context.Context, id, runID uuid.UUID, expected *uuid.UUID) (types.Workspace, bool, error)
	ClearWorkspaceActiveRun(ctx context.Context, id, runID uuid.UUID) (bool, error)
	SetWorkspaceBuiltImage(ctx context.Context, id uuid.UUID, imageRef, builtHash string) (types.Workspace, error)
	// SetWorkspaceImportState advances the import pipeline. FENCED: the write
	// applies only while the import-step slot still holds expectedActive (nil =
	// expected empty); applied=false means it moved and the caller must re-read
	// instead of retrying blindly.
	SetWorkspaceImportState(ctx context.Context, id uuid.UUID, status types.WorkspaceStatus, activeRunID *uuid.UUID, expectedActive *uuid.UUID, verifyResult json.RawMessage, verifiedHash string, verifiedAt *time.Time) (types.Workspace, bool, error)
	SetWorkspaceScanResult(ctx context.Context, id uuid.UUID, profile json.RawMessage, runID uuid.UUID) (types.Workspace, bool, error)
	DeleteWorkspace(ctx context.Context, id uuid.UUID) error

	// CredentialGrant.
	CreateGrant(ctx context.Context, g types.CredentialGrant) (types.CredentialGrant, error)
	ListGrantsByRun(ctx context.Context, runID uuid.UUID) ([]types.CredentialGrant, error)

	// ApprovalRequest.
	CreateApproval(ctx context.Context, a types.ApprovalRequest) (types.ApprovalRequest, error)
	GetApproval(ctx context.Context, id uuid.UUID) (types.ApprovalRequest, error)
	ListApprovals(ctx context.Context, stateFilter types.ApprovalState) ([]types.ApprovalRequest, error)
	DecideApproval(ctx context.Context, id uuid.UUID, state types.ApprovalState, decidedBy, reason string) (types.ApprovalRequest, error)

	// AuditEvent.
	QueryAuditEvents(ctx context.Context, runID uuid.UUID, limit int) ([]types.AuditEvent, error)
	QueryRecentAuditEvents(ctx context.Context, limit int) ([]types.AuditEvent, error)
	LatestAuditEventByAction(ctx context.Context, action string) (types.AuditEvent, error)

	// SiteConfig.
	GetSiteConfig(ctx context.Context) (types.SiteConfig, error)
	PutSiteConfig(ctx context.Context, cfg types.SiteConfig) (types.SiteConfig, error)

	// Sandbox ref -> substrate routing (the orchestrator's RefStore seam; see
	// store_sandbox_ref.go). GetRef reports a missing row as found=false, nil.
	PutRef(ctx context.Context, ref, substrateName string) error
	GetRef(ctx context.Context, ref string) (substrateName string, found bool, err error)
	DeleteRef(ctx context.Context, ref string) error
}

Store is the abstract persistence seam the control plane talks to. The default Postgres implementation is PG, whose methods hold the query bodies directly (no pool param — the receiver carries its own handle); a future pure-Go SQLite backend will satisfy this same interface without touching the API layer.

Out of scope on purpose: the transactional surfaces (broker mint FOR UPDATE, identity revocation) need a real transaction rather than a single-call store and stay on the pool directly.

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.