# Postgres API (Carbon 2.0)

> Every public type in carbon_core::postgres — processors, row types, CRUD traits, migrations, metadata columns and sqlx primitives.

`carbon_core::postgres` persists decoded accounts and instructions to PostgreSQL through [`sqlx`](https://docs.rs/sqlx/0.8). Use this page when you write a Postgres-backed processor, query the tables Carbon creates, or implement your own row type. For a walkthrough, see [Store data in Postgres](https://carbon.sevenlabs.org/guides/postgres.md).

The module is behind the `postgres` feature of `carbon-core`:

`Cargo.toml`:

```toml
[dependencies]
carbon-core = { version = "2.0.0", features = ["postgres"] }
# A generated decoder with its own typed tables:
carbon-meteora-dlmm-decoder = { version = "2.0.0", features = ["postgres"] }
sqlx = { version = "0.8.6", features = ["runtime-tokio-rustls", "postgres"] }
sqlx_migrator = { version = "0.17.0", features = ["postgres"] }
async-trait = "0.1.89"
```

> **Note: Not on docs.rs.** docs.rs builds `carbon-core` 2.0.0 with default features only, so the `postgres` module has no docs.rs page. The links below point to the source at `v2.0.0`. Run `cargo doc -p carbon-core --features postgres --open` for local API docs.

The module has five submodules:

| Module                      | Contents                                                                     |
| --------------------------- | ---------------------------------------------------------------------------- |
| [`processors`](#processors) | Four `Processor` implementations that upsert every decoded update.           |
| [`rows`](#rows)             | `AccountRow<T>` and `InstructionRow<T>` (JSONB tables) and their migrations. |
| [`operations`](#operations) | The `Table`, `Insert`, `Upsert`, `Delete` and `Lookup` traits.               |
| [`metadata`](#metadata)     | The `__`-prefixed metadata columns every row carries.                        |
| [`primitives`](#primitives) | sqlx newtypes for `Pubkey` and unsigned and 128-bit integers.                |

## Two storage models

Carbon offers two ways to store data. Pick one per pipeline.

|                    | Typed tables                                                                                                             | Generic JSONB tables                                                                                                                     |
| ------------------ | ------------------------------------------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------- |
| Processors         | [`PostgresAccountProcessor`](#postgresaccountprocessor), [`PostgresInstructionProcessor`](#postgresinstructionprocessor) | [`PostgresJsonAccountProcessor`](#postgresjsonaccountprocessor), [`PostgresJsonInstructionProcessor`](#postgresjsoninstructionprocessor) |
| Tables             | One per account type (`<name>_account`) and instruction type (`<name>_instruction`), with one column per field           | `accounts` and `instructions`, decoded body in a `JSONB` column                                                                          |
| Migrations         | `<Program>AccountsMigration`, `<Program>InstructionsMigration` from the decoder crate                                    | [`GenericAccountsMigration`](#genericaccountsmigration), [`GenericInstructionMigration`](#genericinstructionmigration)                   |
| Requires           | Decoder built with its `postgres` feature                                                                                | Decoded type implements `serde::Serialize` and `Deserialize` (the decoder's `serde` feature)                                             |
| Works with GraphQL | Yes, the decoder's generated `QueryRoot` reads these tables                                                              | No generated schema; write your own queries                                                                                              |

## Processors

`carbon_core::postgres::processors`. Source: [`processors.rs`](https://github.com/sevenlabs-hq/carbon/blob/v2.0.0/crates/core/src/postgres/processors.rs).

Every processor:

- calls `upsert` once per decoded update and returns its error, so a failed write is logged by the pipeline and counted in `carbon_updates_failed_total`; the pipeline keeps running;
- counts successes and failures and records the upsert duration in milliseconds. See [Postgres metrics](https://carbon.sevenlabs.org/reference/metrics.md) for the metric names;
- holds a `sqlx::PgPool` (cheap to clone; pass `pool.clone()` to each processor).

### `PostgresAccountProcessor`

Upserts each decoded account through a wrapper type `W` that knows how to write it.

```rust
pub struct PostgresAccountProcessor<T, W> { /* private fields */ }

impl<T, W> PostgresAccountProcessor<T, W> {
    pub fn new(pool: sqlx::PgPool) -> Self
}

impl<'a, T, W> Processor<AccountProcessorInputType<'a, T>> for PostgresAccountProcessor<T, W>
where
    T: Clone + Send + Sync + 'static,
    W: From<(T, AccountMetadata)> + Upsert + Send + 'static,
```

**Details.** For every update the processor clones `decoded_account.data` and the [`AccountMetadata`](https://carbon.sevenlabs.org/reference/processor-inputs.md), builds `W::from((data, metadata))` and calls `W::upsert(&pool)`. Lamports, owner and the raw account bytes are not passed to `W`.

| Type parameter | Description                                                                                                                        |
| -------------- | ---------------------------------------------------------------------------------------------------------------------------------- |
| `T`            | The decoder's account type, for example `MeteoraDlmmAccount`.                                                                      |
| `W`            | The row wrapper. Generated decoders provide `<Program>AccountWithMetadata`, which dispatches to one typed row per account variant. |

**Example.** Register a generated decoder's typed account tables:

```rust
.account(
    MeteoraDlmmDecoder,
    PostgresAccountProcessor::<MeteoraDlmmAccount, MeteoraDlmmAccountWithMetadata>::new(pool.clone()),
)
```

See [the complete program](#example-typed-tables) and [a hand-written wrapper](#example-custom-row).

**See also.** [`Upsert`](#upsert), [Generated decoder types](#generated-decoder-types), [`PostgresJsonAccountProcessor`](#postgresjsonaccountprocessor).

### `PostgresInstructionProcessor`

Upserts each decoded instruction through a wrapper type `W`.

```rust
pub struct PostgresInstructionProcessor<T, W> { /* private fields */ }

impl<T, W> PostgresInstructionProcessor<T, W> {
    pub fn new(pool: sqlx::PgPool) -> Self
}

impl<'a, T, W> Processor<InstructionProcessorInputType<'a, T>> for PostgresInstructionProcessor<T, W>
where
    T: Clone + Send + Sync + 'static,
    W: From<(T, InstructionMetadata, Vec<AccountMeta>)> + Upsert + Send + 'static,
```

**Details.** Builds `W::from((decoded_instruction, metadata, raw_instruction.accounts))` and calls `upsert`. The third tuple element is the instruction's full `Vec<solana_instruction::AccountMeta>`; generated rows store it in an `__accounts` `JSONB` column. Nested instructions are not passed to `W`; each inner instruction reaches the processor as its own update.

| Type parameter | Description                                                                      |
| -------------- | -------------------------------------------------------------------------------- |
| `T`            | The decoder's instruction enum, for example `MeteoraDlmmInstruction`.            |
| `W`            | The row wrapper, for example the generated `MeteoraDlmmInstructionWithMetadata`. |

**Example.**

```rust
.instruction(
    MeteoraDlmmDecoder,
    PostgresInstructionProcessor::<MeteoraDlmmInstruction, MeteoraDlmmInstructionWithMetadata>::new(pool.clone()),
)
```

**See also.** [Instruction primary key](#instruction-primary-key), [`InstructionRowMetadata`](#instructionrowmetadata).

### `PostgresJsonAccountProcessor`

Upserts each decoded account into the generic `accounts` table as JSONB.

```rust
pub struct PostgresJsonAccountProcessor<T> { /* private fields */ }

impl<T> PostgresJsonAccountProcessor<T> {
    pub fn new(pool: sqlx::PgPool) -> Self
}

impl<'a, T> Processor<AccountProcessorInputType<'a, T>> for PostgresJsonAccountProcessor<T>
where
    T: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone + Send + Sync + Unpin + 'static,
```

**Details.** Builds an [`AccountRow<T>`](#accountrow) and calls its `upsert`, which runs `INSERT INTO accounts (__pubkey, __slot, data) … ON CONFLICT (__pubkey) DO UPDATE`. Each pubkey keeps one row holding its latest state; the update with the higher slot does not win automatically, the last one processed does. Create the table first with [`GenericAccountsMigration`](#genericaccountsmigration). Generated account enums serialize as `{"type": "<Variant>", "data": {…}}`, so filter with `data->>'type' = 'LbPair'`. Pubkey fields inside the JSON are arrays of 32 numbers.

**Example.**

```rust
.account(
    TokenProgramDecoder,
    PostgresJsonAccountProcessor::<TokenProgramAccount>::new(pool.clone()),
)
```

**See also.** [`AccountRow`](#accountrow), [Two storage models](#two-storage-models).

### `PostgresJsonInstructionProcessor`

Upserts each decoded instruction into the generic `instructions` table as JSONB.

```rust
pub struct PostgresJsonInstructionProcessor<T> { /* private fields */ }

impl<T> PostgresJsonInstructionProcessor<T> {
    pub fn new(pool: sqlx::PgPool) -> Self
}

impl<'a, T> Processor<InstructionProcessorInputType<'a, T>> for PostgresJsonInstructionProcessor<T>
where
    T: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone + Send + Sync + Unpin + 'static,
```

**Details.** Builds an [`InstructionRow<T>`](#instructionrow) and upserts it on `(__signature, __instruction_index, __stack_height)`. The instruction's accounts are not stored separately; they are in the JSON only if the decoded type includes them (generated instruction enums do, under `accounts`). Create the table with [`GenericInstructionMigration`](#genericinstructionmigration). See [the complete program](#example-generic-jsonb-tables).

**Example.**

```rust
.instruction(
    MeteoraDlmmDecoder,
    PostgresJsonInstructionProcessor::<MeteoraDlmmInstruction>::new(pool.clone()),
)
```

**See also.** [Instruction primary key](#instruction-primary-key), [`InstructionRow`](#instructionrow).

### `register_postgres_metrics`

Registers the six Postgres metrics with the global [`MetricsRegistry`](https://carbon.sevenlabs.org/reference/metrics.md).

```rust
pub fn register_postgres_metrics()
```

**Details.** [`PipelineBuilder::build`](https://carbon.sevenlabs.org/reference/pipeline-builder.md#build) calls it whenever `carbon-core` is compiled with `postgres`, so you never call it yourself. The metrics therefore appear in every exporter's output when the feature is on, even if no Postgres processor is registered. Calling it again registers the metrics a second time and duplicates them in snapshots.

**Example.**

```rust
// Only needed if you read MetricsRegistry::global() without building a pipeline.
carbon_core::postgres::processors::register_postgres_metrics();
```

**See also.** [Metrics reference](https://carbon.sevenlabs.org/reference/metrics.md).

## Rows

`carbon_core::postgres::rows`. Source: [`rows.rs`](https://github.com/sevenlabs-hq/carbon/blob/v2.0.0/crates/core/src/postgres/rows.rs).

The generic tables these rows map to:

```sql
CREATE TABLE IF NOT EXISTS accounts (
    __pubkey BYTEA NOT NULL,
    __slot NUMERIC,
    data JSONB NOT NULL,
    PRIMARY KEY (__pubkey)
);

CREATE TABLE IF NOT EXISTS instructions (
    __signature TEXT NOT NULL,
    __instruction_index BIGINT NOT NULL,
    __stack_height BIGINT NOT NULL,
    __slot NUMERIC,
    decoded_instruction JSONB NOT NULL,
    PRIMARY KEY (__signature, __instruction_index, __stack_height)
);
```

### `AccountRow`

One row of the generic `accounts` table.

```rust
#[derive(sqlx::FromRow, Debug, Clone)]
pub struct AccountRow<T>
where
    T: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone + Send + Sync + Unpin + 'static,
{
    #[sqlx(flatten)]
    pub metadata: AccountRowMetadata,
    pub data: sqlx::types::Json<T>,
}

impl<T> AccountRow<T> {
    pub fn from_parts(source: T, metadata: AccountMetadata) -> Self
}
```

**Details.**

| Field      | Type                                        | Column               | Description                                        |
| ---------- | ------------------------------------------- | -------------------- | -------------------------------------------------- |
| `metadata` | [`AccountRowMetadata`](#accountrowmetadata) | `__pubkey`, `__slot` | Account address and slot.                          |
| `data`     | `sqlx::types::Json<T>`                      | `data`               | The decoded account, serialized with `serde_json`. |

Trait implementations:

| Trait               | SQL                                                             | Key                             |
| ------------------- | --------------------------------------------------------------- | ------------------------------- |
| [`Insert`](#insert) | `INSERT INTO accounts …`; fails on an existing pubkey           | —                               |
| [`Upsert`](#upsert) | `… ON CONFLICT (__pubkey) DO UPDATE SET __slot = $2, data = $3` | —                               |
| [`Delete`](#delete) | `DELETE FROM accounts WHERE __pubkey = $1`                      | [`primitives::Pubkey`](#pubkey) |
| [`Lookup`](#lookup) | `SELECT * FROM accounts WHERE __pubkey = $1`                    | [`primitives::Pubkey`](#pubkey) |

Database errors are returned as [`Error::Custom`](https://carbon.sevenlabs.org/reference/errors.md#custom) with the sqlx message.

**Example.** Read back what `PostgresJsonAccountProcessor` wrote:

```rust
use carbon_core::postgres::{operations::Lookup, primitives::Pubkey, rows::AccountRow};

let row = AccountRow::<MeteoraDlmmAccount>::lookup(Pubkey(pubkey), &pool).await?;
if let Some(row) = row {
    println!("slot={:?} data={:?}", row.metadata.slot, row.data.0);
}
```

**See also.** [`PostgresJsonAccountProcessor`](#postgresjsonaccountprocessor), [`GenericAccountsMigration`](#genericaccountsmigration).

### `InstructionRow`

One row of the generic `instructions` table.

```rust
#[derive(sqlx::FromRow, Debug, Clone)]
pub struct InstructionRow<T>
where
    T: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone + Send + Sync + Unpin + 'static,
{
    #[sqlx(flatten)]
    pub metadata: InstructionRowMetadata,
    pub decoded_instruction: sqlx::types::Json<T>,
}

impl<T> InstructionRow<T> {
    pub fn from_parts(decoded_instruction: T, metadata: InstructionMetadata) -> Self
}
```

**Details.**

| Field                 | Type                                                | Column                                                           | Description                      |
| --------------------- | --------------------------------------------------- | ---------------------------------------------------------------- | -------------------------------- |
| `metadata`            | [`InstructionRowMetadata`](#instructionrowmetadata) | `__signature`, `__instruction_index`, `__stack_height`, `__slot` | Where the instruction sits.      |
| `decoded_instruction` | `sqlx::types::Json<T>`                              | `decoded_instruction`                                            | The decoded instruction as JSON. |

`Insert`, `Upsert` (`ON CONFLICT (__signature, __instruction_index, __stack_height) DO UPDATE SET __slot = $4, decoded_instruction = $5`), `Delete` and `Lookup` are implemented. The `Delete` and `Lookup` key is `(String, U32, U32)`: signature in base58, instruction index, stack height.

**Example.**

```rust
use carbon_core::postgres::{operations::Lookup, rows::InstructionRow};

// Top-level instruction 0 of a transaction.
let key = (signature.to_string(), 0u32.into(), 1u32.into());
let row = InstructionRow::<MeteoraDlmmInstruction>::lookup(key, &pool).await?;
```

**See also.** [Instruction primary key](#instruction-primary-key), [`GenericInstructionMigration`](#genericinstructionmigration).

### Instruction primary key

Instruction tables, generic and generated, use the primary key `(__signature, __instruction_index, __stack_height)`. `__instruction_index` comes from [`InstructionMetadata::index`](https://carbon.sevenlabs.org/reference/processor-inputs.md), which for an inner instruction is the index of its **top-level** instruction, not its position among siblings.

> **Warning: Sibling CPIs overwrite each other.** Two inner instructions of the same program at the same stack height under the same top-level instruction (for example, two `Swap` CPIs from one router call) produce the same key. Because the processors upsert, only the last one processed remains in the table. If you need every CPI, write your own wrapper that stores `InstructionMetadata::absolute_path` and keys on it.

### `GenericAccountsMigration`

`sqlx_migrator` migration that creates the generic `accounts` table.

```rust
pub struct GenericAccountsMigration;

impl sqlx_migrator::Migration<sqlx::Postgres> for GenericAccountsMigration
```

**Details.** App `"carbon_core"`, name `"generic_accounts"`, no parents, one operation: [`AccountRowMigrationOperation`](#accountrowmigrationoperation). Rolling it back drops the table.

**Example.**

```rust
use sqlx_migrator::{Info, Migrate, Plan};

let mut migrator = sqlx_migrator::Migrator::default();
migrator
    .add_migration(Box::new(GenericAccountsMigration))
    .map_err(|e| Error::Custom(e.to_string()))?;
let mut conn = pool.acquire().await.map_err(|e| Error::Custom(e.to_string()))?;
migrator
    .run(&mut *conn, &Plan::apply_all())
    .await
    .map_err(|e| Error::Custom(e.to_string()))?;
```

**See also.** [`GenericInstructionMigration`](#genericinstructionmigration).

### `GenericInstructionMigration`

`sqlx_migrator` migration that creates the generic `instructions` table.

```rust
pub struct GenericInstructionMigration;

impl sqlx_migrator::Migration<sqlx::Postgres> for GenericInstructionMigration
```

**Details.** App `"carbon_core"`, name `"generic_instructions"`, no parents, one operation: [`InstructionRowMigrationOperation`](#instructionrowmigrationoperation).

**Example.**

```rust
migrator
    .add_migration(Box::new(GenericInstructionMigration))
    .map_err(|e| Error::Custom(e.to_string()))?;
```

**See also.** [`GenericAccountsMigration`](#genericaccountsmigration).

### `AccountRowMigrationOperation`

The `sqlx_migrator::Operation` behind `GenericAccountsMigration`.

```rust
pub struct AccountRowMigrationOperation;

impl sqlx_migrator::Operation<sqlx::Postgres> for AccountRowMigrationOperation
```

**Details.** `up` runs `CREATE TABLE IF NOT EXISTS accounts (…)` (see [Rows](#rows)); `down` runs `DROP TABLE IF EXISTS accounts`. Use it to include the table in a migration of your own.

**Example.**

```rust
fn operations(&self) -> Vec<Box<dyn sqlx_migrator::Operation<sqlx::Postgres>>> {
    vec![Box::new(AccountRowMigrationOperation)]
}
```

**See also.** [`InstructionRowMigrationOperation`](#instructionrowmigrationoperation).

### `InstructionRowMigrationOperation`

The `sqlx_migrator::Operation` behind `GenericInstructionMigration`.

```rust
pub struct InstructionRowMigrationOperation;

impl sqlx_migrator::Operation<sqlx::Postgres> for InstructionRowMigrationOperation
```

**Details.** `up` creates the `instructions` table; `down` runs `DROP TABLE IF EXISTS instructions`.

**Example.**

```rust
fn operations(&self) -> Vec<Box<dyn sqlx_migrator::Operation<sqlx::Postgres>>> {
    vec![Box::new(InstructionRowMigrationOperation)]
}
```

**See also.** [`AccountRowMigrationOperation`](#accountrowmigrationoperation).

## Operations

`carbon_core::postgres::operations`. Source: [`operations.rs`](https://github.com/sevenlabs-hq/carbon/blob/v2.0.0/crates/core/src/postgres/operations.rs). `Insert`, `Upsert`, `Delete` and `Lookup` are `#[async_trait]` traits: annotate your impls with `#[async_trait::async_trait]`.

### `Table`

Static table name and column list.

```rust
pub trait Table {
    fn table() -> &'static str;
    fn columns() -> Vec<&'static str>;
}
```

**Details.** Nothing in `carbon-core` calls it. Generated decoders implement it on each decoded type (not the row), for example `Swap::table()` returns `"swap_instruction"`. Use it to build queries without hard-coding names.

**Example.**

```rust
use carbon_core::postgres::operations::Table;
use carbon_meteora_dlmm_decoder::instructions::swap::Swap;

let sql = format!("SELECT count(*) FROM {}", Swap::table());
```

**See also.** [Generated decoder types](#generated-decoder-types).

### `Insert`

Inserts a row, failing if the primary key exists.

```rust
#[async_trait::async_trait]
pub trait Insert {
    async fn insert(&self, pool: &sqlx::PgPool) -> CarbonResult<()>;
}
```

**Details.** No built-in processor calls `insert`; they all use [`Upsert`](#upsert). Implemented by `AccountRow`, `InstructionRow` and every generated row and wrapper.

**Example.**

```rust
AccountRow::from_parts(decoded, metadata).insert(&pool).await?;
```

**See also.** [`Upsert`](#upsert).

### `Upsert`

Inserts a row or updates it on primary-key conflict.

```rust
#[async_trait::async_trait]
pub trait Upsert {
    async fn upsert(&self, pool: &sqlx::PgPool) -> CarbonResult<()>;
}
```

**Details.** The bound on `W` in [`PostgresAccountProcessor`](#postgresaccountprocessor) and [`PostgresInstructionProcessor`](#postgresinstructionprocessor). Return `Ok(())` without writing to skip an update.

**Example.** See [a custom row](#example-custom-row).

**See also.** [`Insert`](#insert).

### `Delete`

Deletes the row matching a key.

```rust
#[async_trait::async_trait]
pub trait Delete {
    type Key;

    async fn delete(key: Self::Key, pool: &sqlx::PgPool) -> CarbonResult<()>;
}
```

**Details.** An associated function, not a method: call `AccountRow::<T>::delete(key, &pool)`. No built-in processor deletes rows; account closures arrive as [`AccountDeletion`](https://carbon.sevenlabs.org/reference/updates.md) updates, which you handle with a [`account_deletions`](https://carbon.sevenlabs.org/reference/pipeline-builder.md#account_deletions) processor that calls `delete`.

**Example.**

```rust
use carbon_core::postgres::{operations::Delete, rows::AccountRow};

AccountRow::<MeteoraDlmmAccount>::delete(deletion.pubkey.into(), &pool).await?;
```

**See also.** [`Lookup`](#lookup).

### `Lookup`

Selects the row matching a key, or `None`.

```rust
#[async_trait::async_trait]
pub trait Lookup: Sized {
    type Key;

    async fn lookup(key: Self::Key, pool: &sqlx::PgPool) -> CarbonResult<Option<Self>>;
}
```

**Details.** Generated account rows implement it with `Key = primitives::Pubkey`; the generated GraphQL `QueryRoot` uses it for single-account fields.

**Example.** See [`AccountRow`](#accountrow).

**See also.** [`Delete`](#delete).

## Metadata

`carbon_core::postgres::metadata`. Source: [`metadata.rs`](https://github.com/sevenlabs-hq/carbon/blob/v2.0.0/crates/core/src/postgres/metadata.rs). Row types embed these with `#[sqlx(flatten)]`; the `__` prefix keeps metadata columns from colliding with program field names.

### `AccountRowMetadata`

Account metadata columns.

```rust
#[derive(Debug, Clone, PartialEq, sqlx::FromRow, serde::Serialize, serde::Deserialize)]
pub struct AccountRowMetadata {
    #[sqlx(rename = "__pubkey")]
    pub pubkey: Pubkey,
    #[sqlx(rename = "__slot")]
    pub slot: Option<U64>,
}

impl From<AccountMetadata> for AccountRowMetadata
```

**Details.**

| Field    | Type                           | Column     | Postgres type      | Description                                                              |
| -------- | ------------------------------ | ---------- | ------------------ | ------------------------------------------------------------------------ |
| `pubkey` | [`Pubkey`](#pubkey)            | `__pubkey` | `BYTEA` (32 bytes) | Account address.                                                         |
| `slot`   | `Option<`[`U64`](#u64-u128)`>` | `__slot`   | `NUMERIC`          | Slot of the update. Always `Some` when converted from `AccountMetadata`. |

`AccountMetadata::transaction_signature` is not stored.

**Example.**

```rust
let columns: AccountRowMetadata = input.metadata.clone().into();
```

**See also.** [`InstructionRowMetadata`](#instructionrowmetadata).

### `InstructionRowMetadata`

Instruction metadata columns.

```rust
#[derive(Debug, Clone, PartialEq, sqlx::FromRow, serde::Serialize, serde::Deserialize)]
pub struct InstructionRowMetadata {
    #[sqlx(rename = "__signature")]
    pub signature: String,
    #[sqlx(rename = "__instruction_index")]
    pub instruction_index: U32,
    #[sqlx(rename = "__stack_height")]
    pub stack_height: U32,
    #[sqlx(rename = "__slot")]
    pub slot: Option<U64>,
}

impl From<InstructionMetadata> for InstructionRowMetadata
```

**Details.**

| Field               | Type                           | Column                | Postgres type | Description                                                                                   |
| ------------------- | ------------------------------ | --------------------- | ------------- | --------------------------------------------------------------------------------------------- |
| `signature`         | `String`                       | `__signature`         | `TEXT`        | Transaction signature, base58.                                                                |
| `instruction_index` | [`U32`](#u8-u16-u32)           | `__instruction_index` | `BIGINT`      | `InstructionMetadata::index`: the top-level instruction's index, also for inner instructions. |
| `stack_height`      | [`U32`](#u8-u16-u32)           | `__stack_height`      | `BIGINT`      | `1` for top-level instructions, `2`+ for CPIs.                                                |
| `slot`              | `Option<`[`U64`](#u64-u128)`>` | `__slot`              | `NUMERIC`     | Transaction slot. Always `Some` when converted.                                               |

`absolute_path`, block time and the rest of `TransactionMetadata` are not stored.

**Example.**

```rust
let columns: InstructionRowMetadata = input.metadata.clone().into();
```

**See also.** [Instruction primary key](#instruction-primary-key).

## Primitives

`carbon_core::postgres::primitives`. Source: [`primitives.rs`](https://github.com/sevenlabs-hq/carbon/blob/v2.0.0/crates/core/src/postgres/primitives.rs). Postgres has no unsigned integers and no 32-byte key type, so these newtypes implement sqlx `Type`, `Encode`, `Decode` and `PgHasArrayType` with a lossless mapping. Each derefs to its inner value, is a single public tuple field, and serializes transparently with serde.

| Type                 | Wraps                   | Postgres type | Inner value             |
| -------------------- | ----------------------- | ------------- | ----------------------- |
| [`Pubkey`](#pubkey)  | `solana_pubkey::Pubkey` | `BYTEA`       | `solana_pubkey::Pubkey` |
| [`U8`](#u8-u16-u32)  | `u8`                    | `INT2`        | `i16`                   |
| [`U16`](#u8-u16-u32) | `u16`                   | `INT4`        | `i32`                   |
| [`U32`](#u8-u16-u32) | `u32`                   | `INT8`        | `i64`                   |
| [`U64`](#u64-u128)   | `u64`                   | `NUMERIC`     | `u64`                   |
| [`U128`](#u64-u128)  | `u128`                  | `NUMERIC`     | `u128`                  |
| [`I128`](#i128)      | `i128`                  | `NUMERIC`     | `i128`                  |

Signed types up to 64 bits use sqlx's own `i16`/`i32`/`i64` mappings directly.

### `Pubkey`

A Solana public key stored as 32 raw bytes.

```rust
#[derive(Clone, Copy, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
#[serde(transparent)]
pub struct Pubkey(pub solana_pubkey::Pubkey);

impl From<solana_pubkey::Pubkey> for Pubkey
impl From<Vec<u8>> for Pubkey
impl From<&[u8]> for Pubkey
impl Deref for Pubkey { type Target = solana_pubkey::Pubkey; }
```

**Details.** Encodes as `BYTEA`, arrays as `BYTEA[]`. Decoding fails if the column does not hold exactly 32 bytes. `From<Vec<u8>>` and `From<&[u8]>` return the all-zero key `11111111111111111111111111111111` for input of the wrong length instead of failing. It is a different type from `carbon_core::graphql::primitives::Pubkey`.

**Example.**

```rust
let key: carbon_core::postgres::primitives::Pubkey = solana_pubkey::Pubkey::new_unique().into();
println!("{}", *key); // Deref to solana_pubkey::Pubkey
```

**See also.** [`AccountRowMetadata`](#accountrowmetadata).

### `U8`, `U16`, `U32`

Unsigned integers stored in the next wider signed column.

```rust
pub struct U8(pub i16);   // INT2
pub struct U16(pub i32);  // INT4
pub struct U32(pub i64);  // INT8

// For each of (U8, u8, i16), (U16, u16, i32), (U32, u32, i64):
impl From<u8> for U8
impl From<i16> for U8
impl From<&i16> for U8
impl TryFrom<U8> for u8 { type Error = carbon_core::error::Error; }
impl Deref for U8 { type Target = i16; }
```

**Details.** The inner field is the **signed** wider type, so `*U32` is an `i64`. `TryFrom` back to the unsigned type returns `Error::Custom("out of range")` if the stored value does not fit, which only happens when the column was written by something other than Carbon. Derives `Clone, Copy, PartialEq, Eq, Debug`.

**Example.**

```rust
let decimals: U8 = 255u8.into(); // stored as INT2 255
let back: u8 = decimals.try_into()?;
```

**See also.** [`U64`, `U128`](#u64-u128).

### `U64`, `U128`

Unsigned 64- and 128-bit integers stored as `NUMERIC`.

```rust
pub struct U64(pub u64);
pub struct U128(pub u128);

// For each of (U64, u64), (U128, u128):
impl From<u64> for U64
impl TryFrom<U64> for u64 { type Error = carbon_core::error::Error; } // always Ok
impl Deref for U64 { type Target = u64; }
impl From<U64> for sqlx::types::Decimal
impl From<sqlx::types::Decimal> for U64
impl From<&sqlx::types::Decimal> for U64
```

**Details.** Encodes through `BigDecimal` as a binary `NUMERIC`, so the full range survives (`u64::MAX` round-trips). Decoding fails with `Error::Custom` when the stored value has a fractional part or a minus sign (`"invalid sign/scale for unsigned integer"`) or exceeds the type's maximum (`"value exceeds maximum for target type"`). The `From<Decimal>` conversions are infallible: values that are negative, fractional or out of range become `0` silently. `From<U64> for Decimal` also returns `0` for values a `rust_decimal::Decimal` cannot hold (above about 7.9 × 10²⁸, only reachable with `U128`). Derives `Clone, Debug, PartialEq, Eq` (not `Copy`; bind with `&row.field`).

**Example.**

```rust
let supply: U64 = u64::MAX.into();
let raw: u64 = *supply;
```

**See also.** [`I128`](#i128).

### `I128`

Signed 128-bit integer stored as `NUMERIC`.

```rust
pub struct I128(pub i128);

impl From<i128> for I128
impl Deref for I128 { type Target = i128; }
```

**Details.** Same `NUMERIC` encoding as `U128`. Decoding fails with `Error::Custom("invalid scale for i128")` for fractional values and `"invalid numeric string"` when out of range. Unlike `U64`/`U128`, there is no `TryFrom<I128> for i128` and no `Decimal` conversion; use `*value` or `.0`.

**Example.**

```rust
let delta = I128::from(-1i128);
```

**See also.** [`U64`, `U128`](#u64-u128).

## Generated decoder types

A decoder built with its `postgres` feature (for example `carbon-meteora-dlmm-decoder`, generated by the [CLI](https://carbon.sevenlabs.org/reference/cli.md)) adds these public items. Names use the program's PascalCase name (`MeteoraDlmm`) and each type's snake_case name.

| Item                               | Path                     | Description                                                                                                                                                                                      |
| ---------------------------------- | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `<Program>AccountWithMetadata`     | `accounts::postgres`     | Tuple struct `(Account enum, AccountMetadata)`. `From`, `Insert`, `Upsert`; dispatches to the variant's row. Use as `W` in `PostgresAccountProcessor`.                                           |
| `<Program>AccountsMigration`       | `accounts::postgres`     | `sqlx_migrator` migration creating every `<account>_account` table. App name is the program's kebab-case name.                                                                                   |
| `<Account>Row`                     | `accounts::postgres`     | Typed row: flattened `AccountRowMetadata` plus one column per field. `from_parts(account, metadata)`, `TryFrom<Row> for <Account>`, `Insert`, `Upsert`, `Delete` and `Lookup` keyed by `Pubkey`. |
| `<Account>MigrationOperation`      | `accounts::postgres`     | Creates one account table.                                                                                                                                                                       |
| `<Program>InstructionWithMetadata` | `instructions::postgres` | Tuple struct `(Instruction enum, InstructionMetadata, Vec<AccountMeta>)`. Use as `W` in `PostgresInstructionProcessor`.                                                                          |
| `<Program>InstructionsMigration`   | `instructions::postgres` | Creates every `<instruction>_instruction` table.                                                                                                                                                 |
| `<Instruction>Row`                 | `instructions::postgres` | Flattened `InstructionRowMetadata`, one column per argument, and `__accounts JSONB`.                                                                                                             |
| `<Instruction>MigrationOperation`  | `instructions::postgres` | Creates one instruction table.                                                                                                                                                                   |
| `impl Table for <Type>`            | on the decoded type      | Table name (`swap_instruction`, `lb_pair_account`) and column list.                                                                                                                              |

Generated columns use `NUMERIC(20)` for `u64`, `NUMERIC(39)` for `u128`, `NUMERIC(38)` for `i128` (values with 39 digits, above 10³⁸, fail to insert), `INT2`/`INT4`/`INT8` for `u8`/`u16`/`u32`, `BYTEA` for pubkeys and byte arrays, and `JSONB` for nested structs and enums.

## Examples

### Example: typed tables

Indexes Meteora DLMM accounts and instructions into per-type tables. It uses the public RPC transaction crawler, so only `DATABASE_URL` is required.

`Cargo.toml`:

```toml
[dependencies]
carbon-core = { version = "2.0.0", features = ["postgres"] }
carbon-log-metrics = "2.0.0"
carbon-meteora-dlmm-decoder = { version = "2.0.0", features = ["postgres"] }
carbon-rpc-transaction-crawler-datasource = "2.0.0"
env_logger = "0.11.8"
sqlx = { version = "0.8.6", features = ["runtime-tokio-rustls", "postgres"] }
sqlx_migrator = { version = "0.17.0", features = ["postgres"] }
tokio = { version = "1.48.0", features = ["full"] }
```

`src/main.rs`:

```rust
use {
    carbon_core::{
        error::{CarbonResult, Error},
        pipeline::Pipeline,
        postgres::processors::{PostgresAccountProcessor, PostgresInstructionProcessor},
    },
    carbon_log_metrics::LogMetrics,
    carbon_meteora_dlmm_decoder::{
        accounts::{
            postgres::{MeteoraDlmmAccountWithMetadata, MeteoraDlmmAccountsMigration},
            MeteoraDlmmAccount,
        },
        instructions::{
            postgres::{MeteoraDlmmInstructionWithMetadata, MeteoraDlmmInstructionsMigration},
            MeteoraDlmmInstruction,
        },
        MeteoraDlmmDecoder, PROGRAM_ID,
    },
    carbon_rpc_transaction_crawler_datasource::{ConnectionConfig, Filters, RpcTransactionCrawler},
    sqlx_migrator::{Info, Migrate, Plan},
    std::sync::Arc,
};

#[tokio::main]
async fn main() -> CarbonResult<()> {
    env_logger::init();

    let database_url = std::env::var("DATABASE_URL")
        .map_err(|_| Error::Custom("DATABASE_URL must be set".to_string()))?;
    let pool = sqlx::PgPool::connect(&database_url)
        .await
        .map_err(|e| Error::Custom(e.to_string()))?;

    // Create one table per account type and per instruction type.
    let mut migrator = sqlx_migrator::Migrator::default();
    migrator
        .add_migration(Box::new(MeteoraDlmmAccountsMigration))
        .map_err(|e| Error::Custom(e.to_string()))?;
    migrator
        .add_migration(Box::new(MeteoraDlmmInstructionsMigration))
        .map_err(|e| Error::Custom(e.to_string()))?;
    let mut conn = pool.acquire().await.map_err(|e| Error::Custom(e.to_string()))?;
    migrator
        .run(&mut *conn, &Plan::apply_all())
        .await
        .map_err(|e| Error::Custom(e.to_string()))?;

    let datasource = RpcTransactionCrawler::new(
        "https://api.mainnet-beta.solana.com".to_string(),
        PROGRAM_ID,
        ConnectionConfig::default(),
        Filters::new(None, None, None),
        None,
    );

    Pipeline::builder()
        .datasource(datasource)
        .metrics(Arc::new(LogMetrics::new()))
        .account(
            MeteoraDlmmDecoder,
            PostgresAccountProcessor::<MeteoraDlmmAccount, MeteoraDlmmAccountWithMetadata>::new(
                pool.clone(),
            ),
        )
        .instruction(
            MeteoraDlmmDecoder,
            PostgresInstructionProcessor::<
                MeteoraDlmmInstruction,
                MeteoraDlmmInstructionWithMetadata,
            >::new(pool.clone()),
        )
        .build()?
        .run()
        .await
}
```

```sh
DATABASE_URL=postgres://postgres:postgres@localhost:5432/carbon cargo run
```

The transaction crawler emits transactions only, so the account tables stay empty here; pair the same processors with an account datasource such as [`rpc-program-subscribe`](https://carbon.sevenlabs.org/datasources/rpc-program-subscribe.md) to fill them. Then query a table:

```sql
SELECT __signature, amount_in, min_amount_out FROM swap_instruction ORDER BY __slot DESC LIMIT 5;
```

### Example: generic JSONB tables

The same pipeline writing into the two generic tables. Replace the migrations and the processor:

`src/main.rs`:

```rust
use {
    carbon_core::{
        error::{CarbonResult, Error},
        pipeline::Pipeline,
        postgres::{
            processors::PostgresJsonInstructionProcessor,
            rows::{GenericAccountsMigration, GenericInstructionMigration},
        },
    },
    carbon_log_metrics::LogMetrics,
    carbon_meteora_dlmm_decoder::{
        instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder, PROGRAM_ID,
    },
    carbon_rpc_transaction_crawler_datasource::{ConnectionConfig, Filters, RpcTransactionCrawler},
    sqlx_migrator::{Info, Migrate, Plan},
    std::sync::Arc,
};

#[tokio::main]
async fn main() -> CarbonResult<()> {
    env_logger::init();

    let database_url = std::env::var("DATABASE_URL")
        .map_err(|_| Error::Custom("DATABASE_URL must be set".to_string()))?;
    let pool = sqlx::PgPool::connect(&database_url)
        .await
        .map_err(|e| Error::Custom(e.to_string()))?;

    // Creates the generic `accounts` and `instructions` tables.
    let mut migrator = sqlx_migrator::Migrator::default();
    migrator
        .add_migration(Box::new(GenericAccountsMigration))
        .map_err(|e| Error::Custom(e.to_string()))?;
    migrator
        .add_migration(Box::new(GenericInstructionMigration))
        .map_err(|e| Error::Custom(e.to_string()))?;
    let mut conn = pool.acquire().await.map_err(|e| Error::Custom(e.to_string()))?;
    migrator
        .run(&mut *conn, &Plan::apply_all())
        .await
        .map_err(|e| Error::Custom(e.to_string()))?;

    let datasource = RpcTransactionCrawler::new(
        "https://api.mainnet-beta.solana.com".to_string(),
        PROGRAM_ID,
        ConnectionConfig::default(),
        Filters::new(None, None, None),
        None,
    );

    Pipeline::builder()
        .datasource(datasource)
        .metrics(Arc::new(LogMetrics::new()))
        .instruction(
            MeteoraDlmmDecoder,
            PostgresJsonInstructionProcessor::<MeteoraDlmmInstruction>::new(pool.clone()),
        )
        .build()?
        .run()
        .await
}
```

Generated instruction enums serialize as `{"type": "<Variant>", "data": {"program_id": …, "data": {…}, "accounts": {…}}}`. Pubkeys inside the JSON are arrays of 32 numbers, not base58 strings.

```sql
SELECT __signature, decoded_instruction->'data'->'data'->>'amount_in' AS amount_in
FROM instructions
WHERE decoded_instruction->>'type' = 'Swap'
LIMIT 5;
```

### Example: custom row

A hand-written wrapper that stores only SPL Token mints in a table of your own. It shows `Table`, `Upsert`, `Lookup`, `Delete` and the primitives together. Create the table yourself:

```sql
CREATE TABLE mints (pubkey BYTEA PRIMARY KEY, slot NUMERIC NOT NULL, supply NUMERIC NOT NULL, decimals INT2 NOT NULL);
```

`src/mints.rs`:

```rust
use {
    carbon_core::{
        account::AccountMetadata,
        error::{CarbonResult, Error},
        postgres::{
            operations::{Delete, Lookup, Table, Upsert},
            primitives::{Pubkey, U64, U8},
            processors::PostgresAccountProcessor,
        },
    },
    carbon_token_program_decoder::accounts::TokenProgramAccount,
};

/// One row of a `mints` table. Non-mint accounts map to `None` and are skipped.
#[derive(sqlx::FromRow, Debug, Clone)]
pub struct MintRow {
    pub pubkey: Pubkey,
    pub slot: U64,
    pub supply: U64,
    pub decimals: U8,
}

pub struct MaybeMint(Option<MintRow>);

impl From<(TokenProgramAccount, AccountMetadata)> for MaybeMint {
    fn from((account, metadata): (TokenProgramAccount, AccountMetadata)) -> Self {
        match account {
            TokenProgramAccount::Mint(mint) => MaybeMint(Some(MintRow {
                pubkey: metadata.pubkey.into(),
                slot: metadata.slot.into(),
                supply: mint.supply.into(),
                decimals: mint.decimals.into(),
            })),
            _ => MaybeMint(None),
        }
    }
}

impl Table for MintRow {
    fn table() -> &'static str {
        "mints"
    }

    fn columns() -> Vec<&'static str> {
        vec!["pubkey", "slot", "supply", "decimals"]
    }
}

#[async_trait::async_trait]
impl Upsert for MaybeMint {
    async fn upsert(&self, pool: &sqlx::PgPool) -> CarbonResult<()> {
        let Some(row) = &self.0 else { return Ok(()) };
        sqlx::query(
            "INSERT INTO mints (pubkey, slot, supply, decimals) VALUES ($1, $2, $3, $4)
             ON CONFLICT (pubkey) DO UPDATE SET slot = $2, supply = $3, decimals = $4",
        )
        .bind(row.pubkey)
        .bind(&row.slot)
        .bind(&row.supply)
        .bind(row.decimals)
        .execute(pool)
        .await
        .map_err(|e| Error::Custom(e.to_string()))?;
        Ok(())
    }
}

#[async_trait::async_trait]
impl Lookup for MintRow {
    type Key = Pubkey;

    async fn lookup(key: Pubkey, pool: &sqlx::PgPool) -> CarbonResult<Option<Self>> {
        sqlx::query_as("SELECT * FROM mints WHERE pubkey = $1")
            .bind(key)
            .fetch_optional(pool)
            .await
            .map_err(|e| Error::Custom(e.to_string()))
    }
}

#[async_trait::async_trait]
impl Delete for MintRow {
    type Key = Pubkey;

    async fn delete(key: Pubkey, pool: &sqlx::PgPool) -> CarbonResult<()> {
        sqlx::query("DELETE FROM mints WHERE pubkey = $1")
            .bind(key)
            .execute(pool)
            .await
            .map_err(|e| Error::Custom(e.to_string()))?;
        Ok(())
    }
}

pub fn mint_processor(pool: sqlx::PgPool) -> PostgresAccountProcessor<TokenProgramAccount, MaybeMint> {
    PostgresAccountProcessor::new(pool)
}
```

Register it with `.account(TokenProgramDecoder, mints::mint_processor(pool.clone()))`. It needs `carbon-token-program-decoder = "2.0.0"` and `async-trait = "0.1.89"` in addition to the dependencies above.

---

Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/reference/postgres · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt