From 8816a192c95cf539b65975469a2d61aed46f0414 Mon Sep 17 00:00:00 2001 From: Your Name Date: Tue, 26 May 2026 16:11:05 +0530 Subject: feat: initial implementation of grasp-mirror daemon GRASP mirror daemon that discovers repos from watched npubs and mirrors git data + Nostr events across all known GRASP servers for redundancy. Features: - Configurable npub watch list via .env (MIRROR_NPUBS) - TOML config for GRASP server list, index relays, storage paths - NIP-11 verification of GRASP servers on startup - Discovery of repos via kind:30617 announcements on index relays - Git mirroring (bare clone + push --mirror) to missing GRASP servers - Nostr event forwarding to all GRASP server embedded relays - SQLite state tracking for sync status and event dedup - Optional signing key for updating announcements with new clone URLs - CLI subcommands: daemon, status, verify, mirror-once Architecture: config.rs - TOML + .env config loading db.rs - SQLite state tracking health.rs - NIP-11 GRASP server verification discovery.rs - Relay subscription, kind:30617 parsing git_mirror.rs - Bare clone + push to GRASP servers nostr_mirror.rs - Event forwarding to all GRASP relays signing.rs - Optional announcement updates main.rs - CLI entry point, daemon loop --- src/config.rs | 129 ++++++++++++++++++++++++ src/db.rs | 268 +++++++++++++++++++++++++++++++++++++++++++++++++ src/discovery.rs | 142 ++++++++++++++++++++++++++ src/git_mirror.rs | 159 ++++++++++++++++++++++++++++++ src/health.rs | 136 +++++++++++++++++++++++++ src/main.rs | 279 ++++++++++++++++++++++++++++++++++++++++++++++++++++ src/nostr_mirror.rs | 139 ++++++++++++++++++++++++++ src/signing.rs | 76 ++++++++++++++ 8 files changed, 1328 insertions(+) create mode 100644 src/config.rs create mode 100644 src/db.rs create mode 100644 src/discovery.rs create mode 100644 src/git_mirror.rs create mode 100644 src/health.rs create mode 100644 src/main.rs create mode 100644 src/nostr_mirror.rs create mode 100644 src/signing.rs (limited to 'src') diff --git a/src/config.rs b/src/config.rs new file mode 100644 index 0000000..ceff44d --- /dev/null +++ b/src/config.rs @@ -0,0 +1,129 @@ +use anyhow::{Context, Result}; +use nostr::{FromBech32, PublicKey, ToBech32}; +use serde::Deserialize; +use std::path::PathBuf; + +#[derive(Debug, Deserialize)] +pub struct AppConfig { + pub discovery: DiscoveryConfig, + pub servers: ServersConfig, + pub storage: StorageConfig, + pub signing: Option, +} + +#[derive(Debug, Deserialize)] +pub struct DiscoveryConfig { + pub index_relays: Vec, + #[serde(default = "default_poll_interval")] + pub poll_interval_secs: u64, +} + +fn default_poll_interval() -> u64 { + 300 +} + +#[derive(Debug, Deserialize)] +pub struct ServersConfig { + pub known: Vec, +} + +#[derive(Debug, Deserialize)] +pub struct StorageConfig { + #[serde(default = "default_mirror_dir")] + pub mirror_dir: PathBuf, + #[serde(default = "default_database")] + pub database: PathBuf, +} + +fn default_mirror_dir() -> PathBuf { + dirs::data_dir() + .unwrap_or_else(|| PathBuf::from(".")) + .join("grasp-mirror") + .join("repos") +} + +fn default_database() -> PathBuf { + dirs::data_dir() + .unwrap_or_else(|| PathBuf::from(".")) + .join("grasp-mirror") + .join("mirror.db") +} + +#[derive(Debug, Deserialize)] +pub struct SigningConfig { + pub key_file: PathBuf, +} + +pub struct ResolvedConfig { + pub discovery: DiscoveryConfig, + pub servers: ServersConfig, + pub storage: StorageConfig, + pub signing: Option, + pub npubs: Vec, +} + +impl ResolvedConfig { + pub fn load(config_path: &PathBuf) -> Result { + let _ = dotenvy::dotenv(); + + let config_str = std::fs::read_to_string(config_path) + .with_context(|| format!("failed to read config from {:?}", config_path))?; + let app: AppConfig = toml::from_str(&config_str).context("failed to parse config.toml")?; + + let npubs = Self::load_npubs()?; + + std::fs::create_dir_all(&app.storage.mirror_dir) + .context("failed to create mirror directory")?; + if let Some(parent) = app.storage.database.parent() { + std::fs::create_dir_all(parent).context("failed to create database directory")?; + } + + Ok(Self { + discovery: app.discovery, + servers: app.servers, + storage: app.storage, + signing: app.signing, + npubs, + }) + } + + fn load_npubs() -> Result> { + let raw = std::env::var("MIRROR_NPUBS").unwrap_or_default(); + let mut npubs = Vec::new(); + + for entry in raw.split(',').map(|s| s.trim()).filter(|s| !s.is_empty()) { + let pk = if entry.starts_with("npub1") { + PublicKey::from_bech32(entry) + .with_context(|| format!("invalid npub: {}", entry))? + } else { + let bytes = hex::decode(entry) + .with_context(|| format!("invalid hex pubkey: {}", entry))?; + PublicKey::from_slice(&bytes) + .with_context(|| format!("invalid pubkey bytes: {}", entry))? + }; + npubs.push(pk); + } + + if npubs.is_empty() { + tracing::warn!("MIRROR_NPUBS is empty — no pubkeys to mirror"); + } + + Ok(npubs) + } + + pub fn relay_urls(&self) -> Vec { + let mut relays = self.discovery.index_relays.clone(); + for server in &self.servers.known { + let relay = format!( + "wss://{}", + server + .trim_start_matches("https://") + .trim_start_matches("http://") + ); + if !relays.contains(&relay) { + relays.push(relay); + } + } + relays + } +} diff --git a/src/db.rs b/src/db.rs new file mode 100644 index 0000000..bb1bf31 --- /dev/null +++ b/src/db.rs @@ -0,0 +1,268 @@ +use anyhow::{Context, Result}; +use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; +use sqlx::SqlitePool; +use std::path::Path; + +pub struct MirrorDb { + pool: SqlitePool, +} + +#[derive(Debug, sqlx::FromRow)] +pub struct RepoRecord { + pub id: i64, + pub pubkey: String, + pub identifier: String, + pub announcement_event_id: Option, + pub last_seen_at: i64, +} + +#[derive(Debug, sqlx::FromRow)] +pub struct ServerSyncRecord { + pub id: i64, + pub repo_id: i64, + pub server_domain: String, + pub git_synced: bool, + pub nostr_synced: bool, + pub last_sync_at: Option, + pub error: Option, +} + +#[derive(Debug, sqlx::FromRow)] +pub struct EventRecord { + pub id: i64, + pub event_id: String, + pub first_seen_at: i64, +} + +impl MirrorDb { + pub async fn open(db_path: &Path) -> Result { + let opts = SqliteConnectOptions::new() + .filename(db_path) + .create_if_missing(true) + .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal); + + let pool = SqlitePoolOptions::new() + .max_connections(5) + .connect_with(opts) + .await + .with_context(|| format!("failed to open database at {:?}", db_path))?; + + let db = Self { pool }; + db.run_migrations().await?; + Ok(db) + } + + async fn run_migrations(&self) -> Result<()> { + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS repos ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + pubkey TEXT NOT NULL, + identifier TEXT NOT NULL, + announcement_event_id TEXT, + last_seen_at INTEGER NOT NULL, + UNIQUE(pubkey, identifier) + ); + + CREATE TABLE IF NOT EXISTS server_syncs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + repo_id INTEGER NOT NULL REFERENCES repos(id), + server_domain TEXT NOT NULL, + git_synced INTEGER NOT NULL DEFAULT 0, + nostr_synced INTEGER NOT NULL DEFAULT 0, + last_sync_at INTEGER, + error TEXT, + UNIQUE(repo_id, server_domain) + ); + + CREATE TABLE IF NOT EXISTS seen_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + event_id TEXT NOT NULL UNIQUE, + first_seen_at INTEGER NOT NULL + ); + + CREATE INDEX IF NOT EXISTS idx_repos_pubkey ON repos(pubkey); + CREATE INDEX IF NOT EXISTS idx_server_syncs_repo ON server_syncs(repo_id); + CREATE INDEX IF NOT EXISTS idx_seen_events_id ON seen_events(event_id); + "#, + ) + .execute(&self.pool) + .await + .context("failed to run migrations")?; + + Ok(()) + } + + pub async fn upsert_repo( + &self, + pubkey: &str, + identifier: &str, + event_id: &str, + ) -> Result { + let now = chrono_now_secs(); + let result = sqlx::query_as::<_, RepoRecord>( + "SELECT * FROM repos WHERE pubkey = ? AND identifier = ?", + ) + .bind(pubkey) + .bind(identifier) + .fetch_optional(&self.pool) + .await?; + + if let Some(existing) = result { + sqlx::query("UPDATE repos SET announcement_event_id = ?, last_seen_at = ? WHERE id = ?") + .bind(event_id) + .bind(now) + .bind(existing.id) + .execute(&self.pool) + .await?; + Ok(existing.id) + } else { + sqlx::query("INSERT INTO repos (pubkey, identifier, announcement_event_id, last_seen_at) VALUES (?, ?, ?, ?)") + .bind(pubkey) + .bind(identifier) + .bind(event_id) + .bind(now) + .execute(&self.pool) + .await?; + + let row: (i64,) = sqlx::query_as("SELECT last_insert_rowid()") + .fetch_one(&self.pool) + .await?; + Ok(row.0) + } + } + + pub async fn get_repos_needing_git_sync(&self, known_servers: &[String]) -> Result)>> { + let repos = sqlx::query_as::<_, RepoRecord>("SELECT * FROM repos") + .fetch_all(&self.pool) + .await?; + + let mut result = Vec::new(); + for repo in repos { + let synced: Vec = sqlx::query_scalar::<_, String>( + "SELECT server_domain FROM server_syncs WHERE repo_id = ? AND git_synced = 1", + ) + .bind(repo.id) + .fetch_all(&self.pool) + .await?; + + let missing: Vec = known_servers + .iter() + .filter(|s| !synced.contains(s)) + .cloned() + .collect(); + + if !missing.is_empty() { + result.push((repo, missing)); + } + } + + Ok(result) + } + + pub async fn mark_git_synced( + &self, + repo_id: i64, + server_domain: &str, + ) -> Result<()> { + let now = chrono_now_secs(); + sqlx::query( + r#"INSERT INTO server_syncs (repo_id, server_domain, git_synced, nostr_synced, last_sync_at) + VALUES (?, ?, 1, 0, ?) + ON CONFLICT(repo_id, server_domain) DO UPDATE SET git_synced = 1, last_sync_at = ?, error = NULL"#, + ) + .bind(repo_id) + .bind(server_domain) + .bind(now) + .bind(now) + .execute(&self.pool) + .await?; + Ok(()) + } + + pub async fn mark_nostr_synced( + &self, + repo_id: i64, + server_domain: &str, + ) -> Result<()> { + let now = chrono_now_secs(); + sqlx::query( + r#"INSERT INTO server_syncs (repo_id, server_domain, git_synced, nostr_synced, last_sync_at) + VALUES (?, ?, 0, 1, ?) + ON CONFLICT(repo_id, server_domain) DO UPDATE SET nostr_synced = 1, last_sync_at = ?, error = NULL"#, + ) + .bind(repo_id) + .bind(server_domain) + .bind(now) + .bind(now) + .execute(&self.pool) + .await?; + Ok(()) + } + + pub async fn mark_sync_error( + &self, + repo_id: i64, + server_domain: &str, + error: &str, + ) -> Result<()> { + let now = chrono_now_secs(); + sqlx::query( + r#"INSERT INTO server_syncs (repo_id, server_domain, git_synced, nostr_synced, last_sync_at, error) + VALUES (?, ?, 0, 0, ?, ?) + ON CONFLICT(repo_id, server_domain) DO UPDATE SET error = ?, last_sync_at = ?"#, + ) + .bind(repo_id) + .bind(server_domain) + .bind(now) + .bind(error) + .bind(error) + .bind(now) + .execute(&self.pool) + .await?; + Ok(()) + } + + pub async fn have_seen_event(&self, event_id: &str) -> Result { + let result: Option<(i64,)> = sqlx::query_as( + "SELECT id FROM seen_events WHERE event_id = ?", + ) + .bind(event_id) + .fetch_optional(&self.pool) + .await?; + Ok(result.is_some()) + } + + pub async fn record_event(&self, event_id: &str) -> Result<()> { + let now = chrono_now_secs(); + sqlx::query("INSERT OR IGNORE INTO seen_events (event_id, first_seen_at) VALUES (?, ?)") + .bind(event_id) + .bind(now) + .execute(&self.pool) + .await?; + Ok(()) + } + + pub async fn get_all_repos(&self) -> Result> { + let repos = sqlx::query_as::<_, RepoRecord>("SELECT * FROM repos ORDER BY last_seen_at DESC") + .fetch_all(&self.pool) + .await?; + Ok(repos) + } + + pub async fn get_sync_summary(&self) -> Result> { + let records = sqlx::query_as::<_, ServerSyncRecord>( + "SELECT * FROM server_syncs ORDER BY last_sync_at DESC NULLS LAST", + ) + .fetch_all(&self.pool) + .await?; + Ok(records) + } +} + +fn chrono_now_secs() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64 +} diff --git a/src/discovery.rs b/src/discovery.rs new file mode 100644 index 0000000..59b9f2c --- /dev/null +++ b/src/discovery.rs @@ -0,0 +1,142 @@ +use crate::db::MirrorDb; +use crate::health::GraspServer; +use anyhow::{Context, Result}; +use nostr::Kind; +use nostr_sdk::prelude::*; +use std::collections::HashMap; + +#[derive(Debug, Clone)] +pub struct DiscoveredRepo { + pub pubkey: PublicKey, + pub identifier: String, + pub event_id: EventId, + pub clone_urls: Vec, + pub relay_urls: Vec, +} + +pub async fn discover_repos_from_relays( + client: &nostr_sdk::Client, + npubs: &[PublicKey], + relay_urls: &[String], +) -> Result> { + for url in relay_urls { + if let Err(e) = client.add_relay(url).await { + tracing::warn!(relay = %url, error = %e, "failed to add relay"); + continue; + } + } + client.connect().await; + + let mut repos = Vec::new(); + + for pk in npubs { + let filter = Filter::new() + .kind(Kind::Custom(30617)) + .author(*pk) + .limit(50); + + let events = client + .fetch_events(filter, std::time::Duration::from_secs(30)) + .await + .context("failed to fetch events from relays")?; + + for event in events.into_iter() { + if let Some(repo) = parse_announcement(&event) { + repos.push(repo); + } + } + } + + tracing::info!(count = repos.len(), "discovered repos from relays"); + Ok(repos) +} + +fn parse_announcement(event: &Event) -> Option { + let mut identifier = None; + let mut clone_urls = Vec::new(); + let mut relay_urls = Vec::new(); + + for tag in event.tags.iter() { + let tag_vec = tag.clone().to_vec(); + if tag_vec.len() < 2 { + continue; + } + match tag_vec[0].as_str() { + "d" => identifier = Some(tag_vec[1].clone()), + "clone" => clone_urls.push(tag_vec[1].clone()), + "relays" => relay_urls.push(tag_vec[1].clone()), + _ => {} + } + } + + let identifier = identifier?; + if clone_urls.is_empty() { + tracing::warn!( + identifier = %identifier, + "repo announcement has no clone URLs" + ); + } + + Some(DiscoveredRepo { + pubkey: event.pubkey, + identifier, + event_id: event.id, + clone_urls, + relay_urls, + }) +} + +pub async fn persist_discovered_repos( + db: &MirrorDb, + repos: &[DiscoveredRepo], +) -> Result> { + let mut repo_ids = HashMap::new(); + for repo in repos { + let pk_hex = repo.pubkey.to_hex(); + match db + .upsert_repo(&pk_hex, &repo.identifier, &repo.event_id.to_hex()) + .await + { + Ok(id) => { + tracing::debug!( + identifier = %repo.identifier, + repo_id = id, + "persisted repo" + ); + repo_ids.insert(format!("{}:{}", pk_hex, repo.identifier), id); + } + Err(e) => { + tracing::error!( + identifier = %repo.identifier, + error = %e, + "failed to persist repo" + ); + } + } + } + Ok(repo_ids) +} + +pub fn identify_missing_servers( + repo: &DiscoveredRepo, + servers: &HashMap, +) -> Vec { + let mut missing = Vec::new(); + + for (_domain, server) in servers { + if !server.is_grasp_server() { + continue; + } + + let has_clone = repo + .clone_urls + .iter() + .any(|url| url.contains(&server.domain)); + + if !has_clone { + missing.push(server.clone()); + } + } + + missing +} diff --git a/src/git_mirror.rs b/src/git_mirror.rs new file mode 100644 index 0000000..47c0442 --- /dev/null +++ b/src/git_mirror.rs @@ -0,0 +1,159 @@ +use crate::db::MirrorDb; +use crate::discovery::DiscoveredRepo; +use crate::health::GraspServer; +use anyhow::{Context, Result}; +use git2::RemoteCallbacks; +use std::path::Path; + +pub struct GitMirror { + mirror_dir: std::path::PathBuf, +} + +impl GitMirror { + pub fn new(mirror_dir: &Path) -> Self { + Self { + mirror_dir: mirror_dir.to_path_buf(), + } + } + + fn repo_path(&self, pubkey: &str, identifier: &str) -> std::path::PathBuf { + self.mirror_dir + .join(pubkey) + .join(format!("{}.git", identifier)) + } + + pub async fn mirror_repo_to_servers( + &self, + db: &MirrorDb, + repo: &DiscoveredRepo, + target_servers: &[GraspServer], + ) -> Result<()> { + if target_servers.is_empty() { + tracing::debug!( + identifier = %repo.identifier, + "no missing servers to mirror to" + ); + return Ok(()); + } + + let pk_hex = repo.pubkey.to_hex(); + let repo_path = self.repo_path(&pk_hex, &repo.identifier); + + if !repo_path.exists() { + self.clone_bare(&repo_path, &repo.clone_urls)?; + } + + for server in target_servers { + let target_url = server.clone_url(&pk_hex, &repo.identifier); + + tracing::info!( + identifier = %repo.identifier, + server = %server.domain, + target = %target_url, + "mirroring git data" + ); + + let repo_id = db.get_all_repos().await.ok().and_then(|repos| { + repos + .iter() + .find(|r| r.pubkey == pk_hex && r.identifier == repo.identifier) + .map(|r| r.id) + }); + + match self.push_mirror(&repo_path, &target_url) { + Ok(()) => { + tracing::info!( + identifier = %repo.identifier, + server = %server.domain, + "git mirror succeeded" + ); + if let Some(id) = repo_id { + let _ = db.mark_git_synced(id, &server.domain).await; + } + } + Err(e) => { + tracing::error!( + identifier = %repo.identifier, + server = %server.domain, + error = %e, + "git mirror failed" + ); + if let Some(id) = repo_id { + let _ = db.mark_sync_error(id, &server.domain, &e.to_string()).await; + } + } + } + } + + Ok(()) + } + + fn clone_bare(&self, repo_path: &Path, clone_urls: &[String]) -> Result<()> { + if let Some(parent) = repo_path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("failed to create {:?}", parent))?; + } + + let mut last_error = None; + + for url in clone_urls { + if url.is_empty() { + continue; + } + tracing::info!(url = %url, path = ?repo_path, "cloning bare repo"); + + let mut callbacks = RemoteCallbacks::new(); + callbacks.credentials(|_url, _username, _allowed| git2::Cred::default()); + + let mut fetch_opts = git2::FetchOptions::new(); + fetch_opts.remote_callbacks(callbacks); + + let mut builder = git2::build::RepoBuilder::new(); + builder.bare(true).fetch_options(fetch_opts); + + match builder.clone(url, repo_path) { + Ok(_) => { + tracing::info!(url = %url, "bare clone succeeded"); + return Ok(()); + } + Err(e) => { + tracing::warn!(url = %url, error = %e, "clone failed, trying next URL"); + last_error = Some(e); + if repo_path.exists() { + let _ = std::fs::remove_dir_all(repo_path); + } + } + } + } + + let err = last_error.unwrap_or_else(|| git2::Error::from_str("no clone URLs available")); + Err(err).with_context(|| format!("all clone attempts failed for {:?}", repo_path)) + } + + fn push_mirror(&self, repo_path: &Path, target_url: &str) -> Result<()> { + let repo = git2::Repository::open(repo_path) + .with_context(|| format!("failed to open bare repo at {:?}", repo_path))?; + + let mut remote = repo.remote("push_target", target_url)?; + + let mut callbacks = RemoteCallbacks::new(); + callbacks.credentials(|_url, _username, _allowed| git2::Cred::default()); + callbacks.push_update_reference(|_refname, status| { + if let Some(s) = status { + tracing::warn!(status = %s, "push rejected"); + } + Ok(()) + }); + + let mut push_opts = git2::PushOptions::new(); + push_opts.remote_callbacks(callbacks); + + let refspecs = ["+refs/*:refs/*"]; + + remote + .push(&refspecs, Some(&mut push_opts)) + .with_context(|| format!("failed to push mirror to {}", target_url))?; + + Ok(()) + } +} diff --git a/src/health.rs b/src/health.rs new file mode 100644 index 0000000..2853239 --- /dev/null +++ b/src/health.rs @@ -0,0 +1,136 @@ +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Nip11Info { + pub name: Option, + pub description: Option, + pub supported_nips: Option>, + pub supported_grasps: Option>, + pub software: Option, + pub version: Option, +} + +#[derive(Debug, Clone)] +pub struct GraspServer { + pub domain: String, + pub relay_url: String, + pub clone_url_prefix: String, + pub nip11: Option, + pub healthy: bool, +} + +impl GraspServer { + pub fn from_domain(domain: &str) -> Self { + let clean = domain + .trim_start_matches("https://") + .trim_start_matches("http://") + .trim_end_matches('/') + .to_string(); + Self { + relay_url: format!("wss://{}", clean), + clone_url_prefix: format!("https://{}", clean), + nip11: None, + healthy: false, + domain: clean, + } + } + + pub fn clone_url(&self, npub_hex: &str, identifier: &str) -> String { + format!("{}/{}/{}.git", self.clone_url_prefix, npub_hex, identifier) + } + + pub fn is_grasp_server(&self) -> bool { + self.nip11 + .as_ref() + .map(|info| info.supported_grasps.is_some()) + .unwrap_or(false) + } +} + +pub async fn verify_grasp_server(domain: &str) -> Result { + let mut server = GraspServer::from_domain(domain); + let nip11_url = format!("https://{}", server.domain); + + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(10)) + .build()?; + + let resp = client + .get(&nip11_url) + .header("Accept", "application/nostr+json") + .send() + .await; + + match resp { + Ok(resp) if resp.status().is_success() => { + match resp.json::().await { + Ok(info) => { + let is_grasp = info.supported_grasps.is_some(); + if is_grasp { + tracing::info!( + domain = %server.domain, + grasps = ?info.supported_grasps, + version = ?info.version, + "verified GRASP server" + ); + } else { + tracing::warn!( + domain = %server.domain, + "server responded to NIP-11 but has no supported_grasps" + ); + } + server.healthy = is_grasp; + server.nip11 = Some(info); + } + Err(e) => { + tracing::warn!(domain = %server.domain, error = %e, "failed to parse NIP-11 response"); + } + } + } + Ok(resp) => { + tracing::warn!( + domain = %server.domain, + status = %resp.status(), + "NIP-11 check returned non-success" + ); + } + Err(e) => { + tracing::warn!(domain = %server.domain, error = %e, "NIP-11 check failed"); + } + } + + Ok(server) +} + +pub async fn verify_all_servers(domains: &[String]) -> HashMap { + let mut servers = HashMap::new(); + let mut set = tokio::task::JoinSet::new(); + + for d in domains { + let domain = d.clone(); + set.spawn(async move { + let result = verify_grasp_server(&domain).await; + (domain, result) + }); + } + + while let Some(res) = set.join_next().await { + match res { + Ok((domain, result)) => match result { + Ok(server) => { + servers.insert(domain, server); + } + Err(e) => { + tracing::error!(domain = %domain, error = %e, "failed to verify server"); + } + }, + Err(e) => { + tracing::error!(error = %e, "task panicked during server verification"); + } + } + } + + servers +} diff --git a/src/main.rs b/src/main.rs new file mode 100644 index 0000000..b709d44 --- /dev/null +++ b/src/main.rs @@ -0,0 +1,279 @@ +mod config; +mod db; +mod discovery; +mod git_mirror; +mod health; +mod nostr_mirror; +mod signing; + +use anyhow::Result; +use clap::Parser; +use std::path::PathBuf; +use std::sync::Arc; + +#[derive(Parser, Debug)] +#[command(name = "grasp-mirror", about = "GRASP mirror daemon")] +struct Cli { + #[arg(short, long, default_value = "config.toml")] + config: PathBuf, + + #[command(subcommand)] + command: Option, +} + +#[derive(clap::Subcommand, Debug)] +enum Command { + Daemon, + Status, + Verify, + MirrorOnce, +} + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .init(); + + let cli = Cli::parse(); + let config = config::ResolvedConfig::load(&cli.config)?; + + tracing::info!( + npubs = config.npubs.len(), + servers = config.servers.known.len(), + index_relays = config.discovery.index_relays.len(), + "starting grasp-mirror" + ); + + let db = db::MirrorDb::open(&config.storage.database).await?; + + match cli.command.unwrap_or(Command::Daemon) { + Command::Daemon => run_daemon(config, db).await, + Command::Status => run_status(db).await, + Command::Verify => run_verify(config).await, + Command::MirrorOnce => run_mirror_once(config, db).await, + } +} + +fn build_nostr_client(relay_urls: &[String]) -> nostr_sdk::Client { + let client = nostr_sdk::Client::default(); + for url in relay_urls { + let _ = client.add_relay(url); + } + client +} + +async fn run_daemon(config: config::ResolvedConfig, db: db::MirrorDb) -> Result<()> { + let db = Arc::new(db); + let config = Arc::new(config); + + let servers = health::verify_all_servers(&config.servers.known).await; + let healthy: Vec<_> = servers + .values() + .filter(|s| s.is_grasp_server()) + .cloned() + .collect(); + + tracing::info!( + total = servers.len(), + healthy = healthy.len(), + "server verification complete" + ); + + let healthy = Arc::new(healthy); + + let relay_urls = config.relay_urls(); + let nostr_client = build_nostr_client(&relay_urls); + nostr_client.connect().await; + + let mirror = git_mirror::GitMirror::new(&config.storage.mirror_dir); + let nostr_mirror = nostr_mirror::NostrMirror::new(nostr_client.clone()); + + let mut interval = tokio::time::interval(std::time::Duration::from_secs( + config.discovery.poll_interval_secs, + )); + + tracing::info!( + "daemon started, polling every {}s", + config.discovery.poll_interval_secs + ); + + loop { + tokio::select! { + _ = interval.tick() => { + if let Err(e) = mirror_cycle( + &config, + &db, + &nostr_client, + &mirror, + &nostr_mirror, + &healthy, + ).await { + tracing::error!(error = %e, "mirror cycle failed"); + } + } + _ = tokio::signal::ctrl_c() => { + tracing::info!("shutting down"); + break; + } + } + } + + Ok(()) +} + +async fn mirror_cycle( + config: &Arc, + db: &Arc, + nostr_client: &nostr_sdk::Client, + mirror: &git_mirror::GitMirror, + nostr_mirror: &nostr_mirror::NostrMirror, + servers: &Arc>, +) -> Result<()> { + tracing::info!("starting mirror cycle"); + + let relay_urls = config.relay_urls(); + let repos = discovery::discover_repos_from_relays(nostr_client, &config.npubs, &relay_urls) + .await?; + + tracing::info!(count = repos.len(), "discovered repos"); + + discovery::persist_discovered_repos(db, &repos).await?; + + let server_map: std::collections::HashMap = servers + .iter() + .map(|s| (s.domain.clone(), s.clone())) + .collect(); + + for repo in &repos { + let missing = discovery::identify_missing_servers(repo, &server_map); + + if missing.is_empty() { + tracing::debug!(identifier = %repo.identifier, "repo already on all servers"); + continue; + } + + tracing::info!( + identifier = %repo.identifier, + missing = missing.iter().map(|s| s.domain.as_str()).collect::>().join(", "), + "mirroring to missing servers" + ); + + mirror.mirror_repo_to_servers(db, repo, &missing).await?; + nostr_mirror.forward_repo_events(db, repo, servers).await?; + } + + nostr_mirror + .sync_all_events(db, &config.npubs, servers) + .await?; + + tracing::info!("mirror cycle complete"); + Ok(()) +} + +async fn run_status(db: db::MirrorDb) -> Result<()> { + let repos = db.get_all_repos().await?; + let syncs = db.get_sync_summary().await?; + + println!("=== GRASP Mirror Status ===\n"); + println!("Repos tracked: {}\n", repos.len()); + + for repo in &repos { + println!(" {} ({})", repo.identifier, &repo.pubkey[..12]); + } + + println!("\nSync records: {}\n", syncs.len()); + for sync in &syncs { + let status = if sync.git_synced && sync.nostr_synced { + "OK" + } else if sync.error.is_some() { + "ERR" + } else { + "PENDING" + }; + println!( + " repo:{} server:{} git:{} nostr:{} [{}]", + sync.repo_id, + sync.server_domain, + if sync.git_synced { "Y" } else { "N" }, + if sync.nostr_synced { "Y" } else { "N" }, + status, + ); + if let Some(err) = &sync.error { + println!(" error: {}", err); + } + } + + Ok(()) +} + +async fn run_verify(config: config::ResolvedConfig) -> Result<()> { + let servers = health::verify_all_servers(&config.servers.known).await; + + println!("=== GRASP Server Verification ===\n"); + + for (domain, server) in &servers { + let grasp = server.is_grasp_server(); + let version = server + .nip11 + .as_ref() + .and_then(|i| i.version.clone()) + .unwrap_or_else(|| "unknown".to_string()); + let grasps = server + .nip11 + .as_ref() + .and_then(|i| i.supported_grasps.clone()) + .map(|g| g.join(", ")) + .unwrap_or_else(|| "NONE".to_string()); + + println!( + " {} [{}] v={} GRASP={}", + domain, + if grasp { "OK" } else { "SKIP" }, + version, + grasps, + ); + } + + let healthy = servers.values().filter(|s| s.is_grasp_server()).count(); + println!( + "\n{}/{} verified as GRASP servers", + healthy, + servers.len() + ); + + Ok(()) +} + +async fn run_mirror_once(config: config::ResolvedConfig, db: db::MirrorDb) -> Result<()> { + let db = Arc::new(db); + let config = Arc::new(config); + + let servers = health::verify_all_servers(&config.servers.known).await; + let healthy: Vec<_> = servers + .values() + .filter(|s| s.is_grasp_server()) + .cloned() + .collect(); + + let relay_urls = config.relay_urls(); + let nostr_client = build_nostr_client(&relay_urls); + nostr_client.connect().await; + + let mirror = git_mirror::GitMirror::new(&config.storage.mirror_dir); + let nostr_mirror = nostr_mirror::NostrMirror::new(nostr_client.clone()); + let healthy = Arc::new(healthy); + + mirror_cycle( + &config, + &db, + &nostr_client, + &mirror, + &nostr_mirror, + &healthy, + ) + .await +} diff --git a/src/nostr_mirror.rs b/src/nostr_mirror.rs new file mode 100644 index 0000000..76f66d0 --- /dev/null +++ b/src/nostr_mirror.rs @@ -0,0 +1,139 @@ +use crate::db::MirrorDb; +use crate::discovery::DiscoveredRepo; +use crate::health::GraspServer; +use anyhow::Result; +use nostr::Kind; +use nostr_sdk::prelude::*; + +pub struct NostrMirror { + client: nostr_sdk::Client, +} + +impl NostrMirror { + pub fn new(client: nostr_sdk::Client) -> Self { + Self { client } + } + + pub async fn forward_events_to_servers( + &self, + db: &MirrorDb, + events: &[Event], + servers: &[GraspServer], + ) -> Result<()> { + for event in events { + if db.have_seen_event(&event.id.to_hex()).await? { + continue; + } + + for server in servers { + if !server.is_grasp_server() { + continue; + } + + tracing::debug!( + event_id = %event.id.to_hex(), + kind = event.kind.as_u16(), + server = %server.domain, + "forwarding event" + ); + + let url: RelayUrl = RelayUrl::parse(&server.relay_url)?; + let urls = vec![url]; + + match self.client.send_event_to(urls, event.clone()).await { + Ok(_) => { + tracing::debug!( + event_id = %event.id.to_hex(), + server = %server.domain, + "event forwarded" + ); + } + Err(e) => { + tracing::warn!( + event_id = %event.id.to_hex(), + server = %server.domain, + error = %e, + "failed to forward event" + ); + } + } + } + + let _ = db.record_event(&event.id.to_hex()).await; + } + + Ok(()) + } + + pub async fn forward_repo_events( + &self, + db: &MirrorDb, + repo: &DiscoveredRepo, + servers: &[GraspServer], + ) -> Result<()> { + let filters = vec![ + Filter::new() + .kind(Kind::Custom(30617)) + .author(repo.pubkey) + .identifier(&repo.identifier), + Filter::new() + .kind(Kind::Custom(30618)) + .author(repo.pubkey) + .identifier(&repo.identifier), + ]; + + let mut all_events = Vec::new(); + for filter in filters { + let events = self + .client + .fetch_events(filter, std::time::Duration::from_secs(15)) + .await?; + all_events.extend(events); + } + + if all_events.is_empty() { + tracing::debug!(identifier = %repo.identifier, "no events to forward"); + return Ok(()); + } + + tracing::info!( + identifier = %repo.identifier, + count = all_events.len(), + "forwarding repo events" + ); + + self.forward_events_to_servers(db, &all_events, servers).await + } + + pub async fn sync_all_events( + &self, + db: &MirrorDb, + npubs: &[PublicKey], + servers: &[GraspServer], + ) -> Result<()> { + let git_kinds = [ + Kind::Custom(30617), + Kind::Custom(30618), + Kind::Custom(1631), + Kind::Custom(1642), + Kind::EventDeletion, + ]; + + let mut all_events = Vec::new(); + + for pk in npubs { + for kind in &git_kinds { + let filter = Filter::new().kind(*kind).author(*pk).limit(100); + let events = self + .client + .fetch_events(filter, std::time::Duration::from_secs(30)) + .await?; + all_events.extend(events); + } + } + + tracing::info!(count = all_events.len(), "fetched events for forwarding"); + + self.forward_events_to_servers(db, &all_events, servers).await + } +} diff --git a/src/signing.rs b/src/signing.rs new file mode 100644 index 0000000..7ec796a --- /dev/null +++ b/src/signing.rs @@ -0,0 +1,76 @@ +use crate::discovery::DiscoveredRepo; +use crate::health::GraspServer; +use anyhow::{Context, Result}; +use nostr_sdk::prelude::*; + +pub struct OptionalSigner { + keys: Option, +} + +impl OptionalSigner { + pub fn none() -> Self { + Self { keys: None } + } + + pub fn from_nsec(nsec: &str) -> Result { + let keys = Keys::parse(nsec).context("failed to parse nsec")?; + Ok(Self { keys: Some(keys) }) + } + + pub fn is_available(&self) -> bool { + self.keys.is_some() + } + + pub async fn build_updated_announcement( + &self, + original: &DiscoveredRepo, + additional_servers: &[GraspServer], + ) -> Result> { + let keys = match &self.keys { + Some(k) => k, + None => { + tracing::debug!("no signer available, skipping announcement update"); + return Ok(None); + } + }; + + if additional_servers.is_empty() { + return Ok(None); + } + + let pk_hex = original.pubkey.to_hex(); + + let mut new_clone_urls: Vec = original.clone_urls.clone(); + for server in additional_servers { + let url = server.clone_url(&pk_hex, &original.identifier); + if !new_clone_urls.contains(&url) { + new_clone_urls.push(url); + } + } + + let mut tags: Vec = vec![ + Tag::custom(TagKind::Custom("d".into()), [&original.identifier]), + ]; + + for url in &new_clone_urls { + tags.push(Tag::custom(TagKind::Custom("clone".into()), [url.as_str()])); + } + for url in &original.relay_urls { + tags.push(Tag::custom( + TagKind::Custom("relays".into()), + [url.as_str()], + )); + } + + let builder = EventBuilder::new(Kind::Custom(30617), "").tags(tags); + let event = builder.sign_with_keys(keys)?; + + tracing::info!( + identifier = %original.identifier, + added = additional_servers.len(), + "built updated announcement with additional clone URLs" + ); + + Ok(Some(event)) + } +} -- cgit v1.2.3