mongo

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: 21 Imported by: 0

Documentation

Overview

Package mongo implements the Chronicle store interface using grove ORM with the MongoDB driver.

Index

Constants

This section is empty.

Variables

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

Migrations is the grove migration group for the Chronicle mongo store.

Functions

This section is empty.

Types

type ArchiveModel 

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

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

ArchiveModel is the grove ORM model for the chronicle_archives collection.

type CheckpointModel 

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

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

CheckpointModel is the grove ORM model for the chronicle_checkpoints collection.

type ErasureModel 

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

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

	LegacyKeyRetained bool `grove:"legacy_key_retained" bson:"legacy_key_retained"`
}

ErasureModel is the grove ORM model for the chronicle_erasures collection.

type EventModel 

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

	ID              string         `grove:"id,pk"              bson:"_id"`
	StreamID        string         `grove:"stream_id"          bson:"stream_id"`
	Sequence        uint64         `grove:"sequence"           bson:"sequence"`
	Hash            string         `grove:"hash"               bson:"hash"`
	PrevHash        string         `grove:"prev_hash"          bson:"prev_hash"`
	AppID           string         `grove:"app_id"             bson:"app_id"`
	TenantID        string         `grove:"tenant_id"          bson:"tenant_id"`
	UserID          string         `grove:"user_id"            bson:"user_id"`
	IP              string         `grove:"ip"                 bson:"ip"`
	UserAgent       string         `grove:"user_agent"         bson:"user_agent"`
	RequestID       string         `grove:"request_id"         bson:"request_id"`
	SessionID       string         `grove:"session_id"         bson:"session_id"`
	Action          string         `grove:"action"             bson:"action"`
	Resource        string         `grove:"resource"           bson:"resource"`
	Category        string         `grove:"category"           bson:"category"`
	ResourceID      string         `grove:"resource_id"        bson:"resource_id"`
	Metadata        map[string]any `grove:"metadata"           bson:"metadata,omitempty"`
	Outcome         string         `grove:"outcome"            bson:"outcome"`
	Severity        string         `grove:"severity"           bson:"severity"`
	Reason          string         `grove:"reason"             bson:"reason"`
	SubjectID       string         `grove:"subject_id"         bson:"subject_id"`
	EncryptionKeyID string         `grove:"encryption_key_id"  bson:"encryption_key_id"`
	Erased          bool           `grove:"erased"             bson:"erased"`
	ErasedAt        *time.Time     `grove:"erased_at"          bson:"erased_at,omitempty"`
	ErasureID       string         `grove:"erasure_id"         bson:"erasure_id,omitempty"`
	Timestamp       time.Time      `grove:"timestamp"          bson:"timestamp"`
	TimestampSubMs  int32          `grove:"timestamp_sub_ms"   bson:"timestamp_sub_ms"`
	CreatedAt       time.Time      `grove:"created_at"         bson:"created_at"`
	HashScheme      string         `grove:"hash_scheme"        bson:"hash_scheme,omitempty"`
	HashKeyID       string         `grove:"hash_key_id"        bson:"hash_key_id,omitempty"`
}

EventModel is the grove ORM model for the chronicle_events collection.

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

type ReportModel 

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

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

ReportModel is the grove ORM model for the chronicle_reports collection.

type RetentionPolicyModel 

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

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

RetentionPolicyModel is the grove ORM model for the chronicle_retention_policies collection.

type Store 

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

Store implements the Chronicle store interface using grove ORM with MongoDB.

func New 

func New(db *grove.DB) *Store

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

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.

func (*Store) AppendBatch 

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

AppendBatch persists multiple events atomically.

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.

The unique indexes on (stream_id, to_seq) and (stream_id, from_seq) -- created alongside the collection in migrations.go, see checkpointIndexes for why both -- 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.

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. It deliberately does not use scopeFilter: both fields are always matched, 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. Since MongoDB does not have generate_series, we query existing sequences and compute the gaps in Go.

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 with 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 with 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 creates indexes for all chronicle collections.

It drops the category-only policy index earlier releases created before building anything, because creating the scoped index next to it would leave the old uniqueness in force.

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, upserting on the owning scope plus the category.

The filter matches the full scope, not the category alone, so one app cannot overwrite or take ownership of another app's policy. SetUpsert is what makes the first save persist: UpdateOne against a filter that matches nothing returns MatchedCount 0 and a nil error, so an insert fallback guarded by `if err != nil` would never run and the write would be silently dropped.

func (*Store) SaveReport 

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

SaveReport persists a generated compliance report.

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 updates the stream's head hash and sequence after append.

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"`

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

StreamModel is the grove ORM model for the chronicle_streams collection.

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.