store

package
v0.3.0 Latest Latest
Warning

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

 Go to latest Published: Jul 15, 2026 License: Apache-2.0 Imports: 10 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.

Index

Constants

This section is empty.

Variables

View Source 
var ErrAlreadyDecided = errors.New("store: approval already decided")

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 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 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 core A'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: a foreign-key reference from agent_runs.policy_id can make this fail at the DB level if runs still reference the policy; the wrapped error surfaces.

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) GetGrant 

func (s PG) GetGrant(ctx context.Context, id uuid.UUID) (types.CredentialGrant, error)

GetGrant returns the grant 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) 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) 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) ListRuns 

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

ListRuns returns all runs in reverse creation order.

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) 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) 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) 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, verifyResult json.RawMessage,
	verifiedHash string, verifiedAt *time.Time) (types.Workspace, 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.

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) 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) UpdateRunState 

func (s PG) UpdateRunState(ctx context.Context, id uuid.UUID, state types.RunState) error

UpdateRunState sets the state and bumps updated_at.

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 does NOT touch the scan-owned fields (profile, image_ref, built_profile_hash, status) — those are exclusively written by the scan flow (core A; handleScanWorkspace is a stub as of this wave). Returns ErrNotFound when no workspace has the given id.

Callers must round-trip the fetched row (handleUpdateWorkspace does, 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 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)
	UpdateRunState(ctx context.Context, id uuid.UUID, state types.RunState) 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
	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)
	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(ctx context.Context, id uuid.UUID, status types.WorkspaceStatus, activeRunID *uuid.UUID, verifyResult json.RawMessage, verifiedHash string, verifiedAt *time.Time) (types.Workspace, error)
	DeleteWorkspace(ctx context.Context, id uuid.UUID) error

	// CredentialGrant.
	CreateGrant(ctx context.Context, g types.CredentialGrant) (types.CredentialGrant, error)
	GetGrant(ctx context.Context, id uuid.UUID) (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)
}

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.