sfm-sdk 0.2.32

use std::collections::HashMap;
use std::iter::Map;
use ts_rs::TS;

#[macro_use]
extern crate string_enum;

pub mod helpers;
pub mod processors;
pub mod utils;

use serde::{Deserialize, Serialize};
use solana_program::clock::UnixTimestamp;
use solana_sdk::transaction::TransactionError;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use tokio::select;
use tokio::signal::unix::{signal, SignalKind};
use tokio::sync::watch::{Receiver, Sender};
use tracing::info;

#[derive(Clone, Serialize, Deserialize)]
pub struct QueueItem {
    pub numbers: Vec<u64>,
    pub programs: Vec<String>,
}

#[derive(Clone, Serialize, Deserialize)]
pub struct TransactionProgram {
    /// Transaction hash..
    pub transaction_hash: String,
    /// The programs involved in the transaction
    pub program: String,
    /// Time of transaction
    pub timestamp: u64,
}

pub type AccountMap = Map<String, AccountLatest>;

/// Struct that stashes the latest bits of an account.
pub struct AccountLatest {
    /// The balance in the account.  Modifiable by programs.
    pub balance: u64,
    /// The data held in this account.  Modifiable by programs.
    pub data: Vec<u8>,
    /// Account that owns this account, defaults to System program.
    pub owner: String,
    /// The epoch at which this account will next owe rent
    pub rent_epoch: u64,
    /// The optional mint relevant to this account (At that block in time)
    pub mint: Option<String>,
    /// Timestamp at point in time relevant to block
    pub timestamp: u64,
    /// Block at point in time
    pub block: u64
}

/// Account Hash, Owner (New), Mint (New), Amount (New), Timestamp, Block
/// pub type AccountEntry = (String, Option<String>, Option<String>, Option<u64>, u64, u64);
pub struct AccountEntry {
    /// The unique identifier for the account
    pub hash: String,
    /// The optional amount relevant to this account (At that block in time)
    pub balance: Option<u64>,
    /// The optional data relevant to this account (At that block in time, and if there is any data
    /// for diff generation)
    pub data: Option<Vec<u8>>,
    /// The map that differentiates vLatest and the vUpcoming
    pub data_diff_map: Option<HashMap<usize, u8>>,
    /// The optional owner relevant to this account (At that block in time)
    pub owner: Option<String>,
    /// The optional mint relevant to this account (At that block in time)
    pub mint: Option<String>,
    /// Timestamp at point in time relevant to block
    pub timestamp: u64,
    /// Block at point in time
    pub block: u64
}

#[derive(Clone, Serialize, Deserialize)]
pub struct AccountProgram {
    // The account pub key relating to this transaction.
    pub account: String,
    // The programs involved with the account
    pub program: String,
    // Block number responsible for this instance.
    pub block: u64,
    // Time when it is created according to Solana's on-chain clock
    pub timestamp: u64,
}

#[derive(Clone, Serialize, Deserialize)]
pub struct AccountTransaction {
    // The transaction this transaction belongs to.
    pub transaction_hash: String,
    // The account pub key relating to this transaction.
    pub account: String,
    // Block number responsible for this instance.
    pub block: u64,
    // Time when it is created according to Solana's on-chain clock
    pub timestamp: u64,
}

impl PartialEq for AccountTransaction {
    fn eq(&self, other: &Self) -> bool {
        self.block == other.block
            && self.account == other.account
            && self.timestamp == other.timestamp
    }
}

#[derive(Clone, Serialize, Deserialize)]
pub struct BatchedAccountTransaction {
    // Transaction hashes relevant to this account at this point in time
    pub hashes: Vec<String>,
    // Time when it is created according to Solana's on-chain clock
    pub timestamp: u64,
}

#[derive(Clone, Serialize, Deserialize, TS)]
#[ts(export)]
pub struct Block {
    /// Basically the epoch this block belongs to
    pub epoch: u32,
    /// Parent block hash of the current block
    pub previous_hash: String,
    /// Validator producing said block
    pub producer: String,
    /// This block's hash
    pub hash: String,
    /// Parent's block number
    pub parent_number: u64,
    /// This block's number
    pub number: u64,
    /// Amount of data contained within the block
    pub data_size: u64,
    /// Total count of transactions in the block
    pub number_of_transactions: u32,
    /// Total number of successful transactions
    pub successful_transactions: u32,
    /// Total number of vote-related transactions
    pub vote_transactions: u32,
    /// Total transaction fees accumulated in the transactions within this block
    pub total_tx_fees: u64,
    /// Total number of rewards
    pub number_of_rewards: u32,
    /// Total amount of rewards accrued in this block
    pub total_reward_amount: u64,
    /// Total amount of compute units consumed
    pub total_compute_units_consumed: u64,
    /// Absolute limit of compute units
    pub total_compute_units_limit: u64,
    /// Time this block was proposed
    pub block_time: u64,
}

#[derive(Clone, Serialize, Deserialize)]
pub enum VersionedTransfer {
    V0(Transfer),
    V1(EnrichedTransfer),
}

/// Record an instance of a transaction transfer at any given time.
/// Routing key
/// <account>#<mint>#<timestamp>
/// Schema column qualifiers
/// type —> receive | send
/// amount (in finalised format, e.g. 0.000031 <eth, excluded>)
/// txHash
///
#[derive(Clone, Serialize, Deserialize)]
pub struct Transfer {
    // The transaction this instruction belongs to.
    pub transaction_hash: String,
    // Status of the transaction,
    pub status: u16,
    // The account that will give up the amount.
    pub source: String,
    // Should this be a token-based transfer, this will be the associated token account of the source.
    pub source_association: Option<String>,
    // The account that will receive the amount.
    pub destination: String,
    // Should this be a token-based transfer, this will be the associated token account of the destination.
    pub destination_association: Option<String>,
    // If this is empty, the balance relates to lamports. If its NOT empty, the balance relates to the
    // token in question.
    pub token: Option<String>,
    // The amount transferred
    pub amount: u64,
    // Epoch time for when this input was added to the db.
    pub timestamp: u64,
}

impl From<EnrichedTransfer> for Transfer {
    fn from(transfer: EnrichedTransfer) -> Self {
        Self {
            transaction_hash: "".into(),
            status: transfer.status as u16,
            source: transfer.source,
            source_association: transfer.source_association,
            destination: transfer.destination,
            destination_association: transfer.destination_association,
            token: transfer.token,
            amount: transfer.amount,
            timestamp: transfer.timestamp,
        }
    }
}

impl From<Option<TransactionError>> for EnrichedTransferStatus {
    fn from(err: Option<TransactionError>) -> Self {
        match err {
            Some(error) => match error {
                TransactionError::AccountInUse => Self::AccountInUse,
                TransactionError::AccountLoadedTwice => Self::AccountLoadedTwice,
                TransactionError::AccountNotFound => Self::AccountNotFound,
                TransactionError::ProgramAccountNotFound => Self::ProgramAccountNotFound,
                TransactionError::InsufficientFundsForFee => Self::InsufficientFundsForFee,
                TransactionError::InvalidAccountForFee => Self::InvalidAccountForFee,
                TransactionError::AlreadyProcessed => Self::AlreadyProcessed,
                TransactionError::BlockhashNotFound => Self::BlockhashNotFound,
                TransactionError::InstructionError(_, _) => Self::InstructionError,
                TransactionError::CallChainTooDeep => Self::CallChainTooDeep,
                TransactionError::MissingSignatureForFee => Self::MissingSignatureForFee,
                TransactionError::InvalidAccountIndex => Self::InvalidAccountIndex,
                TransactionError::SignatureFailure => Self::SignatureFailure,
                TransactionError::InvalidProgramForExecution => Self::InvalidProgramForExecution,
                TransactionError::SanitizeFailure => Self::SanitizeFailure,
                TransactionError::ClusterMaintenance => Self::ClusterMaintenance,
                TransactionError::AccountBorrowOutstanding => Self::AccountBorrowOutstanding,
                TransactionError::WouldExceedMaxBlockCostLimit => {
                    Self::WouldExceedMaxBlockCostLimit
                }
                TransactionError::UnsupportedVersion => Self::UnsupportedVersion,
                TransactionError::InvalidWritableAccount => Self::InvalidWritableAccount,
                TransactionError::WouldExceedMaxAccountCostLimit => {
                    Self::WouldExceedMaxAccountCostLimit
                }
                TransactionError::WouldExceedAccountDataBlockLimit => {
                    Self::WouldExceedAccountDataBlockLimit
                }
                TransactionError::TooManyAccountLocks => Self::TooManyAccountLocks,
                TransactionError::AddressLookupTableNotFound => Self::AddressLookupTableNotFound,
                TransactionError::InvalidAddressLookupTableOwner => {
                    Self::InvalidAddressLookupTableOwner
                }
                TransactionError::InvalidAddressLookupTableData => {
                    Self::InvalidAddressLookupTableData
                }
                TransactionError::InvalidAddressLookupTableIndex => {
                    Self::InvalidAddressLookupTableIndex
                }
                TransactionError::InvalidRentPayingAccount => Self::InvalidRentPayingAccount,
                TransactionError::WouldExceedMaxVoteCostLimit => Self::WouldExceedMaxVoteCostLimit,
                TransactionError::WouldExceedAccountDataTotalLimit => {
                    Self::WouldExceedAccountDataTotalLimit
                }
                TransactionError::DuplicateInstruction(_) => Self::DuplicateInstruction,
                TransactionError::InsufficientFundsForRent { .. } => Self::InsufficientFundsForRent,
            },
            None => Self::Successful,
        }
    }
}

#[derive(Copy, Clone, PartialEq, StringEnum)]
pub enum EnrichedTransferStatus {
    /// `Unknown`
    Null = 0,
    /// `Successful`
    Successful = 1,
    /// `AccountInUse`
    AccountInUse = 2,
    /// `AccountLoadedTwice`
    AccountLoadedTwice = 3,
    /// `AccountNotFound`
    AccountNotFound = 4,
    /// `ProgramAccountNotFound`
    ProgramAccountNotFound = 5,
    /// `InsufficientFundsForFee`
    InsufficientFundsForFee = 6,
    /// `InvalidAccountForFee`
    InvalidAccountForFee = 7,
    /// `AlreadyProcessed`
    AlreadyProcessed = 8,
    /// `BlockhashNotFound`
    BlockhashNotFound = 9,
    /// `InstructionError`
    InstructionError = 10,
    /// `CallChainTooDeep`
    CallChainTooDeep = 11,
    /// `MissingSignatureForFee`
    MissingSignatureForFee = 12,
    /// `InvalidAccountIndex`
    InvalidAccountIndex = 13,
    /// `SignatureFailure`
    SignatureFailure = 14,
    /// `InvalidProgramForExecution`
    InvalidProgramForExecution = 15,
    /// `SanitizeFailure`
    SanitizeFailure = 16,
    /// `ClusterMaintenance`
    ClusterMaintenance = 17,
    /// `AccountBorrowOutstanding`
    AccountBorrowOutstanding = 18,
    /// `WouldExceedMaxBlockCostLimit`
    WouldExceedMaxBlockCostLimit = 19,
    /// `UnsupportedVersion`
    UnsupportedVersion = 20,
    /// `InvalidWritableAccount`
    InvalidWritableAccount = 21,
    /// `WouldExceedMaxAccountCostLimit`
    WouldExceedMaxAccountCostLimit = 22,
    /// `WouldExceedAccountDataBlockLimit`
    WouldExceedAccountDataBlockLimit = 23,
    /// `TooManyAccountLocks`
    TooManyAccountLocks = 24,
    /// `AddressLookupTableNotFound`
    AddressLookupTableNotFound = 25,
    /// `InvalidAddressLookupTableOwner`
    InvalidAddressLookupTableOwner = 26,
    /// `InvalidAddressLookupTableData`
    InvalidAddressLookupTableData = 27,
    /// `InvalidAddressLookupTableIndex`
    InvalidAddressLookupTableIndex = 28,
    /// `InvalidRentPayingAccount`
    InvalidRentPayingAccount = 29,
    /// `WouldExceedMaxVoteCostLimit`
    WouldExceedMaxVoteCostLimit = 30,
    /// `WouldExceedAccountDataTotalLimit`
    WouldExceedAccountDataTotalLimit = 31,
    /// `DuplicateInstruction`
    DuplicateInstruction = 32,
    /// `InsufficientFundsForRent`
    InsufficientFundsForRent = 33,
}

impl EnrichedTransferStatus {
    pub fn from_u16(value: u16) -> Self {
        match value {
            1 => Self::Successful,
            2 => Self::AccountInUse,
            3 => Self::AccountLoadedTwice,
            4 => Self::AccountNotFound,
            5 => Self::ProgramAccountNotFound,
            6 => Self::InsufficientFundsForFee,
            7 => Self::InvalidAccountForFee,
            8 => Self::AlreadyProcessed,
            9 => Self::BlockhashNotFound,
            10 => Self::InstructionError,
            11 => Self::CallChainTooDeep,
            12 => Self::MissingSignatureForFee,
            13 => Self::InvalidAccountIndex,
            14 => Self::SignatureFailure,
            15 => Self::InvalidProgramForExecution,
            16 => Self::SanitizeFailure,
            17 => Self::ClusterMaintenance,
            18 => Self::AccountBorrowOutstanding,
            19 => Self::WouldExceedMaxBlockCostLimit,
            20 => Self::UnsupportedVersion,
            21 => Self::InvalidWritableAccount,
            22 => Self::WouldExceedMaxAccountCostLimit,
            23 => Self::WouldExceedAccountDataBlockLimit,
            24 => Self::TooManyAccountLocks,
            25 => Self::AddressLookupTableNotFound,
            26 => Self::InvalidAddressLookupTableOwner,
            27 => Self::InvalidAddressLookupTableData,
            28 => Self::InvalidAddressLookupTableIndex,
            29 => Self::InvalidRentPayingAccount,
            30 => Self::WouldExceedMaxVoteCostLimit,
            31 => Self::WouldExceedAccountDataTotalLimit,
            32 => Self::DuplicateInstruction,
            33 => Self::InsufficientFundsForRent,
            _ => Self::Null,
        }
    }
    pub fn to_16(&self) -> u16 {
        *self as u16
    }
}

/// For (Transaction Hash, ix idx, inner ix idx) to EnrichedTransfer binding
pub type EnrichedTransferPair = (String, i16, i16, EnrichedTransfer);

/// Routing key
/// <account>#<mint>#<timestamp>
/// Schema column qualifiers
/// type —> receive | send
/// amount (in finalised format, e.g. 0.000031 <eth, excluded>)
/// txHash
#[derive(Clone, Serialize, Deserialize)]
pub struct EnrichedTransfer {
    // Dictates the actions resulting in this transfer.
    pub action: String,
    // Status of the transaction,
    pub status: EnrichedTransferStatus,
    // The account that will give up the amount.
    pub source: String,
    // Should this be a token-based transfer, this will be the associated token account of the source.
    pub source_association: Option<String>,
    // The account that will receive the amount.
    pub destination: String,
    // Should this be a token-based transfer, this will be the associated token account of the destination.
    pub destination_association: Option<String>,
    // If this is empty, the balance relates to lamports. If its NOT empty, the balance relates to the
    // token in question.
    pub token: Option<String>,
    // The amount transferred
    pub amount: u64,
    // Epoch time for when this input was added to the db.
    pub timestamp: u64,
}

#[derive(Clone, Deserialize, Serialize)]
pub struct TransactionInfo {
    pub hash: String,
    pub status: u16,
    pub fee: u64,
    pub total_cu_consumed: u64,
    pub total_cu_limit: u64,
    pub transfers: Vec<EnrichedTransferPair>,
    pub account_transactions: Vec<AccountTransaction>,
}

impl TransactionInfo {
    pub fn old_transfers(&mut self) -> Vec<EnrichedTransfer> {
        self.transfers.iter().map(|trf| trf.3.clone()).collect()
    }
}

#[derive(Clone, Deserialize, Serialize)]
pub struct ProgramConsumption {
    pub tx: String,
    pub status: u16,
    pub line: u32,
    pub program: String,
    pub cu_consumed: u64,
    pub cu_limit: u64,
    pub timestamp: UnixTimestamp,
}

pub fn is_not_transfer(address: &str) -> bool {
    !vec![
        // System Program
        "11111111111111111111111111111111",
        // Config Program
        "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA",
        // Associated Token Account Program
        "ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL",
    ]
    .contains(&address)
}

/// Listens for the machine shutdown signals via a SysSigListener with a `broadcast::Receiver`.
///
/// Shutdown is signalled using a `broadcast::Receiver`. Only a single value is
/// ever sent. Once a value has been sent via the broadcast channel, the server
/// should shutdown.
///
/// The `Shutdown` struct listens for the signal and tracks that the signal has
/// been received. Callers may query for whether the shutdown signal has been
/// received or not.
#[derive(Debug)]
pub struct SysSigReceiver {
    /// `true` if the shutdown signal has been received
    shutdown: Arc<AtomicBool>,

    /// The receive half of the channel used to listen for shutdown.
    notify: Receiver<()>,
}

impl SysSigReceiver {
    /// Create a new `Shutdown` backed by the given `broadcast::Receiver`.
    pub fn new(shutdown: Arc<AtomicBool>, notify: Receiver<()>) -> SysSigReceiver {
        SysSigReceiver { shutdown, notify }
    }

    /// Returns `true` if the shutdown signal has been received.
    pub fn is_shutdown(&self) -> bool {
        self.shutdown.load(Ordering::SeqCst)
    }

    /// Receive the shutdown notice, waiting if necessary.
    pub async fn recv(&mut self) {
        // If the shutdown signal has already been received, then return
        // immediately.
        if self.is_shutdown() {
            return;
        }

        // Cannot receive a "lag error" as only one value is ever sent.
        let _ = self.notify.changed().await;

        if !self.is_shutdown() {
            self.shutdown.store(true, Ordering::SeqCst);
        }
    }
}

/// System Signal Listener
/// A simple struct to listen for shutdown signals from system, and for listeners to receive a
/// subscriber pipe.
pub struct SysSigListener {
    shutdown: Arc<AtomicBool>,
    notifier: Sender<()>,
}

impl SysSigListener {
    pub fn new(notifier: Sender<()>) -> Self {
        Self {
            shutdown: Arc::new(AtomicBool::new(false)),
            notifier,
        }
    }

    /// Initiates the watchdog sequence, listening to signals from the host.
    pub async fn watchdog(self) {
        info!("Watchdog turned on!");

        let mut alarm_sig = signal(SignalKind::alarm()).expect("Alarm stream failed.");
        let mut hangup_sig = signal(SignalKind::hangup()).expect("Hangup stream failed.");
        let mut int_sig = signal(SignalKind::interrupt()).expect("Interrupt stream failed.");
        let mut quit_sig = signal(SignalKind::quit()).expect("Quit stream failed.");
        let mut term_sig = signal(SignalKind::terminate()).expect("Terminate stream failed.");

        select! {
            _ = tokio::signal::ctrl_c() => {
                info!("CTRL+C received, terminating now!");
            },
            _ = alarm_sig.recv() => {
                info!("SIGALRM received, terminating now!");
            }
            _ = hangup_sig.recv() => {
                info!("SIGHUP received, terminating now!");
            }
            _ = int_sig.recv() => {
                info!("SIGINT received, terminating now!");
            }
            _ = quit_sig.recv() => {
                info!("SIGQUIT received, terminating now!");
            }
            _ = term_sig.recv() => {
                info!("SIGTERM received, terminating now!");
            }
        }

        self.forced_shutdown();
    }

    pub fn forced_shutdown(self) {
        info!("Shutting down!");
        (*self.shutdown).store(true, Ordering::SeqCst);

        drop(self.notifier);
    }

    /// Retrieves a listener for the watchdog.
    pub fn get_receiver(&mut self) -> SysSigReceiver {
        SysSigReceiver::new(Arc::clone(&self.shutdown), self.notifier.subscribe())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;
    use tokio::sync::watch;

    #[tokio::test]
    async fn test_shutdown() {
        // Establish a channel broadcaster
        let (notify_shutdown, _) = watch::channel(());
        // Establish the SysSigListener with the shutdown notifier (the channel sender)
        let mut listener = SysSigListener::new(notify_shutdown);
        // Establish the receiver in conjunction with the listener by obtaining a receiver.
        let receiver = listener.get_receiver();

        // Simulate listener.watchdog().await; and sleep for 2 seconds before termination.
        tokio::time::sleep(Duration::from_millis(2000)).await;
        listener.forced_shutdown();

        assert_eq!(receiver.shutdown.load(Ordering::SeqCst), true);
    }
}