The highest tagged major version is
README
¶
architecturekit
Building blocks for DDD-based applications with CQRS and event sourcing in Go, on top of EventSourcingDB – a purpose-built database for event sourcing.
architecturekit covers both sides of an event-sourced application: commands, events, and the state to decide on for writing, and projections, views, and queries for reading. An optional package exposes commands and queries over HTTP.
For more information on EventSourcingDB, see its official documentation.
architecturekit includes a test package to test deciders, projections, and queries without a database. For details, see Testing Deciders. For the tests that need a real database, a second package starts one in a container (see Testing with a Database).
Getting Started
Install the packages:
go get github.com/thenativeweb/architecturekit-golang/architecturekit github.com/thenativeweb/eventsourcingdb-client-golang/eventsourcingdb
Import the package, create an EventSourcingDB client, and create a store by providing the client and the source to use for all events you write:
import (
"net/url"
"github.com/thenativeweb/architecturekit-golang/architecturekit"
"github.com/thenativeweb/eventsourcingdb-client-golang/eventsourcingdb"
)
// ...
baseURL, err := url.Parse("http://localhost:3000")
if err != nil {
// ...
}
apiToken := "secret"
client, err := eventsourcingdb.NewClient(baseURL, apiToken)
if err != nil {
// ...
}
store := architecturekit.NewStore(client, "https://library.eventsourcingdb.io")
The NewStore function returns a *Store, which reads and writes the events of all commands. For details on the client, see the client SDK for Go.
Defining Commands
A command describes what someone wants to do. Define it as a struct and implement two functions: Subject, which returns the subject the command acts on, and Preconditions, which returns the conditions under which its events may be written. This makes the struct a Command:
type AcquireBook struct {
BookID string
Title string
Author string
ISBN string
}
func (c AcquireBook) Subject() string {
return "/books/" + c.BookID
}
func (c AcquireBook) Preconditions() []architecturekit.Precondition {
return []architecturekit.Precondition{
architecturekit.OnStateRead(),
}
}
type BorrowBook struct {
BookID string
ReaderID string
BorrowedUntil string
}
func (c BorrowBook) Subject() string {
return "/books/" + c.BookID
}
func (c BorrowBook) Preconditions() []architecturekit.Precondition {
return []architecturekit.Precondition{
architecturekit.OnStateRead(),
}
}
type ReturnBook struct {
BookID string
}
func (c ReturnBook) Subject() string {
return "/books/" + c.BookID
}
func (c ReturnBook) Preconditions() []architecturekit.Precondition {
return []architecturekit.Precondition{
architecturekit.OnStateRead(),
}
}
OnStateRead makes sure that the events are only written if the state the command is decided on still holds. For the other preconditions, see Using Preconditions.
Defining Events
An event describes what has happened. Define it as a struct with JSON annotations and implement the EventType function, which returns the event type. This makes the struct an Event:
type BookAcquired struct {
Title string `json:"title"`
Author string `json:"author"`
ISBN string `json:"isbn"`
}
func (BookAcquired) EventType() string {
return "io.eventsourcingdb.library.book-acquired"
}
type BookBorrowed struct {
BorrowedBy string `json:"borrowedBy"`
BorrowedUntil string `json:"borrowedUntil"`
}
func (BookBorrowed) EventType() string {
return "io.eventsourcingdb.library.book-borrowed"
}
type BookReturned struct{}
func (BookReturned) EventType() string {
return "io.eventsourcingdb.library.book-returned"
}
The struct becomes the event's data. The subject is taken from the command, and the source from the store.
Describing Events with Schemas
Every event type has a JSON schema, which the database checks every event of the type against, once the schema is registered (see Registering Event Schemas). The kit derives it from the struct, so that it describes exactly what encoding/json writes for the event:
- A struct is an object with the fields
encoding/jsonwrites: named by theirjsontags, without fields tagged-and unexported ones, and with the fields of embedded structs in place of the embedded struct. Fields withomitemptyoromitzero, and fields of an embedded pointer, are optional, all others are required, and no other fields are allowed. - A
stringis a string, aboola boolean, an integer an integer, and a floating-point number a number. Thestringoption of ajsontag turns such a field into a string. - A slice is an array, a
[]bytea string, and an array an array of exactly its length. A map is an object whose values all have the same schema. - A pointer, a slice and a map may also be
null, sinceencoding/jsonwritesnullfornil, unless a field of such a type is optional and therefore left out instead. - A
time.Timeis a string in thedate-timeformat, ajson.Numbera number, unless thestringoption makes it a string, and a type with aMarshalTextor anAppendTextfunction a string. An interface allows any value.
For BookBorrowed, this yields the following schema:
{
"type": "object",
"properties": {
"borrowedBy": { "type": "string" },
"borrowedUntil": { "type": "string" }
},
"required": ["borrowedBy", "borrowedUntil"],
"additionalProperties": false
}
These rules do not change, since a registered schema can not change either.
To constrain a value further than its Go type does, declare a type for it with a Schema function, which returns the JSON schema of the type. Wherever a field has that type, the derived schema takes it over. For example, borrowedUntil accepts any string so far. To make sure that it is a date, BookBorrowed could use a Date type for the field:
type Date string
func (Date) Schema() map[string]any {
return map[string]any{"type": "string", "format": "date"}
}
type BookBorrowed struct {
BorrowedBy string `json:"borrowedBy"`
BorrowedUntil Date `json:"borrowedUntil"`
}
Note that the other examples in this README keep BorrowedUntil a string. With the Date type, they would convert between the two, as in Date(cmd.BorrowedUntil).
A type that encodes itself with a MarshalJSON function, or with MarshalJSONTo of encoding/json/v2, which encoding/json calls as well, needs such a Schema function, too, since the kit can not know what the function writes. So does a type that has its MarshalText or AppendText function on a pointer receiver only: encoding/json calls it only for a value it can take the address of, so whether such a value is written as a string depends on where it is. If the schema of an event can not be derived, for example because of such a type, a recursive type, or a channel, Evolve panics and names the field.
If an event needs a schema that its fields can not express, give the event itself a Schema function. It takes precedence over the derived schema. To start from the derived schema, call the DeriveSchema function, which derives the schema of a type without calling its own Schema function. For example, to require at least one of two optional fields:
type BookCorrected struct {
Title string `json:"title,omitempty"`
Author string `json:"author,omitempty"`
}
func (BookCorrected) EventType() string {
return "io.eventsourcingdb.library.book-corrected"
}
func (BookCorrected) Schema() map[string]any {
schema := architecturekit.DeriveSchema[BookCorrected]()
schema["minProperties"] = 1
return schema
}
DeriveSchema returns a new schema on every call, with objects as map[string]any and arrays as []any, as encoding/json decodes them.
Note that a json tag name that encoding/json considers invalid also makes Evolve panic, since encoding/json reads such a name differently depending on the Go version the application declares.
Defining State
The state holds what a command needs to decide on. Define it as a struct, call the NewState function with its initial value, and call the Evolve function for every event type that changes it:
type Book struct {
IsAcquired bool
IsBorrowed bool
}
var bookState = architecturekit.NewState(Book{}).
Evolve(func(book Book, event BookAcquired) Book {
book.IsAcquired = true
return book
}).
Evolve(func(book Book, event BookBorrowed) Book {
book.IsBorrowed = true
return book
}).
Evolve(func(book Book, event BookReturned) Book {
book.IsBorrowed = false
return book
})
The event type is taken from the event's EventType function, so it does not have to be repeated.
Note that calling Evolve twice for the same event type panics.
Reading an event without a rule fails, since it usually points to a missing rule or a wrong subject. If a subject holds events that matter for no decision, such as BookInspected, which only records that somebody looked at a book, call the Ignore function for their type. The state then takes them without changing, and says so, rather than an Evolve function that returns the state unchanged and needs a comment to explain why:
var bookState = architecturekit.NewState(Book{}).
Evolve(func(book Book, event BookAcquired) Book {
book.IsAcquired = true
return book
}).
// ...
Ignore[BookInspected]()
Note that the data of an ignored event is not decoded, but its schema is still part of Schemas, since the event is still written. Ignoring an event type that has an Evolve rule, or ignoring it twice, panics, and so does calling FromLatest for it.
Note that Execute refuses to write an event that the state of the decider has no rule for, since the state could not read the subject any more afterwards (see Executing Commands). To find out whether a state has a rule for an event type, call the HasRule function on the state with the event type.
Making Decisions
A decider connects a state with the decision made on it. Create a Decider, hand over the state, and provide a Decide function that receives the command and the current state, and returns the events to write:
var acquireBook = architecturekit.Decider[AcquireBook, Book]{
State: bookState,
Decide: func(ctx context.Context, cmd AcquireBook, book Book) ([]architecturekit.Event, error) {
if book.IsAcquired {
return nil, architecturekit.NewDomainError("book %s has already been acquired", cmd.BookID)
}
return []architecturekit.Event{
BookAcquired{
Title: cmd.Title,
Author: cmd.Author,
ISBN: cmd.ISBN,
},
}, nil
},
}
To reject a command, return an error created with the NewDomainError function. It takes a format string and arguments, like fmt.Errorf, and returns an error of the type *DomainError, whose message is exactly the formatted text, and which belongs to the category ErrDomain (see Handling Errors).
A decider may check several rules:
var borrowBook = architecturekit.Decider[BorrowBook, Book]{
State: bookState,
Decide: func(ctx context.Context, cmd BorrowBook, book Book) ([]architecturekit.Event, error) {
if !book.IsAcquired {
return nil, architecturekit.NewDomainError("book %s does not exist", cmd.BookID)
}
if book.IsBorrowed {
return nil, architecturekit.NewDomainError("book %s is already borrowed", cmd.BookID)
}
return []architecturekit.Event{
BookBorrowed{
BorrowedBy: cmd.ReaderID,
BorrowedUntil: cmd.BorrowedUntil,
},
}, nil
},
}
If there is nothing to do, return neither events nor an error:
var returnBook = architecturekit.Decider[ReturnBook, Book]{
State: bookState,
Decide: func(ctx context.Context, cmd ReturnBook, book Book) ([]architecturekit.Event, error) {
if !book.IsBorrowed {
return nil, nil
}
return []architecturekit.Event{
BookReturned{},
}, nil
},
}
Executing Commands
To execute a command, call the Execute function with a context, the store, the decider, and the command:
writtenEvents, err := architecturekit.Execute(
context.TODO(),
store,
acquireBook,
AcquireBook{
BookID: "42",
Title: "2001 – A Space Odyssey",
Author: "Arthur C. Clarke",
ISBN: "978-0756906788",
},
)
if err != nil {
// ...
}
Execute reads the events of the command's subject, evolves the state from them, calls the decider, and writes the events it returns. The function returns the written events, including the fields added by the server. If the decider returns no events, nothing is written, and the function returns nil.
Every event the decider returns needs a rule on the state of the decider, an Evolve function or Ignore. Execute writes the events to the subject that the same state reads for the next command, and the database keeps every event, so an event without a rule would leave a subject that the state can not read any more. If the state has no rule for one of the events, Execute writes none of them, and fails with an error of the category ErrPermanent that names the event type (see Handling Errors). The test fixture reports such an event the same way (see Testing Deciders).
The written events come as they are stored, with their data as JSON. To read the data of one of them, call the Decode function with the type of the event. It returns an Envelope, the same a projection gets (see Defining Projections), with the data in its Data field:
for _, event := range writtenEvents {
switch event.Type {
case (BookAcquired{}).EventType():
acquired, err := architecturekit.Decode[BookAcquired](event)
if err != nil {
// ...
}
// acquired.Data.Title, acquired.ID, ...
}
}
Note that Decode fails with an error of the category ErrPermanent for an event of another type, rather than leaving the fields of the wrong struct empty, and for data that does not fit the type.
Note that Execute only reads the events of the command's subject itself, not those of nested subjects.
Using Preconditions
Every command declares at least one precondition, so that writing without any check is always a decision, never an oversight. There are three kinds:
OnStateReadguards the state the command is decided on.Requireturns a precondition of the client SDK into one of the command, for example to check a revision the caller hands over.Unconditionallywrites without any check.
Preconditions can be combined, and all of them must hold. If a precondition does not hold, nothing is written, and Execute returns an error of the category ErrConflict (see Handling Errors). If a command declares no preconditions, or combines Unconditionally with others, Execute returns an error of the category ErrPermanent before reading anything.
Guarding Against Concurrent Changes
If a command may only write events in case nothing has been written to its subject since Execute read the state, use the OnStateRead function. This fits most commands, since the decider decides on exactly that state:
func (c ReturnBook) Preconditions() []architecturekit.Precondition {
return []architecturekit.Precondition{
architecturekit.OnStateRead(),
}
}
Execute fills in the ID of the last event it has read. For a subject without any events, it requires the subject to still be pristine instead. This also holds with several processes writing to the same subjects, and with a state cache (see Caching States).
If something else was written in between, the command fails with an error of the category ErrConflict. Since the decider decided on the state that Execute read, deciding again on the new state is what would have happened had the command arrived a moment later. To have Execute do so, hand over the WithConflictRetries option when creating the store, with the number of times to decide again:
store := architecturekit.NewStore(client, "https://library.eventsourcingdb.io", architecturekit.WithConflictRetries(4))
Execute then reads the state anew, and the decider decides on it, up to four more times. Once the retries are used up, the command fails with ErrConflict, as without the option.
Note that this only applies to commands whose preconditions include OnStateRead. A command that checks only a revision the caller hands over is never decided again, since the caller has to learn about the conflict, and deciding again would fail the same way (see Checking the Revision of the Caller). A command that checks both is decided again, since the database does not say which precondition did not hold. If the revision of the caller is outdated, every attempt fails the same way, until the retries are used up. On the same subject, one of the two is enough: if the revision of the caller holds, so does OnStateRead.
Checking the Revision of the Caller
If a command may only write events in case its subject has not changed since the caller last read it, for example in a user interface, use the NewIsSubjectOnEventIDPrecondition function of the client SDK, and wrap it with the Require function. For that, add a field for the ID of the last event the caller has seen:
type BorrowBook struct {
BookID string
ReaderID string
BorrowedUntil string
ExpectedEventID string
}
func (c BorrowBook) Preconditions() []architecturekit.Precondition {
return []architecturekit.Precondition{
architecturekit.Require(eventsourcingdb.NewIsSubjectOnEventIDPrecondition(c.Subject(), c.ExpectedEventID)),
}
}
Note that the caller has to provide the event ID. A view can keep it for that purpose, as long as its items follow every event of their subject (see Defining Views).
Preventing Duplicates
If a command may only write events in case its subject does not yet have any events, use the NewIsSubjectPristinePrecondition function of the client SDK:
func (c AcquireBook) Preconditions() []architecturekit.Precondition {
return []architecturekit.Precondition{
architecturekit.Require(eventsourcingdb.NewIsSubjectPristinePrecondition(c.Subject())),
}
}
Requiring an Existing Subject
If a command may only write events in case its subject already has at least one event, use the NewIsSubjectPopulatedPrecondition function of the client SDK. Combine it with OnStateRead to also guard the state:
func (c ReturnBook) Preconditions() []architecturekit.Precondition {
return []architecturekit.Precondition{
architecturekit.Require(eventsourcingdb.NewIsSubjectPopulatedPrecondition(c.Subject())),
architecturekit.OnStateRead(),
}
}
Enforcing Rules Across Subjects
If a command may only write events depending on an EventQL query, use the NewIsEventQLQueryTruePrecondition function of the client SDK. For example, to acquire every ISBN only once, extend the preconditions of AcquireBook:
func (c AcquireBook) Preconditions() []architecturekit.Precondition {
return []architecturekit.Precondition{
architecturekit.Require(eventsourcingdb.NewIsSubjectPristinePrecondition(c.Subject())),
architecturekit.Require(eventsourcingdb.NewIsEventQLQueryTruePrecondition(fmt.Sprintf(
"FROM e IN events WHERE e.type == 'io.eventsourcingdb.library.book-acquired' AND e.data.isbn == '%s' PROJECT INTO COUNT() == 0",
c.ISBN,
))),
}
}
Note that the query must return a single row with a single value, which is interpreted as a boolean.
Writing Unconditionally
If a command may write its events whatever has been written to its subject in the meantime, for example because it only records a comment that does not depend on the state, use the Unconditionally function:
func (c CommentOnBook) Preconditions() []architecturekit.Precondition {
return []architecturekit.Precondition{
architecturekit.Unconditionally(),
}
}
Note that Unconditionally can not be combined with other preconditions.
Sharing Code Between Commands
The commands of one aggregate usually act on the same kind of subject, guarded by the same preconditions, and their deciders check the same things first. Write these once instead of for every command.
Sharing Subjects and Preconditions
Define Subject and Preconditions on a struct of their own, and embed it in every command. Go promotes the functions and fields of an embedded struct, so every command that embeds it is a Command:
type BookTarget struct {
BookID string
}
func (t BookTarget) Subject() string {
return "/books/" + t.BookID
}
func (t BookTarget) Preconditions() []architecturekit.Precondition {
return []architecturekit.Precondition{
architecturekit.OnStateRead(),
}
}
type AcquireBook struct {
BookTarget
Title string
Author string
ISBN string
}
type BorrowBook struct {
BookTarget
ReaderID string
BorrowedUntil string
}
type ReturnBook struct {
BookTarget
}
A command then names its book through the embedded struct, as in BorrowBook{BookTarget: BookTarget{BookID: "42"}, ReaderID: "23"}, and a decider reads it as cmd.BookID.
The command that creates a book can embed the struct, too, since OnStateRead requires a subject without any events to still be pristine (see Guarding Against Concurrent Changes). That is different if the struct checks the revision of the caller instead (see Checking the Revision of the Caller): a new book has no revision yet. Then the creating command does not embed the struct, and prevents duplicates itself:
type BookTarget struct {
BookID string
ExpectedEventID string
}
func (t BookTarget) Subject() string {
return "/books/" + t.BookID
}
func (t BookTarget) Preconditions() []architecturekit.Precondition {
return []architecturekit.Precondition{
architecturekit.Require(eventsourcingdb.NewIsSubjectOnEventIDPrecondition(t.Subject(), t.ExpectedEventID)),
}
}
type AcquireBook struct {
BookID string
Title string
Author string
ISBN string
}
func (c AcquireBook) Subject() string {
return "/books/" + c.BookID
}
func (c AcquireBook) Preconditions() []architecturekit.Precondition {
return []architecturekit.Precondition{
architecturekit.Require(eventsourcingdb.NewIsSubjectPristinePrecondition(c.Subject())),
}
}
Sharing Checks
Most deciders of an aggregate first check that it exists. Write that check once, as a function that returns a domain error, and call it at the beginning of every decider that needs it:
func requireAcquired(book Book, bookID string) error {
if !book.IsAcquired {
return architecturekit.NewDomainError("book %s does not exist", bookID)
}
return nil
}
var borrowBook = architecturekit.Decider[BorrowBook, Book]{
State: bookState,
Decide: func(ctx context.Context, cmd BorrowBook, book Book) ([]architecturekit.Event, error) {
if err := requireAcquired(book, cmd.BookID); err != nil {
return nil, err
}
if book.IsBorrowed {
return nil, architecturekit.NewDomainError("book %s is already borrowed", cmd.BookID)
}
return []architecturekit.Event{
BookBorrowed{
BorrowedBy: cmd.ReaderID,
BorrowedUntil: cmd.BorrowedUntil,
},
}, nil
},
}
Note that only the state knows what it means for an aggregate to exist, which is why the kit does not check it. The precondition that requires an existing subject protects the write (see Requiring an Existing Subject), but it fails with an error of the category ErrConflict, which is transient. A check in the decider answers a command on a book that does not exist with a domain error instead.
Using the Current Time
Some rules depend on the current time. For example, a reader may borrow a book for at most four weeks, so the decider has to know what day it is. Rather than calling time.Now in the decider, hand a clock to a function that creates the decider:
type Clock func() time.Time
func borrowBookDecider(now Clock) architecturekit.Decider[BorrowBook, Book] {
return architecturekit.Decider[BorrowBook, Book]{
State: bookState,
Decide: func(ctx context.Context, cmd BorrowBook, book Book) ([]architecturekit.Event, error) {
// ...
today := now()
if cmd.BorrowedUntil < today.Format(time.DateOnly) ||
cmd.BorrowedUntil > today.AddDate(0, 0, 28).Format(time.DateOnly) {
return nil, architecturekit.NewDomainError("book %s can be borrowed for at most four weeks", cmd.BookID)
}
return []architecturekit.Event{
BookBorrowed{
BorrowedBy: cmd.ReaderID,
BorrowedUntil: cmd.BorrowedUntil,
},
}, nil
},
}
}
The application hands over the real clock, and a test a fixed one, so that it does not have to compute around the actual date:
borrowBook := borrowBookDecider(time.Now)
borrowBookOnFirstOfOctober := borrowBookDecider(func() time.Time {
return time.Date(2026, time.October, 1, 0, 0, 0, 0, time.UTC)
})
The parameter shows which deciders depend on the time, and the clock can return what the domain thinks in, such as a date in a particular time zone instead of a moment.
Keeping the Time in Events
Only the decider reads the clock. Whatever it decides based on the time, such as the day a book was borrowed until, it writes into the event. Everything that comes after relies on the event, not on the clock:
Evolvenever reads the clock. It runs again every time the state is loaded, so a state that depends on the clock would be different tomorrow from what it is today, for the very same events.- A projection never reads the clock either. It runs again whenever the read model is rebuilt, which would turn yesterday's events into today's answers.
To know when an event was written, use the time that the database records for every event, which a projection receives in the Time field of the envelope. A question that depends on the moment it is asked, such as which books are overdue, belongs into the query, which gets the current time as a parameter (see Depending on More Than the Read Model).
Writing to Several Subjects
A command acts on a single subject, since it is a decision on the state of that subject. Some events are no such decision, for example a summary derived from many subjects, which has to be written together with a note on each of them, either all or none. To write events to several subjects at once, call the Write function with a context, the store, the events together with their subjects, and the preconditions:
written, err := architecturekit.Write(context.TODO(), store, []architecturekit.EventOn{
{Subject: "/inventories/2026", Event: InventoryTaken{BookCount: 2}},
{Subject: "/books/42", Event: BookCounted{Inventory: "2026"}},
{Subject: "/books/23", Event: BookCounted{Inventory: "2026"}},
}, architecturekit.Require(eventsourcingdb.NewIsSubjectPristinePrecondition("/inventories/2026")))
if errors.Is(err, architecturekit.ErrConflict) {
// Somebody else has taken the inventory already.
}
Write writes the events with the source of the store, and returns them as the database recorded them. If a precondition does not hold, nothing is written, and the error belongs to the category ErrConflict. Other failures belong to the same categories as for Execute (see Handling Errors).
Like a command, a write declares at least one precondition, made with Require, or Unconditionally to write without any. OnStateRead has nothing to guard, since Write reads no state.
Note that Write never decides again, since there is nothing to decide. If nothing is to be written, call it with no events, which writes nothing. It still needs preconditions, though, as any other write, and fails with an error of the category ErrPermanent without them, so declare them, for example with Unconditionally().
Note that Write knows no state, so unlike Execute, it does not refuse an event that a state has no rule for. Every state that reads one of the subjects needs a rule for the events written to it, since reading the subject fails otherwise.
Handling Errors
Every failure of architecturekit itself in reading and writing belongs to one of four categories. Use errors.Is to check for a category rather than for a concrete error:
ErrDomainmeans that a business rule rejected the command, as withNewDomainError.ErrConflictmeans that a precondition did not hold.ErrTransientmeans that trying again may help, for example if the database can not be reached.ErrPermanentmeans that trying again will not help, for example if an event could not be decoded, if it does not match the schema of its type, if a subject contains an event type the state has no rule for, or if a decider returns one.
A failure of the database is sorted by what its answer means, the same way for reading and for writing:
- If the database can not be reached, if the connection breaks, if the database asks to slow down (
429), or if it is unable to answer for now (5xx), for example because it is shutting down, the error belongs toErrTransient. - If the answer does not come from an EventSourcingDB, the error belongs to
ErrTransientas well, since a proxy in front of the database answers on its own while the database restarts. The message says so, so that a wrong address stands out in the log. - If the database rejects the API token (
401), or if it rejects the request itself, for example because it is malformed (400) or too large (413), the error belongs toErrPermanent. For a rejected API token, the message says so. - If a precondition did not hold (
409), the error belongs toErrConflict. If an event does not match its schema, which the database answers with the same status, it belongs toErrPermanent.
writtenEvents, err := architecturekit.Execute(
// ...
)
switch {
case errors.Is(err, architecturekit.ErrDomain):
// A business rule rejected the command.
case errors.Is(err, architecturekit.ErrConflict):
// A precondition did not hold.
case errors.Is(err, architecturekit.ErrTransient):
// Trying again may help.
case errors.Is(err, architecturekit.ErrPermanent):
// Trying again will not help.
}
Note that ErrConflict is a special case of ErrTransient, so check for it first.
An error that your own code returns, for example from a decider or a projection, passes through unchanged, so it belongs to a category only if you wrap it with one, as NewDomainError does. The httpapi and query packages have errors of their own, which StatusFor maps to status codes (see Mapping Errors to Status Codes and Getting a Single Item).
If the context ends, reading and writing stop, and the error is the one of the context, context.Canceled or context.DeadlineExceeded, which belongs to no category. Check for it with errors.Is as well. This is never a partial success: a read that the context cut short fails rather than handing out part of a state, and Execute writes nothing once the context has ended, also if it ends while the decider decides.
Note that ErrUnverified is a special case of ErrPermanent, which means that an event failed its verification (see Verifying Events). Since that may point to a security incident rather than a mistake, check for it before ErrPermanent if you want to treat it differently, for example to raise an alarm.
Note that Execute does not try again by itself, unless the store decides again on conflicts (see Guarding Against Concurrent Changes). To try again, for example after a transient failure, call Execute again.
Registering Event Schemas
The database only checks events against a schema once it is registered. The Evolve function collects the schemas of all events of a state, derived or their own (see Describing Events with Schemas). To get them as a slice of EventSchema, each with the fields EventType and Schema, call the Schemas function on the state. Then hand them over to the RegisterSchemas function, together with a context and the store:
err := architecturekit.RegisterSchemas(context.TODO(), store, bookState.Schemas())
if err != nil {
// ...
}
RegisterSchemas accepts the schemas of several states at once. Call it on every start, before the application serves requests: for an event type the database knows already, it checks that the registered schema is exactly the one from the code. If the context ends first, it returns the error of the context. Since the client registers a schema without a context, a registration that has begun is finished, but none begins once the context has ended.
A registered schema can not change. If it differs from the one from the code, RegisterSchemas returns an error of the category ErrPermanent, and so it does if the database refuses a schema, for example because stored events of the type do not match it. To change the shape of an event, introduce a new event type instead (see Versioning Events).
Note that this also holds if you remove the Schema function of an event whose schema is registered already: the derived schema has to be exactly the registered one, or RegisterSchemas fails. Keep the Schema function of such an event, unless you have compared both.
Once a schema is registered, the database rejects every event of the type that does not match it. Execute then returns an error of the category ErrPermanent, because writing the same event again gives the same result.
Versioning Events
The database keeps the schema of an event type forever. If the shape of an event changes, introduce a new event type, and translate the stored events of the old type with an upcaster.
Suppose an earlier version of the library wrote events of the type io.eventsourcingdb.library.book-lent, with the fields lentTo and until. To translate them into BookBorrowed events, create a set of upcasters with the NewUpcasters function. Then call the Upcast function on the set, and hand over the old event type and a function that receives the stored event and returns the translated events:
var libraryUpcasters = architecturekit.NewUpcasters().
Upcast(
"io.eventsourcingdb.library.book-lent",
func(event eventsourcingdb.Event) ([]eventsourcingdb.Event, error) {
var old struct {
LentTo string `json:"lentTo"`
Until string `json:"until"`
}
if err := json.Unmarshal(event.Data, &old); err != nil {
return nil, err
}
data, err := json.Marshal(BookBorrowed{
BorrowedBy: old.LentTo,
BorrowedUntil: old.Until,
})
if err != nil {
return nil, err
}
event.Type = BookBorrowed{}.EventType()
event.Data = data
return []eventsourcingdb.Event{event}, nil
},
)
The function has the type Upcaster. Upcasters may return more than one event. If a returned event has an upcaster of its own, that one runs as well, so every version needs only a single step to the next one. The translated events are never written back.
To use the upcasters, call the UpcastWith function on the state and hand over the set. The upcasters then run before the Evolve rules:
bookState.UpcastWith(libraryUpcasters)
Upcasting belongs to the event types, not to a single state, so register the upcasters once and hand the same set to every state, and to every projection that reads these events (see Defining Projections). That way, the write side and the read side see the same events.
Note that calling Upcast twice for the same event type panics, as does calling UpcastWith twice, or with nil.
Reading Long Streams
By default, Execute reads all events of a subject every time it runs a command. For a subject that collects many events over time, this gets slower with every event. If an event of one type carries everything the state needs from the events before it, the state can start from the latest event of that type instead.
Suppose every book is audited regularly, and every audit records the complete status of the book:
type BookAudited struct {
IsAcquired bool `json:"isAcquired"`
IsBorrowed bool `json:"isBorrowed"`
}
func (BookAudited) EventType() string {
return "io.eventsourcingdb.library.book-audited"
}
Add an Evolve rule that sets the state from the event alone, without looking at the state before it. Then call the FromLatest function on the state, with the event type as type parameter:
bookState.
Evolve(func(book Book, event BookAudited) Book {
return Book{
IsAcquired: event.IsAcquired,
IsBorrowed: event.IsBorrowed,
}
}).
FromLatest[BookAudited]()
Execute then reads the events of a subject from its latest BookAudited event onwards. If the subject has no such event yet, it reads all events, as before. Replay and ReplayStored start from the same event, so tests of a decider see exactly what Execute sees (see Testing Deciders).
Note that the state is only correct if the Evolve rule of that event type does not depend on the state before it. Events before the latest one of that type are never read.
Note that the database looks for the type under which an event is stored. If the event type is the result of an upcaster, events stored under the older type are not found, and all events are read.
Note that calling FromLatest for an event type without an Evolve rule, or calling it twice, panics.
Caching States
If no event carries the whole state, the store can keep the states of the most recently used subjects in memory instead, so that the next command on one of them reads only the events written since. To do so, hand over the WithStateCache option with the number of subjects to keep when creating the store:
store := architecturekit.NewStore(client, "https://library.eventsourcingdb.io", architecturekit.WithStateCache(10_000))
Once the cache is full, the least recently used subject makes room. Such a subject is read in full again the next time, or from the latest event of the type given to FromLatest, if there is one (see Reading Long Streams).
The cache only holds what was read, never what a command has written. Every command reads all events after the ones its state was built from, including those written by other processes, so the cache stays correct if several processes write to the same subjects.
The cache tells states apart by their type, not by the object. A state that is built anew for every command, for example by a function that returns it, is cached as well. This is why two different states that read the same subject need two different types: if one of them counts the loans of a book and the other one its returns, declare types such as LoanCount and ReturnCount rather than using int for both. If two states of the same type meet on the same subject, but differ in their initial value, in the event types they have Evolve rules or upcasters for, or in FromLatest, Execute and Load return an error of the category ErrPermanent (see Handling Errors).
Note that the cache can not compare the Evolve functions themselves. Two states of the same type that are built alike, but compute something else, are not told apart.
A cached state is handed to several commands, possibly at the same time. That is safe for a state that consists of values only, such as the Book state above. A state that holds slices, maps or pointers is only cached if it has a Clone function, which returns a copy that shares no data with the original:
type Shelf struct {
BookIDs []string
}
var shelfState = architecturekit.NewState(Shelf{}).
Evolve(func(shelf Shelf, event BookShelved) Shelf {
shelf.BookIDs = append(shelf.BookIDs, event.BookID)
return shelf
}).
Clone(func(shelf Shelf) Shelf {
return Shelf{BookIDs: slices.Clone(shelf.BookIDs)}
})
Without a Clone function, such a state is read as without a cache. The same function lets Step and StepStored leave a state unchanged (see Stepping Through States).
Note that a time.Time counts as a value, since its location never changes.
Note that values below 1 count as 1, and that calling Clone twice panics.
Loading States
To read the state of a single subject without executing a command, for example to answer a query about it, call the Load function with a context, the store, the state, and the subject:
book, err := architecturekit.Load(context.TODO(), store, bookState, "/books/42")
if err != nil {
// ...
}
Load reads the events exactly the way Execute does before it decides, including FromLatest and the state cache. For a query across many subjects, use a view instead (see Defining Views).
Reading Events
Some reads need the events themselves rather than a state, and fit neither Load nor a projection, for example one page of a long history, the events up to a certain one, or a single event. To read the events of a subject as they are stored, call the Read function with a context, the store, the subject, and the options of the client SDK. It returns an iterator over the events and errors:
options := eventsourcingdb.ReadEventsOptions{
LowerBound: &eventsourcingdb.Bound{ID: afterEventID, Type: eventsourcingdb.BoundTypeExclusive},
}
for event, err := range architecturekit.Read(context.TODO(), store, "/books/42", options) {
if err != nil {
// ...
}
// ...
}
Every event is verified, like everything else the store reads, before the loop sees it (see Verifying Events). A failure belongs to a category, as with Load (see Handling Errors), and ends the iteration. The store stops reading as soon as the loop ends, so breaking out of it after a page is fine.
The events come as they are stored, without upcasters or rules, since there is no state. To get at the data of an event, call the Decode function (see Executing Commands).
Stepping Through States
Sometimes the states before and after an event are needed, not only the latest one, for example for a history that tells what each event changed. To advance a state by a single stored event, call the StepStored function with the state, the state so far, and the event. It runs the upcasters and the Evolve rules exactly as reading from the database does, and returns the next state:
books := map[string]Book{}
projection := architecturekit.ProjectionFunc(func(ctx context.Context, event eventsourcingdb.Event) error {
before := books[event.Subject]
after, err := architecturekit.StepStored(bookState, before, event)
if err != nil {
return err
}
if !before.IsBorrowed && after.IsBorrowed {
// The book was borrowed by this event.
}
books[event.Subject] = after
return nil
})
For typed events, for example in a test, call the Step function instead.
The given state stays unchanged, so both states are at hand afterwards. For a state that holds slices, maps or pointers, that takes a Clone function (see Caching States). Without one, Step and StepStored return an error of the category ErrPermanent, rather than a next state that may share data with the one before. A state that consists of values only needs no Clone function.
Note that a stored event that the upcasters turn into several events is applied in full, and that FromLatest has no effect on a single step.
Verifying Events
EventSourcingDB gives every event a hash, which covers its metadata, its data, and the hash of the event before it. If the database runs with a signing key, it also signs every event it hands out. The store checks the hash of every event it reads, without being asked to.
To check the signatures as well, hand over the WithSignatureVerification option with the verification key of the database:
store := architecturekit.NewStore(client, "https://library.eventsourcingdb.io", architecturekit.WithSignatureVerification(verificationKey))
The verification key is an ed25519.PublicKey. To read it from a PEM file, use the pem and x509 packages of the standard library:
block, _ := pem.Decode(pemBytes)
if block == nil {
// ...
}
publicKey, err := x509.ParsePKIXPublicKey(block.Bytes)
if err != nil {
// ...
}
verificationKey, ok := publicKey.(ed25519.PublicKey)
if !ok {
// ...
}
The store checks every event it reads, for Execute, Load, and Read as well as for every kind of projection, and it does so before any upcaster sees the event. The events that Execute has just written are not checked, since they are not read. If an event fails its verification, reading fails with an error of the category ErrUnverified, which is a special case of ErrPermanent (see Handling Errors). A projection stops rather than skipping the event.
The two checks prove different things:
- A matching hash proves that an event is what was written. It proves no more than that, since whoever can change the stored data can compute a new hash as well.
- A matching signature proves that an event comes from a database that holds the signing key. The database signs events when handing them out, so the signature guards the way from the database to your application, but not the stored data itself.
The hashes are enough wherever reading is under your control. Check the signatures as well where events cross a trust boundary, for example when reading from a database that another organization runs. For details, see Verifying Event Signatures.
Note that checking a hash takes about a microsecond per event, so there is rarely a reason to turn it off. If there is one, hand over the WithoutHashVerification option, which can not be combined with WithSignatureVerification. Checking a signature, on the other hand, takes some tens of microseconds per event, which adds up when a projection catches up on millions of events.
Note that the database signs with the key it has at the moment, so after the signing key is rotated, the store needs the new verification key.
Note that neither check detects a history that has been rewritten as a whole. For that, audit the chain of hashes (see Auditing the Event Store).
Composing Subjects
So far, subjects have been composed by hand. To define their structure once, call the NewSubjectScheme function with a pattern, and use placeholders in braces for the variable parts:
var bookSubject = architecturekit.NewSubjectScheme("/books/{book}")
Each segment of a subject may only contain the characters that EventSourcingDB allows: the letters A-Z and a-z, the digits 0-9, underscores, and hyphens. This applies to the literal segments of the pattern as well as to the values that fill its placeholders.
The function returns a *SubjectScheme. To compose a subject, call the Build function with one value per placeholder, in the order in which they appear in the pattern. Use it in every command that acts on a book:
func (c AcquireBook) Subject() string {
return bookSubject.Build(c.BookID)
}
func (c BorrowBook) Subject() string {
return bookSubject.Build(c.BookID)
}
func (c ReturnBook) Subject() string {
return bookSubject.Build(c.BookID)
}
To take a subject apart, call the Match function. It returns the values by placeholder name, and false if the subject does not follow the pattern or has a value that Build would refuse:
values, ok := bookSubject.Match("/books/42")
// values["book"] == "42"
To get the pattern and the names of the placeholders, call the Pattern and the Placeholders function respectively.
To read or observe the events of all books, for example in a projection, start from the subject that all of them lie under. Call the Root function to get it from the scheme, rather than writing it down a second time. It returns the literal segments before the first placeholder, here /books, or / if the pattern starts with a placeholder:
run := architecturekit.StartProjection(ctx, store, architecturekit.SubjectTree(bookSubject.Root()), projection)
Note that other subjects may lie under the same root, such as /books/42/reviews/7 under /books. Use Match in the projection to tell them apart.
Note that a malformed pattern panics, including one with a literal segment that contains a character EventSourcingDB does not allow, as does calling Build with the wrong number of values, with an empty value, or with a value that contains such a character, for example a slash, a dot, or a space.
Values that come from outside, such as an ID in a request, may well be empty or contain such characters, and that is not a programming error. So always check them before building a subject: call the Check function with the same values as Build. It returns an error that says what is wrong, such as which characters a value may contain, instead of panicking:
if err := bookSubject.Check(bookID); err != nil {
// ...
}
Defining Views
A view holds the data that queries read. Define the shape of an item as a struct, and call the NewInMemoryView function with a function that returns the key of an item, to create a view that holds such items in memory:
type BookItem struct {
ID string
Title string
Author string
IsBorrowed bool
BorrowedUntil string
EventID string
}
func newCatalog() *architecturekit.InMemoryView[string, BookItem] {
return architecturekit.NewInMemoryView(
func(item BookItem) string { return item.ID },
architecturekit.RevisionIn(func(item *BookItem) *string { return &item.EventID }),
)
}
catalog := newCatalog()
An item carries no JSON annotations, since what a caller sees is decided by a query and its answer, not by the view (see Handling Queries over HTTP).
Every item has a revision of its own, which is the ID of the last event that changed it. The RevisionIn option makes the view keep it in a field of the item, so that a caller can hand it over to a command that uses the NewIsSubjectOnEventIDPrecondition function (see Checking the Revision of the Caller). The view sets the field whenever it changes an item, so you never set it yourself. Without the option, the view keeps the revisions to itself.
Note that calling RevisionIn with nil, or twice, panics.
The revision of an item fits such a precondition only if the item stands for exactly one subject, and the projection applies every event type of that subject to the item. The precondition checks the last event of the subject, so as soon as an event lands in the subject that the view does not apply to the item, the two drift apart, and every command with the revision of the item fails with an error of the category ErrConflict, until an event changes the item again. Apply an event type that does not change the item, such as one the state ignores, with a change that does nothing (see Defining Projections). For an item that gathers several subjects, such as all books a reader has borrowed, there is no single subject its revision could stand for, so use OnStateRead for the commands instead (see Guarding Against Concurrent Changes).
Every function that changes the view takes the ID of the event it applies. An event that is not newer than the item it is about is skipped, so applying the same event twice changes nothing. All functions take a context and return an error, which the view in memory hardly needs, but a view in a database would. So a view in a database can offer the same functions later on, without the projections that write to it having to change.
Note that the view as a whole has a revision as well, which is the last event it has seen at all, rather than the last one that changed a particular item (see Reading Your Own Writes).
Adding Items
To add an item, call the Insert function with a context, the ID of the event, and the item:
err := catalog.Insert(ctx, event.ID, BookItem{
ID: "42",
Title: "2001 – A Space Odyssey",
Author: "Arthur C. Clarke",
})
if err != nil {
// ...
}
If the key of the item is already taken, and the event is newer than the item with that key, Insert fails with an error of the category ErrPermanent, since two items with the same key point to a mistake in the events or in the key. To add an item or change the existing one, call the Upsert function instead, with the key and a function that changes the item. It starts from the existing item, or from an empty one if there is none, so it has to set the fields that make up the key:
err := catalog.Upsert(ctx, "42", event.ID, func(item *BookItem) {
item.ID = "42"
item.IsBorrowed = true
})
Note that an item whose key, after the change, differs from the given one fails with an error of the category ErrPermanent, which also catches a function that forgets to set the key.
Reading Items
To read the item with a given key, call the Get function. It returns false if there is none:
book, isFound, err := catalog.Get(ctx, "42")
if err != nil {
// ...
}
To read all items, call the All function. It returns an iterator over a copy of the items, in the order in which they were added, which you can use e.g. inside a for range loop:
items, err := catalog.All(ctx)
if err != nil {
// ...
}
for item := range items {
// ...
}
Note that the items you get share their slices, maps, and pointees with the view, so never change them (see Sharing Items with Readers).
Changing and Removing Items
To change the item with a given key, call the Update function with a function that changes it. To remove it, call the Delete function. Both return an Outcome, which tells what they did:
updated, err := catalog.Update(ctx, "42", event.ID, func(item *BookItem) {
item.IsBorrowed = true
})
deleted, err := catalog.Delete(ctx, "42", event.ID)
| Outcome | Meaning |
|---|---|
architecturekit.Applied |
The item was changed or removed. |
architecturekit.Missing |
There is no item with the key. |
architecturekit.AlreadyApplied |
The item has seen the event, or a newer one, so nothing was changed. |
Neither of the latter is an error, since both happen when events are applied a second time, as a later event may have removed the item already. If an event about an item that does not exist means that the view and the database disagree, say so in the projection:
outcome, err := catalog.Update(ctx, bookID, event.ID, func(item *BookItem) {
item.IsBorrowed = true
})
if err != nil {
return err
}
if outcome == architecturekit.Missing {
return fmt.Errorf("event %s is about book %s, which was never acquired", event.ID, bookID)
}
Note that changing the key of an item fails with an error of the category ErrPermanent.
To change or remove several items at once, call the UpdateWhere or the DeleteWhere function with a function that selects them. Both return the number of items they changed or removed:
isByClarke := func(item BookItem) bool {
return item.Author == "Arthur C. Clarke"
}
changed, err := catalog.UpdateWhere(ctx, isByClarke, event.ID, func(item *BookItem) {
item.Author = "Sir Arthur C. Clarke"
})
removed, err := catalog.DeleteWhere(ctx, isByClarke, event.ID)
Sharing Items with Readers
Get and All hand out plain copies of the items, which share their slices, maps, and pointees with the view. So an item is shared with every reader who got it, also while a projection changes it. That is why a reader must never change an item it got, and why a change replaces a field that holds a slice, a map, or a pointer, rather than writing into it. Writing into it would change what the readers hold, possibly while they are reading it, and writing into a map while somebody reads it may crash the process. Replacing the field with one built anew is safe:
type ReaderItem struct {
ID string
BookIDs []string
DueDates map[string]string
}
outcome, err := readers.Update(ctx, event.Data.BorrowedBy, event.ID, func(item *ReaderItem) {
item.BookIDs = append(slices.Clone(item.BookIDs), bookID)
})
Note that append(item.BookIDs, bookID) alone does not build the slice anew: if its array has room left, append writes into the array that the readers share.
To have the view take care of that, hand over the CloneWith option with a function that returns a copy of an item that shares no data with the original, like the Clone function of a state (see Caching States). The view then clones the item before every change, and hands the clone to the change, which may write into it freely:
readers := architecturekit.NewInMemoryView(
func(item ReaderItem) string { return item.ID },
architecturekit.CloneWith(func(item ReaderItem) ReaderItem {
item.BookIDs = slices.Clone(item.BookIDs)
item.DueDates = maps.Clone(item.DueDates)
return item
}),
)
outcome, err := readers.Update(ctx, event.Data.BorrowedBy, event.ID, func(item *ReaderItem) {
item.DueDates[bookID] = event.Data.BorrowedUntil
})
The view clones an item whenever it changes an existing one, with Update, Upsert, UpdateWhere, and the Update function of an index, but not when it adds one, since nobody has read a new item yet. Readers still never change an item they got, since the view hands out what it holds.
Note that an index whose value is taken from a slice, a map, or a pointee loses track of an item if a change writes into that data without a clone, since the old item then holds the new value as well (see Indexing Items).
Note that a time.Time counts as a value, since its location never changes, so an item whose only pointer is the one inside a time.Time needs no CloneWith.
Note that calling CloneWith with nil, or twice, panics.
Indexing Items
Selecting items with a function reads every item. To find items by a value directly, for example all books by an author, add a secondary index with the Index function. It takes a function that returns the value of an item, and returns an index, which offers the Lookup, Update, and Delete functions:
byAuthor := catalog.Index(func(item BookItem) string { return item.Author })
books, err := byAuthor.Lookup(ctx, "Arthur C. Clarke")
changed, err := byAuthor.Update(ctx, "Arthur C. Clarke", event.ID, func(item *BookItem) {
item.IsBorrowed = false
})
removed, err := byAuthor.Delete(ctx, "Arthur C. Clarke", event.ID)
Several items may share a value. The index follows every change to the view, also when the value of an item changes, and hands out items in the order in which they were added.
Note that adding an index reads every item, so add indexes before the view is used.
To keep items somewhere else, for example in a database, implement the View interface, which consists of the All function:
type BookTable struct {
// ...
}
func (t *BookTable) All(ctx context.Context) (iter.Seq[BookItem], error) {
// ...
}
If the view can find a single item by its key, also implement the Get function, which makes it a KeyedView:
func (t *BookTable) Get(ctx context.Context, id string) (BookItem, bool, error) {
// ...
}
A query that reads a single item can then take an architecturekit.KeyedView[string, BookItem] instead of running over all items, and work with the view in memory and the one in a database alike. InMemoryView is a KeyedView, too.
Defining Projections
A projection turns events into a view. Call the NewProjection function, and call the On function for every event type the view depends on. Each handler receives an Envelope, which holds the metadata of the event, such as its ID, Time, and Subject, and its data, decoded into the Go type of the event:
func newCatalogProjection(catalog *architecturekit.InMemoryView[string, BookItem]) *architecturekit.TypedProjection {
return architecturekit.NewProjection().
On(func(ctx context.Context, event architecturekit.Envelope[BookAcquired]) error {
return catalog.Insert(ctx, event.ID, BookItem{
ID: bookIDOf(event.Subject),
Title: event.Data.Title,
Author: event.Data.Author,
})
}).
On(func(ctx context.Context, event architecturekit.Envelope[BookBorrowed]) error {
_, err := catalog.Update(ctx, bookIDOf(event.Subject), event.ID, func(item *BookItem) {
item.IsBorrowed = true
item.BorrowedUntil = event.Data.BorrowedUntil
})
return err
}).
On(func(ctx context.Context, event architecturekit.Envelope[BookReturned]) error {
_, err := catalog.Update(ctx, bookIDOf(event.Subject), event.ID, func(item *BookItem) {
item.IsBorrowed = false
item.BorrowedUntil = ""
})
return err
})
}
catalogProjection := newCatalogProjection(catalog)
The event type is taken from the event's EventType function, so it does not have to be repeated, and the type of an event and the type its data is decoded into can not drift apart.
Which parts of the subject a view needs is up to the application. In this example, a small function takes the ID of the book out of the subject (see Composing Subjects):
func bookIDOf(subject string) string {
values, _ := bookSubject.Match(subject)
return values["book"]
}
NewProjection returns a *TypedProjection, which is a Projection like any other, so you can run, track, and test it as described below. Events without a handler are skipped, since a projection usually reads more events than it depends on. If the data of an event can not be decoded, the projection returns an error of the category ErrPermanent. An error returned by a handler is passed on unchanged.
If the view keeps the revisions of its items for a precondition, a skipped event makes the revision of its item fall behind the subject (see Defining Views). In that case, give every event type of the subject a handler, also one that does not change the item, such as BookInspected, which the state ignores. Its handler calls Update with a change that does nothing, which only moves the revision of the item on:
func newCatalogProjection(catalog *architecturekit.InMemoryView[string, BookItem]) *architecturekit.TypedProjection {
return architecturekit.NewProjection().
// ...
On(func(ctx context.Context, event architecturekit.Envelope[BookInspected]) error {
_, err := catalog.Update(ctx, bookIDOf(event.Subject), event.ID, func(*BookItem) {})
return err
})
}
To have the projection see the same events as the state, hand over the same set of upcasters with the UpcastWith function (see Versioning Events):
catalogProjection.UpcastWith(libraryUpcasters)
Note that calling On twice for the same event type panics.
Handling Every Event
A projection that has to see every event, for example to log it, implements the Projection interface directly. Its Apply function receives every event as it is stored, without running any upcasters:
type LogProjection struct{}
func (LogProjection) Apply(ctx context.Context, event eventsourcingdb.Event) error {
log.Println(event.Subject, event.Type)
return nil
}
For a projection that needs no type of its own, use ProjectionFunc, which turns a function into a projection:
logProjection := architecturekit.ProjectionFunc(func(ctx context.Context, event eventsourcingdb.Event) error {
log.Println(event.Subject, event.Type)
return nil
})
Running Projections
To run a projection, call the StartProjection function with a context, the store, the subjects to read, and the projection. Say which subjects with the SubjectTree function, which stands for the given subject together with every subject below it, or with the ExactSubject function, which stands for the given subject alone. A projection usually reads a tree, such as every book below /books. There is no default, so that every projection says which one it means, since the client SDK reads a single subject unless told otherwise.
The function runs the projection in the background and returns a *ProjectionRun at once. The run first applies all events that are already stored, then observes new events until the context is canceled. An application usually answers queries only once its views have caught up, since a half-built view answers wrongly rather than slowly, so wait for that:
ctx, cancel := context.WithCancel(context.TODO())
defer cancel()
run := architecturekit.StartProjection(ctx, store, architecturekit.SubjectTree("/books"), catalogProjection,
architecturekit.Named("catalog"),
)
select {
case <-run.CaughtUp():
// The view holds every event that was stored when the run started.
case <-run.Done():
// The run ended before it caught up.
return run.Err()
}
// Start to answer queries.
The options after the projection are optional. Named gives the projection a name, by which the observer of reconnects and the health checks report it (see Checking Health over HTTP). The Name function of the run returns it.
CaughtUp returns a channel that is closed once the run has applied the events that were stored when it started. It is closed only once, and stays closed while the run reconnects later on. Done returns a channel that is closed once the run has ended, which happens when the context ends, or on a failure that trying again will not fix. Err returns why the run has ended. It returns nil as long as the run has not ended, and if it ended because its context did, since canceling the context is how a projection is stopped. If Apply returns an error that trying again will not fix, the run ends, and Err returns it.
To wait until the run ends, as a process does that runs nothing else, wait for Done:
<-run.Done()
return run.Err()
Note that if the database can not be reached at the start, the run keeps trying, and CaughtUp stays open. To wait for a limited time only, add a case with time.After to the select statement.
If reading fails with an error of the category ErrTransient, or if the database ends the stream, for example because it restarts, the run waits and continues after the last event it has applied, until the context is canceled. The delay starts at one second, doubles with every attempt in a row, and never exceeds one minute. It starts over once the projection has applied an event again, or has followed the stream for longer than the delay had grown to, even if no event arrived. That way, a load balancer that ends long-lived connections regularly does not hold back a quiet projection, while a database that fails before the projection has caught up, or right after, is given ever more time.
To use other delays, or to learn about every attempt, for example to log it, hand over the WithReconnectDelays and WithReconnectObserver options when creating the store:
store := architecturekit.NewStore(client, "https://library.eventsourcingdb.io",
architecturekit.WithReconnectDelays(500*time.Millisecond, 30*time.Second),
architecturekit.WithReconnectObserver(func(reconnect architecturekit.Reconnect) {
log.Println("observing again", reconnect.Projection, reconnect.Attempt, reconnect.Err, reconnect.Delay)
}),
)
Note that an initial delay of zero or less panics, since the projections would then read again without any pause, and so does a maximum delay below the initial one.
The observer receives a Reconnect with these fields:
Projectionis the name the projection was given withNamed, or empty if it has none.Subjectis the subject the projection reads.Erris the reason, which isnilif the database ended the stream.Delayis how long the projection waits before the next attempt.Attemptcounts the attempts in a row, starting at one. It starts over together with the delay once the projection has applied an event, or has followed the stream for longer than the delay had grown to.
Note that a database that can not be reached is retried as well, since that is usually transient. The observer is how to notice a database that stays unreachable. A failure that trying again will not fix, for example a rejected API token, ends the run (see Handling Errors).
To find out where a run stands, call the Status function. It returns a ProjectionStatus with these fields:
PhaseisPhaseCatchingUp,PhaseLive,PhaseReconnecting, orPhaseStopped.Sinceis when the phase began. ForPhaseReconnecting, that is when the disruption began, not when the latest attempt did.Erris why the run is reconnecting or has stopped. It isnilif the database ended the stream, or if the run stopped because its context ended.Attemptscounts the attempts to read again within the current disruption. UnlikeAttemptofReconnect, it starts over whenever the run has caught up, even if the stream ends again right after.Revisionis the ID of the last event the run has applied and committed.HasCaughtUptells whether the run has caught up at least once. LikeCaughtUp, it staystruewhile the run reconnects later on.
status := run.Status()
if status.Phase == architecturekit.PhaseReconnecting && time.Since(status.Since) > 5*time.Minute {
// The view has not been up to date for more than five minutes.
}
To only apply the events that are already stored, for example for a batch job or in a test, call the CatchUpProjection function instead. It takes the same arguments and returns once all stored events have been applied. If the context ends before that, it returns the error of the context, so that a read model that is only partly built does not look complete:
err := architecturekit.CatchUpProjection(context.TODO(), store, architecturekit.SubjectTree("/books"), catalogProjection)
if err != nil {
// ...
}
For a transactional projection, call the StartTransactionalProjection and the CatchUpTransactionalProjection function instead (see Resuming Projections).
Resuming Projections
By default, a projection starts from the first event every time it runs, which fits a view held in memory. For a view that keeps its data, the projection can resume where it stopped instead.
If the view can store a checkpoint, but not together with the data, additionally implement the Resumable interface. Checkpoint returns the ID of the last event saved, or an empty string if there is none, and SaveCheckpoint saves it:
type BookTableProjection struct {
// ...
}
func (p *BookTableProjection) Apply(ctx context.Context, event eventsourcingdb.Event) error {
// ...
}
func (p *BookTableProjection) Checkpoint(ctx context.Context) (string, error) {
// ...
}
func (p *BookTableProjection) SaveCheckpoint(ctx context.Context, eventID string) error {
// ...
}
To make a projection created with NewProjection resumable, embed it in a type of your own, and add the two functions there:
type BookTable struct {
*architecturekit.TypedProjection
// ...
}
func (t *BookTable) Checkpoint(ctx context.Context) (string, error) {
// ...
}
func (t *BookTable) SaveCheckpoint(ctx context.Context, eventID string) error {
// ...
}
Note that the checkpoint is saved after the events have been applied. After a crash, events may therefore be applied a second time, so Apply must be idempotent.
To find out how a projection will be run, call the ModeOf function. It returns a Mode, which is ModeRebuild or ModeResumable:
mode := architecturekit.ModeOf(catalogProjection)
// architecturekit.ModeRebuild
Note that the mode depends on which interfaces a projection implements. If a function's signature does not match, the projection silently runs in ModeRebuild. To catch that, check the mode in a test (see Testing Projections).
If the view can store the data and the checkpoint together, as a relational database can, implement the Transactional interface instead of Projection. A transactional projection applies events only within a transaction, so it has no Apply function of its own. Begin starts a transaction and returns a Tx, which applies the events, and commits them together with the ID of the last event, or rolls them back:
type TransactionalBookTableProjection struct {
// ...
}
func (p *TransactionalBookTableProjection) Checkpoint(ctx context.Context) (string, error) {
// ...
}
func (p *TransactionalBookTableProjection) Begin(ctx context.Context) (architecturekit.Tx, error) {
// ...
}
type bookTableTx struct {
// ...
}
func (tx *bookTableTx) Apply(ctx context.Context, event eventsourcingdb.Event) error {
// ...
}
func (tx *bookTableTx) Commit(ctx context.Context, lastEventID string) error {
// ...
}
func (tx *bookTableTx) Rollback(ctx context.Context) error {
// ...
}
Instead of implementing Apply on the Tx yourself, you can use handlers created with NewProjection. Build them in Begin, so that they write into the transaction that has just been started, and embed them in the Tx, which then only needs Commit and Rollback:
func (p *TransactionalBookTableProjection) Begin(ctx context.Context) (architecturekit.Tx, error) {
tx, err := p.db.BeginTx(ctx, nil)
if err != nil {
return nil, err
}
return &bookTableTx{
tx: tx,
TypedProjection: architecturekit.NewProjection().
On(func(ctx context.Context, event architecturekit.Envelope[BookAcquired]) error {
_, err := tx.ExecContext(ctx, "INSERT INTO books ...")
return err
}),
}, nil
}
type bookTableTx struct {
*architecturekit.TypedProjection
tx *sql.Tx
}
func (tx *bookTableTx) Commit(ctx context.Context, lastEventID string) error {
_, err := tx.tx.ExecContext(ctx, "UPDATE checkpoint ...", lastEventID)
if err != nil {
return errors.Join(err, tx.tx.Rollback())
}
return tx.tx.Commit()
}
func (tx *bookTableTx) Rollback(ctx context.Context) error {
return tx.tx.Rollback()
}
Note that Rollback is not called after Commit fails, so Commit has to roll back itself if it can not finish, as above.
To run a transactional projection, call the StartTransactionalProjection or the CatchUpTransactionalProjection function instead of StartProjection or CatchUpProjection. They take the same arguments:
run := architecturekit.StartTransactionalProjection(ctx, store, architecturekit.SubjectTree("/books"), &TransactionalBookTableProjection{},
architecturekit.Named("book-table"),
)
Note that StartProjection, CatchUpProjection, and Tracking panic for a projection that implements Transactional in addition to Apply, since calling Apply would bypass the transactions.
Batching Events
By default, the checkpoint is saved, or the transaction is committed, after every event. While a projection catches up, for example when it starts for the first time, that takes a lot longer than needed. To apply several events at once while catching up, implement the Batched interface on a resumable or transactional projection, and return how many:
func (p *BookTableProjection) CatchUpBatchSize() int {
return 1000
}
Values below 1 count as 1. For a transactional projection, a batch is one transaction, so a failure rolls back all of its events, and the next attempt starts after the last commit.
Once the projection has caught up, it saves the checkpoint, or commits, after every event again. Events then arrive one at a time, and a batch that waited to fill up would hold back the ones that have already arrived.
Note that a resumable projection may apply up to that many events a second time after a crash.
Publishing Events
Other systems often need to learn about events, for example to send an email once a book has been borrowed, to call a webhook, to fill a search index, or to tell another team. How they learn about them depends on whether they can read the database themselves.
If the other system can read the EventSourcingDB, let it observe the events itself. It then gets every event in the order it was stored, and after a restart, it resumes from the last event it has seen. Nothing needs to be forwarded (see Observing Events).
If it can not, because it is a mail server, a webhook, or a message broker, or if it should not depend on how the events of your application look, forward the events with a projection that publishes them instead of writing a view. Do not publish from where a command is executed: if the application stops right after Execute, the event is stored, but never published, and nothing tells.
A projection that publishes needs to be resumable (see Resuming Projections), and it reports a failure of the other system as an error of the category ErrTransient:
type LoanMailer struct {
*architecturekit.TypedProjection
checkpoints CheckpointStore
}
func NewLoanMailer(mailer Mailer, checkpoints CheckpointStore) *LoanMailer {
return &LoanMailer{
TypedProjection: architecturekit.NewProjection().
On(func(ctx context.Context, event architecturekit.Envelope[BookBorrowed]) error {
// The ID of the event lets the receiver recognize a mail it got before.
err := mailer.Send(ctx, event.ID, event.Data.BorrowedBy, "You have borrowed a book.")
if err != nil {
return fmt.Errorf("%w: sending a mail: %v", architecturekit.ErrTransient, err)
}
return nil
}),
checkpoints: checkpoints,
}
}
func (m *LoanMailer) Checkpoint(ctx context.Context) (string, error) {
return m.checkpoints.Load(ctx, "loan-mailer")
}
func (m *LoanMailer) SaveCheckpoint(ctx context.Context, eventID string) error {
return m.checkpoints.Save(ctx, "loan-mailer", eventID)
}
Here, Mailer and CheckpointStore stand for whatever your application uses to send mails and to keep a value. Start the projection as any other (see Running Projections):
run := architecturekit.StartProjection(ctx, store, architecturekit.SubjectTree("/books"), NewLoanMailer(mailer, checkpoints),
architecturekit.Named("loan-mailer"),
)
Since publishing rides on a projection, it behaves like one:
- It resumes only with a checkpoint. A projection that is not resumable starts from the first event whenever the application starts, which suits a view held in memory. A publisher would send every event again on every start. Within a run, the position is kept either way, so reconnecting repeats nothing.
- An event may be published twice. The checkpoint is saved after an event has been applied, so an event published right before the application stopped is published again after the restart. Hand over the ID of the event, so that the receiver can recognize an event it got before.
- A transient failure is tried again. For an error of the category
ErrTransient, the run tries the failed event again, with a growing delay. Any other error ends the run. - Events arrive in order, one at a time, as they were stored.
Note that the mode of a projection depends on the functions it implements. To make sure that a publisher is resumable, check its mode in a test (see Testing Projections).
Defining Queries
A query describes what someone wants to know. Define it as a struct, and answer it with a function that reads a view. To turn the items into a slice, use slices.Collect:
type ListBooks struct {
OnlyAvailable bool
Limit int
}
func listBooks(catalog architecturekit.View[BookItem]) func(context.Context, ListBooks) ([]BookItem, error) {
return func(ctx context.Context, q ListBooks) ([]BookItem, error) {
items, err := catalog.All(ctx)
if err != nil {
return nil, err
}
// ...
return slices.Collect(items), nil
}
}
To filter, order, page, and transform the items, use the query package:
import "github.com/thenativeweb/architecturekit-golang/architecturekit/query"
All of its functions take an iterator, and those that return items return an iterator again, so they can be combined without collecting anything in between.
Filtering and Transforming Items
To keep only some items, call the Where function with a function that selects them:
if q.OnlyAvailable {
items = query.Where(items, func(item BookItem) bool {
return !item.IsBorrowed
})
}
To turn every item into something else, call the Select function:
titles := query.Select(items, func(item BookItem) string {
return item.Title
})
Ordering Items
To order items by a key, call the OrderBy function, or the OrderByDescending function for the reverse order:
items = query.OrderBy(items, func(item BookItem) string {
return item.Title
})
To order items by a comparison function, for example by several fields, call the OrderByFunc function. It expects the same kind of function as slices.SortFunc, so cmp.Or and strings.Compare work with it:
items = query.OrderByFunc(items, func(left, right BookItem) int {
return cmp.Or(
strings.Compare(left.Author, right.Author),
strings.Compare(left.Title, right.Title),
)
})
Note that ordering reads all items, and that it is stable, so items that compare as equal keep their order.
Paging Items
To skip a number of items, call the Skip function. To stop after a number of items, call the Take function:
if q.Limit > 0 {
items = query.Take(items, q.Limit)
}
Both can be combined, for example to get the third page of ten items:
items = query.Take(query.Skip(items, 20), 10)
Getting a Single Item
To get the first item, call the First function. It returns false if there is none:
first, ok := query.First(items)
To get the only item, call the Single function. It returns query.ErrNoItems if there is none, and query.ErrTooManyItems if there are several:
type GetBook struct {
BookID string
}
func getBook(catalog architecturekit.View[BookItem]) func(context.Context, GetBook) (BookItem, error) {
return func(ctx context.Context, q GetBook) (BookItem, error) {
items, err := catalog.All(ctx)
if err != nil {
return BookItem{}, err
}
return query.Single(query.Where(items, func(item BookItem) bool {
return item.ID == q.BookID
}))
}
}
Note that the query package reads every item it is handed. To get an item by its key, call the Get function of a KeyedView instead, and to get items by the value of a secondary index, call the Lookup function of the index (see Defining Views).
Counting Items
To count items, call the Count function. To check whether at least one item matches, call the Any function, which stops at the first match:
count := query.Count(items)
hasBorrowedBooks := query.Any(items, func(item BookItem) bool {
return item.IsBorrowed
})
Reading Your Own Writes
A view lags behind the events that have been written, by however long its projection takes. To read your own writes, wait until the view has seen the events you have written.
The ID of the last event a view has seen is its revision. Since the database assigns event IDs in ascending order across all subjects, revisions can be compared.
Tracking Revisions
To track the revision of a view, wrap the projection with the Tracking function and hand over the view. It records every event that reaches the projection, including the ones the projection ignores:
trackedProjection := architecturekit.Tracking(catalogProjection, catalog)
If the projection writes to several views, hand over all of them, so that each one knows how far it has come:
trackedProjection := architecturekit.Tracking(libraryProjection, catalog, readers, loans)
Then run trackedProjection instead of catalogProjection (see Running Projections).
Tracking accepts every view that implements the RevisionSink interface, which consists of the Seen function. InMemoryView implements it.
Note that calling Tracking without any view panics.
The tracked projection keeps the mode and the batch size of the projection it wraps. A transactional projection can not be tracked, since it has no Apply function. Record its revision within the transaction instead.
To get the revision your own write has produced, call the RevisionOf function with the written events. It returns the highest event ID, or an empty string if no events were written:
revision := architecturekit.RevisionOf(writtenEvents)
Waiting for Revisions
To wait until a view has reached a revision, call the WaitFor function with a context and the revision. It returns immediately if the view has already reached the revision, and otherwise once it does, or when the context ends:
ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
defer cancel()
err := catalog.WaitFor(ctx, revision)
if err != nil {
// ...
}
To get the current revision of a view, call the Revision function. It returns an empty string as long as the view has not seen any event:
current := catalog.Revision()
Both functions form the Revisioned interface, which InMemoryView implements. To wait for revisions of a view of your own, implement it as well.
Comparing Revisions
To compare two revisions, call the CompareRevisions function. Like cmp.Compare, it returns -1, 0, or 1. An empty revision comes before every other one. If a value is not a revision, it returns ErrNotARevision:
result, err := architecturekit.CompareRevisions("9", "10")
// result == -1
Setting Up an HTTP API
To expose commands and queries over HTTP, use the httpapi package:
import "github.com/thenativeweb/architecturekit-golang/architecturekit/httpapi"
Define a type that describes who is making a request, and a function that determines it from the request. Then call the NewAPI function with the store and this function, and create a mux:
type User struct {
ID string
IsLibrarian bool
}
func userFrom(r *http.Request) (User, error) {
// ...
}
api := httpapi.NewAPI(store, userFrom)
mux := http.NewServeMux()
If the function returns an error, neither a command nor a query is run, and the request is answered with 401 Unauthorized. An error that has a status code of its own keeps it, though (see Mapping Errors to Status Codes), and so does an error of the category ErrPermanent, which is answered with 500 Internal Server Error. So if the function can not determine the user because the session store is down, for example, it returns an error of the category ErrTransient. The request is then answered with 503 Service Unavailable, and the failure is logged, rather than sending the caller off to sign in again.
Note that to answer an error that has a status code of its own with 401 Unauthorized all the same, the function wraps it with httpapi.ErrUnauthorized itself, for example with fmt.Errorf("%w: %v", httpapi.ErrUnauthorized, err).
For an application without authentication, call the NewPublicAPI function instead. Commands and queries then receive httpapi.NoUser as user:
api := httpapi.NewPublicAPI(store)
Everything that answers through an API logs every error it does not explain to the caller in full, once, with the method and the route of the request: the routes it wires up, and the functions that answer in a handler of your own. A failure of the server is logged at level Error, and a refusal with 401 or 409, whose details the caller is not told, at level Info (see Handling Commands over HTTP). By default, they use the default logger of log/slog. To use the logger of your application instead, hand over the WithLogger option, which NewPublicAPI accepts as well:
api := httpapi.NewAPI(store, userFrom, httpapi.WithLogger(logger))
Determining the User
To determine the user in a handler of your own, call the UserOf function. If the user cannot be determined, it returns an error that wraps httpapi.ErrUnauthorized, unless the error of the function has a status code of its own, which it then returns as it is (see Setting Up an HTTP API):
mux.HandleFunc("GET /api/me", func(w http.ResponseWriter, r *http.Request) {
user, err := httpapi.UserOf(r, api)
if err != nil {
// ...
}
// ...
})
Handling Commands over HTTP
To accept a command over HTTP, define a request type with JSON annotations, and implement the ToCommand function, which receives the user and returns the command. For example, for the BorrowBook that checks the revision of the caller (see Checking the Revision of the Caller):
type borrowBookRequest struct {
BookID string `json:"bookId"`
BorrowedUntil string `json:"borrowedUntil"`
ExpectedEventID string `json:"expectedEventId"`
}
func (r borrowBookRequest) ToCommand(user User) (BorrowBook, error) {
if err := bookSubject.Check(r.BookID); err != nil {
return BorrowBook{}, err
}
if _, err := time.Parse(time.DateOnly, r.BorrowedUntil); err != nil {
return BorrowBook{}, errors.New("borrowedUntil must be a date")
}
return BorrowBook{
BookID: r.BookID,
ReaderID: user.ID,
BorrowedUntil: r.BorrowedUntil,
ExpectedEventID: r.ExpectedEventID,
}, nil
}
ToCommand is the place to validate a request, since an error it returns is answered with 400 Bad Request. Check at least what would otherwise fail later: the ID of the book becomes part of a subject, and Build panics on an empty ID or one with a character that a subject may not contain, such as a slash or a dot (see Composing Subjects). And a value that does not match the schema of its event is refused by the database, which is a permanent failure answered with 500 Internal Server Error – although it is the caller's mistake.
Then call the Route function with the request type, the API, the mux, a pattern, and the decider:
httpapi.Route[borrowBookRequest](api, mux, "POST /api/borrow-book", borrowBook)
The route decodes the request body, builds the command, and executes it:
curl -X POST http://localhost:8080/api/borrow-book \
-H "Content-Type: application/json" \
-d '{"bookId":"42","borrowedUntil":"2026-10-24","expectedEventId":"0"}'
If this succeeds, it answers with 200 OK and the revision it has written, which is the ID of the last written event:
{ "revision": "1" }
The revision is empty if the command did not write anything. A caller hands it to a query to read its own writes (see Reading Your Own Writes over HTTP).
Otherwise, it answers with the status code that matches the error (see Mapping Errors to Status Codes) and a message:
{ "message": "book 42 is already borrowed" }
The message is the error message if the error is written for the caller, such as a broken business rule or what is wrong with the request. Other errors may name internals, such as the key a token failed to verify with, or the subject a precondition guarded, so their message is fixed:
| Status code | Message |
|---|---|
401 Unauthorized |
unauthorized |
409 Conflict |
conflict: the data has changed since it was read |
500 and above |
internal server error |
The actual error is logged, so that it does not vanish (see Setting Up an HTTP API): at level Info for 401 and 409, since the server did not fail, and at level Error for 500 and above.
To answer this way in a handler of your own, call the Respond function with the response writer, the request, the API, the written events, and the error. Like the route, it logs through the logger of the API, with the route of the request.
Adding to the Answer
To answer with more than the revision, for example with the ID of a new book that ToCommand has generated, hand over the Adding option. It takes a function that receives a Handled value with the command and the written events, and returns the fields to add, usually as a struct with JSON annotations, and an error:
type acquireBookRequest struct {
Title string `json:"title"`
Author string `json:"author"`
ISBN string `json:"isbn"`
}
func (r acquireBookRequest) ToCommand(user User) (AcquireBook, error) {
return AcquireBook{
BookID: rand.Text(),
Title: r.Title,
Author: r.Author,
ISBN: r.ISBN,
}, nil
}
httpapi.Route[acquireBookRequest](api, mux, "POST /api/acquire-book", acquireBook,
httpapi.Adding(func(handled httpapi.Handled[AcquireBook]) (any, error) {
return struct {
ID string `json:"id"`
}{handled.Command.BookID}, nil
}))
The route then answers with both:
{ "id": "…", "revision": "1" }
The function is only called if the command has succeeded. If it returns an error, the events are written all the same, so the route still answers with 200 OK and the revision, which the caller needs to read its own writes, and must not take for a reason to send the command again. The answer then holds whatever fields the function returned along with the error, or none, and the error is logged through the logger of the API, with the route.
The kit adds the revision itself, so the fields must not contain one, and they must encode to a JSON object. Otherwise, the route answers with 500 Internal Server Error and logs why, although the events have been written, since that is a mistake in the code rather than something that happens at runtime.
Note that the written events are available in Handled as well. Add them only deliberately: they are the inner model of the application, every caller that reads them depends on their shape, and they may contain data that is not meant for the caller.
Answering Commands in Your Own Format
To answer in a format of your own, for example with another status code, call the Handle function in a handler of your own. It does the same as a route, but writes nothing to the response. Instead, it returns a Handled value with the command it has built and the written events:
mux.HandleFunc("POST /api/acquire-book", func(w http.ResponseWriter, r *http.Request) {
handled, err := httpapi.Handle[acquireBookRequest](r, api, acquireBook)
if err != nil {
httpapi.Respond(w, r, api, nil, err)
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusCreated)
json.NewEncoder(w).Encode(map[string]string{
"id": handled.Command.BookID,
})
})
Note that Handled contains the command even if executing it fails.
Authorizing Commands
To refuse a command, return httpapi.ErrForbidden from ToCommand. The request is then answered with 403 Forbidden, and the command is not executed:
func (r acquireBookRequest) ToCommand(user User) (AcquireBook, error) {
if !user.IsLibrarian {
return AcquireBook{}, httpapi.ErrForbidden
}
return AcquireBook{
BookID: rand.Text(),
Title: r.Title,
Author: r.Author,
ISBN: r.ISBN,
}, nil
}
The same applies to the other errors of the httpapi package, such as httpapi.ErrNotFound, and to errors of the category ErrDomain: they keep their status code. Any other error returned from ToCommand is answered with 400 Bad Request.
Validating Requests
Before a request reaches ToCommand, it is validated:
- The
Content-Typeheader must beapplication/json, otherwise the request is answered with415 Unsupported Media Type, and the error ishttpapi.ErrUnsupportedMediaType. - The body must not be larger than
httpapi.MaxRequestBody, which is one mebibyte, otherwise the request is answered with413 Request Entity Too Large, and the error ishttpapi.ErrTooLarge. - The body must be a single valid JSON value without unknown fields, otherwise the request is answered with
400 Bad Request, and the error ishttpapi.ErrMalformed. Nothing but whitespace may follow the value, and no name may occur twice in an object. Names match fields regardless of case, as withencoding/json, so two names that match the same field count as the same name, even if they differ in case.
Note that parsers disagree on what a name that occurs twice means, and on data after the value: one takes the first value, another the last, and one stops after the value, while another reads on. A filter or a proxy in front of the application might then check another value than the one the application uses, which is why both are refused.
To read a body by the same rules elsewhere, call the BodyOf function (see Reading Queries from the Body).
Handling Queries over HTTP
To answer a query over HTTP, define a function that receives the request and the user, and returns the query:
toListBooks := func(r *http.Request, user User) (ListBooks, error) {
return ListBooks{
OnlyAvailable: r.URL.Query().Get("available") == "true",
}, nil
}
Items carry no JSON annotations (see Defining Views). So to answer with JSON, define a response type with JSON annotations, as the counterpart of the request types that commands use (see Handling Commands over HTTP), and map the items to it. Here, the revision of a book goes along as eventId, so that a caller can send it back as expectedEventId:
type bookBody struct {
ID string `json:"id"`
Title string `json:"title"`
Author string `json:"author"`
IsBorrowed bool `json:"isBorrowed"`
BorrowedUntil string `json:"borrowedUntil"`
EventID string `json:"eventId"`
}
func bookBodyOf(book BookItem) bookBody {
return bookBody{
ID: book.ID,
Title: book.Title,
Author: book.Author,
IsBorrowed: book.IsBorrowed,
BorrowedUntil: book.BorrowedUntil,
EventID: book.EventID,
}
}
To answer with such bodies, wrap the function that answers the query with items, such as listBooks (see Defining Queries), and map what it returns. Since every query that lists books returns items, one wrapper serves all of them:
func answerBooks[TQuery any](ask httpapi.Answer[TQuery, []BookItem]) httpapi.Answer[TQuery, []bookBody] {
return func(ctx context.Context, q TQuery) ([]bookBody, error) {
books, err := ask(ctx, q)
if err != nil {
return nil, err
}
bodies := make([]bookBody, 0, len(books))
for _, book := range books {
bodies = append(bodies, bookBodyOf(book))
}
return bodies, nil
}
}
Then call the Query function with the API, the mux, a pattern, the function that returns the query, and the function that answers it:
httpapi.Query(api, mux, "GET /api/books", toListBooks, answerBooks(listBooks(catalog)))
The route answers with 200 OK and the result as JSON. A result without items is answered with an empty list, [], even as the nil slice that slices.Collect returns when there are no items. Errors are answered as for commands, and errors returned from the first function are treated as they are from ToCommand (see Authorizing Commands).
To answer this way in a handler of your own, call the RespondResult function with the response writer, the request, the API, the result, and the error.
Note that the functions have the types httpapi.ToQuery and httpapi.Answer. The answering function receives neither the request nor the user.
Answering Queries in Your Own Format
To answer in a format of your own, call the Ask function in a handler of your own. It does the same as a route, but writes nothing to the response. Instead, it returns the result:
mux.HandleFunc("GET /api/books", func(w http.ResponseWriter, r *http.Request) {
books, err := httpapi.Ask(r, api, toListBooks, listBooks(catalog))
if err != nil {
// ...
}
// ...
})
Reading Queries from the Body
Some queries need more input than fits into the query string, for example a list of books to check at once. Send such a query as the body of a POST request, and call the BodyOf function with the type of the body to read it:
type CheckAvailability struct {
BookIDs []string `json:"bookIds"`
}
toCheckAvailability := func(r *http.Request, user User) (CheckAvailability, error) {
return httpapi.BodyOf[CheckAvailability](r)
}
httpapi.Query(api, mux, "POST /api/check-availability", toCheckAvailability, checkAvailability(catalog))
The function reads the body by the same rules as for a command (see Validating Requests), and returns the same errors, so the request is answered with 415, 413, or 400 as a command would be. It works in a handler of your own as well.
Reporting Missing Items
If the answering function returns query.ErrNoItems, as query.Single does if no item matches, the request is answered with 404 Not Found. So a wrapper that maps a single item hands on the error as it is:
func answerBook[TQuery any](ask httpapi.Answer[TQuery, BookItem]) httpapi.Answer[TQuery, bookBody] {
return func(ctx context.Context, q TQuery) (bookBody, error) {
book, err := ask(ctx, q)
if err != nil {
return bookBody{}, err
}
return bookBodyOf(book), nil
}
}
httpapi.Query(
api,
mux,
"GET /api/books/{id}",
func(r *http.Request, user User) (GetBook, error) {
return GetBook{BookID: r.PathValue("id")}, nil
},
answerBook(getBook(catalog)),
)
To report a missing item yourself, return httpapi.ErrNotFound.
Mapping Errors to Status Codes
To get the status code that matches an error, call the StatusFor function:
status := httpapi.StatusFor(err)
It checks the categories in this order:
| Error | Status code |
|---|---|
nil |
200 OK |
httpapi.ErrUnauthorized |
401 Unauthorized |
httpapi.ErrForbidden |
403 Forbidden |
httpapi.ErrTooLarge |
413 Request Entity Too Large |
httpapi.ErrUnsupportedMediaType |
415 Unsupported Media Type |
httpapi.ErrMalformed |
400 Bad Request |
httpapi.ErrNotFound, query.ErrNoItems |
404 Not Found |
architecturekit.ErrDomain |
422 Unprocessable Entity |
architecturekit.ErrConflict |
409 Conflict |
architecturekit.ErrTransient |
503 Service Unavailable |
context.Canceled |
499 Client Closed Request |
context.DeadlineExceeded |
503 Service Unavailable |
| any other error | 500 Internal Server Error |
Note that context.Canceled means that the caller went away before it got an answer. HTTP has no status code for that, so 499 is the one that nginx introduced, and which logs and metrics commonly know. Since nothing failed, it is not logged.
Note that the status code says nothing about what to tell the caller. If you answer in a format of your own, leave out the error for 401, 409, and 500 and above, as Respond and RespondResult do, since it may name internals (see Handling Commands over HTTP).
Reading Your Own Writes over HTTP
To let a caller read its own writes over HTTP, hand over the Revisioned option to Query, with a view that implements Revisioned, whose projection is tracked (see Tracking Revisions), and how long to wait at most:
httpapi.Query(api, mux, "GET /api/books", toListBooks, answerBooks(listBooks(catalog)),
httpapi.Revisioned(catalog, httpapi.DefaultWait),
)
After sending a command, the caller takes the revision from the answer and sends it in the Wait-For-Revision header of the query:
curl http://localhost:8080/api/books \
-H "Wait-For-Revision: 1"
The route waits until the view has reached this revision, but at most for the given duration, which is five seconds for httpapi.DefaultWait. Then it answers with what the view holds, even if the time has run out. Without the header, it does not wait at all. If the header holds something that is not a revision, the request is answered with 400 Bad Request.
Once the view has seen at least one event, the response contains the revision it shows in the Revision header, as well as an ETag header and Cache-Control: private, no-cache. If the caller sends the ETag in the If-None-Match header, asks the same, and the view has not changed since, the request is answered with 304 Not Modified. private keeps shared caches, such as proxies, from keeping the answer.
The ETag holds the query that was asked, with every field. So two callers get the same ETag only if their queries are equal: a query that holds the user, or anything else that tells callers apart, gets an ETag of its own for each of them. That matters as soon as callers share a browser one after the other, since the browser asks with the ETag it kept for the one before. The query is built before the route waits or answers 304 Not Modified, so a caller who may not ask is refused first.
This holds as long as the answer depends on nothing but the query and the view, which is why the answer sees neither the request nor the user. Three things get past it:
- The clock, or anything else outside the events. See Depending on More Than the Read Model.
- Another view. The revision is that of the view handed over, so an answer that also reads from another view does not notice when that one changes.
- The context. A value that a middleware puts into the context, such as the user, never shows up in the
ETag. Put it into the query instead.
A query that holds a function or a channel can not be written into an ETag, and its answer goes without one.
Note that the constants httpapi.HeaderWaitFor and httpapi.HeaderRevision contain the names of the two headers.
Depending on More Than the Read Model
If an answer depends on more than the view, for example on the current date, the simplest way is to put that value into the query, as Today below. Since the query is part of the ETag, the ETag changes with it, and Revisioned is all it takes:
type ListOverdueBooks struct {
Today string
}
func listOverdueBooks(catalog architecturekit.View[BookItem]) func(context.Context, ListOverdueBooks) ([]BookItem, error) {
return func(ctx context.Context, q ListOverdueBooks) ([]BookItem, error) {
items, err := catalog.All(ctx)
if err != nil {
return nil, err
}
return slices.Collect(query.Where(items, func(item BookItem) bool {
return item.IsBorrowed && item.BorrowedUntil < q.Today
})), nil
}
}
httpapi.Query(api, mux, "GET /api/overdue-books",
func(r *http.Request, user User) (ListOverdueBooks, error) {
return ListOverdueBooks{Today: time.Now().Format(time.DateOnly)}, nil
},
answerBooks(listOverdueBooks(catalog)),
httpapi.Revisioned(catalog, httpapi.DefaultWait),
)
If the answer takes such a value from elsewhere instead, for example because it reads the clock itself, hand over the Varying option as well, with a function of the type httpapi.Volatile. It receives the request and returns a value that changes whenever the answer would, and that becomes part of the ETag as well:
func today(*http.Request) string {
return time.Now().Format(time.DateOnly)
}
httpapi.Query(api, mux, "GET /api/books-due-today", toListBooksDueToday, answerBooks(listBooksDueToday(catalog)),
httpapi.Revisioned(catalog, httpapi.DefaultWait),
httpapi.Varying(today),
)
Waiting for a Revision in a Handler of Your Own
To read its own writes in a handler of your own, for example one that answers in another format than JSON, call the Await function with the request, the view, and how long to wait at most. It waits for the revision the request asks for, within the context of the request. Running out of time is not an error. It returns an error if the header holds something that is not a revision, or if waiting fails for another reason. Determine the caller first, so that nobody can make the server wait without being allowed to ask:
mux.HandleFunc("GET /api/books.csv", func(w http.ResponseWriter, r *http.Request) {
if _, err := httpapi.UserOf(r, api); err != nil {
httpapi.RespondResult(w, r, api, struct{}{}, err)
return
}
if err := httpapi.Await(r, catalog, httpapi.DefaultWait); err != nil {
httpapi.RespondResult(w, r, api, struct{}{}, err)
return
}
// ...
})
Note that such a handler answers with neither an ETag nor 304 Not Modified. Those come with the Revisioned option of Query, which ties the ETag to the query, so that callers never share one by accident.
Checking Health over HTTP
An orchestrator such as Kubernetes regularly asks an application whether it can serve requests, and whether it is alive. To answer both by the state of the projections, hand the runs started with StartProjection over to the Readiness and Liveness functions, and serve the handlers they return on paths of your choice. They list each projection by the name it was given with Named:
run := architecturekit.StartProjection(ctx, store, architecturekit.SubjectTree("/books"), catalogProjection,
architecturekit.Named("catalog"),
)
mux.Handle("GET /ready", httpapi.Readiness(run))
mux.Handle("GET /live", httpapi.Liveness(run))
Note that a run without a name, or two runs of the same name, make Readiness and Liveness panic.
Both answer with 200 OK or 503 Service Unavailable, depending on where the projections stand:
| Projection | Readiness |
Liveness |
|---|---|---|
| catches up for the first time | 503 |
200 |
| is live | 200 |
200 |
| reconnects after it has caught up | 200 |
200 |
| has stopped | 503 |
503 |
The application is ready once every projection has caught up, since a half-built view answers wrongly. A projection that reconnects later on, for example because the database restarts, keeps it ready: its view is behind, but consistent, and every instance shares the database, so taking them all out would answer nothing instead of something that is behind. A projection that has stopped makes the application neither ready nor alive, since its view never changes again. The orchestrator then restarts the application, which builds the view anew, with a configuration that may have been fixed in the meantime. There is no time limit for reconnecting, since a restart does not bring the database back.
The body of Readiness tells whether the application is ready, and where each projection stands:
{
"isReady": true,
"projections": {
"catalog": {
"phase": "live",
"since": "2026-09-30T12:00:00Z",
"hasCaughtUp": true,
"attempts": 0,
"revision": "42"
}
}
}
The body of Liveness has the same shape, but tells whether the application is alive, with isAlive instead of isReady.
Note that the body does not tell why a projection reconnects or has stopped, since health checks are usually reachable without signing in, and the reason may name internal addresses. Log it instead, for example by waiting for Done and calling Err.
Testing Deciders
To test deciders without a database, use the architecturekittest package:
import "github.com/thenativeweb/architecturekit-golang/architecturekit/architecturekittest"
Call the Given function with a *testing.T, the decider, and the events that have happened so far. Then call the When function with the command, and check the outcome:
func TestBorrowBook(t *testing.T) {
architecturekittest.Given(t, borrowBook,
BookAcquired{
Title: "2001 – A Space Odyssey",
Author: "Arthur C. Clarke",
ISBN: "978-0756906788",
},
).
When(BorrowBook{
BookID: "42",
ReaderID: "23",
BorrowedUntil: "2026-10-24",
}).
ThenEvents(BookBorrowed{
BorrowedBy: "23",
BorrowedUntil: "2026-10-24",
})
}
Given returns a *Fixture, and When returns an *Outcome. The functions that check the outcome return the outcome again, so they can be chained.
Like Execute, When refuses an event that the state of the decider has no rule for (see Executing Commands). The outcome is then the same error of the category ErrPermanent that Execute returns, so ThenEvents and the other functions that expect events, or nothing, fail and name the event type, and ThenFailed(architecturekit.ErrPermanent) matches.
Note that Given accepts any value that provides the Helper and Fatalf functions, as described by the TestingT interface.
Expecting Events
To expect exactly the given events, in the given order, call the ThenEvents function, as shown above. To expect neither events nor an error, call the ThenNothing function:
architecturekittest.Given(t, returnBook, BookAcquired{}).
When(ReturnBook{BookID: "42"}).
ThenNothing()
To check the events with a function, call the ThenSomeEvent function to expect at least one matching event, the ThenEveryEvent function to expect only matching events, and at least one, or the ThenNoEvent function to expect no matching event:
isBookBorrowed := func(event architecturekit.Event) bool {
_, ok := event.(BookBorrowed)
return ok
}
architecturekittest.Given(t, borrowBook, BookAcquired{}).
When(BorrowBook{BookID: "42", ReaderID: "23"}).
ThenEveryEvent(isBookBorrowed)
Expecting Rejections
To expect that a command is rejected with exactly the given message, call the ThenRejected function:
architecturekittest.Given(t, acquireBook, BookAcquired{}).
When(AcquireBook{BookID: "42"}).
ThenRejected("book 42 has already been acquired")
To expect an error of a category instead, call the ThenFailed function:
architecturekittest.Given(t, borrowBook).
When(BorrowBook{BookID: "42"}).
ThenFailed(architecturekit.ErrDomain)
Expecting Preconditions
To expect exactly the given preconditions, in the given order, call the ThenPreconditions function. Describe the preconditions of the kit with the OnStateRead and Unconditionally functions, and those of the client SDK with the OnPristineSubject, OnPopulatedSubject, OnEventID, and OnQuery functions. For example, the BorrowBook that checks the revision of the caller requires its subject to be on the event ID the command carries (see Checking the Revision of the Caller):
architecturekittest.Given(t, borrowBook, BookAcquired{}).
When(BorrowBook{BookID: "42", ReaderID: "23", ExpectedEventID: "0"}).
ThenPreconditions(architecturekittest.OnEventID("/books/42", "0"))
OnPristineSubject describes a precondition created with NewIsSubjectPristinePrecondition, and OnPopulatedSubject one created with NewIsSubjectPopulatedPrecondition. A test that expects the one fails for a command that declares the other, and the failure names both. For example, AcquireBook requires a pristine subject (see Preventing Duplicates):
architecturekittest.Given(t, acquireBook).
When(AcquireBook{BookID: "42"}).
ThenPreconditions(architecturekittest.OnPristineSubject("/books/42"))
To get the preconditions of a command directly, call the PreconditionsOf function. It returns a slice of Precondition, with the fields Subject, Pristine, Populated, EventID, Query, OnStateRead, and Unconditional:
preconditions := architecturekittest.PreconditionsOf(ReturnBook{BookID: "42"})
Inspecting State
To check the state the command has been decided on, call the ThenState function with a function that receives the state:
architecturekittest.Given(t, returnBook, BookAcquired{}, BookBorrowed{}).
When(ReturnBook{BookID: "42"}).
ThenState(func(book Book) {
if !book.IsBorrowed {
t.Fatal("expected the book to be borrowed")
}
})
Testing Upcasters
To test an upcaster, call the GivenStored function instead of Given, and hand over the events as they are stored. They run through the upcasters, as they do when reading from the database. To turn a typed event into a stored one, call the StoredEvent function with the subject, the event ID, and the event:
architecturekittest.GivenStored(t, borrowBook,
architecturekittest.StoredEvent("/books/42", "0", BookAcquired{
Title: "2001 – A Space Odyssey",
Author: "Arthur C. Clarke",
ISBN: "978-0756906788",
}),
eventsourcingdb.Event{
Subject: "/books/42",
Type: "io.eventsourcingdb.library.book-lent",
ID: "1",
Data: json.RawMessage(`{"lentTo":"23","until":"2026-10-24"}`),
},
).
When(BorrowBook{BookID: "42", ReaderID: "17"}).
ThenRejected("book 42 is already borrowed")
Replaying Events Directly
To evolve a state from events without a decider, call the Replay function with the state and typed events, or the ReplayStored function with the state and stored events. To check the state after each event, advance it one event at a time with the Step function (see Stepping Through States):
book, err := architecturekit.Replay(bookState, BookAcquired{}, BookBorrowed{})
if err != nil {
// ...
}
Testing Projections
To test a projection without a database, call the Project function with a *testing.T, the projection, and the stored events. To turn typed events for the same subject into stored ones, call the StoredEvents function, which numbers them from 0:
func TestCatalogProjection(t *testing.T) {
catalog := newCatalog()
catalogProjection := newCatalogProjection(catalog)
architecturekittest.Project(t, catalogProjection,
architecturekittest.StoredEvents("/books/42",
BookAcquired{
Title: "2001 – A Space Odyssey",
Author: "Arthur C. Clarke",
ISBN: "978-0756906788",
},
BookBorrowed{
BorrowedBy: "23",
BorrowedUntil: "2026-10-24",
},
)...,
)
architecturekittest.ExpectItems(t, catalog, BookItem{
ID: "42",
Title: "2001 – A Space Odyssey",
Author: "Arthur C. Clarke",
IsBorrowed: true,
BorrowedUntil: "2026-10-24",
EventID: "1",
})
}
Project hands over the events as they are stored, so a projection that uses upcasters runs them, just as it does with a database. To test that a projection handles an older event type, hand over an event of that type.
The ExpectItems function expects the view to hold exactly the given items, in the given order. It requires an item type that is comparable. To get the items as a slice instead, call the ItemsOf function:
items := architecturekittest.ItemsOf(t, catalog)
If a projection reads the time of an event, or the events of several subjects have to follow one another, call the StoredEventsAt function instead of StoredEvents. It numbers the events from the given ID on, and times them one minute apart from the given time on:
noon := time.Date(2026, time.March, 2, 12, 0, 0, 0, time.UTC)
events := append(
architecturekittest.StoredEventsAt("/books/42", 0, noon, BookAcquired{ /* ... */ }),
architecturekittest.StoredEventsAt("/books/23", 1, noon.Add(time.Minute), BookAcquired{ /* ... */ })...,
)
To check how a projection will be run, call the ExpectMode function:
architecturekittest.ExpectMode(t, catalogProjection, architecturekit.ModeRebuild)
To test a transactional projection, call the ProjectTransactional function instead of Project. It begins a transaction, applies the events, and commits the transaction with the ID of the last event. If an event is refused, it rolls the transaction back:
architecturekittest.ProjectTransactional(t, &TransactionalBookTableProjection{},
architecturekittest.StoredEvents("/books/42",
BookAcquired{
Title: "2001 – A Space Odyssey",
Author: "Arthur C. Clarke",
ISBN: "978-0756906788",
},
)...,
)
Note that Project fails the test for a projection that implements Transactional in addition to Apply.
Testing Queries
To test a query, call the function that answers it directly, with a view filled by a projection:
books, err := listBooks(catalog)(context.TODO(), ListBooks{OnlyAvailable: true})
if err != nil {
// ...
}
To test the queries of an HTTP API without a database, create the API without a store, with nil. It answers queries, and answers a command with 500 Internal Server Error, and logs that the API has no store:
api := httpapi.NewAPI(nil, userFrom)
Testing with a Database
Some tests need a real database, for example to run commands from end to end. To get one, use the dbtest package:
import "github.com/thenativeweb/architecturekit-golang/architecturekit/architecturekittest/dbtest"
Call the Store function with a *testing.T, the source, and the schemas to register. It returns a store on a database that all tests of the package share, which the first test that asks for it starts in a container. Options for the store, such as WithStateCache, follow the schemas, so that a test runs with the same store as the application:
func TestAcquireBook(t *testing.T) {
store := dbtest.Store(t, "https://library.eventsourcingdb.io", bookState.Schemas())
_, err := architecturekit.Execute(context.TODO(), store, acquireBook, AcquireBook{
BookID: uuid.NewString(),
// ...
})
if err != nil {
t.Fatal(err)
}
}
To stop the database once all tests have run, call the Main function from TestMain:
func TestMain(m *testing.M) {
dbtest.Main(m)
}
If TestMain has more to do once the tests have run, such as closing a browser, run the tests yourself, and call the StopSharedDatabase function afterwards. It does nothing if no test has started the database:
func TestMain(m *testing.M) {
code := m.Run()
closeBrowser()
if err := dbtest.StopSharedDatabase(); err != nil {
fmt.Fprintln(os.Stderr, err)
code = 1
}
os.Exit(code)
}
Note that the schemas come as a single slice, so that the options can follow. To hand over the schemas of several states, join them with slices.Concat, for example slices.Concat(bookState.Schemas(), readerState.Schemas()).
The tests share the events as well, so a test writes to subjects of its own, for example with a random ID in them, and reads only from those. A test that reads more than that, such as a projection from /, needs a database of its own. Call the IsolatedStore function to start one for the test alone, which is stopped once the test is over. It takes the same arguments as Store, and a few seconds to start, so use it only where the shared one would not do.
For a test that connects by itself, such as one that starts a whole server, call the SharedDatabase or the IsolatedDatabase function. Each returns a *Database, whose URL and APIToken fields are what a client needs. Its Client function returns a client, for example to write an event that no command would, and its Store function returns a store, as above:
database := dbtest.SharedDatabase(t)
config := server.Config{
DatabaseURL: database.URL.String(),
APIToken: database.APIToken,
}
Note that with -short every test that asks for a database is skipped, so that the other tests run without Docker.
Note that dbtest is a package of its own because it starts the database with Testcontainers, which brings along the Docker client. A package whose tests only import architecturekittest, for example to test deciders and projections, builds without either.
Directories
¶
| Path | Synopsis |
|---|---|
|
Package architecturekit provides building blocks for CQRS and event-sourced applications on top of the EventSourcingDB.
|
Package architecturekit provides building blocks for CQRS and event-sourced applications on top of the EventSourcingDB. |
|
architecturekittest
Package architecturekittest provides given-when-then for deciders and projections.
|
Package architecturekittest provides given-when-then for deciders and projections. |
|
architecturekittest/dbtest
Package dbtest starts an EventSourcingDB in a container for the tests that need a real one, and hands out stores on it.
|
Package dbtest starts an EventSourcingDB in a container for the tests that need a real one, and hands out stores on it. |
|
httpapi
Package httpapi exposes commands and queries over HTTP, lets a caller read its own writes (see Revisioned and Await), and answers the health checks of an orchestrator (see Readiness and Liveness).
|
Package httpapi exposes commands and queries over HTTP, lets a caller read its own writes (see Revisioned and Await), and answers the health checks of an orchestrator (see Readiness and Liveness). |
|
query
Package query runs queries over a view.
|
Package query runs queries over a view. |