Projections

Defining Projections#

A projection turns events into a view. Call the NewTypedProjection 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.NewTypedProjection().
    On(func(ctx context.Context, event architecturekit.Envelope[BookAcquired]) error {
      _, err := catalog.Insert(ctx, event.ID, BookItem{
        ID:     bookIDOf(event.Subject),
        Title:  event.Data.Title,
        Author: event.Data.Author,
      })
      return err
    }).
    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 {
  bookID, _ := bookSubject.Value(subject, "book")
  return bookID
}

NewTypedProjection 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.NewTypedProjection().
    // ...
    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, and so does calling it with nil as the function, or with a pointer or an interface as the event type, as with Evolve. Calling UpcastWith twice, or with nil, panics as well, as it does on a state.

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.

Note that a subject that does not start with a slash makes SubjectTree and ExactSubject panic, and that the zero value of Subjects, which names no subject, makes StartProjection, the other functions that run a projection, and Read panic, as does a nil projection. A nil pointer or a nil ProjectionFunc counts as a nil projection.

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 with the WaitCaughtUp function, which takes a context and the runs:

ctx, cancel := context.WithCancel(context.TODO())
defer cancel()

run := architecturekit.StartProjection(ctx, store, architecturekit.SubjectTree("/books"), catalogProjection,
  architecturekit.Named("catalog"),
)

if err := architecturekit.WaitCaughtUp(ctx, run); err != nil {
  return 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.

Note that an empty name makes Named panic, and that giving Named twice makes StartProjection and the other functions that run a projection panic, and so does a nil option.

WaitCaughtUp waits until every run has applied the events that were stored when it started, and returns nil then, and only then, so that nil always means that the views are complete. To wait for several projections, hand over all of their runs, as in architecturekit.WaitCaughtUp(ctx, catalogRun, readersRun). It waits for all of them at once, not one after the other, and returns as soon as one of them ends before it has caught up:

• If a run ends before it catches up, WaitCaughtUp returns the error the run ended with. If the run has a name, the error names it, as in projection "catalog": …, and wraps the error of the run, so that errors.Is still finds what that wraps.
• If a run ends without an error before it catches up, because its context ended, WaitCaughtUp returns an error that wraps context.Canceled and says that the projection stopped before it caught up, naming it if it has a name.
• If the context handed to WaitCaughtUp ends first, it returns the error of that context.

A run that has caught up counts as caught up, even if it has ended since, or the context handed to WaitCaughtUp has ended. If several of the other cases have happened by the time WaitCaughtUp looks, the error a run ended with comes first, since it tells why, that of the first such run among the ones handed over, and the error of the context handed to WaitCaughtUp comes next. So if the run and WaitCaughtUp share a context, as in the example above, and that context times out, WaitCaughtUp returns its error, context.DeadlineExceeded, which tells more than context.Canceled.

Note that WaitCaughtUp returns nil at once without any run, since none of them is behind, and that a nil run makes it panic.

For finer control, the run offers what WaitCaughtUp waits for. 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. If the run ends before it catches up, the channel is never closed, so wait for Done as well. 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.

If the projection panics, for example because Apply writes into a map that was never made, the run ends as well, rather than the whole process. Err then returns an error of the category ErrPermanent, since a panic is a mistake in the code that trying again will not fix. Its message holds the value and the stack of the panic, so log it to find out where the panic happened. This holds for every function of the projection that the run calls, and for the observer of reconnects (see below). Once the run has ended, Liveness answers 503, so that the orchestrator restarts the application (see Checking Health over HTTP).

Note that canceling the context stops the projection, so defer cancel() stops it as soon as the surrounding function returns. That is too early for a function that only sets up the application (see Putting It Together).

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, hand WaitCaughtUp a context with a deadline, such as one from context.WithTimeout. Its end stops the waiting, but not the run (see Putting It Together).

Changed in 2.0If reading fails with an error of the category ErrTransient, the run waits and continues after the last event it has applied, until the context is canceled. That includes a stream that ends, for example because the database restarts, or because a load balancer ends long-lived connections: the database never ends a stream on its own, so the client reports that it broke off. 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 caught up and followed the stream for longer than the initial delay, 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 within the initial delay after, is given ever more time.

Note that how long the projection followed the stream is compared with the initial delay, not with the delay it has grown to. Otherwise, once an outage had let the delay grow to one minute, a load balancer that ends connections every 30 seconds would keep a quiet projection waiting for a minute after each of them, and a caller who wants to read their own write in the meantime would not see it (see Reading Your Own Writes). In return, a database that ends every stream a few seconds after the projection has caught up is tried again every few seconds rather than once a minute. That is not a loop without a pause, and catching up, the expensive part, succeeds each time.

A connection can also stall without being closed, for example behind a proxy that keeps it open but no longer passes anything on. While there is nothing else to send, the database sends a heartbeat every second, so the client ends a stream on which neither an event nor a heartbeat has arrived for 30 seconds, and the run reconnects as after any other failure of the category ErrTransient. Without that, the run would wait for the next event forever, while Status kept reporting PhaseLive.

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. Calling WithReconnectObserver with nil panics as well.

The observer receives a Reconnect with these fields:

• Projection is the name the projection was given with Named, or empty if it has none.
• Subject is the subject the projection reads.
• Changed in 2.0Err is the reason, such as a stream that broke off.
• Delay is how long the projection waits before the next attempt.
• Attempt counts 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 initial delay. So it stays at 1 while a load balancer ends long-lived connections regularly.

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:

• Phase is PhaseCatchingUp, PhaseLive, PhaseReconnecting, or PhaseStopped.
• Since is when the phase began. For PhaseReconnecting, that is when the disruption began, not when the latest attempt did.
• Changed in 2.0Err is why the run is reconnecting or has stopped, such as a stream that broke off. It is nil if the run stopped because its context ended.
• Attempts counts the attempts to read again within the current disruption. Unlike Attempt of Reconnect, it starts over whenever the run has caught up, even if the stream ends again within the initial delay.
• Revision is the ID of the last event the run has applied and committed.
• HasCaughtUp tells whether the run has caught up at least once. Like CaughtUp, it stays true while 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 an error that wraps the one of the context, so that a read model that is only partly built does not look complete. If the projection panics, it returns the same error a run ends with, rather than panicking:

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 NewTypedProjection resumable, embed it in a type of your own, and add the two functions there:

type TypedBookTableProjection struct {
  *architecturekit.TypedProjection
  // ...
}

func (p *TypedBookTableProjection) Checkpoint(ctx context.Context) (string, error) {
  // ...
}

func (p *TypedBookTableProjection) 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 NewTypedProjection. 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 {
    // The database can not be reached, which may pass.
    return nil, fmt.Errorf("%w: beginning a transaction: %v", architecturekit.ErrTransient, err)
  }

  return &bookTableTx{
    tx: tx,
    TypedProjection: architecturekit.NewTypedProjection().
      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, or panics, 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.

The place where a view keeps its data can fail as well, and such a failure counts like an error of Apply, whether it comes from Checkpoint, SaveCheckpoint, Begin, Commit, or the Apply function of a Tx. An error that does not belong to ErrTransient ends the run, so Liveness answers 503, and the orchestrator restarts the application (see Checking Health over HTTP). That fits a mistake, such as a statement that the database refuses, but not a failure that may pass, such as a lost connection or a deadlock. Report such a failure as an error of the category ErrTransient, as Begin does above. The run then waits and tries again from where it stopped, with the same growing delay as when reading from EventSourcingDB fails (see Running Projections). A panic in any of these functions, on the other hand, always ends the run, since it is a mistake in the code, even if it panics with an error of the category ErrTransient. A panic in the Apply function of a Tx rolls the transaction back first, as an error does.

Note that which failures may pass depends on the database and its driver, so mark only those. A failure that is marked as transient but never passes keeps the run trying forever, while Liveness keeps answering 200.

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.