18694098093

Committed 21 Oct 2025 06:38PM UTC coverage: 45.23% (+0.6%) from 44.656%

Build # 18694098093

Build Type

push

github

Committed by

0xmichalis

Commit Message

refactor: consolidate config validation under config.rs

Coverage Stats

26 of 42 new or added lines in 3 files covered. (61.9%)

2138 of 4727 relevant lines covered (45.23%)

7.86 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::config::{load_and_validate_config, Config};
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 configuration file
    #[arg(short = 'c', long, default_value = "config.toml")]
    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,
}

#[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, !args.no_color);
    info!(
        "Version: {} {} (commit {})",
        env!("CARGO_BIN_NAME"),
        env!("CARGO_PKG_VERSION"),
        env!("GIT_COMMIT")
    );
    info!("Initializing server with options: {:?}", args);

    // Load unified configuration
    let auth_token = env::var("NFTBK_AUTH_TOKEN").ok();
    info!(
        "Symmetric authentication enabled: {}",
        is_defined(&auth_token)
    );

    let Config {
        chain_config,
        jwt_credentials,
        x402_config,
        ipfs_pinning_configs,
    } = match load_and_validate_config(&args.config) {
        Ok(config) => config,
        Err(e) => {
            error!("Failed to load and validate config: {}", 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(
        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;

    // 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(),
        jwt_credentials
            .into_iter()
            .map(|c| (c.issuer, c.audience, c.verification_key))
            .collect(),
        x402_config.clone(),
    );
    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::config::{load_and_validate_config, Config};
13	use nftbk::envvar::is_defined;
14	use nftbk::logging;
15	use nftbk::logging::LogLevel;
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 configuration file
31	#[arg(short = 'c', long, default_value = "config.toml")]
32	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
79	#[tokio::main]
80	async fn main() {	×
81	// We are consuming config both from the environment and from the command line
82	dotenv().ok();	×
83	let args = Args::parse();	×
84	logging::init(args.log_level, !args.no_color);	×
85	info!(	×
86	"Version: {} {} (commit {})",	×
87	env!("CARGO_BIN_NAME"),
88	env!("CARGO_PKG_VERSION"),
89	env!("GIT_COMMIT")
90	);
91	info!("Initializing server with options: {:?}", args);	×
92
93	// Load unified configuration
94	let auth_token = env::var("NFTBK_AUTH_TOKEN").ok();	×
95	info!(	×
96	"Symmetric authentication enabled: {}",	×
97	is_defined(&auth_token)	×
98	);
99
100	let Config {
NEW 101	chain_config,	×
NEW 102	jwt_credentials,	×
NEW 103	x402_config,	×
NEW 104	ipfs_pinning_configs,	×
NEW 105	} = match load_and_validate_config(&args.config) {	×
NEW 106	Ok(config) => config,	×
107	Err(e) => {	×
NEW 108	error!("Failed to load and validate config: {}", e);	×
109	std::process::exit(1);	×
110	}
111	};
112
113	let (backup_task_sender, backup_task_receiver) =	×
114	mpsc::channel::<BackupTaskOrShutdown>(args.backup_queue_size);	×
115	let db_url =	×
116	std::env::var("DATABASE_URL").expect("DATABASE_URL env var must be set for Postgres");	×
117	let shutdown_flag = Arc::new(AtomicBool::new(false));	×
118	let state = AppState::new(
NEW 119	chain_config,	×
120	&args.base_dir,	×
121	args.unsafe_skip_checksum_check,	×
122	auth_token.clone(),	×
123	args.enable_pruner,	×
124	args.pruner_retention_days,	×
125	backup_task_sender.clone(),	×
126	&db_url,	×
127	(args.backup_queue_size + 1) as u32,	×
128	shutdown_flag.clone(),	×
129	ipfs_pinning_configs,	×
130	)
131	.await;	×
132
133	// Spawn worker pool for backup tasks
134	let worker_handles =	×
135	spawn_backup_workers(args.backup_parallelism, backup_task_receiver, state.clone());	×
136
137	// Recover incomplete backup tasks from previous server runs
138	match recover_incomplete_tasks(&*state.db, &state.backup_task_sender).await {	×
139	Ok(count) => {	×
140	if count > 0 {	×
141	info!("Successfully recovered {} incomplete backup tasks", count);	×
142	}
143	}
144	Err(e) => {	×
145	error!("Failed to recover incomplete backup tasks: {}", e);	×
146	// Don't exit the server, just log the error and continue
147	}
148	}
149
150	// Start the pruner thread
151	let pruner_handle = if !args.enable_pruner {	×
152	None	×
153	} else {
154	let db = state.db.clone();	×
155	let base_dir = args.base_dir.clone();	×
156	let interval = args.pruner_interval_seconds;	×
157	let shutdown_flag = state.shutdown_flag.clone();	×
158	Some(tokio::spawn(async move {	×
159	run_pruner(db, base_dir, interval, shutdown_flag).await;	×
160	}))
161	};
162
163	// Start the pin monitor thread if IPFS providers are configured
164	let pin_monitor_handle = if state.ipfs_pinning_instances.is_empty() {	×
165	None	×
166	} else {
167	let db = state.db.clone();	×
168	let providers = state.ipfs_pinning_instances.clone();	×
169	let interval = args.pin_monitor_interval_seconds;	×
170	let shutdown_flag = state.shutdown_flag.clone();	×
171	info!(	×
172	"Starting pin monitor with {} IPFS provider(s) and {} second interval",	×
173	providers.len(),	×
174	interval
175	);
176	Some(tokio::spawn(async move {	×
177	run_pin_monitor(db, providers.to_vec(), interval, shutdown_flag).await;	×
178	}))
179	};
180
181	// Add graceful shutdown
182	let shutdown_signal = async move {	×
183	let mut sigterm = signal::unix::signal(signal::unix::SignalKind::terminate())	×
184	.expect("failed to install SIGTERM handler");
185
186	tokio::select! {	×
187	_ = signal::ctrl_c() => {	×
188	info!("Received SIGINT (Ctrl+C), shutting down server...");	×
189	}
190	_ = sigterm.recv() => {	×
191	info!("Received SIGTERM, shutting down server...");	×
192	}
193	}
194	shutdown_flag.store(true, Ordering::SeqCst);	×
195	};
196
197	// Start the server
198	let addr: SocketAddr = args.listen_address.parse().expect("Invalid listen address");	×
199	let listener = tokio::net::TcpListener::bind(addr).await.unwrap();	×
200	let app = build_router(
201	state.clone(),	×
202	jwt_credentials	×
203	.into_iter()	×
204	.map(\|c\| (c.issuer, c.audience, c.verification_key))	×
205	.collect(),	×
206	x402_config.clone(),	×
207	);
208	info!("Listening on {}", addr);	×
209	axum::serve(listener, app)	×
210	.with_graceful_shutdown(shutdown_signal)	×
211	.await	×
212	.unwrap();
213	info!("Server has exited");	×
214
215	if let Some(handle) = pruner_handle {	×
216	let _ = handle.await;	×
217	}
218	info!("Pruner has exited");	×
219
220	if let Some(handle) = pin_monitor_handle {	×
221	let _ = handle.await;	×
222	}
223	info!("Pin monitor has exited");	×
224
225	// On shutdown, send one Shutdown message per worker
226	for _ in 0..args.backup_parallelism {	×
227	let _ = state	×
228	.backup_task_sender	×
229	.send(BackupTaskOrShutdown::Shutdown)	×
230	.await;	×
231	}
232	// Drop the last sender to close the channel and signal workers to exit
233	drop(state.backup_task_sender);	×
234	info!("Backup task sender has exited");	×
235
236	// Wait for all workers to finish
237	for handle in worker_handles {	×
238	let _ = handle.await;	×
239	}
240	info!("Backup workers have exited");	×
241
242	// Give time for final logs to flush
243	let _ = std::io::stdout().flush();	×
244	std::thread::sleep(std::time::Duration::from_millis(200));	×
245	}

0xmichalis / nftbk / 18694098093

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