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

## The tables

```ts 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] })],
)
```

```ts 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](/server/consumers): 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](/stores/postgres#with-drizzle).

## 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

```ts 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:

```sql
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

```ts 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](#read-your-writes) 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](#rebuild-it).

## Read it

```ts 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

```ts 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](/server/consumers#run-a-consumer-now-with-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:

```ts 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:

```ts 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

```ts 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](/client/react). 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

```ts 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](/stores/sqlite). 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](/reference/telemetry#the-span-catalogue). 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:

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