Write a custom datasource

use {
    async_trait::async_trait,
    carbon_core::{
        account::AccountProcessorInputType,
        datasource::{AccountDeletion, AccountUpdate, Datasource, DatasourceId, Update, UpdateType},
        error::CarbonResult,
        pipeline::Pipeline,
        processor::Processor,
    },
    carbon_token_program_decoder::{accounts::TokenProgramAccount, TokenProgramDecoder},
    solana_client::nonblocking::rpc_client::RpcClient,
    solana_commitment_config::CommitmentConfig,
    solana_pubkey::Pubkey,
    std::{collections::HashMap, time::Duration},
    tokio::sync::mpsc::Sender,
    tokio_util::sync::CancellationToken,
};

/// Polls `getMultipleAccounts` and emits an update whenever an account changes.
pub struct AccountPoller {
    pub rpc_url: String,
    pub accounts: Vec<Pubkey>,
    pub poll_interval: Duration,
}

#[async_trait]
impl Datasource for AccountPoller {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: Sender<(Update, DatasourceId)>,
        cancellation_token: CancellationToken,
    ) -> CarbonResult<()> {
        let rpc = RpcClient::new(self.rpc_url.clone());
        let mut last_seen = HashMap::new();

        loop {
            let result = tokio::select! {
                _ = cancellation_token.cancelled() => return Ok(()),
                result = rpc.get_multiple_accounts_with_commitment(
                    &self.accounts,
                    CommitmentConfig::confirmed(),
                ) => result,
            };

            match result {
                Ok(response) => {
                    let slot = response.context.slot;
                    for (pubkey, account) in self.accounts.iter().zip(response.value) {
                        // A missing account is reported with zero lamports, so
                        // `into_update` turns it into an `AccountDeletion`.
                        let account = account.unwrap_or_default();
                        if last_seen.get(pubkey) == Some(&account) {
                            continue; // unchanged since the last poll
                        }
                        last_seen.insert(*pubkey, account.clone());

                        let update = AccountUpdate {
                            pubkey: *pubkey,
                            account,
                            slot,
                            transaction_signature: None,
                        }
                        .into_update();

                        if sender.send((update, id.clone())).await.is_err() {
                            return Ok(());
                        }
                    }
                }
                Err(error) => log::warn!("getMultipleAccounts failed: {error}"),
            }

            tokio::select! {
                _ = cancellation_token.cancelled() => return Ok(()),
                _ = tokio::time::sleep(self.poll_interval) => {}
            }
        }
    }

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

struct MintLogger;

impl Processor<AccountProcessorInputType<'_, TokenProgramAccount>> for MintLogger {
    async fn process(
        &mut self,
        input: &AccountProcessorInputType<'_, TokenProgramAccount>,
    ) -> CarbonResult<()> {
        if let TokenProgramAccount::Mint(mint) = &input.decoded_account.data {
            println!(
                "slot={} mint={} supply={} decimals={}",
                input.metadata.slot, input.metadata.pubkey, mint.supply, mint.decimals
            );
        }
        Ok(())
    }
}

struct DeletionLogger;

impl Processor<AccountDeletion> for DeletionLogger {
    async fn process(&mut self, deletion: &AccountDeletion) -> CarbonResult<()> {
        println!("slot={} closed {}", deletion.slot, deletion.pubkey);
        Ok(())
    }
}

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

    let datasource = AccountPoller {
        rpc_url: "https://api.mainnet-beta.solana.com".to_string(),
        accounts: vec![
            Pubkey::from_str_const("EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"), // USDC mint
            Pubkey::from_str_const("Es9vMFrzaCERmJfrF4H2FYD4KCoNkY11McCe8BenwNYB"), // USDT mint
        ],
        poll_interval: Duration::from_secs(2),
    };

    Pipeline::builder()
        .datasource(datasource)
        .account(TokenProgramDecoder, MintLogger)
        .account_deletions(DeletionLogger)
        .build()?
        .run()
        .await
}

use {
    carbon_core::{
        account::AccountProcessorInputType,
        datasource::AccountUpdate,
        error::CarbonResult,
        pipeline::Pipeline,
        processor::Processor,
    },
    carbon_stream_message_datasource::{StreamMessageClient, UnifiedMessage},
    carbon_token_program_decoder::{accounts::TokenProgramAccount, TokenProgramDecoder},
    solana_client::nonblocking::rpc_client::RpcClient,
    solana_pubkey::Pubkey,
};

const USDC_MINT: Pubkey = Pubkey::from_str_const("EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v");

struct MintLogger;

impl Processor<AccountProcessorInputType<'_, TokenProgramAccount>> for MintLogger {
    async fn process(
        &mut self,
        input: &AccountProcessorInputType<'_, TokenProgramAccount>,
    ) -> CarbonResult<()> {
        if let TokenProgramAccount::Mint(mint) = &input.decoded_account.data {
            println!("slot={} supply={}", input.metadata.slot, mint.supply);
        }
        Ok(())
    }
}

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

    let (tx, rx) = tokio::sync::mpsc::channel::<UnifiedMessage>(1_000);

    // Your producer: a Kafka consumer, a websocket, a file replay... Here, three RPC reads.
    tokio::spawn(async move {
        let rpc = RpcClient::new("https://api.mainnet-beta.solana.com".to_string());
        for _ in 0..3 {
            if let Ok(response) = rpc
                .get_account_with_commitment(&USDC_MINT, rpc.commitment())
                .await
            {
                if let Some(account) = response.value {
                    let update = AccountUpdate {
                        pubkey: USDC_MINT,
                        account,
                        slot: response.context.slot,
                        transaction_signature: None,
                    };
                    if tx.send(UnifiedMessage::Account(update)).await.is_err() {
                        break;
                    }
                }
            }
            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
        }
        // Dropping `tx` closes the stream; the pipeline then finishes.
    });

    Pipeline::builder()
        .datasource(StreamMessageClient::new(rx))
        .account(TokenProgramDecoder, MintLogger)
        .build()?
        .run()
        .await
}