Querying Indexed Data

View as Markdown

Use this page when the indexer is already writing rows and you want to consume them from your app or internal tooling. This page covers app-owned queries over tables created by @thru/indexer; package-level replay behavior lives in the @thru/replay reference.

Direct Table Queries

Export the stream tables from your schema package, then query them with Drizzle like any other app table.

import { desc, eq } from "drizzle-orm";
import { db } from "./db";
import { tokenAccountsTable, tokenTransferEvents } from "./schema";

const recentTransfers = await db
  .select()
  .from(tokenTransferEvents)
  .where(eq(tokenTransferEvents.dest, "ta..."))
  .orderBy(desc(tokenTransferEvents.slot))
  .limit(20);

const ownerBalances = await db
  .select()
  .from(tokenAccountsTable)
  .where(eq(tokenAccountsTable.owner, "ta..."));

Query Shape

Keep product-specific query logic in your backend, not inside stream definitions. Stream definitions decide what rows exist. Repository functions and routes decide which rows a user can see, how they are joined, and how response objects are shaped.

import { desc, eq, or } from "drizzle-orm";
import { db } from "./db";
import { tokenAccountsTable, tokenTransferEvents } from "./schema";

export async function getWalletActivity(accountAddress: string, limit = 25) {
  return await db
    .select({
      id: tokenTransferEvents.id,
      slot: tokenTransferEvents.slot,
      amount: tokenTransferEvents.amount,
      source: tokenTransferEvents.source,
      dest: tokenTransferEvents.dest,
      sourceAmount: tokenAccountsTable.amount,
    })
    .from(tokenTransferEvents)
    .leftJoin(
      tokenAccountsTable,
      eq(tokenTransferEvents.source, tokenAccountsTable.address),
    )
    .where(or(
      eq(tokenTransferEvents.source, accountAddress),
      eq(tokenTransferEvents.dest, accountAddress),
    ))
    .orderBy(desc(tokenTransferEvents.slot), desc(tokenTransferEvents.id))
    .limit(limit);
}

export async function getCurrentTokenAccounts(owner: string) {
  return await db
    .select()
    .from(tokenAccountsTable)
    .where(eq(tokenAccountsTable.owner, owner));
}

Cursor Pagination

For user-facing history, prefer cursor pagination over offset pagination. Slots can contain multiple rows, so use a stable tie-breaker such as the event row id or another unique column from your stream schema.

import { and, desc, eq, lt, or } from "drizzle-orm";
import { db } from "./db";
import { tokenTransferEvents } from "./schema";

interface TransferCursor {
  slot: bigint;
  id: string;
}

export async function listTransfersPage(options: {
  accountAddress: string;
  cursor?: TransferCursor;
  limit?: number;
}) {
  const limit = options.limit ?? 50;
  const accountFilter = or(
    eq(tokenTransferEvents.source, options.accountAddress),
    eq(tokenTransferEvents.dest, options.accountAddress),
  );
  const before = options.cursor
    ? and(
        accountFilter,
        or(
          lt(tokenTransferEvents.slot, options.cursor.slot),
          and(
            eq(tokenTransferEvents.slot, options.cursor.slot),
            lt(tokenTransferEvents.id, options.cursor.id),
          ),
        ),
      )
    : accountFilter;

  const rows = await db
    .select()
    .from(tokenTransferEvents)
    .where(before)
    .orderBy(desc(tokenTransferEvents.slot), desc(tokenTransferEvents.id))
    .limit(limit + 1);

  const page = rows.slice(0, limit);
  const next = rows.length > limit ? page.at(-1) : null;

  return {
    rows: page,
    nextCursor: next ? { slot: next.slot, id: next.id } : null,
  };
}

If your stream schema does not have a natural unique event ID, add one when you define the stream. Avoid paginating only by slot because that can skip or duplicate rows at slot boundaries.

Freshness

Indexed rows are eventually consistent with the chain endpoint used by the worker. If the API needs to expose freshness, read the stream checkpoint or runtime status from an internal endpoint and return the latest processed slot alongside the query response.

import { getCheckpoint } from "@thru/indexer";
import { db } from "./db";

export async function getTransfersWithFreshness(accountAddress: string) {
  const [rows, checkpoint] = await Promise.all([
    listTransfersPage({ accountAddress }),
    getCheckpoint(db, "token-transfers"),
  ]);

  return {
    ...rows,
    indexedThroughSlot: checkpoint?.slot ?? null,
  };
}

Event streams checkpoint under the stream name. Account streams checkpoint under account:${stream.name}.

When To Query Tables Directly

Direct queries are a good fit when:

the backend route is internal to your app
the query needs joins across multiple indexed tables
you need auth, aggregation, or business-specific response shapes
the result shape should evolve with your product rather than with stream definitions

Practical Recommendation

For most apps:

query tables directly inside the backend for core product endpoints
paginate by (slot, uniqueRowId) rather than offset
wrap common queries in small repository functions
keep auth, aggregation, and response shaping in application code
expose indexed freshness when stale reads matter to the product