​

Context capabilities

import { Actor, type ActorRef, type Caller } from "@rikalabs/akter"
import { pgTable, text, timestamp } from "drizzle-orm/pg-core"
import { Context, DateTime, Effect, Schema } from "effect"

class RoomClosed extends Schema.TaggedError<RoomClosed>()("RoomClosed", {}) {}
class NotAMember extends Schema.TaggedError<NotAMember>()("NotAMember", {}) {}
const MessageAdded = Actor.event("MessageAdded", {
  id: Schema.String,
  body: Schema.String,
})
const messages = Actor.table(
  pgTable("chat_messages", {
    id: text("id").primaryKey(),
    author_id: text("author_id").notNull(),
    body: text("body").notNull(),
    sent_at: timestamp("sent_at", { withTimezone: true }).notNull(),
  }),
)
const SendMessage = Actor.command("SendMessage", {
  payload: Schema.Struct({ body: Schema.String }),
  success: Schema.String,
  error: NotAMember,
})
const Chat = Actor.make("Chat", {
  key: Schema.String,
  state: Actor.state({ closed: Schema.Boolean }),
  tables: [messages],
  events: [MessageAdded],
  api: { SendMessage },
})
class Access extends Context.Service<
  Access,
  { readonly requireMember: (caller: Caller, ref: ActorRef) => Effect.Effect<string, NotAMember> }
>()("Access") {}

export const ChatLive = Chat.toLayer(
  Effect.gen(function* () {
    const access = yield* Access
    return {
      SendMessage: Effect.fn(function* ({ body }) {
        const turn = yield* Chat.Turn
        const author = yield* access.requireMember(turn.caller, turn.ref)
        yield* turn.rows(messages).insert({
          id: turn.commandId,
          author_id: author,
          body,
          sent_at: DateTime.toDateUtc(yield* DateTime.now),
        })
        yield* turn.emit(MessageAdded.make({ id: turn.commandId, body }))
        return turn.commandId
      }),
    }
  }),
)

// A helper's requirement states where it may run.
const requireOpen: Effect.Effect<
  void,
  RoomClosed,
  Context.Service.Identifier<typeof Chat.Turn>
> = Effect.gen(function* () {
  const turn = yield* Chat.Turn
  if (turn.state.closed) return yield* new RoomClosed()
})

​

Phases

​

Subscriptions

​

Owned rows

​

Blobs

import { Actor } from "@rikalabs/akter"
import { Effect, Option, Schema } from "effect"

const Attachments = Actor.blob("attachments")

const Attach = Actor.command("Attach", {
  payload: Schema.Struct({ id: Schema.String, bytes: Schema.Uint8Array, line: Schema.Uint8Array }),
})
const Download = Actor.query("Download", {
  payload: Schema.String,
  success: Schema.Option(Schema.Uint8Array),
})
const Room = Actor.make("Room", {
  key: Schema.String,
  blobs: [Attachments],
  api: { Attach, Download },
})

export const RoomLive = Room.toLayer({
  Attach: Effect.fn(function* ({ id, bytes, line }) {
    const files = (yield* Room.Turn).blob(Attachments)
    yield* files.set(id, bytes)
    yield* files.append("log", line) // a new chunk; earlier chunks are not rewritten
    yield* files.compact("log") // one chunk, same bytes
  }),
})

export const RoomReads = Room.toQueryLayer({
  Download: Effect.fn(function* (id) {
    const file: Option.Option<Uint8Array> = yield* (yield* Room.Read).blob(Attachments).get(id)
    return file
  }),
})

​

Command turns

​

Minted child ids

​

Events

import type { DateTime } from "effect"

interface EventEntry<E> {
  readonly cursor: string // pass as `after` to resume after this event
  readonly event: E
  readonly commandId: string // the command whose turn emitted it
  readonly timestamp: DateTime.Utc // database clock when the event was appended at commit
}

• UnknownCursor { cursor }: the cursor is malformed or ahead of every event this actor has committed.
• RetentionGap { cursor }: some event after the cursor has been pruned, whatever its class. The reader has to resynchronize from state.

​

Connections and streams

• conn.session.get and conn.session.set(patch) read and change this connection’s own state; conn.state is the actor’s committed state, as in every read context. A changed session is written once, after the handler returns, fenced by the actor’s generation; above 16 KiB encoded the handler is a defect that closes the connection and writes nothing.
• conn.resumed is true when this activation did not run the connection’s open: after hibernation, a move to another runner, or a restart.
• conn.send(frame) sends to this connection. conn.broadcast(frame, { except?, to? }) sends to this member’s open connections, parked or not, wherever they are held. turn.broadcast(Member, frame, { except?, to? }) does the same from a command and flushes after commit; turn.connections(Member) lists that member’s open connections so a turn can filter them into to.
• conn.connections({ session? }) lists this member’s open connections as { connectionId, caller, session? }, at most 1,000.
• A connection member may add resync({ after }), which runs on the new owner after an ungraceful owner death, with resumed === true, to replay what the client may have missed (it may not call session.set); the client receives a Resync { after, reason, deadline } control frame first and answers ResyncDone { through } within 30 seconds.
• Every member frame reaches the client as { frame, cursor?, event? }. cursor is the activation’s flushed-through watermark, the highest commit whose broadcasts to that holder all went out before this frame, and is how Resync.after is known; event is the cursor of the event the frame was sent from (conn.send(entry), or turn.broadcast(Member, entry) with an EventEntry), which is what clients deduplicate on. After Resync, the client replays events after after, reloads state derived from frames that carry no event, waits for ResyncReplayed when the member has a resync handler, and answers ResyncDone { through } within 30 seconds. A member declared with stampCursor: false hides every cursor (cursor, event, the open baseline, Resync.after, and ResyncReplayed.through), cannot declare a resync handler, and its clients resync from state. The resync handler runs under the connection’s current authorization, at bounded concurrency, and never without a cursor.
• conn.close closes this connection once the handler returns. The client sees ActorError reason SessionEnded with cause ServerClosed. A close handler runs after the client is gone: its conn.broadcast reaches the member’s other connections, and its conn.send is discarded.
• Connection handlers cannot write actor state, rows, events, or blobs. They call commands through X.get, including on the connection’s own actor. Handlers run with the actor’s tenant and the caller stored at open; the build Effect runs as a System caller, so handles acquired there carry no connection’s identity. Frame handlers are at least once, so these calls get command ids derived by HMAC from a per-connection secret, the frame sequence, and the call index: a redelivered frame replays its commands’ receipts, and no other caller can predict the ids.
• conn.connections() returns session only when asked. It shows every connection’s caller, so don’t send it to clients unfiltered.

​

Activation-local values

​

Callers