Test decoders and processors

[package]
name = "swap-stats"
version = "0.1.0"
edition = "2021"

[dependencies]
carbon-core = "2.0.0"
carbon-meteora-dlmm-decoder = "2.0.0"
hex = "0.4.3"
serde_json = "1.0.145"
solana-pubkey = "4.2.1"
solana-transaction-status = "=4.2.2"

[dev-dependencies]
async-trait = "0.1.89"
carbon-test-utils = "2.0.0"
solana-transaction-error = "3.3.2"
tokio = { version = "1.48.0", features = ["full"] }
tokio-util = "0.7.17"

use {
    carbon_core::{
        datasource::TransactionUpdate,
        error::{CarbonResult, Error},
        instruction::InstructionProcessorInputType,
        processor::Processor,
        transformers::transaction_metadata_from_original_meta,
    },
    carbon_meteora_dlmm_decoder::instructions::MeteoraDlmmInstruction,
    solana_pubkey::Pubkey,
    solana_transaction_status::EncodedConfirmedTransactionWithStatusMeta,
    std::{
        collections::HashMap,
        path::Path,
        sync::{Arc, Mutex},
    },
};

#[derive(Debug, Default, Clone, PartialEq)]
pub struct PairTotals {
    pub swaps: u64,
    pub amount_in: u128,
}

/// The pool and input amount of a swap instruction, if it is one.
pub fn swap_amount(instruction: &MeteoraDlmmInstruction) -> Option<(Pubkey, u64)> {
    match instruction {
        MeteoraDlmmInstruction::Swap { data, accounts, .. } => Some((accounts.lb_pair, data.amount_in)),
        MeteoraDlmmInstruction::Swap2 { data, accounts, .. } => {
            Some((accounts.lb_pair, data.amount_in))
        }
        _ => None,
    }
}

/// Sums swap volume per DLMM pool. Clone it to keep a handle on the totals.
#[derive(Clone, Default)]
pub struct SwapStats {
    pub totals: Arc<Mutex<HashMap<Pubkey, PairTotals>>>,
}

impl Processor<InstructionProcessorInputType<'_, MeteoraDlmmInstruction>> for SwapStats {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, MeteoraDlmmInstruction>,
    ) -> CarbonResult<()> {
        if input.metadata.transaction_metadata.meta.status.is_err() {
            return Ok(()); // failed transactions moved no funds
        }
        if let Some((pair, amount_in)) = swap_amount(input.decoded_instruction) {
            let mut totals = self.totals.lock().map_err(|e| Error::Custom(e.to_string()))?;
            let entry = totals.entry(pair).or_default();
            entry.swaps += 1;
            entry.amount_in += u128::from(amount_in);
        }
        Ok(())
    }
}

/// Loads a `getTransaction` result saved as JSON (Base64 encoding) into a `TransactionUpdate`.
pub fn load_transaction(path: impl AsRef<Path>) -> TransactionUpdate {
    let path = path.as_ref();
    let json = std::fs::read(path).unwrap_or_else(|e| panic!("read {}: {e}", path.display()));
    let fetched: EncodedConfirmedTransactionWithStatusMeta =
        serde_json::from_slice(&json).expect("a getTransaction result");

    let transaction = fetched
        .transaction
        .transaction
        .decode()
        .expect("Base64-encoded transaction");
    let meta = transaction_metadata_from_original_meta(fetched.transaction.meta.expect("meta"))
        .expect("convertible meta");

    TransactionUpdate {
        signature: transaction.signatures[0],
        transaction,
        meta,
        is_vote: false,
        slot: fetched.slot,
        index: None,
        block_time: fetched.block_time,
        block_hash: None,
    }
}

//! Writes every Meteora DLMM instruction of a saved transaction as a
//! `carbon-test-utils` instruction fixture.
use {
    carbon_core::{transaction::TransactionMetadata, transformers::extract_instructions_with_metadata},
    carbon_meteora_dlmm_decoder::PROGRAM_ID,
    serde_json::json,
    std::sync::Arc,
};

fn main() {
    let path = std::env::args().nth(1).expect("usage: extract-instructions <tx.json>");
    let update = swap_stats::load_transaction(&path);
    let metadata = Arc::new(TransactionMetadata::try_from(update.clone()).expect("metadata"));

    let instructions = extract_instructions_with_metadata(&metadata, &update).expect("instructions");
    for (n, (meta, ix)) in instructions.iter().enumerate() {
        if ix.program_id != PROGRAM_ID {
            continue;
        }
        let fixture = json!({
            "program_id": ix.program_id.to_string(),
            "accounts": ix.accounts.iter().map(|a| json!({
                "pubkey": a.pubkey.to_string(),
                "is_signer": a.is_signer,
                "is_writable": a.is_writable,
            })).collect::<Vec<_>>(),
            "data": hex::encode(&ix.data),
        });
        let out = format!("{}.ix{n}.json", path.trim_end_matches(".json"));
        std::fs::write(&out, serde_json::to_string_pretty(&fixture).unwrap()).unwrap();
        println!("{out}: path={:?} stack_height={}", meta.absolute_path, meta.stack_height);
    }
}

use {
    carbon_core::{account::AccountDecoder, instruction::InstructionDecoder},
    carbon_meteora_dlmm_decoder::{
        accounts::MeteoraDlmmAccount,
        instructions::{MeteoraDlmmInstruction, Swap2},
        MeteoraDlmmDecoder,
    },
    carbon_test_utils::{base58_deserialize, read_account, read_instruction},
    solana_pubkey::Pubkey,
};

const POOL: Pubkey = Pubkey::from_str_const("Bnoev1Gp1o247d7zTwSbznRwqjkRmbrx3R6Qt4P16YbQ");
const WSOL: Pubkey = Pubkey::from_str_const("So11111111111111111111111111111111111111112");

#[test]
fn decodes_swap2_data_from_base58() {
    // Instruction data as an explorer shows it (base58).
    let data = base58_deserialize::ix_data("fx9RHbGFfZA1Se5ZmKqQdyyUnuKrrydfLZcDLj");

    let swap = Swap2::decode(&data).expect("Swap2 data");
    assert_eq!(swap.amount_in, 91_444_688);
    assert_eq!(swap.min_amount_out, 0);
}

#[test]
fn decodes_swap2_instruction_fixture() {
    let instruction = read_instruction("tests/fixtures/swap2_first.json").expect("fixture");

    let decoded = MeteoraDlmmDecoder
        .decode_instruction(&instruction)
        .expect("a DLMM instruction");

    let MeteoraDlmmInstruction::Swap2 { data, accounts, .. } = decoded else {
        panic!("expected Swap2, got {decoded:?}");
    };
    assert_eq!(data.amount_in, 91_444_688);
    assert_eq!(accounts.lb_pair, POOL);
    assert_eq!(accounts.token_y_mint, WSOL);
}

#[test]
fn decodes_lb_pair_account_fixture() {
    let account = read_account("tests/fixtures/lb_pair.json").expect("fixture");

    let decoded = MeteoraDlmmDecoder.decode_account(&account).expect("a DLMM account");

    let MeteoraDlmmAccount::LbPair(pair) = decoded.data else {
        panic!("expected LbPair");
    };
    assert_eq!(pair.token_y_mint, WSOL);
}

use {
    async_trait::async_trait,
    carbon_core::{
        datasource::{Datasource, DatasourceId, Update, UpdateType},
        error::{CarbonResult, Error},
        pipeline::Pipeline,
    },
    carbon_meteora_dlmm_decoder::MeteoraDlmmDecoder,
    solana_pubkey::Pubkey,
    swap_stats::{load_transaction, PairTotals, SwapStats},
    tokio::sync::mpsc::Sender,
    tokio_util::sync::CancellationToken,
};

/// Sends saved updates once, then returns. The pipeline stops when it has processed them.
struct FixtureDatasource(Vec<Update>);

#[async_trait]
impl Datasource for FixtureDatasource {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: Sender<(Update, DatasourceId)>,
        _cancellation_token: CancellationToken,
    ) -> CarbonResult<()> {
        for update in &self.0 {
            sender
                .send((update.clone(), id.clone()))
                .await
                .map_err(|e| Error::Custom(e.to_string()))?;
        }
        Ok(())
    }

    fn update_types(&self) -> Vec<UpdateType> {
        vec![UpdateType::Transaction]
    }
}

#[tokio::test]
async fn sums_both_swaps_of_a_routed_transaction() {
    let transaction = load_transaction("tests/fixtures/route.json");
    let stats = SwapStats::default();

    Pipeline::builder()
        .datasource(FixtureDatasource(vec![Update::Transaction(Box::new(transaction))]))
        .instruction(MeteoraDlmmDecoder, stats.clone())
        .build()
        .unwrap()
        .run()
        .await
        .unwrap();

    let totals = stats.totals.lock().unwrap();
    assert_eq!(totals.len(), 2);
    assert_eq!(
        totals[&Pubkey::from_str_const("Bnoev1Gp1o247d7zTwSbznRwqjkRmbrx3R6Qt4P16YbQ")],
        PairTotals { swaps: 1, amount_in: 91_444_688 }
    );
    assert_eq!(
        totals[&Pubkey::from_str_const("EZyszDEx1LZDt7TsSFV8xdPi49sDKC3mdfv2MVMEQLtU")],
        PairTotals { swaps: 1, amount_in: 405_807_562 }
    );
}

#[tokio::test]
async fn ignores_failed_transactions() {
    let mut transaction = load_transaction("tests/fixtures/route.json");
    transaction.meta.status = Err(solana_transaction_error::TransactionError::AccountInUse);
    let stats = SwapStats::default();

    Pipeline::builder()
        .datasource(FixtureDatasource(vec![Update::Transaction(Box::new(transaction))]))
        .instruction(MeteoraDlmmDecoder, stats.clone())
        .build()
        .unwrap()
        .run()
        .await
        .unwrap();

    assert!(stats.totals.lock().unwrap().is_empty());
}