archive

type Archiver ¶

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

Archiver copies the audit and event rings out to object storage.

func New(src Source, cursors CursorStore, sealer secrets.Sealer, dest func() (volumes.ObjectStore, bool), clk clock.Clock, log *slog.Logger) *Archiver

func (a *Archiver) Sweep(ctx context.Context) int

type Cursor ¶

type Cursor struct {
	Ms  int64    `json:"ms"`
	IDs []string `json:"ids,omitempty"`
}

Cursor is the archiver's resume point for one stream, persisted in the replicated KV.

It is a millisecond watermark plus the ids already written at exactly that millisecond. A bare watermark cannot be both complete and duplicate-free: exclusive (> ms) drops entries sharing the boundary millisecond, inclusive (>= ms) re-writes them. Audit ids are minted before the raft round trip so they are not a strict watermark either, hence time plus an id set.

type CursorStore ¶

type CursorStore interface {
	PutCursor(ctx context.Context, key string, value []byte) error
}

CursorStore persists the resume cursor through raft.

type Reader ¶

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

Reader fetches archived records back out of object storage.

func NewReader(store volumes.ObjectStore, sealer secrets.Sealer) *Reader

func (r *Reader) Audit(ctx context.Context, sinceMs, untilMs int64, limit int, keep func(*zatterav1.AuditEntry) bool) ([]*zatterav1.AuditEntry, error)

func (r *Reader) Events(ctx context.Context, sinceMs, untilMs int64, limit int, keep func(*zatterav1.Event) bool) ([]*zatterav1.Event, error)

type Source ¶

type Source interface {
	AuditSince(sinceMs int64) []*zatterav1.AuditEntry
	EventsSince(sinceMs int64) []*zatterav1.Event
	KV(key string) ([]byte, int64, int64, bool)
}

Source is the state the archiver reads. *state.Store satisfies it.

type Stream ¶

type Stream string

Stream names the two archived histories; it is also the object key prefix.

const (
	StreamAudit  Stream = "audit"
	StreamEvents Stream = "events"
)