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

sqlite

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

Documentation

Overview

Package sqlite implements the Chronicle store interface using grove ORM with the SQLite driver.

Index

Constants

This section is empty.

Variables

View Source 
var Migrations = func() *migrate.Group {
	g := migrate.NewGroup("chronicle")
	g.MustRegister(
		&migrate.Migration{
			Name:    "create_streams_table",
			Version: "20240101000000",
			Comment: "Create chronicle_streams table",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS chronicle_streams (
    id          TEXT PRIMARY KEY,
    app_id      TEXT NOT NULL,
    tenant_id   TEXT NOT NULL DEFAULT '',
    head_hash   TEXT NOT NULL DEFAULT '',
    head_seq    INTEGER NOT NULL DEFAULT 0,
    created_at  TEXT NOT NULL DEFAULT (datetime('now')),
    updated_at  TEXT NOT NULL DEFAULT (datetime('now')),

    UNIQUE(app_id, tenant_id)
);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `DROP TABLE IF EXISTS chronicle_streams;`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_events_table",
			Version: "20240101000001",
			Comment: "Create chronicle_events table with hash chain columns",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS chronicle_events (
    id              TEXT PRIMARY KEY,
    stream_id       TEXT NOT NULL REFERENCES chronicle_streams(id),
    sequence        INTEGER NOT NULL,
    hash            TEXT NOT NULL,
    prev_hash       TEXT NOT NULL DEFAULT '',

    -- Scope
    app_id          TEXT NOT NULL,
    tenant_id       TEXT NOT NULL DEFAULT '',
    user_id         TEXT NOT NULL DEFAULT '',
    ip              TEXT NOT NULL DEFAULT '',

    -- What
    action          TEXT NOT NULL,
    resource        TEXT NOT NULL,
    category        TEXT NOT NULL,
    resource_id     TEXT NOT NULL DEFAULT '',
    metadata        TEXT DEFAULT '{}',
    outcome         TEXT NOT NULL DEFAULT 'success',
    severity        TEXT NOT NULL DEFAULT 'info',
    reason          TEXT NOT NULL DEFAULT '',

    -- GDPR
    subject_id          TEXT NOT NULL DEFAULT '',
    encryption_key_id   TEXT NOT NULL DEFAULT '',
    erased              INTEGER NOT NULL DEFAULT 0,
    erased_at           TEXT,
    erasure_id          TEXT,

    timestamp       TEXT NOT NULL,
    created_at      TEXT NOT NULL DEFAULT (datetime('now')),

    UNIQUE(stream_id, sequence)
);

CREATE INDEX IF NOT EXISTS idx_chronicle_events_scope
    ON chronicle_events (app_id, tenant_id, timestamp);
CREATE INDEX IF NOT EXISTS idx_chronicle_events_category
    ON chronicle_events (category, timestamp);
CREATE INDEX IF NOT EXISTS idx_chronicle_events_action
    ON chronicle_events (action, outcome, timestamp);
CREATE INDEX IF NOT EXISTS idx_chronicle_events_user
    ON chronicle_events (user_id, timestamp);
CREATE INDEX IF NOT EXISTS idx_chronicle_events_subject
    ON chronicle_events (subject_id);
CREATE INDEX IF NOT EXISTS idx_chronicle_events_severity
    ON chronicle_events (severity, timestamp);
CREATE INDEX IF NOT EXISTS idx_chronicle_events_resource
    ON chronicle_events (resource, resource_id, timestamp);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `DROP TABLE IF EXISTS chronicle_events;`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_erasures_table",
			Version: "20240101000002",
			Comment: "Create chronicle_erasures table",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS chronicle_erasures (
    id              TEXT PRIMARY KEY,
    subject_id      TEXT NOT NULL,
    reason          TEXT NOT NULL,
    requested_by    TEXT NOT NULL,
    events_affected INTEGER NOT NULL DEFAULT 0,
    key_destroyed   INTEGER NOT NULL DEFAULT 0,
    app_id          TEXT NOT NULL,
    tenant_id       TEXT NOT NULL DEFAULT '',
    created_at      TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_chronicle_erasures_subject ON chronicle_erasures (subject_id);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `DROP TABLE IF EXISTS chronicle_erasures;`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_retention_tables",
			Version: "20240101000003",
			Comment: "Create chronicle_retention_policies and chronicle_archives tables",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS chronicle_retention_policies (
    id          TEXT PRIMARY KEY,
    category    TEXT NOT NULL UNIQUE,
    duration    INTEGER NOT NULL,
    archive     INTEGER NOT NULL DEFAULT 0,
    app_id      TEXT NOT NULL DEFAULT '',
    created_at  TEXT NOT NULL DEFAULT (datetime('now')),
    updated_at  TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE TABLE IF NOT EXISTS chronicle_archives (
    id              TEXT PRIMARY KEY,
    policy_id       TEXT NOT NULL,
    category        TEXT NOT NULL,
    event_count     INTEGER NOT NULL,
    from_timestamp  TEXT NOT NULL,
    to_timestamp    TEXT NOT NULL,
    sink_name       TEXT NOT NULL,
    sink_ref        TEXT NOT NULL DEFAULT '',
    created_at      TEXT NOT NULL DEFAULT (datetime('now'))
);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
DROP TABLE IF EXISTS chronicle_archives;
DROP TABLE IF EXISTS chronicle_retention_policies;
`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_reports_table",
			Version: "20240101000004",
			Comment: "Create chronicle_reports table",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS chronicle_reports (
    id          TEXT PRIMARY KEY,
    title       TEXT NOT NULL,
    type        TEXT NOT NULL,
    period_from TEXT NOT NULL,
    period_to   TEXT NOT NULL,
    app_id      TEXT NOT NULL,
    tenant_id   TEXT NOT NULL DEFAULT '',
    format      TEXT NOT NULL DEFAULT 'json',
    data        TEXT NOT NULL,
    generated_by TEXT NOT NULL DEFAULT '',
    created_at  TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_chronicle_reports_scope ON chronicle_reports (app_id, tenant_id, created_at);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `DROP TABLE IF EXISTS chronicle_reports;`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "scope_retention_to_app_and_tenant",
			Version: "20240101000005",
			Comment: "Key retention policies per (app_id, tenant_id, category) and scope archives",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				_, err := exec.Exec(ctx, `
CREATE TABLE chronicle_retention_policies_new (
    id          TEXT PRIMARY KEY,
    category    TEXT NOT NULL,
    duration    INTEGER NOT NULL,
    archive     INTEGER NOT NULL DEFAULT 0,
    app_id      TEXT NOT NULL DEFAULT '',
    tenant_id   TEXT NOT NULL DEFAULT '',
    created_at  TEXT NOT NULL DEFAULT (datetime('now')),
    updated_at  TEXT NOT NULL DEFAULT (datetime('now')),

    UNIQUE(app_id, tenant_id, category)
);

INSERT INTO chronicle_retention_policies_new
    (id, category, duration, archive, app_id, created_at, updated_at)
SELECT id, category, duration, archive, app_id, created_at, updated_at
FROM chronicle_retention_policies;

DROP TABLE chronicle_retention_policies;

ALTER TABLE chronicle_retention_policies_new RENAME TO chronicle_retention_policies;

ALTER TABLE chronicle_archives ADD COLUMN app_id TEXT NOT NULL DEFAULT '';
ALTER TABLE chronicle_archives ADD COLUMN tenant_id TEXT NOT NULL DEFAULT '';

CREATE INDEX IF NOT EXISTS idx_chronicle_archives_scope
    ON chronicle_archives (app_id, tenant_id, created_at);

CREATE INDEX IF NOT EXISTS idx_chronicle_events_retention
    ON chronicle_events (app_id, tenant_id, category, timestamp);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
DROP INDEX IF EXISTS idx_chronicle_events_retention;
DROP INDEX IF EXISTS idx_chronicle_archives_scope;

CREATE TABLE chronicle_archives_old (
    id              TEXT PRIMARY KEY,
    policy_id       TEXT NOT NULL,
    category        TEXT NOT NULL,
    event_count     INTEGER NOT NULL,
    from_timestamp  TEXT NOT NULL,
    to_timestamp    TEXT NOT NULL,
    sink_name       TEXT NOT NULL,
    sink_ref        TEXT NOT NULL DEFAULT '',
    created_at      TEXT NOT NULL DEFAULT (datetime('now'))
);
INSERT INTO chronicle_archives_old
    (id, policy_id, category, event_count, from_timestamp, to_timestamp, sink_name, sink_ref, created_at)
SELECT id, policy_id, category, event_count, from_timestamp, to_timestamp, sink_name, sink_ref, created_at
FROM chronicle_archives;
DROP TABLE chronicle_archives;
ALTER TABLE chronicle_archives_old RENAME TO chronicle_archives;

CREATE TABLE chronicle_retention_policies_old (
    id          TEXT PRIMARY KEY,
    category    TEXT NOT NULL UNIQUE,
    duration    INTEGER NOT NULL,
    archive     INTEGER NOT NULL DEFAULT 0,
    app_id      TEXT NOT NULL DEFAULT '',
    created_at  TEXT NOT NULL DEFAULT (datetime('now')),
    updated_at  TEXT NOT NULL DEFAULT (datetime('now'))
);
INSERT INTO chronicle_retention_policies_old
    (id, category, duration, archive, app_id, created_at, updated_at)
SELECT id, category, duration, archive, app_id, created_at, updated_at
FROM chronicle_retention_policies
GROUP BY category;
DROP TABLE chronicle_retention_policies;
ALTER TABLE chronicle_retention_policies_old RENAME TO chronicle_retention_policies;
`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "record_hash_scheme",
			Version: "20240101000006",
			Comment: "Record the digest scheme per event and pin it per stream",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				_, err := exec.Exec(ctx, `
ALTER TABLE chronicle_events  ADD COLUMN hash_scheme  TEXT NOT NULL DEFAULT '';
ALTER TABLE chronicle_events  ADD COLUMN hash_key_id  TEXT NOT NULL DEFAULT '';
ALTER TABLE chronicle_streams ADD COLUMN scheme       TEXT NOT NULL DEFAULT 'chronicle/v2';
ALTER TABLE chronicle_streams ADD COLUMN scheme_since INTEGER NOT NULL DEFAULT 0;
-- head_seq is not trusted to equal the true high-water mark: Append derives
-- the next sequence from MAX(sequence), not head_seq, precisely because a
-- crash between the event insert and the head update can leave head_seq
-- lagging. Pinning from a lagging head_seq alone would misclassify the
-- already-written events above it as a permanent downgrade rather than as
-- history. MAX(a, b) below is SQLite's two-argument scalar max, not the
-- single-argument aggregate.
UPDATE chronicle_streams SET scheme_since = MAX(
    head_seq,
    (SELECT COALESCE(MAX(sequence), 0) FROM chronicle_events e WHERE e.stream_id = chronicle_streams.id)
) + 1 WHERE scheme_since = 0;
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
ALTER TABLE chronicle_streams DROP COLUMN scheme_since;
ALTER TABLE chronicle_streams DROP COLUMN scheme;
ALTER TABLE chronicle_events  DROP COLUMN hash_key_id;
ALTER TABLE chronicle_events  DROP COLUMN hash_scheme;
`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_checkpoints_table",
			Version: "20240101000007",
			Comment: "Signed checkpoints over a stream's sequence range",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS chronicle_checkpoints (
    id              TEXT PRIMARY KEY,
    stream_id       TEXT NOT NULL REFERENCES chronicle_streams(id),
    app_id          TEXT NOT NULL,
    tenant_id       TEXT NOT NULL DEFAULT '',
    from_seq        INTEGER NOT NULL,
    to_seq          INTEGER NOT NULL,
    from_hash       TEXT NOT NULL,
    to_hash         TEXT NOT NULL,
    event_count     INTEGER NOT NULL,
    prev_checkpoint TEXT NOT NULL DEFAULT '',
    algorithm       TEXT NOT NULL,
    sign_key_id     TEXT NOT NULL,
    signature       BLOB NOT NULL,
    signed_payload  TEXT NOT NULL,
    created_at      TEXT NOT NULL DEFAULT (datetime('now')),

    -- Two constraints, not one. UNIQUE(stream_id, to_seq) only decides a
    -- race between checkpointers that read the same head, and two replicas
    -- on a busy stream routinely do not: each snapshots the stream list at
    -- the top of its own tick, so replica A can commit 6-10 while replica
    -- B, whose read predated it, is still about to commit 6-15. Both
    -- succeed, and the stream then reports tampered forever, because
    -- checkpoints are append-only and the overlap cannot be removed.
    -- UNIQUE(stream_id, from_seq) is what stops it: both racers derive the
    -- same from_seq from the same stale latest checkpoint, so the loser
    -- fails cleanly with checkpoint.ErrExists.
    UNIQUE(stream_id, to_seq),
    UNIQUE(stream_id, from_seq)
);

CREATE INDEX IF NOT EXISTS idx_chronicle_checkpoints_stream
    ON chronicle_checkpoints (stream_id, to_seq DESC);
CREATE INDEX IF NOT EXISTS idx_chronicle_checkpoints_scope
    ON chronicle_checkpoints (app_id, tenant_id, created_at DESC);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `DROP TABLE IF EXISTS chronicle_checkpoints;`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "fixed_width_timestamps",
			Version: "20240101000008",
			Comment: "Rewrite RFC3339Nano timestamps with a fixed nine-digit fraction",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				for _, tc := range timestampColumns {
					if _, err := exec.Exec(ctx, normalizeTimestampSQL(tc[0], tc[1])); err != nil {
						return fmt.Errorf("normalize %s.%s: %w", tc[0], tc[1], err)
					}
				}
				return nil
			},
			Down: func(context.Context, migrate.Executor) error {

				return nil
			},
		},
		&migrate.Migration{
			Name:    "add_request_correlation_columns",
			Version: "20260922000001",
			Comment: "Add user_agent, request_id and session_id to chronicle_events",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				for _, stmt := range []string{
					`ALTER TABLE chronicle_events ADD COLUMN user_agent TEXT NOT NULL DEFAULT ''`,
					`ALTER TABLE chronicle_events ADD COLUMN request_id TEXT NOT NULL DEFAULT ''`,
					`ALTER TABLE chronicle_events ADD COLUMN session_id TEXT NOT NULL DEFAULT ''`,
					`CREATE INDEX IF NOT EXISTS idx_chronicle_events_session ON chronicle_events (session_id, timestamp DESC) WHERE session_id != ''`,
				} {
					if _, err := exec.Exec(ctx, stmt); err != nil {
						return err
					}
				}
				return nil
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				for _, stmt := range []string{
					`DROP INDEX IF EXISTS idx_chronicle_events_session`,
					`ALTER TABLE chronicle_events DROP COLUMN session_id`,
					`ALTER TABLE chronicle_events DROP COLUMN request_id`,
					`ALTER TABLE chronicle_events DROP COLUMN user_agent`,
				} {
					if _, err := exec.Exec(ctx, stmt); err != nil {
						return err
					}
				}
				return nil
			},
		},
		&migrate.Migration{
			Name:    "add_erasure_status",
			Version: "20260929000001",
			Comment: "Add status and legacy_key_retained to chronicle_erasures",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				for _, stmt := range []string{
					`ALTER TABLE chronicle_erasures ADD COLUMN status TEXT NOT NULL DEFAULT 'completed'`,
					`ALTER TABLE chronicle_erasures ADD COLUMN legacy_key_retained INTEGER NOT NULL DEFAULT 0`,
				} {
					if _, err := exec.Exec(ctx, stmt); err != nil {
						return err
					}
				}
				return nil
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				for _, stmt := range []string{
					`ALTER TABLE chronicle_erasures DROP COLUMN legacy_key_retained`,
					`ALTER TABLE chronicle_erasures DROP COLUMN status`,
				} {
					if _, err := exec.Exec(ctx, stmt); err != nil {
						return err
					}
				}
				return nil
			},
		},
	)
	return g
}()

Migrations is the grove migration group for the Chronicle SQLite 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 string `grove:"from_timestamp"` // TEXT, written with timeLayout
	ToTimestamp   string `grove:"to_timestamp"`   // TEXT, written with timeLayout
	SinkName      string `grove:"sink_name"`
	SinkRef       string `grove:"sink_ref"`
	AppID         string `grove:"app_id"`
	TenantID      string `grove:"tenant_id"`
	CreatedAt     string `grove:"created_at"` // TEXT, written with timeLayout
}

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        uint64 `grove:"from_seq"`
	ToSeq          uint64 `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"` // BLOB
	SignedPayload  string `grove:"signed_payload"`
	CreatedAt      string `grove:"created_at"` // TEXT, written with timeLayout
}

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   int    `grove:"key_destroyed"` // INTEGER boolean
	AppID          string `grove:"app_id"`
	TenantID       string `grove:"tenant_id"`
	CreatedAt      string `grove:"created_at"` // TEXT, written with timeLayout
	Status         string `grove:"status"`

	LegacyKeyRetained int `grove:"legacy_key_retained"` // INTEGER boolean
}

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        uint64 `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        string `grove:"metadata"` // JSON TEXT in SQLite
	Outcome         string `grove:"outcome"`
	Severity        string `grove:"severity"`
	Reason          string `grove:"reason"`
	SubjectID       string `grove:"subject_id"`
	EncryptionKeyID string `grove:"encryption_key_id"`
	Erased          int    `grove:"erased"`    // INTEGER boolean in SQLite
	ErasedAt        string `grove:"erased_at"` // TEXT nullable timestamp
	ErasureID       string `grove:"erasure_id"`
	Timestamp       string `grove:"timestamp"`  // TEXT, written with timeLayout
	CreatedAt       string `grove:"created_at"` // TEXT, written with timeLayout
	HashScheme      string `grove:"hash_scheme"`
	HashKeyID       string `grove:"hash_key_id"`
}

EventModel is the grove ORM model for the chronicle_events table.

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 inside its transaction, 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  string `grove:"period_from"` // TEXT, written with timeLayout
	PeriodTo    string `grove:"period_to"`   // TEXT, written with timeLayout
	AppID       string `grove:"app_id"`
	TenantID    string `grove:"tenant_id"`
	Format      string `grove:"format"`
	Data        string `grove:"data"` // JSON TEXT from compliance.EncodeReportBody
	GeneratedBy string `grove:"generated_by"`
	CreatedAt   string `grove:"created_at"` // TEXT, written with timeLayout
}

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   int    `grove:"archive"`  // INTEGER boolean
	AppID     string `grove:"app_id"`
	TenantID  string `grove:"tenant_id"`
	CreatedAt string `grove:"created_at"` // TEXT, written with timeLayout
	UpdatedAt string `grove:"updated_at"` // TEXT, written with timeLayout
}

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 with SQLite.

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) 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.

This mirrors the postgres backend: the sequence is a positional counter and is not part of the hash chain, so the store owns its allocation. Deriving the next value from the greater of the stored head and the actual MAX(sequence) makes a desynced head self-healing, and doing it inside the transaction that inserts the event keeps concurrent appends from colliding on UNIQUE(stream_id, sequence).

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) 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. Since SQLite does not have generate_series by default, 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 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 runs grove migrations for the Chronicle schema.

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 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,alias:s"`

	ID          string `grove:"id,pk"`
	AppID       string `grove:"app_id"`
	TenantID    string `grove:"tenant_id"`
	HeadHash    string `grove:"head_hash"`
	HeadSeq     uint64 `grove:"head_seq"`
	CreatedAt   string `grove:"created_at"` // TEXT, written with timeLayout
	UpdatedAt   string `grove:"updated_at"` // TEXT, written with timeLayout
	Scheme      string `grove:"scheme"`
	SchemeSince uint64 `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.