block_store.rs

use std::{collections::HashMap, sync::Arc, time::Duration};

use graph::{components::store::BLOCK_CACHE_SIZE, parking_lot::RwLock};

use anyhow::anyhow;
use async_trait::async_trait;
use diesel::{sql_query, ExpressionMethods as _, QueryDsl};
use diesel_async::{scoped_futures::ScopedFutureExt, RunQueryDsl};
use graph::{
    blockchain::ChainIdentifier,
    components::store::{BlockStore as BlockStoreTrait, QueryPermit},
    derive::CheapClone,
    prelude::{error, info, warn, BlockNumber, BlockPtr, Logger, ENV_VARS},
    slog::o,
};
use graph::{
    components::{network_provider::ChainName, store::ChainIdStore},
    prelude::ChainStore as _,
};
use graph::{internal_error, prelude::CheapClone};
use graph::{prelude::StoreError, util::timed_cache::TimedCache};

use crate::{
    chain_head_listener::ChainHeadUpdateSender,
    chain_store::{ChainStoreMetrics, Storage},
    pool::ConnectionPool,
    primary::Mirror as PrimaryMirror,
    AsyncPgConnection, ChainStore, NotificationSender, Shard, PRIMARY_SHARD,
};

use self::primary::Chain;

#[cfg(debug_assertions)]
pub const FAKE_NETWORK_SHARED: &str = "fake_network_shared";

// Highest version of the database that the executable supports.
// To be incremented on each breaking change to the database.
const SUPPORTED_DB_VERSION: i64 = 3;

pub mod primary {
    use std::convert::TryFrom;

    use diesel::{delete, insert_into, update, ExpressionMethods, OptionalExtension, QueryDsl};
    use diesel_async::RunQueryDsl;
    use graph::{
        blockchain::{BlockHash, ChainIdentifier},
        internal_error,
        prelude::StoreError,
    };

    use crate::{chain_store::Storage, AsyncPgConnection};
    use crate::{ConnectionPool, Shard};

    table! {
            chains(id) {
                id                 -> Integer,
                name               -> Text,
                net_version        -> Text,
                genesis_block_hash -> Text,
                shard              -> Text,
                namespace          -> Text,
        }
    }

    /// Information about the mapping of chains to storage shards. We persist
    /// this information in the database to make it possible to detect a
    /// change in the configuration file that doesn't match what is in the database
    #[derive(Clone, Queryable)]
    pub struct Chain {
        pub id: i32,
        pub name: String,
        pub net_version: String,
        pub genesis_block: String,
        pub shard: Shard,
        pub storage: Storage,
    }

    impl Chain {
        pub fn network_identifier(&self) -> Result<ChainIdentifier, StoreError> {
            Ok(ChainIdentifier {
                net_version: self.net_version.clone(),
                genesis_block_hash: BlockHash::try_from(self.genesis_block.as_str()).map_err(
                    |e| {
                        internal_error!(
                            "the genesis block hash `{}` for chain `{}` is not a valid hash: {}",
                            self.genesis_block,
                            self.name,
                            e
                        )
                    },
                )?,
            })
        }
    }

    pub async fn load_chains(conn: &mut AsyncPgConnection) -> Result<Vec<Chain>, StoreError> {
        Ok(chains::table.load(conn).await?)
    }

    pub async fn find_chain(
        conn: &mut AsyncPgConnection,
        name: &str,
    ) -> Result<Option<Chain>, StoreError> {
        Ok(chains::table
            .filter(chains::name.eq(name))
            .first(conn)
            .await
            .optional()?)
    }

    pub async fn add_chain(
        conn: &mut AsyncPgConnection,
        name: &str,
        shard: &Shard,
        ident: ChainIdentifier,
    ) -> Result<Chain, StoreError> {
        // For tests, we want to have a chain that still uses the
        // shared `ethereum_blocks` table
        #[cfg(debug_assertions)]
        if name == super::FAKE_NETWORK_SHARED {
            insert_into(chains::table)
                .values((
                    chains::name.eq(name),
                    chains::namespace.eq("public"),
                    chains::net_version.eq(&ident.net_version),
                    chains::genesis_block_hash.eq(ident.genesis_block_hash.hash_hex()),
                    chains::shard.eq(shard.as_str()),
                ))
                .returning(chains::namespace)
                .get_result::<Storage>(conn)
                .await
                .map_err(StoreError::from)?;
            return Ok(chains::table
                .filter(chains::name.eq(name))
                .first(conn)
                .await?);
        }

        insert_into(chains::table)
            .values((
                chains::name.eq(name),
                chains::net_version.eq(&ident.net_version),
                chains::genesis_block_hash.eq(ident.genesis_block_hash.hash_hex()),
                chains::shard.eq(shard.as_str()),
            ))
            .returning(chains::namespace)
            .get_result::<Storage>(conn)
            .await
            .map_err(StoreError::from)?;
        Ok(chains::table
            .filter(chains::name.eq(name))
            .first(conn)
            .await?)
    }

    pub(super) async fn drop_chain(pool: &ConnectionPool, name: &str) -> Result<(), StoreError> {
        let mut conn = pool.get_permitted().await?;

        delete(chains::table.filter(chains::name.eq(name)))
            .execute(&mut conn)
            .await?;
        Ok(())
    }

    // update chain name where chain name is 'name'
    pub async fn update_chain_name(
        conn: &mut AsyncPgConnection,
        name: &str,
        new_name: &str,
    ) -> Result<(), StoreError> {
        update(chains::table.filter(chains::name.eq(name)))
            .set(chains::name.eq(new_name))
            .execute(conn)
            .await?;
        Ok(())
    }
}

/// The store that chains use to maintain their state and cache often used
/// data from a chain. The `BlockStore` maintains information about all
/// configured chains, and serves as a directory of chains, whereas the
/// [ChainStore] manages chain-specific data. The `BlockStore` is sharded so
/// that each chain can use, depending on configuration, a separate database.
/// Regardless of configuration, the data for each chain is stored in its
/// own database namespace, though for historical reasons, the code also deals
/// with a shared table for the block and call cache for chains in the
/// `public` namespace.
///
/// The `BlockStore` uses the table `public.chains` to keep immutable
/// information about each chain, including a mapping from the chain name
/// to the shard holding the chain specific data, and the database namespace
/// for that chain.
///
/// Chains are identified by the name with which they are configured in the
/// configuration file. Once a chain has been used with the system, it is
/// not possible to change its configuration, in particular, the database
/// shard and namespace, and the genesis block and net version must not
/// change between runs of `graph-node`
#[derive(Clone, CheapClone)]
pub struct BlockStore {
    inner: Arc<Inner>,
}

impl std::ops::Deref for BlockStore {
    type Target = Inner;

    fn deref(&self) -> &Self::Target {
        &self.inner
    }
}

pub struct Inner {
    logger: Logger,
    /// Map chain names to the corresponding store. This map is updated
    /// dynamically with new chains if an operation would require a chain
    /// that is not yet in `stores`. It is initialized with all chains
    /// known to the system at startup, either from configuration or from
    /// previous state in the database.
    stores: RwLock<HashMap<String, Arc<ChainStore>>>,
    /// We keep this information so we can create chain stores during
    /// startup. The triple is (network, shard, cache_size)
    shards: Vec<(String, Shard, BlockNumber)>,
    pools: HashMap<Shard, ConnectionPool>,
    sender: Arc<NotificationSender>,
    mirror: PrimaryMirror,
    chain_head_cache: TimedCache<String, HashMap<String, BlockPtr>>,
    chain_store_metrics: Arc<ChainStoreMetrics>,
}

impl BlockStore {
    /// Create a new `BlockStore` by creating a `ChainStore` for each entry
    /// in `networks`. The creation process checks that the configuration for
    /// existing chains has not changed from the last time `graph-node` ran, and
    /// creates new entries in its chain directory for chains we had not used
    /// previously. It also creates a `ChainStore` for each chain that was used
    /// in previous runs of the node, regardless of whether it is mentioned in
    /// `chains` to ensure that queries against such chains will succeed.
    ///
    /// Each entry in `chains` gives the chain name, the network identifier,
    /// and the name of the database shard for the chain. The `ChainStore` for
    /// a chain uses the pool from `pools` for the given shard.
    pub async fn new(
        logger: Logger,
        // (network, shard, cache_size)
        shards: Vec<(String, Shard, BlockNumber)>,
        // shard -> pool
        pools: HashMap<Shard, ConnectionPool>,
        sender: Arc<NotificationSender>,
        chain_store_metrics: Arc<ChainStoreMetrics>,
    ) -> Result<Self, StoreError> {
        // Cache chain head pointers for this long when returning
        // information from `chain_head_pointers`
        const CHAIN_HEAD_CACHE_TTL: Duration = Duration::from_secs(2);

        let mirror = PrimaryMirror::new(&pools);
        let existing_chains = mirror
            .read_async(|conn| primary::load_chains(conn).scope_boxed())
            .await?;
        let chain_head_cache = TimedCache::new(CHAIN_HEAD_CACHE_TTL);
        let chains = shards.clone();

        let inner = Arc::new(Inner {
            logger,
            stores: RwLock::new(HashMap::new()),
            shards,
            pools,
            sender,
            mirror,
            chain_head_cache,
            chain_store_metrics,
        });
        let block_store = Self { inner };

        // For each configured chain, add a chain store
        for (chain_name, shard, _cache_size) in chains {
            if let Some(chain) = existing_chains
                .iter()
                .find(|chain| chain.name == chain_name)
            {
                if chain.shard != shard {
                    warn!(
                        &block_store.logger,
                        "The chain `{0}` is stored in shard `{1}` but is configured for shard `{2}`; ignoring config and using shard `{1}`",
                        chain.name,
                        chain.shard,
                        shard,
                    );
                }
                block_store.add_chain_store(chain, false).await?;
            };
        }

        // There might be chains we have in the database that are not yet/
        // no longer configured. Add a chain store for each of them, too
        let configured_chains = block_store
            .stores
            .read()
            .keys()
            .cloned()
            .collect::<Vec<_>>();
        for chain in existing_chains
            .iter()
            .filter(|chain| !configured_chains.contains(&chain.name))
        {
            block_store.add_chain_store(chain, false).await?;
        }
        Ok(block_store)
    }

    pub(crate) async fn query_permit_primary(&self) -> QueryPermit {
        self.mirror.primary().query_permit().await
    }

    pub async fn allocate_chain(
        conn: &mut AsyncPgConnection,
        name: &str,
        shard: &Shard,
        ident: &ChainIdentifier,
    ) -> Result<Chain, StoreError> {
        #[derive(QueryableByName, Debug)]
        struct ChainIdSeq {
            #[diesel(sql_type = diesel::sql_types::BigInt)]
            last_value: i64,
        }

        // Fetch the current last_value from the sequence
        let result = sql_query("SELECT last_value FROM chains_id_seq")
            .get_result::<ChainIdSeq>(conn)
            .await?;

        let last_val = result.last_value;

        let next_val = last_val + 1;
        let namespace = format!("chain{}", next_val);
        let storage =
            Storage::new(namespace.to_string()).map_err(|e| StoreError::Unknown(anyhow!(e)))?;

        let chain = Chain {
            id: next_val as i32,
            name: name.to_string(),
            shard: shard.clone(),
            net_version: ident.net_version.clone(),
            genesis_block: ident.genesis_block_hash.hash_hex(),
            storage: storage.clone(),
        };

        Ok(chain)
    }

    pub async fn add_chain_store(
        &self,
        chain: &primary::Chain,
        create: bool,
    ) -> Result<Arc<ChainStore>, StoreError> {
        let pool = self
            .pools
            .get(&chain.shard)
            .ok_or_else(|| internal_error!("there is no pool for shard {}", chain.shard))?
            .clone();
        let sender = ChainHeadUpdateSender::new(
            self.mirror.primary().clone(),
            chain.name.clone(),
            self.sender.clone(),
        );
        let ident = chain.network_identifier()?;
        let logger = self.logger.new(o!("network" => chain.name.clone()));
        let cache_size = self
            .shards
            .iter()
            .find_map(|(network, _, chain_size)| {
                if network == &chain.name {
                    Some(*chain_size)
                } else {
                    None
                }
            })
            .unwrap_or(BLOCK_CACHE_SIZE);
        let store = ChainStore::new(
            logger,
            chain.name.clone(),
            chain.storage.clone(),
            sender,
            pool,
            ENV_VARS.store.recent_blocks_cache_capacity,
            self.chain_store_metrics.clone(),
            cache_size,
        );
        if create {
            store.create(&ident).await?;
        }
        let store = Arc::new(store);
        self.stores
            .write()
            .insert(chain.name.clone(), store.clone());
        Ok(store)
    }

    /// Return a map from network name to the network's chain head pointer.
    /// The information is cached briefly since this method is used heavily
    /// by the indexing status API
    pub async fn chain_head_pointers(&self) -> Result<HashMap<String, BlockPtr>, StoreError> {
        let mut map = HashMap::new();
        for (shard, pool) in &self.pools {
            let cached = match self.chain_head_cache.get(shard.as_str()) {
                Some(cached) => cached,
                None => {
                    let mut conn = match pool.get_permitted().await {
                        Ok(conn) => conn,
                        Err(StoreError::DatabaseUnavailable) => continue,
                        Err(e) => return Err(e),
                    };
                    let heads = Arc::new(ChainStore::chain_head_pointers(&mut conn).await?);
                    self.chain_head_cache.set(shard.to_string(), heads.clone());
                    heads
                }
            };
            map.extend(
                cached
                    .iter()
                    .map(|(chain, ptr)| (chain.clone(), ptr.clone())),
            );
        }
        Ok(map)
    }

    pub async fn chain_head_block(&self, chain: &str) -> Result<Option<BlockNumber>, StoreError> {
        let store = self
            .store(chain)
            .await
            .ok_or_else(|| internal_error!("unknown network `{}`", chain))?;
        store.chain_head_block(chain).await
    }

    async fn lookup_chain(&self, chain: &str) -> Result<Option<Arc<ChainStore>>, StoreError> {
        // See if we have that chain in the database even if it wasn't one
        // of the configured chains
        let chain = chain.to_string();
        let this = self.cheap_clone();
        self.mirror
            .read_async(|conn| {
                async {
                    match primary::find_chain(conn, &chain).await? {
                        Some(chain) => {
                            let chain_store = this.add_chain_store(&chain, false).await?;
                            Ok(Some(chain_store))
                        }
                        None => Ok(None),
                    }
                }
                .scope_boxed()
            })
            .await
    }

    async fn store(&self, chain: &str) -> Option<Arc<ChainStore>> {
        let store = self.stores.read().get(chain).map(CheapClone::cheap_clone);
        if store.is_some() {
            return store;
        }
        // The chain is not in the cache yet, load it from the database; we
        // suppress errors here since it will be very rare that we look up
        // a chain from the database as most of them will be set up when
        // the block store is created
        self.lookup_chain(chain).await.unwrap_or_else(|e| {
                error!(&self.logger, "Error getting chain from store"; "network" => chain, "error" => e.to_string());
                None
            })
    }

    pub async fn drop_chain(&self, chain: &str) -> Result<(), StoreError> {
        let chain_store = self
            .store(chain)
            .await
            .ok_or_else(|| internal_error!("unknown chain {}", chain))?;

        // Delete from the primary first since that's where
        // deployment_schemas has a fk constraint on chains
        primary::drop_chain(self.mirror.primary(), chain).await?;

        chain_store.drop_chain().await?;

        self.stores.write().remove(chain);

        Ok(())
    }

    // Helper to clone the list of chain stores to avoid holding the lock
    // while awaiting
    fn stores(&self) -> Vec<Arc<ChainStore>> {
        self.stores
            .read()
            .values()
            .map(CheapClone::cheap_clone)
            .collect()
    }

    // cleanup_ethereum_shallow_blocks will delete cached blocks previously produced by firehose on
    // an ethereum chain that is not currently configured to use firehose provider.
    //
    // This is to prevent an issue where firehose stores "shallow" blocks (with null data) in `chainX.blocks`
    // table but RPC provider requires those blocks to be full.
    //
    // - This issue only affects ethereum chains.
    // - This issue only happens when switching providers from firehose back to RPC. it is gated by
    // the presence of a cursor in the public.ethereum_networks table for a chain configured without firehose.
    // - Only the shallow blocks close to HEAD need to be deleted, the older blocks don't need data.
    // - Deleting everything or creating an index on empty data would cause too much performance
    // hit on graph-node startup.
    //
    // Discussed here: https://github.com/graphprotocol/graph-node/pull/4790
    pub async fn cleanup_ethereum_shallow_blocks(
        &self,
        eth_rpc_only_nets: Vec<String>,
    ) -> Result<(), StoreError> {
        for store in self.stores() {
            if !eth_rpc_only_nets.contains(&store.chain) {
                continue;
            };

            if let Some(head_block) = store.remove_cursor(&store.chain).await? {
                let lower_bound = head_block.saturating_sub(ENV_VARS.reorg_threshold() * 2);
                info!(&self.logger, "Removed cursor for non-firehose chain, now cleaning shallow blocks"; "network" => &store.chain, "lower_bound" => lower_bound);
                store.cleanup_shallow_blocks(lower_bound).await?;
            }
        }
        Ok(())
    }

    async fn truncate_block_caches(&self) -> Result<(), StoreError> {
        for store in self.stores() {
            store.truncate_block_cache().await?;
        }
        Ok(())
    }

    pub async fn update_db_version(&self) -> Result<(), StoreError> {
        use crate::primary::db_version as dbv;

        let primary_pool = self.pools.get(&*PRIMARY_SHARD).unwrap();
        let mut conn = primary_pool.get_permitted().await?;
        let version: i64 = dbv::table
            .select(dbv::version)
            .get_result(&mut conn)
            .await?;
        if version < 3 {
            self.truncate_block_caches().await?;
            diesel::update(dbv::table)
                .set(dbv::version.eq(3))
                .execute(&mut conn)
                .await?;
        };
        if version < SUPPORTED_DB_VERSION {
            // Bump it to make sure that all executables are working with the same DB format
            diesel::update(dbv::table)
                .set(dbv::version.eq(SUPPORTED_DB_VERSION))
                .execute(&mut conn)
                .await?;
        };
        if version > SUPPORTED_DB_VERSION {
            panic!(
                "The executable is too old and doesn't support the database version: {}",
                version
            )
        }
        Ok(())
    }

    pub async fn create_chain_store(
        &self,
        network: &str,
        ident: ChainIdentifier,
    ) -> anyhow::Result<Arc<ChainStore>> {
        if let Some(chain_store) = self.store(network).await {
            return Ok(chain_store);
        }

        let mut conn = self.mirror.primary().get().await?;
        let shard = self
            .shards
            .iter()
            .find_map(|(chain_id, shard, _cache_size)| {
                if chain_id.as_str().eq(network) {
                    Some(shard)
                } else {
                    None
                }
            })
            .ok_or_else(|| anyhow!("unable to find shard for network {}", network))?;
        let chain = primary::add_chain(&mut conn, network, shard, ident).await?;
        self.add_chain_store(&chain, true)
            .await
            .map_err(anyhow::Error::from)
    }
}

#[async_trait]
impl BlockStoreTrait for BlockStore {
    type ChainStore = ChainStore;

    async fn chain_store(&self, network: &str) -> Option<Arc<Self::ChainStore>> {
        self.store(network).await
    }
}

#[async_trait]
impl ChainIdStore for BlockStore {
    async fn chain_identifier(
        &self,
        chain_name: &ChainName,
    ) -> Result<ChainIdentifier, anyhow::Error> {
        let chain_store = self
            .chain_store(chain_name)
            .await
            .ok_or_else(|| anyhow!("unable to get store for chain '{chain_name}'"))?;

        chain_store.chain_identifier().await
    }

    async fn set_chain_identifier(
        &self,
        chain_name: &ChainName,
        ident: &ChainIdentifier,
    ) -> Result<(), anyhow::Error> {
        use primary::chains as c;

        // Update the block shard first since that contains a copy from the primary
        let chain_store = self
            .chain_store(chain_name)
            .await
            .ok_or_else(|| anyhow!("unable to get store for chain '{chain_name}'"))?;

        chain_store.set_chain_identifier(ident).await?;

        // Update the master copy in the primary
        let primary_pool = self.pools.get(&*PRIMARY_SHARD).unwrap();
        let mut conn = primary_pool.get_permitted().await?;

        diesel::update(c::table.filter(c::name.eq(chain_name.as_str())))
            .set((
                c::genesis_block_hash.eq(ident.genesis_block_hash.hash_hex()),
                c::net_version.eq(&ident.net_version),
            ))
            .execute(&mut conn)
            .await?;

        Ok(())
    }
}