Help us shape the future of Go! Take the 2026 Go Developer Survey

postgres

package
v1.10.0 Latest Latest
Warning

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

 Go to latest Published: Nov 3, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Overview

Package postgres implements the Chronicle store interface using grove ORM.

Index

Constants

This section is empty.

Variables

View Source 
var Migrations = migrate.NewGroup("chronicle")

Migrations is the grove migration group for the Chronicle store.

Functions

This section is empty.

Types

type ArchiveModel 

type ArchiveModel struct {
	grove.BaseModel `grove:"table:chronicle_archives,alias:a"`

	ID            string    `grove:"id,pk"`
	PolicyID      string    `grove:"policy_id"`
	Category      string    `grove:"category"`
	EventCount    int64     `grove:"event_count"`
	FromTimestamp time.Time `grove:"from_timestamp"`
	ToTimestamp   time.Time `grove:"to_timestamp"`
	SinkName      string    `grove:"sink_name"`
	SinkRef       string    `grove:"sink_ref"`
	AppID         string    `grove:"app_id"`
	TenantID      string    `grove:"tenant_id"`
	CreatedAt     time.Time `grove:"created_at"`
}

ArchiveModel is the grove ORM model for the chronicle_archives table.

type CheckpointModel 

type CheckpointModel struct {
	grove.BaseModel `grove:"table:chronicle_checkpoints,alias:cp"`

	ID             string    `grove:"id,pk"`
	StreamID       string    `grove:"stream_id"`
	AppID          string    `grove:"app_id"`
	TenantID       string    `grove:"tenant_id"`
	FromSeq        int64     `grove:"from_seq"`
	ToSeq          int64     `grove:"to_seq"`
	FromHash       string    `grove:"from_hash"`
	ToHash         string    `grove:"to_hash"`
	EventCount     int64     `grove:"event_count"`
	PrevCheckpoint string    `grove:"prev_checkpoint"`
	Algorithm      string    `grove:"algorithm"`
	SignKeyID      string    `grove:"sign_key_id"`
	Signature      []byte    `grove:"signature"`
	SignedPayload  string    `grove:"signed_payload"`
	CreatedAt      time.Time `grove:"created_at"`
}

CheckpointModel is the grove ORM model for the chronicle_checkpoints table.

type ErasureModel 

type ErasureModel struct {
	grove.BaseModel `grove:"table:chronicle_erasures,alias:er"`

	ID             string    `grove:"id,pk"`
	SubjectID      string    `grove:"subject_id"`
	Reason         string    `grove:"reason"`
	RequestedBy    string    `grove:"requested_by"`
	EventsAffected int64     `grove:"events_affected"`
	KeyDestroyed   bool      `grove:"key_destroyed"`
	AppID          string    `grove:"app_id"`
	TenantID       string    `grove:"tenant_id"`
	CreatedAt      time.Time `grove:"created_at"`
	Status         string    `grove:"status"`

	LegacyKeyRetained bool `grove:"legacy_key_retained"`
}

ErasureModel is the grove ORM model for the chronicle_erasures table.

type EventModel 

type EventModel struct {
	grove.BaseModel `grove:"table:chronicle_events,alias:e"`

	ID               string        `grove:"id,pk"`
	StreamID         string        `grove:"stream_id"`
	Sequence         int64         `grove:"sequence"`
	Hash             string        `grove:"hash"`
	PrevHash         string        `grove:"prev_hash"`
	AppID            string        `grove:"app_id"`
	TenantID         string        `grove:"tenant_id"`
	UserID           string        `grove:"user_id"`
	IP               string        `grove:"ip"`
	UserAgent        string        `grove:"user_agent"`
	RequestID        string        `grove:"request_id"`
	SessionID        string        `grove:"session_id"`
	Action           string        `grove:"action"`
	Resource         string        `grove:"resource"`
	Category         string        `grove:"category"`
	ResourceID       string        `grove:"resource_id"`
	Metadata         exactMetadata `grove:"metadata,type:jsonb"` // decoded without numeric rounding
	LosslessMetadata bool          `grove:"lossless_metadata"`
	Outcome          string        `grove:"outcome"`
	Severity         string        `grove:"severity"`
	Reason           string        `grove:"reason"`
	SubjectID        string        `grove:"subject_id"`
	EncryptionKeyID  string        `grove:"encryption_key_id"`
	Erased           bool          `grove:"erased"`
	ErasedAt         *time.Time    `grove:"erased_at"`
	ErasureID        string        `grove:"erasure_id"`
	Timestamp        time.Time     `grove:"timestamp"`
	TimestampSubUs   int32         `grove:"timestamp_sub_us"`
	CreatedAt        time.Time     `grove:"created_at"`
	HashScheme       string        `grove:"hash_scheme"`
	HashKeyID        string        `grove:"hash_key_id"`
}

EventModel is the grove ORM model for the chronicle_events table.

Timestamp is split across two columns. See splitTimestamp for why.

type Option 

type Option func(*Store)

Option configures a Store constructed by New.

func WithHasher 

func WithHasher(h *hash.Chain) Option

WithHasher sets the chain the store re-links with.

Append re-derives the sequence and prev_hash while holding the stream's row lock, which means it must also recompute the digest. That recomputation has to use the same scheme Chronicle was configured with; a store left on the default plain chain would quietly downgrade every event it re-linked.

type ReportModel 

type ReportModel struct {
	grove.BaseModel `grove:"table:chronicle_reports,alias:r"`

	ID          string    `grove:"id,pk"`
	Title       string    `grove:"title"`
	Type        string    `grove:"type"`
	PeriodFrom  time.Time `grove:"period_from"`
	PeriodTo    time.Time `grove:"period_to"`
	AppID       string    `grove:"app_id"`
	TenantID    string    `grove:"tenant_id"`
	Format      string    `grove:"format"`
	Data        []byte    `grove:"data,type:jsonb"` // compliance.EncodeReportBody: sections, stats and verification
	GeneratedBy string    `grove:"generated_by"`
	CreatedAt   time.Time `grove:"created_at"`
}

ReportModel is the grove ORM model for the chronicle_reports table.

type RetentionPolicyModel 

type RetentionPolicyModel struct {
	grove.BaseModel `grove:"table:chronicle_retention_policies,alias:rp"`

	ID        string    `grove:"id,pk"`
	Category  string    `grove:"category"`
	Duration  int64     `grove:"duration"` // nanoseconds
	Archive   bool      `grove:"archive"`
	AppID     string    `grove:"app_id"`
	TenantID  string    `grove:"tenant_id"`
	CreatedAt time.Time `grove:"created_at"`
	UpdatedAt time.Time `grove:"updated_at"`
}

RetentionPolicyModel is the grove ORM model for the chronicle_retention_policies table.

type Store 

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

Store implements the Chronicle store interface using grove ORM.

func New 

func New(db *grove.DB, opts ...Option) *Store

New creates a new grove ORM store with the given database connection.

Without WithHasher, Append re-links under a zero-value hash.Chain, which is SchemePlainV4 with no key provider: unkeyed, but with the unambiguous content encoding, so a default-configured store is not writing forgeable digests.

func (*Store) Accept 

func (s *Store) Accept(ctx context.Context, request acceptance.Request, h *hash.Chain, prepare func(*audit.Event) error) (*acceptance.Receipt, error)

Accept commits stream creation, pin, event, head and receipt in one transaction.

func (*Store) Aggregate 

func (s *Store) Aggregate(ctx context.Context, q *audit.AggregateQuery) (*audit.AggregateResult, error)

Aggregate returns grouped event statistics.

func (*Store) Append 

func (s *Store) Append(ctx context.Context, event *audit.Event) error

Append persists a single audit event, allocating its sequence number authoritatively inside a transaction.

The sequence is a positional counter and is NOT part of the hash chain (see hash.Chain.Compute), so the store owns its allocation. Chronicle.Record precomputes event.Sequence from the stream's tracked head, but that head can fall behind the events table — e.g. a crash between Append and UpdateStreamHead, or a bulk import — after which every subsequent append recomputes the same value and collides on UNIQUE(stream_id, sequence), permanently wedging the stream. To be both self-healing and safe under concurrent appends, derive the next sequence from the greater of the stored head and the actual MAX(sequence) while holding a row lock on the stream, and advance the head in the same transaction so it can never lag again.

func (*Store) AppendBatch 

func (s *Store) AppendBatch(ctx context.Context, events []*audit.Event) error

AppendBatch persists multiple events atomically in a transaction.

func (*Store) AppendCheckpoint 

func (s *Store) AppendCheckpoint(ctx context.Context, cp *checkpoint.Checkpoint) error

AppendCheckpoint persists a checkpoint. Checkpoints are append-only: there is no update or delete, since a checkpoint that could be revised would assert nothing.

UNIQUE(stream_id, to_seq) and UNIQUE(stream_id, from_seq) on chronicle_checkpoints are the structural backstop against two checkpointers racing on one stream -- a background ticker, an operator-triggered run, or a second replica's ticker -- so the loser of that race gets checkpoint.ErrExists straight from the database rather than from a read-then-insert check a concurrent writer could still slip past.

Both constraints are load-bearing. See migration 007 for why to_seq alone lets two replicas reading different heads write overlapping checkpoints.

func (*Store) AppendWithChain 

func (s *Store) AppendWithChain(ctx context.Context, event *audit.Event, h *chash.Chain) error

AppendWithChain uses the caller's configured hasher under the stream lock.

func (*Store) ByUser 

func (s *Store) ByUser(ctx context.Context, userID string, opts audit.TimeRange) (*audit.QueryResult, error)

ByUser returns events for a specific user within a time range.

func (*Store) CheckpointsInRange 

func (s *Store) CheckpointsInRange(
	ctx context.Context, streamID id.ID, fromSeq, toSeq uint64,
) ([]*checkpoint.Checkpoint, error)

CheckpointsInRange returns every checkpoint whose range overlaps [fromSeq, toSeq], ascending by ToSeq.

Overlap rather than containment: a caller verifying sequences 150 to 160 still needs the checkpoint covering 101 to 200, because that is the one whose signed assertion those events fall under. Containment would return nothing for a range that sits inside a wider checkpoint.

func (*Store) Close 

func (s *Store) Close() error

Close closes the database connection.

func (*Store) CompleteErasure 

func (s *Store) CompleteErasure(ctx context.Context, erasureID id.ID, o erasure.Outcome) error

CompleteErasure writes an erasure's outcome and marks it completed.

func (*Store) Count 

func (s *Store) Count(ctx context.Context, q *audit.CountQuery) (int64, error)

Count returns the total number of events matching filters.

func (*Store) CountBySubject 

func (s *Store) CountBySubject(ctx context.Context, sq erasure.SubjectQuery) (int64, error)

CountBySubject returns the number of events for a subject within the query's scope.

Security-critical: without the scope filter this reveals how many events other tenants hold on the subject.

func (*Store) CountErasures 

func (s *Store) CountErasures(ctx context.Context, sc erasure.Scope) (int64, error)

CountErasures returns the number of erasure records in the given scope.

func (*Store) CreateStream 

func (s *Store) CreateStream(ctx context.Context, st *stream.Stream) error

CreateStream initializes a new hash chain stream.

func (*Store) DeletePolicy 

func (s *Store) DeletePolicy(ctx context.Context, policyID id.ID) error

DeletePolicy removes a retention policy.

func (*Store) DeleteReport 

func (s *Store) DeleteReport(ctx context.Context, reportID id.ID) error

DeleteReport removes a report by ID.

func (*Store) EventRange 

func (s *Store) EventRange(ctx context.Context, streamID id.ID, fromSeq, toSeq uint64) ([]*audit.Event, error)

EventRange returns events in a sequence range for chain verification.

func (*Store) EventsOlderThan 

func (s *Store) EventsOlderThan(
	ctx context.Context, pq retention.PurgeQuery,
) ([]*audit.Event, error)

EventsOlderThan returns the events the purge query selects.

Security-critical: the scope filter is what keeps one tenant's policy from selecting, and therefore purging, every tenant's history. Both columns are always compared, so an empty TenantID selects untenanted events rather than every tenant. The bound keeps a large backlog from being loaded into memory all at once.

func (*Store) Gaps 

func (s *Store) Gaps(ctx context.Context, streamID id.ID, fromSeq, toSeq uint64) ([]uint64, error)

Gaps detects missing sequence numbers in a range.

func (*Store) Get 

func (s *Store) Get(ctx context.Context, eventID id.ID) (*audit.Event, error)

Get returns a single event by ID.

func (*Store) GetCheckpoint 

func (s *Store) GetCheckpoint(ctx context.Context, checkpointID id.ID) (*checkpoint.Checkpoint, error)

GetCheckpoint returns one checkpoint by ID.

func (*Store) GetErasure 

func (s *Store) GetErasure(ctx context.Context, erasureID id.ID) (*erasure.Erasure, error)

GetErasure returns an erasure record by ID.

func (*Store) GetPolicy 

func (s *Store) GetPolicy(ctx context.Context, policyID id.ID) (*retention.Policy, error)

GetPolicy returns a retention policy by ID.

func (*Store) GetReport 

func (s *Store) GetReport(ctx context.Context, reportID id.ID) (*compliance.Report, error)

GetReport returns a report by ID.

func (*Store) GetStream 

func (s *Store) GetStream(ctx context.Context, streamID id.ID) (*stream.Stream, error)

GetStream returns a stream by ID.

func (*Store) GetStreamByScope 

func (s *Store) GetStreamByScope(ctx context.Context, appID, tenantID string) (*stream.Stream, error)

GetStreamByScope returns the stream for a given app+tenant scope.

func (*Store) LastHash 

func (s *Store) LastHash(ctx context.Context, streamID id.ID) (string, error)

LastHash returns the hash of the most recent event in a stream.

func (*Store) LastSequence 

func (s *Store) LastSequence(ctx context.Context, streamID id.ID) (uint64, error)

LastSequence returns the highest sequence number for a stream.

func (*Store) LatestCheckpoint 

func (s *Store) LatestCheckpoint(ctx context.Context, streamID id.ID) (*checkpoint.Checkpoint, error)

LatestCheckpoint returns the checkpoint with the highest ToSeq for a stream.

func (*Store) ListArchives 

func (s *Store) ListArchives(ctx context.Context, opts retention.ListOpts) ([]*retention.Archive, error)

ListArchives returns archive records matching opts, scoped before pagination.

func (*Store) ListCheckpoints 

func (s *Store) ListCheckpoints(
	ctx context.Context, streamID id.ID, opts checkpoint.ListOpts,
) ([]*checkpoint.Checkpoint, error)

ListCheckpoints returns a stream's checkpoints, newest first.

func (*Store) ListErasures 

func (s *Store) ListErasures(ctx context.Context, opts erasure.ListOpts) ([]*erasure.Erasure, error)

ListErasures returns erasure records matching opts, scoped before pagination.

func (*Store) ListPolicies 

func (s *Store) ListPolicies(
	ctx context.Context, opts retention.ListPoliciesOpts,
) ([]*retention.Policy, error)

ListPolicies returns retention policies matching opts, scoped before pagination so a caller's own rows are never hidden behind another tenant's.

func (*Store) ListReports 

func (s *Store) ListReports(ctx context.Context, opts compliance.ListOpts) ([]*compliance.Report, error)

ListReports returns reports matching opts, scoped before pagination.

func (*Store) ListStreams 

func (s *Store) ListStreams(ctx context.Context, opts stream.ListOpts) ([]*stream.Stream, error)

ListStreams returns all streams with pagination.

func (*Store) MarkErased 

func (s *Store) MarkErased(
	ctx context.Context, sq erasure.SubjectQuery, erasureID id.ID,
) (int64, error)

MarkErased flags a subject's events as erased within the query's scope.

Security-critical: without the scope filter any caller could flag every tenant's events for a guessed subject ID.

func (*Store) Migrate 

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

Migrate applies the Chronicle migrations this database hasn't had yet.

grove records each one in grove_migrations as it lands, so on a current schema this runs no DDL at all. It used to run every ALTER TABLE on each call, and ALTER TABLE waits for an exclusive lock on the table even when the column is already there, so a process starting up could deadlock with one that was appending. Concurrent callers take turns on grove's advisory lock, and the ones that go second find nothing left to apply.

A database migrated before this kept no record, so its first call here runs every migration once more. They are all written to be re-run.

func (*Store) Ping 

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

Ping checks database connectivity.

func (*Store) PurgeEvents 

func (s *Store) PurgeEvents(ctx context.Context, eventIDs []id.ID) (int64, error)

PurgeEvents permanently deletes events by IDs.

func (*Store) Query 

func (s *Store) Query(ctx context.Context, q *audit.Query) (*audit.QueryResult, error)

Query returns events matching filters with pagination.

func (*Store) RecordArchive 

func (s *Store) RecordArchive(ctx context.Context, a *retention.Archive) error

RecordArchive records that a batch of events was archived.

func (*Store) RecordErasure 

func (s *Store) RecordErasure(ctx context.Context, e *erasure.Erasure) error

RecordErasure persists an erasure event.

func (*Store) SavePolicy 

func (s *Store) SavePolicy(ctx context.Context, p *retention.Policy) error

SavePolicy persists a retention policy (INSERT or UPDATE on conflict).

The conflict target is the owning scope plus the category, not the category alone: a policy must never overwrite, or take ownership of, another app or tenant's policy for the same category.

func (*Store) SaveReport 

func (s *Store) SaveReport(ctx context.Context, r *compliance.Report) error

SaveReport persists a generated compliance report.

func (*Store) SetHasher 

func (s *Store) SetHasher(h *hash.Chain)

SetHasher replaces the chain the store re-links with after construction.

This exists for a store built directly (e.g. passed to extension.WithStore) rather than through New with WithHasher: it lets a caller who already holds the Store give it the chain later, which is what lets that path be brought up to the configured digest scheme instead of silently staying on the default plain one.

func (*Store) SubjectKeyUsage 

func (s *Store) SubjectKeyUsage(ctx context.Context, subjectID string) ([]erasure.KeyUsage, error)

SubjectKeyUsage groups a subject's events across every scope by app, tenant, encryption key ID and erased flag.

Security-critical in the other direction: it is unscoped on purpose, so the erasure service can see every scope sharing a legacy key. It must never back a response to a caller; see erasure.Store.

func (*Store) UpdateStreamHead 

func (s *Store) UpdateStreamHead(ctx context.Context, streamID id.ID, hash string, seq uint64) error

UpdateStreamHead ignores late writers and rejects conflicting equal positions.

func (*Store) UpdateStreamScheme 

func (s *Store) UpdateStreamScheme(ctx context.Context, streamID id.ID, scheme string, since uint64) error

UpdateStreamScheme moves the stream's digest pin to scheme, applying from sequence since.

type StreamModel 

type StreamModel struct {
	grove.BaseModel `grove:"table:chronicle_streams,alias:s"`

	ID          string    `grove:"id,pk"`
	AppID       string    `grove:"app_id"`
	TenantID    string    `grove:"tenant_id"`
	HeadHash    string    `grove:"head_hash"`
	HeadSeq     int64     `grove:"head_seq"`
	CreatedAt   time.Time `grove:"created_at"`
	UpdatedAt   time.Time `grove:"updated_at"`
	Scheme      string    `grove:"scheme"`
	SchemeSince int64     `grove:"scheme_since"`
}

StreamModel is the grove ORM model for the chronicle_streams table.

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.