store: Add rebuild_storage to ChainStore and BlockStore

Adds ChainStore::rebuild_storage, which drops the chain's schema (if
present), upserts the ethereum_networks row to reset head tracking
columns, and recreates the schema with empty tables — all in a single
transaction.

Adds BlockStore::has_namespace and BlockStore::rebuild_chain_storage as
the public entry points used by the graphman command.

Author: dimitrovmaksim

store/postgres/src/block_store.rs

@@ -484,6 +484,29 @@ impl BlockStore {
         Ok(())
     }
 
+    pub async fn has_namespace(&self, chain: &primary::Chain) -> Result<bool, StoreError> {
+        let pool = self
+            .pools
+            .get(&chain.shard)
+            .ok_or_else(|| internal_error!("no pool for shard {}", chain.shard))?;
+        let nsp = crate::primary::Namespace::special(chain.storage.to_string());
+        let mut conn = pool.get_permitted().await?;
+        crate::catalog::has_namespace(&mut conn, &nsp).await
+    }
+
+    pub async fn rebuild_chain_storage(
+        &self,
+        chain: &str,
+        ident: &ChainIdentifier,
+    ) -> Result<(), StoreError> {
+        let chain_store = self
+            .store(chain)
+            .await
+            .ok_or_else(|| internal_error!("No chain store found for {}", chain))?;
+
+        Ok(chain_store.rebuild_storage(ident).await?)
+    }
+
     // Helper to clone the list of chain stores to avoid holding the lock
     // while awaiting
     fn stores(&self) -> Vec<Arc<ChainStore>> {

store/postgres/src/chain_store.rs

@@ -3,7 +3,7 @@ use async_trait::async_trait;
 use diesel::sql_types::Text;
 use diesel::{insert_into, update, ExpressionMethods, OptionalExtension, QueryDsl};
 use diesel_async::AsyncConnection;
-use diesel_async::{scoped_futures::ScopedFutureExt, RunQueryDsl};
+use diesel_async::{scoped_futures::ScopedFutureExt, RunQueryDsl, SimpleAsyncConnection};
 
 use graph::components::store::ChainHeadStore;
 use graph::data::store::ethereum::call;
@@ -12,7 +12,7 @@ use graph::parking_lot::RwLock;
 use graph::prelude::alloy::primitives::B256;
 use graph::prelude::MetricsRegistry;
 use graph::prometheus::{CounterVec, GaugeVec};
-use graph::slog::{info, o, Logger};
+use graph::slog::{debug, info, o, Logger};
 use graph::stable_hash::crypto_stable_hash;
 use graph::util::herd_cache::HerdCache;
 
@@ -2479,6 +2479,55 @@ impl ChainStore {
         .await
     }
 
+    /// Drop the chain's storage schema (if it exists), reset head
+    /// metadata in `ethereum_networks`, and rebuild the schema with
+    /// empty tables. If the `ethereum_networks` row is missing, it is
+    /// created from the provided `ident`.
+    pub(crate) async fn rebuild_storage(&self, ident: &ChainIdentifier) -> Result<(), Error> {
+        use public::ethereum_networks as n;
+
+        let nsp = self.storage.to_string();
+
+        debug!(&self.logger, "Rebuilding storage for chain"; "chain" => &self.chain, "namespace" => &nsp);
+
+        let mut conn = self.pool.get_permitted().await?;
+        conn.transaction(|conn| {
+            async {
+                debug!(&self.logger, "Dropping existing schema if present"; "namespace" => &nsp);
+                conn.batch_execute(&format!("DROP SCHEMA IF EXISTS {nsp} CASCADE"))
+                    .await?;
+
+                debug!(&self.logger, "Upserting ethereum_networks row"; "chain" => &self.chain);
+                insert_into(n::table)
+                    .values((
+                        n::name.eq(&self.chain),
+                        n::namespace.eq(&self.storage),
+                        n::net_version.eq(&ident.net_version),
+                        n::genesis_block_hash.eq(ident.genesis_block_hash.hash_hex()),
+                    ))
+                    .on_conflict(n::name)
+                    .do_update()
+                    .set((
+                        n::namespace.eq(&self.storage),
+                        n::net_version.eq(&ident.net_version),
+                        n::genesis_block_hash.eq(ident.genesis_block_hash.hash_hex()),
+                        n::head_block_hash.eq(None::<String>),
+                        n::head_block_number.eq(None::<i64>),
+                        n::head_block_cursor.eq(None::<String>),
+                    ))
+                    .execute(conn)
+                    .await?;
+
+                debug!(&self.logger, "Creating storage schema and tables"; "namespace" => &nsp);
+                self.storage.create(conn).await?;
+
+                Ok(())
+            }
+            .scope_boxed()
+        })
+        .await
+    }
+
     pub async fn chain_head_pointers(
         conn: &mut AsyncPgConnection,
     ) -> Result<HashMap<String, BlockPtr>, StoreError> {