18683462560

Committed 21 Oct 2025 12:15PM UTC coverage: 40.115% (+0.09%) from 40.026%

Build # 18683462560

Build Type

Pull #83

github

Committed by

web-flow

Commit Message

Merge 32c5b2f54 into 106419232

Pull Request Pull Request #83: feat: add x402 support in the CLI

Run Details

45 of 106 new or added lines in 2 files covered. (42.45%)

3 existing lines in 1 file now uncovered.

1891 of 4714 relevant lines covered (40.11%)

7.36 hits per line

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

0.0

/src/cli/commands/server/create.rs

use std::env;
use std::fs::File;
use std::io::Cursor;
use std::path::PathBuf;

use anyhow::{Context, Result};
use flate2::read::GzDecoder;
use reqwest::Client;
use tar::Archive;
use zip::ZipArchive;

use crate::cli::config::load_token_config;
use crate::cli::x402::X402PaymentHandler;
use crate::envvar::is_defined;
use crate::server::api::{self, BackupCreateResponse, BackupRequest, BackupResponse, Tokens};
use crate::server::archive::archive_format_from_user_agent;

const BACKUPS_API_PATH: &str = "/v1/backups";

enum BackupStart {
    Created(BackupCreateResponse),
    Exists(BackupCreateResponse),
    Conflict {
        task_id: String,
        retry_url: String,
        message: String,
    },
}

pub async fn run(
    tokens_config_path: PathBuf,
    server_address: String,
    output_path: Option<PathBuf>,
    force: bool,
    user_agent: String,
    _ipfs_config: Option<String>,
    pin_on_ipfs: bool,
) -> Result<()> {
    let token_config = load_token_config(&tokens_config_path).await?;
    let auth_token = env::var("NFTBK_AUTH_TOKEN").ok();
    let x402_private_key = env::var("NFTBK_X402_PRIVATE_KEY").ok();
    let client = Client::new();

    // First, try to create the backup
    let start = request_backup(
        &token_config,
        &server_address,
        &client,
        auth_token.as_deref(),
        &user_agent,
        pin_on_ipfs,
        x402_private_key.as_deref(),
    )
    .await?;

    let task_id = match start {
        BackupStart::Created(resp) => {
            let task_id = resp.task_id.clone();
            println!("Task ID: {task_id}");
            task_id
        }
        BackupStart::Exists(resp) => {
            let task_id = resp.task_id.clone();
            println!("Task ID (exists): {task_id}");
            if force {
                // Delete and re-request
                let _ =
                    delete_backup(&client, &server_address, &task_id, auth_token.as_deref()).await;
                let second_try = request_backup(
                    &token_config,
                    &server_address,
                    &client,
                    auth_token.as_deref(),
                    &user_agent,
                    pin_on_ipfs,
                    x402_private_key.as_deref(),
                )
                .await?;
                match second_try {
                    BackupStart::Created(resp2) | BackupStart::Exists(resp2) => {
                        let task_id2 = resp2.task_id.clone();
                        println!("Task ID (after delete): {task_id2}");
                        task_id2
                    }
                    BackupStart::Conflict {
                        message,
                        retry_url,
                        task_id: _,
                    } => {
                        anyhow::bail!(
                            "{}\nCould not create new task after delete. Try POST to {}{}",
                            message,
                            server_address.trim_end_matches('/'),
                            retry_url
                        );
                    }
                }
            } else {
                task_id
            }
        }
        BackupStart::Conflict {
            task_id,
            retry_url,
            message,
        } => {
            println!("Task ID: {task_id}");
            if force {
                println!("Server response: {message}");
                // Try to delete the existing task by task_id, then request again
                let _ =
                    delete_backup(&client, &server_address, &task_id, auth_token.as_deref()).await;
                let second_try = request_backup(
                    &token_config,
                    &server_address,
                    &client,
                    auth_token.as_deref(),
                    &user_agent,
                    pin_on_ipfs,
                    x402_private_key.as_deref(),
                )
                .await?;
                match second_try {
                    BackupStart::Created(resp2) | BackupStart::Exists(resp2) => {
                        let task_id2 = resp2.task_id.clone();
                        println!("Task ID (after delete): {task_id2}");
                        task_id2
                    }
                    BackupStart::Conflict {
                        message,
                        retry_url,
                        task_id: _,
                    } => {
                        anyhow::bail!(
                            "{}\nCould not create new task after delete. Try POST to {}{}",
                            message,
                            server_address.trim_end_matches('/'),
                            retry_url
                        );
                    }
                }
            } else {
                anyhow::bail!(
                    "{}\nRun the CLI with --force true, or POST to {}{}",
                    message,
                    server_address.trim_end_matches('/'),
                    retry_url
                );
            }
        }
    };

    wait_for_done_backup(&client, &server_address, &task_id, auth_token.as_deref()).await?;

    return download_backup(
        &client,
        &server_address,
        &task_id,
        output_path.as_ref(),
        auth_token.as_deref(),
        &archive_format_from_user_agent(&user_agent),
    )
    .await;
}

async fn delete_backup(
    client: &Client,
    server_address: &str,
    task_id: &str,
    auth_token: Option<&str>,
) -> Result<()> {
    let server = server_address.trim_end_matches('/');
    let url = format!("{server}{BACKUPS_API_PATH}/{task_id}",);
    let mut req = client.delete(url);
    if is_defined(&auth_token.as_ref().map(|s| s.to_string())) {
        req = req.header("Authorization", format!("Bearer {}", auth_token.unwrap()));
    }
    let resp = req
        .send()
        .await
        .context("Failed to send DELETE to server")?;

    match resp.status().as_u16() {
        202 => {
            println!("Deletion request sent for backup {task_id}, waiting for completion...");
            // Poll until backup is actually deleted (returns 404)
            loop {
                tokio::time::sleep(std::time::Duration::from_secs(2)).await;
                let status_url = format!("{server}{BACKUPS_API_PATH}/{task_id}");
                let mut status_req = client.get(&status_url);
                if is_defined(&auth_token.as_ref().map(|s| s.to_string())) {
                    status_req = status_req
                        .header("Authorization", format!("Bearer {}", auth_token.unwrap()));
                }
                match status_req.send().await {
                    Ok(status_resp) => {
                        if status_resp.status().as_u16() == 404 {
                            println!("Backup {task_id} successfully deleted");
                            return Ok(());
                        }
                        // Still exists, continue polling
                    }
                    Err(_) => {
                        // Network error, continue polling
                    }
                }
            }
        }
        404 => {
            println!("Backup {task_id} already deleted");
            Ok(())
        }
        409 => {
            // In progress; proceed with new request anyway
            println!("Existing backup {task_id} is in progress; proceeding to create new request");
            Ok(())
        }
        code => {
            let text = resp.text().await.unwrap_or_default();
            println!("Warning: failed to delete existing backup ({code}): {text}");
            Ok(())
        }
    }
}

async fn request_backup(
    token_config: &crate::backup::TokenConfig,
    server_address: &str,
    client: &Client,
    auth_token: Option<&str>,
    user_agent: &str,
    pin_on_ipfs: bool,
    x402_private_key: Option<&str>,
) -> Result<BackupStart> {
    let mut backup_req = BackupRequest {
        tokens: Vec::new(),
        pin_on_ipfs,
        create_archive: true,
    };
    for (chain, tokens) in &token_config.chains {
        backup_req.tokens.push(Tokens {
            chain: chain.clone(),
            tokens: tokens.clone(),
        });
    }

    let server = server_address.trim_end_matches('/');
    println!("Submitting backup request to server at {server}{BACKUPS_API_PATH} ...");
    let mut req_builder = client
        .post(format!("{server}{BACKUPS_API_PATH}"))
        .json(&backup_req);
    req_builder = req_builder.header("User-Agent", user_agent);
    if is_defined(&auth_token.as_ref().map(|s| s.to_string())) {
        req_builder =
            req_builder.header("Authorization", format!("Bearer {}", auth_token.unwrap()));
    }
    let req = req_builder.build().context("Failed to build request")?;
    let resp = client
        .execute(req.try_clone().context("Failed to clone request")?)
        .await
        .context("Failed to send backup request to server")?;
    let status = resp.status();

    // Handle 402 Payment Required response
    if status.as_u16() == 402 {
        return handle_402_response(client, resp, &req, x402_private_key).await;
    }

    if status.is_success() {
        let backup_resp: BackupCreateResponse =
            resp.json().await.context("Invalid server response")?;
        if status.as_u16() == 201 {
            return Ok(BackupStart::Created(backup_resp));
        } else {
            return Ok(BackupStart::Exists(backup_resp));
        }
    }
    if status.as_u16() == 409 {
        let body: serde_json::Value = resp
            .json()
            .await
            .context("Invalid conflict response from server")?;
        let task_id = body
            .get("task_id")
            .and_then(|v| v.as_str())
            .unwrap_or_default()
            .to_string();
        let retry_url = body
            .get("retry_url")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string();
        let message = body
            .get("error")
            .and_then(|v| v.as_str())
            .unwrap_or(&format!("Server returned conflict for {BACKUPS_API_PATH}"))
            .to_string();
        return Ok(BackupStart::Conflict {
            task_id,
            retry_url,
            message,
        });
    }
    let text = resp.text().await.unwrap_or_default();
    anyhow::bail!("Server error: {}", text);
}

/// Handle 402 Payment Required response by creating a payment and retrying the request
async fn handle_402_response(
    client: &Client,
    response: reqwest::Response,
    original_request: &reqwest::Request,
    x402_private_key: Option<&str>,
) -> Result<BackupStart> {
    // Parse PaymentRequirements from the response
    let payment_requirements: x402_rs::types::PaymentRequirements = response
        .json()
        .await
        .context("Failed to parse PaymentRequirements from 402 response")?;

    // Create x402 payment handler
    let payment_handler = X402PaymentHandler::new(x402_private_key)
        .context("Failed to create x402 payment handler")?;

    // Handle the payment and retry the request
    let response = payment_handler
        .handle_402_response(client, original_request, payment_requirements)
        .await
        .context("Failed to handle x402 payment")?;

    // Process the retry response
    let status = response.status();
    if status.is_success() {
        let backup_resp: BackupCreateResponse = response
            .json()
            .await
            .context("Invalid server response after payment")?;
        if status.as_u16() == 201 {
            return Ok(BackupStart::Created(backup_resp));
        } else {
            return Ok(BackupStart::Exists(backup_resp));
        }
    }

    let text = response.text().await.unwrap_or_default();
    anyhow::bail!("Server error after payment: {}", text);
}

async fn wait_for_done_backup(
    client: &reqwest::Client,
    server_address: &str,
    task_id: &str,
    auth_token: Option<&str>,
) -> Result<()> {
    let status_url = format!(
        "{}{}/{}",
        server_address.trim_end_matches('/'),
        BACKUPS_API_PATH,
        task_id
    );
    let mut in_progress_logged = false;
    loop {
        let mut req = client.get(&status_url);
        if is_defined(&auth_token.as_ref().map(|s| s.to_string())) {
            req = req.header("Authorization", format!("Bearer {}", auth_token.unwrap()));
        }
        let resp = req.send().await;
        match resp {
            Ok(r) => {
                if r.status().is_success() {
                    let status: BackupResponse = r.json().await.unwrap_or({
                        // Fallback shape with nulls
                        BackupResponse {
                            task_id: task_id.to_string(),
                            created_at: String::new(),
                            storage_mode: String::new(),
                            tokens: Vec::new(),
                            total_tokens: 0,
                            page: 1,
                            limit: 50,
                            archive: Some(api::Archive {
                                status: api::SubresourceStatus {
                                    status: None,
                                    fatal_error: None,
                                    error_log: None,
                                    deleted_at: None,
                                },
                                format: None,
                                expires_at: None,
                            }),
                            pins: Some(api::Pins {
                                status: api::SubresourceStatus {
                                    status: None,
                                    fatal_error: None,
                                    error_log: None,
                                    deleted_at: None,
                                },
                            }),
                        }
                    });
                    // Aggregate a coarse "overall" view for UX: in_progress if any subresource is in_progress
                    let archive_status = status
                        .archive
                        .as_ref()
                        .and_then(|a| a.status.status.as_deref());
                    let ipfs_status = status
                        .pins
                        .as_ref()
                        .and_then(|p| p.status.status.as_deref());
                    let any_in_progress = matches!(archive_status, Some("in_progress"))
                        || matches!(ipfs_status, Some("in_progress"));
                    let any_error = matches!(archive_status, Some("error"))
                        || matches!(ipfs_status, Some("error"))
                        || status
                            .archive
                            .as_ref()
                            .and_then(|a| a.status.fatal_error.as_ref())
                            .is_some()
                        || status
                            .pins
                            .as_ref()
                            .and_then(|p| p.status.fatal_error.as_ref())
                            .is_some();
                    let all_done = matches!(archive_status, Some("done"))
                        && (ipfs_status.is_none() || matches!(ipfs_status, Some("done")));
                    if any_in_progress {
                        if !in_progress_logged {
                            println!("Waiting for backup to complete...");
                            in_progress_logged = true;
                        }
                    } else if all_done {
                        println!("Backup complete.");
                        if let Some(ref a) = status
                            .archive
                            .as_ref()
                            .and_then(|a| a.status.error_log.as_ref())
                        {
                            if !a.is_empty() {
                                tracing::warn!("{}", a);
                            }
                        }
                        if let Some(ref i) = status
                            .pins
                            .as_ref()
                            .and_then(|p| p.status.error_log.as_ref())
                        {
                            if !i.is_empty() {
                                tracing::warn!("{}", i);
                            }
                        }
                        break;
                    } else if any_error {
                        let msg = status
                            .archive
                            .as_ref()
                            .and_then(|a| a.status.fatal_error.clone())
                            .or(status
                                .pins
                                .as_ref()
                                .and_then(|p| p.status.fatal_error.clone()))
                            .unwrap_or_else(|| "Unknown error".to_string());
                        anyhow::bail!("Server error: {}", msg);
                    } else {
                        println!(
                            "Unknown status: archive={:?} ipfs={:?}",
                            status
                                .archive
                                .as_ref()
                                .and_then(|a| a.status.status.as_deref()),
                            status
                                .pins
                                .as_ref()
                                .and_then(|p| p.status.status.as_deref())
                        );
                    }
                } else {
                    println!("Failed to get status: {}", r.status());
                }
            }
            Err(e) => {
                println!("Error polling status: {e}");
            }
        }
        tokio::time::sleep(std::time::Duration::from_secs(10)).await;
    }
    Ok(())
}

async fn download_backup(
    client: &Client,
    server_address: &str,
    task_id: &str,
    output_path: Option<&PathBuf>,
    _auth_token: Option<&str>,
    archive_format: &str,
) -> Result<()> {
    // Step 1: Get download token
    let token_url = format!(
        "{}{}/{}/download-tokens",
        server_address.trim_end_matches('/'),
        BACKUPS_API_PATH,
        task_id
    );
    let mut token_req = client.post(&token_url);
    if is_defined(&_auth_token.as_ref().map(|s| s.to_string())) {
        token_req = token_req.header("Authorization", format!("Bearer {}", _auth_token.unwrap()));
    }
    let token_resp = token_req
        .send()
        .await
        .context("Failed to get download token")?;
    if !token_resp.status().is_success() {
        anyhow::bail!("Failed to get download token: {}", token_resp.status());
    }
    let token_json: serde_json::Value =
        token_resp.json().await.context("Invalid token response")?;
    let download_token = token_json
        .get("token")
        .and_then(|v| v.as_str())
        .ok_or_else(|| anyhow::anyhow!("No token in response"))?;

    // Step 2: Download using token
    let download_url = format!(
        "{}{}/{}/download?token={}",
        server_address.trim_end_matches('/'),
        BACKUPS_API_PATH,
        task_id,
        urlencoding::encode(download_token)
    );
    println!("Downloading archive ...");
    let resp = client
        .get(&download_url)
        .send()
        .await
        .context("Failed to download archive")?;
    if !resp.status().is_success() {
        anyhow::bail!("Failed to download archive: {}", resp.status());
    }
    let bytes = resp.bytes().await.context("Failed to read archive bytes")?;
    let output_path = output_path.cloned().unwrap_or_else(|| PathBuf::from("."));

    println!("Extracting backup to {}...", output_path.display());
    match archive_format {
        "tar.gz" => {
            let gz = GzDecoder::new(Cursor::new(bytes));
            let mut archive = Archive::new(gz);
            archive
                .unpack(&output_path)
                .context("Failed to extract backup archive (tar.gz)")?;
        }
        "zip" => {
            let mut zip =
                ZipArchive::new(Cursor::new(bytes)).context("Failed to read zip archive")?;
            for i in 0..zip.len() {
                let mut file = zip.by_index(i).context("Failed to access file in zip")?;
                let outpath = match file.enclosed_name() {
                    Some(path) => output_path.join(path),
                    None => continue,
                };
                if file.name().ends_with('/') {
                    std::fs::create_dir_all(&outpath)
                        .context("Failed to create directory from zip")?;
                } else {
                    if let Some(p) = outpath.parent() {
                        std::fs::create_dir_all(p)
                            .context("Failed to create parent directory for zip file")?;
                    }
                    let mut outfile =
                        File::create(&outpath).context("Failed to create file from zip")?;
                    std::io::copy(&mut file, &mut outfile)
                        .context("Failed to extract file from zip")?;
                }
            }
        }
        _ => anyhow::bail!("Unknown archive format: {archive_format}"),
    }
    println!("Backup extracted to {}", output_path.display());
    Ok(())
}

1	use std::env;
2	use std::fs::File;
3	use std::io::Cursor;
4	use std::path::PathBuf;
5
6	use anyhow::{Context, Result};
7	use flate2::read::GzDecoder;
8	use reqwest::Client;
9	use tar::Archive;
10	use zip::ZipArchive;
11
12	use crate::cli::config::load_token_config;
13	use crate::cli::x402::X402PaymentHandler;
14	use crate::envvar::is_defined;
15	use crate::server::api::{self, BackupCreateResponse, BackupRequest, BackupResponse, Tokens};
16	use crate::server::archive::archive_format_from_user_agent;
17
18	const BACKUPS_API_PATH: &str = "/v1/backups";
19
20	enum BackupStart {
21	Created(BackupCreateResponse),
22	Exists(BackupCreateResponse),
23	Conflict {
24	task_id: String,
25	retry_url: String,
26	message: String,
27	},
28	}
29
30	pub async fn run(	×
31	tokens_config_path: PathBuf,
32	server_address: String,
33	output_path: Option<PathBuf>,
34	force: bool,
35	user_agent: String,
36	_ipfs_config: Option<String>,
37	pin_on_ipfs: bool,
38	) -> Result<()> {
39	let token_config = load_token_config(&tokens_config_path).await?;	×
40	let auth_token = env::var("NFTBK_AUTH_TOKEN").ok();	×
NEW 41	let x402_private_key = env::var("NFTBK_X402_PRIVATE_KEY").ok();	×
UNCOV 42	let client = Client::new();	×
43
44	// First, try to create the backup
45	let start = request_backup(
46	&token_config,	×
47	&server_address,	×
48	&client,	×
49	auth_token.as_deref(),	×
50	&user_agent,	×
51	pin_on_ipfs,	×
NEW 52	x402_private_key.as_deref(),	×
53	)
54	.await?;	×
55
56	let task_id = match start {	×
57	BackupStart::Created(resp) => {	×
58	let task_id = resp.task_id.clone();	×
59	println!("Task ID: {task_id}");	×
60	task_id	×
61	}
62	BackupStart::Exists(resp) => {	×
63	let task_id = resp.task_id.clone();	×
64	println!("Task ID (exists): {task_id}");	×
65	if force {	×
66	// Delete and re-request
67	let _ =
68	delete_backup(&client, &server_address, &task_id, auth_token.as_deref()).await;	×
69	let second_try = request_backup(
70	&token_config,	×
71	&server_address,	×
72	&client,	×
73	auth_token.as_deref(),	×
74	&user_agent,	×
75	pin_on_ipfs,	×
NEW 76	x402_private_key.as_deref(),	×
77	)
78	.await?;	×
79	match second_try {	×
80	BackupStart::Created(resp2) \| BackupStart::Exists(resp2) => {	×
81	let task_id2 = resp2.task_id.clone();	×
82	println!("Task ID (after delete): {task_id2}");	×
83	task_id2	×
84	}
85	BackupStart::Conflict {
86	message,	×
87	retry_url,	×
88	task_id: _,
89	} => {
90	anyhow::bail!(	×
91	"{}\nCould not create new task after delete. Try POST to {}{}",	×
92	message,
93	server_address.trim_end_matches('/'),	×
94	retry_url
95	);
96	}
97	}
98	} else {
99	task_id	×
100	}
101	}
102	BackupStart::Conflict {
103	task_id,	×
104	retry_url,	×
105	message,	×
106	} => {
107	println!("Task ID: {task_id}");	×
108	if force {	×
109	println!("Server response: {message}");	×
110	// Try to delete the existing task by task_id, then request again
111	let _ =
112	delete_backup(&client, &server_address, &task_id, auth_token.as_deref()).await;	×
113	let second_try = request_backup(
114	&token_config,	×
115	&server_address,	×
116	&client,	×
117	auth_token.as_deref(),	×
118	&user_agent,	×
119	pin_on_ipfs,	×
NEW 120	x402_private_key.as_deref(),	×
121	)
122	.await?;	×
123	match second_try {	×
124	BackupStart::Created(resp2) \| BackupStart::Exists(resp2) => {	×
125	let task_id2 = resp2.task_id.clone();	×
126	println!("Task ID (after delete): {task_id2}");	×
127	task_id2	×
128	}
129	BackupStart::Conflict {
130	message,	×
131	retry_url,	×
132	task_id: _,
133	} => {
134	anyhow::bail!(	×
135	"{}\nCould not create new task after delete. Try POST to {}{}",	×
136	message,
137	server_address.trim_end_matches('/'),	×
138	retry_url
139	);
140	}
141	}
142	} else {
143	anyhow::bail!(	×
144	"{}\nRun the CLI with --force true, or POST to {}{}",	×
145	message,
146	server_address.trim_end_matches('/'),	×
147	retry_url
148	);
149	}
150	}
151	};
152
153	wait_for_done_backup(&client, &server_address, &task_id, auth_token.as_deref()).await?;	×
154
155	return download_backup(	×
156	&client,	×
157	&server_address,	×
158	&task_id,	×
159	output_path.as_ref(),	×
160	auth_token.as_deref(),	×
161	&archive_format_from_user_agent(&user_agent),	×
162	)
163	.await;	×
164	}
165
166	async fn delete_backup(	×
167	client: &Client,
168	server_address: &str,
169	task_id: &str,
170	auth_token: Option<&str>,
171	) -> Result<()> {
172	let server = server_address.trim_end_matches('/');	×
173	let url = format!("{server}{BACKUPS_API_PATH}/{task_id}",);	×
174	let mut req = client.delete(url);	×
175	if is_defined(&auth_token.as_ref().map(\|s\| s.to_string())) {	×
176	req = req.header("Authorization", format!("Bearer {}", auth_token.unwrap()));	×
177	}
178	let resp = req	×
179	.send()
180	.await	×
181	.context("Failed to send DELETE to server")?;
182
183	match resp.status().as_u16() {	×
184	202 => {
185	println!("Deletion request sent for backup {task_id}, waiting for completion...");	×
186	// Poll until backup is actually deleted (returns 404)
187	loop {
188	tokio::time::sleep(std::time::Duration::from_secs(2)).await;	×
189	let status_url = format!("{server}{BACKUPS_API_PATH}/{task_id}");	×
190	let mut status_req = client.get(&status_url);	×
191	if is_defined(&auth_token.as_ref().map(\|s\| s.to_string())) {	×
192	status_req = status_req	×
193	.header("Authorization", format!("Bearer {}", auth_token.unwrap()));	×
194	}
195	match status_req.send().await {	×
196	Ok(status_resp) => {	×
197	if status_resp.status().as_u16() == 404 {	×
198	println!("Backup {task_id} successfully deleted");	×
199	return Ok(());	×
200	}
201	// Still exists, continue polling
202	}
203	Err(_) => {	×
204	// Network error, continue polling
205	}
206	}
207	}
208	}
209	404 => {
210	println!("Backup {task_id} already deleted");	×
211	Ok(())	×
212	}
213	409 => {
214	// In progress; proceed with new request anyway
215	println!("Existing backup {task_id} is in progress; proceeding to create new request");	×
216	Ok(())	×
217	}
218	code => {	×
219	let text = resp.text().await.unwrap_or_default();	×
220	println!("Warning: failed to delete existing backup ({code}): {text}");	×
221	Ok(())	×
222	}
223	}
224	}
225
226	async fn request_backup(	×
227	token_config: &crate::backup::TokenConfig,
228	server_address: &str,
229	client: &Client,
230	auth_token: Option<&str>,
231	user_agent: &str,
232	pin_on_ipfs: bool,
233	x402_private_key: Option<&str>,
234	) -> Result<BackupStart> {
235	let mut backup_req = BackupRequest {
236	tokens: Vec::new(),	×
237	pin_on_ipfs,
238	create_archive: true,
239	};
240	for (chain, tokens) in &token_config.chains {	×
241	backup_req.tokens.push(Tokens {	×
242	chain: chain.clone(),	×
243	tokens: tokens.clone(),	×
244	});
245	}
246
247	let server = server_address.trim_end_matches('/');	×
248	println!("Submitting backup request to server at {server}{BACKUPS_API_PATH} ...");	×
NEW 249	let mut req_builder = client	×
250	.post(format!("{server}{BACKUPS_API_PATH}"))	×
251	.json(&backup_req);	×
NEW 252	req_builder = req_builder.header("User-Agent", user_agent);	×
253	if is_defined(&auth_token.as_ref().map(\|s\| s.to_string())) {	×
NEW 254	req_builder =	×
NEW 255	req_builder.header("Authorization", format!("Bearer {}", auth_token.unwrap()));	×
256	}
NEW 257	let req = req_builder.build().context("Failed to build request")?;	×
NEW 258	let resp = client	×
NEW 259	.execute(req.try_clone().context("Failed to clone request")?)	×
UNCOV 260	.await	×
261	.context("Failed to send backup request to server")?;
262	let status = resp.status();	×
263
264	// Handle 402 Payment Required response
NEW 265	if status.as_u16() == 402 {	×
NEW 266	return handle_402_response(client, resp, &req, x402_private_key).await;	×
267	}
268
269	if status.is_success() {	×
270	let backup_resp: BackupCreateResponse =	×
271	resp.json().await.context("Invalid server response")?;	×
272	if status.as_u16() == 201 {	×
273	return Ok(BackupStart::Created(backup_resp));	×
274	} else {
275	return Ok(BackupStart::Exists(backup_resp));	×
276	}
277	}
278	if status.as_u16() == 409 {	×
279	let body: serde_json::Value = resp	×
280	.json()
281	.await	×
282	.context("Invalid conflict response from server")?;
283	let task_id = body	×
284	.get("task_id")
285	.and_then(\|v\| v.as_str())	×
286	.unwrap_or_default()
287	.to_string();
288	let retry_url = body	×
289	.get("retry_url")
290	.and_then(\|v\| v.as_str())	×
291	.unwrap_or("")
292	.to_string();
293	let message = body	×
294	.get("error")
295	.and_then(\|v\| v.as_str())	×
296	.unwrap_or(&format!("Server returned conflict for {BACKUPS_API_PATH}"))	×
297	.to_string();
298	return Ok(BackupStart::Conflict {	×
299	task_id,	×
300	retry_url,	×
301	message,	×
302	});
303	}
304	let text = resp.text().await.unwrap_or_default();	×
305	anyhow::bail!("Server error: {}", text);	×
306	}
307
308	/// Handle 402 Payment Required response by creating a payment and retrying the request
NEW 309	async fn handle_402_response(	×
310	client: &Client,
311	response: reqwest::Response,
312	original_request: &reqwest::Request,
313	x402_private_key: Option<&str>,
314	) -> Result<BackupStart> {
315	// Parse PaymentRequirements from the response
NEW 316	let payment_requirements: x402_rs::types::PaymentRequirements = response	×
317	.json()
NEW 318	.await	×
319	.context("Failed to parse PaymentRequirements from 402 response")?;
320
321	// Create x402 payment handler
NEW 322	let payment_handler = X402PaymentHandler::new(x402_private_key)	×
323	.context("Failed to create x402 payment handler")?;
324
325	// Handle the payment and retry the request
NEW 326	let response = payment_handler	×
NEW 327	.handle_402_response(client, original_request, payment_requirements)	×
NEW 328	.await	×
329	.context("Failed to handle x402 payment")?;
330
331	// Process the retry response
NEW 332	let status = response.status();	×
NEW 333	if status.is_success() {	×
NEW 334	let backup_resp: BackupCreateResponse = response	×
335	.json()
NEW 336	.await	×
337	.context("Invalid server response after payment")?;
NEW 338	if status.as_u16() == 201 {	×
NEW 339	return Ok(BackupStart::Created(backup_resp));	×
340	} else {
NEW 341	return Ok(BackupStart::Exists(backup_resp));	×
342	}
343	}
344
NEW 345	let text = response.text().await.unwrap_or_default();	×
NEW 346	anyhow::bail!("Server error after payment: {}", text);	×
347	}
348
UNCOV 349	async fn wait_for_done_backup(	×
350	client: &reqwest::Client,
351	server_address: &str,
352	task_id: &str,
353	auth_token: Option<&str>,
354	) -> Result<()> {
355	let status_url = format!(	×
356	"{}{}/{}",
357	server_address.trim_end_matches('/'),	×
358	BACKUPS_API_PATH,
359	task_id
360	);
361	let mut in_progress_logged = false;	×
362	loop {
363	let mut req = client.get(&status_url);	×
364	if is_defined(&auth_token.as_ref().map(\|s\| s.to_string())) {	×
365	req = req.header("Authorization", format!("Bearer {}", auth_token.unwrap()));	×
366	}
367	let resp = req.send().await;	×
368	match resp {	×
369	Ok(r) => {	×
370	if r.status().is_success() {	×
371	let status: BackupResponse = r.json().await.unwrap_or({	×
372	// Fallback shape with nulls
373	BackupResponse {	×
374	task_id: task_id.to_string(),	×
375	created_at: String::new(),	×
376	storage_mode: String::new(),	×
377	tokens: Vec::new(),	×
378	total_tokens: 0,	×
379	page: 1,	×
380	limit: 50,	×
381	archive: Some(api::Archive {	×
382	status: api::SubresourceStatus {	×
383	status: None,	×
384	fatal_error: None,	×
385	error_log: None,	×
386	deleted_at: None,	×
387	},
388	format: None,	×
389	expires_at: None,	×
390	}),
391	pins: Some(api::Pins {	×
392	status: api::SubresourceStatus {	×
393	status: None,	×
394	fatal_error: None,	×
395	error_log: None,	×
396	deleted_at: None,	×
397	},
398	}),
399	}
400	});
401	// Aggregate a coarse "overall" view for UX: in_progress if any subresource is in_progress
402	let archive_status = status	×
403	.archive	×
404	.as_ref()
405	.and_then(\|a\| a.status.status.as_deref());	×
406	let ipfs_status = status	×
407	.pins	×
408	.as_ref()
409	.and_then(\|p\| p.status.status.as_deref());	×
410	let any_in_progress = matches!(archive_status, Some("in_progress"))	×
411	\|\| matches!(ipfs_status, Some("in_progress"));	×
412	let any_error = matches!(archive_status, Some("error"))	×
413	\|\| matches!(ipfs_status, Some("error"))	×
414	\|\| status	×
415	.archive	×
416	.as_ref()	×
417	.and_then(\|a\| a.status.fatal_error.as_ref())	×
418	.is_some()	×
419	\|\| status	×
420	.pins	×
421	.as_ref()	×
422	.and_then(\|p\| p.status.fatal_error.as_ref())	×
423	.is_some();	×
424	let all_done = matches!(archive_status, Some("done"))	×
425	&& (ipfs_status.is_none() \|\| matches!(ipfs_status, Some("done")));	×
426	if any_in_progress {	×
427	if !in_progress_logged {	×
428	println!("Waiting for backup to complete...");	×
429	in_progress_logged = true;	×
430	}
431	} else if all_done {	×
432	println!("Backup complete.");	×
433	if let Some(ref a) = status	×
434	.archive	×
435	.as_ref()
436	.and_then(\|a\| a.status.error_log.as_ref())	×
437	{
438	if !a.is_empty() {	×
439	tracing::warn!("{}", a);	×
440	}
441	}
442	if let Some(ref i) = status	×
443	.pins	×
444	.as_ref()
445	.and_then(\|p\| p.status.error_log.as_ref())	×
446	{
447	if !i.is_empty() {	×
448	tracing::warn!("{}", i);	×
449	}
450	}
451	break;	×
452	} else if any_error {	×
453	let msg = status	×
454	.archive	×
455	.as_ref()
456	.and_then(\|a\| a.status.fatal_error.clone())	×
457	.or(status	×
458	.pins	×
459	.as_ref()	×
460	.and_then(\|p\| p.status.fatal_error.clone()))	×
461	.unwrap_or_else(\|\| "Unknown error".to_string());	×
462	anyhow::bail!("Server error: {}", msg);	×
463	} else {
464	println!(	×
465	"Unknown status: archive={:?} ipfs={:?}",	×
466	status	×
467	.archive	×
468	.as_ref()	×
469	.and_then(\|a\| a.status.status.as_deref()),	×
470	status	×
471	.pins	×
472	.as_ref()	×
473	.and_then(\|p\| p.status.status.as_deref())	×
474	);
475	}
476	} else {
477	println!("Failed to get status: {}", r.status());	×
478	}
479	}
480	Err(e) => {	×
481	println!("Error polling status: {e}");	×
482	}
483	}
484	tokio::time::sleep(std::time::Duration::from_secs(10)).await;	×
485	}
486	Ok(())	×
487	}
488
489	async fn download_backup(	×
490	client: &Client,
491	server_address: &str,
492	task_id: &str,
493	output_path: Option<&PathBuf>,
494	_auth_token: Option<&str>,
495	archive_format: &str,
496	) -> Result<()> {
497	// Step 1: Get download token
498	let token_url = format!(	×
499	"{}{}/{}/download-tokens",
500	server_address.trim_end_matches('/'),	×
501	BACKUPS_API_PATH,
502	task_id
503	);
504	let mut token_req = client.post(&token_url);	×
505	if is_defined(&_auth_token.as_ref().map(\|s\| s.to_string())) {	×
506	token_req = token_req.header("Authorization", format!("Bearer {}", _auth_token.unwrap()));	×
507	}
508	let token_resp = token_req	×
509	.send()
510	.await	×
511	.context("Failed to get download token")?;
512	if !token_resp.status().is_success() {	×
513	anyhow::bail!("Failed to get download token: {}", token_resp.status());	×
514	}
515	let token_json: serde_json::Value =	×
516	token_resp.json().await.context("Invalid token response")?;	×
517	let download_token = token_json	×
518	.get("token")
519	.and_then(\|v\| v.as_str())	×
520	.ok_or_else(\|\| anyhow::anyhow!("No token in response"))?;	×
521
522	// Step 2: Download using token
523	let download_url = format!(	×
524	"{}{}/{}/download?token={}",
525	server_address.trim_end_matches('/'),	×
526	BACKUPS_API_PATH,
527	task_id,
528	urlencoding::encode(download_token)	×
529	);
530	println!("Downloading archive ...");	×
531	let resp = client	×
532	.get(&download_url)	×
533	.send()
534	.await	×
535	.context("Failed to download archive")?;
536	if !resp.status().is_success() {	×
537	anyhow::bail!("Failed to download archive: {}", resp.status());	×
538	}
539	let bytes = resp.bytes().await.context("Failed to read archive bytes")?;	×
540	let output_path = output_path.cloned().unwrap_or_else(\|\| PathBuf::from("."));	×
541
542	println!("Extracting backup to {}...", output_path.display());	×
543	match archive_format {	×
544	"tar.gz" => {	×
545	let gz = GzDecoder::new(Cursor::new(bytes));	×
546	let mut archive = Archive::new(gz);	×
547	archive	×
548	.unpack(&output_path)	×
549	.context("Failed to extract backup archive (tar.gz)")?;
550	}
551	"zip" => {	×
552	let mut zip =	×
553	ZipArchive::new(Cursor::new(bytes)).context("Failed to read zip archive")?;	×
554	for i in 0..zip.len() {	×
555	let mut file = zip.by_index(i).context("Failed to access file in zip")?;	×
556	let outpath = match file.enclosed_name() {	×
557	Some(path) => output_path.join(path),	×
558	None => continue,	×
559	};
560	if file.name().ends_with('/') {	×
561	std::fs::create_dir_all(&outpath)	×
562	.context("Failed to create directory from zip")?;
563	} else {
564	if let Some(p) = outpath.parent() {	×
565	std::fs::create_dir_all(p)	×
566	.context("Failed to create parent directory for zip file")?;
567	}
568	let mut outfile =	×
569	File::create(&outpath).context("Failed to create file from zip")?;	×
570	std::io::copy(&mut file, &mut outfile)	×
571	.context("Failed to extract file from zip")?;
572	}
573	}
574	}
575	_ => anyhow::bail!("Unknown archive format: {archive_format}"),	×
576	}
577	println!("Backup extracted to {}", output_path.display());	×
578	Ok(())	×
579	}

0xmichalis / nftbk / 18683462560

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