Consume changes into PostgreSQL

This example copies USDC/WETH processor changes into PostgreSQL. It commits each change and its cursor in one database transaction, then acknowledges Leani. Restarting the consumer resumes from its acknowledgement without duplicating committed rows.

Start the matching Leani node

Follow query once, then follow through the backfill and serve commands. Keep that node running at http://127.0.0.1:8080. It must contain the usdc-weth-latest instance with delivery enabled. The earlier windowed.toml event example has delivery disabled and cannot be used for this consumer.

You also need Bun and the local SDK setup from Install Leani.

Prepare PostgreSQL

If you have PostgreSQL running, create an empty leani_example database and set DATABASE_URL to its connection string. Alternatively, with Docker running:

docker run --name leani-example-postgres \
  -e POSTGRES_PASSWORD=leani-example \
  -e POSTGRES_DB=leani_example \
  -p 127.0.0.1:54329:5432 -d postgres:17-alpine
docker exec leani-example-postgres pg_isready -U postgres -d leani_example
export DATABASE_URL='postgres://postgres:[email protected]:54329/leani_example'

Wait until pg_isready reports that it is accepting connections. The container is an isolated example database; use your own credentials for other deployments.

Run the consumer

In the terminal with DATABASE_URL, run:

cd "$LEANI_SOURCE"
export LEANI_URL=http://127.0.0.1:8080
export LEANI_PROCESSOR=usdc-weth-latest
export LEANI_CONSUMER=postgres-example
export LEANI_CONSUMER_CREDENTIAL="$(openssl rand -hex 32)"
bun examples/sdk-postgres/index.ts

The application creates its tables and registers the named consumer with the secret you supplied. Registration does not generate this secret. Preserve it when restarting or moving to another terminal; do not generate a different secret for the same consumer identity.

Expect Consuming usdc-weth-latest into PostgreSQL; press Ctrl-C to stop. The historical example then drains the retained changes and waits for new ones. The runnable application consists of examples/sdk-postgres/index.ts and its imported helper, examples/sdk-subscription/index.ts.

Inspect the result and restart

Using psql with your connection string:

psql "$DATABASE_URL" -c 'select count(*), max(sequence) from leani_applied_changes'
psql "$DATABASE_URL" -c 'select processor, network, sequence from leani_cursors'
curl -s http://127.0.0.1:8080/v1/processors/usdc-weth-latest/consumers/postgres-example

For the 1,000-block example, the checked run committed 491 changes. PostgreSQL’s highest sequence should match Leani’s acknowledgedSequence, with lagChanges zero after draining. A change is a processor mutation or finality transition; it is not necessarily one row per Ethereum block.

Press Ctrl-C in the consumer terminal. Run bun examples/sdk-postgres/index.ts again in that same terminal, keeping the exported identity and credential. The row count and cursor should stay unchanged until new changes arrive.

Keep the transaction boundary

The destination implementation is:

Atomic PostgreSQL destination

class PostgresDestination implements Destination<unknown> {
  constructor(
    private readonly sql: SQL,
    private readonly processor: string,
    private readonly network: string,
  ) {}

  async migrate(): Promise<void> {
    await this.sql`
      create table if not exists leani_applied_changes (
        processor text not null,
        network text not null,
        sequence numeric(20, 0) not null,
        cursor text not null,
        operation text not null,
        kind text not null,
        entity_key text,
        payload jsonb,
        block_number bigint,
        primary key (processor, network, sequence)
      )
    `;
    await this.sql`
      create table if not exists leani_cursors (
        processor text not null,
        network text not null,
        cursor text not null,
        sequence numeric(20, 0) not null,
        updated_at timestamptz not null default now(),
        primary key (processor, network)
      )
    `;
  }

  async transaction(
    work: (transaction: DestinationTransaction<unknown>) => Promise<void>,
  ): Promise<void> {
    await this.sql.begin(async (sql) => {
      await work({
        apply: async (change: ChangeEnvelope<unknown>) => {
          await sql`
            insert into leani_applied_changes (
              processor, network, sequence, cursor, operation, kind,
              entity_key, payload, block_number
            ) values (
              ${this.processor}, ${this.network}, ${change.sequence},
              ${change.cursor}, ${change.operation}, ${change.kind},
              ${change.key}, ${JSON.stringify(change.data)}::jsonb,
              ${change.block?.number ?? null}
            )
            on conflict (processor, network, sequence) do nothing
          `;
        },
        storeLeaniCursor: async (cursor: string, sequence: string) => {
          await sql`
            insert into leani_cursors (
              processor, network, cursor, sequence, updated_at
            ) values (
              ${this.processor}, ${this.network}, ${cursor}, ${sequence}, now()
            )
            on conflict (processor, network) do update set
              cursor = excluded.cursor,
              sequence = excluded.sequence,
              updated_at = excluded.updated_at
            where leani_cursors.sequence <= excluded.sequence
          `;
        },
      });
    });
  }
}

Apply an idempotent change and persist its opaque cursor in one PostgreSQL transaction.Verified source

The (processor, network, sequence) key makes replay idempotent. The cursor is stored in the same transaction as the event. A failed acknowledgement after commit can cause redelivery; a committed event must remain safe to apply again. This example stores a change log. When building domain tables, apply both forward and inverse mutations to keep those tables correct through reorgs.

Retention remains a node policy

The query/follow example uses a bounded change window. A consumer that falls behind that window must rebuild. For a destination that must prevent pruning until it acknowledges, use delivery.mode = "until_acknowledged" with a configured required consumer; see operating profiles and durable delivery.

The reusable consumer loop explains leases, reset handling, and commit-before-ack in detail. Stop this example’s Docker database when finished with docker stop leani-example-postgres.