Event sourcing

lith/events keeps an event log in the app's own database, beside its other tables. A command appends events to a stream inside its Apply, and the events, the stream's state and the inline projections commit in the command's savepoint, or not at all. The module adds nothing between a handler and the writer: it is a few functions that a command calls, four tables, and a Runner that rebuilds and catches up in the background.

The writer applies one command at a time. A stream's version cannot change between the read and the append, so there is no retry loop. Positions are handed out and committed in one order, so a reader that keeps the last position it saw misses nothing. And an inline projection commits with its events, so a page never sees an event without its effect.

The board example event-sources one thing, a card's title. A card is a row of cards, which AddCard makes and MoveCard moves; its stream holds its renames, and two projectors keep cards.title and a history table current. The code is in site/internal/docs/example/board/events.go.

Streams and event types

A kind of stream has a key type, int64 or string, and an interface that its events implement. Each event type is defined once, under the name the table stores, through a helper of one line that makes the compiler check that the type belongs to the stream:

// CardEvent is an event of a card's stream. A card itself is a row of
// cards, which AddCard makes; its stream holds its renames.
type CardEvent interface{ cardEvent() }

// Renamed is a card's rename.
type Renamed struct {
	Title string `json:"title"`
}

func (Renamed) cardEvent() {}

// Cards is the kind of the cards' streams: a stream per card, by its id.
var Cards = events.NewStream[int64, CardEvent]("card")

func defineCard[T CardEvent](name string) events.Type[int64, CardEvent, T] {
	return events.Define(Cards, name, func(t T) CardEvent { return t })
}

// CardRenamed is the type of Renamed, stored as card.renamed.
var CardRenamed = defineCard[Renamed]("card.renamed")

Go cannot constrain one type parameter by another, so events.Define takes the conversion to the stream's interface as a function: a type that does not implement CardEvent fails to compile in defineCard. A stream's id comes from the stream, Cards.ID(7), and takes only the stream's key type. Events are stored as JSON; the kind and the type names are what the table holds, so they never change once events are stored.

A decider

A decider is what a command needs to decide, folded from the stream's own events, never from a table shaped for pages: the functional decider pattern. Decide turns the command and the state into events, Evolve folds one event into the state. Both are pure:

// titled is what a rename decides on: the title of the card's last
// rename, folded from its stream.
type titled struct {
	Title string
}

// Rename is the decider of RenameCard.
var Rename = events.Decider[RenameCard, CardEvent, titled]{
	Decide: decideRename,
	Evolve: func(s titled, e CardEvent) titled {
		switch e := e.(type) {
		case Renamed:
			s.Title = e.Title
		}
		return s
	},
	Version: 1,
}

// decideRename is the decision alone: no database, so a test runs it
// with eventstest.
func decideRename(c RenameCard, s titled) ([]CardEvent, error) {
	title := strings.TrimSpace(c.Title)
	switch {
	case title == "" || utf8.RuneCountInString(title) > MaxTitle:
		return nil, &writer.Refusal{Msg: fmt.Sprintf("A title has 1 to %d characters.", MaxTitle)}
	case title == s.Title:
		return nil, nil
	}
	return []CardEvent{Renamed{Title: title}}, nil
}

The store keeps each stream's state as JSON in a row of lith_event_state, written in the same savepoint as the events, so a command reads one row instead of replaying the stream. A decider is made for one command type; the commands of one stream kind share the state type, Evolve and Version, each with a Decide of its own, and Execute refuses a decider whose state or version differs from the first it saw for the kind.

Version is the version of Evolve and of the state's shape. Each row records the version it was folded with, and a stream whose row has another version is folded again from its events at its next command. Bump it whenever Evolve or the state change; a new state field that old events fill needs the bump too. Fields of the state must survive JSON: unexported fields are lost when the row is written.

Running a command on a stream

events.Execute does the rest in the command's Apply: it reads the state row, checks the expected version, decides, appends the events, saves the row and runs the inline projectors:

func (c RenameCard) Apply(tx *writer.Tx) error {
	var one int
	err := tx.QueryRow(`SELECT 1 FROM cards WHERE id = ?`, c.Card).Scan(&one)
	if errors.Is(err, sql.ErrNoRows) {
		return ErrNotFound
	}
	if err != nil {
		return err
	}
	return events.Execute(Events, tx, Rename, Cards.ID(c.Card), events.Any, c, events.Meta{At: c.At})
}

The card's existence is a fact of cards, not of its stream, so the command checks it there. An error from any step fails the command, and its savepoint takes everything back: no event, no state row, no projection change, no topic. A refusal from Decide is an error like any other, here a *writer.Refusal that the board's Act shows in the tab.

The expected version is the third argument after the stream's id. A stream's version counts its events, and the version that matters is the one the page showed when the person acted:

The check comes before the decision. A stream at another version fails the command with an error that wraps events.ErrConflict, and nothing changes. The board's pages show a card's title but not its stream's version, so a rename takes the last word; a page that shows the version sends it with the action, and the command passes events.Exactly(c.Seen).

events.Meta records who and what caused the event: the actor (a user, a session), a correlation id (the request) and a causation id (the command's op id, which writer.Once deduplicates). At defaults to the time of the append and is stored in Unix milliseconds, so events of the same second keep their order on a page; their order in the log comes from their positions.

events.Append appends events that need no decision, such as an import's, with the same version check and the same projectors. It drops the stream's state row, so the stream's next Execute refolds it from all its events. Go infers the stream's event type from its id and refuses a struct in its place, so a single event goes through events.One[CardEvent](Renamed{...}), and several through []CardEvent{...}....

Testing a decider

lith/events/eventstest tests a decider as given, when, then: the events the stream holds, a command, and the events or the error it causes. It runs the fold and the decision without a database:

func TestDecideRename(t *testing.T) {
	named := eventstest.For(board.Rename).Given(board.Renamed{Title: "Draft"})
	named.When(board.RenameCard{Card: 1, Title: " Final "}).Then(t, board.Renamed{Title: "Final"})
	named.When(board.RenameCard{Card: 1, Title: "Draft"}).Then(t)
	named.When(board.RenameCard{Card: 1, Title: " "}).ThenError(t, &writer.Refusal{Msg: "A title has 1 to 200 characters."})
}

Then(t) with no events checks that the command changes nothing. ThenError matches with errors.Is, or by the message, as for a refusal. Each case also decides on the state as the store keeps it, read back from JSON, and fails when the two decisions differ, as they do when the state has a field that JSON drops.

Inline projectors

A projector keeps a read model current from one kind of stream. Its inline handlers run in the transaction of the command that appends, after the events are stored: pages read the model in their next tick, and never see an event without its effect. Each handler is tied to one event type, so it gets its event's struct without a type switch:

// history keeps card_history, every title a card had, for its panel. The
// table is the projector's own, so a new version of it is rebuilt from
// the events while pages read the old table.
var history = events.NewProjector(Cards, "card_history", 1,
	events.Handle(CardRenamed, func(tx *writer.Tx, r events.Record[int64, Renamed]) error {
		_, err := tx.Exec(`INSERT INTO card_history (card_id, version, title, at) VALUES (?, ?, ?, ?)`,
			r.ID, r.Version, r.Data.Title, r.At.Unix())
		tx.Touch(CardTopic(r.ID))
		return err
	}),
).Tables("card_history")

// Events is the board's event store. The app runs an events.Runner on it,
// which rebuilds a projector whose version changed.
var Events = events.Must(events.New(titles, history))

A projector handles or ignores (events.Ignore) every event type of its stream: events.New refuses one that forgets a type, two projectors with one name, and two streams with one kind, and events.Must turns that into a panic as the program starts. A handler touches the topics of what it changed, as a command does; events.Execute touches events.Topic besides, for views over the whole log.

Tables names the tables the projector writes, and only it: the app's migrations create them (003_card_history.sql here), and a rebuild replaces them. The board's other projector, card_titles, updates cards.title, a table that moves change too, so it names no tables and a new version of it rebuilds nothing.

Inline handlers keep read models and do nothing else. A rebuild runs them again for every stored event, so a handler that appended to another stream, sent a command or called out would do it again with each rebuild. A reaction to an event, a notification for an issue's assignee say, is a catch-up projector: it handles each event once its commit is in, and sends its command with writer.Once keyed by the event's position (w.Do(ctx, writer.Once("notify:"+strconv.FormatInt(r.Position, 10), cmd))), so that a retry after a failure does not send it twice.

The Runner

events.Runner keeps a store's projectors current in one database: it rebuilds what changed and runs the catch-up projectors. The app starts it beside the writer, on a context that ends before the writer closes:

runCtx, stop := context.WithCancel(ctx)
runner := events.NewRunner(board.Events, db, w, events.Options{})
done := make(chan error, 1)
go func() { done <- runner.Run(runCtx) }()
defer func() {
	stop() // before the writer closes
	<-done
}()

Run first records new projectors and changed versions in lith_projections, in one command, and returns an error there, such as a projector's missing table. In commands, a missing table stops only its own stream kind: Execute and Append on that kind fail with an error that names the table, and the other kinds go on. Without a runner, inline projectors still run in every command; nothing is rebuilt and no catch-up projector moves. runner.Stats() lists each projector with its version, its status (live, rebuilding or parked), its position and its last error.

Rebuilds

A new version of an inline projector, or a projector new to a database whose stream has events, is rebuilt from the events:

1. The runner creates an empty copy of each of its tables, card_history__v2 with its indexes, from the table's own CREATE statements.
2. It fills the copy in batches of Options.Batch events (500), each a command of its own, so other commands run between them. The handlers write their tables by name, so for the length of each batch the copy takes the table's name and the table steps aside. The batches run in tx.Untouched, which drops the topics the handlers touch: no page reads the copy.
3. The batch that finds fewer events left is the swap: it applies them, and the copy keeps the table's name. Pages read the old table until that batch commits and the new one from then on, never a table half built. The swap touches the projector's topic, lith:projection:card_history, for views that list it.
4. The next command drops the old table. That is one DROP TABLE, as long as SQLite takes to free the table's pages: for a large projection, one long command that the other commands wait for.

While it rebuilds, pages stay current: the events of new commands reach the old table too, through the projector's new handlers, each projector in a savepoint of its own (tx.Savepoint), and touch their topics. The new handlers may not fit the old table, a column the new version adds, say: a failure there is logged and undone, and the person's command goes on. The copy gets every event all the same, and the swap replaces the table. Commands decide from their streams' state rows, not from projections.

So a projection table guards nothing: while its projector rebuilds, a UNIQUE constraint on it, a project's slug say, fails only a handler on the old table, which is logged, and the command that broke the rule goes on. A rule across streams that must hold is a table that no projector writes, project_slugs (slug UNIQUE, project), written by the command itself in the same transaction, where the constraint refuses the command. A rebuild that stops, with a deploy, goes on from its last batch with the next runner.

On the development machine, a rebuild of 100,000 events took about a second in 200 commands. Every batch renames tables, a schema change, so each statement a reader has prepared is prepared again after each batch, 53 µs for a lookup by id, which did not show in the readers' mean lookup of 4.4 µs (the numbers).

The tables are renamed with PRAGMA legacy_alter_table on, so views and triggers elsewhere that name the table go on naming whatever table holds the name. What the module asks of a projector's tables: only the projector writes them, they have no triggers of their own, and no foreign key points at them. After a rebuild to version 2, an index x of the table is named x__v2, since SQLite cannot rename an index. A migration that changes such an index drops both names, DROP INDEX IF EXISTS x and DROP INDEX IF EXISTS x__v2, before it creates the new one.

Catch-up projectors

A catch-up projector's handlers (events.After) run after the commit, outside the writer, from a checkpoint the projector keeps in lith_projections: for mail, a search service or another system's API. The writer wakes them when a commit appends their events; nothing polls. A projector handles an event at least once, so the event's position goes along as the idempotency key:

hook := events.NewProjector(Cards, "card_hook", 1,
	events.After(Created, func(ctx context.Context, r events.Record[int64, CardCreated]) error {
		body := strings.NewReader(fmt.Sprintf(`{"card": %d, "title": %q}`, r.ID, r.Data.Title))
		req, err := http.NewRequestWithContext(ctx, "POST", "https://hooks.example.com/cards", body)
		if err != nil {
			return err
		}
		req.Header.Set("Idempotency-Key", strconv.FormatInt(r.Position, 10))
		res, err := http.DefaultClient.Do(req)
		if err != nil {
			return err // retried with backoff, then parked
		}
		res.Body.Close()
		if res.StatusCode >= 300 {
			return fmt.Errorf("the hook answered %s", res.Status)
		}
		return nil
	}),
	events.Ignore(Renamed),
	events.Ignore(Archived),
).SkipHistory() // a new hook posts the cards made from now on

A failed event is tried again after Options.Backoff (a second), then after twice that, up to MaxBackoff (a minute). After Retries failures (8, about two minutes) the projector parks: it stops at the event, lith_projections shows parked: and the error, and so does Stats. One broken mail holds up only its own projector. runner.Resume(name) tries the event again, and so does the next Run. A new catch-up projector, and one whose version changed, starts at the first event, or after the last one with SkipHistory.

Monitoring

runner.Stats() has the database's head, the last position committed, and each projector's lag: how many positions a catch-up projector or a rebuild is behind the last event of its stream kind. Every kind shares the positions, so at most that many events are left; a live inline projector commits with its events and has no lag. The runner reads the head as it starts and moves it after each commit that appends, so reading it costs no query. store.Stats() counts the events committed by stream kind, the appends that ErrConflict refused, and the refolds: streams whose state Execute folded again from their events, after a decider's Version changed or Append dropped the state row.

lith/prom exports them as <namespace>_events_* when the app passes the runner and the store, prom.Sources{Writer: w, Hub: hub, Events: runner, Store: board.Events}. The lith apps dashboard shows them in its Events row, and two alerts fire for a projector parked for a minute and for a live catch-up projector more than 1,000 positions behind for 10 minutes.

Reading events

events.Load yields a stream's events and events.Since every event after a position, as iterators over the transaction they are given: a GET's, a render's sqlite.QuerierFrom(ctx), a command's. Breaking off the loop closes the rows. Load decodes and upcasts each event; Since gives them as stored, as events.Raw, for an activity feed whose view lists events.Topic, and events.Decode turns one into its type.

The tables come with the first command that needs them, so a page can render before they exist: the first page of a new app, or one that renders while the app starts. Until then the reads find no events and return no error, Load, LoadUntil, Since, SinceUntil, StateAt and Bounds alike, and the page needs no check of its own.

Pages render from projections. A render that replays a stream grows with its history: the review checklist flags it.

Looking back to a position

A position names a moment of the whole log, so a page can show the app as it stood at any of them, the position of a slider, say. Three reads take an upper position:

• events.LoadUntil(ctx, store, q, id, after, upto) yields a stream's events stored up to position upto, and events.SinceUntil(ctx, q, after, upto) every event between two positions.
• events.StateAt(ctx, store, q, decider, id, upto) folds a stream's events up to upto into the decider's state: what a command on it decided from at the time, with the stream's version then.
• events.Bounds(ctx, q, Cards) returns the first and the last position of a kind, 0 and 0 without events: the ends of a slider, with a seek in the log's index for each and no query of the app's own on lith_events. After Compact, the first is the newest snapshot (Retention).
• A projection as of a position is a table of its own whose rows carry the positions between which they held. A change ends the current row at the event's position and starts the next row there, and one query reads the rows that held at a position:

var titleHistory = events.NewProjector(Cards, "title_history", 1,
	events.Handle(Created, func(tx *writer.Tx, r events.Record[int64, CardCreated]) error {
		_, err := tx.Exec(`INSERT INTO title_history (card_id, title, valid_from) VALUES (?, ?, ?)`, r.ID, r.Data.Title, r.Position)
		return err
	}),
	events.Handle(Renamed, func(tx *writer.Tx, r events.Record[int64, CardRenamed]) error {
		if err := endRow(tx, r.ID, r.Position); err != nil {
			return err
		}
		_, err := tx.Exec(`INSERT INTO title_history (card_id, title, archived, valid_from)
			SELECT card_id, ?3, archived, ?1 FROM title_history WHERE card_id = ?2 AND valid_to = ?1`, r.Position, r.ID, r.Data.To)
		return err
	}),
	events.Handle(Archived, func(tx *writer.Tx, r events.Record[int64, CardArchived]) error {
		if err := endRow(tx, r.ID, r.Position); err != nil {
			return err
		}
		_, err := tx.Exec(`INSERT INTO title_history (card_id, title, archived, valid_from)
			SELECT card_id, title, 1, ?1 FROM title_history WHERE card_id = ?2 AND valid_to = ?1`, r.Position, r.ID)
		return err
	}),
).Tables("title_history")

// endRow ends a card's current row at position pos.
func endRow(tx *writer.Tx, card, pos int64) error {
	_, err := tx.Exec(`UPDATE title_history SET valid_to = ? WHERE card_id = ? AND valid_to IS NULL`, pos, card)
	return err
}

rows, err := q.QueryContext(ctx, `SELECT card_id, title FROM title_history
	WHERE valid_from <= ?1 AND (valid_to IS NULL OR valid_to > ?1)`, pos)

The table's key is (card_id, valid_from), so one card's row at a position is a lookup. The test beside it checks, at every position of a random history, that the table, a replay of the log and each card's StateAt agree. A rebuild fills the table like any other, so the history table can come long after the events.

What a look costs on the development machine, for 100 cards and a log of 10,000 events at a random position, in one read transaction: 0.46 ms from the history table, 12.6 ms for a replay of the log up to the position (SinceUntil, Decode and the fold, 2.5 µs an event), 0.12 ms for one card's StateAt over its events up to the position. A slider gives each tab a position of its own, so the view that shows the past is the tab's, rendered for that tab in every tick its slider moves: plan the history table for whole pages and StateAt for one stream's page, and key a shared view by the position when tabs can land on the same ones. The site's Town in time is such a page: a town of tiles, a stream per tile, and a slider whose position is the tab's state, with the town as of each position a view of its own.

Event shapes that change

A new optional field needs nothing: old events decode without it, at its zero value. A change beyond that gets a new type with a new name, such as card.renamed.v2, and an upcaster reads the old type as the new one: events.Upcast(old, new, fn), passed to events.New, with fn turning the old struct into the new. Stored events are never changed. Projectors and deciders see only the newest type, and a command can no longer append the old one. The one exception to stored events never changing is Compact's snapshots (Retention). Upcasters chain, from v1 to v2 to v3.

Each event type keeps a JSON sample in testdata, as it was first stored, and a test decodes and upcasts every sample. A field renamed instead of upcast then fails in CI, not on old rows in production:

func TestEventSamples(t *testing.T) {
	eventstest.Samples(t, board.Events, board.Cards, os.DirFS("testdata/card"))
}

Samples fails for a type without a sample, and for a sample with a field its type lacks.

Retention

Event tables grow like any other, and each stream kind needs a rule for what it keeps:

• events.Forget(store, tx, id) deletes a stream's events and its state row: a person's erasure, or a card archived long ago. Nothing can be rebuilt from them afterwards, so the command deletes the projections' rows of the stream too.
• events.Prune(store, tx, Cards, keep) deletes the events of a kind that an events.Retention lets go, oldest first, for a sweep. Since keeps the events stored from a time on and Newest the newest so many; with both, an event goes only when both let it go. Limit caps the events one call takes, so that a sweep spreads a long backlog over its runs. Prune keeps the state rows and the projections, so commands and pages go on as before, but a rebuild or a new projector sees only the events left: a stream whose events all went comes back as if it had none, and a decider whose version changes keeps the state of a stream that lost its first events as it was.
• events.Compact(store, tx, snapshot, keep, decider, of) is Prune that closes the books: the streams that lose events keep their state in a snapshot (below).

Neither Prune nor Compact deletes an event that a catch-up projector of the kind, or a rebuild under way, has not passed. Keep personal data out of events where history must stay, or delete whole streams with their projections' rows. The board's sweep forgets the streams of cards archived a month ago, in the same command that deletes their history:

func forgetCards(tx *writer.Tx, before int64) error {
	rows, err := tx.Query(`SELECT id FROM cards WHERE archived < ?`, before)
	if err != nil {
		return err
	}
	var gone []int64
	for rows.Next() {
		var id int64
		if err := rows.Scan(&id); err != nil {
			return err
		}
		gone = append(gone, id)
	}
	if err := rows.Err(); err != nil {
		return err
	}
	rows.Close()
	for _, id := range gone {
		if _, err := events.Forget(Events, tx, Cards.ID(id)); err != nil {
			return err
		}
	}
	return nil
}

Positions use AUTOINCREMENT, so a position is never handed out twice, also after retention deleted the newest events: a catch-up projector that keeps its position never skips the event that would otherwise take a deleted one's place.

Closing the books

A rebuild after Prune misses what the deleted events said. events.Compact keeps it: each stream that loses events keeps one, its last event that the retention lets go, rewritten in place as a snapshot of the stream's state as of that event. The snapshot is an event type of the stream, defined like any other. The decider's Evolve sets the state from it, and every projector of the stream handles or ignores it, which events.New checks as for every type. Compact folds each stream's state with the decider, as StateAt does, and a function of the app makes the snapshot from it. The site's Town in time keeps its newest 5,000 events and a snapshot of each tile from before them:

// snapshotOf is a tile's snapshot of its state.
func snapshotOf(s tile) Snapshot { return Snapshot{Kind: s.Kind} }

	if _, err := events.Compact(Events, tx, TileSnapshot, events.Retention{Newest: keep}, PlaceTile, snapshotOf); err != nil {
		return err
	}

Rewriting a stored event is the one deliberate exception to events never changing. The snapshot keeps the event's position, version, time and metadata, and only its type and data change: it holds what the events it replaces held, at the place of the last of them. A rebuild, a new projector, and a decider whose version changed all fold from it. The town's test rebuilds both its projectors after a compaction and finds the tables they had; without the snapshots, a tile whose last change was older than the newest 5,000 events comes back as grass.

Positions stay in order, so reads as of a position stay right from the newest snapshot on. events.Bounds returns it as the kind's first position, and from there StateAt, LoadUntil, a replay of the log and a history table read what they read before; StateAt refuses a position before a stream's snapshot, whose state there is gone. A stream keeps one snapshot at most, so the snapshots that stay are bounded by the number of streams, and a later Compact rewrites only the streams that lost events since. It refuses a stream whose first events Prune took, as it cannot fold that stream's state.

A bound per stream, kept as an index row

Compact cuts a kind at one position, the first event stored at or after Since. Some apps need a bound per stream: a co-edited document may fold only the commits that one of its saved versions holds, so each document has its own oldest commit that must stay. Scanning every stream for it on each sweep grows with the streams. Keep the bound in a small table instead, one row per stream that has events past its bound:

CREATE TABLE uncut (
    kind TEXT NOT NULL,
    id   TEXT NOT NULL,
    at   INTEGER NOT NULL,  -- the stream's oldest event that must stay, Unix ms
    PRIMARY KEY (kind, id)
) STRICT, WITHOUT ROWID;
CREATE INDEX uncut_at ON uncut (kind, at);

The command that appends to a stream inserts its row with the event's time when there is none (INSERT OR IGNORE), so the row keeps the oldest. The command that moves a stream's bound, saving a version say, deletes the row and inserts the first event past the new bound, read from lith_events by the stream and its version, if there is one. The sweep reads SELECT min(at) FROM uncut WHERE kind = ?, one step down the index, and passes it as Since, or a fixed age when that is earlier: every event at or after the oldest bound stays, in every stream. Give Compact a Limit and send it again while it deletes events. Compact remembers where its lookup of Since ended, so while the bound only moves on, a call reads neither the events a Limit left nor those the streams keep. A stream whose bound never moves holds every other one back, so the app moves it on a schedule as well, such as a version every few minutes of editing.

The tables

The first command that needs them creates four tables, as writer.Once creates lith_ops:

The tables are the app's to read: a query of its own over lith_events, such as the newest events first for a feed, goes through the read pool like any other.

What an event costs

An Execute with one inline projector writes six B-trees: the event's row, its two indexes and sqlite_sequence, the state row and the projection. On the development machine, an Intel i5-13500, on 16 October 2026 with other jobs running, BenchmarkExecute appended 6,675 events a second for one producer and 18,342 for 200 with synchronous=NORMAL, where BenchmarkWriter's transfers of four statements reached 24,404 and 81,087 in the same hour (the numbers).