cqrs

README

cqrs

A small, opinionated seam between event-sourced domain entities and the infrastructure they happen to run on.

The whole idea is four interfaces. An entity is written against those and nothing else — it never names a broker, a driver, or a database handle. What you get for that discipline is entities you can lift from one service into another unchanged, and command/projection logic you can unit-test as a pure function of its input events, with no database and no message broker in sight.

import "github.com/jrgensen/cqrs"

go get github.com/jrgensen/cqrs

Requires Go 1.25 or later (inherited from github.com/jrgensen/stream).

---

The seam

Interface	Side	Method set
cqrs.Publisher	command	Publish(Message) error, MessageFunc() MessageFunc
cqrs.Consumer	projection	Consumes() []Subject, HandleMessage(Message) error
cqrs.Writer	projection	Consume(statement string) error
cqrs.Reader	query	the read-only method set of *sql.DB

A command decides something and publishes an event. A consumer folds that event into the read model by handing statements to a Writer. A query reads the read model back through a Reader. That is the entire model.

flowchart TD
    H[HTTP handler / caller]
    H -->|command| C[commander]
    C -->|Publish| P[cqrs.Publisher - the event stream]
    P -->|replay + live| K[cqrs.Consumer]
    K -->|Consume statement| DL[deadletter.Writer]
    DL -->|on success| SP[sqlpersister.Writer]
    DL -->|on failure, once armed| T[(deadletter table)]
    SP --> DB[(read model)]
    DB --> R[cqrs.Reader]
    R --> Q[querier]
    Q --> H

Three properties of this shape are worth stating outright, because entities are written on the assumption that they hold:

• Writer returns no result. Deliberately. A Writer may execute the statement, buffer it, log it, or divert a failing one to a dead-letter table, and the caller cannot tell. That is what makes the decorator in deadletter possible. It also means a Writer is only ever used to project, never to read back.
• HandleMessage must be idempotent. Projections are rebuilt by replaying the stream from sequence zero on every boot, so every event will be handled again, and an event may be redelivered within a single run. In practice this means upserts (ON DUPLICATE KEY UPDATE, INSERT IGNORE) rather than plain inserts.
• *sql.DB satisfies Reader as it stands, so no adapter is needed when wiring up. Reader is also a superset of goqu.SQLDatabase, so a Reader can be handed straight to goqu.New. Note that *sql.Tx does not satisfy it: it has no Begin, since a transaction cannot start another one.

Relationship to the stream package

Message, MutableMessage, MessageFunc and Subject are aliases for the corresponding github.com/jrgensen/stream types, and Publisher and Consumer are declared with method sets identical to stream.Publisher and stream.Consumer.

Aliases rather than wrappers, on purpose: the jetstream transport and the streamtest fakes satisfy these interfaces implicitly, so there is no adapter layer that can go stale. Compile-time assertions at the bottom of cqrs.go keep both directions honest — add a method to either side and one of them stops compiling.

This is the one place in the module that names a concrete messaging library. Replace the transport and cqrs.go changes; no entity does.

Subjects

SubjectFromStr parses the canonical string form, where the domain is separated from the type by either : or .:

s := cqrs.SubjectFromStr("NATHEJK:2026.patrulje.abc123.updated")

s.Domain()  // "NATHEJK"
s.Type()    // "2026.patrulje.abc123.updated"
s.Subject() // "NATHEJK.2026.patrulje.abc123.updated"  — ':' is canonicalised to '.'
s.Match("nathejk.*.patrulje.*.updated") // true

It exists so entities can name subjects without importing a transport package for the constructor alone.

---

Anatomy of an entity

The convention this seam grew up with is one package per entity, holding three small types that share nothing but the tables they touch: a commander (write side), a consumer (projection), and a querier (read side). A single constructor takes the three collaborators and hands back all three roles.

package patrulje

import (
	"context"
	"fmt"

	"github.com/jrgensen/cqrs"
)

// --- command side: decisions become events -------------------------------

type commander struct{ p cqrs.Publisher }

func (c *commander) Rename(ctx context.Context, teamID, name string) error {
	// Messages are never constructed directly: ask the Publisher for one bound
	// to a subject, so the message takes whatever form the transport needs.
	msg := c.p.MessageFunc()(cqrs.SubjectFromStr(
		fmt.Sprintf("NATHEJK:2026.patrulje.%s.updated", teamID)))
	if err := msg.SetBody(&Renamed{TeamID: teamID, Name: name}); err != nil {
		return err
	}
	return c.p.Publish(msg)
}

// --- projection side: events become rows ---------------------------------

type consumer struct{ w cqrs.Writer }

func (c *consumer) Consumes() []cqrs.Subject {
	return []cqrs.Subject{cqrs.SubjectFromStr("NATHEJK:*.patrulje.*.updated")}
}

func (c *consumer) HandleMessage(msg cqrs.Message) error {
	if !msg.Subject().Match("NATHEJK.*.patrulje.*.updated") {
		return nil
	}
	var body Renamed
	if err := msg.Body(&body); err != nil {
		return err
	}
	// Idempotent: replay must be able to run this again.
	return c.w.Consume(fmt.Sprintf(
		`INSERT INTO patrulje SET teamId=%q, name=%q
		 ON DUPLICATE KEY UPDATE name=VALUES(name)`, body.TeamID, body.Name))
}

// --- query side: rows become answers -------------------------------------

type querier struct{ db cqrs.Reader }

func (q querier) ByID(ctx context.Context, teamID string) (*Patrulje, error) {
	var p Patrulje
	err := q.db.QueryRowContext(ctx,
		"SELECT teamId, name FROM patrulje WHERE teamId = ?", teamID).
		Scan(&p.TeamID, &p.Name)
	return &p, err
}

// --- assembly ------------------------------------------------------------

type table struct {
	commander
	consumer
	querier
}

func New(p cqrs.Publisher, w cqrs.Writer, r cqrs.Reader) (*table, error) {
	t := &table{
		commander: commander{p: p},
		consumer:  consumer{w: w},
		querier:   querier{db: r},
	}
	// Own your schema: creating it is just another statement for the Writer.
	// This runs before the writer is armed, so a broken DDL still fails loudly.
	return t, w.Consume(createTableSQL)
}

Note what is not in that package: no driver import, no broker import, no *sql.DB. That is the point.

---

Wiring it up

The application is the only place that picks concrete infrastructure. The ordering below matters, and the comments say why.

package main

import (
	"context"
	"database/sql"
	"log"

	_ "github.com/go-sql-driver/mysql"
	"github.com/jrgensen/cqrs/deadletter"
	"github.com/jrgensen/cqrs/sqlpersister"
	"github.com/jrgensen/stream/jetstream"
	"github.com/jrgensen/stream/xstream"
)

func main() {
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()

	// jetstream.New returns a stream.Stream, which embeds stream.Publisher and
	// therefore satisfies cqrs.Publisher directly. No adapter.
	js, err := jetstream.New(jetstreamDSN)
	if err != nil {
		log.Fatal(err)
	}

	// The three collaborators. These are the only concrete infrastructure
	// choices in the program.
	publisher := js
	reader := db // *sql.DB satisfies cqrs.Reader
	writer := deadletter.New(sqlpersister.New(db), db)

	// 1. Create the safety net's own table — loudly, before Arm.
	if err := writer.Consume(writer.CreateTableSql()); err != nil {
		log.Fatal(err)
	}
	// 2. Projections replay from sequence zero on every boot, so anything
	//    captured during the previous run's replay is stale. Clear it.
	if err := writer.Reset(); err != nil {
		log.Fatal(err)
	}

	// 3. Construct entities. Their CREATE TABLEs and any seeding run through
	//    the still-unarmed writer, so a mistake here is fatal, as it should be.
	teams, err := patrulje.New(publisher, writer, reader)
	if err != nil {
		log.Fatal(err)
	}

	// 4. Startup is done. From here a single bad statement is captured in the
	//    deadletter table instead of killing the consumer loop.
	writer.Arm()

	mux := xstream.NewMux(js)
	mux.AddConsumer(teams)
	if err := mux.Run(context.Background()); err != nil {
		log.Fatal(err)
	}

	// ... serve HTTP, using teams for both commands and queries.
}

To stamp every published message with default metadata (a producer tag, a build version) rather than setting it at each call site, wrap the transport in stream/metatagger — it also satisfies cqrs.Publisher.

---

Sub-packages

Everything that names a specific technology lives below the root package, so nothing above it has to.

sqlpersister — the production Writer

Executes each statement against a *sql.DB, echoing any statement that fails to a log sink before returning the error. This is the bottom of the Writer stack; it only ever reports failure upwards and never decides what to do about it.

w := sqlpersister.New(db)
err := w.Consume("UPDATE patrulje SET name='Ulven' WHERE teamId='abc'")

The echo goes to os.Stderr by default. WithLogWriter redirects it into whatever logger the application already has:

w := sqlpersister.New(db, sqlpersister.WithLogWriter(logger.Writer()))

// Or silence it — reasonable under deadletter, which persists the failing
// statement and its error to a table anyway.
w := sqlpersister.New(db, sqlpersister.WithLogWriter(io.Discard))

The echo is a diagnostic, not a contract: a log sink that fails, or a nil one, never masks or replaces the error Consume reports. If the Writer is used concurrently, the sink must be safe for concurrent use.

It is dialect-agnostic, and not by effort: a statement arrives as an opaque string and is handed to Exec. Be clear about what that does and does not buy. The package works against any database/sql driver; the statements flowing through it are only as portable as whoever produced them.

deadletter — the safety net

A cqrs.Writer decorator. A single event whose payload cannot be written (a value that overflows its VARCHAR, say) must not take down the whole consumer loop — and because the stream uses an ordered consumer with no redelivery, it must not be silently dropped either. This package captures such failures in a deadletter table so they can be inspected and replayed.

w := deadletter.New(sqlpersister.New(db), db)

w.Consume(w.CreateTableSql()) // create its table, before arming
w.Reset()                     // discard last run's stale captures

// ... startup DDL and seeding: failures here are still fatal ...

w.Arm() // from here, failures are captured and Consume returns nil

n, err := w.Count() // how much broke during replay?

The two-phase behaviour is the whole design:

	before Arm	after Arm
inner Consume succeeds	nil	nil
inner Consume fails	error returned unchanged	captured, nil returned
capture itself fails	—	logged, nil returned

Nothing after Arm is ever fatal. Startup work, which happens before it, still fails loudly and visibly.

The captured table is deadletter (id, createdAt, query, error, resolvedAt). MySQL/MariaDB (the default), PostgreSQL and SQLite ship as Dialect values:

w := deadletter.New(inner, db, deadletter.WithDialect(deadletter.Postgres))

A Dialect is plain data — two strings, the CREATE TABLE and the capture INSERT — because there is nothing to dispatch on. A database with no built-in variant is supported from the outside without changing the package:

w := deadletter.New(inner, db, deadletter.WithDialect(deadletter.Dialect{
	CreateTable: `CREATE TABLE IF NOT EXISTS deadletter (...)`,
	Insert:      `INSERT INTO deadletter (query, error) VALUES (:1, :2)`,
}))

Insert must take exactly two bind parameters: the statement, then the error message. Reset and Count are plain ANSI SQL and need no variant.

The dialect concerns this package's own table only. Statements flowing through Consume come from your projections and are passed through verbatim.

cqrstest — in-memory fakes

Writer records the statements a projection produces; Publisher records the events a command produces. Neither needs a database or a broker, so an entity's logic can be tested directly:

func TestRenamePublishesUpdated(t *testing.T) {
	pub := &cqrstest.Publisher{}
	teams, _ := patrulje.New(pub, &cqrstest.Writer{}, nil)

	if err := teams.Rename(ctx, "abc123", "Ulven"); err != nil {
		t.Fatal(err)
	}

	if got := pub.Subjects(); got[0] != "NATHEJK.2026.patrulje.abc123.updated" {
		t.Errorf("subject = %q", got[0])
	}
	var body patrulje.Renamed
	pub.Messages[0].Body(&body) // decoded exactly as a real consumer would
}

func TestProjectsRename(t *testing.T) {
	w := &cqrstest.Writer{}
	teams, _ := patrulje.New(&cqrstest.Publisher{}, w, nil)

	w.Reset() // drop the CREATE TABLE from construction
	teams.HandleMessage(evt)

	if !strings.Contains(w.Last(), "name='Ulven'") {
		t.Errorf("statement = %q", w.Last())
	}
}

Useful details:

• Writer.Statements holds every statement in order, including failed ones, so a test can assert on what a broken projection attempted. Last() and Matching(substr) narrow it down; Reset() clears it.
• Setting Err on either fake makes every call fail, which is how error paths get exercised. The statement or message is still recorded.
• Bodies and metadata round-trip through real JSON, so a test sees the same field-tag behaviour a real consumer would — notably omitempty.
• Reading a body or metadata that was never set is not an error; it decodes as an empty object.
• None of these types are safe for concurrent use. Tests should not need them to be.

---

Schema migration helpers

EnsureColumn and EnsureIndex are the missing piece of the CREATE TABLE IF NOT EXISTS pattern the projections use: that statement creates new tables but never adds columns to existing ones. Both are idempotent and safe to call on every startup.

// ddl is the fragment following "ADD COLUMN".
cqrs.EnsureColumn(reader, writer, "product", "sizes",
	"sizes VARCHAR(255) NOT NULL DEFAULT '' AFTER eligibleFor")

// ddl is the complete statement, so UNIQUE / FULLTEXT / composite are yours.
cqrs.EnsureIndex(reader, writer, "order_line", "idx_order_line_member",
	"ALTER TABLE order_line ADD INDEX idx_order_line_member (memberId)")

The existence check goes through the Reader and the change through the Writer, matching the split everywhere else. A failed check never leads to a speculative ALTER.

These two are MySQL/MariaDB-specific, unlike the rest of the root package. They use INFORMATION_SCHEMA lookups, which SQLite does not have at all (it uses PRAGMA), and SQLite cannot ALTER TABLE ADD INDEX. They live here because every projection that has outgrown its original schema needs them and there is no dialect-neutral way to express them worth the indirection.

---

Testing this module

go test ./...

The tests use go-sqlmock rather than a real database, and mock Reader that way too — *sql.Rows and *sql.Row are concrete types and cannot be fabricated by hand.

goqu appears in go.mod only for a compile-time assertion in cqrs_test.go that Reader remains usable as a goqu.SQLDatabase. Both it and go-sqlmock are test-only dependencies of this module: neither is imported by any package you can import, so neither ends up in your binary (go list -deps on a consumer confirms it). They may still show up as checksums in your go.sum, which is normal for the module graph and costs you nothing at build time.

---

Extending it

• New Writer behaviour (metrics, batching, a statement log): write a decorator like deadletter and slot it into the stack. Because Consume returns no result, nothing upstream notices.
• A different database: implement Writer yourself, or keep sqlpersister and supply a deadletter.Dialect. Remember that your projections' statements are the part that actually has to be portable.
• A different transport: anything satisfying stream.Publisher / stream.Consumer already satisfies cqrs.Publisher / cqrs.Consumer. If the underlying messaging library changes, cqrs.go is the only file that moves.
• Reader: it mirrors the *sql.DB method set and is a superset of goqu.SQLDatabase. Add methods if you must; removing one breaks every entity that builds queries with goqu.

---

License

MIT © Klaus Jørgensen

Documentation

Overview

Package cqrs defines the seam between event-sourced domain entities and the infrastructure they happen to run on.

An entity in this architecture has three collaborators, and nothing else:

Publisher — the command side. Appends domain events to the event stream.
Writer    — the projection side. Applies statements that build a read model.
Reader    — the query side. Reads the read model back out.

Entities depend on these interfaces only. They never name a broker, a driver, or a database handle. That is what makes them portable between repositories and trivially testable.

Implementations

The concrete adapters live in sub-packages, so everything that names a specific technology sits below this one and nothing above it has to:

sqlpersister — Writer that executes statements against a database.
deadletter   — Writer decorator that diverts failing statements to a
               table instead of letting them kill the projection loop.
cqrstest     — in-memory Writer and Publisher for tests.

The production Publisher and Reader have no adapter here: a metatagger over JetStream and an *sql.DB respectively satisfy the interfaces as they stand.

Relationship to the stream package

Message, MutableMessage, MessageFunc and Subject are aliases for the corresponding github.com/jrgensen/stream types, and Publisher and Consumer are declared with method sets identical to stream.Publisher and stream.Consumer. Aliases rather than wrappers, deliberately: the existing jetstream transport and the streamtest fakes satisfy these interfaces implicitly, so no adapter layer exists to go stale. The compile-time assertions at the bottom of this file keep both directions honest.

The aliases are the one place that names a concrete messaging library. When the event transport is replaced, this file changes and no entity does.

Example

package main

import (
	"context"
	"fmt"

	"github.com/jrgensen/cqrs"
	"github.com/jrgensen/cqrs/cqrstest"
)

// This example is the shape every entity in this architecture takes: a command
// side that turns a decision into an event, and a projection side that folds
// that event into the read model. Neither names a broker or a driver — only the
// cqrs interfaces — which is what lets the whole round trip be exercised here
// with the in-memory fakes from cqrstest.
//
// The query side is omitted for brevity; it is a plain struct holding a
// cqrs.Reader, which *sql.DB satisfies directly.

// renamed is the event body.
type renamed struct {
	TeamID string `json:"teamId"`
	Name   string `json:"name"`
}

type team struct {
	p cqrs.Publisher
	w cqrs.Writer
}

// Rename is the command side. Messages are not constructed directly: ask the
// Publisher for one bound to a subject, so it takes whatever form the transport
// needs.
func (t *team) Rename(ctx context.Context, teamID, name string) error {
	msg := t.p.MessageFunc()(cqrs.SubjectFromStr(
		fmt.Sprintf("NATHEJK:2026.patrulje.%s.updated", teamID)))
	if err := msg.SetBody(&renamed{TeamID: teamID, Name: name}); err != nil {
		return err
	}
	return t.p.Publish(msg)
}

// Consumes declares the subjects this projection cares about.
func (t *team) Consumes() []cqrs.Subject {
	return []cqrs.Subject{cqrs.SubjectFromStr("NATHEJK:*.patrulje.*.updated")}
}

// HandleMessage is the projection side. It is idempotent, because the stream is
// replayed from sequence zero on every boot.
func (t *team) HandleMessage(msg cqrs.Message) error {
	if !msg.Subject().Match("NATHEJK.*.patrulje.*.updated") {
		return nil
	}
	var body renamed
	if err := msg.Body(&body); err != nil {
		return err
	}
	return t.w.Consume(fmt.Sprintf(
		"INSERT INTO patrulje SET teamId=%q, name=%q"+
			" ON DUPLICATE KEY UPDATE name=VALUES(name)", body.TeamID, body.Name))
}

var _ cqrs.Consumer = (*team)(nil)

func main() {
	pub, w := &cqrstest.Publisher{}, &cqrstest.Writer{}
	t := &team{p: pub, w: w}

	if err := t.Rename(context.Background(), "abc123", "Ulven"); err != nil {
		fmt.Println("rename:", err)
		return
	}
	fmt.Println(pub.Subjects()[0])

	// Feed the published events back in, as the stream mux would.
	for _, msg := range pub.Messages {
		if err := t.HandleMessage(msg); err != nil {
			fmt.Println("handle:", err)
			return
		}
	}
	fmt.Println(w.Last())

}

Output:
NATHEJK.2026.patrulje.abc123.updated
INSERT INTO patrulje SET teamId="abc123", name="Ulven" ON DUPLICATE KEY UPDATE name=VALUES(name)

Index

• func EnsureColumn(r Reader, w Writer, table, column, ddl string) error
• func EnsureIndex(r Reader, w Writer, table, index, ddl string) error
• type Consumer
• type Message
• type MessageFunc
• type MutableMessage
• type Publisher
• type Reader
• type Subject
  • func SubjectFromStr(s string) Subject
• type Writer

Examples

• Package

Functions

func EnsureColumn

func EnsureColumn(r Reader, w Writer, table, column, ddl string) error

EnsureColumn adds a column to a table if it doesn't already exist. It's the missing piece of the "CREATE TABLE IF NOT EXISTS" pattern the projections use: that statement creates new tables but never adds columns to existing ones.

ddl is the part of an ALTER TABLE statement that follows "ADD COLUMN", for example:

"sizes VARCHAR(255) NOT NULL DEFAULT '' AFTER eligibleFor"

The existence check goes through the Reader and the change through the Writer, matching the split everywhere else: reads return values, writes are statements the Writer may divert or defer.

The check uses INFORMATION_SCHEMA.COLUMNS, so it works on both MySQL and MariaDB without requiring a specific server version. EnsureColumn is idempotent and safe to call on every startup.

func EnsureIndex

func EnsureIndex(r Reader, w Writer, table, index, ddl string) error

EnsureIndex adds a named index to a table if it doesn't already exist. Like EnsureColumn but for indexes; ddl is the full ALTER TABLE statement so the caller has full control over UNIQUE, FULLTEXT, composite columns, etc.:

"ALTER TABLE order_line ADD INDEX idx_order_line_member (memberId)"

Types

type Consumer

type Consumer interface {
	Consumes() []Subject
	HandleMessage(msg Message) error
}

Consumer is the projection side of an entity: it declares the subjects it cares about and folds each matching event into the read model, normally by handing statements to a Writer.

HandleMessage must be idempotent. Projections are rebuilt by replaying the stream from the beginning on every boot, so every event will be handled again, and an event may be redelivered within a single run.

type Message

type Message = stream.Message

Message is a single immutable event read off the stream.

type MessageFunc

type MessageFunc = stream.MessageFunc

MessageFunc builds a MutableMessage for a subject. A Publisher hands one out so callers construct messages in whatever form that particular transport requires.

type MutableMessage

type MutableMessage = stream.MutableMessage

MutableMessage is an event being composed, before it is published.

type Publisher

type Publisher interface {
	Publish(msg Message) error
	MessageFunc() MessageFunc
}

Publisher is the command side of an entity: it turns a decision into an event on the stream.

Messages are not constructed directly. Ask MessageFunc for a message bound to a subject, set its body, then Publish it:

msg := p.MessageFunc()(cqrs.SubjectFromStr("NATHEJK:2026.patrulje.abc.updated"))
msg.SetBody(&body)
return p.Publish(msg)

Publish may be synchronous or asynchronous, and must be safe for concurrent use.

type Reader

type Reader interface {
	Begin() (*sql.Tx, error)
	BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error)
	Exec(query string, args ...interface{}) (sql.Result, error)
	ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
	Prepare(query string) (*sql.Stmt, error)
	PrepareContext(ctx context.Context, query string) (*sql.Stmt, error)
	Query(query string, args ...interface{}) (*sql.Rows, error)
	QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
	QueryRow(query string, args ...interface{}) *sql.Row
	QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row
}

Reader is the query side of the read model.

The method set is that of *sql.DB, minus the connection-lifecycle calls (Close, Ping, SetMaxOpenConns, Stats, ...) that belong to whoever owns the pool rather than to an entity. Two consequences worth knowing about:

• *sql.DB satisfies Reader as it stands, so no adapter is needed at the call site in main. Note that *sql.Tx does not: it has no Begin, since a transaction cannot start another one.
• Reader is a superset of goqu.SQLDatabase, so a Reader can be handed straight to goqu.New. Removing a method here will break the entities that build queries with goqu; add, don't subtract.

Mock it with a database/sql mock (github.com/DATA-DOG/go-sqlmock) rather than by hand — the *sql.Rows and *sql.Row return types are concrete and cannot be fabricated otherwise.

type Subject

type Subject = stream.Subject

Subject addresses an event, e.g. "NATHEJK:2026.patrulje.abc123.updated".

func SubjectFromStr

func SubjectFromStr(s string) Subject

SubjectFromStr parses the canonical string form of a subject, where the domain is separated from the type by either ':' or '.':

NATHEJK:2026.patrulje.abc123.updated

It exists so entities can name subjects without importing a transport package for the constructor alone.

type Writer

type Writer interface {
	Consume(statement string) error
}

Writer applies a single statement to the read model.

The contract is deliberately this thin. A Writer is free to execute the statement, log it, buffer it, or divert a failing one to a dead-letter queue; callers cannot tell and must not care. Because there is no result value, a Writer is only ever used to project — never to read back.