node: Add graphman restore command

Add `graphman restore` CLI with options:
  --directory  Path to dump directory
  --shard      Target shard (default: primary)
  --name       Subgraph name for deployment rule matching
  --replace    Drop and recreate if exists in target shard
  --add        Create copy in a different shard
  --force      Restore regardless of current state

Author: lutter

node/src/bin/manager.rs

@@ -342,6 +342,33 @@ pub enum Command {
         /// The name of the directory to dump to
         directory: String,
     },
+
+    /// Restore a subgraph deployment from a dump directory
+    ///
+    /// EXPERIMENTAL - NOT FOR PRODUCTION USE
+    ///
+    /// Restore a subgraph deployment from a dump created with the `dump`
+    /// command.
+    Restore {
+        /// Path to the dump directory
+        directory: String,
+        /// The database shard to restore into (default: primary)
+        #[clap(long)]
+        shard: Option<String>,
+        /// Subgraph name for deployment rule matching and node assignment.
+        /// If omitted, uses an existing name from the database; errors if none found.
+        #[clap(long)]
+        name: Option<String>,
+        /// Drop and recreate if the deployment already exists in the target shard
+        #[clap(long, conflicts_with_all = ["add", "force"])]
+        replace: bool,
+        /// Create a copy in a shard that doesn't have this deployment (requires --shard)
+        #[clap(long, conflicts_with_all = ["replace", "force"])]
+        add: bool,
+        /// Restore no matter what: replace if exists in target shard, add if not
+        #[clap(long, conflicts_with_all = ["replace", "add"])]
+        force: bool,
+    },
 }
 
 impl Command {
@@ -1757,6 +1784,21 @@ async fn main() -> anyhow::Result<()> {
 
             commands::dump::run(subgraph_store, primary_pool, deployment, directory).await
         }
+
+        Restore {
+            directory,
+            shard,
+            name,
+            replace,
+            add,
+            force,
+        } => {
+            let store = ctx.store().await;
+            let subgraph_store = store.subgraph_store();
+
+            commands::restore::run(subgraph_store, directory, shard, name, replace, add, force)
+                .await
+        }
     }
 }
 

node/src/manager/commands/mod.rs

@@ -14,6 +14,7 @@ pub mod provider_checks;
 pub mod prune;
 pub mod query;
 pub mod remove;
+pub mod restore;
 pub mod rewind;
 pub mod run;
 pub mod stats;

node/src/manager/commands/restore.rs

@@ -0,0 +1,54 @@
+use std::sync::Arc;
+
+use graph::{bail, prelude::anyhow::Result};
+
+use graph::prelude::SubgraphName;
+use graph_store_postgres::{RestoreMode, Shard, SubgraphStore, PRIMARY_SHARD};
+
+pub async fn run(
+    subgraph_store: Arc<SubgraphStore>,
+    directory: String,
+    shard: Option<String>,
+    name: Option<String>,
+    replace: bool,
+    add: bool,
+    force: bool,
+) -> Result<()> {
+    if add && shard.is_none() {
+        bail!("--add requires --shard");
+    }
+
+    let directory = std::path::Path::new(&directory).canonicalize()?;
+    let stat = std::fs::metadata(&directory)?;
+
+    if !stat.is_dir() {
+        bail!(
+            "The path `{}` is not a directory",
+            directory.to_string_lossy()
+        );
+    }
+
+    let shard = match shard {
+        Some(s) => Shard::new(s)?,
+        None => PRIMARY_SHARD.clone(),
+    };
+
+    let name = name
+        .map(|n| SubgraphName::new(n.clone()).map_err(|_| anyhow::anyhow!("invalid name `{n}`")))
+        .transpose()?;
+
+    let mode = if replace {
+        RestoreMode::Replace
+    } else if add {
+        RestoreMode::Add
+    } else if force {
+        RestoreMode::Force
+    } else {
+        RestoreMode::Default
+    };
+
+    subgraph_store
+        .restore(&directory, shard, name, mode)
+        .await?;
+    Ok(())
+}