History and replay

Every actor keeps a journal of the commands it ran: their ids, payloads, principals, times and outcomes. Queries aren't journaled. A command whose handler fails is journaled too, but a request refused before the handler runs isn't.

This is a record of commands, not event sourcing. The actor opens its SQLite database to get its state; it doesn't rebuild state from history.

Read the journal⁠

The contract types each entry's payload and outcome:

counter.contract.ts

import { Schema } from "effect"
import { Akter } from "@akterjs/effect/contract"

export class NegativeAmountError extends Schema.TaggedError<NegativeAmountError>()("NegativeAmountError", {}) {}

export const Counter = Akter.make("Counter", {
  key: Schema.String,
  state: Schema.Struct({ count: Schema.Int }),
  initial: () => ({ count: 0 }),
  commands: {
    add: Akter.command({ payload: { by: Schema.Int }, success: Schema.Int, error: NegativeAmountError }),
  },
})

This handler writes an audit row with an id and time from Akter. Those values are recorded with the command, so replay can give the handler the same ones:

counter.actor.ts

import { Effect } from "effect"
import { Akter } from "@akterjs/effect/contract"
import { Counter, NegativeAmountError } from "./counter.contract.ts"

export const CounterLive = Counter.toLayer({
  migrations: {
    1: (sql) => sql`CREATE TABLE additions (id TEXT PRIMARY KEY, at INTEGER NOT NULL, amount INTEGER NOT NULL)`,
  },
  commands: {
    add: (counter, { by }) =>
      Effect.gen(function* () {
        if (by < 0) {
          return yield* new NegativeAmountError()
        }
        const sql = yield* Akter.sql
        const id = yield* Akter.id
        const at = yield* Akter.now
        yield* sql`INSERT INTO additions ${sql.insert({ id, at, amount: by })}`
        const count = counter.count + by
        yield* Counter.update({ count })
        return count
      }),
  },
})

Counter.history(key) returns { entries, coveredSince }. Each entry has a kind: recorded for a command, created for creation, result for a job result, or unreadable when today's contract can't decode an old entry. Narrow by kind before using its typed payload and outcome.

This script serves the counter on the local engine, runs three commands and reads their journal:

history.ts

import { Effect, Layer, Option } from "effect"
import { HttpServer } from "effect/http"
import { Akter, Verify } from "@akterjs/effect"
import { akterClient } from "@akterjs/client/effect"
import { Counter } from "./counter.contract.ts"
import { CounterLive } from "./counter.actor.ts"

const local = Akter.local(".akter").pipe(Layer.provide(CounterLive))
const server = Akter.serve({ actors: [Counter], port: 0 }).pipe(Layer.provideMerge(local))
const client = Layer.unwrap(HttpServer.addressFormattedWith((url) => Effect.succeed(akterClient({ url }))))

const program = Effect.gen(function* () {
  const counter = yield* Counter.client("clicks")
  yield* counter.add({ by: 2 })
  yield* counter.add({ by: 3 })
  yield* Effect.flip(counter.add({ by: -1 }))

  const history = yield* Counter.history("clicks")
  for (const entry of history.entries) {
    if (entry.kind === "recorded") {
      console.log(`${entry.command} ${entry.payload.by} ${entry.exit._tag}`)
    }
  }

  const replay = yield* Verify.replay(CounterLive, { actor: Counter, directory: ".akter", key: "clicks" })
  console.log(`replayed ${replay.window.commands}, matched ${Option.isNone(replay.difference)}`)
})

Effect.runPromise(program.pipe(Effect.scoped, Effect.provide(client.pipe(Layer.provideMerge(server)))))

Text

add 2 Success
add 3 Success
add -1 Failure
replayed 3, matched true

Read a window⁠

Pass { since: "1 hour" } to read recent entries. Counter.historyStream(key, options) reads the same history as a finite stream of pages, rather than collecting it all in memory. It isn't a live subscription.

coveredSince says how far back the retained journal covers. Counter.historyCoveredSince(key, options) reads that boundary without collecting entries. Set retention with Akter.historyRetention(duration) when constructing the local engine; removing old entries limits both history and replay.

Replay a command history⁠

Verify.replay(handlers, { actor, directory, key }) reruns the stored commands in a new in-memory database using the handlers you pass. It supplies the ids and clock values recorded by Akter.id and Akter.now, and compares command outcomes with their recorded ones. It doesn't change the original database.

The report gives the replayed window and the first differing outcome, if any, in difference. It also reports a difference in file digests under files. A matching report isn't a comparison of every database row, so keep assertions about your domain state in tests too.

In a story, call story.replay(Counter, key) instead of passing the directory yourself. A pruned journal fails replay with HistoryPrunedError, because it no longer holds every command needed to start from the beginning.