Introducing FuelDataIngester (?)

[Jurshsmith]

References: #1

First, it would be great if we came up with a feature-inclined name for this new System -- whose responsibility is subscribing to important events in a Fuel's Node and publishing to consumer systems/apps. Since NATS can be used for other use cases, instead of NatsPublisher/FuelCoreNats, would it be better if we used FuelDataIngester/ FuelEventsIngester/ FuelDataPublisher/ (suggestions)? To avoid being blocked, FuelDataIngester will be used in the demonstrations below, which, of course, is subject to change once we decide.

The simplest way to decouple the FuelDataIngester process from fuel-core is by using an environment variable such that when present, the process starts, otherwise, it simply skips with a warning log. Please see the code below for what it will look like approximately:

In our data-systems repo, we could expose this from a fuel-data-ingester crate:

struct FuelDataIngester {
    nats_url: Option<String>,
    chain_id: ChainId, // As discussed, will be used for all subject prefixes 
    data_subscription: tokio::sync::broadcast::Receiver<SharedImportResult>,
}

impl FuelDataIngester {
    pub fn new(
        nats_url: Option<String>,
        chain_id: ChainId,
        data_subscription: tokio::sync::broadcast::Receiver<SharedImportResult>,
    ) -> Self {
        Self {
            nats_url,
            chain_id,
            data_subscription,
        }
    }

    pub fn start(&self) { // This should return a `Result<(), FuelDataIngesterError>` eventually for other runtime validations
        if let Some(nats_url) = &self.nats_url {
            info!("Starting FuelDataIngester...");

            tokio::spawn(async {
             // Starting logic resides here...
            });
        } else {
            warn!("NATS_URL not present: FuelDataIngester did not start.")
        }
    }
}

Then, fuel-core's exec function will start the ingester system this way:

pub async fn exec(command: Command) -> anyhow::Result<()> {
    let service = get_service(command)?;

    let block_importer = service.shared.block_importer.block_importer.clone();
    let subscription = block_importer.subscribe();
    FuelDataIngester::new(
        env::var("FUEL_DATA_INGESTER_NATS_URL").ok(),
        block_importer.chain_id,
        subscription,
    )
    .start();
 
 
 // ...The rest of the code
}

Please feel free to share any feedback to keep this discussion going 🚀

[Jurshsmith]

Using the CLI approach and following our current nomenclature, fuel-core's exec will look like this:

pub async fn exec(command: Command) -> anyhow::Result<()> {
    let service = get_service(command)?;

    if let Command {
        nats_url: Some(nats_url),
        ..
    } = command
    {
        fuel_nats_stream::start_publisher(nats_url, subscription);
    }
    
    // ...rest of the code
}

[pedronauck]

I'm closing this discussion in favor of new updates and patterns already implemented