Capsule Database

New Lakebed artifacts use database API v1. Database calls are async, queries require declared indexes, and every handler runs inside one parent-owned transaction.

Indexes

import { boolean, capsule, id, query, string, table } from "lakebed/server";

export default capsule({
  schema: {
    rooms: table({ name: string() }),
    users: table({ displayName: string() }),
    messages: table({
      roomId: id("rooms"),
      authorId: id("users"),
      body: string(),
      pinned: boolean().default(false)
    })
      .index("by_room", ["roomId"])
      .index("by_room_pinned", ["roomId", "pinned"])
      .index("by_author", ["authorId"])
  },
  queries: {
    recent: query(async (ctx, args: { roomId: string }) =>
      ctx.db.messages
        .withIndex("by_room", (q) => q.eq("roomId", args.roomId))
        .order("desc")
        .take(20)
    )
  }
});

Every index has an implicit createdAt, id suffix. This makes traversal deterministic even when declared fields and timestamps are equal. Ranges support an equality prefix followed by gt, gte, lt, or lte on the next index field. by_creation is available on every table.

Each declared index field has one schema-defined scalar type: string, boolean, number, id, or userId. IDs and user references are encoded as strings. Lakebed uses the same canonical byte encoding for those values in memory, source-runtime, and Postgres execution.

Field types

Chain .default(value) to fill a value when an insert omits the field. Chain .optional() to allow the field to be absent. The two chain in either order.

Numbers

scores: table({
  playerId: id("players"),
  points: number(),
  streak: number().default(0)
}).index("by_points", ["points"])

number() stores a JavaScript number. Index order is numeric, so -10 sorts before 2 and 2 before 10. Range clauses compare numerically: q.gte("points", 100) returns rows with points of 100 or more.

Store timestamps with number() as Unix epoch milliseconds. Use Date.now() for the current time and Date.parse(value) for an external timestamp, then reject a non-finite result. Numeric index ranges compare times without depending on string formats or time zone offsets.

Optional fields

todos: table({
  text: string(),
  dueAt: number().optional(),
  priority: number().optional()
}).index("by_priority", ["priority"])

An optional field can be left out of insert. The stored row has no key for it, so row.priority reads as undefined and the row type is priority?: number. Pass null to insert or update to mean "no value". An update with null removes the field from the row. Stored rows never contain null.

An optional field with a default always has a value. number().optional().default(0) reads as number, and clearing it with null restores the default.

Sort rules

Within one index field, values sort in this order:

1. Absent (an optional field with no value). Absent sorts first in asc and last in desc.
2. false, then true.
3. Strings by UTF-8 byte order.
4. Numbers by numeric value. -0 and 0 are the same key.

To find rows where an optional field is absent, use q.eq("priority", null). Range clauses such as gt and lte only match present values.

Schema changes with existing rows

Adding a required field to a table that already has rows fails on deploy with Missing value for <table>.<field>. Make the new field .optional() or give it a .default(value) instead. Both work without rewriting stored rows, and both can be indexed right away.

Collect or paginate

count() returns the number of matching index entries without loading row bodies:

const count = await ctx.db.todos.withIndex("by_creation", q => q.gte("createdAt", since)).count();

Each match counts as one row read. The call uses one scan and does not use the rows returned or bytes read budgets. The shared handler limit of 5000 rows read still applies, including across several count() calls. A count that exceeds the remaining budget throws instead of returning a partial count. Use stored summary rows for larger ranges.

collect() returns every matching row if the handler stays within its database limits. The default limit is 1000 rows returned by database calls per handler. collect() throws collect() exceeded the 1000 row return limit. Use paginate(). when one call finds too many rows. Several calls can throw Database rowsReturned limit exceeded (1000). when their combined result exceeds the budget. Both failures roll back the transaction. Use collect() only when your app keeps the result set below that limit.

For a growing result set, use take(n) for a fixed window or paginate() for a scrolling list. The default limits also allow at most 5000 rows read and 4 MiB read per handler, so an unindexed collect() can fail before it reaches 1000 returned rows. See limits.md for the other resource limits.

first() returns the first matching row, or null when no row matches. Use .order("desc").first() for the latest row in an index range. take(n) returns at most n rows. Both use the same database read budgets as collect().

Several paginate() calls share the row return budget within one handler. A pagination loop inside one handler does not bypass it. Paginate across separate calls for a growing list. Neither take(n) nor one page gives an exact count of rows outside that window. See dashboard counts for totals stored during ingest.

Pagination

paginate({ cursor, numItems }) uses an authenticated keyset cursor. A cursor is tied to the deploy, schema, query arguments, index, range, and order. Each page has a repeatable snapshot; the cursor does not preserve one historical snapshot across separate requests.

const result = await ctx.db.messages
  .withIndex("by_room", (q) => q.eq("roomId", args.roomId))
  .order("desc")
  .paginate(args.pagination);

On the client, usePaginatedQuery(name, args) supplies a trailing pagination argument, keeps every loaded page subscribed, accumulates results, and resets to the first page after reactive invalidation.

Dashboard counts

For totals larger than one handler can read, store a count per UTC day alongside the source rows. Insert the source row and increment its count in the same writable handler. Writes on one deploy are serialized, so an indexed lookup followed by an insert is safe from concurrent ingest calls. Indexes do not enforce uniqueness. All writers must use this check.

This server fetches one page of closed GitHub PRs, keeps merged PRs, and updates daily counts. It uses the existing API. No data passes through a browser.

import { capsule, endpoint, json, number, query, table, text } from "lakebed/server";

const DAY = 86_400_000;
const PAGE_SIZE = 25;

export default capsule({
  auth: { requireSignIn: false },
  schema: {
    prs: table({ number: number(), mergedAt: number() })
      .index("by_number", ["number"])
      .index("by_merged", ["mergedAt"]),
    days: table({ day: number(), count: number() }).index("by_day", ["day"])
  },
  endpoints: {
    ingest: endpoint({ method: "POST", path: "/api/ingest", readOnly: false }, async (ctx, req) => {
      const secret = ctx.env.INGEST_SECRET;
      if (!secret || req.headers.get("authorization") !== `Bearer ${secret}`) {
        return text("unauthorized", { status: 401 });
      }
      const page = Number(req.query.get("page") ?? "1");
      if (!Number.isSafeInteger(page) || page < 1) {
        return text("invalid page", { status: 400 });
      }
      const token = ctx.env.GITHUB_TOKEN;
      if (!token) throw new Error("Set GITHUB_TOKEN in .env.lakebed.server.");
      const response = await fetch(
        `https://api.github.com/repos/pingdotgg/t3code/pulls?state=closed&sort=updated&direction=desc&per_page=${PAGE_SIZE}&page=${page}`,
        { headers: { authorization: `Bearer ${token}`, accept: "application/vnd.github+json" } }
      );
      if (!response.ok) throw new Error(`GitHub returned ${response.status}.`);
      const items: unknown = await response.json();
      if (!Array.isArray(items) || items.length > PAGE_SIZE) throw new Error("Invalid GitHub page.");
      let inserted = 0;
      for (const value of items) {
        const item: unknown = value;
        if (typeof item !== "object" || item === null || !("number" in item) || !("merged_at" in item)) {
          throw new Error("Invalid GitHub PR.");
        }
        if (item.merged_at === null) continue;
        if (typeof item.number !== "number" || !Number.isSafeInteger(item.number) ||
            typeof item.merged_at !== "string") throw new Error("Invalid merged PR.");
        const mergedAt = Date.parse(item.merged_at);
        if (!Number.isFinite(mergedAt)) throw new Error("Invalid merge time.");
        const existing = await ctx.db.prs.withIndex("by_number", (q) => q.eq("number", item.number)).first();
        if (existing) continue;
        const day = Math.floor(mergedAt / DAY) * DAY;
        const bucket = await ctx.db.days.withIndex("by_day", (q) => q.eq("day", day)).first();
        await ctx.db.prs.insert({ number: item.number, mergedAt });
        if (bucket) {
          if (typeof bucket.count !== "number") throw new Error("Invalid daily count.");
          await ctx.db.days.update(bucket.id, { count: bucket.count + 1 });
        } else {
          await ctx.db.days.insert({ day, count: 1 });
        }
        inserted += 1;
      }
      ctx.log.info("GitHub ingest", { page, inserted });
      return json({ inserted, nextPage: items.length === PAGE_SIZE ? page + 1 : null });
    })
  },
  queries: {
    daily: query(async (ctx, start: number) => {
      if (!Number.isSafeInteger(start) || start % DAY !== 0) throw new Error("Use a UTC day boundary.");
      return ctx.db.days
        .withIndex("by_day", (q) => q.gte("day", start).lt("day", start + 14 * DAY))
        .take(14);
    })
  }
});

Set INGEST_SECRET and GITHUB_TOKEN in .env.lakebed.server, keep the file out of Git, and claim and redeploy before hosted use. Keep the ingest secret in the external caller too. Call POST /api/ingest?page=1 with Authorization: Bearer <INGEST_SECRET>, then follow nextPage across separate requests. Retry failed pages with backoff. Repeating a page does not count a PR twice. Each call uses at most two index scans and two writes per PR, so this batch size leaves room within the handler limits.

For the client, pass a UTC midnight timestamp to client.useQuery("daily", start). Fill missing days with zero and draw the returned counts. Sum complete day buckets for a phase that starts at midnight. For a phase boundary within a day, query source rows for the partial day. Use a separate paginated query on by_merged for recent PR details. Update time-based query arguments when the visible window changes, because the clock alone does not invalidate a subscription.

This is a batching and count example, not a complete GitHub mirror. GitHub pages can move while ingestion runs. Revisit overlapping pages, periodically reconcile the full retained window, and record the last successful refresh for the UI. Do not stop at the first old merge when sorting by update time. For a first backfill, the external caller must continue across pages. Lakebed has no durable continuation queue or built-in scheduler yet.

This example keeps only PR numbers and merge times and does not prune. Add bounded retention work before running it indefinitely. Delete old source rows and their expired count buckets in the same transaction. Keep buckets needed for historic rates. Rolling windows with partial days require source rows for those days, or finer buckets. A query cannot read unlimited rows by selecting fewer fields. There is no projection API. Subscriptions can share query execution, but there is no configurable time-based query cache. Each subscriber still uses the request quota.

Relationships

Use id("table") fields and bounded application-level traversal. Lakebed does not expose SQL joins.

const page = await ctx.db.messages
  .withIndex("by_room", (q) => q.eq("roomId", args.roomId))
  .take(20);

const enriched = [];
for (const message of page) {
  enriched.push({ ...message, author: await ctx.db.users.get(message.authorId) });
}

Many-to-many relationships use an explicit join table with indexes for both directions.

User references

Use userId() for a Lakebed user reference. Use id("table") for a row in an app table. These are different references, even if your app has a table named users.

import { string, table, userId } from "lakebed/server";

const todos = table({
  text: string(),
  ownerId: userId()
}).index("by_owner", ["ownerId"]);

userId() stores an opaque nonempty string. It marks the field for automatic transfer when a protected guest signs in. Plain strings do not get this behavior. Writes and queries store and match the value you give them. Do not write a retired guest ID after its upgrade. Use the caller's current userId.

The field does not authorize reads or writes. Use ctx.auth.requireIdentity() for a verified guest or account, filter by its userId, and check ownership before updates or deletes. Use ctx.auth.requireSignedIn() when guests must not use an operation.

Automatic transfer preserves existing account rows. Use auth.onGuestUpgrade for app-specific merge rules, such as keeping existing account settings. The hook and reference transfer run in one transaction. See the auth guide.

User references and upgrade hooks require database API v1. Rebuild and redeploy older capsules before using them. Changing a field from string() to userId() does not assign old shared guest data to a visitor.

Consistency

• Query handlers receive a read-only database and one repeatable snapshot.
• Mutations and endpoints commit row and index writes atomically.
• An endpoint with readOnly: true uses read-only database access and cannot write.
• Mutations read their own writes.
• Writes are serialized per deploy in v1.
• Worker failure, timeout, quota failure, or store errors roll back the session.

Migrating Legacy Code

Add async to handlers, add await to every database terminal, declare indexes, and replace:

• where(...).orderBy(...).limit(...).all() with withIndex(...).order(...).take(...)
• all() with collect() or paginate()
• synchronous get, insert, update, and delete with awaited calls

Existing immutable artifacts keep the legacy snapshot runtime. Redeploying compiles database API v1 and backfills durable index entries from the existing JSON rows without rewriting application data.

See the Database API v1 migration guide for complete before-and-after examples and a copy-paste prompt for migrating a capsule with an agent.