redis

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

Documentation

Overview

Package redis implements the Chronicle store interface using Grove KV with the Redis driver.

Index

Constants

View Source 
const DefaultKeyPrefix = "chronicle:"

DefaultKeyPrefix is what every key starts with unless New is given WithKeyPrefix. The names below are relative to it: with the default, the event index zEventAll is the key "chronicle:z:evt:all".

Variables

View Source 
var (
	// ErrScopeKeysNotMigrated is returned by GetStreamByScope when Migrate has
	// not yet moved the scope indexes to the v2 key format. See Migrate.
	ErrScopeKeysNotMigrated = errors.New("chronicle/redis: scope keys are in the pre-v2 format, run Migrate")

	// ErrScopeCollision is returned by the Migrate call that finds streams
	// holding events from more than one app+tenant scope. The migration itself
	// has completed when this is returned. See Migrate.
	ErrScopeCollision = errors.New("chronicle/redis: streams hold events from more than one scope")
)

Functions

This section is empty.

Types

type Option 

type Option func(*Store)

Option configures a Store.

func WithKeyPrefix 

func WithKeyPrefix(prefix string) Option

WithKeyPrefix puts every key the store reads or writes under prefix instead of DefaultKeyPrefix, so two stores can share one redis database without seeing each other's data. Migrate, the indexes and every count stay inside the prefix.

The prefix goes in front of each key as it is. End it with a separator such as ":", and don't give two stores prefixes where one starts with the other ("chronicle:" and "chronicle:evt:"), because the shorter one's scans would reach into the longer one's keys.

type Store 

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

Store implements the Chronicle store interface using Redis via Grove KV.

func New 

func New(kvStore *kv.Store, opts ...Option) *Store

New creates a new Redis store backed by Grove KV. Without WithKeyPrefix its keys start with DefaultKeyPrefix, the layout every earlier version used.

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 using a pipeline.

func (*Store) AppendCheckpoint 

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

AppendCheckpoint is not supported by this backend.

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(
	context.Context, id.ID, uint64, uint64,
) ([]*checkpoint.Checkpoint, error)

CheckpointsInRange is not supported by this backend.

func (*Store) Close 

func (s *Store) Close() error

Close closes the KV store.

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.

The read and write are not atomic. Only the erasure service writes a record after creating it, once, so there is no competing writer to lose.

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.

The per-subject index is not partitioned by scope, so when a scope is given each event is read to confirm ownership. ZCard is only correct unscoped.

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.

Unscoped this is a single ZCARD. Scoped, the erasure index is not partitioned by app/tenant, so each record is read to confirm ownership.

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. The scope matches exactly, so an empty TenantID means 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(context.Context, id.ID) (*checkpoint.Checkpoint, error)

GetCheckpoint is not supported by this backend.

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.

Chronicle.resolveStream creates a new stream on ErrStreamNotFound, so a miss has to be true. Before Migrate has moved the scope index to the v2 format, every existing scope misses, and reporting that would start a second chain for each tenant. Until then a miss returns ErrScopeKeysNotMigrated instead. The check costs one EXISTS, and only on a miss.

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(context.Context, id.ID) (*checkpoint.Checkpoint, error)

LatestCheckpoint is not supported by this backend.

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(
	context.Context, id.ID, checkpoint.ListOpts,
) ([]*checkpoint.Checkpoint, error)

ListCheckpoints is not supported by this backend.

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 moves the scope indexes from the bare-joined key format to the v2 format built by scopeSuffix, once. A marker key records that it has run, so every call after the first costs one EXISTS.

The old keys joined app, tenant and category with ":", so app "a:b" with tenant "c" and app "a" with tenant "b:c" shared a key. Their names cannot be split back into parts, so Migrate does not rename them. It rebuilds each v2 index from the entities themselves (streams, events, policies), which store every scope part as a separate field, and then deletes the old keys.

This runs in Migrate rather than as a read-through fallback in GetStreamByScope. The per-scope event index has to be rebuilt anyway, and a sorted set cannot be patched up one lookup at a time: until it is rebuilt, a scoped Query, Count or retention purge silently misses every event recorded before the upgrade. A fallback on the stream key alone would make skipping Migrate look safe when it is not. GetStreamByScope instead returns ErrScopeKeysNotMigrated on a miss until the marker exists. A deployment that skips Migrate gets that error on its first event, and no tenant's chain forks.

Rebuilding is idempotent, so a Migrate interrupted part way is safe to run again. Run it with every writer on this version: a process still on the old version keeps writing old-format keys, which nothing reads once the marker is set. If that happened, delete the key "meta:scope-key-format" under the store's key prefix ("chronicle:meta:scope-key-format" by default) and run Migrate again.

A collision that already happened cannot be undone here. When a second scope found the first scope's stream, it appended its events into that chain, and every later hash covers them. Moving them out would break verification for the rest of the chain. Migrate still finishes and indexes every event under the scope it was recorded with, so each tenant's queries see only its own events again, and the scope that created the stream keeps it. The other scope's next event starts a chain of its own. The shared chains are listed in the error, which wraps ErrScopeCollision, and kept in the set "meta:scope-collisions" under the store's key prefix ("chronicle:meta:scope-collisions" by default) for whoever has to investigate. Only the Migrate call that does the migration returns that error.

A retention policy lost to a collision cannot be found either: the old SavePolicy deleted it when the colliding scope saved its own. Re-save it.

func (*Store) Ping 

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

Ping checks Redis 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, replacing any existing policy for the same (app, tenant, category).

The uniqueness key includes the owning scope. Keying on category alone let a save from one app evict another app's policy for that category.

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.

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.