​

Drizzle integration

​

Declaring an owned table

import { Actor } from "@rikalabs/akter"
import { index, pgTable, text, timestamp } from "drizzle-orm/pg-core"
import { Schema } from "effect"

const RoomId = Schema.String
const Post = Actor.command("Post", { payload: Schema.Struct({ body: Schema.String }) })
const Recent = Actor.query("Recent", { success: Schema.Array(Schema.String) })

export const messages = Actor.table(
  pgTable(
    "chat_messages",
    {
      id: text("id").primaryKey(),
      author: text("author").notNull(),
      body: text("body").notNull(),
      sentAt: timestamp("sent_at", { withTimezone: true }).notNull(),
    },
    (table) => [index("chat_messages_sent").on(table.sentAt)],
  ),
)

export const Room = Actor.make("Room", { key: RoomId, tables: [messages], api: { Post, Recent } })

​

Adopting an existing table

import { Actor } from "@rikalabs/akter"
import { pgTable, text } from "drizzle-orm/pg-core"

const existingInvoices = pgTable("invoices", {
  id: text("id").primaryKey(),
  orgId: text("org_id").notNull(),
  accountId: text("account_id").notNull(),
})
const existingContacts = pgTable("contacts", {
  id: text("id").primaryKey(),
  tenant: text("tenant").notNull(),
  owner: text("owner").notNull(),
})

const InvoiceRows = Actor.table(existingInvoices, {
  owner: { tenant: existingInvoices.orgId, actor: existingInvoices.accountId },
})

const Contacts = Actor.table(existingContacts, {
  owner: { tenant: existingContacts.tenant, actor: existingContacts.owner },
  access: "read",
})

​

Scoped rows

import { Effect } from "effect"
import { messages, Room } from "./room.ts"

declare const id: string
declare const author: string
declare const body: string
declare const sentAt: Date

const inATurn = Effect.gen(function* () {
  const turn = yield* Room.Turn
  yield* turn.rows(messages).insert({ id, author, body, sentAt })
  yield* turn.rows(messages).update({ body }).where({ id })
  yield* turn.rows(messages).delete().where({ id })
  yield* turn.rows(messages).upsert({ id, author, body, sentAt })
})

const inAQuery = Effect.gen(function* () {
  const read = yield* Room.Read
  const recent = yield* read.rows(messages).all({ orderBy: { sentAt: "desc" }, limit: 20 })
  const one = yield* read.rows(messages).one({ where: { id } }) // Option<Row>
  const total = yield* read.rows(messages).count({ where: { author } })
})

• ownership columns in insert values, update values, upsert values, filters, or orderBy;
• unknown columns, RAW filters, and SQL values (sql, columns, subqueries, placeholders) in values or filters;
• rows(table) for a table the actor type does not list;
• use of a rows or group capability after its turn or query ended (including from a fiber forked during it), or from another actor’s turn;
• mutation methods on read.rows, and anything other than select on group.

​

Placement group reads

import { Actor } from "@rikalabs/akter"
import { eq, inArray } from "drizzle-orm"
import { pgTable, text } from "drizzle-orm/pg-core"
import { Effect, Schema } from "effect"

const notes = Actor.table(pgTable("notes", { id: text("id").primaryKey() }))
const labels = Actor.table(
  pgTable("labels", { noteId: text("note_id").primaryKey(), label: text("label").notNull() }),
)
const Library = Actor.make("Library", { key: Schema.String, tables: [notes, labels], api: {} })
declare const ids: ReadonlyArray<string>

const inAQuery = Effect.gen(function* () {
  const read = yield* Library.Read
  const rows = yield* read.group((db) =>
    db
      .select({ note: notes.id, label: labels.label })
      .from(notes)
      .leftJoin(labels, eq(labels.noteId, notes.id))
      .where(inArray(notes.id, ids))
      .orderBy(notes.id),
  )
})

​

Fleet views

import { Actor, Fleet } from "@rikalabs/akter"
import { Actors } from "@rikalabs/akter/runtime"
import { Effect } from "effect"
import { bigint, boolean, pgTable, text } from "drizzle-orm/pg-core"

const OrderRows = Actor.table(
  pgTable("orders", {
    id: text("id").primaryKey(),
    status: text("status").notNull(),
    amountCents: bigint("amount_cents", { mode: "number" }).notNull(),
    archived: boolean("archived").notNull(),
  }),
)
const authorize = () => Effect.succeed(true)

const OrdersByStatus = Fleet.view("OrdersByStatus", {
  from: OrderRows,
  where: { archived: false },
  groupBy: ["status"],
  select: { orders: Fleet.count(), total: Fleet.sum("amountCents") },
})

Actors.layer({ authorize, fleet: [OrdersByStatus] })

• from may be an adopted table (ADR 0054) with access: "write": the view reads its mapped tenant column and its routing_key, so rows are counted once backfilled, and legacy writers’ changes reach the view through the WAL. A read-only adoption has no routing_key and is refused.
• where is the scoped-read filter algebra; groupBy names business columns, which must be NOT NULL; select names aggregates. Fleet.count() and Fleet.sum(column) are bigint (a sum takes an integer column, and a sum past the bigint range poisons the view), Fleet.avg(column) is double precision over a numeric column, and Fleet.min and Fleet.max keep the column’s type and are null when every value is. Joins, SQL, and cross-tenant views are not offered.
• OrdersByStatus.table is the derived Drizzle table, fleet_<snake_case name> in the source’s schema: tenant_id, the group columns, the aggregates, and as_of (the change-feed position of the batch that last wrote the row), keyed by (tenant_id, …group columns) and carrying the durable_tenant policy. Put it in the drizzle-kit schema. The maintainer is its only writer; plain SQL may read it, outside authorize.
• The source needs an index leading with (routing_key, tenant_id, …group columns), so recomputing a group is one indexed read. Actor.table prefixes its own indexes with actor_id as well, so create this one with a hand-written migration; startup refuses without it and prints the CREATE INDEX.
• A view is eventually consistent: it reflects committed changes only, each group at or after its last applied change. actor_fleet_views shows each view’s status (building, ready, or stale), applied_lsn (comparable with a command’s durable-version: once applied_lsn is at least the token, the view has seen that command), and last_error.
• Fleet.subscribe(View, filter?, { limit? }) follows a view in process for the ambient tenant and caller: a Stream of pages { asOf, stale, rows }, the tenant’s groups in group-key order (at most limit, default 100, up to 1,000), first at once and again whenever the view’s rows or staleness change; asOf alone moving on sends nothing. filter compares group keys with each column’s text form. Actors.serve({ fleet: [View] }) serves it at GET /fleet/{View}?{groupKey}=value&limit= as server-sent events, and fleetClient([View], { baseUrl, headers }) from @rikalabs/akter/client reads it as client.View.subscribe(filter?, options?), an AsyncIterable. A subscription is authorized with kind: "fleet" and command the view, its ref naming the caller’s tenant and the view’s source actor type with the view as its id, so the source’s access policy and the authorize hook both apply, and again every reauthorizeEvery of the source’s actor type with kind: "reauthorize", of: "fleet". Its tenant is always the caller’s: no parameter names one. One runner holds at most 1,000 subscriptions of a view (RunnerAtCapacity). With row-level security on, the page runs as the tenant role, which must be able to read the derived table.
• Views need Postgres with wal_level=logical, a login with REPLICATION, and akter fleet setup; see deployment. PGlite refuses fleet at layer build.

​

Transactions and backends