Skip to content

Streams

View as Markdown

Use this page when the indexer runtime is clear but you need to design or extend the actual streams.

Event Streams Vs Account Streams

Section titled “Event Streams Vs Account Streams”
Stream typeBest forData model
Event streamImmutable logs such as transfers, mints, fills, or lifecycle eventsAppend-only rows keyed by event identity
Account streamCurrent on-chain account state such as balances, configuration accounts, or inventoryCurrent-state rows keyed by account identity

Use an event stream when the chain emits the thing you want directly.

Use an account stream when the important answer is “what is the latest state of this account now?”

Event Stream Shape

Section titled “Event Stream Shape”

An event stream needs:

  • name
  • schema
  • filter or filterFactory
  • parse(event)

It can also define filterBatch(events, ctx) and onCommit(batch, ctx) hooks for event-only workflows that need database-aware filtering or side effects.

Token transfer example:

import { create } from "@bufbuild/protobuf";
import { decodeAddress, encodeAddress, encodeSignature } from "@thru/sdk/helpers";
import { defineEventStream, t } from "@thru/indexer";
import { FilterParamValueSchema, FilterSchema } from "@thru/replay";
import { TokenEvent } from "./abi/thru/program/token/types";


const tokenTransfers = defineEventStream({
  name: "token-transfers",
  schema: {
    id: t.text().primaryKey(),
    slot: t.bigint().notNull().index(),
    txnSignature: t.text().notNull(),
    source: t.text().notNull().index(),
    dest: t.text().notNull().index(),
    amount: t.bigint().notNull(),
  },
  filterFactory: () => {
    const programBytes = new Uint8Array(decodeAddress(process.env.TOKEN_PROGRAM_ID!));
    return create(FilterSchema, {
      expression: "event.program.value == params.address",
      params: {
        address: create(FilterParamValueSchema, {
          kind: { case: "bytesValue", value: programBytes },
        }),
      },
    });
  },
  parse: (event) => {
    if (!event.payload || event.slot === undefined) return null;
    const tokenEvent = TokenEvent.from_array(event.payload);
    const transfer = tokenEvent?.payload()?.asTransfer();
    if (!transfer) return null;


    return {
      id: event.eventId,
      slot: event.slot,
      txnSignature: encodeSignature(event.transactionSignature?.value ?? new Uint8Array()),
      source: encodeAddress(new Uint8Array(transfer.source.get_bytes())),
      dest: encodeAddress(new Uint8Array(transfer.dest.get_bytes())),
      amount: transfer.amount,
    };
  },
});

Account Stream Shape

Section titled “Account Stream Shape”

An account stream needs:

  • name
  • schema
  • ownerProgram or ownerProgramFactory
  • optional expectedSize or dataSizes
  • parse(account)

Token account example:

import { decodeAddress, encodeAddress } from "@thru/sdk/helpers";
import { defineAccountStream, t } from "@thru/indexer";
import { TokenAccount } from "@thru/programs/token";


const tokenAccounts = defineAccountStream({
  name: "token-accounts",
  ownerProgramFactory: () => new Uint8Array(decodeAddress(process.env.TOKEN_PROGRAM_ID!)),
  expectedSize: 73,
  schema: {
    address: t.text().primaryKey(),
    mint: t.text().notNull().index(),
    owner: t.text().notNull().index(),
    amount: t.bigint().notNull(),
    isFrozen: t.boolean().notNull(),
    slot: t.bigint().notNull(),
    seq: t.bigint().notNull(),
  },
  parse: (account) => {
    if (account.data.length !== 73) return null;
    const parsed = TokenAccount.from_array(account.data);
    if (!parsed) return null;
    return {
      address: encodeAddress(account.address),
      mint: encodeAddress(new Uint8Array(parsed.mint.get_bytes())),
      owner: encodeAddress(new Uint8Array(parsed.owner.get_bytes())),
      amount: parsed.amount,
      isFrozen: parsed.is_frozen !== 0,
      slot: account.slot,
      seq: account.seq,
    };
  },
});

Practical Rules

Section titled “Practical Rules”
  • Use filterFactory and ownerProgramFactory when values come from environment or config so migration tooling can still import the schema files safely.
  • Use expectedSize when the account layout is fixed and size mismatches should be skipped early.
  • Return null from parse when an event or non-delete account update should be ignored. Account delete updates are handled by the runtime before parsing and remove the row identified by api.idField or address.
  • Export the generated .table from every stream so Drizzle can include it in migrations.

Event Hooks

Section titled “Event Hooks”

Use filterBatch when a parsed event needs to be filtered against current database state before insert.

const tokenTransfers = defineEventStream({
  name: "token-transfers",
  schema,
  filterFactory,
  parse,
  filterBatch: async (events, { db }) => {
    const allowed = await loadAllowedAccounts(db);
    return events.filter(
      (event) => allowed.has(event.source) || allowed.has(event.dest)
    );
  },
});

If filterBatch returns an empty array, the runtime still checkpoints the batch so already-seen events are not replayed forever. If filterBatch throws, the stream fails and the supervisor restarts it from the last durable checkpoint.

Use onCommit for side effects after rows are inserted.

const tokenTransfers = defineEventStream({
  name: "token-transfers",
  schema,
  filterFactory,
  parse,
  onCommit: async (batch, { db }) => {
    await notifyTransferSubscribers(db, batch.events);
  },
});

onCommit receives only rows that were actually inserted. Duplicate rows skipped by onConflictDoNothing() are not passed to the hook. Hook failures are logged and counted, but they do not block indexing.

Account Stream Deletes

Section titled “Account Stream Deletes”

Account streams use slot-aware upserts for normal updates and delete rows for account delete updates. The delete key is taken from stream.api?.idField and defaults to address, so custom account-stream primary keys should set api.idField.

const tokenAccounts = defineAccountStream({
  name: "token-accounts",
  ownerProgramFactory,
  expectedSize: TOKEN_ACCOUNT_SIZE,
  schema,
  parse,
  api: { idField: "address" },
});

The account table must include a slot column for the runtime’s slot-aware upsert guard. Older account updates do not overwrite newer rows.

Next Steps

Section titled “Next Steps”