ClickHouse for Solana Data

-- Core transactions table
CREATE TABLE solana.transactions
(
    signature         String,
    slot              UInt64,
    block_time        DateTime,
    success           Bool,
    fee               UInt64,
    compute_units     UInt32,
    
    -- Account keys (up to 64 accounts per transaction)
    account_keys      Array(String),
    
    -- Program IDs involved
    program_ids       Array(String),
    
    -- Pre/post token balances for DEX analysis
    pre_token_balances  Array(Tuple(
        account_index UInt8,
        mint          String,
        amount        UInt64,
        decimals      UInt8
    )),
    post_token_balances Array(Tuple(
        account_index UInt8,
        mint          String,
        amount        UInt64,
        decimals      UInt8
    )),
    
    -- Log messages for event parsing
    log_messages      Array(String),
    
    -- Instruction data (base58 encoded)
    instructions      Array(Tuple(
        program_id    String,
        accounts      Array(UInt8),
        data          String
    ))
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(block_time)
ORDER BY (block_time, slot, signature)
SETTINGS index_granularity = 8192;

-- Parsed DEX swap events (materialized from transactions)
CREATE TABLE solana.dex_swaps
(
    signature         String,
    block_time        DateTime,
    slot              UInt64,
    
    -- DEX identification
    program_id        String,
    pool_address      String,
    
    -- Swap details
    trader            String,
    token_in_mint     String,
    token_out_mint    String,
    amount_in         UInt64,
    amount_out        UInt64,
    
    -- Derived metrics
    price_impact      Float64,
    fee_amount        UInt64
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(block_time)
ORDER BY (block_time, program_id, pool_address)
SETTINGS index_granularity = 8192;

-- Materialized view to auto-populate from transactions
CREATE MATERIALIZED VIEW solana.dex_swaps_mv
TO solana.dex_swaps
AS SELECT
    signature,
    block_time,
    slot,
    -- Parse swap data from transactions
    -- (parsing logic depends on specific DEX program)
    program_ids[1] as program_id
FROM solana.transactions
WHERE has(program_ids, '675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8'); -- Raydium

-- 24-hour trading volume by DEX program
SELECT 
    program_id,
    count() as swap_count,
    sum(amount_in) / 1e9 as volume_sol,
    uniqExact(trader) as unique_traders
FROM solana.dex_swaps
WHERE block_time > now() - INTERVAL 24 HOUR
GROUP BY program_id
ORDER BY volume_sol DESC;

-- Top wallets by transaction count in last 7 days
SELECT 
    arrayJoin(account_keys) as wallet,
    count() as tx_count,
    sum(fee) / 1e9 as total_fees_sol
FROM solana.transactions
WHERE 
    block_time > now() - INTERVAL 7 DAY
    AND success = true
GROUP BY wallet
HAVING tx_count > 100
ORDER BY tx_count DESC
LIMIT 100;

-- 1-minute OHLCV for a token pair
SELECT 
    toStartOfMinute(block_time) as minute,
    argMin(price, block_time) as open,
    max(price) as high,
    min(price) as low,
    argMax(price, block_time) as close,
    sum(amount_in) as volume
FROM (
    SELECT 
        block_time,
        toFloat64(amount_out) / toFloat64(amount_in) as price,
        amount_in
    FROM solana.dex_swaps
    WHERE 
        token_in_mint = 'So11111111111111111111111111111111111111112'  -- SOL
        AND token_out_mint = 'YourTokenMint'
        AND block_time > now() - INTERVAL 24 HOUR
)
GROUP BY minute
ORDER BY minute;

import { createClient } from '@clickhouse/client';

const client = createClient({
  url: 'http://localhost:8123',
  database: 'solana',
  // Enable async inserts for high-throughput ingestion
  clickhouse_settings: {
    async_insert: 1,
    wait_for_async_insert: 0,
    async_insert_max_data_size: '10000000', // 10MB batches
    async_insert_busy_timeout_ms: '1000',   // Flush every 1 second
  },
});

// Buffer transactions and insert in batches
const buffer: TransactionRecord[] = [];
const BATCH_SIZE = 1000;

async function ingestTransaction(tx: ParsedTransaction) {
  buffer.push(transformTransaction(tx));
  
  if (buffer.length >= BATCH_SIZE) {
    await flushBuffer();
  }
}

async function flushBuffer() {
  if (buffer.length === 0) return;
  
  const batch = buffer.splice(0, buffer.length);
  await client.insert({
    table: 'transactions',
    values: batch,
    format: 'JSONEachRow',
  });
  
  console.log(`Inserted ${batch.length} transactions`);
}