Live queries

A client can subscribe to a query. It gets the current answer first, then a new one each time a command changes it. Nothing polls: the host is the only writer of an actor, so it knows which tables each command wrote, and it runs a subscribed query again only when one of those tables is a table the query read.

Every query on a client has subscribe, which takes the query's payload and returns a Stream. Each update is { value, version }, where version counts the actor's commands that the value includes.

This page uses one actor:

room.contract.ts

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

export const Room = Akter.make("Room", {
  key: Schema.String,
  state: Schema.Struct({ title: Schema.String }),
  initial: () => ({ title: "Untitled" }),
  commands: {
    post: Akter.command({ payload: { text: Schema.String }, success: Schema.Void }),
    rename: Akter.command({ payload: { title: Schema.String }, success: Schema.Void }),
  },
  queries: {
    recent: Akter.query({ payload: { limit: Schema.Int }, success: Schema.Array(Schema.String) }),
  },
})

room.actor.ts

import { Effect } from "effect"
import { Akter } from "@akterjs/effect/contract"
import { Room } from "./room.contract.ts"

export const RoomLive = Room.toLayer({
  migrations: {
    1: (sql) => sql`CREATE TABLE messages (seq INTEGER PRIMARY KEY, text TEXT NOT NULL)`,
  },
  commands: {
    post: (_room, { text }) => Effect.flatMap(Akter.sql, (sql) => sql`INSERT INTO messages ${sql.insert({ text })}`),
    rename: (_room, { title }) => Room.update({ title }),
  },
  queries: {
    recent: (_room, { limit }) =>
      Effect.flatMap(Akter.sql, (sql) =>
        Effect.map(
          sql<{ text: string }>`SELECT text FROM messages ORDER BY seq DESC LIMIT ${limit}`,
          (rows) => rows.map((row) => row.text),
        ),
      ),
  },
})

This script serves the room on the local engine, subscribes to recent, and posts a message after each update. Before the second post it renames the room:

watch.ts

import { Effect, Layer, Stream } from "effect"
import { HttpServer } from "effect/http"
import { Akter } from "@akterjs/effect"
import { akterClient } from "@akterjs/client/effect"
import { Room } from "./room.contract.ts"
import { RoomLive } from "./room.actor.ts"

const server = Akter.serve({ actors: [Room], port: 0 }).pipe(
  Layer.provide(Akter.local(".akter").pipe(Layer.provide(RoomLive))),
)

const client = Layer.unwrap(HttpServer.addressFormattedWith((url) => Effect.succeed(akterClient({ url }))))

const program = Effect.gen(function* () {
  const room = yield* Room.client("lobby")

  yield* room.recent.subscribe({ limit: 2 }).pipe(
    Stream.take(3),
    Stream.zipWithIndex,
    Stream.runForEach(([{ value, version }, index]) =>
      Effect.gen(function* () {
        console.log(`version ${version}: [${value.join(", ")}]`)
        if (index === 0) yield* room.post({ text: "hi" })
        if (index === 1) {
          yield* room.rename({ title: "Lunch" })
          yield* room.post({ text: "lunch?" })
        }
      }),
    ),
  )
})

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

Text

version 0: []
version 1: [hi]
version 3: [lunch?, hi]

There's no update at version 2. rename changed only the room's state, which recent doesn't read, so the host didn't run the query again.

What you can rely on⁠

Every value comes from committed commands. A subscriber never sees a change that could still roll back.
A command that changes the answer always sends an update. A command that leaves it the same sends none, so two updates in a row always differ.
room.state is a Stream of the actor's state that works the same way: the current state first, then a new value each time a command changes it.

Subscribe⁠

What you can rely on⁠