postgres

package
v0.0.0-...-918c0af Latest Latest
Warning

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

 Go to latest Published: Aug 7, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Overview

Package postgres bulk insert helpers (P3-PERF-01a).

These methods implement the Put*Bulk / Append*Bulk additions on the Store interface. Each one builds a chunked multi-row `INSERT ... VALUES (...),(...) ON CONFLICT ...` statement so the control-plane batch writer can flush a full buffer in a single round-trip instead of N individual INSERTs.

pgx.CopyFrom is not used here because the project talks to Postgres through database/sql + pgx/v5/stdlib, not through pgxpool, and CopyFrom requires a native pgx connection. Multi-row INSERT is the next-best option and still delivers an order-of-magnitude speedup over per-row Exec.

Chunking: Postgres allows up to 65535 bind parameters per query. We chunk at 250 rows — the widest row (server_load, 27 columns) uses 250 * 27 = 6750 params, well under the 65535 cap. 250 was picked after the P3-PERF-01b chunk-size sweep: per-row throughput peaks around 100-250 rows and regresses at 500+ because the generated SQL and argument slice both grow super-linearly with chunk size. Every bulk method runs inside a single transaction so partial failure rolls the whole batch back.

Package postgres telemetry bulk helpers (P6-6.1a, audit #10).

Multi-row/one-tx variants of the per-agent telemetry writers, mirroring sqlite/bulk_telemetry.go. Dedup helpers are duplicated here because the two backend packages share no internal path (same as dedupAgents).

internal/controlplane/storage/postgres/clients_repository.go

clients.Repository implementation backed by Postgres via dbsqlc. This is the authoritative persistence layer for the clients domain; it replaces the ad-hoc raw-SQL methods on Store for new callers.

internal/controlplane/storage/postgres/discovered_repository.go

discovered.Repository implementation backed by Postgres via dbsqlc. This is the authoritative persistence layer for the discovered domain; it replaces ad-hoc raw-SQL methods on Store for new callers.

Package postgres hosts the PostgreSQL-backed storage.Store implementation. This file owns schema management — it delegates entirely to goose, which discovers versioned .sql migrations from an embedded FS and records applied versions in the goose_db_version table. Historically this package contained a hand-rolled Migrate() with a single big initialSchema string plus a handful of idempotent ALTERs; that approach left no audit trail of which migrations had run (see DF-20 / M-F8 in the security review).

internal/controlplane/storage/postgres/reposet.go

txRepoSet wires the domain repositories to a single *sql.Tx so that all Repository calls inside a UnitOfWork.Do belong to the same transaction.

internal/controlplane/storage/postgres/uow.go

Postgres-backed UnitOfWork. The retry and rollback contract mirrors (*Store).Transact in tx.go — retryable conflicts (serialization_failure 40001 / deadlock_detected 40P01, see isRetryableTxError) are retried up to maxTransactRetries times with jittered backoff; panics inside fn cause rollback + re-raise; nested Do returns storage.ErrNestedTransact.

Index

Constants

View Source 
const (
	EnvMaxOpenConns    = "PANVEX_DB_MAX_OPEN_CONNS"
	EnvMaxIdleConns    = "PANVEX_DB_MAX_IDLE_CONNS"
	EnvConnMaxLifetime = "PANVEX_DB_CONN_MAX_LIFETIME"
	EnvConnMaxIdleTime = "PANVEX_DB_CONN_MAX_IDLE_TIME"
)

Env var names for tuning the database/sql connection pool. Defaults below were sized to support ~50 concurrent agents on a single CP replica without hitting `connection pool exhausted`. See docs/REMEDIATION_PLAN.md §0.7.

Variables

View Source 
var (
	// ErrDSNRequired reports a missing PostgreSQL connection string.
	ErrDSNRequired = errors.New("postgres dsn is required")
)

Functions

func Migrate 

func Migrate(db *sql.DB) error

Migrate brings the database schema up to the latest embedded migration. Safe to call repeatedly: goose skips versions already recorded in goose_db_version.

func MigrateContext 

func MigrateContext(ctx context.Context, db *sql.DB) error

MigrateContext is the context-aware variant of Migrate.

func NewClientsRepository 

func NewClientsRepository(db dbsqlc.DBTX) clients.Repository

NewClientsRepository wires a clients.Repository against a Postgres connection or transaction. db may be *sql.DB (pool) or *sql.Tx.

func NewDiscoveredRepository 

func NewDiscoveredRepository(db dbsqlc.DBTX) discovered.Repository

NewDiscoveredRepository wires a discovered.Repository against a Postgres connection or transaction. db may be *sql.DB (pool) or *sql.Tx.

func NewUoW 

func NewUoW(db *sql.DB) uow.UnitOfWork

NewUoW constructs a Postgres-backed UnitOfWork bound to the given pool. Pass store.DB() for production wiring.

func Status 

func Status(ctx context.Context, db *sql.DB) error

Status writes the applied/pending migration list to stdout via goose's default logger. The operator invokes this through the `migrate-schema status` subcommand on the control-plane binary.

Types

type PoolConfig 

type PoolConfig struct {
	MaxOpenConns    int
	MaxIdleConns    int
	ConnMaxLifetime time.Duration
	ConnMaxIdleTime time.Duration
}

PoolConfig captures the four knobs database/sql exposes for connection pool sizing. Zero values are not valid: an unset or empty env var falls back to the package defaults via loadPoolConfigFromEnv.

type Store 

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

Store persists control-plane records in a PostgreSQL database.

Store methods reference s.db via the dbExecutor interface so the same method bodies can run against a *sql.DB (outside Transact) or a *sql.Tx (inside Transact). Every domain method MUST go through s.db — the storagetest Transact contract exercises one representative per domain on the tx-bound store. s.sqlDB is the pool handle reserved for lifecycle and pool-only concerns (Ping, Close, PoolStats, Queries, DB, BeginTx in Transact/execInTx); it is nil on transaction-bound Stores to prevent accidental escape from the transaction boundary.

func Open 

func Open(dsn string) (*Store, error)

Open opens a PostgreSQL connection, applies the schema, and returns a storage backend.

Open uses context.Background() for migrations and the initial Ping; callers that need cancellation during startup should use OpenContext instead.

func OpenContext 

func OpenContext(ctx context.Context, dsn string) (*Store, error)

OpenContext is the context-aware variant of Open. It threads ctx through schema migration and the initial connectivity check so startup work can be cancelled by the caller.

func (*Store) ActiveConfigApplyBatchForGroup 

func (s *Store) ActiveConfigApplyBatchForGroup(ctx context.Context, fleetGroupID string) (storage.ConfigApplyBatchRecord, bool, error)

ActiveConfigApplyBatchForGroup returns the running batch for a fleet group, if any. The bool is false (with a zero-value record) when the group has no batch in storage.ConfigApplyBatchStatusRunning.

func (*Store) AggregateClientIPHistory 

func (s *Store) AggregateClientIPHistory(ctx context.Context, clientID string, from time.Time, to time.Time, limit int) ([]storage.ClientIPAggregateRecord, error)

AggregateClientIPHistory pushes the per-IP fold into the database: one row per IP, with MIN(first_seen) / MAX(last_seen) across all agents that reported it. Limit is applied in SQL so a high-cardinality client never streams millions of raw rows back to the control plane. A zero or negative limit disables the cap.

func (*Store) AppendAuditEvent 

func (s *Store) AppendAuditEvent(ctx context.Context, event storage.AuditEventRecord) error

AppendAuditEvent persists one audit row.

R-Q-03: routed through dbsqlc.AppendAuditEvent. The details field flows through the encodeJSON helper so legacy callers keep their untyped `map[string]any` shape — sqlc owns the column-level types for everything else.

func (*Store) AppendAuditEventsBulk 

func (s *Store) AppendAuditEventsBulk(ctx context.Context, events []storage.AuditEventRecord) error

AppendAuditEventsBulk inserts a batch of audit rows in one transaction (P6-6.1b). Mirrors the sqlite implementation; see store.go for the hash-chain rationale.

func (*Store) AppendDCHealthPoint 

func (s *Store) AppendDCHealthPoint(ctx context.Context, record storage.DCHealthPointRecord) error

func (*Store) AppendDCHealthPointsBulk 

func (s *Store) AppendDCHealthPointsBulk(ctx context.Context, records []storage.DCHealthPointRecord) error

AppendDCHealthPointsBulk inserts a batch of DC-health points. Same ON CONFLICT DO NOTHING semantics as the single-row variant.

func (*Store) AppendMetricSnapshot 

func (s *Store) AppendMetricSnapshot(ctx context.Context, snapshot storage.MetricSnapshotRecord) error

func (*Store) AppendMetricSnapshotsBulk 

func (s *Store) AppendMetricSnapshotsBulk(ctx context.Context, snapshots []storage.MetricSnapshotRecord) error

AppendMetricSnapshotsBulk inserts a batch of metric snapshots. Rows have a synthetic ID primary key so no ON CONFLICT clause is needed — same as the single-row AppendMetricSnapshot.

func (*Store) AppendServerLoadPoint 

func (s *Store) AppendServerLoadPoint(ctx context.Context, record storage.ServerLoadPointRecord) error

func (*Store) AppendServerLoadPointsBulk 

func (s *Store) AppendServerLoadPointsBulk(ctx context.Context, records []storage.ServerLoadPointRecord) error

AppendServerLoadPointsBulk inserts a batch of server-load points. Matches the single-row INSERT ... ON CONFLICT (agent_id, captured_at) DO NOTHING semantics so duplicate (agent,capture) pairs do not error.

func (*Store) AppendTelemetryRuntimeEvents 

func (s *Store) AppendTelemetryRuntimeEvents(ctx context.Context, agentID string, records []storage.TelemetryRuntimeEventRecord) error

func (*Store) AppendTelemetryRuntimeEventsBulk 

func (s *Store) AppendTelemetryRuntimeEventsBulk(ctx context.Context, records []storage.TelemetryRuntimeEventRecord) error

func (*Store) Close 

func (s *Store) Close() error

Close releases the database handle owned by the store.

func (*Store) CloseAgentCertOverlap 

func (s *Store) CloseAgentCertOverlap(ctx context.Context, agentID string) error

CloseAgentCertOverlap drops the previous credential.

func (*Store) ConsumeEnrollmentToken 

func (s *Store) ConsumeEnrollmentToken(ctx context.Context, value string, consumedAt time.Time) (storage.EnrollmentTokenRecord, error)

func (*Store) CountFleetGroupMembers 

func (s *Store) CountFleetGroupMembers(ctx context.Context, fleetGroupID string) (storage.ReassignCounts, error)

func (*Store) CountUniqueClientIPs 

func (s *Store) CountUniqueClientIPs(ctx context.Context, clientID string) (int, error)

func (*Store) CountUniqueClientIPsForClients 

func (s *Store) CountUniqueClientIPsForClients(ctx context.Context, clientIDs []string) (map[string]int, error)

CountUniqueClientIPsForClients computes the unique-IP count for each client ID in one query so the /api/clients listing avoids the N+1 pattern (Q2.U-P-03).

func (*Store) CreateConfigApplyBatch 

func (s *Store) CreateConfigApplyBatch(ctx context.Context, b storage.ConfigApplyBatchRecord, targets []storage.ConfigApplyBatchTargetRecord) error

CreateConfigApplyBatch inserts a batch row and its full target set inside a single transaction via the store's internal-tx helper (beginInternalTx): either every row lands or none does.

func (*Store) CreateFleetGroup 

func (s *Store) CreateFleetGroup(ctx context.Context, group storage.FleetGroupRecord) error

func (*Store) CreateFleetGroupIntegration 

func (s *Store) CreateFleetGroupIntegration(ctx context.Context, i storage.FleetGroupIntegrationRecord) error

func (*Store) CreateIntegrationProvider 

func (s *Store) CreateIntegrationProvider(ctx context.Context, p storage.IntegrationProviderRecord) error

CreateIntegrationProvider inserts a new provider row. config is opaque TEXT, not JSONB: fleet.Service.encryptProviderConfig may seal the caller-supplied JSON into a "PVS2:"/"PVS3:"-prefixed ciphertext string before it reaches this store (see db/migrations/postgres/0052_integration_providers_config_text.sql for why a jsonb column/cast cannot hold that). The table's CHECK constraint enforces "plain JSON OR PVS_:%-prefixed ciphertext" at write time in place of JSONB's native validation.

func (*Store) DB 

func (s *Store) DB() *sql.DB

DB returns the underlying *sql.DB. Used by adapters that need raw SQL (e.g. settings.NewDBStore for column-keyed UPDATE). Returns nil when the store is tx-bound (no pool of its own).

func (*Store) DeleteAgent 

func (s *Store) DeleteAgent(ctx context.Context, agentID string) error

func (*Store) DeleteAgentConfigTarget 

func (s *Store) DeleteAgentConfigTarget(ctx context.Context, scopeType, scopeID string) (int64, error)

DeleteAgentConfigTarget removes the config target row for one scope. Returns the number of rows deleted (0 if the row did not exist).

func (*Store) DeleteAgentFallbackState 

func (s *Store) DeleteAgentFallbackState(ctx context.Context, agentID string) error

DeleteAgentFallbackState clears the fallback marker — invoked when the agent's MERuntimeReady flag returns to true. Idempotent.

func (*Store) DeleteAgentUpdateStrategy 

func (s *Store) DeleteAgentUpdateStrategy(ctx context.Context, agentID string) error

DeleteAgentUpdateStrategy removes the update strategy row for an agent. Idempotent: deleting an absent row is not an error.

func (*Store) DeleteClientUsageByClient 

func (s *Store) DeleteClientUsageByClient(ctx context.Context, clientID string) error

func (*Store) DeleteExpiredAgentRevocations 

func (s *Store) DeleteExpiredAgentRevocations(ctx context.Context, before time.Time) (int64, error)

DeleteExpiredAgentRevocations removes entries whose cert has already expired — once the cert can no longer authenticate, the revocation entry is no longer useful and can shrink the table.

func (*Store) DeleteExpiredConsumedTotp 

func (s *Store) DeleteExpiredConsumedTotp(ctx context.Context, before time.Time) error

func (*Store) DeleteExpiredLoginLockouts 

func (s *Store) DeleteExpiredLoginLockouts(ctx context.Context, before time.Time) (int64, error)

func (*Store) DeleteExpiredSessions 

func (s *Store) DeleteExpiredSessions(ctx context.Context, before time.Time) error

func (*Store) DeleteFleetGroup 

func (s *Store) DeleteFleetGroup(ctx context.Context, id string) error

func (*Store) DeleteFleetGroupIntegration 

func (s *Store) DeleteFleetGroupIntegration(ctx context.Context, id string) error

func (*Store) DeleteInstancesByAgent 

func (s *Store) DeleteInstancesByAgent(ctx context.Context, agentID string) error

func (*Store) DeleteIntegrationProvider 

func (s *Store) DeleteIntegrationProvider(ctx context.Context, id string) error

func (*Store) DeleteLoginLockout 

func (s *Store) DeleteLoginLockout(ctx context.Context, username string) error

func (*Store) DeleteSession 

func (s *Store) DeleteSession(ctx context.Context, sessionID string) error

func (*Store) DeleteUser 

func (s *Store) DeleteUser(ctx context.Context, userID string) error

func (*Store) EarliestAgentCertExpiry 

func (s *Store) EarliestAgentCertExpiry(ctx context.Context) (*time.Time, error)

ListAgents returns every agent the panel knows about, ordered by last_seen_at + id for stable pagination.

Phase-3 §3.1: this is the first method to consume the sqlc-generated dbsqlc.Queries surface. Conversion from dbsqlc.ListAgentsRow to the storage.AgentRecord shape lives in agentRecordFromRow below; if a future query gets migrated, that helper stays the only place that knows about the SQL → domain mapping. EarliestAgentCertExpiry returns MIN(cert_expires_at) or nil when no agent carries an expiry (P6-6.3f).

func (*Store) GetAgentCertPin 

func (s *Store) GetAgentCertPin(ctx context.Context, agentID string) ([]byte, error)

GetAgentCertPin returns the SPKI pin for the agent. Returns ErrNotFound when no agent with the given ID exists; returns empty bytes (no error) when the agent exists but is not yet pinned.

func (*Store) GetAgentCertPins 

func (s *Store) GetAgentCertPins(ctx context.Context, agentID string) (storage.AgentCertPins, error)

GetAgentCertPins returns the credentials the panel accepts for the agent.

func (*Store) GetAgentCertSerial 

func (s *Store) GetAgentCertSerial(ctx context.Context, agentID string) (string, error)

GetAgentCertSerial returns the pinned serial for the given agent.

func (*Store) GetAgentCertificateRecoveryGrant 

func (s *Store) GetAgentCertificateRecoveryGrant(ctx context.Context, agentID string) (storage.AgentCertificateRecoveryGrantRecord, error)

func (*Store) GetAgentConfigTarget 

func (s *Store) GetAgentConfigTarget(ctx context.Context, scopeType, scopeID string) (storage.AgentConfigTargetRecord, error)

GetAgentConfigTarget returns the operator-desired Telemt config for one scope. Returns storage.ErrNotFound when no row exists for the given scopeType+scopeID pair.

func (*Store) GetAgentFallbackState 

func (s *Store) GetAgentFallbackState(ctx context.Context, agentID string) (storage.AgentFallbackStateRecord, error)

GetAgentFallbackState returns the persisted fallback entry for one agent. Returns storage.ErrNotFound when the agent is not currently in fallback.

func (*Store) GetAgentUpdateStrategy 

func (s *Store) GetAgentUpdateStrategy(ctx context.Context, agentID string) (storage.AgentUpdateStrategyRecord, error)

GetAgentUpdateStrategy returns the persisted Telemt update strategy for one agent. Returns storage.ErrNotFound when no row exists for the agent.

func (*Store) GetCPSecret 

func (s *Store) GetCPSecret(ctx context.Context, key string) ([]byte, error)

func (*Store) GetCertificateAuthority 

func (s *Store) GetCertificateAuthority(ctx context.Context) (storage.CertificateAuthorityRecord, error)

func (*Store) GetConfigApplyBatch 

func (s *Store) GetConfigApplyBatch(ctx context.Context, id string) (storage.ConfigApplyBatchRecord, []storage.ConfigApplyBatchTargetRecord, error)

GetConfigApplyBatch returns the batch plus every target row, ordered by wave_index then agent_id. Returns storage.ErrNotFound when no batch with the given id exists.

func (*Store) GetEnrollmentToken 

func (s *Store) GetEnrollmentToken(ctx context.Context, value string) (storage.EnrollmentTokenRecord, error)

func (*Store) GetFleetGroup 

func (s *Store) GetFleetGroup(ctx context.Context, id string) (storage.FleetGroupRecord, error)

func (*Store) GetFleetGroupByName 

func (s *Store) GetFleetGroupByName(ctx context.Context, name string) (storage.FleetGroupRecord, error)

func (*Store) GetFleetGroupIntegration 

func (s *Store) GetFleetGroupIntegration(ctx context.Context, id string) (storage.FleetGroupIntegrationRecord, error)

func (*Store) GetGeoIPSettings 

func (s *Store) GetGeoIPSettings(ctx context.Context) (json.RawMessage, error)

func (*Store) GetGeoIPState 

func (s *Store) GetGeoIPState(ctx context.Context) (json.RawMessage, error)

func (*Store) GetIntegrationProvider 

func (s *Store) GetIntegrationProvider(ctx context.Context, id string) (storage.IntegrationProviderRecord, error)

func (*Store) GetJob 

func (s *Store) GetJob(ctx context.Context, id string) (storage.JobRecord, error)

GetJob returns one job row by primary key, or storage.ErrNotFound. See storage.JobStore for the contract.

func (*Store) GetLoginLockout 

func (s *Store) GetLoginLockout(ctx context.Context, username string) (storage.LoginLockoutRecord, error)

func (*Store) GetPanelSelfUpdate 

func (s *Store) GetPanelSelfUpdate(ctx context.Context) (json.RawMessage, error)

func (*Store) GetPanelSettings 

func (s *Store) GetPanelSettings(ctx context.Context) (storage.PanelSettingsRecord, error)

func (*Store) GetPendingAgentUpdates 

func (s *Store) GetPendingAgentUpdates(ctx context.Context) (json.RawMessage, error)

func (*Store) GetPendingTelemtUpdates 

func (s *Store) GetPendingTelemtUpdates(ctx context.Context) (json.RawMessage, error)

func (*Store) GetRetentionSettings 

func (s *Store) GetRetentionSettings(ctx context.Context) (storage.RetentionSettings, error)

func (*Store) GetSession 

func (s *Store) GetSession(ctx context.Context, sessionID string) (storage.SessionRecord, error)

func (*Store) GetTelemetryDiagnosticsCurrent 

func (s *Store) GetTelemetryDiagnosticsCurrent(ctx context.Context, agentID string) (storage.TelemetryDiagnosticsCurrentRecord, error)

func (*Store) GetTelemetryRuntimeCurrent 

func (s *Store) GetTelemetryRuntimeCurrent(ctx context.Context, agentID string) (storage.TelemetryRuntimeCurrentRecord, error)

func (*Store) GetTelemetrySecurityInventoryCurrent 

func (s *Store) GetTelemetrySecurityInventoryCurrent(ctx context.Context, agentID string) (storage.TelemetrySecurityInventoryCurrentRecord, error)

func (*Store) GetUpdateSettings 

func (s *Store) GetUpdateSettings(ctx context.Context) (json.RawMessage, error)

func (*Store) GetUpdateState 

func (s *Store) GetUpdateState(ctx context.Context) (json.RawMessage, error)

func (*Store) GetUserAppearance 

func (s *Store) GetUserAppearance(ctx context.Context, userID string) (storage.UserAppearanceRecord, error)

func (*Store) GetUserByID 

func (s *Store) GetUserByID(ctx context.Context, userID string) (storage.UserRecord, error)

func (*Store) GetUserByUsername 

func (s *Store) GetUserByUsername(ctx context.Context, username string) (storage.UserRecord, error)

func (*Store) LatestAuditChainHash 

func (s *Store) LatestAuditChainHash(ctx context.Context) (string, error)

LatestAuditChainHash returns the EventHash of the most recently persisted audit row. Empty string when the table is empty.

Producers read this once per batch flush so each row is chained onto the tail of the existing chain. See AuditStore.LatestAuditChainHash.

func (*Store) ListAgentCertificateRecoveryGrants 

func (s *Store) ListAgentCertificateRecoveryGrants(ctx context.Context) ([]storage.AgentCertificateRecoveryGrantRecord, error)

func (*Store) ListAgentConfigTargets 

func (s *Store) ListAgentConfigTargets(ctx context.Context) ([]storage.AgentConfigTargetRecord, error)

ListAgentConfigTargets returns all operator-desired Telemt config targets, ordered by scope_type ASC, scope_id ASC.

func (*Store) ListAgentFallbackState 

func (s *Store) ListAgentFallbackState(ctx context.Context) ([]storage.AgentFallbackStateRecord, error)

ListAgentFallbackState returns every agent currently flagged as in fallback, oldest entry first. Drives the panel-side severity refresh loop on startup.

func (*Store) ListAgentRevocations 

func (s *Store) ListAgentRevocations(ctx context.Context) ([]storage.AgentRevocationRecord, error)

func (*Store) ListAgents 

func (s *Store) ListAgents(ctx context.Context) ([]storage.AgentRecord, error)

func (*Store) ListAllJobTargets 

func (s *Store) ListAllJobTargets(ctx context.Context) ([]storage.JobTargetRecord, error)

ListAllJobTargets returns every job_targets row in one round-trip so the service-level restore loop can hydrate Job.Targets without per-job N+1 SELECTs.

func (*Store) ListAllTelemetryRuntimeDCs 

func (s *Store) ListAllTelemetryRuntimeDCs(ctx context.Context) ([]storage.TelemetryRuntimeDCRecord, error)

ListAllTelemetryRuntimeDCs returns DC rows for every agent in a single query so cold-start rehydration groups by agent_id in memory instead of issuing one query per agent (A2).

func (*Store) ListAllTelemetryRuntimeEventsPerAgent 

func (s *Store) ListAllTelemetryRuntimeEventsPerAgent(ctx context.Context, perAgentLimit int) ([]storage.TelemetryRuntimeEventRecord, error)

ListAllTelemetryRuntimeEventsPerAgent returns the most recent perAgentLimit events PER agent for every agent in one query. The per-agent window is enforced by ROW_NUMBER() OVER (PARTITION BY agent_id ...) — NOT a global LIMIT — so each agent gets its own newest-N slice. perAgentLimit <= 0 returns all events.

func (*Store) ListAllTelemetryRuntimeUpstreams 

func (s *Store) ListAllTelemetryRuntimeUpstreams(ctx context.Context) ([]storage.TelemetryRuntimeUpstreamRecord, error)

ListAllTelemetryRuntimeUpstreams returns upstream rows for every agent in a single query (A2 cold-start rehydration).

func (*Store) ListAllUserFleetGroupScopes 

func (s *Store) ListAllUserFleetGroupScopes(ctx context.Context) ([]storage.UserFleetGroupScopeRecord, error)

ListAllUserFleetGroupScopes returns every scope grant with provenance. Offline-migrate only. Uses raw SQL rather than dbsqlc because the migrate-complete listing has no other caller and adding a sqlc query would force a baseline regen.

func (*Store) ListAuditEvents 

func (s *Store) ListAuditEvents(ctx context.Context, limit int) ([]storage.AuditEventRecord, error)

func (*Store) ListAuditEventsCursor 

func (s *Store) ListAuditEventsCursor(ctx context.Context, params storage.ListAuditEventsCursorParams) ([]storage.AuditEventRecord, storage.ListAuditEventsCursorParams, error)

ListAuditEventsCursor returns one keyset-paginated page in (created_at DESC, id DESC) order — newest first. Hand-written SQL (not sqlc) so the tuple-comparison form ports cleanly across drivers without regenerating the entire dbsqlc tree. See storage.AuditStore for the contract.

func (*Store) ListCPSecrets 

func (s *Store) ListCPSecrets(ctx context.Context) ([]storage.CPSecretRecord, error)

ListCPSecrets enumerates every cp_secrets row for the offline migrate tooling. Values are returned verbatim as raw bytes. Uses raw SQL rather than dbsqlc because the migrate-complete listing has no other caller and adding a sqlc query would force a baseline regen.

func (*Store) ListClientAssignments 

func (s *Store) ListClientAssignments(ctx context.Context, clientID string) ([]storage.ClientAssignmentRecord, error)

func (*Store) ListClientDeployments 

func (s *Store) ListClientDeployments(ctx context.Context, clientID string) ([]storage.ClientDeploymentRecord, error)

func (*Store) ListClientIPHistory 

func (s *Store) ListClientIPHistory(ctx context.Context, clientID string, from time.Time, to time.Time) ([]storage.ClientIPHistoryRecord, error)

ListClientIPHistory returns the per-(agent, ip) seen rows for `clientID` inside the time window [from, to]. Capped at storage.DefaultListLimit rows (P-7) so a high-cardinality client cannot stream millions of rows when a caller forgets to pre-aggregate. Operators that genuinely need every row should use AggregateClientIPHistory or a cursor-paginated follow-up query.

func (*Store) ListClientUsage 

func (s *Store) ListClientUsage(ctx context.Context) ([]storage.ClientUsageRecord, error)

func (*Store) ListClients 

func (s *Store) ListClients(ctx context.Context) ([]storage.ClientRecord, error)

func (*Store) ListConsumedTotp 

func (s *Store) ListConsumedTotp(ctx context.Context) ([]storage.ConsumedTotpRecord, error)

func (*Store) ListDCHealthPoints 

func (s *Store) ListDCHealthPoints(ctx context.Context, agentID string, from time.Time, to time.Time) ([]storage.DCHealthPointRecord, error)

func (*Store) ListEnrollmentTokens 

func (s *Store) ListEnrollmentTokens(ctx context.Context) ([]storage.EnrollmentTokenRecord, error)

ListEnrollmentTokens returns every token, ordered by issued_at + value for stable pagination.

R-Q-03: routed through dbsqlc.ListEnrollmentTokens. Conversion from dbsqlc.EnrollmentToken to the storage shape lives in enrollmentTokenFromRow.

func (*Store) ListFleetGroupIntegrations 

func (s *Store) ListFleetGroupIntegrations(ctx context.Context, fleetGroupID string) ([]storage.FleetGroupIntegrationRecord, error)

func (*Store) ListFleetGroups 

func (s *Store) ListFleetGroups(ctx context.Context) ([]storage.FleetGroupRecord, error)

func (*Store) ListInstances 

func (s *Store) ListInstances(ctx context.Context) ([]storage.InstanceRecord, error)

func (*Store) ListIntegrationProviders 

func (s *Store) ListIntegrationProviders(ctx context.Context) ([]storage.IntegrationProviderRecord, error)

func (*Store) ListIntegrationProvidersByKind 

func (s *Store) ListIntegrationProvidersByKind(ctx context.Context, kind string) ([]storage.IntegrationProviderRecord, error)

func (*Store) ListJobTargets 

func (s *Store) ListJobTargets(ctx context.Context, jobID string) ([]storage.JobTargetRecord, error)

ListJobTargets returns every delivery row for one job, ordered by agent_id. Wired through dbsqlc.ListJobTargets.

func (*Store) ListJobs 

func (s *Store) ListJobs(ctx context.Context) ([]storage.JobRecord, error)

ListJobs returns every job ordered by created_at + id for stable pagination. Phase-3 §3.1 (continued): wired through dbsqlc.ListJobs; the SQL definition in db/queries/jobs.sql is the single source of truth for column set + ORDER BY.

S25 T1: defensive cap. dbsqlc.ListJobs is unbounded; we read into a slice and trim to DefaultListLimit so a long-lived control plane cannot stream millions of rows even if a caller forgets to paginate. Operator-facing list APIs should call ListJobsCursor instead.

func (*Store) ListJobsCursor 

func (s *Store) ListJobsCursor(ctx context.Context, params storage.ListJobsCursorParams) ([]storage.JobRecord, storage.ListJobsCursorParams, error)

ListJobsCursor returns one keyset-paginated page of jobs in (created_at DESC, id DESC) order. See storage.JobStore for the contract. The query is hand-written rather than going through sqlc so the cursor variant can ship without regenerating the entire dbsqlc tree (sqlc parametrises tuple comparisons differently across drivers — keeping this local keeps the change footprint small).

func (*Store) ListLoginLockouts 

func (s *Store) ListLoginLockouts(ctx context.Context) ([]storage.LoginLockoutRecord, error)

func (*Store) ListMetricSnapshots 

func (s *Store) ListMetricSnapshots(ctx context.Context) ([]storage.MetricSnapshotRecord, error)

func (*Store) ListRunningConfigApplyBatches 

func (s *Store) ListRunningConfigApplyBatches(ctx context.Context) ([]storage.ConfigApplyBatchRecord, error)

ListRunningConfigApplyBatches returns every batch in storage.ConfigApplyBatchStatusRunning, ordered by created_at then id.

func (*Store) ListServerLoadHourly 

func (s *Store) ListServerLoadHourly(ctx context.Context, agentID string, from time.Time, to time.Time) ([]storage.ServerLoadHourlyRecord, error)

func (*Store) ListServerLoadPoints 

func (s *Store) ListServerLoadPoints(ctx context.Context, agentID string, from time.Time, to time.Time) ([]storage.ServerLoadPointRecord, error)

func (*Store) ListServerLoadPointsForAgents 

func (s *Store) ListServerLoadPointsForAgents(ctx context.Context, agentIDs []string, from time.Time, to time.Time) (map[string][]storage.ServerLoadPointRecord, error)

ListServerLoadPointsForAgents returns load points for a batch of agents (Q2.U-P-01). Each agent's slice is sorted by captured_at ascending; missing agents are absent from the map. Chunked so the IN-list never approaches the Postgres 65535-parameter ceiling.

func (*Store) ListSessions 

func (s *Store) ListSessions(ctx context.Context) ([]storage.SessionRecord, error)

func (*Store) ListTelemetryRuntimeCurrent 

func (s *Store) ListTelemetryRuntimeCurrent(ctx context.Context) ([]storage.TelemetryRuntimeCurrentRecord, error)

func (*Store) ListTelemetryRuntimeDCs 

func (s *Store) ListTelemetryRuntimeDCs(ctx context.Context, agentID string) ([]storage.TelemetryRuntimeDCRecord, error)

func (*Store) ListTelemetryRuntimeEvents 

func (s *Store) ListTelemetryRuntimeEvents(ctx context.Context, agentID string, limit int) ([]storage.TelemetryRuntimeEventRecord, error)

func (*Store) ListTelemetryRuntimeUpstreams 

func (s *Store) ListTelemetryRuntimeUpstreams(ctx context.Context, agentID string) ([]storage.TelemetryRuntimeUpstreamRecord, error)

func (*Store) ListUserAppearances 

func (s *Store) ListUserAppearances(ctx context.Context) ([]storage.UserAppearanceRecord, error)

func (*Store) ListUserFleetGroupScopes 

func (s *Store) ListUserFleetGroupScopes(ctx context.Context, userID string) ([]string, error)

ListUserFleetGroupScopes returns every fleet_group_id the user is scoped to. An empty slice means "global".

func (*Store) ListUsers 

func (s *Store) ListUsers(ctx context.Context) ([]storage.UserRecord, error)

func (*Store) Ping 

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

Ping verifies that the database connection is alive.

func (*Store) PoolStats 

func (s *Store) PoolStats() sql.DBStats

PoolStats returns the current sql.DBStats for this store, or the zero value when the store is tx-bound (no pool of its own). Used by the metrics publisher to expose panvex_db_pool_* gauges.

func (*Store) PruneAuditEvents 

func (s *Store) PruneAuditEvents(ctx context.Context, before time.Time) (int64, error)

PruneAuditEvents deletes audit_events rows with created_at strictly before the cutoff and returns the RowsAffected count (P2-REL-04 / finding M-R2). Relies on idx_audit_events_created_at (added in P2-DB-02) for efficiency.

R-Q-03: routed through dbsqlc.PruneAuditEvents.

func (*Store) PruneClientIPHistory 

func (s *Store) PruneClientIPHistory(ctx context.Context, olderThan time.Time) (int64, error)

func (*Store) PruneConfigApplyBatches 

func (s *Store) PruneConfigApplyBatches(ctx context.Context, before time.Time) (int64, error)

PruneConfigApplyBatches deletes batches in a terminal status (succeeded/failed/halted) whose updated_at predates before. Targets are removed via ON DELETE CASCADE.

func (*Store) PruneDCHealthPoints 

func (s *Store) PruneDCHealthPoints(ctx context.Context, olderThan time.Time) (int64, error)

func (*Store) PruneEnrollmentTokens 

func (s *Store) PruneEnrollmentTokens(ctx context.Context, before time.Time) (int64, error)

PruneEnrollmentTokens implements the EnrollmentStore prune contract (C4). R-Q-03: routed through dbsqlc.PruneEnrollmentTokens.

func (*Store) PruneMetricSnapshots 

func (s *Store) PruneMetricSnapshots(ctx context.Context, before time.Time) (int64, error)

PruneMetricSnapshots deletes metric_snapshots rows with captured_at strictly before the cutoff and returns the RowsAffected count (P2-REL-05). Relies on idx_metric_snapshots_captured_at (added in P2-DB-02) for efficiency.

func (*Store) PruneServerLoadHourly 

func (s *Store) PruneServerLoadHourly(ctx context.Context, olderThan time.Time) (int64, error)

func (*Store) PruneServerLoadPoints 

func (s *Store) PruneServerLoadPoints(ctx context.Context, olderThan time.Time) (int64, error)

func (*Store) PruneTelemetryRuntimeEvents 

func (s *Store) PruneTelemetryRuntimeEvents(ctx context.Context, olderThan time.Time) (int64, error)

func (*Store) PruneTerminalJobs 

func (s *Store) PruneTerminalJobs(ctx context.Context, before time.Time) (int64, error)

PruneTerminalJobs deletes jobs in a finished status whose created_at predates the cutoff (Q2.U-P-02). job_targets is cleaned up via ON DELETE CASCADE in the schema.

func (*Store) PutAgent 

func (s *Store) PutAgent(ctx context.Context, agent storage.AgentRecord) error

PutAgent upserts one agent row.

Phase-3 §3.1 (continued): now goes through dbsqlc.UpsertAgent. agentRecordToUpsertParams below is the domain-DTO → SQL-row bridge — future PutAgent callers gain compile-time type safety on every column from the sqlc-generated UpsertAgentParams.

Uses s.db (the dbExecutor) rather than the pool-only s.sqlDB so the upsert composes inside Transact — the inbound enrollment flow calls this as tx.PutAgent(...) from within a transaction. dbsqlc.New accepts any DBTX, which dbExecutor satisfies (both *sql.DB and *sql.Tx fit).

func (*Store) PutAgentCertificateRecoveryGrant 

func (s *Store) PutAgentCertificateRecoveryGrant(ctx context.Context, grant storage.AgentCertificateRecoveryGrantRecord) error

func (*Store) PutAgentFallbackState 

func (s *Store) PutAgentFallbackState(ctx context.Context, rec storage.AgentFallbackStateRecord) error

PutAgentFallbackState marks the agent as having entered ME→Direct fallback at rec.EnteredAt. The first writer wins: subsequent calls while a row is already present are a no-op so the original transition timestamp survives across observer churn (Phase 4 §4.3).

func (*Store) PutAgentRevocation 

func (s *Store) PutAgentRevocation(ctx context.Context, r storage.AgentRevocationRecord) error

PutAgentRevocation upserts a revocation so repeated deregistrations are idempotent and cert_expires_at is kept fresh if the caller knows a newer cert existed.

func (*Store) PutAgentsBulk 

func (s *Store) PutAgentsBulk(ctx context.Context, agents []storage.AgentRecord) error

PutAgentsBulk upserts a batch of agents in a single transaction using chunked multi-row INSERT. See Store.PutAgentsBulk in storage/store.go for the full contract.

func (*Store) PutCPSecret 

func (s *Store) PutCPSecret(ctx context.Context, key string, value []byte) error

func (*Store) PutCertificateAuthority 

func (s *Store) PutCertificateAuthority(ctx context.Context, authority storage.CertificateAuthorityRecord) error

func (*Store) PutClient 

func (s *Store) PutClient(ctx context.Context, client storage.ClientRecord) error

func (*Store) PutClientAssignment 

func (s *Store) PutClientAssignment(ctx context.Context, assignment storage.ClientAssignmentRecord) error

func (*Store) PutClientDeployment 

func (s *Store) PutClientDeployment(ctx context.Context, deployment storage.ClientDeploymentRecord) error

func (*Store) PutEnrollmentToken 

func (s *Store) PutEnrollmentToken(ctx context.Context, token storage.EnrollmentTokenRecord) error

PutEnrollmentToken upserts one enrollment_tokens row.

R-Q-03: routed through dbsqlc.UpsertEnrollmentToken so the postgres path gains compile-time type safety on every column. The dead value_hash column was dropped in migration 0044 (L-4).

func (*Store) PutFleetGroup 

func (s *Store) PutFleetGroup(ctx context.Context, group storage.FleetGroupRecord) error

func (*Store) PutGeoIPSettings 

func (s *Store) PutGeoIPSettings(ctx context.Context, data json.RawMessage) error

func (*Store) PutGeoIPState 

func (s *Store) PutGeoIPState(ctx context.Context, data json.RawMessage) error

func (*Store) PutInstance 

func (s *Store) PutInstance(ctx context.Context, instance storage.InstanceRecord) error

func (*Store) PutInstancesBulk 

func (s *Store) PutInstancesBulk(ctx context.Context, instances []storage.InstanceRecord) error

PutInstancesBulk upserts a batch of Telemt instances. See Store.PutInstancesBulk.

func (*Store) PutJob 

func (s *Store) PutJob(ctx context.Context, job storage.JobRecord) error

func (*Store) PutJobTarget 

func (s *Store) PutJobTarget(ctx context.Context, target storage.JobTargetRecord) error

func (*Store) PutPanelSelfUpdate 

func (s *Store) PutPanelSelfUpdate(ctx context.Context, data json.RawMessage) error

func (*Store) PutPanelSettings 

func (s *Store) PutPanelSettings(ctx context.Context, settings storage.PanelSettingsRecord) error

func (*Store) PutPendingAgentUpdates 

func (s *Store) PutPendingAgentUpdates(ctx context.Context, data json.RawMessage) error

func (*Store) PutPendingTelemtUpdates 

func (s *Store) PutPendingTelemtUpdates(ctx context.Context, data json.RawMessage) error

func (*Store) PutRetentionSettings 

func (s *Store) PutRetentionSettings(ctx context.Context, settings storage.RetentionSettings) error

func (*Store) PutSession 

func (s *Store) PutSession(ctx context.Context, session storage.SessionRecord) error

func (*Store) PutTelemetryDiagnosticsCurrent 

func (s *Store) PutTelemetryDiagnosticsCurrent(ctx context.Context, record storage.TelemetryDiagnosticsCurrentRecord) error

func (*Store) PutTelemetryDiagnosticsCurrentBulk 

func (s *Store) PutTelemetryDiagnosticsCurrentBulk(ctx context.Context, records []storage.TelemetryDiagnosticsCurrentRecord) error

func (*Store) PutTelemetryRuntimeCurrent 

func (s *Store) PutTelemetryRuntimeCurrent(ctx context.Context, record storage.TelemetryRuntimeCurrentRecord) error

func (*Store) PutTelemetryRuntimeCurrentBulk 

func (s *Store) PutTelemetryRuntimeCurrentBulk(ctx context.Context, records []storage.TelemetryRuntimeCurrentRecord) error

func (*Store) PutTelemetrySecurityInventoryCurrent 

func (s *Store) PutTelemetrySecurityInventoryCurrent(ctx context.Context, record storage.TelemetrySecurityInventoryCurrentRecord) error

func (*Store) PutTelemetrySecurityInventoryCurrentBulk 

func (s *Store) PutTelemetrySecurityInventoryCurrentBulk(ctx context.Context, records []storage.TelemetrySecurityInventoryCurrentRecord) error

func (*Store) PutUpdateSettings 

func (s *Store) PutUpdateSettings(ctx context.Context, data json.RawMessage) error

func (*Store) PutUpdateState 

func (s *Store) PutUpdateState(ctx context.Context, data json.RawMessage) error

func (*Store) PutUser 

func (s *Store) PutUser(ctx context.Context, user storage.UserRecord) error

PutUser upserts one users row.

R-Q-03: routed through dbsqlc.UpsertUser. The created_at column is no longer touched by the upsert path so an UPDATE keeps the original timestamp — this matches the prior behaviour where ON CONFLICT set created_at to EXCLUDED.created_at and callers passed the same value they originally inserted; the column is stable across upserts so dropping it from the SET keeps the existing semantic for every observed callsite.

func (*Store) PutUserAppearance 

func (s *Store) PutUserAppearance(ctx context.Context, appearance storage.UserAppearanceRecord) error

func (*Store) Queries 

func (s *Store) Queries() *dbsqlc.Queries

Queries returns a *dbsqlc.Queries bound to this store's connection pool. Callers that need transport/bootstrap DB access (agenttransport.Manager, bootstrap.EnrollDriver, the provision-outbound handler) use this instead of opening a second sql.DB. Returns nil when the store is tx-bound (no pool).

func (*Store) ReassignFleetGroupMembers 

func (s *Store) ReassignFleetGroupMembers(ctx context.Context, fromID, toID string) (storage.ReassignCounts, error)

ReassignFleetGroupMembers is NOT atomic on its own — callers must wrap the full delete flow in Store.Transact. See fleet.Service.Delete.

func (*Store) ReplaceTelemetryRuntimeDCs 

func (s *Store) ReplaceTelemetryRuntimeDCs(ctx context.Context, agentID string, records []storage.TelemetryRuntimeDCRecord) error

func (*Store) ReplaceTelemetryRuntimeDCsBulk 

func (s *Store) ReplaceTelemetryRuntimeDCsBulk(ctx context.Context, byAgent map[string][]storage.TelemetryRuntimeDCRecord) error

func (*Store) ReplaceTelemetryRuntimeUpstreams 

func (s *Store) ReplaceTelemetryRuntimeUpstreams(ctx context.Context, agentID string, records []storage.TelemetryRuntimeUpstreamRecord) error

func (*Store) ReplaceTelemetryRuntimeUpstreamsBulk 

func (s *Store) ReplaceTelemetryRuntimeUpstreamsBulk(ctx context.Context, byAgent map[string][]storage.TelemetryRuntimeUpstreamRecord) error

func (*Store) RevokeAgentCertificateRecoveryGrant 

func (s *Store) RevokeAgentCertificateRecoveryGrant(ctx context.Context, agentID string, revokedAt time.Time) (storage.AgentCertificateRecoveryGrantRecord, error)

func (*Store) RevokeEnrollmentToken 

func (s *Store) RevokeEnrollmentToken(ctx context.Context, value string, revokedAt time.Time) (storage.EnrollmentTokenRecord, error)

func (*Store) RollupServerLoadHourly 

func (s *Store) RollupServerLoadHourly(ctx context.Context, bucketHour time.Time) error

func (*Store) RotateAgentCert 

func (s *Store) RotateAgentCert(ctx context.Context, agentID string, serial string, spki []byte, overlapUntil time.Time, presentedSerial string) error

RotateAgentCert records a new credential and keeps the previous one valid until overlapUntil. See storage.Store for why the overlap exists.

The previous credential is only retained when there IS one: a first issuance (enrollment) has nothing to fall back to, and an empty pin is what the verifier already treats as fail-closed.

When the renewal was presented over the PREVIOUS credential while the overlap window is still open (presentedSerial = cert_serial_prev, deadline set), prev and its deadline are kept as they are: the presenter just proved the agent never received the current certificate, so shifting prev := current would evict the only credential the agent holds (R11-1). Each CASE arm repeats that condition because every right-hand side of an UPDATE sees the pre-update row, so the three columns stay consistent.

func (*Store) SetConfigApplyBatchTargetJob 

func (s *Store) SetConfigApplyBatchTargetJob(ctx context.Context, batchID, agentID, jobID, status string) error

SetConfigApplyBatchTargetJob records the job enqueued for one target (wave enqueue) and updates its status in the same write.

func (*Store) SetUserFleetGroupScopes 

func (s *Store) SetUserFleetGroupScopes(ctx context.Context, userID string, fleetGroupIDs []string, grantedBy string, grantedAt time.Time) error

SetUserFleetGroupScopes replaces the user's scope set with the supplied list. Wrapped in a single transaction so a partially applied update cannot leave the operator stuck halfway between scopes.

func (*Store) TouchSession 

func (s *Store) TouchSession(ctx context.Context, sessionID string, lastSeenAt time.Time) error

TouchSession updates only last_seen_at so the sliding idle timeout survives restart (Q2.U-S-12).

func (*Store) Transact 

func (s *Store) Transact(ctx context.Context, fn storage.TxFn) error

Transact runs fn inside a single database transaction with read-committed isolation. On serialization failures it retries up to maxTransactRetries times. See storage.Store.Transact for the full contract.

func (*Store) UpdateAgentCertPin 

func (s *Store) UpdateAgentCertPin(ctx context.Context, agentID string, pin []byte) error

UpdateAgentCertPin persists the SPKI SHA-256 hash for an agent (S-02).

func (*Store) UpdateAgentCertSerial 

func (s *Store) UpdateAgentCertSerial(ctx context.Context, agentID string, serial string) error

UpdateAgentCertSerial pins the latest issued client cert serial (Q4.U-S-04). Called after each successful issuance.

func (*Store) UpdateAgentFleetGroup 

func (s *Store) UpdateAgentFleetGroup(ctx context.Context, agentID, fleetGroupID string) error

func (*Store) UpdateAgentNodeName 

func (s *Store) UpdateAgentNodeName(ctx context.Context, agentID string, nodeName string) error

func (*Store) UpdateAgentTransportMode 

func (s *Store) UpdateAgentTransportMode(ctx context.Context, agentID, transportMode, dialAddress string) error

func (*Store) UpdateConfigApplyBatchStatus 

func (s *Store) UpdateConfigApplyBatchStatus(ctx context.Context, id, status string, now time.Time) error

UpdateConfigApplyBatchStatus transitions a batch's status and bumps updated_at. Returns storage.ErrNotFound when no batch with the given id exists.

func (*Store) UpdateConfigApplyBatchTargetStatus 

func (s *Store) UpdateConfigApplyBatchTargetStatus(ctx context.Context, batchID, agentID, status, message string) error

UpdateConfigApplyBatchTargetStatus updates one target's delivery status and message without touching its job id.

func (*Store) UpdateFleetGroup 

func (s *Store) UpdateFleetGroup(ctx context.Context, group storage.FleetGroupRecord) error

UpdateFleetGroup mutates editable fields only; `name` is the immutable slug and is not in the SET list.

func (*Store) UpdateFleetGroupIntegration 

func (s *Store) UpdateFleetGroupIntegration(ctx context.Context, i storage.FleetGroupIntegrationRecord) error

func (*Store) UpdateIntegrationProvider 

func (s *Store) UpdateIntegrationProvider(ctx context.Context, p storage.IntegrationProviderRecord) error

UpdateIntegrationProvider — see CreateIntegrationProvider's doc comment for why config binds as plain TEXT rather than a ::jsonb cast.

func (*Store) UpsertAgentConfigTarget 

func (s *Store) UpsertAgentConfigTarget(ctx context.Context, rec storage.AgentConfigTargetRecord) error

UpsertAgentConfigTarget inserts or updates the operator-desired Telemt config for one scope. On conflict (scope_type, scope_id) the sections_json and updated_at are overwritten.

func (*Store) UpsertAgentUpdateStrategy 

func (s *Store) UpsertAgentUpdateStrategy(ctx context.Context, rec storage.AgentUpdateStrategyRecord) error

UpsertAgentUpdateStrategy inserts or replaces the update strategy for an agent. On conflict (agent_id) the mode/restart_spec/binary_path/ asset_flavor/updated_at are overwritten; created_at is left untouched.

func (*Store) UpsertClientIPHistory 

func (s *Store) UpsertClientIPHistory(ctx context.Context, record storage.ClientIPHistoryRecord) error

func (*Store) UpsertClientIPHistoryBulk 

func (s *Store) UpsertClientIPHistoryBulk(ctx context.Context, records []storage.ClientIPHistoryRecord) error

UpsertClientIPHistoryBulk upserts a batch of client-ip history rows. Same ON CONFLICT (agent_id, client_id, ip_address) DO UPDATE SET last_seen as the single-row variant; when the same (agent, client, ip) key appears twice in one batch, the last row's last_seen wins.

func (*Store) UpsertClientUsage 

func (s *Store) UpsertClientUsage(ctx context.Context, r storage.ClientUsageRecord) error

UpsertClientUsage inserts or updates one (client, agent) usage row. Unconditional last-write-wins upsert (P4): ordering/duplicate protection lives upstream in the panel's watermark derivation, not in SQL.

func (*Store) UpsertConsumedTotp 

func (s *Store) UpsertConsumedTotp(ctx context.Context, record storage.ConsumedTotpRecord) error

func (*Store) UpsertLoginLockout 

func (s *Store) UpsertLoginLockout(ctx context.Context, record storage.LoginLockoutRecord) error

func (*Store) UseAgentCertificateRecoveryGrant 

func (s *Store) UseAgentCertificateRecoveryGrant(ctx context.Context, agentID string, usedAt time.Time) (storage.AgentCertificateRecoveryGrantRecord, error)

type WebhookStore 

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

WebhookStore implements webhooks.Storage on top of pgx-backed *sql.DB. The Postgres dialect lets multiple worker replicas share the outbox safely via FOR UPDATE SKIP LOCKED on the claim path — a feature SQLite doesn't have, hence the separate file.

func NewWebhookStore 

func NewWebhookStore(db *sql.DB, decrypt webhooks.SecretDecrypter) *WebhookStore

NewWebhookStore wires a webhook storage backend over the given pool. decrypt mirrors sqlite.NewWebhookStore — see its godoc.

func (*WebhookStore) ClaimReady 

func (s *WebhookStore) ClaimReady(ctx context.Context, now time.Time, max int) ([]webhooks.Delivery, error)

ClaimReady atomically reserves up to max ready rows in a single transaction using FOR UPDATE SKIP LOCKED, which lets multiple workers race without re-delivering. The transaction commits as soon as the rows are read; the worker performs the actual HTTP POST outside the transaction.

SKIP LOCKED + ORDER BY may produce slightly out-of-order delivery when contention is high — acceptable for an at-least-once webhook queue (receivers must be idempotent on X-Panvex-Delivery anyway).

func (*WebhookStore) CreateEndpoint 

func (s *WebhookStore) CreateEndpoint(ctx context.Context, in webhooks.EndpointInput, now time.Time) error

func (*WebhookStore) DeleteEndpoint 

func (s *WebhookStore) DeleteEndpoint(ctx context.Context, id string) error

func (*WebhookStore) GetEndpointMeta 

func (s *WebhookStore) GetEndpointMeta(ctx context.Context, id string) (webhooks.Endpoint, error)

func (*WebhookStore) InsertOutbox 

func (s *WebhookStore) InsertOutbox(ctx context.Context, row webhooks.OutboxRow) error

InsertOutbox creates a pending delivery row.

func (*WebhookStore) InsertOutboxBatch 

func (s *WebhookStore) InsertOutboxBatch(ctx context.Context, rows []webhooks.OutboxRow) error

InsertOutboxBatch writes every fan-out row in ONE transaction so a failure on any row leaves the outbox untouched (R4 §1.7).

func (*WebhookStore) ListEnabledEndpoints 

func (s *WebhookStore) ListEnabledEndpoints(ctx context.Context) ([]webhooks.Endpoint, error)

ListEnabledEndpoints returns every enabled webhook_endpoints row with the secret already decrypted.

func (*WebhookStore) ListEndpointMeta 

func (s *WebhookStore) ListEndpointMeta(ctx context.Context) ([]webhooks.Endpoint, error)

func (*WebhookStore) MarkDelivered 

func (s *WebhookStore) MarkDelivered(ctx context.Context, id string, deliveredAt time.Time) error

func (*WebhookStore) MarkFailed 

func (s *WebhookStore) MarkFailed(ctx context.Context, id string, attempt int, nextAttempt time.Time, errMsg string, dead bool) error

func (*WebhookStore) PruneOutbox 

func (s *WebhookStore) PruneOutbox(ctx context.Context, before time.Time) (int64, error)

PruneOutbox deletes terminal rows per the webhooks.Storage contract (see the sqlite twin for the retention semantics).

func (*WebhookStore) UpdateEndpoint 

func (s *WebhookStore) UpdateEndpoint(ctx context.Context, in webhooks.EndpointInput, now time.Time) error

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.