Serve indexed data over GraphQL

Carbon's graphql feature serves the tables written by the Postgres processors as a read-only GraphQL API. A decoder built with its graphql feature contains a complete query schema (QueryRoot) over its typed tables, and carbon_core::graphql::server turns any Juniper schema into an Axum router with /graphql and a GraphiQL playground at /graphiql.

This guide runs a Meteora DLMM pipeline that writes to Postgres and serves it on port 8080, then adds a custom query next to the generated ones.

You need:

• Rust 1.96.1 or newer.
• PostgreSQL and a DATABASE_URL, as in Store decoded data in Postgres. This guide was run against PostgreSQL 16.4.
• A decoder with both postgres and graphql features. The generated GraphQL module reads the typed Postgres tables, so graphql doesn't work without postgres.

What you get#

See the GraphQL reference for the signatures.

cargo new dlmm-graphql
cd dlmm-graphql
cargo add carbon-core@2.0.0 --features postgres,graphql
cargo add carbon-meteora-dlmm-decoder@2.0.0 --features postgres,graphql
cargo add carbon-rpc-transaction-crawler-datasource@2.0.0
cargo add sqlx@0.8.6 --features postgres,runtime-tokio-rustls
cargo add sqlx_migrator@0.17.0 --features postgres
cargo add axum@0.8 juniper@0.16 solana-pubkey@4.2.1
cargo add tokio --features full
cargo add env_logger log

[dependencies]
axum = "0.8"
carbon-core = { version = "2.0.0", features = ["postgres", "graphql"] }
carbon-meteora-dlmm-decoder = { version = "2.0.0", features = ["postgres", "graphql"] }
carbon-rpc-transaction-crawler-datasource = "2.0.0"
env_logger = "0.11.11"
juniper = "0.16"
log = "0.4.35"
solana-pubkey = "4.2.1"
sqlx = { version = "0.8.6", features = ["postgres", "runtime-tokio-rustls"] }
sqlx_migrator = { version = "0.17.0", features = ["postgres"] }
tokio = { version = "1.54.1", features = ["full"] }

use {
    carbon_core::{
        error::{CarbonResult, Error},
        graphql::server::{build_schema, graphql_router},
        pipeline::Pipeline,
        postgres::processors::PostgresInstructionProcessor,
    },
    carbon_meteora_dlmm_decoder::{
        accounts::postgres::MeteoraDlmmAccountsMigration,
        graphql::{context::GraphQLContext, QueryRoot},
        instructions::{
            postgres::{MeteoraDlmmInstructionWithMetadata, MeteoraDlmmInstructionsMigration},
            MeteoraDlmmInstruction,
        },
        MeteoraDlmmDecoder, PROGRAM_ID,
    },
    carbon_rpc_transaction_crawler_datasource::{ConnectionConfig, Filters, RpcTransactionCrawler},
    sqlx::postgres::PgPoolOptions,
    sqlx_migrator::{Info, Migrate, Migrator, Plan},
    std::sync::Arc,
};

fn custom(e: impl std::fmt::Display) -> Error {
    Error::Custom(e.to_string())
}

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

    let database_url = std::env::var("DATABASE_URL").map_err(custom)?;
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect(&database_url)
        .await
        .map_err(custom)?;

    // QueryRoot reads both account and instruction tables, so create both.
    let mut migrator = Migrator::default();
    migrator
        .add_migration(Box::new(MeteoraDlmmAccountsMigration))
        .map_err(custom)?;
    migrator
        .add_migration(Box::new(MeteoraDlmmInstructionsMigration))
        .map_err(custom)?;
    let mut conn = pool.acquire().await.map_err(custom)?;
    migrator
        .run(&mut *conn, &Plan::apply_all())
        .await
        .map_err(custom)?;
    drop(conn);

    // GraphQL API over the tables the pipeline writes.
    let schema = build_schema(QueryRoot);
    let context = GraphQLContext { pool: Arc::new(pool.clone()) };
    let app = graphql_router(schema, context);
    let listener = tokio::net::TcpListener::bind("0.0.0.0:8080")
        .await
        .map_err(custom)?;
    log::info!("GraphiQL at http://localhost:8080/graphiql");
    tokio::spawn(async move {
        if let Err(e) = axum::serve(listener, app).await {
            log::error!("GraphQL server stopped: {e}");
        }
    });

    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)
        .instruction(
            MeteoraDlmmDecoder,
            PostgresInstructionProcessor::<MeteoraDlmmInstruction, MeteoraDlmmInstructionWithMetadata>::new(pool),
        )
        .build()?
        .run()
        .await
}

export DATABASE_URL=postgres://carbon:carbon@localhost:5432/carbon
RUST_LOG=info cargo run

[… INFO  dlmm_graphql] GraphiQL at http://localhost:8080/graphiql
[… INFO  carbon_core::pipeline] starting pipeline. num_datasources: 1, num_exporters: 0, num_account_pipes: 0, num_account_deletion_pipes: 0, num_instruction_pipes: 1, num_transaction_pipes: 0

curl -s localhost:8080/graphql -H 'content-type: application/json' \
  -d '{"query":"{ listSwap2(limit: 1, offset: 0) { instructionMetadata { signature instructionIndex stackHeight slot } amountIn minAmountOut } }"}'

{"data":{"listSwap2":[{"instructionMetadata":{"signature":"Ez2XgG9LDAvjnQnkmTNZM2rLTcjDAdLcSAv9jPCZJ7qSVo9gcyN8riocyJcwkY5FEvBn7mhWn49SvQ8AEagyUHH","instructionIndex":"2","stackHeight":"2","slot":"467111933"},"amountIn":"124337400","minAmountOut":"0"}]}}

{
  swap2(signature: "SZE5CFK6gkG91yUWfovgDWd7XYFT9DFaHphCnQq2TbghcBY1c1uSzdksAxsayxgh27aBPhrvbBAahUbUNiNB757", instructionIndex: 4) {
    instructionMetadata { stackHeight }
    amountIn
  }
}

Add your own queries#

The generated fields are lookups and paginated lists. For anything else (filters, joins, aggregates) write your own query root with Juniper, reuse the decoder's row and GraphQL types, and expose the generated root as one field so you keep both. This server runs on its own (port 8081) against the same database, which is how you deploy API replicas separately from the indexer:

//! A read-only API with your own queries, next to the generated ones.
use {
    carbon_core::graphql::server::{build_schema, graphql_router},
    carbon_meteora_dlmm_decoder::{
        graphql::{context::GraphQLContext, QueryRoot},
        instructions::{graphql::Swap2GraphQL, postgres::Swap2Row},
    },
    juniper::{graphql_object, FieldError, FieldResult, Value},
    sqlx::postgres::PgPoolOptions,
    solana_pubkey::Pubkey,
    std::{str::FromStr, sync::Arc},
};

pub struct Query;

#[graphql_object(context = GraphQLContext)]
impl Query {
    /// Swap2 instructions in one pool, newest first.
    async fn pool_swaps(
        context: &GraphQLContext,
        lb_pair: String,
        limit: Option<i32>,
    ) -> FieldResult<Vec<Swap2GraphQL>> {
        let lb_pair = Pubkey::from_str(&lb_pair)
            .map_err(|e| FieldError::new(e.to_string(), Value::null()))?;
        // `__accounts` stores each pubkey as a JSON array of 32 bytes; the pool is account 0.
        let rows: Vec<Swap2Row> = sqlx::query_as(
            "SELECT * FROM swap2_instruction \
             WHERE __accounts -> 0 -> 'pubkey' = $1 \
             ORDER BY __slot DESC LIMIT $2",
        )
        .bind(sqlx::types::Json(lb_pair.to_bytes()))
        .bind(i64::from(limit.unwrap_or(20).clamp(1, 100)))
        .fetch_all(&*context.pool)
        .await
        .map_err(|e| FieldError::new(e.to_string(), Value::null()))?;

        Ok(rows.into_iter().filter_map(|row| row.try_into().ok()).collect())
    }

    /// Every query the decoder generates, under `generated { ... }`.
    fn generated() -> QueryRoot {
        QueryRoot
    }
}

#[tokio::main]
async fn main() {
    let pool = PgPoolOptions::new()
        .connect(&std::env::var("DATABASE_URL").expect("DATABASE_URL"))
        .await
        .expect("connect to Postgres");

    let app = graphql_router(
        build_schema(Query),
        GraphQLContext { pool: Arc::new(pool) },
    )
    .route("/healthz", axum::routing::get(|| async { "ok" }));
    let listener = tokio::net::TcpListener::bind("0.0.0.0:8081").await.unwrap();
    println!("GraphiQL at http://localhost:8081/graphiql");
    axum::serve(listener, app).await.unwrap();
}

cargo run --bin api

curl -s localhost:8081/graphql -H 'content-type: application/json' \
  -d '{"query":"{ poolSwaps(lbPair: \"EZyszDEx1LZDt7TsSFV8xdPi49sDKC3mdfv2MVMEQLtU\", limit: 5) { instructionMetadata { signature slot } amountIn } }"}'

{"data":{"poolSwaps":[{"instructionMetadata":{"signature":"SZE5CFK6gkG91yUWfovgDWd7XYFT9DFaHphCnQq2TbghcBY1c1uSzdksAxsayxgh27aBPhrvbBAahUbUNiNB757","slot":"467107994"},"amountIn":"405807562"}]}}

The generated fields are now under generated: { generated { listSwap2(limit: 2, offset: 0) { amountIn } } }. graphql_router returns a plain axum::Router, so .route(..), .merge(..) and Tower layers work as usual; /healthz above answers ok.

The SQL compares __accounts -> 0 -> 'pubkey' with the pool's 32 bytes as JSON because the generated tables store instruction accounts as JSON with each pubkey as a byte array. For frequent filters, add an index on that expression in your own migration.

Before you expose it#

• Authentication. The router has none. Put it behind a reverse proxy or add an Axum middleware layer.
• Cost limits. Juniper doesn't limit query depth or result size here. The generated list… fields pass limit to SQL as given; cap it in your own fields, as poolSwaps does with clamp(1, 100).
• Indexes. The generated lists sort by __slot and the generated tables only index their primary key. Add indexes for the queries you serve.
• Pool. API requests and the pipeline share connections when they share a pool; size max_connections for both, or give the API its own pool.
• Errors. Database errors come back in the GraphQL errors array with the database's message, including table names.

Common problems#

relation "lb_pair_account" does not exist#

An account field was queried but the account migration never ran. QueryRoot covers accounts and instructions; run both MeteoraDlmmAccountsMigration and MeteoraDlmmInstructionsMigration, as main.rs does.

Field "listSwap2" argument "offset" of type "Int!" is required but not provided#

Generated list fields take both limit and offset. Pass offset: 0.

Unknown field "listSwap2" on type "Query"#

You're querying a custom root that nests the generated fields. Use { generated { listSwap2(..) } }, or the field names of your own root.

cannot find `graphql` in `carbon_meteora_dlmm_decoder` #

The decoder's graphql feature is off. Enable postgres and graphql on the decoder and on carbon-core.

Mismatched Router or RootNode types#

Your axum or juniper major version differs from Carbon's. Use axum 0.8 and juniper 0.16.

Address already in use (os error 98)#

Another process listens on 8080. Change the bind address.

Next steps#