Index token transfers

use {
    carbon_core::{
        error::CarbonResult, instruction::InstructionProcessorInputType, pipeline::Pipeline,
        processor::Processor,
    },
    carbon_log_metrics::LogMetrics,
    carbon_rpc_transaction_crawler_datasource::{ConnectionConfig, Filters, RpcTransactionCrawler},
    carbon_token_2022_decoder::{instructions::Token2022Instruction, Token2022Decoder},
    carbon_token_program_decoder::{instructions::TokenProgramInstruction, TokenProgramDecoder},
    solana_pubkey::{pubkey, Pubkey},
    std::sync::Arc,
};

/// The mint whose transfers we index: JUP.
const MINT: Pubkey = pubkey!("JUPyiwrYJFskUPiHa7hkeR8VUtAeFoSYbKedZNsDvCN");

#[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());

    // Every transaction that references the mint, newest first, then new ones as they land.
    let datasource = RpcTransactionCrawler::new(
        rpc_url,
        MINT,
        ConnectionConfig::default(),
        Filters::new(None, None, None),
        None,
    );

    Pipeline::builder()
        .datasource(datasource)
        .metrics(Arc::new(LogMetrics::new()))
        // A mint belongs to one token program; register both so the code works for any mint.
        .instruction(TokenProgramDecoder, TransferLogger)
        .instruction(Token2022Decoder, TransferLogger)
        .build()?
        .run()
        .await
}

struct TransferLogger;

fn print_transfer(amount: u64, decimals: u8, mint: Pubkey, from: Pubkey, to: Pubkey) {
    if mint != MINT {
        return; // a transfer of another token in the same transaction
    }
    let ui_amount = amount as f64 / 10f64.powi(decimals as i32);
    println!("{ui_amount:>14.6} from {from} to {to}");
}

impl Processor<InstructionProcessorInputType<'_, TokenProgramInstruction>> for TransferLogger {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, TokenProgramInstruction>,
    ) -> CarbonResult<()> {
        if let TokenProgramInstruction::TransferChecked { data, accounts, .. } = input.decoded_instruction {
            print_transfer(data.amount, data.decimals, accounts.mint, accounts.source, accounts.destination);
        }
        Ok(())
    }
}

impl Processor<InstructionProcessorInputType<'_, Token2022Instruction>> for TransferLogger {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, Token2022Instruction>,
    ) -> CarbonResult<()> {
        if let Token2022Instruction::TransferChecked { data, accounts, .. } = input.decoded_instruction {
            print_transfer(data.amount, data.decimals, accounts.mint, accounts.source, accounts.destination);
        }
        Ok(())
    }
}