Streaming Mint Accounts

let mut request = helius_laserstream::grpc::SubscribeRequest::default();

// 1. Account sub: mint state tracking + cold-to-hot detection.
//    account_type == 1 (Mint) at byte offset 165.
request.accounts.insert(
    "light_mints".to_string(),
    SubscribeRequestFilterAccounts {
        owner: vec![LIGHT_TOKEN_PROGRAM_ID.to_string()],
        filters: vec![SubscribeRequestFilterAccountsFilter {
            filter: Some(Filter::Memcmp(SubscribeRequestFilterAccountsFilterMemcmp {
                offset: ACCOUNT_TYPE_OFFSET,
                data: Some(Data::Bytes(vec![1])),
            })),
        }],
        nonempty_txn_signature: Some(true),
        ..Default::default()
    },
);

// 2. Transaction sub: hot-to-cold detection.
request.transactions.insert(
    "light_token_txns".to_string(),
    SubscribeRequestFilterTransactions {
        vote: Some(false),
        failed: Some(false),
        account_include: vec![LIGHT_TOKEN_PROGRAM_ID.to_string()],
        ..Default::default()
    },
);

let (stream, _handle) = subscribe(config, request);
tokio::pin!(stream);

use helius_laserstream::grpc::subscribe_update::UpdateOneof;

Some(UpdateOneof::Account(account_update)) => {
    if let Some(account_info) = account_update.account {
        let pubkey: [u8; 32] = account_info.pubkey.as_slice().try_into().unwrap();

        match Mint::deserialize(&mut account_info.data.as_slice()) {
            Ok(mint) => {
                cold_cache.remove(&pubkey); // no longer cold
                cache.insert(pubkey, mint);
            }
            Err(e) => {
                eprintln!(
                    "Failed to deserialize mint {}: {}",
                    bs58::encode(&pubkey).into_string(),
                    e
                );
            }
        }
    }
}

use helius_laserstream::grpc::subscribe_update::UpdateOneof;

Some(UpdateOneof::Transaction(tx_update)) => {
    if let Some(ref tx_info) = tx_update.transaction {
        for pubkey in find_closed_accounts(tx_info) {
            if cache.remove(&pubkey).is_some() {
                // Async: fetch AccountInterface with ColdContext.
                // Cold accounts are inactive, so this completes well
                // before anyone tries to swap through them.
                let rpc = rpc.clone();
                let cold_cache = cold_cache.clone();
                tokio::spawn(async move {
                    if let Ok(Some(iface)) = rpc.get_account_interface(&pubkey, None).await {
                        cold_cache.insert(pubkey, iface);
                    }
                });
            }
        }
    }
}

fn find_closed_accounts(
    tx_info: &helius_laserstream::grpc::SubscribeUpdateTransactionInfo,
) -> Vec<[u8; 32]> {
    let meta = match &tx_info.meta {
        Some(m) => m,
        None => return vec![],
    };
    let msg = match tx_info.transaction.as_ref().and_then(|t| t.message.as_ref()) {
        Some(m) => m,
        None => return vec![],
    };

    let mut all_keys: Vec<&[u8]> = msg.account_keys.iter().map(|k| k.as_slice()).collect();
    all_keys.extend(meta.loaded_writable_addresses.iter().map(|k| k.as_slice()));
    all_keys.extend(meta.loaded_readonly_addresses.iter().map(|k| k.as_slice()));

    let mut closed = Vec::new();
    for (i, key) in all_keys.iter().enumerate() {
        if key.len() == 32
            && meta.pre_balances.get(i).copied().unwrap_or(0) > 0
            && meta.post_balances.get(i).copied().unwrap_or(1) == 0
        {
            closed.push(<[u8; 32]>::try_from(*key).unwrap());
        }
    }
    closed
}

fn extract_metadata(mint: &Mint) -> Option<(String, String, String)> {
    let extensions = mint.extensions.as_ref()?;

    for ext in extensions {
        if let ExtensionStruct::TokenMetadata(m) = ext {
            let name = String::from_utf8_lossy(&m.name).to_string();
            let symbol = String::from_utf8_lossy(&m.symbol).to_string();
            let uri = String::from_utf8_lossy(&m.uri).to_string();
            return Some((name, symbol, uri));
        }
    }
    None
}