Contractless/src/torrent/torrenting_system/torrent_requests.rs

use crate::io;
use crate::log::warn;
use crate::records::memory::response_channels::{Byte3, Command};
use crate::records::memory::torrent_status::{set_torrent_status, TorrentStatus};
use crate::rpc::command_maps::RPC_TORRENT_BY_HEIGHT;
use crate::rpc::responses::RpcResponse;
use crate::sled::Db;
use crate::torrent::create_metadata::broadcast_new_torrent_to_peers;
use crate::torrent::structs::Torrent;
use crate::torrent::torrenting_system::save_torrent::save_staged_torrent;
use crate::torrent::torrenting_system::setup_block_download::setup_download;
use crate::verifications::verification_service::global_verification_service;
use crate::wallets::structures::Wallet;
use crate::Arc;
use crate::Mutex;
use crate::TcpStream;

pub async fn send_request_torrent_message(
    stream: Arc<Mutex<TcpStream>>,
    local_height: u32,
    hashmap_key: Byte3,
    connections_key: String,
) -> io::Result<()> {
    // Ask the remote node for the torrent metadata for the requested
    // block height using the shared response hashmap key.
    let request_torrent: u8 = RPC_TORRENT_BY_HEIGHT;
    let request_torrent_binary = request_torrent.to_le_bytes();
    let get_height_binary = local_height.to_le_bytes();
    let mut message = Vec::new();
    message.extend(request_torrent_binary);
    message.extend(hashmap_key);
    message.extend(get_height_binary);
    RpcResponse::send_raw(&stream, Some(&connections_key), &message).await;
    Ok(())
}

pub async fn handle_response_and_save_torrent(
    height: u32,
    db: &Db,
    torrent: Torrent,
    wallet: Arc<Wallet>,
    map: Arc<Mutex<Command>>,
    allow_during_reorg: bool,
    rebroadcast: bool,
) -> Result<(), String> {
    let Some((torrent, staged_path)) =
        stage_and_verify_torrent(height, db, torrent, wallet, true).await?
    else {
        return Ok(());
    };
    let torrent_bytes = torrent.clone().to_bytes().await;

    setup_download_for_torrent(
        height,
        torrent,
        staged_path,
        allow_during_reorg,
        db.clone(),
        map.clone(),
    )
    .await?;

    if rebroadcast {
        // A requested torrent is only forwarded after this node has the
        // complete validated block available for piece requests.
        broadcast_new_torrent_to_peers(height, &torrent_bytes, map).await;
    }
    Ok(())
}

#[derive(Clone)]
pub struct ProcessTorrentResponse {
    pub height: u32,
    pub db: Db,
    pub torrent: Torrent,
    pub wallet: Arc<Wallet>,
    pub map: Arc<Mutex<Command>>,
    pub allow_during_reorg: bool,
    pub process_now: bool,
}

pub async fn process_torrent_response(params: ProcessTorrentResponse) -> Result<(), String> {
    let Some((torrent, staged_path)) = stage_and_verify_torrent(
        params.height,
        &params.db,
        params.torrent,
        params.wallet,
        params.process_now,
    )
    .await?
    else {
        return Ok(());
    };
    let torrent_bytes = torrent.clone().to_bytes().await;

    setup_download_for_torrent(
        params.height,
        torrent,
        staged_path,
        params.allow_during_reorg,
        params.db,
        params.map.clone(),
    )
    .await?;

    // Successful replay/download means this node can now seed the block
    // behind this torrent, so rebroadcasting is safe.
    broadcast_new_torrent_to_peers(params.height, &torrent_bytes, params.map).await;
    Ok(())
}

pub async fn stage_and_verify_torrent(
    height: u32,
    db: &Db,
    torrent: Torrent,
    wallet: Arc<Wallet>,
    process_now: bool,
) -> Result<Option<(Torrent, String)>, String> {
    // Stage the torrent first so a parseable candidate is never lost just
    // because the current chain state is in the middle of syncing or
    // orphan correction. Immediate validation/download only happens when
    // this torrent is actionable right now.
    let torrent_bytes = torrent.clone().to_bytes().await;
    let staged_path = save_staged_torrent(height, &torrent_bytes)
        .await
        .map_err(|err| format!("Failed to save staged torrent: {err}"))?;
    set_torrent_status(height, &torrent.info.info_hash, TorrentStatus::Pending).await;

    if !process_now {
        return Ok(None);
    }

    if let Err(error) = torrent.verify(height, db, wallet).await {
        warn!("[torrent] validation failed: height={height} err={error}");
        return Err(error);
    }

    Ok(Some((torrent, staged_path)))
}

pub async fn setup_download_for_torrent(
    height: u32,
    torrent: Torrent,
    staged_path: String,
    allow_during_reorg: bool,
    db: Db,
    map: Arc<Mutex<Command>>,
) -> Result<(), String> {
    let verification_service = global_verification_service()
        .ok_or_else(|| "Verification service not initialized".to_string())?;

    // Hand the staged torrent off to the download pipeline so the
    // full block can be assembled, verified, and saved.
    setup_download(
        height,
        torrent,
        staged_path,
        allow_during_reorg,
        db,
        Arc::new(verification_service),
        map,
    )
    .await
}