postgres

package
v0.1.5 Latest Latest
Warning

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

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

Documentation

Overview

Package postgres provides PostgreSQL-backed repositories (pgx/v5) and applies migrations via goose. Used when SYNAPSE_DB_DSN is set; otherwise the server falls back to in-memory persistence for dev.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func AcquireSingletonLock 

func AcquireSingletonLock(ctx context.Context, pool *pgxpool.Pool, role string) (*pgxpool.Conn, bool, error)

AcquireSingletonLock takes a session-level advisory lock (keyed by role) on a DEDICATED connection the caller holds for the whole process lifetime – releasing it drops the lock. A second instance OF THE SAME ROLE gets ok=false so it can fail fast (the repos still ignore tenant_id, so two same-role writers would race). Returns the held connection (retain it; Release at shutdown), whether the lock was obtained, and any error.

func Connect 

func Connect(ctx context.Context, dsn string) (*pgxpool.Pool, error)

Connect opens a pgx pool with default sizing (back-compat wrapper).

func ConnectPool 

func ConnectPool(ctx context.Context, dsn string, pc PoolConfig) (*pgxpool.Pool, error)

ConnectPool opens a sized pgx pool and verifies connectivity.

func Migrate 

func Migrate(ctx context.Context, dsn string) error

Migrate applies all pending goose migrations (idempotent; tracked in goose_db_version).

Types

type AUPStore 

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

AUPStore persists Acceptable-Use-Policy acceptances to PostgreSQL.

func NewAUPStore 

func NewAUPStore(pool *pgxpool.Pool) *AUPStore

NewAUPStore returns an AUP store backed by the given pool.

func (*AUPStore) Accepted 

func (s *AUPStore) Accepted(ctx context.Context, version string) (bool, error)

Accepted reports whether the given policy version has been accepted.

func (*AUPStore) Save 

func (s *AUPStore) Save(ctx context.Context, a aup.Acceptance) error

Save records an acceptance, idempotent per (actor, version) – this keeps per-actor history (the file dev sink keeps one record per version; both gate identically via Accepted's EXISTS-by-version). RBAC is enforced at the API edge. Should actor identifiers ever become attacker- influenced (e.g. a future external OIDC subject), key idempotency on a UNIQUE(actor, policy_version) constraint instead of a concatenated id.

type AdvisoryRepository 

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

AdvisoryRepository persists the OWNED normalized-advisory store to PostgreSQL. It is GLOBAL reference data (NOT tenant-scoped): the full advisory is a JSONB blob in `advisories`, with one `advisory_affects` row per affected (ecosystem, package) for the indexed ByPackage lookup.

func NewAdvisoryRepository 

func NewAdvisoryRepository(pool *pgxpool.Pool) *AdvisoryRepository

NewAdvisoryRepository returns a repository backed by the given pool.

func (*AdvisoryRepository) ByPackage 

func (r *AdvisoryRepository) ByPackage(ctx context.Context, ecosystem, name string) ([]advisory.Advisory, error)

ByPackage returns the advisories that affect (ecosystem, name), decoded from their JSONB blobs. The caller runs advisory.Match to decide which actually hit the component's version. Deterministic id order.

func (*AdvisoryRepository) Upsert 

func (r *AdvisoryRepository) Upsert(ctx context.Context, a advisory.Advisory) error

Upsert inserts or replaces an advisory by id and rebuilds its (ecosystem, package) index rows, in one transaction. Idempotent – advisories are re-syncable reference data (a re-ingest REPLACES in place), not an append-only ledger. The affected (ecosystem, package) keys must be ingester-normalized per the ports.AdvisoryStore KEY CONTRACT. The full domain advisory round-trips through the JSONB `data` blob.

type AgentDecisionStore 

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

AgentDecisionStore is the durable ports.DecisionStore on PostgreSQL (migration 0032). seq is a monotonic per-session counter (MAX+1). Idempotency is enforced by partial unique indexes – one row per (session_id, action_id) for steps and one stop per session – so a redelivered drive that re-records a decision is a no-op (ON CONFLICT DO NOTHING), never a forked log. Decisions are written by a single driver under the session run lock, so the MAX+1 read/insert pair is race-free.

func NewAgentDecisionStore 

func NewAgentDecisionStore(pool *pgxpool.Pool) *AgentDecisionStore

NewAgentDecisionStore returns a Postgres-backed decision store.

func (*AgentDecisionStore) AppendDecision 

func (s *AgentDecisionStore) AppendDecision(ctx context.Context, d agent.AgentDecision) error

func (*AgentDecisionStore) ListBySession 

func (s *AgentDecisionStore) ListBySession(ctx context.Context, sessionID shared.ID) ([]agent.AgentDecision, error)

type AgentPlanStore 

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

AgentPlanStore is the durable ports.PlanStore on PostgreSQL (migration 0031). One plan per session (session_id UNIQUE → CreatePlan on a redelivery hits a unique violation → ErrConflict, preventing a forked second plan). SavePlan is a guarded UPDATE (… WHERE revision=$expected) that bumps the revision, so a node claim is an atomic compare-and-swap; a lost CAS (0 rows) returns ErrConflict.

func NewAgentPlanStore 

func NewAgentPlanStore(pool *pgxpool.Pool) *AgentPlanStore

NewAgentPlanStore returns a Postgres-backed plan store.

func (*AgentPlanStore) CreatePlan 

func (s *AgentPlanStore) CreatePlan(ctx context.Context, p agent.Plan) error

func (*AgentPlanStore) GetBySession 

func (s *AgentPlanStore) GetBySession(ctx context.Context, sessionID shared.ID) (agent.Plan, bool, error)

func (*AgentPlanStore) SavePlan 

func (s *AgentPlanStore) SavePlan(ctx context.Context, p agent.Plan) error

type AgentSessionStore 

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

AgentSessionStore is the durable ports.AgentSessionStore on PostgreSQL: agent_sessions + agent_messages (migration 0027). The (session_id, seq) primary key is the transcript fork-guard – a duplicate seq is a unique violation → ErrConflict.

func NewAgentSessionStore 

func NewAgentSessionStore(pool *pgxpool.Pool) *AgentSessionStore

NewAgentSessionStore returns a Postgres-backed agent session store.

func (*AgentSessionStore) AppendMessage 

func (s *AgentSessionStore) AppendMessage(ctx context.Context, sessionID shared.ID, seq int, m agent.Message) error

func (*AgentSessionStore) GetSession 

func (s *AgentSessionStore) GetSession(ctx context.Context, id shared.ID) (agent.Session, error)

func (*AgentSessionStore) ListByEngagement 

func (s *AgentSessionStore) ListByEngagement(ctx context.Context, engagementID shared.ID) ([]agent.Session, error)

func (*AgentSessionStore) ListResumable 

func (s *AgentSessionStore) ListResumable(ctx context.Context, staleFor time.Duration, now time.Time, limit int) ([]agent.Session, error)

func (*AgentSessionStore) Messages 

func (s *AgentSessionStore) Messages(ctx context.Context, sessionID shared.ID) ([]agent.Message, error)

func (*AgentSessionStore) SaveSession 

func (s *AgentSessionStore) SaveSession(ctx context.Context, e agent.Session) error

type ApprovalStore 

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

ApprovalStore is the durable ports.ApprovalStore on PostgreSQL: the HITL approval queue (migration 0028). Decide is a guarded UPDATE (… WHERE decision_state= 'pending') so the first decision wins; a second hits 0 rows and returns ErrConflict.

func NewApprovalStore 

func NewApprovalStore(pool *pgxpool.Pool) *ApprovalStore

NewApprovalStore returns a Postgres-backed approval store.

func (*ApprovalStore) Decide 

func (s *ApprovalStore) Decide(ctx context.Context, d agent.ApprovalDecision) error

func (*ApprovalStore) EngagementsWithPending 

func (s *ApprovalStore) EngagementsWithPending(ctx context.Context) ([]shared.ID, error)

func (*ApprovalStore) Enqueue 

func (s *ApprovalStore) Enqueue(ctx context.Context, a agent.ProposedAction) error

func (*ApprovalStore) Get 

func (s *ApprovalStore) Get(ctx context.Context, actionID shared.ID) (agent.ProposedAction, agent.ApprovalDecision, error)

func (*ApprovalStore) Pending 

func (s *ApprovalStore) Pending(ctx context.Context, engagementID shared.ID) ([]agent.ProposedAction, error)

type AuditLog 

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

AuditLog is an append-only, attributable audit log on PostgreSQL.

func NewAuditLog 

func NewAuditLog(pool *pgxpool.Pool) *AuditLog

NewAuditLog returns an audit log backed by the given pool.

func (*AuditLog) List 

func (l *AuditLog) List(ctx context.Context, limit int) ([]ports.AuditEntry, error)

List returns the most recent audit entries (newest first), capped at limit.

func (*AuditLog) Record 

func (l *AuditLog) Record(ctx context.Context, e ports.AuditEntry) error

Record appends an immutable audit entry (INSERT only – never update or delete), chaining it to the previous row. A transaction-scoped advisory lock serializes the read-head/insert so concurrent writers cannot fork the chain. The fork-guard unique index (migration 0033) is defense-in-depth on top of the lock: if the lock is ever bypassed, a concurrent append yields a 23505 unique violation – Record then re-reads the advanced head and re-chains (bounded), parity with the evidence store, rather than surfacing an opaque error. On the normal locked path the conflict is unreachable and the loop runs once.

func (*AuditLog) Verify 

func (l *AuditLog) Verify(ctx context.Context) (audit.Report, error)

Verify re-derives the hash chain over the entire log (oldest-first, id order) and reports whether it is intact. It is an explicit integrity check, so it reads every row rather than a capped window.

type CommentRepository 

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

CommentRepository persists the per-finding comment thread to PostgreSQL.

func NewCommentRepository 

func NewCommentRepository(pool *pgxpool.Pool) *CommentRepository

NewCommentRepository returns a repository backed by the given pool.

func (*CommentRepository) Add 

func (r *CommentRepository) Add(ctx context.Context, c finding.Comment) error

Add inserts a comment (append-only; comments are not edited or deleted in app code).

func (*CommentRepository) ListByEngagementFinding 

func (r *CommentRepository) ListByEngagementFinding(ctx context.Context, engagementID, findingID shared.ID) ([]finding.Comment, error)

ListByEngagementFinding returns a finding's comments oldest-first, scoped to the engagement (no cross-engagement read).

type EngagementRepository 

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

EngagementRepository persists engagements and their scope to PostgreSQL.

func NewEngagementRepository 

func NewEngagementRepository(pool *pgxpool.Pool) *EngagementRepository

NewEngagementRepository returns a repository backed by the given pool.

func (*EngagementRepository) Create 

func (r *EngagementRepository) Create(ctx context.Context, e *engagement.Engagement) error

Create inserts the engagement and its scope targets in one transaction.

func (*EngagementRepository) Delete 

func (r *EngagementRepository) Delete(ctx context.Context, id shared.ID) error

Delete removes an engagement; ON DELETE CASCADE drops its scope, findings, comments, evidence, recon runs, and retests. Idempotent (no error if absent). Used to roll back a partially-materialized import.

func (*EngagementRepository) GetByID 

func (r *EngagementRepository) GetByID(ctx context.Context, id shared.ID) (*engagement.Engagement, error)

GetByID returns the engagement with its full scope WITHOUT a tenant predicate. It is the INTERNAL execution-gate read (see ports.EngagementRepository.GetByID): the scope/window/RoE guard and the worker/agent execution paths, which act on an engagement a queued/authorized run already belongs to. User-facing access uses GetByIDInTenant (below), which adds the tenant predicate that blocks cross-tenant reads.

func (*EngagementRepository) GetByIDInTenant 

func (r *EngagementRepository) GetByIDInTenant(ctx context.Context, tenantID, id shared.ID) (*engagement.Engagement, error)

GetByIDInTenant loads an engagement scoped to tenantID (tenant isolation). A caller tenant of ” (single-tenant / default-tenant admin) matches any row; a non-empty tenant matches only its own – tenant A cannot read tenant B's engagement (ErrNotFound, existence not revealed).

func (*EngagementRepository) GetByProjectID 

func (r *EngagementRepository) GetByProjectID(ctx context.Context, tenantID, projectID shared.ID) (*engagement.Engagement, error)

func (*EngagementRepository) List 

func (r *EngagementRepository) List(ctx context.Context, tenantID shared.ID) ([]*engagement.Engagement, error)

List returns the tenant's engagements, each with its scope loaded (consistent with the in-memory repository; the UI and the scope gate both rely on scope).

func (*EngagementRepository) ProjectContexts 

func (r *EngagementRepository) ProjectContexts(ctx context.Context, tenantID shared.ID, projectIDs []shared.ID) (map[shared.ID]*engagement.Engagement, error)

Update persists an existing engagement aggregate: the engagement row and its full scope target set, replaced atomically in one transaction (E1 scope CRUD + lifecycle). Returns shared.ErrNotFound if the engagement does not exist. Unlike Create's deterministic scope PKs, the replace path uses generated IDs.

func (*EngagementRepository) Update 

func (r *EngagementRepository) Update(ctx context.Context, e *engagement.Engagement) error

type EvidenceStore 

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

EvidenceStore persists the per-engagement hash-chained evidence ledger.

func NewEvidenceStore 

func NewEvidenceStore(pool *pgxpool.Pool) *EvidenceStore

NewEvidenceStore returns a store backed by the given pool.

func (*EvidenceStore) Append 

func (r *EvidenceStore) Append(ctx context.Context, items []evidence.Evidence) error

Append inserts sealed evidence items in order, in one transaction (append-only).

func (*EvidenceStore) Head 

func (r *EvidenceStore) Head(ctx context.Context, engagementID shared.ID) (string, error)

Head returns the most recent sealed hash for an engagement ("" if the chain is empty). A real query error is returned (NOT swallowed as "empty") so the caller never forks the append-only chain on a transient DB failure.

func (*EvidenceStore) ListByEngagement 

func (r *EvidenceStore) ListByEngagement(ctx context.Context, engagementID shared.ID) ([]evidence.Evidence, error)

ListByEngagement returns the engagement's evidence in chain order (oldest first).

type FindingRepository 

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

FindingRepository persists findings to PostgreSQL, deduped per engagement.

func NewFindingRepository 

func NewFindingRepository(pool *pgxpool.Pool) *FindingRepository

NewFindingRepository returns a repository backed by the given pool.

func (*FindingRepository) ListByEngagement 

func (r *FindingRepository) ListByEngagement(ctx context.Context, engagementID shared.ID) ([]finding.Finding, error)

ListByEngagement returns the engagement's findings, highest risk first (CISA KEV, then EPSS x CVSS, then severity).

func (*FindingRepository) ListPublishableByEngagement 

func (r *FindingRepository) ListPublishableByEngagement(ctx context.Context, engagementID shared.ID) ([]finding.Finding, error)

ListPublishableByEngagement returns only the engagement's findings that clear the evidence gate. It reuses ListByEngagement and the single domain rule finding.Publishable, so the publishability policy lives in exactly one place (the domain) rather than being re-encoded in SQL.

func (*FindingRepository) SetAssignee 

func (r *FindingRepository) SetAssignee(ctx context.Context, engagementID, findingID shared.ID, assignee string, expectedVersion int) (finding.Finding, error)

SetAssignee sets the assignee with the same optimistic-concurrency guard.

func (*FindingRepository) SetEvidenceScore 

func (r *FindingRepository) SetEvidenceScore(ctx context.Context, engagementID, findingID shared.ID, score, expectedVersion int) (finding.Finding, error)

SetEvidenceScore sets a finding's evidence score with the same optimistic-concurrency guard as UpdateStatus (the adversarial-verdict path): the row is updated only if version matches, then version is bumped. Note the Upsert ON CONFLICT set deliberately omits evidence_score, so this is the only path that moves it for an already-stored finding.

func (*FindingRepository) UpdateStatus 

func (r *FindingRepository) UpdateStatus(ctx context.Context, engagementID, findingID shared.ID, status finding.Status, expectedVersion int) (finding.Finding, error)

UpdateStatus sets the triage status with optimistic concurrency: the row is updated only if version matches expectedVersion, then version is bumped. On a miss it distinguishes ErrConflict (exists, version moved) from ErrNotFound.

func (*FindingRepository) Upsert 

func (r *FindingRepository) Upsert(ctx context.Context, findings []finding.Finding) error

Upsert inserts or updates findings, deduped on (engagement_id, dedup_key). On conflict it updates the data fields but preserves id, status (triage), assignee, and created_at – and bumps version (a re-scan IS a concurrent change) – so human triage state is never clobbered.

type ImportedSBOMStore 

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

ImportedSBOMStore persists the active imported SBOM per engagement.

func NewImportedSBOMStore 

func NewImportedSBOMStore(pool *pgxpool.Pool) *ImportedSBOMStore

NewImportedSBOMStore returns a Postgres-backed imported-SBOM store.

func (*ImportedSBOMStore) LatestByEngagement 

func (s *ImportedSBOMStore) LatestByEngagement(ctx context.Context, tenantID, engagementID shared.ID) (importedsbom.Record, error)

LatestByEngagement returns the active imported SBOM for a tenant-scoped engagement.

func (*ImportedSBOMStore) SaveActive 

func (s *ImportedSBOMStore) SaveActive(ctx context.Context, record importedsbom.Record) error

SaveActive upserts the active imported SBOM for an engagement.

type JobQueue 

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

JobQueue is the durable ports.JobQueue on PostgreSQL. Claim uses FOR UPDATE SKIP LOCKED so concurrent workers never hand the same job to two claimants, and an expired lease (claimed_until < now) makes a job claimable again – at-least-once delivery with crash recovery.

func NewJobQueue 

func NewJobQueue(pool *pgxpool.Pool, ids ports.IDGenerator) *JobQueue

NewJobQueue returns a Postgres-backed job queue.

func (*JobQueue) Claim 

func (q *JobQueue) Claim(ctx context.Context, visibility time.Duration, kinds ...string) (*ports.QueuedJob, error)

func (*JobQueue) Complete 

func (q *JobQueue) Complete(ctx context.Context, id string) error

func (*JobQueue) Deadletter 

func (q *JobQueue) Deadletter(ctx context.Context, id string) error

func (*JobQueue) Depth 

func (q *JobQueue) Depth(ctx context.Context, kinds ...string) (int, error)

Depth counts not-yet-terminal jobs (queued or claimed) – the durable-backpressure admission signal. 'done' and 'failed' are terminal and excluded. Optional kind filter.

func (*JobQueue) Enqueue 

func (q *JobQueue) Enqueue(ctx context.Context, kind string, payload []byte) (string, error)

func (*JobQueue) Fail 

func (q *JobQueue) Fail(ctx context.Context, id string, retryIn time.Duration) error

func (*JobQueue) Heartbeat 

func (q *JobQueue) Heartbeat(ctx context.Context, id string, extend time.Duration) error

type JudgmentRepository 

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

JudgmentRepository persists AI judgments to PostgreSQL, engagement-scoped.

func NewJudgmentRepository 

func NewJudgmentRepository(pool *pgxpool.Pool) *JudgmentRepository

NewJudgmentRepository returns a repository backed by the given pool.

func (*JudgmentRepository) ListByEngagement 

func (r *JudgmentRepository) ListByEngagement(ctx context.Context, engagementID shared.ID) ([]judgment.Judgment, error)

ListByEngagement returns the engagement's judgments, oldest first (deterministic order).

func (*JudgmentRepository) ListBySubject 

func (r *JudgmentRepository) ListBySubject(ctx context.Context, engagementID, subjectID shared.ID) ([]judgment.Judgment, error)

ListBySubject returns the engagement's judgments about a given subject id, oldest first.

func (*JudgmentRepository) Save 

func (r *JudgmentRepository) Save(ctx context.Context, j judgment.Judgment) error

Save inserts a proposed judgment (idempotent by id; never clobbers an existing row – score/state move only via SetScoreState). The typed claim is stored as its fail-closed discriminated envelope (JSONB).

func (*JudgmentRepository) SetScoreState 

func (r *JudgmentRepository) SetScoreState(ctx context.Context, engagementID, id shared.ID, score int, state judgment.State, expectedVersion int) (judgment.Judgment, error)

SetScoreState moves a judgment's evidence score + state under optimistic concurrency (the verify/accept path): the row updates only if version matches expectedVersion, then version is bumped. This is the ONLY path that moves a stored judgment's score/state, and it is deliberately off the broad ports.JudgmentStore interface (a read-only consumer cannot reach it). On a miss it distinguishes ErrConflict (exists, version moved) from ErrNotFound.

type LeaseRunLock 

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

LeaseRunLock implements ports.RunLocker via a jobs_run_lock ROW lease instead of a session advisory lock. Unlike RunLock it does NOT hold a pooled connection for the duration of the run – each operation borrows a connection transiently – so N concurrent runs cannot starve the pool (the ≤8-default-pool hazard the hostile review flagged). A background renewer extends the lease while the run is live; a crash lets the lease expire so another worker can reclaim it. Used for RECON; the agent SESSION lock stays the advisory RunLock (it must not expire mid-LLM-loop). owner is a per-process id so renew/release are owner-scoped.

func NewLeaseRunLock 

func NewLeaseRunLock(pool *pgxpool.Pool, owner string, lease time.Duration) *LeaseRunLock

NewLeaseRunLock returns a row-lease run locker. lease is the claim TTL (set it comfortably above the longest run, e.g. ReconTimeout + a minute); the renewer ticks at lease/4 so several renews fall inside one TTL.

func (*LeaseRunLock) TryLock 

func (l *LeaseRunLock) TryLock(ctx context.Context, runID string) (func(), bool, error)

TryLock claims the lease (see TryLockLeased) and discards the lease-loss context – for callers that don't observe lease loss (e.g. the stale-run sweeper's liveness probe).

func (*LeaseRunLock) TryLockLeased 

func (l *LeaseRunLock) TryLockLeased(ctx context.Context, runID string) (context.Context, func(), bool, error)

TryLockLeased claims the lease for runID if it is free or expired. On success it starts a renewer and returns: a leaseCtx cancelled when the lease is LOST (so the caller aborts the in-flight run), plus a release that stops the renewer, cancels leaseCtx, and deletes the owner's row. A claim held by a live owner returns ok=false (the at-least-once queue retries).

type PoolConfig 

type PoolConfig struct {
	MaxConns          int32
	MinConns          int32
	MaxConnLifetime   time.Duration
	MaxConnIdleTime   time.Duration
	HealthCheckPeriod time.Duration
}

PoolConfig sizes the pgx connection pool. Zero values get sane defaults. Sizing the pool explicitly (the default pgx cap is max(4, NumCPU) ≈ 8) is required now that the durable agent path holds a connection-bearing advisory lock per active run – an unsized pool would starve HTTP handlers at low-tens concurrency.

type ProjectAnalysisStore 

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

ProjectAnalysisStore persists immutable Project analysis snapshots.

func NewProjectAnalysisStore 

func NewProjectAnalysisStore(pool *pgxpool.Pool) *ProjectAnalysisStore

func (*ProjectAnalysisStore) CurrentAnalysisHotspotSummary 

func (r *ProjectAnalysisStore) CurrentAnalysisHotspotSummary(ctx context.Context, tenantID, projectID, analysisID shared.ID, lens hotspot.Lens) (hotspot.Summary, error)

func (*ProjectAnalysisStore) Get 

func (r *ProjectAnalysisStore) Get(ctx context.Context, tenantID, projectID, analysisID shared.ID) (projectanalysis.Analysis, error)

func (*ProjectAnalysisStore) GetHotspot 

func (r *ProjectAnalysisStore) GetHotspot(ctx context.Context, tenantID, projectID, hotspotID shared.ID) (hotspot.Hotspot, error)

func (*ProjectAnalysisStore) GetIssue 

func (r *ProjectAnalysisStore) GetIssue(ctx context.Context, tenantID, projectID, issueID shared.ID) (issue.Issue, error)

func (*ProjectAnalysisStore) HotspotHistory 

func (r *ProjectAnalysisStore) HotspotHistory(ctx context.Context, tenantID, projectID, hotspotID shared.ID) ([]hotspot.ReviewEvent, error)

func (*ProjectAnalysisStore) IssueHistory 

func (r *ProjectAnalysisStore) IssueHistory(ctx context.Context, tenantID, projectID, issueID shared.ID) ([]issue.ReviewEvent, error)

func (*ProjectAnalysisStore) LatestForProjects 

func (r *ProjectAnalysisStore) LatestForProjects(ctx context.Context, tenantID shared.ID, projectIDs []shared.ID) (map[shared.ID]projectanalysis.Analysis, error)

func (*ProjectAnalysisStore) LatestWithResult 

func (r *ProjectAnalysisStore) LatestWithResult(ctx context.Context, tenantID, projectID shared.ID) (projectanalysis.Analysis, []byte, error)

func (*ProjectAnalysisStore) List 

func (r *ProjectAnalysisStore) List(ctx context.Context, tenantID, projectID shared.ID, limit int, beforeCreatedAt time.Time, beforeID shared.ID) ([]projectanalysis.Analysis, bool, error)

func (*ProjectAnalysisStore) ListAnalysisHotspots 

func (r *ProjectAnalysisStore) ListAnalysisHotspots(ctx context.Context, tenantID, projectID, analysisID shared.ID, lens hotspot.Lens, filter hotspot.ListFilter) (hotspot.Page, hotspot.Summary, error)

func (*ProjectAnalysisStore) ListHotspots 

func (r *ProjectAnalysisStore) ListHotspots(ctx context.Context, tenantID, projectID shared.ID, filter hotspot.ListFilter) (hotspot.Page, error)

func (*ProjectAnalysisStore) ListIssues 

func (r *ProjectAnalysisStore) ListIssues(ctx context.Context, tenantID, projectID shared.ID, filter issue.ListFilter) (issue.Page, error)

func (*ProjectAnalysisStore) ResolvedIssueKeys 

func (r *ProjectAnalysisStore) ResolvedIssueKeys(ctx context.Context, tenantID, projectID shared.ID) (map[string]bool, error)

func (*ProjectAnalysisStore) Save 

func (r *ProjectAnalysisStore) Save(ctx context.Context, analysis projectanalysis.Analysis) error

func (*ProjectAnalysisStore) SaveWithResult 

func (r *ProjectAnalysisStore) SaveWithResult(ctx context.Context, analysis projectanalysis.Analysis, result []byte) error

func (*ProjectAnalysisStore) SaveWithResultAndHotspots 

func (r *ProjectAnalysisStore) SaveWithResultAndHotspots(ctx context.Context, analysis projectanalysis.Analysis, result []byte, candidates []hotspot.Candidate) error

SaveWithResultAndHotspots commits the immutable analysis and its Security Hotspot projection in one PostgreSQL transaction. It delegates to SaveWithResultAndProjections with no issue projection, so both write paths share the same single-transaction body: a projection write failure rolls the analysis back, and the scan worker cannot publish a successful analysis without its projections.

func (*ProjectAnalysisStore) SaveWithResultAndProjections 

func (r *ProjectAnalysisStore) SaveWithResultAndProjections(ctx context.Context, analysis projectanalysis.Analysis, result []byte, hotspots []hotspot.Candidate, issues []issue.Candidate) error

SaveWithResultAndProjections commits the immutable analysis, its Security Hotspot projection, and its code-quality issue projection in a single PostgreSQL transaction. A projection write failure rolls the analysis back, so the scan worker cannot publish a successful analysis without both projections.

func (*ProjectAnalysisStore) TransitionHotspot 

func (r *ProjectAnalysisStore) TransitionHotspot(ctx context.Context, cmd hotspot.TransitionCommand) (hotspot.Hotspot, hotspot.ReviewEvent, error)

func (*ProjectAnalysisStore) TransitionIssue 

func (r *ProjectAnalysisStore) TransitionIssue(ctx context.Context, cmd issue.TransitionCommand) (issue.Issue, issue.ReviewEvent, error)

type ProjectRepository 

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

func NewProjectRepository 

func NewProjectRepository(pool *pgxpool.Pool) *ProjectRepository

func (*ProjectRepository) AssignProfile 

func (r *ProjectRepository) AssignProfile(ctx context.Context, tenantID shared.ID, projectKey, language, profileKey string) error

AssignProfile sets or clears the quality profile for a language in the project's JSONB default_profile_by_lang map, atomically at the column level (no read-modify-write race).

func (*ProjectRepository) CountByGate 

func (r *ProjectRepository) CountByGate(ctx context.Context, tenantID shared.ID, gateID string) (int, error)

func (*ProjectRepository) Create 

func (r *ProjectRepository) Create(ctx context.Context, p *project.Project) error

func (*ProjectRepository) DeleteByKey 

func (r *ProjectRepository) DeleteByKey(ctx context.Context, tenantID shared.ID, key string) error

func (*ProjectRepository) GetByID 

func (r *ProjectRepository) GetByID(ctx context.Context, tenantID, projectID shared.ID) (*project.Project, error)

func (*ProjectRepository) GetByKey 

func (r *ProjectRepository) GetByKey(ctx context.Context, tenantID shared.ID, key string) (*project.Project, error)

func (*ProjectRepository) List 

func (r *ProjectRepository) List(ctx context.Context, tenantID shared.ID) ([]*project.Project, error)

func (*ProjectRepository) UpdateGate 

func (r *ProjectRepository) UpdateGate(ctx context.Context, tenantID shared.ID, key, gateID string) error

type QualityGateMutator 

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

QualityGateMutator commits managed-gate writes and their audit records together.

func NewQualityGateMutator 

func NewQualityGateMutator(pool *pgxpool.Pool) *QualityGateMutator

func (*QualityGateMutator) AssignProjectGate 

func (m *QualityGateMutator) AssignProjectGate(ctx context.Context, tenantID shared.ID, projectKey, gateID string, audit ports.AuditEntry) error

func (*QualityGateMutator) CreateGate 

func (m *QualityGateMutator) CreateGate(ctx context.Context, tenantID shared.ID, gate qualitygate.Gate, audit ports.AuditEntry) error

func (*QualityGateMutator) CreateProjectWithGate 

func (m *QualityGateMutator) CreateProjectWithGate(ctx context.Context, p *project.Project) error

func (*QualityGateMutator) DeleteGate 

func (m *QualityGateMutator) DeleteGate(ctx context.Context, tenantID shared.ID, key string, audit ports.AuditEntry) error

func (*QualityGateMutator) UpdateGate 

func (m *QualityGateMutator) UpdateGate(ctx context.Context, tenantID shared.ID, gate qualitygate.Gate, audit ports.AuditEntry) error

type QualityGateStore 

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

QualityGateStore persists tenant-scoped custom quality gates.

func NewQualityGateStore 

func NewQualityGateStore(pool *pgxpool.Pool) *QualityGateStore

func (*QualityGateStore) Create 

func (s *QualityGateStore) Create(ctx context.Context, tenantID shared.ID, gate qualitygate.Gate) error

func (*QualityGateStore) Delete 

func (s *QualityGateStore) Delete(ctx context.Context, tenantID shared.ID, key string) error

func (*QualityGateStore) DeleteIfUnassigned 

func (s *QualityGateStore) DeleteIfUnassigned(ctx context.Context, tenantID shared.ID, key string) error

func (*QualityGateStore) Get 

func (s *QualityGateStore) Get(ctx context.Context, tenantID shared.ID, key string) (qualitygate.Gate, error)

func (*QualityGateStore) List 

func (s *QualityGateStore) List(ctx context.Context, tenantID shared.ID) ([]qualitygate.Gate, error)

func (*QualityGateStore) Update 

func (s *QualityGateStore) Update(ctx context.Context, tenantID shared.ID, gate qualitygate.Gate) error

type QualityProfileStore 

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

QualityProfileStore persists tenant-scoped custom quality profiles. Built-in profiles are generated from the rule catalog and never stored here.

func NewQualityProfileStore 

func NewQualityProfileStore(pool *pgxpool.Pool) *QualityProfileStore

func (*QualityProfileStore) Create 

func (s *QualityProfileStore) Create(ctx context.Context, tenantID shared.ID, profile qualityprofile.Profile) error

func (*QualityProfileStore) Delete 

func (s *QualityProfileStore) Delete(ctx context.Context, tenantID shared.ID, key string) error

func (*QualityProfileStore) Get 

func (s *QualityProfileStore) Get(ctx context.Context, tenantID shared.ID, key string) (qualityprofile.Profile, error)

func (*QualityProfileStore) List 

func (s *QualityProfileStore) List(ctx context.Context, tenantID shared.ID) ([]qualityprofile.Profile, error)

func (*QualityProfileStore) Update 

func (s *QualityProfileStore) Update(ctx context.Context, tenantID shared.ID, profile qualityprofile.Profile) error

type ReconRunStore 

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

ReconRunStore persists recon-run records.

func NewReconRunStore 

func NewReconRunStore(pool *pgxpool.Pool) *ReconRunStore

NewReconRunStore returns a store backed by the given pool.

func (*ReconRunStore) Get 

func (r *ReconRunStore) Get(ctx context.Context, id shared.ID) (recon.Run, error)

Get returns a run by id, or shared.ErrNotFound.

func (*ReconRunStore) ListByEngagement 

func (r *ReconRunStore) ListByEngagement(ctx context.Context, engagementID shared.ID) ([]recon.Run, error)

ListByEngagement returns an engagement's runs, newest first.

func (*ReconRunStore) ListStaleRunning 

func (r *ReconRunStore) ListStaleRunning(ctx context.Context, olderThan time.Time, limit int) ([]recon.Run, error)

ListStaleRunning returns runs still 'running' that started before olderThan (≤ limit), oldest first – the stale-run sweeper's input.

func (*ReconRunStore) Save 

func (r *ReconRunStore) Save(ctx context.Context, run recon.Run) error

Save upserts a run (used on create and on every stage/status update).

type RetestRepository 

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

RetestRepository persists the per-finding retest history to PostgreSQL.

func NewRetestRepository 

func NewRetestRepository(pool *pgxpool.Pool) *RetestRepository

NewRetestRepository returns a repository backed by the given pool.

func (*RetestRepository) Add 

func (r *RetestRepository) Add(ctx context.Context, rt finding.Retest) error

Add inserts a retest (append-only; retests are not edited or deleted in app code).

func (*RetestRepository) ListByEngagementFinding 

func (r *RetestRepository) ListByEngagementFinding(ctx context.Context, engagementID, findingID shared.ID) ([]finding.Retest, error)

ListByEngagementFinding returns a finding's retests oldest-first, scoped to the engagement (no cross-engagement read).

type RunLock 

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

RunLock implements ports.RunLocker with a PostgreSQL session advisory lock keyed by the run id (F9). The lock is held on a dedicated pooled connection for the duration of the execution and released (with the connection) afterwards – so across the API + the worker, at most one delivery of a given run executes at a time. A redelivery that finds the lock held gets ok=false and skips, preventing a duplicate live scan.

func NewRunLock 

func NewRunLock(pool *pgxpool.Pool) *RunLock

NewRunLock returns a Postgres-backed run locker.

func (*RunLock) TryLock 

func (l *RunLock) TryLock(ctx context.Context, runID string) (func(), bool, error)

type ScanJobStore 

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

ScanJobStore persists asynchronous scan-job status.

func NewScanJobStore 

func NewScanJobStore(pool *pgxpool.Pool) *ScanJobStore

NewScanJobStore returns a store backed by the given pool.

func (*ScanJobStore) CreateRunning 

func (r *ScanJobStore) CreateRunning(ctx context.Context, j ports.ScanJob) error

func (*ScanJobStore) GetJob 

func (r *ScanJobStore) GetJob(ctx context.Context, id string) (ports.ScanJob, error)

GetJob returns a scan job by its own id, or ErrNotFound.

func (*ScanJobStore) LatestForEngagement 

func (r *ScanJobStore) LatestForEngagement(ctx context.Context, engagementID shared.ID) (ports.ScanJob, error)

func (*ScanJobStore) LatestForEngagements 

func (r *ScanJobStore) LatestForEngagements(ctx context.Context, engagementIDs []shared.ID) (map[shared.ID]ports.ScanJob, error)

LatestForEngagement returns the engagement's most recent scan job, or ErrNotFound.

func (*ScanJobStore) ListStaleRunning 

func (r *ScanJobStore) ListStaleRunning(ctx context.Context, olderThan time.Time, limit int) ([]ports.ScanJob, error)

ListStaleRunning returns scan jobs still 'running' that started before olderThan (≤ limit), oldest first – the stale-scan sweeper's input.

func (*ScanJobStore) Save 

func (r *ScanJobStore) Save(ctx context.Context, j ports.ScanJob) error

Save upserts a scan job (used on create and on every stage/status update).

type ScanRepository 

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

ScanRepository persists SCA scans (SBOM + components + vulnerabilities).

func NewScanRepository 

func NewScanRepository(pool *pgxpool.Pool) *ScanRepository

NewScanRepository returns a repository backed by the given pool.

func (*ScanRepository) SaveScan 

func (r *ScanRepository) SaveScan(ctx context.Context, engagementID shared.ID, doc *sbom.SBOM, vulns []vulnerability.Vulnerability, snap ports.ScanSnapshot) (int, error)

SaveScan stores the SBOM, its components, and the vulnerabilities found against them in one transaction – a new immutable snapshot per scan. It returns the number of vulns that could not be linked to a component in this SBOM (skipped, never orphaned); the caller surfaces a non-zero count on the audit log so a dropped advisory is never invisible on a chain-of-custody tool.

type ScanResultStore 

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

ScanResultStore caches the latest full scan result (JSON) per engagement.

func NewScanResultStore 

func NewScanResultStore(pool *pgxpool.Pool) *ScanResultStore

NewScanResultStore returns a store backed by the given pool.

func (*ScanResultStore) LatestResult 

func (r *ScanResultStore) LatestResult(ctx context.Context, engagementID shared.ID) ([]byte, error)

LatestResult returns the engagement's cached scan result, or shared.ErrNotFound.

func (*ScanResultStore) SaveResult 

func (r *ScanResultStore) SaveResult(ctx context.Context, engagementID shared.ID, result []byte) error

SaveResult upserts the engagement's latest scan result JSON.

type ScanRunStore 

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

ScanRunStore persists scan-run manifests + finding keys.

func NewScanRunStore 

func NewScanRunStore(pool *pgxpool.Pool) *ScanRunStore

NewScanRunStore returns a store backed by the given pool.

func (*ScanRunStore) Get 

func (r *ScanRunStore) Get(ctx context.Context, runID string) (ports.ScanRun, error)

func (*ScanRunStore) List 

func (r *ScanRunStore) List(ctx context.Context, engagementID shared.ID) ([]ports.ScanRun, error)

func (*ScanRunStore) Save 

func (r *ScanRunStore) Save(ctx context.Context, run ports.ScanRun) error

type ThreatModelRepository 

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

ThreatModelRepository persists the architecture-input threat model per engagement to PostgreSQL: one row per engagement, the validated domain model stored as a JSONB blob. Tenant-scoped customer data (tenant_id recorded for the row-scoping sweep); reads are engagement-scoped (the tenant gate runs upstream at the child route).

func NewThreatModelRepository 

func NewThreatModelRepository(pool *pgxpool.Pool) *ThreatModelRepository

NewThreatModelRepository returns a repository backed by the given pool.

func (*ThreatModelRepository) Get 

func (r *ThreatModelRepository) Get(ctx context.Context, engagementID shared.ID) (threatmodel.Model, bool, error)

Get decodes the engagement's model from its JSONB blob; ok=false when none has been ingested.

func (*ThreatModelRepository) Save 

func (r *ThreatModelRepository) Save(ctx context.Context, engagementID, tenantID shared.ID, m threatmodel.Model) error

Save upserts the engagement's model (the usecase has already bounded size + validated it), bumping version on each re-ingest. The model round-trips through the JSONB `data` blob.

type TimestampStore 

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

TimestampStore persists external RFC-3161 tokens for chain heads on PostgreSQL, out-of-band from the report. One token per (chain, engagement, head).

func NewTimestampStore 

func NewTimestampStore(pool *pgxpool.Pool) *TimestampStore

NewTimestampStore returns a timestamp store backed by the given pool.

func (*TimestampStore) Get 

func (s *TimestampStore) Get(ctx context.Context, chain string, eng shared.ID, head string) (*ports.TimestampToken, error)

Get returns the stored token for a head, or nil if it is not yet anchored.

func (*TimestampStore) LatestHead 

func (s *TimestampStore) LatestHead(ctx context.Context, chain string, eng shared.ID) (string, bool, error)

LatestHead returns the most-recently-anchored head for a chain (ok=false if none) – the retained head for out-of-band tail-truncation detection.

func (*TimestampStore) Put 

func (s *TimestampStore) Put(ctx context.Context, chain string, eng shared.ID, head string, token ports.TimestampToken) error

Put stores a token for a head, idempotent per (chain, engagement, head).

type UserRepository 

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

UserRepository persists operator identities to PostgreSQL.

func NewUserRepository 

func NewUserRepository(pool *pgxpool.Pool) *UserRepository

NewUserRepository returns a repository backed by the given pool.

func (*UserRepository) Create 

func (r *UserRepository) Create(ctx context.Context, u *user.User) error

func (*UserRepository) GetByAPIKeyHash 

func (r *UserRepository) GetByAPIKeyHash(ctx context.Context, hash string) (*user.User, error)

func (*UserRepository) GetByID 

func (r *UserRepository) GetByID(ctx context.Context, id shared.ID) (*user.User, error)

func (*UserRepository) List 

func (r *UserRepository) List(ctx context.Context) ([]*user.User, error)

func (*UserRepository) Upsert 

func (r *UserRepository) Upsert(ctx context.Context, u *user.User) error

type WriteupDraftRepository 

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

WriteupDraftRepository persists AI-proposed, human-gated finding write-up drafts to PostgreSQL, engagement-scoped.

func NewWriteupDraftRepository 

func NewWriteupDraftRepository(pool *pgxpool.Pool) *WriteupDraftRepository

NewWriteupDraftRepository returns a repository backed by the given pool.

func (*WriteupDraftRepository) Get 

func (r *WriteupDraftRepository) Get(ctx context.Context, engagementID, id shared.ID) (writeupdraft.Draft, error)

Get returns the engagement's draft by id, or shared.ErrNotFound.

func (*WriteupDraftRepository) ListByEngagement 

func (r *WriteupDraftRepository) ListByEngagement(ctx context.Context, engagementID shared.ID) ([]writeupdraft.Draft, error)

ListByEngagement returns the engagement's drafts, oldest first (deterministic order).

func (*WriteupDraftRepository) Save 

func (r *WriteupDraftRepository) Save(ctx context.Context, d writeupdraft.Draft) error

Save upserts a draft by id. Unlike a Judgment (insert-only), a Draft is mutable working data, so on conflict the mutable fields (text, state, decided_by, updated_at) are replaced; the immutable fields (engagement_id, finding_id, proposed_by, created_at) are never moved. tenant_id is written as the empty-string default tenant (mirrors judgments/findings); reads are tenant-isolated via the engagement gate, and the column is present so the P5/E22 row-scoping sweep covers it.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.