# Projections

This page shows how to define projections, which turn events into views, how to run them in the background and wait until they have caught up, how to resume them where they stopped, and how to apply events in batches. A projection can also forward events to other systems (see **[Publishing Events](/docs/architecturekit/v2.0/publishing-events)**).

## 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:

```go
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](/docs/architecturekit/v2.0/composing-subjects)**):

```go
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](/docs/architecturekit/v2.0/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:

```go
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](/docs/architecturekit/v2.0/versioning-events)**):

```go
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:

```go
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:

```go
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:

```go
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](/docs/architecturekit/v2.0/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](/docs/architecturekit/v2.0/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](/docs/architecturekit/v2.0/putting-it-together)**).*

To wait until the run ends, as a process does that runs nothing else, wait for `Done`:

```go
<-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](/docs/architecturekit/v2.0/putting-it-together)**).*

*(Changed in 2.0)* If 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](/docs/architecturekit/v2.0/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:

```go
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.0)* `Err` 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](/docs/architecturekit/v2.0/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.0)* `Err` 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.

```go
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:

```go
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](/docs/architecturekit/v2.0/projections#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:

```go
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:

```go
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`:

```go
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](/docs/architecturekit/v2.0/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:

```go
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`:

```go
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:

```go
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](/docs/architecturekit/v2.0/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](/docs/architecturekit/v2.0/projections#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:

```go
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.*