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

eventlog

package
v1.18.0 Latest Latest
Warning

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

 Go to latest Published: Oct 25, 2026 License: MIT Imports: 12 Imported by: 0

README

Retained events

Each product owns an independent scoped log and its authentication, payloads and HTTP routing. A scoped eventlog.Reader is the consumer seam for both embedded and HTTP clients. No global ordering, delivery workers, registrations or ACKs.

Query uses an opaque After cursor, exact Types (unknown values are valid), and Limit (default 100, maximum 500). Filters are sorted/deduplicated; at most 100 input values, each 1–128 bytes after trimming. Keep the same filter when resuming a cursor. ErrInvalidCursor means malformed/wrong-scope/wrong-filter; ErrExpiredCursor means retained history cannot continue and current state must be resynchronized. Product HTTP adapters map these to 400 and 410 respectively. OccurredAt is domain time; RecordedAt is storage time.

Log.Source and PostgreSQL Config.Source name the stable product/log namespace (for example openrails). Event.Source preserves product provenance (host or notification in billing). A consumer subscription source must identify the deployment/product log. Deduplication keys combine that source, its scope and the event ID; Event.Source or a bare ID does not identify a fact globally.

Consumer.Poll loads the owner's checkpoint, reads a page and calls the owner's commit function. The commit function applies idempotent effects and saves NextCursor atomically, comparing the previous checkpoint or serializing writers. Checkpoint keys include source, scope, subscription and canonical filter. Empty pages must also commit: their cursor can advance across irrelevant events. If effects are external, use source/scope/event-ID keys for idempotency before saving the checkpoint. The owner controls polling, retries and resynchronization; errors never reset a cursor automatically.

Log centralizes paging and retention for existing storage through Snapshot. Adapters must provide one repeatable-read snapshot and preserve their legacy cursor codec. SkipFloorBoundary exists only for legacy logs that never stored a pruned boundary ID; new logs must retain and validate it. Imported history must preserve IDs, positions, epochs, boundaries and recording timestamps. If that is impossible, explicitly return 410; never silently reset a cursor.

PostgreSQL

postgres.Migrate installs versioned tables/functions in a host-created schema. postgres.New binds a namespace/source and scope. Create the initial empty stream with EnsureScope during scope provisioning (append also creates it). AppendTx queues a fact inside the mutation transaction. SQL capture triggers can insert the same columns into eventlog_pending, omitting transaction_id and ordinal. Never force its deferred constraint immediate or change those internal columns. The deferred trigger locks all touched scopes in sorted order, then assigns positions and monotonic recording timestamps. Rolled-back mutations leave no facts. An identical retained fact is an idempotent no-op. Reusing its ID with different content aborts the transaction; records are immutable. Products must never reuse an ID for a different fact, even after pruning. The log does not keep an unbounded ID ledger.

The pending identity is only intra-transaction ordering, never a public cursor. There is no ordering between independent scopes. Existing SQL journal capture, restore suppression, scope checks and domain snapshot functions remain owned by the product; adopting the shared queue must preserve them.

Retention defaults to 90 days and is enforced by reads even without pruning. Call Prune periodically with a bounded batch (1–10,000 rows per scoped call). Hosts own pool lifetime, schema permissions and scheduling. Test PostgreSQL with:

EVENTLOG_TEST_DATABASE_URL=postgres://... GOWORK=off go test ./eventlog/...

Products with existing data can embed postgres.MigrationSQL(schema, 1) verbatim in their own transactional migration, verify that generated segment in a contract test, add product-owned foreign keys, copy history unchanged and switch source capture in that same transaction. The shared row's internal boundary may differ from its public ID; imports must preserve both. New pending facts default their boundary to the ID. Imported nonmonotonic recording timestamps use a retained prefix for expiry/pruning plus per-row time filtering, so a later expired row cannot hide an earlier retained fact. Individually expired imported rows are omitted even behind a younger position; this retained-only compatibility mode is not lossless catch-up of expired history. New writes record monotonic times.

Documentation

Overview

Package eventlog defines retained, independently scoped event logs and consumer-owned checkpoints. It does not register consumers or deliver events.

Index

Constants

View Source 
const (
	DefaultLimit     = 100
	MaxLimit         = 500
	MaxTypes         = 100
	DefaultRetention = 90 * 24 * time.Hour
)

Variables

View Source 
var (
	ErrInvalidQuery  = errors.New("eventlog: invalid query")
	ErrInvalidCursor = errors.New("eventlog: invalid cursor")
	ErrExpiredCursor = errors.New("eventlog: cursor expired; resynchronize current state")
)

Functions

This section is empty.

Types

type Consumer 

type Consumer struct {
	Reader       Reader
	Subscription Subscription
	Limit        int
	Load         func(ctx context.Context, key string) (string, error)
	Commit       func(ctx context.Context, key, previous string, page Page) error
}

Consumer polls one page. The owner schedules retries and polling. Load returns an empty cursor only for a new subscription. Commit must atomically apply idempotent effects and persist page.NextCursor, conditional on the checkpoint still equaling previous (or serialize consumers). It must commit empty pages, too. With external effects, scope each event ID by subscription source and scope when forming idempotency keys before saving. Neither Load nor Commit is producer-side state.

func (Consumer) Poll 

func (c Consumer) Poll(ctx context.Context) (hasMore bool, err error)

type Cursor 

type Cursor struct {
	Version  int      `json:"v"`
	Source   string   `json:"source"`
	Scope    string   `json:"scope"`
	Epoch    string   `json:"epoch"`
	Types    []string `json:"types,omitempty"`
	Position int64    `json:"position"`
	Boundary string   `json:"boundary,omitempty"`
}

type CursorCodec 

type CursorCodec interface {
	Encode(Cursor) string
	Decode(string) (Cursor, error)
}

CursorCodec allows a product to retain its already-issued cursor envelope during adoption. Decode must reject malformed/oversized input, unsupported versions and negative positions. Missing boundaries must never be invented.

type Event 

type Event struct {
	ID string `json:"id"`
	// Source is product-owned provenance (for example host or notification).
	// It is not the log namespace and is not a global deduplication key.
	Source     string          `json:"source"`
	Scope      string          `json:"scope"`
	Type       string          `json:"type"`
	Version    int             `json:"version"`
	OccurredAt time.Time       `json:"occurred_at"`
	RecordedAt time.Time       `json:"recorded_at"`
	Subject    Subject         `json:"subject"`
	Data       json.RawMessage `json:"data"`
}

Event is a product-owned fact. IDs are opaque; only cursors define order. RecordedAt is storage time; OccurredAt may describe an earlier domain event.

type JSONCursor 

type JSONCursor struct{}

func (JSONCursor) Decode 

func (JSONCursor) Decode(raw string) (Cursor, error)

func (JSONCursor) Encode 

func (JSONCursor) Encode(c Cursor) string

type Log 

type Log struct {
	// NormalizeQuery preserves historical filter semantics during migration.
	// Nil uses Normalize. Legacy normalizers must retain bounded input and limits.
	NormalizeQuery func(Query) (Query, error)

	// SkipFloorBoundary preserves legacy logs that never stored a pruned boundary ID.
	// Only enable for an existing cursor contract that already omitted this check.
	SkipFloorBoundary bool

	// Source names the product/log namespace, independent of Event.Source.
	Source       string
	Scope        string
	Retention    time.Duration
	Now          func() time.Time
	Codec        CursorCodec
	ReadSnapshot func(context.Context, func(Snapshot) error) error
}

Log centralizes paging, retention and cursor validation over product storage. A codec is optional for new logs; existing logs supply their legacy codec. ReadSnapshot must run the callback once using a single committed snapshot.

func (Log) ListEvents 

func (l Log) ListEvents(ctx context.Context, q Query) (Page, error)

type Page 

type Page struct {
	Events           []Event `json:"events"`
	NextCursor       string  `json:"next_cursor"`
	HasMore          bool    `json:"has_more"`
	RetentionSeconds int64   `json:"retention_seconds"`
}

type Position 

type Position struct {
	Number int64
	ID     string
}

Position and Stream describe storage state within one consistent snapshot. Floor includes records expired by read-time retention even before pruning.

type Query 

type Query struct {
	After string   `json:"after,omitempty"`
	Types []string `json:"types,omitempty"`
	Limit int      `json:"limit,omitempty"`
}

func Normalize 

func Normalize(q Query) (Query, error)

Normalize copies, sorts and deduplicates exact filters. Unknown types are valid, allowing older producers and newer consumers to coexist.

type Reader 

type Reader interface {
	ListEvents(context.Context, Query) (Page, error)
}

Reader is already bound to an authorized source and scope. Embedded and HTTP product clients implement this same interface; authentication stays with them.

type ReaderFunc 

type ReaderFunc func(context.Context, Query) (Page, error)

func (ReaderFunc) ListEvents 

func (f ReaderFunc) ListEvents(ctx context.Context, q Query) (Page, error)

type Record 

type Record struct {
	Event    Event
	Position int64
	// Boundary overrides Event.ID for legacy stores with independent boundary tokens.
	Boundary string
}

type Snapshot 

type Snapshot interface {
	Stream(context.Context, time.Time) (Stream, error)
	Boundary(context.Context, int64) (string, error)
	Events(context.Context, int64, []string, int) ([]Record, error)
}

Snapshot must expose committed rows from one repeatable-read snapshot. Events returns strictly increasing positions above after, filtered by exact types (empty means all), with at most limit records. Head's Floor is the last expired position at cutoff. Legacy storage with nonmonotonic RecordedAt must compute the retained prefix floor and additionally exclude expired rows in Events.

type Stream 

type Stream struct {
	Epoch string
	Head  Position
	Floor Position
}

type Subject 

type Subject struct {
	Type string `json:"type"`
	ID   string `json:"id"`
}

type Subscription 

type Subscription struct {
	Source, Scope, Name string
	Types               []string
}

Subscription identifies a consumer checkpoint. Changing any field selects a different checkpoint. Source names the deployment/product log, not Event.Source provenance. A filter change must not reuse the old cursor.

func (Subscription) Key 

func (s Subscription) Key() (string, error)

Source Files

View all Source files

Directories

Path Synopsis
Package postgres stores independent scoped logs in a host-owned PostgreSQL schema.
 Package postgres stores independent scoped logs in a host-owned PostgreSQL schema.

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.