postgres

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Nov 3, 2026 License: AGPL-3.0 Imports: 26 Imported by: 0

Documentation

Overview

Package postgres is the PostgreSQL adapter under internal/storage.

M1: package skeleton (doc.go only). M3 lands the storage layer:

  • migrations runner (golang-migrate v4, database/pgx/v5 driver, source/iofs embedded migrations; one atomic transaction per migration) — migrate.go, lock.go
  • bounded advisory-lock waiting (session-level statement_timeout; DB_MIGRATION_LOCK_TIMEOUT on expiry) — lock.go
  • administrative migration commands backing `gseer db migrate {status,diagnose,force}` — admin.go
  • pgxpool lifecycle (created AFTER migrations succeed) — pool.go
  • reachability + PG version checks (E_DB_UNREACHABLE, PD-8) — reachability.go, version.go
  • credential redaction for diagnostics (M3 discipline until SEC-6 lands in M10) — redact.go
  • doctor migration-status checks (QS-4) — doctor.go

M4 lands the repositories and the search_index union view.

Search ≠ Scan (ADR-0010): this package contains no egress code and imports no collector package. The M1 import-graph test pins that invariant; it must hold for every future change here.

Index

Constants

View Source 
const (
	StateUninitialized = "UNINITIALIZED" // no schema_migrations table yet
	StateAtHead        = "AT_HEAD"
	StateBehind        = "BEHIND"
	StateAhead         = "AHEAD"
	StateDirty         = "DIRTY"
)

Migration states reported by GetMigrationStatus.

View Source 
const (
	// CodeDBUnreachable — PostgreSQL is not reachable (SPEC-03 §7).
	CodeDBUnreachable = "E_DB_UNREACHABLE"

	// CodeDBVersion — the PostgreSQL major version is below the
	// supported floor (PD-8: >= 15).
	CodeDBVersion = "E_DB_VERSION"

	// CodeDBMigrationFailed — a migration failed mid-execution; its
	// transaction was rolled back and the version is marked dirty.
	CodeDBMigrationFailed = "E_DB_MIGRATION_FAILED"

	// CodeDBMigrationDirty — the schema_migrations table is in a dirty
	// state from a previously failed migration; recovery is operator-
	// driven via `gseer db migrate diagnose` + `force`.
	CodeDBMigrationDirty = "E_DB_MIGRATION_DIRTY"

	// CodeDBMigrationLockTimeout — the migration advisory lock could not
	// be acquired within the configured timeout (owner directive §1,
	// M3 authorization round).  Named per the owner's exact identifier.
	CodeDBMigrationLockTimeout = "DB_MIGRATION_LOCK_TIMEOUT"

	// CodeDBUsage — an administrative command was invoked with invalid
	// arguments (e.g. force --version beyond head).
	CodeDBUsage = "E_DB_USAGE"

	// CodeDBSuppressed — a write was refused because the record matches
	// an enabled suppression rule (migration 0005; M13).  Surfaced by
	// the DB backstop trigger (custom ERRCODE GSUP1) when a write path
	// bypasses the Go gate; the gate itself skips suppressed facts
	// cleanly and reports them via IngestStats.Suppressed.
	CodeDBSuppressed = "E_DB_SUPPRESSED"
)

Stable storage-layer error codes.

View Source 
const (
	DefaultPoolMaxConns        = 4
	DefaultPoolMinConns        = 1
	DefaultPoolMaxConnLifetime = time.Hour
	DefaultPoolMaxConnIdleTime = 15 * time.Minute
	DefaultPoolHealthCheck     = 30 * time.Second
)

Default pool settings (conservative for the v0.0.1 dev/preview; owner decision D-M3-9).

View Source 
const RetractLimit = 100_000

RetractLimit bounds one retraction run. A rule that matches more assets than this is a scope mistake (a /8, a TLD-wide suffix); the operator narrows the rule instead of the tool silently deleting millions of rows.

View Source 
const ServerVersionNumQuery = `SELECT current_setting('server_version_num')::int`

ServerVersionNumQuery returns the numeric server version (e.g. 170002 for 17.2); the major is num / 10000.

Variables

View Source 
var ErrSourceNotFound = errors.New("data source not registered")

ErrSourceNotFound — registry lookup miss.

Functions

func CheckPGVersion 

func CheckPGVersion(ctx context.Context, rawDSN string, minMajor int) (int, error)

CheckPGVersion opens a connection, verifies the server major version is at least minMajor (PD-8 floor), and closes. It returns the detected major version on success (even when too low — callers may report it) and an E_DB_VERSION DBError when below the floor.

func CheckReachable 

func CheckReachable(ctx context.Context, rawDSN string) error

CheckReachable verifies the database is reachable. It returns nil on success or an E_DB_UNREACHABLE DBError.

func ForceMigrationVersion 

func ForceMigrationVersion(rawDSN string, lockTimeoutMs, version int) error

ForceMigrationVersion sets schema_migrations to (version, dirty=false) WITHOUT executing any SQL migration (owner directive §2: the force operation "only changes migration version metadata", "never executes or repairs SQL migrations").

The CLI layer (internal/cli/db.go) adds the confirmation gate; this primitive enforces the argument contract: 0 <= version <= head.

func HeadVersion 

func HeadVersion() (uint, error)

HeadVersion returns the highest embedded migration version. Doctor compares it against the database's recorded version to classify AT_HEAD / BEHIND / AHEAD (owner directive §2.5).

func MigrationFiles 

func MigrationFiles() ([]string, error)

MigrationFiles returns the embedded up-migration file names in ascending version order (used by status/diagnose output).

func MustRedactDSN 

func MustRedactDSN(rawDSN string) string

MustRedactDSN is the panic-free convenience wrapper for rendering paths where a malformed DSN means the diagnostic omits the DSN segment rather than failing the whole command. It returns the redacted form, or "" when the DSN cannot be parsed.

func NewPool 

func NewPool(ctx context.Context, rawDSN string) (*pgxpool.Pool, error)

NewPool builds a pgxpool with the GSEER defaults. The DSN is the operator's raw connection string (never logged — redact.go).

func PGMajorVersion 

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

PGMajorVersion queries the connected server's major version.

func PingPool 

func PingPool(ctx context.Context, pool *pgxpool.Pool) error

PingPool is a helper for the server-start readiness path: the pool is created after migrations; this re-verifies liveness.

func RedactDSN 

func RedactDSN(rawDSN string) (string, error)

RedactDSN strips the userinfo (user, password) from a URL-shaped PostgreSQL DSN and returns the display-safe form: 

postgres://host:port/dbname?sslmode=...

The M2 fmtDSN builds DSNs via net/url, so operator-provided database.dsn values are expected to be URL-shaped too (documented in .env.example and docs/INSTALL.md). A non-URL DSN returns an error — callers then fall back to omitting the DSN from output entirely (never a partial or best-effort redaction).

func RunMigrations 

func RunMigrations(rawDSN string, lockTimeoutMs int) (applied int, head uint, err error)

RunMigrations applies pending embedded migrations to the database at rawDSN. It returns the number of migrations applied (0 when the schema was already at head — idempotent re-run).

Error translation (owner directives):

  • advisory lock not acquired within the timeout → DB_MIGRATION_LOCK_TIMEOUT (lock.go);
  • a migration failed mid-execution → E_DB_MIGRATION_FAILED (the failed migration's transaction was rolled back; the version is dirty — recovery via `gseer db migrate diagnose`);
  • the database was already dirty → E_DB_MIGRATION_DIRTY.

Types

type CheckOutcome 

type CheckOutcome string

CheckOutcome classifies a doctor check line. 

const (
	CheckPass CheckOutcome = "PASS"
	CheckFail CheckOutcome = "FAIL"
	CheckHint CheckOutcome = "HINT"
	CheckSkip CheckOutcome = "SKIP"
)

type CheckResult 

type CheckResult struct {
	Name    string
	Outcome CheckOutcome
	Detail  string // human detail; built from redacted DSN only
	Fix     string // exact next command on FAIL/HINT
	// Err carries the structured DBError for FAIL outcomes (nil other-
	// wise) so the CLI can render the full Code/Why/Try shape.
	Err error
}

CheckResult is one doctor line: what was checked, how it fared, and the fix instruction on failure.

func DoctorDBChecks 

func DoctorDBChecks(ctx context.Context, rawDSN string, minMajor int) []CheckResult

DoctorDBChecks runs the M3 database checks in QS-4 order: reachable → PG version → migrations. Each check that depends on a live connection SKIPs when the reachability check fails (doctor never hides the root cause behind a cascade of secondary failures).

type CollectionJob 

type CollectionJob struct {
	ID         int64
	SourceID   int64
	Kind       string
	Status     string
	StartedAt  time.Time
	FinishedAt *time.Time
	Stats      map[string]any
	Error      string
}

CollectionJob is one connector run row.

type DBError 

type DBError struct {
	Code   string
	Detail string
	Why    string
	Next   string
}

DBError is the structured storage-layer error. It mirrors the config.Error shape so the CLI renders one consistent style.

func (*DBError) Error 

func (e *DBError) Error() string

Error renders per the SPEC-03 §7 style guide.

type DataSourceRow 

type DataSourceRow struct {
	ID         int64
	Name       string
	Connector  string
	TermsRef   string
	ConfigJSON []byte
	Enabled    bool
	CreatedAt  time.Time
}

DataSourceRow mirrors the data_sources row with the config kept as raw JSON (decoded by the connector layer).

type DeadLetterRecord 

type DeadLetterRecord struct {
	ID        int64
	SourceID  int64
	Payload   map[string]any
	Reason    string
	Attempts  int
	FirstSeen time.Time
	LastSeen  time.Time
}

DeadLetterRecord is one pending dead-letter row handed to the re-drive.

type Diagnosis 

type Diagnosis struct {
	Status        *MigrationStatus
	FailedFile    string   // embedded up-file for the dirty version, if any
	InspectionSQL []string // suggested read-only inspection queries
	ForceHint     string   // the exact force command, when applicable
}

Diagnosis is the read-only recovery guidance for a dirty database.

func DiagnoseMigration 

func DiagnoseMigration(ctx context.Context, rawDSN string) (*Diagnosis, error)

DiagnoseMigration assembles the operator-facing recovery guidance for a dirty (or otherwise non-head) database. Read-only.

type Fact 

type Fact = storage.Fact

Fact, FactResult and IngestStats moved to internal/storage in v0.0.2 (M14: the collector framework names them without importing the pgx-heavy store package). These aliases keep every reference in this package — and every caller — compiling unchanged; []storage.Fact IS []Fact because the types are identical aliases.

type FactResult 

type FactResult = storage.FactResult

FactResult reports what one IngestFact applied (see storage.FactResult).

type IngestStats 

type IngestStats = storage.IngestStats

IngestStats aggregates an IngestFacts run (see storage.IngestStats).

type MigrationStatus 

type MigrationStatus struct {
	State   string // one of the State* constants
	Version int64  // recorded version; -1 when uninitialized
	Dirty   bool
	Head    uint     // highest embedded migration version
	Pending []string // up-file names above the recorded version
}

MigrationStatus is the read-only migration state snapshot.

func GetMigrationStatus 

func GetMigrationStatus(ctx context.Context, rawDSN string) (*MigrationStatus, error)

GetMigrationStatus reads schema_migrations (if it exists) and classifies the state against the embedded head version. Read-only: SELECT queries only.

type PurgeStats 

type PurgeStats struct {
	Observations, Services, Assets int64
	FixtureSourcePresent           bool
}

PurgeStats reports what PurgeFixtureData removed.

func PurgeFixtureData 

func PurgeFixtureData(ctx context.Context, pool *pgxpool.Pool) (PurgeStats, error)

PurgeFixtureData removes every row the fixture loader wrote, in ONE transaction, scoped so real observations are NEVER deleted:

  1. the observation DELETE binds to the fixture source's registry id AND the provenance='fixture' tag — a real source's observation is unreachable by both predicates;
  2. services are deleted only if a fixture observation referenced them AND no observation references them anymore AFTER the observation deletion (separate statement so it sees the post-deletion state) — a real source's service always has a live reference and survives;
  3. assets are deleted only if they are fixture-sourced AND have no observation AND no service left — a real source's asset keeps its merged sources[] and/or live references and survives;
  4. surviving assets have 'fixture' stripped from sources[] (registry honesty: the fixture source no longer claims them).

The fixture source registry row itself is left in place (it is configuration, not data); FixtureSourcePresent reports whether it existed. A database with no fixture source row is a no-op success (nothing was ever loaded), not an error.

type RetractStats 

type RetractStats struct {
	AssetsRetracted     int
	ServicesRemoved     int
	ObservationsRemoved int
	AuditRows           int
}

RetractStats reports one RetractSuppressed run.

type Store 

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

Store is the M4 write-side repository over the application pool (the pool is created after migrations succeed — D-M3-8).

func NewStore 

func NewStore(pool *pgxpool.Pool) *Store

NewStore returns a Store over pool.

func (*Store) AddDeadLetter 

func (s *Store) AddDeadLetter(ctx context.Context, sourceID int64, payload map[string]any, reason string) error

AddDeadLetter preserves a record that could not be ingested. An identical pending payload (same source + reason + payload hash) bumps attempts/last_seen instead of adding a new row (bounded growth under retry storms).

func (*Store) AddSuppression 

func (s *Store) AddSuppression(ctx context.Context, in SuppressInput) (rule suppress.Rule, existed bool, err error)

AddSuppression inserts a rule (value canonicalized first) and returns it with its database id. existed=true when an identical (kind, value) rule was already present — the existing row is returned unchanged (idempotent re-adds are not audit events).

func (*Store) DeadLetterStats 

func (s *Store) DeadLetterStats(ctx context.Context) (map[int64]int, error)

DeadLetterStats counts the pending dead letters per source.

func (*Store) DiscardDeadLetters 

func (s *Store) DiscardDeadLetters(ctx context.Context, sourceID int64, limit int) (int64, error)

DiscardDeadLetters marks pending rows discarded (operator's explicit judgment that a record will never apply — e.g. a malformed payload from a retired source). Returns the number of rows marked.

func (*Store) EnsureFixtureSource 

func (s *Store) EnsureFixtureSource(ctx context.Context) (sourceID int64, err error)

EnsureFixtureSource upserts the fixture source registry row and returns its id (D-M3-1: the fixture source is itself a data_sources row). Re-running refreshes terms_ref but deliberately leaves enabled untouched: v0.0.1 sources are never enabled (SPEC-09 §1).

func (*Store) EnsureSource 

func (s *Store) EnsureSource(ctx context.Context, name, connector, termsRef string, cfg map[string]any) (int64, error)

EnsureSource upserts a data_sources registry row and returns its id. Enablement is NEVER flipped here: it is an explicit operator decision (config + review status), never a side effect of registration.

func (*Store) FailDeadLetter 

func (s *Store) FailDeadLetter(ctx context.Context, id int64, failure string) error

FailDeadLetter records a failed re-drive attempt (attempts bump, still pending — the record is never silently dropped).

func (*Store) FinishCollectionJob 

func (s *Store) FinishCollectionJob(ctx context.Context, jobID int64, status string, stats map[string]any, errMsg string) error

FinishCollectionJob closes a job with status + counters + error text.

func (*Store) GetCheckpoint 

func (s *Store) GetCheckpoint(ctx context.Context, sourceID int64, key string) (string, error)

GetCheckpoint returns the stored cursor for (source, key); "" when none.

func (*Store) GetSource 

func (s *Store) GetSource(ctx context.Context, name string) (DataSourceRow, error)

GetSource loads one registered source by name.

func (*Store) IndexSeedValues 

func (s *Store) IndexSeedValues(ctx context.Context, kind string, limit int) ([]string, error)

IndexSeedValues returns up to limit asset values of the given kind, freshest first (seed_mode "index": the connector enriches identifiers ALREADY in the operator's index — no third-party target list is ever consulted; P1 honesty contract 1).

func (*Store) IngestFact 

func (s *Store) IngestFact(ctx context.Context, f *Fact) (FactResult, error)

IngestFact applies one Fact atomically: asset upsert → optional service upsert → observation insert, all in ONE transaction. The observation write is last, so a failure there (e.g. an FK violation for an unknown source) rolls the whole fact back — a service can never be left without its asset, nor an observation without its service.

Suppression gate (M13): the fact is checked against the compiled rule set BEFORE any write — source exclusion first, then asset rules. A suppressed fact writes nothing (not even asset metadata): it returns FactResult{Suppressed: true} with a nil error, and one ingest-block row lands in suppression_applications.

Deliberate side effect: the caller's f.Observation.AssetID, f.Service.AssetID and — when a service is present — f.Observation.ServiceID are updated to the resolved row ids, so follow-up code can reference them without a re-read.

func (*Store) IngestFacts 

func (s *Store) IngestFacts(ctx context.Context, facts []Fact) (IngestStats, error)

IngestFacts applies facts one at a time — each fact commits separately because facts are independent: a failure aborts only its own transaction, and a re-run is idempotent (L1/L2 identities) — and aggregates the results. On the first error it stops and returns the stats accumulated so far with the error. Service counters advance only for facts that carry a service (an asset-level fact is neither a service insert nor a service update).

The suppression rule set is compiled ONCE for the whole batch (the gate then applies per fact, writing the ingest-block audit rows); suppressed facts count into IngestStats.Suppressed and are skipped.

func (*Store) InsertObservation 

func (s *Store) InsertObservation(ctx context.Context, o *storage.Observation) (id int64, inserted bool, err error)

InsertObservation inserts one observation at pool scope. The id is CALLER-PROVIDED — observations.id has no sequence default because the fixture corpus loader assigns deterministic ids (D-M3-2) — and is passed through as-is.

Dedup semantics: on any unique-index conflict the already-stored row WINS (its content and provenance are kept) and inserted=false is returned with the surviving row's id, resolved via the L1 delivery lookup first, then the L2 fact lookup. An empty Provenance defaults to the fixture tag (QS-6: unlabeled rows must never look real).

func (*Store) ListSources 

func (s *Store) ListSources(ctx context.Context) ([]DataSourceRow, error)

ListSources returns every registered source.

func (*Store) ListSuppressions 

func (s *Store) ListSuppressions(ctx context.Context) ([]suppress.Rule, error)

ListSuppressions returns every rule (including disabled and expired — audit sees everything; the Matcher applies the filters).

func (*Store) LoadSuppressions 

func (s *Store) LoadSuppressions(ctx context.Context) ([]suppress.Rule, error)

LoadSuppressions reads every rule row (enabled, disabled and expired — the Matcher applies the filters so a rule flip is visible in one place).

func (*Store) PendingDeadLetters 

func (s *Store) PendingDeadLetters(ctx context.Context, sourceID int64, limit int) ([]DeadLetterRecord, error)

PendingDeadLetters lists pending dead-letter rows, oldest activity last (re-drive order: oldest first). sourceID 0 = every source.

func (*Store) RemoveSuppression 

func (s *Store) RemoveSuppression(ctx context.Context, id int64) (bool, error)

RemoveSuppression deletes rule id. Returns false when the id does not exist. NOTE: removing a rule does NOT resurrect previously retracted records — they were deleted with audit rows in suppression_applications.

func (*Store) ResolveDeadLetter 

func (s *Store) ResolveDeadLetter(ctx context.Context, id int64, note string) error

ResolveDeadLetter marks a dead letter resolved after a successful re-drive (the note appends to the reason for the audit trail).

func (*Store) RetractSuppressed 

func (s *Store) RetractSuppressed(ctx context.Context, rule suppress.Rule) (RetractStats, error)

RetractSuppressed removes the already-indexed records matched by rule and writes one retroactive-retract audit row per retracted asset (plus one summary row for source retractions). The whole retraction is ONE transaction: either the matching records leave the index together with their audit rows, or nothing changes.

Scope by rule kind:

  • ip/cidr: kind='ip' assets by exact / containment match
  • domain: kind='domain'|'hostname' assets by exact name
  • domain_suffix: those kinds by strict-label suffix
  • source: every observation of that source, then orphaned services, then the source name stripped from assets.sources[], then assets left with no observation, no service and no source name

A rule matching more than RetractLimit assets is refused (E_DB_USAGE): narrowing a runaway rule is an operator decision, never a silent bulk delete. Retraction deletes by MATCH only — provenance is never read — so the fixture purge (provenance='fixture'-scoped) and this path stay orthogonal (fixture-purge isolation directive).

func (*Store) SetCheckpoint 

func (s *Store) SetCheckpoint(ctx context.Context, sourceID int64, key, cursor string) error

SetCheckpoint stores the cursor for (source, key) upsert-style.

func (*Store) SetFingerprintEngine 

func (s *Store) SetFingerprintEngine(reg *fprule.Registry)

SetFingerprintEngine attaches the rule engine used to enrich ingested observation evidence. A nil engine disables enrichment. The engine only fires when the evidence has http/tls/ssh payload and no source-provided `fingerprint` key (the source always wins).

func (*Store) SetSourceEnabled 

func (s *Store) SetSourceEnabled(ctx context.Context, name string, enabled bool) error

SetSourceEnabled flips the enablement flag — the explicit operator action. The collect runner re-verifies the connector's review verdict before actually fetching, so a hand-flipped row cannot bypass the terms gate.

func (*Store) StartCollectionJob 

func (s *Store) StartCollectionJob(ctx context.Context, sourceID int64, kind string) (int64, error)

StartCollectionJob opens a 'running' job row and returns its id.

func (*Store) UpsertAsset 

func (s *Store) UpsertAsset(ctx context.Context, a *storage.Asset) (id int64, inserted bool, err error)

UpsertAsset upserts one asset at pool scope (single-statement auto-commit; callers wanting per-fact atomicity use IngestFact).

Validation: Kind must be one of ip|domain|hostname, the kind's typed pointer (IP/Domain/Hostname) must be non-nil and Value non-empty — usage errors are E_DB_USAGE, not raw SQLSTATEs from the schema's assets_kind_fields_check.

On return a.ID, a.FirstSeen and a.LastSeen carry the stored row's values (on conflict FirstSeen stays at the original, since RETURNING exposes the new row and first_seen is not in the update set); a.Sources is NOT re-read — the DB-side union may differ in order from the caller's slice. A zero LastSeen means "seen now" (the column is written explicitly, so the year-1 zero time must never reach SQL). ASN 0 = unset (the CASE preserves the stored value on conflict).

func (*Store) UpsertService 

func (s *Store) UpsertService(ctx context.Context, sv *storage.Service) (id int64, inserted bool, err error)

UpsertService upserts one service at pool scope. Validation: AssetID must be a positive asset id, Port 1..65535, Transport tcp|udp and Protocol non-empty (E_DB_USAGE otherwise). On return sv.ID, sv.FirstSeen and sv.LastSeen carry the stored row's values; a zero LastSeen means "seen now".

No suppression gate here BY DESIGN: a service hangs off an existing asset, and the asset's identifier was gated when the asset was written (and is re-gated by the backstop trigger on every asset write). A suppression retraction removes the asset row, cascading its services — so no service can outlive a suppression that covers its asset.

type SuppressInput 

type SuppressInput struct {
	Kind      suppress.Kind
	Value     string // raw (any case, trailing dot, non-punycode); normalized here
	Reason    string
	CreatedBy string
	ExpiresAt *time.Time
}

SuppressInput is one new suppression rule.

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.