serverstore

package
v0.1.1 Latest Latest
Warning

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

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

Documentation

Overview

Package serverstore is the csx-server persistence layer: a Store interface over PostgreSQL (plan contract C4) plus the pure ingest logic — batch validation and delta-merge dedup semantics — which unit-tests without a database. This file must stay free of pgx imports; the pgx implementation lives in pg.go.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Migrate 

func Migrate(ctx context.Context, conn *pgx.Conn) error

Migrate applies every embedded migration not yet recorded in schema_migrations, each inside its own transaction. It is idempotent and safe to run on every server start.

func ValidateBatch 

func ValidateBatch(b domain.ObservationBatch) error

ValidateBatch checks one ObservationBatch against the wire contract before it may touch storage. It is pure and shared with the HTTP ingest handler.

Enforced here (binding):

  • PURL parses and its ecosystem is on the public allowlist (§8.1).
  • Stage is an observation stage. SYMBOL_EXECUTED / SYMBOL_CALL are A3-only runtime-instrumentation stages and no Public v1 adapter claims A3, so they are rejected (docs/execution-context.md §3).
  • symbolConfidence is EXACT, PROBABLE or UNKNOWN (empty allowed).
  • errorFingerprint must be a sha256 content id — raw error text can never ride in on that field.

Types

type Changes added in v0.1.1 

type Changes struct {
	Targets     []SnapshotTarget
	SamplePURLs []string
}

NetworkCounts are the honest headline numbers behind /v1/stats (goal.md §14.5). Estimated values are computed elsewhere and always labeled; these are raw counts only. Changes is everything that could have altered a materialized view since a given time: evidence targets whose rows were touched, and the packages of samples that were added or that gained a receipt.

Receipts are the complete signal for sample state: SetSampleStatus is only ever called from the receipt handler, so a status transition always has a receipt behind it and never needs its own timestamp column.

func (Changes) Empty added in v0.1.1 

func (c Changes) Empty() bool

Empty reports that nothing moved, so a pass has no work beyond stats.

type ClusterRow 

type ClusterRow struct {
	ID                  int64
	Ecosystem           string
	PackageName         string
	Symbol              string
	Stage               string
	ErrorFingerprint    string
	ErrorCode           string
	ObservationCount    int64
	EnvSummaryJSON      string
	HypothesesJSON      string // [{domain,confidence}] — never a definitive cause
	RegressionCandidate bool
	VersionsJSON        string
	FirstSeen           time.Time
	LastSeen            time.Time
}

ClusterRow is one failure_clusters-table row.

type EvidenceRow 

type EvidenceRow struct {
	PURL                 string
	Symbol               string
	SymbolConfidence     string
	EnvHash              string
	EnvJSON              string
	Stage                string
	Result               string
	ErrorFingerprint     string
	ErrorCode            string
	ObservationCount     int64
	UniquePeerBuckets    int
	UniqueProjectBuckets int
	FirstSeen            time.Time
	LastSeen             time.Time
}

EvidenceRow is one aggregated evidence_agg row.

type Fake 

type Fake struct {

	// NowFn is the test seam for time-dependent behavior; nil means time.Now.
	NowFn func() time.Time
	// ChangedSinceFn overrides change detection. The fake keeps no per-row
	// timestamps, so incremental-rebuild tests script it directly.
	ChangedSinceFn func(context.Context, time.Time) (Changes, error)
	// contains filtered or unexported fields
}

Fake is a complete in-memory Store implementation. Handler and builder tests (and e2e harnesses that want a server without PostgreSQL) use it; its ingest semantics are the mergeState reference implementation that pg.go is held to, so both stores behave identically.

func NewFake 

func NewFake() *Fake

NewFake returns an empty in-memory Store.

func (*Fake) AnnouncePeer 

func (f *Fake) AnnouncePeer(_ context.Context, p PeerRow) error

func (*Fake) ChangedSince added in v0.1.1 

func (f *Fake) ChangedSince(ctx context.Context, since time.Time) (Changes, error)

ChangedSince reports everything as changed. The fake keeps no per-row timestamps, and over-reporting is the safe direction: a builder test then exercises the full path, and no test can pass because a change was silently missed.

func (*Fake) ClaimJob 

func (f *Fake) ClaimJob(_ context.Context, id int64, peerID string) (bool, error)

func (*Fake) Close 

func (f *Fake) Close()

Close implements Store; the fake has nothing to release.

func (*Fake) CompleteJob 

func (f *Fake) CompleteJob(_ context.Context, id int64) error

func (*Fake) CreateJob 

func (f *Fake) CreateJob(_ context.Context, j JobRow) (int64, error)

func (*Fake) EvidenceForTarget 

func (f *Fake) EvidenceForTarget(_ context.Context, purl, symbol string) ([]EvidenceRow, error)

func (*Fake) ExpirePeers 

func (f *Fake) ExpirePeers(_ context.Context, now time.Time) (int64, error)

func (*Fake) GetLatestStats 

func (f *Fake) GetLatestStats(_ context.Context) (string, bool, error)

func (*Fake) GetPackage 

func (f *Fake) GetPackage(_ context.Context, purl string) (PackageRow, bool, error)

func (*Fake) GetSample 

func (f *Fake) GetSample(_ context.Context, sampleID string) (SampleRow, bool, error)

func (*Fake) GetShard 

func (f *Fake) GetShard(_ context.Context, key string) (string, string, bool, error)

func (*Fake) GetSnapshot 

func (f *Fake) GetSnapshot(_ context.Context, purl, symbol string) (string, bool, error)

func (*Fake) HotShardKeys 

func (f *Fake) HotShardKeys(_ context.Context, limit int) ([]string, error)

HotShardKeys returns the stored keys in sorted order — the fake has no traffic model, only determinism.

func (*Fake) IdentityByAPIToken 

func (f *Fake) IdentityByAPIToken(_ context.Context, apiTokenHash string) (IdentityRow, bool, error)

func (*Fake) IngestBatches 

func (f *Fake) IngestBatches(_ context.Context, batches []domain.ObservationBatch) (int, []RejectedBatch, error)

IngestBatches mirrors PG.IngestBatches: validate, then delta-merge.

func (*Fake) JobsForSample 

func (f *Fake) JobsForSample(_ context.Context, sampleID string) ([]JobRow, error)

func (*Fake) ListFailureClusters 

func (f *Fake) ListFailureClusters(_ context.Context, packageName string) ([]ClusterRow, error)

func (*Fake) ListPackageVersions 

func (f *Fake) ListPackageVersions(_ context.Context, ecosystem, name string) ([]PackageRow, error)

func (*Fake) ListSamples 

func (f *Fake) ListSamples(_ context.Context, limit int) ([]SampleRow, error)

func (*Fake) ListSnapshotTargets 

func (f *Fake) ListSnapshotTargets(_ context.Context) ([]SnapshotTarget, error)

func (*Fake) NetworkCounts 

func (f *Fake) NetworkCounts(_ context.Context, now time.Time) (NetworkCounts, error)

func (*Fake) OpenJobs 

func (f *Fake) OpenJobs(_ context.Context, capability string, limit int) ([]JobRow, error)

func (*Fake) PeersForSample 

func (f *Fake) PeersForSample(_ context.Context, sampleID string) ([]PeerRow, error)

func (*Fake) PurgeDedupOlderThan 

func (f *Fake) PurgeDedupOlderThan(_ context.Context, days int) (int64, error)

PurgeDedupOlderThan drops rotating dedup ledger entries older than the retention window; accumulated counts and unique-bucket sets stay frozen, matching pg.go.

func (*Fake) PutShard 

func (f *Fake) PutShard(_ context.Context, key, etag, shardJSON string) error

func (*Fake) PutSnapshot 

func (f *Fake) PutSnapshot(_ context.Context, purl, symbol, snapshotJSON string) error

func (*Fake) ReceiptsForSample 

func (f *Fake) ReceiptsForSample(_ context.Context, sampleID string) ([]ReceiptRow, error)

func (*Fake) SaveCase 

func (f *Fake) SaveCase(_ context.Context, c domain.Case) error

func (*Fake) SaveIdentity 

func (f *Fake) SaveIdentity(_ context.Context, login string, githubID int64, tokenHash, apiTokenHash string) error

func (*Fake) SaveReceipt 

func (f *Fake) SaveReceipt(_ context.Context, r ReceiptRow) error

func (*Fake) SaveSample 

func (f *Fake) SaveSample(_ context.Context, s SampleRow) error

func (*Fake) SetSampleStatus 

func (f *Fake) SetSampleStatus(_ context.Context, sampleID, status string) error

func (*Fake) SetStatsDaily 

func (f *Fake) SetStatsDaily(_ context.Context, day string, statsJSON string) error

func (*Fake) ShardWrites added in v0.1.1 

func (f *Fake) ShardWrites() int

ShardWrites counts PutShard calls so tests can assert that an idle aggregation pass does no work — the whole point of incremental rebuilds.

func (*Fake) UpsertFailureCluster 

func (f *Fake) UpsertFailureCluster(_ context.Context, c ClusterRow) error

func (*Fake) UpsertPackage 

func (f *Fake) UpsertPackage(_ context.Context, p PackageRow) error

type IdentityRow 

type IdentityRow struct {
	Login        string
	GithubID     int64
	Display      string
	TokenHash    string
	APITokenHash string
	CreatedAt    time.Time
}

IdentityRow is one identities-table row (persistent seeder/verifier identity; automatic evidence never uses these).

type JobRow 

type JobRow struct {
	ID          int64
	SampleID    string
	Reason      string // "cross" | "matrix"
	WantEnvJSON string // "" ⇒ NULL (any environment)
	Status      string // open | claimed | done
	ClaimedBy   string
	ClaimedAt   time.Time
	CreatedAt   time.Time
}

JobRow is one verification_jobs-table row.

type Migration 

type Migration struct {
	Version    string
	Statements []string
}

Migration is one embedded .sql file, split into executable statements. Version is the file name; lexicographic order is execution order (0001_..., 0002_...).

func LoadMigrations 

func LoadMigrations() ([]Migration, error)

LoadMigrations reads the embedded migrations directory in order.

type NetworkCounts 

type NetworkCounts struct {
	// Peers is the distinct anonymous peer buckets that contributed
	// evidence in the current epoch: who is actually using the network
	// today. Buckets rotate daily (goal.md §8.6), so a longer window
	// would count one person many times.
	Peers int64
	// ProjectsMonth is the distinct project buckets seen this calendar
	// month. Those buckets rotate MONTHLY rather than daily, so counting
	// them over the month is the longest honest participation window the
	// identity scheme allows — a peer id summed across days would count
	// one person once per day and report thirty where there is one.
	ProjectsMonth   int64
	Packages        int64 // distinct (ecosystem, name) pairs
	Symbols         int64 // distinct non-empty symbol families with evidence
	Observations    int64 // total aggregated observation count
	VerifiedSamples int64 // contract-verified or independently reproduced
	// ServingPeers is the P2P blob tracker population — nodes that opted
	// into peerListen. Most peers contribute evidence without ever
	// serving blobs, so this is much smaller and is not the headline.
	ServingPeers int64
}

type PG 

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

PG implements Store on PostgreSQL via pgx/v5.

It manages its own small *pgx.Conn pool (connPool below) instead of pgxpool: the repo pins its dependency set and pgxpool's puddle dependency is not part of it. The pool is deliberately tiny — the production target is a 2GB VM with max_connections 40.

func Open 

func Open(ctx context.Context, dsn string) (*PG, error)

Open connects to PostgreSQL and returns a ready Store. It validates the DSN and dials one connection eagerly so misconfiguration fails fast.

func (*PG) AnnouncePeer 

func (p *PG) AnnouncePeer(ctx context.Context, peer PeerRow) error

func (*PG) ChangedSince added in v0.1.1 

func (p *PG) ChangedSince(ctx context.Context, since time.Time) (Changes, error)

ChangedSince implements the incremental-rebuild query. Both halves are index-friendly single scans; on an idle network they return nothing and the aggregation pass does no work at all.

func (*PG) ClaimJob 

func (p *PG) ClaimJob(ctx context.Context, id int64, peerID string) (bool, error)

func (*PG) Close 

func (p *PG) Close()

Close releases every pooled connection.

func (*PG) CompleteJob 

func (p *PG) CompleteJob(ctx context.Context, id int64) error

func (*PG) CreateJob 

func (p *PG) CreateJob(ctx context.Context, j JobRow) (int64, error)

func (*PG) EvidenceForTarget 

func (p *PG) EvidenceForTarget(ctx context.Context, purl, symbol string) ([]EvidenceRow, error)

func (*PG) ExpirePeers 

func (p *PG) ExpirePeers(ctx context.Context, now time.Time) (int64, error)

func (*PG) GetLatestStats 

func (p *PG) GetLatestStats(ctx context.Context) (string, bool, error)

func (*PG) GetPackage 

func (p *PG) GetPackage(ctx context.Context, purl string) (PackageRow, bool, error)

func (*PG) GetSample 

func (p *PG) GetSample(ctx context.Context, sampleID string) (SampleRow, bool, error)

func (*PG) GetShard 

func (p *PG) GetShard(ctx context.Context, key string) (string, string, bool, error)

func (*PG) GetSnapshot 

func (p *PG) GetSnapshot(ctx context.Context, purl, symbol string) (string, bool, error)

func (*PG) HotShardKeys 

func (p *PG) HotShardKeys(ctx context.Context, limit int) ([]string, error)

HotShardKeys ranks built shards by the samples they carry first, then by the observation volume of the packages they cover. Keys with no shard row are dropped: handing a client a key that 404s wastes the one warm-up it gets.

func (*PG) IdentityByAPIToken 

func (p *PG) IdentityByAPIToken(ctx context.Context, apiTokenHash string) (IdentityRow, bool, error)

func (*PG) IngestBatches 

func (p *PG) IngestBatches(ctx context.Context, batches []domain.ObservationBatch) (int, []RejectedBatch, error)

IngestBatches validates each batch and delta-merges the valid ones with the semantics documented in merge.go: per (bucket, agg row, epoch) only count growth is added, so a re-sent identical batch adds exactly 0. IngestBatches applies a whole request in ONE transaction. Per-batch transactions meant a 500-batch upload paid 500 commits: on a small instance that took longer than the client's timeout, so a first sync after scanning a machine could never finish. One commit is also the honest unit — a request lands completely or not at all, and the delta-merge makes a retried request add nothing twice.

func (*PG) JobsForSample 

func (p *PG) JobsForSample(ctx context.Context, sampleID string) ([]JobRow, error)

JobsForSample lists every job for a sample regardless of status.

func (*PG) ListFailureClusters 

func (p *PG) ListFailureClusters(ctx context.Context, packageName string) ([]ClusterRow, error)

func (*PG) ListPackageVersions 

func (p *PG) ListPackageVersions(ctx context.Context, ecosystem, name string) ([]PackageRow, error)

func (*PG) ListSamples 

func (p *PG) ListSamples(ctx context.Context, limit int) ([]SampleRow, error)

func (*PG) ListSnapshotTargets 

func (p *PG) ListSnapshotTargets(ctx context.Context) ([]SnapshotTarget, error)

func (*PG) Migrate 

func (p *PG) Migrate(ctx context.Context) error

Migrate applies the embedded migrations (see migrate.go).

func (*PG) NetworkCounts 

func (p *PG) NetworkCounts(ctx context.Context, now time.Time) (NetworkCounts, error)

NetworkCounts computes the raw /v1/stats numbers in one round trip.

Peers counts the distinct anonymous peer buckets that contributed evidence in the current epoch — the people actually using the network today. It deliberately does NOT count rows in the peers table: that is the P2P blob tracker, which a node only joins by opting into peerListen, so it read 0 while evidence was arriving every minute. Peer identities rotate daily by design (goal.md §8.6), so one day is the longest window that can be counted without inflating one user into many.

func (*PG) OpenJobs 

func (p *PG) OpenJobs(ctx context.Context, capability string, limit int) ([]JobRow, error)

OpenJobs lists claimable jobs. A job that pins want_env.sandboxCapability only matches peers reporting that capability; capability "" matches all.

func (*PG) PeersForSample 

func (p *PG) PeersForSample(ctx context.Context, sampleID string) ([]PeerRow, error)

func (*PG) PurgeDedupOlderThan 

func (p *PG) PurgeDedupOlderThan(ctx context.Context, days int) (int64, error)

PurgeDedupOlderThan deletes dedup buckets whose epoch day is older than the retention window. Aggregate counts are kept — only the rotating bucket linkage is erased (goal.md §14.4), so unique_*_buckets freeze at their accumulated values rather than shrinking retroactively.

func (*PG) PutShard 

func (p *PG) PutShard(ctx context.Context, key, etag, shardJSON string) error

func (*PG) PutSnapshot 

func (p *PG) PutSnapshot(ctx context.Context, purl, symbol, snapshotJSON string) error

func (*PG) ReceiptsForSample 

func (p *PG) ReceiptsForSample(ctx context.Context, sampleID string) ([]ReceiptRow, error)

func (*PG) SaveCase 

func (p *PG) SaveCase(ctx context.Context, cse domain.Case) error

func (*PG) SaveIdentity 

func (p *PG) SaveIdentity(ctx context.Context, login string, githubID int64, tokenHash, apiTokenHash string) error

func (*PG) SaveReceipt 

func (p *PG) SaveReceipt(ctx context.Context, r ReceiptRow) error

func (*PG) SaveSample 

func (p *PG) SaveSample(ctx context.Context, s SampleRow) error

func (*PG) SetSampleStatus 

func (p *PG) SetSampleStatus(ctx context.Context, sampleID, status string) error

func (*PG) SetStatsDaily 

func (p *PG) SetStatsDaily(ctx context.Context, day string, statsJSON string) error

func (*PG) UpsertFailureCluster 

func (p *PG) UpsertFailureCluster(ctx context.Context, cl ClusterRow) error

func (*PG) UpsertPackage 

func (p *PG) UpsertPackage(ctx context.Context, pkg PackageRow) error

type PackageRow 

type PackageRow struct {
	PURL       string
	Ecosystem  string
	Name       string
	Version    string
	Major      string
	Publicness string    // PUBLIC | PRIVATE | UNKNOWN
	CheckedAt  time.Time // zero = publicness never checked
	FirstSeen  time.Time
	LastSeen   time.Time
}

PackageRow is one packages-table row.

type PeerRow 

type PeerRow struct {
	PeerID           string
	Addr             string
	Port             int
	CapabilitiesJSON string // JSON array, e.g. ["CONTAINER_RUN"]
	SampleIDsJSON    string // JSON array of sample ids the peer seeds
	AnnouncedAt      time.Time
	ExpiresAt        time.Time
}

PeerRow is one peers-table row (tracker state, TTL-expired).

type ReceiptRow 

type ReceiptRow struct {
	ReceiptID      string
	SampleID       string
	PeerID         string
	EnvHash        string
	ReceiptJSON    string
	ContractResult string // PASS | FAIL | SKIPPED | ""
	CreatedAt      time.Time
}

ReceiptRow is one receipts-table row. ReceiptJSON is the full signed VerificationReceipt document.

type RejectedBatch 

type RejectedBatch struct {
	Index  int    `json:"index"`
	Reason string `json:"reason"`
}

RejectedBatch reports one refused batch of an ingest call, mirroring the C5 wire shape {index, reason}.

type SampleRow 

type SampleRow struct {
	SampleID     string
	CaseID       string // "" ⇒ NULL (case unknown)
	ManifestJSON string
	Status       string // PUBLISHED | CROSS_PASS | MATRIX_PASS | STABLE
	OriginSeeder string
	License      string
	SizeBytes    int64
	HotScore     float64
	CreatedAt    time.Time
}

SampleRow is one samples-table row. ManifestJSON is the csx.json document.

type ServerConfig 

type ServerConfig struct {
	DSN                string        // CSX_DSN — PostgreSQL connection string, required to run
	Listen             string        // CSX_LISTEN — default ":8080"
	BlobDir            string        // CSX_BLOB_DIR — sample artifact blob directory
	PublicURL          string        // CSX_PUBLIC_URL — canonical external base URL
	PublicCheck        string        // CSX_PUBLIC_CHECK — "strict" (default) | "trust" (dev/e2e only)
	SnapshotInterval   time.Duration // CSX_SNAPSHOT_INTERVAL — default 5m
	GithubClientID     string        // CSX_GITHUB_CLIENT_ID — empty ⇒ device flow returns 501
	GithubClientSecret string        // CSX_GITHUB_CLIENT_SECRET
	// BlobBudgetBytes caps total artifact storage (CSX_BLOB_BUDGET_MB, 0 =
	// unlimited). Sample upload is anonymous, so this is the only ceiling
	// on how much disk an unauthenticated caller can take — and the volume
	// is shared with PostgreSQL, which stops working when it fills.
	BlobBudgetBytes int64
}

ServerConfig collects every csx-server environment setting (plan contract C9). It lives here so cmd/csx-server and the Wave C HTTP layer share one definition.

func ConfigFromEnv 

func ConfigFromEnv() ServerConfig

ConfigFromEnv reads the CSX_* server environment with safe defaults. DSN deliberately has no default: pointing at a database must be explicit.

type SnapshotTarget 

type SnapshotTarget struct {
	PURL   string
	Symbol string
}

SnapshotTarget is one (purl, symbol) pair that has evidence and therefore needs a compatibility snapshot. Symbol "" is the package-level row.

type Store 

type Store interface {
	// IngestBatches validates and delta-merges observation batches
	// (semantics in merge.go). Invalid batches are reported in rejected and
	// skipped; a storage failure aborts with err.
	IngestBatches(ctx context.Context, batches []domain.ObservationBatch) (accepted int, rejected []RejectedBatch, err error)

	UpsertPackage(ctx context.Context, p PackageRow) error
	GetPackage(ctx context.Context, purl string) (PackageRow, bool, error)
	ListPackageVersions(ctx context.Context, ecosystem, name string) ([]PackageRow, error)

	GetSnapshot(ctx context.Context, purl, symbol string) (snapshotJSON string, ok bool, err error)
	PutSnapshot(ctx context.Context, purl, symbol, snapshotJSON string) error
	// ChangedSince reports what moved since a point in time, so the
	// aggregation pass can rebuild only what is stale instead of the whole
	// network on every tick.
	ChangedSince(ctx context.Context, since time.Time) (Changes, error)

	// ListSnapshotTargets returns the distinct (purl, symbol) pairs that
	// have aggregated evidence.
	ListSnapshotTargets(ctx context.Context) ([]SnapshotTarget, error)
	EvidenceForTarget(ctx context.Context, purl, symbol string) ([]EvidenceRow, error)

	SaveCase(ctx context.Context, c domain.Case) error
	SaveSample(ctx context.Context, s SampleRow) error
	GetSample(ctx context.Context, sampleID string) (SampleRow, bool, error)
	ListSamples(ctx context.Context, limit int) ([]SampleRow, error)
	SetSampleStatus(ctx context.Context, sampleID, status string) error

	SaveReceipt(ctx context.Context, r ReceiptRow) error
	ReceiptsForSample(ctx context.Context, sampleID string) ([]ReceiptRow, error)

	// OpenJobs lists open verification jobs a peer with the given sandbox
	// capability may claim ("" ⇒ any). Jobs whose want_env pins a
	// sandboxCapability only match that capability.
	OpenJobs(ctx context.Context, capability string, limit int) ([]JobRow, error)
	// JobsForSample lists every verification job (any status) for a sample,
	// oldest first — the aggregation builder uses it to avoid creating
	// duplicate matrix jobs.
	JobsForSample(ctx context.Context, sampleID string) ([]JobRow, error)
	CreateJob(ctx context.Context, j JobRow) (int64, error)
	// ClaimJob atomically moves an open job to claimed; false means someone
	// else got there first (or the job is gone).
	ClaimJob(ctx context.Context, id int64, peerID string) (bool, error)
	CompleteJob(ctx context.Context, id int64) error

	AnnouncePeer(ctx context.Context, p PeerRow) error
	PeersForSample(ctx context.Context, sampleID string) ([]PeerRow, error)
	ExpirePeers(ctx context.Context, now time.Time) (removed int64, err error)

	PutShard(ctx context.Context, key, etag, shardJSON string) error
	GetShard(ctx context.Context, key string) (etag, shardJSON string, ok bool, err error)
	// HotShardKeys lists the shard keys worth warming first, most active
	// first. A fresh install has no local package history, so this is the
	// only thing that fills its cache before the first search.
	HotShardKeys(ctx context.Context, limit int) ([]string, error)

	SaveIdentity(ctx context.Context, login string, githubID int64, tokenHash, apiTokenHash string) error
	IdentityByAPIToken(ctx context.Context, apiTokenHash string) (IdentityRow, bool, error)

	UpsertFailureCluster(ctx context.Context, c ClusterRow) error
	ListFailureClusters(ctx context.Context, packageName string) ([]ClusterRow, error)

	SetStatsDaily(ctx context.Context, day string, statsJSON string) error
	GetLatestStats(ctx context.Context) (statsJSON string, ok bool, err error)
	// NetworkCounts computes the raw stats-rollup numbers as of now.
	NetworkCounts(ctx context.Context, now time.Time) (NetworkCounts, error)

	// PurgeDedupOlderThan deletes rotating dedup buckets older than the
	// given number of days (goal.md §14.4: 30). Aggregates keep their
	// accumulated counts; only the bucket↔evidence linkage is erased.
	PurgeDedupOlderThan(ctx context.Context, days int) (removed int64, err error)

	Close()
}

Store is everything csx-server needs from PostgreSQL. Handlers depend on this interface, never on pgx, so tests can substitute fakes.

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.