18525274937

Committed 15 Oct 2025 10:05AM UTC coverage: 35.433% (+0.007%) from 35.426%

Build # 18525274937

Build Type

push

github

Committed by

0xmichalis

Commit Message

refactor: give more accurate names to ipfs pinning-related vars

Coverage Stats

14 of 34 new or added lines in 10 files covered. (41.18%)

1406 of 3968 relevant lines covered (35.43%)

6.02 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/bin/server.rs

use clap::Parser;
use dotenv::dotenv;
use std::env;
use std::io::Write;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use tokio::signal;
use tokio::sync::mpsc;
use tracing::{error, info};

use nftbk::envvar::is_defined;
use nftbk::logging;
use nftbk::logging::LogLevel;

use nftbk::server::pin_monitor::run_pin_monitor;
use nftbk::server::pruner::run_pruner;
use nftbk::server::router::build_router;
use nftbk::server::{
    recover_incomplete_tasks, spawn_backup_workers, AppState, BackupTaskOrShutdown,
};

#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None)]
struct Args {
    /// The address to listen on
    #[arg(long, default_value = "127.0.0.1:8080")]
    listen_address: String,

    /// The path to the chains configuration file
    #[arg(short = 'c', long, default_value = "config_chains.toml")]
    chain_config: String,

    /// The base directory to save the backup to
    #[arg(long, default_value = "/tmp")]
    base_dir: String,

    /// Set the log level
    #[arg(short, long, value_enum, default_value = "info")]
    log_level: LogLevel,

    /// Skip checksum verification
    #[arg(long, default_value_t = false, action = clap::ArgAction::Set)]
    unsafe_skip_checksum_check: bool,

    /// Enable the pruner thread
    #[arg(long, default_value_t = true, action = clap::ArgAction::Set)]
    enable_pruner: bool,

    /// Pruner retention period in days
    #[arg(long, default_value_t = 3)]
    pruner_retention_days: u64,

    /// Pruner interval in seconds
    #[arg(long, default_value_t = 3600)]
    pruner_interval_seconds: u64,

    /// Pruner regex pattern for file names to prune
    #[arg(long, default_value = "^nftbk-")]
    pruner_pattern: String,

    /// Pin monitor interval in seconds
    #[arg(long, default_value_t = 120)]
    pin_monitor_interval_seconds: u64,

    /// Number of backup worker threads to run in parallel
    #[arg(long, default_value_t = 4)]
    backup_parallelism: usize,

    /// Maximum number of backup tasks to queue before blocking
    #[arg(long, default_value_t = 10000)]
    backup_queue_size: usize,

    /// Disable colored log output
    #[arg(long, default_value_t = false, action = clap::ArgAction::Set)]
    no_color: bool,

    /// Path to a TOML file with one or more Privy credential sets
    /// When provided, these are used in addition to any PRIVY_* env vars
    #[arg(long)]
    privy_config: Option<String>,

    /// Path to a TOML file with IPFS provider configuration
    /// When provided, this is used instead of IPFS_* env vars
    #[arg(long)]
    ipfs_config: Option<String>,
}

#[derive(serde::Deserialize, Clone, Debug)]
struct PrivyCredential {
    app_id: String,
    verification_key: String,
}

#[derive(serde::Deserialize)]
struct PrivyFile {
    privy: Vec<PrivyCredential>,
}

#[derive(serde::Deserialize)]
struct IpfsConfigFile {
    ipfs_pinning_configs: Vec<nftbk::ipfs::IpfsPinningConfig>,
}

#[tokio::main]
async fn main() {
    // We are consuming config both from the environment and from the command line
    dotenv().ok();
    let args = Args::parse();
    logging::init(args.log_level.clone(), !args.no_color);
    info!(
        "Starting {} {} (commit {})",
        env!("CARGO_BIN_NAME"),
        env!("CARGO_PKG_VERSION"),
        env!("GIT_COMMIT")
    );
    let auth_token = env::var("NFTBK_AUTH_TOKEN").ok();

    // Load Privy credentials from file if provided
    let mut privy_credentials: Vec<PrivyCredential> = Vec::new();
    if let Some(path) = &args.privy_config {
        match std::fs::read_to_string(path) {
            Ok(contents) => match toml::from_str::<PrivyFile>(&contents) {
                Ok(file) => {
                    privy_credentials = file
                        .privy
                        .into_iter()
                        .map(|mut c| {
                            // Allow \n escaping in inline keys if users choose to
                            c.verification_key = c.verification_key.replace("\\n", "\n");
                            c
                        })
                        .collect();
                }
                Err(e) => {
                    tracing::error!("Failed to parse Privy config file '{}': {}", path, e);
                }
            },
            Err(e) => {
                tracing::error!("Failed to read Privy config file '{}': {}", path, e);
            }
        }
    }

    // Load IPFS provider configuration from file if provided
    let ipfs_pinning_configs = if args.ipfs_config.is_none() {
        // No config file, use empty list (AppState will fall back to env vars)
        Vec::new()
    } else {
        let path = args.ipfs_config.as_ref().unwrap();
        match std::fs::read_to_string(path) {
            Ok(contents) => match toml::from_str::<IpfsConfigFile>(&contents) {
                Ok(file) => {
                    info!(
                        "Loaded {} IPFS pinning provider(s) from config file '{}'",
                        file.ipfs_pinning_configs.len(),
                        path
                    );
                    file.ipfs_pinning_configs
                }
                Err(e) => {
                    error!("Failed to parse IPFS config file '{}': {}", path, e);
                    std::process::exit(1);
                }
            },
            Err(e) => {
                error!("Failed to read IPFS config file '{}': {}", path, e);
                std::process::exit(1);
            }
        }
    };

    let (backup_task_sender, backup_task_receiver) =
        mpsc::channel::<BackupTaskOrShutdown>(args.backup_queue_size);
    let db_url =
        std::env::var("DATABASE_URL").expect("DATABASE_URL env var must be set for Postgres");
    let shutdown_flag = Arc::new(AtomicBool::new(false));
    let state = AppState::new(
        &args.chain_config,
        &args.base_dir,
        args.unsafe_skip_checksum_check,
        auth_token.clone(),
        args.enable_pruner,
        args.pruner_retention_days,
        backup_task_sender.clone(),
        &db_url,
        (args.backup_queue_size + 1) as u32,
        shutdown_flag.clone(),
        ipfs_pinning_configs,
    )
    .await;

    info!("Starting server with options: {:?}", args);
    info!(
        "Symmetric authentication enabled: {}",
        is_defined(&auth_token)
    );
    info!(
        "Privy JWT authentication enabled: {} ({} credential set(s))",
        !privy_credentials.is_empty(),
        privy_credentials.len()
    );

    // Spawn worker pool for backup tasks
    let worker_handles =
        spawn_backup_workers(args.backup_parallelism, backup_task_receiver, state.clone());

    // Recover incomplete backup tasks from previous server runs
    match recover_incomplete_tasks(&state.db, &state.backup_task_sender).await {
        Ok(count) => {
            if count > 0 {
                info!("Successfully recovered {} incomplete backup tasks", count);
            }
        }
        Err(e) => {
            error!("Failed to recover incomplete backup tasks: {}", e);
            // Don't exit the server, just log the error and continue
        }
    }

    // Start the pruner thread
    let pruner_handle = if !args.enable_pruner {
        None
    } else {
        let db = state.db.clone();
        let base_dir = args.base_dir.clone();
        let interval = args.pruner_interval_seconds;
        let shutdown_flag = state.shutdown_flag.clone();
        Some(tokio::spawn(async move {
            run_pruner(db, base_dir, interval, shutdown_flag).await;
        }))
    };

    // Start the pin monitor thread if IPFS providers are configured
    let pin_monitor_handle = if state.ipfs_pinning_instances.is_empty() {
        None
    } else {
        let db = state.db.clone();
        let providers = state.ipfs_pinning_instances.clone();
        let interval = args.pin_monitor_interval_seconds;
        let shutdown_flag = state.shutdown_flag.clone();
        info!(
            "Starting pin monitor with {} IPFS provider(s) and {} second interval",
            providers.len(),
            interval
        );
        Some(tokio::spawn(async move {
            run_pin_monitor(db, providers.to_vec(), interval, shutdown_flag).await;
        }))
    };

    // Add graceful shutdown
    let shutdown_signal = async move {
        let mut sigterm = signal::unix::signal(signal::unix::SignalKind::terminate())
            .expect("failed to install SIGTERM handler");

        tokio::select! {
            _ = signal::ctrl_c() => {
                info!("Received SIGINT (Ctrl+C), shutting down server...");
            }
            _ = sigterm.recv() => {
                info!("Received SIGTERM, shutting down server...");
            }
        }
        shutdown_flag.store(true, Ordering::SeqCst);
    };

    // Start the server
    let addr: SocketAddr = args.listen_address.parse().expect("Invalid listen address");
    let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
    let app = build_router(
        state.clone(),
        privy_credentials
            .into_iter()
            .map(|c| (c.app_id, c.verification_key))
            .collect(),
    );
    info!("Listening on {}", addr);
    axum::serve(listener, app)
        .with_graceful_shutdown(shutdown_signal)
        .await
        .unwrap();
    info!("Server has exited");

    if let Some(handle) = pruner_handle {
        let _ = handle.await;
    }
    info!("Pruner has exited");

    if let Some(handle) = pin_monitor_handle {
        let _ = handle.await;
    }
    info!("Pin monitor has exited");

    // On shutdown, send one Shutdown message per worker
    for _ in 0..args.backup_parallelism {
        let _ = state
            .backup_task_sender
            .send(BackupTaskOrShutdown::Shutdown)
            .await;
    }
    // Drop the last sender to close the channel and signal workers to exit
    drop(state.backup_task_sender);
    info!("Backup task sender has exited");

    // Wait for all workers to finish
    for handle in worker_handles {
        let _ = handle.await;
    }
    info!("Backup workers have exited");

    // Give time for final logs to flush
    let _ = std::io::stdout().flush();
    std::thread::sleep(std::time::Duration::from_millis(200));
}

1	use clap::Parser;
2	use dotenv::dotenv;
3	use std::env;
4	use std::io::Write;
5	use std::net::SocketAddr;
6	use std::sync::atomic::{AtomicBool, Ordering};
7	use std::sync::Arc;
8	use tokio::signal;
9	use tokio::sync::mpsc;
10	use tracing::{error, info};
11
12	use nftbk::envvar::is_defined;
13	use nftbk::logging;
14	use nftbk::logging::LogLevel;
15
16	use nftbk::server::pin_monitor::run_pin_monitor;
17	use nftbk::server::pruner::run_pruner;
18	use nftbk::server::router::build_router;
19	use nftbk::server::{
20	recover_incomplete_tasks, spawn_backup_workers, AppState, BackupTaskOrShutdown,
21	};
22
23	#[derive(Parser, Debug)]
24	#[command(author, version, about, long_about = None)]
25	struct Args {
26	/// The address to listen on
27	#[arg(long, default_value = "127.0.0.1:8080")]
28	listen_address: String,
29
30	/// The path to the chains configuration file
31	#[arg(short = 'c', long, default_value = "config_chains.toml")]
32	chain_config: String,
33
34	/// The base directory to save the backup to
35	#[arg(long, default_value = "/tmp")]
36	base_dir: String,
37
38	/// Set the log level
39	#[arg(short, long, value_enum, default_value = "info")]
40	log_level: LogLevel,
41
42	/// Skip checksum verification
43	#[arg(long, default_value_t = false, action = clap::ArgAction::Set)]
44	unsafe_skip_checksum_check: bool,
45
46	/// Enable the pruner thread
47	#[arg(long, default_value_t = true, action = clap::ArgAction::Set)]
48	enable_pruner: bool,
49
50	/// Pruner retention period in days
51	#[arg(long, default_value_t = 3)]
52	pruner_retention_days: u64,
53
54	/// Pruner interval in seconds
55	#[arg(long, default_value_t = 3600)]
56	pruner_interval_seconds: u64,
57
58	/// Pruner regex pattern for file names to prune
59	#[arg(long, default_value = "^nftbk-")]
60	pruner_pattern: String,
61
62	/// Pin monitor interval in seconds
63	#[arg(long, default_value_t = 120)]
64	pin_monitor_interval_seconds: u64,
65
66	/// Number of backup worker threads to run in parallel
67	#[arg(long, default_value_t = 4)]
68	backup_parallelism: usize,
69
70	/// Maximum number of backup tasks to queue before blocking
71	#[arg(long, default_value_t = 10000)]
72	backup_queue_size: usize,
73
74	/// Disable colored log output
75	#[arg(long, default_value_t = false, action = clap::ArgAction::Set)]
76	no_color: bool,
77
78	/// Path to a TOML file with one or more Privy credential sets
79	/// When provided, these are used in addition to any PRIVY_* env vars
80	#[arg(long)]
81	privy_config: Option<String>,
82
83	/// Path to a TOML file with IPFS provider configuration
84	/// When provided, this is used instead of IPFS_* env vars
85	#[arg(long)]
86	ipfs_config: Option<String>,
87	}
88
89	#[derive(serde::Deserialize, Clone, Debug)]
90	struct PrivyCredential {
91	app_id: String,
92	verification_key: String,
93	}
94
95	#[derive(serde::Deserialize)]
96	struct PrivyFile {
97	privy: Vec<PrivyCredential>,
98	}
99
100	#[derive(serde::Deserialize)]
101	struct IpfsConfigFile {
102	ipfs_pinning_configs: Vec<nftbk::ipfs::IpfsPinningConfig>,
103	}
104
105	#[tokio::main]
106	async fn main() {	×
107	// We are consuming config both from the environment and from the command line
108	dotenv().ok();	×
109	let args = Args::parse();	×
110	logging::init(args.log_level.clone(), !args.no_color);	×
111	info!(	×
112	"Starting {} {} (commit {})",	×
113	env!("CARGO_BIN_NAME"),
114	env!("CARGO_PKG_VERSION"),
115	env!("GIT_COMMIT")
116	);
117	let auth_token = env::var("NFTBK_AUTH_TOKEN").ok();	×
118
119	// Load Privy credentials from file if provided
120	let mut privy_credentials: Vec<PrivyCredential> = Vec::new();	×
121	if let Some(path) = &args.privy_config {	×
122	match std::fs::read_to_string(path) {	×
123	Ok(contents) => match toml::from_str::<PrivyFile>(&contents) {	×
124	Ok(file) => {	×
125	privy_credentials = file	×
126	.privy	×
127	.into_iter()	×
128	.map(\|mut c\| {	×
129	// Allow \n escaping in inline keys if users choose to
130	c.verification_key = c.verification_key.replace("\\n", "\n");	×
131	c	×
132	})
133	.collect();	×
134	}
135	Err(e) => {	×
136	tracing::error!("Failed to parse Privy config file '{}': {}", path, e);	×
137	}
138	},
139	Err(e) => {	×
140	tracing::error!("Failed to read Privy config file '{}': {}", path, e);	×
141	}
142	}
143	}
144
145	// Load IPFS provider configuration from file if provided
NEW 146	let ipfs_pinning_configs = if args.ipfs_config.is_none() {	×
147	// No config file, use empty list (AppState will fall back to env vars)
148	Vec::new()	×
149	} else {
150	let path = args.ipfs_config.as_ref().unwrap();	×
151	match std::fs::read_to_string(path) {	×
152	Ok(contents) => match toml::from_str::<IpfsConfigFile>(&contents) {	×
153	Ok(file) => {	×
154	info!(	×
NEW 155	"Loaded {} IPFS pinning provider(s) from config file '{}'",	×
NEW 156	file.ipfs_pinning_configs.len(),	×
157	path
158	);
NEW 159	file.ipfs_pinning_configs	×
160	}
161	Err(e) => {	×
162	error!("Failed to parse IPFS config file '{}': {}", path, e);	×
163	std::process::exit(1);	×
164	}
165	},
166	Err(e) => {	×
167	error!("Failed to read IPFS config file '{}': {}", path, e);	×
168	std::process::exit(1);	×
169	}
170	}
171	};
172
173	let (backup_task_sender, backup_task_receiver) =	×
174	mpsc::channel::<BackupTaskOrShutdown>(args.backup_queue_size);	×
175	let db_url =	×
176	std::env::var("DATABASE_URL").expect("DATABASE_URL env var must be set for Postgres");	×
177	let shutdown_flag = Arc::new(AtomicBool::new(false));	×
178	let state = AppState::new(
179	&args.chain_config,	×
180	&args.base_dir,	×
181	args.unsafe_skip_checksum_check,	×
182	auth_token.clone(),	×
183	args.enable_pruner,	×
184	args.pruner_retention_days,	×
185	backup_task_sender.clone(),	×
186	&db_url,	×
187	(args.backup_queue_size + 1) as u32,	×
188	shutdown_flag.clone(),	×
NEW 189	ipfs_pinning_configs,	×
190	)
191	.await;	×
192
193	info!("Starting server with options: {:?}", args);	×
194	info!(	×
195	"Symmetric authentication enabled: {}",	×
196	is_defined(&auth_token)	×
197	);
198	info!(	×
199	"Privy JWT authentication enabled: {} ({} credential set(s))",	×
200	!privy_credentials.is_empty(),	×
201	privy_credentials.len()	×
202	);
203
204	// Spawn worker pool for backup tasks
205	let worker_handles =	×
206	spawn_backup_workers(args.backup_parallelism, backup_task_receiver, state.clone());	×
207
208	// Recover incomplete backup tasks from previous server runs
209	match recover_incomplete_tasks(&state.db, &state.backup_task_sender).await {	×
210	Ok(count) => {	×
211	if count > 0 {	×
212	info!("Successfully recovered {} incomplete backup tasks", count);	×
213	}
214	}
215	Err(e) => {	×
216	error!("Failed to recover incomplete backup tasks: {}", e);	×
217	// Don't exit the server, just log the error and continue
218	}
219	}
220
221	// Start the pruner thread
222	let pruner_handle = if !args.enable_pruner {	×
223	None	×
224	} else {
225	let db = state.db.clone();	×
226	let base_dir = args.base_dir.clone();	×
227	let interval = args.pruner_interval_seconds;	×
228	let shutdown_flag = state.shutdown_flag.clone();	×
229	Some(tokio::spawn(async move {	×
230	run_pruner(db, base_dir, interval, shutdown_flag).await;	×
231	}))
232	};
233
234	// Start the pin monitor thread if IPFS providers are configured
NEW 235	let pin_monitor_handle = if state.ipfs_pinning_instances.is_empty() {	×
236	None	×
237	} else {
238	let db = state.db.clone();	×
NEW 239	let providers = state.ipfs_pinning_instances.clone();	×
240	let interval = args.pin_monitor_interval_seconds;	×
241	let shutdown_flag = state.shutdown_flag.clone();	×
242	info!(	×
243	"Starting pin monitor with {} IPFS provider(s) and {} second interval",	×
244	providers.len(),	×
245	interval
246	);
247	Some(tokio::spawn(async move {	×
248	run_pin_monitor(db, providers.to_vec(), interval, shutdown_flag).await;	×
249	}))
250	};
251
252	// Add graceful shutdown
253	let shutdown_signal = async move {	×
254	let mut sigterm = signal::unix::signal(signal::unix::SignalKind::terminate())	×
255	.expect("failed to install SIGTERM handler");
256
257	tokio::select! {	×
258	_ = signal::ctrl_c() => {	×
259	info!("Received SIGINT (Ctrl+C), shutting down server...");	×
260	}
261	_ = sigterm.recv() => {	×
262	info!("Received SIGTERM, shutting down server...");	×
263	}
264	}
265	shutdown_flag.store(true, Ordering::SeqCst);	×
266	};
267
268	// Start the server
269	let addr: SocketAddr = args.listen_address.parse().expect("Invalid listen address");	×
270	let listener = tokio::net::TcpListener::bind(addr).await.unwrap();	×
271	let app = build_router(
272	state.clone(),	×
273	privy_credentials	×
274	.into_iter()	×
275	.map(\|c\| (c.app_id, c.verification_key))	×
276	.collect(),	×
277	);
278	info!("Listening on {}", addr);	×
279	axum::serve(listener, app)	×
280	.with_graceful_shutdown(shutdown_signal)	×
281	.await	×
282	.unwrap();
283	info!("Server has exited");	×
284
285	if let Some(handle) = pruner_handle {	×
286	let _ = handle.await;	×
287	}
288	info!("Pruner has exited");	×
289
290	if let Some(handle) = pin_monitor_handle {	×
291	let _ = handle.await;	×
292	}
293	info!("Pin monitor has exited");	×
294
295	// On shutdown, send one Shutdown message per worker
296	for _ in 0..args.backup_parallelism {	×
297	let _ = state	×
298	.backup_task_sender	×
299	.send(BackupTaskOrShutdown::Shutdown)	×
300	.await;	×
301	}
302	// Drop the last sender to close the channel and signal workers to exit
303	drop(state.backup_task_sender);	×
304	info!("Backup task sender has exited");	×
305
306	// Wait for all workers to finish
307	for handle in worker_handles {	×
308	let _ = handle.await;	×
309	}
310	info!("Backup workers have exited");	×
311
312	// Give time for final logs to flush
313	let _ = std::io::stdout().flush();	×
314	std::thread::sleep(std::time::Duration::from_millis(200));	×
315	}

0xmichalis / nftbk / 18525274937

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous