SQL projections

Keep tables in your own database from A2's logs with a consumer, so lists, filters, and joins are SQL queries.

The tables

db/schema.ts

import { bigint, pgTable, primaryKey, text } from 'drizzle-orm/pg-core'

/** The order list: one row per order. */
export const orderList = pgTable('order_list', {
  orderId: text('order_id').primaryKey(),
  shopId: text('shop_id').notNull(),
  status: text('status').notNull(),
})

/** The last event each consumer applied from each session's log. */
export const positions = pgTable(
  'positions',
  {
    consumer: text('consumer').notNull(),
    sessionId: text('session_id').notNull(),
    logIndex: bigint('log_index', { mode: 'number' }).notNull(),
  },
  (table) => [primaryKey({ columns: [table.consumer, table.sessionId] })],
)

db/client.ts

import { drizzle } from 'drizzle-orm/node-postgres'
import * as schema from '@/db/schema'

export const db = drizzle({
  connection: { connectionString: process.env.DATABASE_URL },
  schema,
})

A2 keeps each session's log. When people list, filter, and join across sessions, read tables in your database instead, kept from the logs by a consumer: a projection. This page keeps an order list with Drizzle on Postgres, as an example. The pattern is the same with any SQL database and client.

The order list has a row per order. positions has a row per consumer and session: the index of the last event that consumer applied to your tables. Your migrations create both, like any table. A2's own tables can live in the same database; see Postgres.

Why a position row

A2 keeps each consumer's cursor, its place in a session's log, in A2's store, and moves it after run returns. It fences its own writes by attempt: an append, a returned batch, or a cursor move from an attempt that lost its claim is refused. Your SQL commits in your own transaction, which A2 can't fence or make atomic with its cursor. Two things follow:

A crash between your commit and the cursor move retries the event. Your rows are in, the cursor isn't, and A2 runs the event again.
An attempt that lost its claim can still commit. Its claim lapsed while it ran, a newer attempt applied the event and the ones after it, and then the old attempt commits an older event over newer rows.

The position row fences both. It commits in the same transaction as your rows, and a transaction writes only when it moves the position forward, so each event lands in your tables once, in log order.

Claim the event first

server/order-list.ts

import { eq, lt } from 'drizzle-orm'
import { consumer } from 'experimental-a2/server'
import { orders } from '@/contracts'
import { db } from '@/db/client'
import { orderList, positions } from '@/db/schema'

const name = 'order-list-v1'

export const orderListTable = consumer({
  contract: orders,
  name,
  lane: 'order-list',
  backfill: true,
}).run(async ({ event }) => {
  await db.transaction(async (tx) => {
    const claimed = await tx
      .insert(positions)
      .values({ consumer: name, sessionId: event.sessionId, logIndex: event.index })
      .onConflictDoUpdate({
        target: [positions.consumer, positions.sessionId],
        set: { logIndex: event.index },
        setWhere: lt(positions.logIndex, event.index),
      })
      .returning({ logIndex: positions.logIndex })
    // this event, or a later one, is already in:
    if (claimed.length === 0) return

    if (event.type === 'created') {
      await tx.insert(orderList).values({
        orderId: event.sessionId,
        shopId: event.payload.shopId,
        status: 'created',
      })
    } else {
      await tx
        .update(orderList)
        .set({ status: event.type })
        .where(eq(orderList.orderId, event.sessionId))
    }
  })
})

The transaction's first statement claims the event. It inserts the session's position, or moves it to this event's index if it's behind, and returns the row when it did. No row back means this event, or a later one, is already in your tables: return, and the transaction changes nothing. Otherwise write your rows. They commit with the position, or none of it does.

In plain SQL, it's the same statement on Postgres and SQLite:

insert into positions (consumer, session_id, log_index)
values ($1, $2, $3)
on conflict (consumer, session_id) do update
  set log_index = excluded.log_index
  where positions.log_index < excluded.log_index
returning 1

It comes first so a duplicate waits before it does any work. On Postgres, at the default read committed level, two transactions that claim the same event can't both get the row back: the second one's insert waits for the first to commit, then checks the condition again against the committed position, finds it caught up, and returns nothing. If the first rolls back, the second claims the event. Reading the position first with select ... for update doesn't cover a session's first event: there's no row to lock yet, so both transactions read nothing and both apply it.

The consumer

server/orders.ts

import { createServer } from 'experimental-a2/server'
import { postgres } from 'experimental-a2/store-postgres'
import { orders } from '@/contracts'
import { orderListTable } from '@/server/order-list'

export const ordersServer = createServer({
  contract: orders,
  store: postgres({ connectionString: process.env.DATABASE_URL }),
  consumers: [orderListTable],
})

A string lane runs a session's events in order. One at a time, in log order, one transaction each. Sessions run independently.
backfill fills it from history. It reads each session's log from its first event, not only events appended after it exists. A session starts on its next append, or when you consume it. Nothing enumerates sessions, so to fill the table for orders that never change again, consume each one.
A rebuild is a new name. See Rebuild it.

Read it

server/order-list.ts

export function shopOrders(shopId: string) {
  return db.select().from(orderList).where(eq(orderList.shopId, shopId))
}

Lists, filters, and joins are queries against your tables.

Read your writes

server/place-order.ts

import { orderListTable, shopOrders } from '@/server/order-list'
import { ordersServer } from '@/server/orders'

export async function placeOrder(orderId: string, shopId: string) {
  const order = ordersServer.session(orderId)
  const [created] = await order.append({
    type: 'created',
    payload: { shopId, items: [] },
  })
  await order.consume(orderListTable, { through: created.index })
  // the new order is in the list:
  return shopOrders(shopId)
}

Your tables lag the log a little. Before a read that has to include a write you just made, consume the projection through the index the append returned. It returns once your tables have that event.

A consumer that reads your tables waits the same way:

server/receipts.ts

import { consumer } from 'experimental-a2/server'
import { orders } from '@/contracts'
import { orderListTable } from '@/server/order-list'

export const receipts = consumer({
  contract: orders,
  name: 'receipts',
  types: ['shop.notified'],
}).run(async ({ event, session }) => {
  await session.consume(orderListTable, { through: event.index })
  // the order list has this event; read it:
  // const [order] = await db.select().from(orderList).where(eq(orderList.orderId, event.sessionId))
})

Pass event.index - 1 instead when the consumer applies the triggering event itself and needs your tables only up to it. A consumer can't consume itself.

consume takes the session's lock in A2's store on every call, even when your tables are caught up, and while another invocation holds the next event, it polls until that one finishes. On a hot path, read your position row first, and consume only when it's behind the index you need:

server/read-through.ts

import { and, eq } from 'drizzle-orm'
import { db } from '@/db/client'
import { positions } from '@/db/schema'
import { orderListTable } from '@/server/order-list'
import { ordersServer } from '@/server/orders'

export async function readThrough(orderId: string, index: number) {
  const [position] = await db
    .select({ logIndex: positions.logIndex })
    .from(positions)
    .where(
      and(
        eq(positions.consumer, orderListTable.name),
        eq(positions.sessionId, orderId),
      ),
    )
  if ((position?.logIndex ?? 0) < index) {
    await ordersServer.session(orderId).consume(orderListTable, { through: index })
  }
}

Live views from the table

server/order-view.ts

import { and, eq } from 'drizzle-orm'
import { db } from '@/db/client'
import { orderList, positions } from '@/db/schema'
import { orderListTable } from '@/server/order-list'

/** One order's row, and the log index it reflects. */
export function orderView(orderId: string) {
  return db.transaction(
    async (tx) => {
      const [position] = await tx
        .select({ logIndex: positions.logIndex })
        .from(positions)
        .where(
          and(
            eq(positions.consumer, orderListTable.name),
            eq(positions.sessionId, orderId),
          ),
        )
      const [order] = await tx
        .select()
        .from(orderList)
        .where(eq(orderList.orderId, orderId))
      return { order, through: position?.logIndex ?? 0 }
    },
    { isolationLevel: 'repeatable read' },
  )
}

A view of one session can load its rows from your tables and stay live with the session's stream. Read the rows and the position in one snapshot, a repeatable read transaction on Postgres, and return them together. The client follows the stream and applies only the events after through, so it neither misses an event nor applies one twice.

SQLite

server/order-list-sqlite.ts

import { DatabaseSync } from 'node:sqlite'
import { consumer } from 'experimental-a2/server'
import { orders } from '@/contracts'

// your migrations create order_list and positions in this file:
const db = new DatabaseSync('.a2/app.db')
const name = 'order-list-v1'

const claim = db.prepare(`
  insert into positions (consumer, session_id, log_index)
  values (?, ?, ?)
  on conflict (consumer, session_id) do update
    set log_index = excluded.log_index
    where positions.log_index < excluded.log_index
  returning 1
`)
const insertOrder = db.prepare(
  'insert into order_list (order_id, shop_id, status) values (?, ?, ?)',
)
const setStatus = db.prepare(
  'update order_list set status = ? where order_id = ?',
)

export const orderListTable = consumer({
  contract: orders,
  name,
  lane: 'order-list',
  backfill: true,
}).run(async ({ event }) => {
  // synchronous from begin to commit, with nothing awaited in between:
  db.exec('begin immediate')
  try {
    if (claim.get(name, event.sessionId, event.index) !== undefined) {
      if (event.type === 'created') {
        insertOrder.run(event.sessionId, event.payload.shopId, 'created')
      } else {
        setStatus.run(event.type, event.sessionId)
      }
    }
    db.exec('commit')
  } catch (error) {
    db.exec('rollback')
    throw error
  }
})

The claim is the same statement. Run the transaction synchronously, from begin immediate to commit, with nothing awaited in between: node:sqlite's DatabaseSync runs each statement on the calling thread, so nothing else runs in the middle, and two of your own transactions never interleave on one handle.

It's required when your tables share a file with A2's SQLite store. A2 writes on the same thread, through its own handle, and SQLite lets one writer hold a file at a time. If your transaction awaited while it held the file, an A2 write to it would block the thread on A2's five-second busy timeout, and your transaction couldn't reach commit until that write gave up.

Throughput and lag

A string lane runs a session's events one at a time, one transaction each, so a projection keeps up with a session as long as its transactions outpace that session's appends. In the playground's a2slack drills, with Redis and Postgres on the same machine, the feed projection, which keeps a whole workspace in one session, ran an event in about 0.4 ms and rebuilt at about 1,800 events a second. Posting 10,000 messages in under a second put it up to about six seconds behind. A remote database adds at least a round trip to every event.

Watch it with each a2.event span's a2.event.lag, how many events the session had appended after the one it claimed, and a2.event.age_ms, how old that event was when its run started, grouped by a2.event.consumer. See Telemetry. When a session's appends keep outrunning its projection, keep each transaction to the claim and your writes, and do slower work, like calling an API, in a consumer of its own.

Rebuild it

To change what a projection writes, or to fill it again, give its consumer a new name, such as order-list-v2. A new name is a new consumer: it has no cursors in A2 and no rows in positions, and backfill fills it from each session's first event, as each session next runs.

Point it at new tables, created by the migration that ships it, and switch your reads to them once it has caught up. A deployment still running order-list-v1 keeps writing to the old ones until it's gone. To reuse the old tables instead, truncate them, along with the old name's rows in positions, in a migration that runs once no deployment runs the old name: an earlier deployment that still does keeps writing to tables you emptied.

Point in time

The order list holds each order as of its latest event. To read one as of an earlier index, such as next to an old event, keep a versions table too: a row per event, keyed by session and log index, written in the same transaction as the rest. The row as of index N is the newest at or below it:

select status from order_versions
where order_id = $1 and log_index <= $2
order by log_index desc
limit 1