feat: Add from-genesis subcommand

README.md

@@ -86,11 +86,9 @@ Command to run NEAR Indexer for Explorer have to include the chain-id and start
 You can choose NEAR Indexer for Explorer start options:
  - `from-latest` - start indexing blocks from the latest finalized block
  - `from-interruption` - start indexing blocks from the block NEAR Indexer was interrupted last time but earlier for `<number_of_blocks>` if provided
+ - `from-genesis` - download and store accounts/access keys in genesis file and start indexing from the genesis block
  - `from-block --height <block_height>` - start indexing blocks from the specific block height
 
-#### Storing genesis file
-Unlike the original NEAR Indexer for Explorer you **can't** tell Indexer to store data from genesis (Accounts and Access Keys) by adding key `--store-genesis` to the `run` command. So please, ensure you took care about the genesis data in your database in order this indexer to work properly. This capability will be implemented eventually, it's progress can be tracked here: #327.
-
 #### Strict mode
 NEAR Indexer for Explorer works in strict mode by default. In strict mode, the Indexer will ensure parent data exists before storing children, infinitely retrying until this condition is met. This is necessary as a parent (i.e. `block`) may still be processing while a child (i.e. `receipt`) is ready to be stored. This scenario will likely occur if you have not stored the genesis file or do not have all data prior to the block you start indexing from. In this case, you can disable strict mode to store data prior to the block you are concerned about, and then re-enable it once you have passed this block.
 

database/Cargo.toml

@@ -25,6 +25,7 @@ actix-diesel = { git = "https://github.com/frol/actix-diesel", rev = "3a001986c8
 near-indexer-primitives = "0.16.0"
 near-primitives = "0.16.0"
 near-crypto = "0.16.0"
+near-chain-configs = "0.16.0"
 
 [features]
 default = []

database/src/adapters/access_keys.rs

@@ -4,6 +4,7 @@ use actix_diesel::dsl::AsyncRunQueryDsl;
 
 use diesel::{ExpressionMethods, PgConnection, QueryDsl};
 use futures::try_join;
+use tracing::info;
 
 use crate::models;
 use crate::schema;
@@ -145,3 +146,26 @@ pub async fn handle_access_keys(
 
     Ok(())
 }
+
+pub(crate) async fn store_access_keys_from_genesis(
+    pool: &actix_diesel::Database<PgConnection>,
+    access_keys_models: Vec<models::access_keys::AccessKey>,
+) -> anyhow::Result<()> {
+    info!(
+        target: crate::EXPLORER_DATABASE,
+        "Adding/updating {} access keys from genesis...",
+        access_keys_models.len(),
+    );
+
+    crate::await_retry_or_panic!(
+        diesel::insert_into(schema::access_keys::table)
+            .values(access_keys_models.clone())
+            .on_conflict_do_nothing()
+            .execute_async(pool),
+        10,
+        "Failed to store AccessKeys from genesis".to_string(),
+        &access_keys_models
+    );
+
+    Ok(())
+}

database/src/adapters/accounts.rs

@@ -6,6 +6,7 @@ use bigdecimal::BigDecimal;
 
 use diesel::{BoolExpressionMethods, ExpressionMethods, PgConnection, QueryDsl};
 use futures::try_join;
+use tracing::info;
 
 use crate::models;
 use crate::schema;
@@ -249,3 +250,26 @@ pub async fn get_lockup_account_ids_at_block_height(
                 .collect()
         })
 }
+
+pub(crate) async fn store_accounts_from_genesis(
+    pool: &actix_diesel::Database<PgConnection>,
+    accounts_models: Vec<models::accounts::Account>,
+) -> anyhow::Result<()> {
+    info!(
+        target: crate::EXPLORER_DATABASE,
+        "Adding/updating {} accounts from genesis...",
+        accounts_models.len(),
+    );
+
+    crate::await_retry_or_panic!(
+        diesel::insert_into(schema::accounts::table)
+            .values(accounts_models.clone())
+            .on_conflict_do_nothing()
+            .execute_async(pool),
+        10,
+        "Failed to store Accounts from genesis".to_string(),
+        &accounts_models
+    );
+
+    Ok(())
+}

database/src/adapters/genesis.rs

@@ -0,0 +1,59 @@
+use actix_diesel::Database;
+use diesel::PgConnection;
+use near_chain_configs::{Genesis, GenesisValidationMode};
+
+use crate::adapters::access_keys::store_access_keys_from_genesis;
+use crate::adapters::accounts::store_accounts_from_genesis;
+
+pub async fn store_genesis_records(
+    pool: &Database<PgConnection>,
+    genesis_file_path: String,
+) -> anyhow::Result<()> {
+    let mut accounts_to_store: Vec<crate::models::accounts::Account> = vec![];
+    let mut access_keys_to_store: Vec<crate::models::access_keys::AccessKey> = vec![];
+
+    let genesis = Genesis::from_file(genesis_file_path, GenesisValidationMode::Full);
+
+    let genesis_height = genesis.config.genesis_height;
+
+    for record in genesis.records.0 {
+        if accounts_to_store.len() == 5_000 {
+            let mut accounts_to_store_chunk = vec![];
+            std::mem::swap(&mut accounts_to_store, &mut accounts_to_store_chunk);
+            store_accounts_from_genesis(pool, accounts_to_store_chunk).await?;
+        }
+
+        if access_keys_to_store.len() == 5_000 {
+            let mut access_keys_to_store_chunk = vec![];
+            std::mem::swap(&mut access_keys_to_store, &mut access_keys_to_store_chunk);
+            store_access_keys_from_genesis(pool, access_keys_to_store_chunk).await?;
+        }
+
+        match record {
+            near_primitives::state_record::StateRecord::Account { account_id, .. } => {
+                accounts_to_store.push(crate::models::accounts::Account::new_from_genesis(
+                    &account_id,
+                    genesis_height,
+                ));
+            }
+            near_primitives::state_record::StateRecord::AccessKey {
+                account_id,
+                public_key,
+                access_key,
+            } => {
+                access_keys_to_store.push(crate::models::access_keys::AccessKey::from_genesis(
+                    &public_key,
+                    &account_id,
+                    &access_key,
+                    genesis_height,
+                ));
+            }
+            _ => {}
+        };
+    }
+
+    store_accounts_from_genesis(pool, accounts_to_store).await?;
+    store_access_keys_from_genesis(pool, access_keys_to_store).await?;
+
+    Ok(())
+}

database/src/adapters/mod.rs

@@ -6,6 +6,7 @@ pub mod assets;
 pub mod blocks;
 pub mod chunks;
 pub mod execution_outcomes;
+pub mod genesis;
 pub mod receipts;
 pub mod transactions;
 

indexer/Cargo.toml

@@ -24,6 +24,7 @@ lazy_static = "^1.4"
 near-sdk = { git = "https://github.com/near/near-sdk-rs", rev="03487c184d37b0382dd9bd41c57466acad58fc1f" }
 openssl-probe = { version = "0.1.2" }
 prometheus = "0.13.0"
+reqwest = "0.11.14"
 r2d2 = "0.8.8"
 tokio = { version = "1.1", features = ["sync", "time"] }
 tokio-stream = { version = "0.1" }

indexer/src/configs.rs

@@ -50,14 +50,16 @@ pub enum ChainId {
 }
 
 #[allow(clippy::enum_variant_names)]
-#[derive(Subcommand, Debug, Clone)]
+#[derive(Subcommand, Debug, Clone, PartialEq, Eq)]
 pub enum StartOptions {
     /// Start from specific block height
     FromBlock { height: u64 },
     /// Start from interruption (last_indexed_block value from Redis)
     FromInterruption,
     /// Start from the final block on the network (queries JSON RPC for finality: final)
     FromLatest,
+    /// Store genesis data and start from first block
+    FromGenesis,
 }
 
 impl Opts {
@@ -77,6 +79,13 @@ impl Opts {
             ChainId::Betanet(_) => "https://rpc.betanet.near.org",
         }
     }
+
+    pub fn genesis_file_url(&self) -> &str {
+        match self.chain_id {
+            ChainId::Mainnet(_) => "https://s3-us-west-1.amazonaws.com/build.nearprotocol.com/nearcore-deploy/mainnet/genesis.json",
+            ChainId::Testnet(_) => "https://s3-us-west-1.amazonaws.com/build.nearprotocol.com/nearcore-deploy/testnet/genesis.json",
+        }
+    }
 }
 
 impl Opts {
@@ -114,6 +123,9 @@ async fn get_start_block_height(opts: &Opts) -> u64 {
             }
         }
         StartOptions::FromLatest => final_block_height(opts).await,
+        // Since NEAR Lake stores blocks in ascending order, using 0 here forces
+        // near-lake-framework to start from the first block, i.e. genesis
+        StartOptions::FromGenesis => 0,
     }
 }
 

indexer/src/main.rs

@@ -1,3 +1,5 @@
+use std::io::Write;
+
 use clap::Parser;
 
 pub use cached::SizedCache;
@@ -8,7 +10,7 @@ use tracing::{debug, info};
 
 use explorer_database::{adapters, models, receipts_cache};
 
-use crate::configs::Opts;
+use crate::configs::{ChainId, Opts, StartOptions};
 
 mod configs;
 mod metrics;
@@ -139,6 +141,27 @@ async fn handle_message(
     Ok(())
 }
 
+async fn download_genesis_file(opts: &configs::Opts) -> anyhow::Result<String> {
+    let chain = match opts.chain_id {
+        ChainId::Mainnet(_) => "mainnet",
+        ChainId::Testnet(_) => "testnet",
+    };
+
+    info!(
+        target: INDEXER_FOR_EXPLORER,
+        "Downloading {} genesis file", chain
+    );
+
+    let resp = reqwest::get(opts.genesis_file_url()).await?;
+
+    let file_path = format!("{}-genesis.json", chain);
+
+    let mut file = std::fs::File::create(&file_path)?;
+    file.write_all(&resp.bytes().await?)?;
+
+    Ok(file_path.to_string())
+}
+
 #[actix::main]
 async fn main() -> anyhow::Result<()> {
     // We use it to automatically search the for root certificates to perform HTTPS calls
@@ -165,16 +188,21 @@ async fn main() -> anyhow::Result<()> {
     let receipts_cache_arc: receipts_cache::ReceiptsCacheArc =
         std::sync::Arc::new(Mutex::new(SizedCache::with_size(100_000)));
 
-    let config: near_lake_framework::LakeConfig = opts.to_lake_config().await;
-    let (sender, stream) = near_lake_framework::streamer(config);
-
-    tokio::spawn(metrics::init_server(opts.port).expect("Failed to start metrics server"));
-
     tracing::info!(
         target: INDEXER_FOR_EXPLORER,
         "Starting Indexer for Explorer (lake)...",
     );
 
+    tokio::spawn(metrics::init_server(opts.port).expect("Failed to start metrics server"));
+
+    if opts.start_options() == &StartOptions::FromGenesis {
+        let genesis_file_path = download_genesis_file(&opts).await?;
+        adapters::genesis::store_genesis_records(&pool, genesis_file_path).await?;
+    }
+
+    let config: near_lake_framework::LakeConfig = opts.to_lake_config().await;
+    let (sender, stream) = near_lake_framework::streamer(config);
+
     let mut handlers = tokio_stream::wrappers::ReceiverStream::new(stream)
         .map(|streamer_message| {
             info!(