readview

eventsourced Readview handler

Package readview provides materialized view projections built from event streams for use with the eventsourced framework.

Install

go get gitlab.com/unboundsoftware/eventsourced/readview

Prerequisites

• PostgreSQL 12+
• An eventsourced.EventStore implementation (e.g., pg)

Getting Started

Define an event handler and a reset handler, then create and start the readview:

import (
    "context"
    "database/sql"

    "github.com/jmoiron/sqlx"

    "gitlab.com/unboundsoftware/eventsourced/eventsourced"
    "gitlab.com/unboundsoftware/eventsourced/readview"
)

// Event handler: update read model within a transaction
func userEventHandler(ctx context.Context, tx *sql.Tx, event eventsourced.Event) error {
    switch e := event.(type) {
    case *UserCreated:
        _, err := tx.ExecContext(ctx,
            "INSERT INTO users (id, name) VALUES ($1, $2)",
            e.AggregateIdentity(), e.Name,
        )
        return err
    }
    return nil
}

// Reset handler: clear data before a full rebuild
func userResetHandler(ctx context.Context, name string, tx *sqlx.Tx) error {
    _, err := tx.ExecContext(ctx, "TRUNCATE TABLE users")
    return err
}

// Create and start readview
db := sqlx.NewDb(sqlDB, "postgres")
rv, err := readview.New(db, store,
    readview.Register("user_list", userEventHandler, userResetHandler),
)
if err != nil {
    log.Fatal(err)
}

if err := rv.Start(ctx); err != nil {
    log.Fatal(err)
}

Backfill

On startup, Start checks each readview's current position against the event store's maximum global sequence number. If behind, it replays historical events through the event handler.

Progress is committed in batches (configurable via WithBackFillBatchSize) and tracked with heartbeat timestamps to detect stale backfills.

AMQP Integration

After backfill completes, wire up the readview handlers with AMQP for live event processing:

handlers := rv.Handlers()

// Regular events (filters out external source events)
conn.Handle("user_list", rv.RegularMappingHandler(conn, "user_list", handlers["user_list"]))

// External events (no filtering)
conn.Handle("external_list", rv.ExternalMappingHandler(conn, handlers["external_list"]))

Distributed Coordination

Multiple application instances coordinate via PostgreSQL row-level locks (FOR UPDATE NOWAIT). Only one instance can backfill a given readview at a time. Others wait and retry.

The heartbeat mechanism detects stale backfills: if an instance crashes during backfill, another instance can take over after the heartbeat duration expires.

Resetting a Readview

Call Reset to clear a readview's data and re-backfill from scratch:

err := rv.Reset(ctx, "user_list")

This calls the registered ResetHandler, deletes the status row, and initiates a full backfill.

Database Schema

The readview creates and manages a readview_status table (migrations tracked in goose_db_version_readview):

Configuration Options

Error Handling