store

package
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: Jul 22, 2026 License: Apache-2.0 Imports: 13 Imported by: 0

Documentation

Overview

Package store is AgentGuard's durable, multi-tenant persistence tier — the "cold path" behind the in-memory fast path described in docs/v0.6-ARCHITECTURE-PLAN.md (§2.3 write-behind dual-tier).

CONTRACT (see CLAUDE.md):

  • Implementations MUST NOT be called on the streaming proxy hot path. The in-memory maps in pkg/ratelimit, pkg/policy (Engine.sessionCosts) and pkg/proxy (ApprovalQueue) stay authoritative for /v1/check; a Store is reconciled asynchronously (write-behind by the syncer) and read only at boot (hydration) or from operator/query paths.
  • Zero-trust: every persisted row carries a non-empty tenant_id. The mutating batch methods reject any record whose TenantID is empty with ErrTenantRequired; the request-driven read (QueryAudit) takes an explicit tenantID. The all-tenant Load*/Purge* methods are SYSTEM operations (boot hydration / GC), never reached from a per-request handler.

Index

Constants

View Source 
const TenantLocal = "local"

TenantLocal is the canonical id of the default single-tenant deployment. The proxy stores the local tenant as "" on the wire for byte-identity; the syncer normalizes that to TenantLocal before it reaches a Store so every persisted row is attributed.

Variables

View Source 
var ErrTenantRequired = errors.New("store: tenant_id is required")

ErrTenantRequired is returned by a mutating Store method when a record's TenantID is empty. Enforces the zero-trust rule at the lowest layer so a caller that forgets to stamp the tenant fails loudly rather than silently writing an unattributed row. The default tenant is the literal "local" (TenantLocal) — never the empty string at the persistence layer.

Functions

func EffectiveTenant 

func EffectiveTenant(t string) string

EffectiveTenant coerces an empty tenant id to TenantLocal. Callers building Store records from in-memory state (where local may be "") route through this so persisted rows are never unattributed.

func NewAuditLogger 

func NewAuditLogger(s Store) audit.Logger

NewAuditLogger returns an audit.Logger backed by the given Store.

Types

type ApprovalRecord 

type ApprovalRecord struct {
	TenantID   string
	ID         string
	Request    policy.ActionRequest
	Result     policy.CheckResult
	CreatedAt  time.Time
	Resolved   bool
	Decision   string
	ResolvedAt time.Time
	// ConsumedAt persists the one-shot consumption stamp so a restart can
	// never resurrect an already-spent ALLOW as replayable. Zero while
	// unconsumed.
	ConsumedAt time.Time
	// ResolvedVia / ResolvedFrom persist the resolution actor stamp
	// ("bearer" / "session" / "open" + peer host) for incident
	// reconstruction. Empty while pending.
	ResolvedVia  string
	ResolvedFrom string
}

ApprovalRecord is the persisted form of pkg/proxy.PendingAction plus its owning tenant. Kept here (rather than importing the proxy type) to avoid a pkg/store → pkg/proxy import cycle; the syncer maps between the two.

type ApprovalStore 

type ApprovalStore interface {
	// UpsertApprovals writes new/updated approvals (write-behind from
	// ApprovalQueue snapshots). Idempotent on (tenant_id, id). Rejects any
	// record with an empty TenantID.
	//
	// The upsert is a MONOTONIC merge, not last-write-wins — with multiple
	// nodes flushing independent snapshots of the same (tenant_id, id), a
	// stale node's view must never regress shared state another node
	// advanced (the in-memory analog is proxy.ApplyRemote's merge):
	//   - a resolved row is never overwritten by an unresolved one
	//     (resolutions are write-once / terminal);
	//   - a resolved DENY is never overwritten by a non-DENY
	//     (sticky-DENY: the cluster, not just each node, converges to
	//     DENY on a same-ID conflict);
	//   - a non-empty consumed_at is never cleared (a spent one-shot ALLOW
	//     stays spent, so a lagging node's flush cannot make it replayable
	//     again for later hydrates/reconciles).
	UpsertApprovals(ctx context.Context, recs []ApprovalRecord) error
	// LoadApprovals returns every approval (pending and still-retained
	// resolved) across all tenants — boot hydration only.
	LoadApprovals(ctx context.Context) ([]ApprovalRecord, error)
	// PurgeResolvedApprovals deletes resolved approvals resolved before cutoff
	// (durable analogue of the in-memory LRU eviction). Returns rows deleted.
	PurgeResolvedApprovals(ctx context.Context, cutoff time.Time) (int, error)
}

ApprovalStore persists the approval queue (restart-survival + per-tenant pending lists).

type AuditSink 

type AuditSink interface {
	AppendAudit(ctx context.Context, entries []audit.Entry) error
	QueryAudit(ctx context.Context, tenantID string, filter audit.QueryFilter) ([]audit.Entry, error)
}

AuditSink persists the audit trail. AppendAudit is batched write-behind fed by the existing BufferedAsyncLogger workers; QueryAudit backs the tenant-scoped /v1/audit read.

type BucketState 

type BucketState struct {
	TenantID   string
	Key        string
	Tokens     int
	Max        int
	Window     time.Duration
	LastRefill time.Time
}

BucketState is one token-bucket's persisted state. Key is the limiter's opaque "scope:tenant:agent" key; TenantID is the parsed-out tenant so the row is attributable and queryable without re-parsing the key.

type CostConsumption added in v1.0.0 

type CostConsumption struct {
	TenantID  string
	SessionID string
	NodeID    string
	Consumed  float64
	UpdatedAt time.Time
}

CostConsumption is one node's ABSOLUTE cumulative cost reservation for a (tenant, session). Session cost is monotonic within a session, so no window is needed. tenant_id is part of the PK (zero-trust §3).

type CostState 

type CostState struct {
	TenantID    string
	SessionID   string
	Cost        float64
	LastUpdated time.Time
}

CostState is one session's persisted cost accumulator.

type CostStore 

type CostStore interface {
	UpsertCosts(ctx context.Context, costs []CostState) error
	LoadCosts(ctx context.Context) ([]CostState, error)
	// PurgeCosts deletes cost rows last updated before cutoff (durable analogue
	// of Engine.SweepSessionCosts). Returns rows deleted.
	PurgeCosts(ctx context.Context, cutoff time.Time) (int, error)
}

CostStore persists session-cost accumulators.

type PostgresStore added in v1.0.0 

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

PostgresStore is the multi-node Store backend (docs/v0.6-ARCHITECTURE-PLAN.md §2.4 forward-looks to this for the "validated v1.0" Postgres/multi-node target). It mirrors SQLiteStore column-for-column and semantics-for-semantics, adapting only the SQL dialect (positional $N placeholders, native types, BIGSERIAL audit id). Times are stored as RFC3339Nano TEXT and integers/bools via the same helpers SQLiteStore uses, so a row round-trips identically across the two backends and the Purge* string comparisons behave the same.

Concurrency: unlike SQLiteStore (single-file writer pinned to one conn), Postgres handles concurrent sessions, so a small pool is used. This is still a COLD-PATH component — the streaming proxy /v1/check path NEVER touches a Store (CLAUDE.md §1/§2); PostgresStore is reached only by the write-behind syncer, boot hydration, GC, and operator audit queries.

func NewPostgresStore added in v1.0.0 

func NewPostgresStore(dsn string) (*PostgresStore, error)

NewPostgresStore opens the Postgres database identified by dsn (a "postgres://" / "postgresql://" URL or a libpq keyword/value string), sets a modest cold-path connection pool, and runs the schema migration. Connectivity is verified lazily by Migrate's first round-trip, mirroring how NewSQLiteStore surfaces open errors through Migrate.

func (*PostgresStore) AppendAudit added in v1.0.0 

func (s *PostgresStore) AppendAudit(ctx context.Context, entries []audit.Entry) error

func (*PostgresStore) Close added in v1.0.0 

func (s *PostgresStore) Close() error

Close closes the database. Idempotent at the sql.DB layer.

func (*PostgresStore) DSN added in v1.0.0 

func (s *PostgresStore) DSN() string

DSN returns the connection string (for logging / health). Mirrors SQLiteStore.Path().

func (*PostgresStore) DeletePolicy added in v1.0.0 

func (s *PostgresStore) DeletePolicy(ctx context.Context, tenantID string) (bool, error)

DeletePolicy removes a tenant's policy. ok reports whether a row existed.

func (*PostgresStore) GetPolicyYAML added in v1.0.0 

func (s *PostgresStore) GetPolicyYAML(ctx context.Context, tenantID string) ([]byte, bool, error)

GetPolicyYAML returns the stored policy document for a tenant. ok=false when the tenant has no policy row (distinct from an error).

func (*PostgresStore) ListPolicyTenants added in v1.0.0 

func (s *PostgresStore) ListPolicyTenants(ctx context.Context) ([]string, error)

ListPolicyTenants returns every tenant id that has a stored policy, sorted. Used by MultiTenantProvider to eager-load all tenants on boot.

func (*PostgresStore) LoadApprovals added in v1.0.0 

func (s *PostgresStore) LoadApprovals(ctx context.Context) ([]ApprovalRecord, error)

func (*PostgresStore) LoadBuckets added in v1.0.0 

func (s *PostgresStore) LoadBuckets(ctx context.Context) ([]BucketState, error)

func (*PostgresStore) LoadCostConsumption added in v1.0.0 

func (s *PostgresStore) LoadCostConsumption(ctx context.Context) ([]CostConsumption, error)

func (*PostgresStore) LoadCosts added in v1.0.0 

func (s *PostgresStore) LoadCosts(ctx context.Context) ([]CostState, error)

func (*PostgresStore) LoadRateConsumption added in v1.0.0 

func (s *PostgresStore) LoadRateConsumption(ctx context.Context) ([]RateConsumption, error)

func (*PostgresStore) Migrate added in v1.0.0 

func (s *PostgresStore) Migrate(ctx context.Context) error

Migrate creates the schema. Idempotent (every statement is IF NOT EXISTS), so it is safe to run on every boot. Runs inside one transaction so a partial failure leaves the schema untouched.

func (*PostgresStore) Ping added in v1.0.0 

func (s *PostgresStore) Ping(ctx context.Context) error

Ping verifies connectivity.

func (*PostgresStore) PurgeBuckets added in v1.0.0 

func (s *PostgresStore) PurgeBuckets(ctx context.Context, cutoff time.Time) (int, error)

func (*PostgresStore) PurgeCostConsumption added in v1.0.0 

func (s *PostgresStore) PurgeCostConsumption(ctx context.Context, cutoff time.Time) (int, error)

func (*PostgresStore) PurgeCosts added in v1.0.0 

func (s *PostgresStore) PurgeCosts(ctx context.Context, cutoff time.Time) (int, error)

func (*PostgresStore) PurgeRateConsumption added in v1.0.0 

func (s *PostgresStore) PurgeRateConsumption(ctx context.Context, cutoff time.Time) (int, error)

func (*PostgresStore) PurgeResolvedApprovals added in v1.0.0 

func (s *PostgresStore) PurgeResolvedApprovals(ctx context.Context, cutoff time.Time) (int, error)

func (*PostgresStore) PutPolicy added in v1.0.0 

func (s *PostgresStore) PutPolicy(ctx context.Context, tenantID string, policyYAML []byte) error

PutPolicy stores (or replaces) the policy document for a tenant. The bytes are stored verbatim; validation is the caller's responsibility (the CLI and MultiTenantProvider validate before/after). Rejects an empty tenant.

func (*PostgresStore) QueryAudit added in v1.0.0 

func (s *PostgresStore) QueryAudit(ctx context.Context, tenantID string, filter audit.QueryFilter) ([]audit.Entry, error)

func (*PostgresStore) UpsertApprovals added in v1.0.0 

func (s *PostgresStore) UpsertApprovals(ctx context.Context, recs []ApprovalRecord) error

func (*PostgresStore) UpsertBuckets added in v1.0.0 

func (s *PostgresStore) UpsertBuckets(ctx context.Context, buckets []BucketState) error

func (*PostgresStore) UpsertCostConsumption added in v1.0.0 

func (s *PostgresStore) UpsertCostConsumption(ctx context.Context, rows []CostConsumption) error

func (*PostgresStore) UpsertCosts added in v1.0.0 

func (s *PostgresStore) UpsertCosts(ctx context.Context, costs []CostState) error

func (*PostgresStore) UpsertRateConsumption added in v1.0.0 

func (s *PostgresStore) UpsertRateConsumption(ctx context.Context, rows []RateConsumption) error

type RateConsumption added in v1.0.0 

type RateConsumption struct {
	TenantID    string
	Key         string
	WindowEpoch time.Time
	NodeID      string
	Consumed    int
	UpdatedAt   time.Time
}

RateConsumption is one node's ABSOLUTE cumulative consumption of a rate-limit bucket within a fixed window (epoch). Rows are summed per (tenant, key, epoch) across nodes to derive cluster-wide consumption; each node upserts only its own (…, node_id) row, so writes are idempotent last-writer-wins with no read-modify-write race. The tenant_id is part of the PK (zero-trust §3).

type RateLimitStore 

type RateLimitStore interface {
	UpsertBuckets(ctx context.Context, buckets []BucketState) error
	LoadBuckets(ctx context.Context) ([]BucketState, error)
	// PurgeBuckets deletes buckets whose last refill predates cutoff (fully
	// refilled, equivalent to evicted). Returns rows deleted.
	PurgeBuckets(ctx context.Context, cutoff time.Time) (int, error)
}

RateLimitStore persists token buckets so a restart does not reset every limiter to full. Snapshot/restore only — the authoritative check stays in pkg/ratelimit (CLAUDE.md rule #2: async syncs only for rate limits/costs).

type SQLiteStore 

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

SQLiteStore is the single-node, zero-config Store backend (docs/v0.6- ARCHITECTURE-PLAN.md §2.4). It owns one database file holding every persistence table, opened in WAL mode so the eventual direct-reading dashboard can read concurrently with the write-behind syncer.

Concurrency: MaxOpenConns is pinned to 1. The Store is a cold-path component (write-behind flushes + boot hydration + occasional audit queries), so serializing its handle is the simplest correct guard against SQLITE_BUSY and costs nothing on the proxy's request path, which never touches it.

func NewSQLiteStore 

func NewSQLiteStore(path string) (*SQLiteStore, error)

NewSQLiteStore opens (creating if absent) the SQLite database at path, sets WAL + a busy timeout, and runs the schema migration. A ":memory:" path (or any modernc in-memory DSN) yields an ephemeral store, used by tests.

The database file and its -wal/-shm sidecars are held at mode 0600: they carry the full audit trail plus approval/cost/bucket state, and the security brief requires owner-only access (audit 2026-06, M2). The file is pre-created 0600 so SQLite inherits that mode for the sidecars it creates, and existing files from pre-fix versions are tightened on every open.

func (*SQLiteStore) AppendAudit 

func (s *SQLiteStore) AppendAudit(ctx context.Context, entries []audit.Entry) error

func (*SQLiteStore) Close 

func (s *SQLiteStore) Close() error

Close closes the database. Idempotent at the sql.DB layer.

func (*SQLiteStore) DeletePolicy 

func (s *SQLiteStore) DeletePolicy(ctx context.Context, tenantID string) (bool, error)

DeletePolicy removes a tenant's policy. ok reports whether a row existed.

func (*SQLiteStore) GetPolicyYAML 

func (s *SQLiteStore) GetPolicyYAML(ctx context.Context, tenantID string) ([]byte, bool, error)

GetPolicyYAML returns the stored policy document for a tenant. ok=false when the tenant has no policy row (distinct from an error).

func (*SQLiteStore) ListPolicyTenants 

func (s *SQLiteStore) ListPolicyTenants(ctx context.Context) ([]string, error)

ListPolicyTenants returns every tenant id that has a stored policy, sorted. Used by MultiTenantProvider to eager-load all tenants on boot.

func (*SQLiteStore) LoadApprovals 

func (s *SQLiteStore) LoadApprovals(ctx context.Context) ([]ApprovalRecord, error)

func (*SQLiteStore) LoadBuckets 

func (s *SQLiteStore) LoadBuckets(ctx context.Context) ([]BucketState, error)

func (*SQLiteStore) LoadCostConsumption added in v1.0.0 

func (s *SQLiteStore) LoadCostConsumption(ctx context.Context) ([]CostConsumption, error)

func (*SQLiteStore) LoadCosts 

func (s *SQLiteStore) LoadCosts(ctx context.Context) ([]CostState, error)

func (*SQLiteStore) LoadRateConsumption added in v1.0.0 

func (s *SQLiteStore) LoadRateConsumption(ctx context.Context) ([]RateConsumption, error)

func (*SQLiteStore) Migrate 

func (s *SQLiteStore) Migrate(ctx context.Context) error

Migrate creates the schema. Idempotent (CREATE ... IF NOT EXISTS), so it is safe to run on every boot. Additive columns introduced after a table first shipped are applied with guarded ALTERs — CREATE IF NOT EXISTS skips existing tables, so a DB created by an older build would otherwise never gain them.

func (*SQLiteStore) Path 

func (s *SQLiteStore) Path() string

Path returns the database path (for logging / health).

func (*SQLiteStore) Ping 

func (s *SQLiteStore) Ping(ctx context.Context) error

Ping verifies connectivity.

func (*SQLiteStore) PurgeBuckets 

func (s *SQLiteStore) PurgeBuckets(ctx context.Context, cutoff time.Time) (int, error)

func (*SQLiteStore) PurgeCostConsumption added in v1.0.0 

func (s *SQLiteStore) PurgeCostConsumption(ctx context.Context, cutoff time.Time) (int, error)

func (*SQLiteStore) PurgeCosts 

func (s *SQLiteStore) PurgeCosts(ctx context.Context, cutoff time.Time) (int, error)

func (*SQLiteStore) PurgeRateConsumption added in v1.0.0 

func (s *SQLiteStore) PurgeRateConsumption(ctx context.Context, cutoff time.Time) (int, error)

func (*SQLiteStore) PurgeResolvedApprovals 

func (s *SQLiteStore) PurgeResolvedApprovals(ctx context.Context, cutoff time.Time) (int, error)

func (*SQLiteStore) PutPolicy 

func (s *SQLiteStore) PutPolicy(ctx context.Context, tenantID string, policyYAML []byte) error

PutPolicy stores (or replaces) the policy document for a tenant. The bytes are stored verbatim; validation is the caller's responsibility (the CLI and MultiTenantProvider validate before/after). Rejects an empty tenant.

func (*SQLiteStore) QueryAudit 

func (s *SQLiteStore) QueryAudit(ctx context.Context, tenantID string, filter audit.QueryFilter) ([]audit.Entry, error)

func (*SQLiteStore) UpsertApprovals 

func (s *SQLiteStore) UpsertApprovals(ctx context.Context, recs []ApprovalRecord) error

func (*SQLiteStore) UpsertBuckets 

func (s *SQLiteStore) UpsertBuckets(ctx context.Context, buckets []BucketState) error

func (*SQLiteStore) UpsertCostConsumption added in v1.0.0 

func (s *SQLiteStore) UpsertCostConsumption(ctx context.Context, rows []CostConsumption) error

func (*SQLiteStore) UpsertCosts 

func (s *SQLiteStore) UpsertCosts(ctx context.Context, costs []CostState) error

func (*SQLiteStore) UpsertRateConsumption added in v1.0.0 

func (s *SQLiteStore) UpsertRateConsumption(ctx context.Context, rows []RateConsumption) error

type Store 

type Store interface {
	ApprovalStore
	RateLimitStore
	CostStore
	AuditSink

	// Migrate creates/updates the schema. Idempotent; safe on every boot.
	Migrate(ctx context.Context) error
	// Ping verifies connectivity for health checks without touching data.
	Ping(ctx context.Context) error
	// Close releases the underlying handle(s). Idempotent.
	Close() error
}

Store is the composed, durable persistence abstraction. SQLiteStore (and a future PostgresStore) satisfy it.

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.