Index swaps across DEXs

use {
    borsh::BorshDeserialize,
    carbon_core::{
        deserialize::CarbonDeserialize,
        error::CarbonResult,
        filter::{DeduplicationFilter, Filter},
        instruction::InstructionProcessorInputType,
        pipeline::Pipeline,
        processor::Processor,
    },
    carbon_log_metrics::LogMetrics,
    carbon_meteora_dlmm_decoder::{
        MeteoraDlmmDecoder,
        instructions::{CpiEvent as DlmmEvent, MeteoraDlmmInstruction},
    },
    carbon_pump_swap_decoder::{
        PumpSwapDecoder,
        instructions::{CpiEvent as PumpSwapEvent, PumpSwapInstruction},
    },
    carbon_raydium_cpmm_decoder::{
        RaydiumCpmmDecoder, events::swap_event::SwapEventEvent,
        instructions::RaydiumCpmmInstruction,
    },
    carbon_rpc_transaction_crawler_datasource::{ConnectionConfig, Filters, RpcTransactionCrawler},
    solana_pubkey::Pubkey,
    std::{sync::Arc, time::Duration},
};

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

    let rpc_url = std::env::var("RPC_URL")
        .unwrap_or_else(|_| "https://api.mainnet-beta.solana.com".to_string());
    // One crawler per program, all feeding the same pipeline. One request in flight each
    // keeps the three of them under the public endpoint's per-IP rate limit.
    let crawler = |program: Pubkey| {
        RpcTransactionCrawler::new(
            rpc_url.clone(),
            program,
            ConnectionConfig {
                max_concurrent_requests: 1,
                ..ConnectionConfig::default()
            },
            Filters::new(None, None, None),
            None,
        )
    };

    Pipeline::builder()
        .datasource(crawler(carbon_raydium_cpmm_decoder::PROGRAM_ID))
        .datasource(crawler(carbon_pump_swap_decoder::PROGRAM_ID))
        .datasource(crawler(carbon_meteora_dlmm_decoder::PROGRAM_ID))
        .metrics(Arc::new(LogMetrics::new()))
        .instruction_with_filters(RaydiumCpmmDecoder, SwapIndexer, dedupe())
        .instruction_with_filters(PumpSwapDecoder, SwapIndexer, dedupe())
        .instruction_with_filters(MeteoraDlmmDecoder, SwapIndexer, dedupe())
        .build()?
        .run()
        .await
}

/// A transaction that calls two of the programs is fetched by both crawlers. This drops the
/// second copy of each instruction, keyed by signature and position in the transaction.
fn dedupe() -> Vec<Box<dyn Filter>> {
    vec![Box::new(DeduplicationFilter::new(Duration::from_secs(600)))]
}

/// One executed swap, the same shape for every DEX. Amounts are raw token units.
struct Swap {
    dex: &'static str,
    pool: Pubkey,
    amount_in: u64,
    amount_out: u64,
}

struct SwapIndexer;

impl SwapIndexer {
    fn record(&self, signature: impl std::fmt::Display, swap: Swap) {
        // Replace with a database write; see /guides/postgres.
        println!(
            "{:<12} pool={} in={} out={} tx={signature}",
            swap.dex, swap.pool, swap.amount_in, swap.amount_out,
        );
    }
}

/// Raydium CPMM publishes `SwapEvent` with Anchor's `emit!`, as a `Program data:` log line,
/// not as a self-CPI. Wrapping the generated type lets `decode_log_events` read it.
#[derive(BorshDeserialize)]
struct LoggedCpmmSwap(SwapEventEvent);

impl CarbonDeserialize for LoggedCpmmSwap {
    const DISCRIMINATOR: &'static [u8] = &[64, 198, 205, 232, 38, 8, 113, 226];

    fn deserialize(data: &[u8]) -> Option<Self> {
        SwapEventEvent::decode(data).map(LoggedCpmmSwap)
    }
}

impl Processor<InstructionProcessorInputType<'_, RaydiumCpmmInstruction>> for SwapIndexer {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, RaydiumCpmmInstruction>,
    ) -> CarbonResult<()> {
        if !matches!(
            input.decoded_instruction,
            RaydiumCpmmInstruction::SwapBaseInput { .. }
                | RaydiumCpmmInstruction::SwapBaseOutput { .. }
        ) {
            return Ok(());
        }
        // The swap instruction's own log lines hold its SwapEvent.
        for LoggedCpmmSwap(e) in input.metadata.decode_log_events::<LoggedCpmmSwap>() {
            let swap = Swap {
                dex: "raydium-cpmm",
                pool: e.pool_id,
                amount_in: e.input_amount,
                amount_out: e.output_amount,
            };
            self.record(input.metadata.transaction_metadata.signature, swap);
        }
        Ok(())
    }
}

impl Processor<InstructionProcessorInputType<'_, PumpSwapInstruction>> for SwapIndexer {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, PumpSwapInstruction>,
    ) -> CarbonResult<()> {
        let PumpSwapInstruction::CpiEvent { data, .. } = input.decoded_instruction else {
            return Ok(());
        };
        // A buy pays the quote token (usually SOL) for the base token; a sell is the reverse.
        let swap = match data {
            PumpSwapEvent::BuyEvent(e) => Swap {
                dex: "pump-swap",
                pool: e.pool,
                amount_in: e.quote_amount_in,
                amount_out: e.base_amount_out,
            },
            PumpSwapEvent::SellEvent(e) => Swap {
                dex: "pump-swap",
                pool: e.pool,
                amount_in: e.base_amount_in,
                amount_out: e.quote_amount_out,
            },
            _ => return Ok(()),
        };
        self.record(input.metadata.transaction_metadata.signature, swap);
        Ok(())
    }
}

impl Processor<InstructionProcessorInputType<'_, MeteoraDlmmInstruction>> for SwapIndexer {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, MeteoraDlmmInstruction>,
    ) -> CarbonResult<()> {
        let MeteoraDlmmInstruction::CpiEvent { data, .. } = input.decoded_instruction else {
            return Ok(());
        };
        // DLMM emits both `Swap` and `Swap2Evt` for each swap, with the same amounts: read one.
        let DlmmEvent::Swap(e) = data else {
            return Ok(());
        };
        let swap = Swap {
            dex: "meteora-dlmm",
            pool: e.lb_pair,
            amount_in: e.amount_in,
            amount_out: e.amount_out,
        };
        self.record(input.metadata.transaction_metadata.signature, swap);
        Ok(())
    }
}