mod.rs

mod instance;

use crate::polling_monitor::{
    spawn_monitor, ArweaveService, IpfsRequest, IpfsService, PollingMonitor, PollingMonitorMetrics,
};
use anyhow::{self, Error};
use bytes::Bytes;
use graph::{
    blockchain::Blockchain,
    components::{store::DeploymentId, subgraph::HostMetrics},
    data::subgraph::SubgraphManifest,
    data_source::{
        causality_region::CausalityRegionSeq,
        offchain::{self, Base64},
        CausalityRegion, DataSource, DataSourceTemplate,
    },
    derive::CheapClone,
    ipfs::IpfsContext,
    prelude::{
        BlockNumber, CancelGuard, CheapClone, DeploymentHash, MetricsRegistry, RuntimeHostBuilder,
        SubgraphCountMetric, TriggerProcessor,
    },
    slog::Logger,
};
use std::collections::HashMap;
use std::sync::Arc;

use graph::parking_lot::RwLock;
use tokio::sync::mpsc;

use self::instance::SubgraphInstance;
use super::Decoder;

#[derive(Clone, CheapClone, Debug)]
pub struct SubgraphKeepAlive {
    alive_map: Arc<RwLock<HashMap<DeploymentId, CancelGuard>>>,
    sg_metrics: Arc<SubgraphCountMetric>,
}

impl SubgraphKeepAlive {
    pub fn new(sg_metrics: Arc<SubgraphCountMetric>) -> Self {
        Self {
            sg_metrics,
            alive_map: Arc::new(RwLock::new(HashMap::default())),
        }
    }

    pub fn remove(&self, deployment_id: &DeploymentId) {
        self.alive_map.write().remove(deployment_id);
        self.sg_metrics.running_count.dec();
    }
    pub fn insert(&self, deployment_id: DeploymentId, guard: CancelGuard) {
        let old = self.alive_map.write().insert(deployment_id, guard);
        if old.is_none() {
            self.sg_metrics.running_count.inc();
        }
    }

    pub fn contains(&self, deployment_id: &DeploymentId) -> bool {
        self.alive_map.read().contains_key(deployment_id)
    }
}

// The context keeps track of mutable in-memory state that is retained across blocks.
//
// Currently most of the changes are applied in `runner.rs`, but ideally more of that would be
// refactored into the context so it wouldn't need `pub` fields. The entity cache should probably
// also be moved here.
pub struct IndexingContext<C, T>
where
    T: RuntimeHostBuilder<C>,
    C: Blockchain,
{
    pub(crate) instance: SubgraphInstance<C, T>,
    pub instances: SubgraphKeepAlive,
    pub offchain_monitor: OffchainMonitor,
    pub(crate) trigger_processor: Box<dyn TriggerProcessor<C, T>>,
    pub(crate) decoder: Box<Decoder<C, T>>,
}

impl<C: Blockchain, T: RuntimeHostBuilder<C>> IndexingContext<C, T> {
    pub fn new(
        manifest: SubgraphManifest<C>,
        host_builder: T,
        host_metrics: Arc<HostMetrics>,
        causality_region_seq: CausalityRegionSeq,
        instances: SubgraphKeepAlive,
        offchain_monitor: OffchainMonitor,
        trigger_processor: Box<dyn TriggerProcessor<C, T>>,
        decoder: Box<Decoder<C, T>>,
    ) -> Self {
        let instance = SubgraphInstance::new(
            manifest,
            host_builder,
            host_metrics.clone(),
            causality_region_seq,
        );

        Self {
            instance,
            instances,
            offchain_monitor,
            trigger_processor,
            decoder,
        }
    }

    /// Removes data sources hosts with a creation block greater or equal to `reverted_block`, so
    /// that they are no longer candidates for `process_trigger`.
    ///
    /// This does not currently affect the `offchain_monitor` or the `filter`, so they will continue
    /// to include data sources that have been reverted. This is not ideal for performance, but it
    /// does not affect correctness since triggers that have no matching host will be ignored by
    /// `process_trigger`.
    ///
    /// File data sources that have been marked not done during this process will get re-queued
    pub fn revert_data_sources(&mut self, reverted_block: BlockNumber) {
        let removed = self.instance.revert_data_sources(reverted_block);

        removed
            .into_iter()
            .for_each(|source| self.offchain_monitor.add_source(source))
    }

    pub fn add_dynamic_data_source(
        &mut self,
        logger: &Logger,
        data_source: DataSource<C>,
    ) -> Result<Option<Arc<T::Host>>, Error> {
        let offchain_fields = data_source
            .as_offchain()
            .map(|ds| (ds.source.clone(), ds.is_processed()));
        let host = self.instance.add_dynamic_data_source(logger, data_source)?;

        if host.is_some() {
            if let Some((source, is_processed)) = offchain_fields {
                // monitor data source only if it has not yet been processed.
                if !is_processed {
                    self.offchain_monitor.add_source(source);
                }
            }
        }

        Ok(host)
    }

    pub fn causality_region_next_value(&mut self) -> CausalityRegion {
        self.instance.causality_region_next_value()
    }

    pub fn hosts_len(&self) -> usize {
        self.instance.hosts_len()
    }

    pub fn onchain_data_sources(&self) -> impl Iterator<Item = &C::DataSource> + Clone {
        self.instance.onchain_data_sources()
    }

    pub fn static_data_sources(&self) -> &[DataSource<C>] {
        &self.instance.static_data_sources
    }

    pub fn templates(&self) -> &[DataSourceTemplate<C>] {
        &self.instance.templates
    }
}

pub struct OffchainMonitor {
    ipfs_monitor: PollingMonitor<IpfsRequest>,
    ipfs_monitor_rx: mpsc::UnboundedReceiver<(IpfsRequest, Bytes)>,
    arweave_monitor: PollingMonitor<Base64>,
    arweave_monitor_rx: mpsc::UnboundedReceiver<(Base64, Bytes)>,
    deployment_hash: DeploymentHash,
    logger: Logger,
}

impl OffchainMonitor {
    pub fn new(
        logger: Logger,
        registry: Arc<MetricsRegistry>,
        subgraph_hash: &DeploymentHash,
        ipfs_service: IpfsService,
        arweave_service: ArweaveService,
    ) -> Self {
        let metrics = Arc::new(PollingMonitorMetrics::new(registry, subgraph_hash));
        // The channel is unbounded, as it is expected that `fn ready_offchain_events` is called
        // frequently, or at least with the same frequency that requests are sent.
        let (ipfs_monitor_tx, ipfs_monitor_rx) = mpsc::unbounded_channel();
        let (arweave_monitor_tx, arweave_monitor_rx) = mpsc::unbounded_channel();

        let ipfs_monitor = spawn_monitor(
            ipfs_service,
            ipfs_monitor_tx,
            logger.cheap_clone(),
            metrics.cheap_clone(),
        );

        let arweave_monitor = spawn_monitor(
            arweave_service,
            arweave_monitor_tx,
            logger.cheap_clone(),
            metrics,
        );

        Self {
            ipfs_monitor,
            ipfs_monitor_rx,
            arweave_monitor,
            arweave_monitor_rx,
            deployment_hash: subgraph_hash.to_owned(),
            logger,
        }
    }

    fn add_source(&mut self, source: offchain::Source) {
        match source {
            offchain::Source::Ipfs(path) => self.ipfs_monitor.monitor(IpfsRequest {
                ctx: IpfsContext::new(&self.deployment_hash, &self.logger),
                path,
            }),
            offchain::Source::Arweave(base64) => self.arweave_monitor.monitor(base64),
        };
    }

    pub fn ready_offchain_events(&mut self) -> Result<Vec<offchain::TriggerData>, Error> {
        use tokio::sync::mpsc::error::TryRecvError;

        let mut triggers = vec![];
        loop {
            match self.ipfs_monitor_rx.try_recv() {
                Ok((req, data)) => triggers.push(offchain::TriggerData {
                    source: offchain::Source::Ipfs(req.path),
                    data: Arc::new(data),
                }),
                Err(TryRecvError::Disconnected) => {
                    anyhow::bail!("ipfs monitor unexpectedly terminated")
                }
                Err(TryRecvError::Empty) => break,
            }
        }

        loop {
            match self.arweave_monitor_rx.try_recv() {
                Ok((base64, data)) => triggers.push(offchain::TriggerData {
                    source: offchain::Source::Arweave(base64),
                    data: Arc::new(data),
                }),
                Err(TryRecvError::Disconnected) => {
                    anyhow::bail!("arweave monitor unexpectedly terminated")
                }
                Err(TryRecvError::Empty) => break,
            }
        }

        Ok(triggers)
    }
}