use crate::records::memory::response_channels::{reserve_entry_with_context, Command};
use crate::rpc::command_maps::RPC_TORRENT_CANDIDATES;
use crate::rpc::responses::RpcResponse;
use crate::torrent::torrenting_system::save_torrent::save_staged_torrent;
use crate::{timeout, Arc, Duration, Mutex, TcpStream};

pub async fn hydrate_torrent_candidates(
    stream: Arc<Mutex<TcpStream>>,
    map: Arc<Mutex<Command>>,
    connections_key: String,
) -> Result<usize, String> {
    // Reserve a reply slot and send a small request packet asking the peer for
    // its staged/local torrent candidates.
    let (hashmap_key, _tx, rx) = reserve_entry_with_context(
        map.clone(),
        Some(RPC_TORRENT_CANDIDATES),
        Some(connections_key.clone()),
    )
    .await;
    let mut message = Vec::with_capacity(4);
    message.push(RPC_TORRENT_CANDIDATES);
    message.extend_from_slice(&hashmap_key);

    RpcResponse::send_raw(&stream, Some(&connections_key), &message).await;

    let response = {
        let mut rx = rx.lock().await;
        timeout(Duration::from_secs(30), rx.recv())
            .await
            .map_err(|_| "Timed out waiting for torrent candidates".to_string())?
            .ok_or_else(|| "Torrent candidate response channel closed".to_string())?
    };

    if response.len() < 4 {
        return Err("Torrent candidate response was too short".to_string());
    }

    // The response starts with the number of entries, then each entry is
    // height + byte length + raw torrent bytes.
    let mut offset = 0_usize;
    let candidate_count = u32::from_le_bytes(
        response[offset..offset + 4]
            .try_into()
            .map_err(|_| "Failed to read torrent candidate count".to_string())?,
    );
    offset += 4;

    let mut imported = 0_usize;
    for _ in 0..candidate_count {
        // Refuse truncated entries rather than trying to save partial torrent
        // bytes into staging.
        if response.len().saturating_sub(offset) < 8 {
            return Err("Torrent candidate entry was truncated".to_string());
        }

        let height = u32::from_le_bytes(
            response[offset..offset + 4]
                .try_into()
                .map_err(|_| "Failed to read torrent candidate height".to_string())?,
        );
        offset += 4;

        let torrent_len = u32::from_le_bytes(
            response[offset..offset + 4]
                .try_into()
                .map_err(|_| "Failed to read torrent candidate length".to_string())?,
        ) as usize;
        offset += 4;

        if response.len().saturating_sub(offset) < torrent_len {
            return Err("Torrent candidate bytes were truncated".to_string());
        }

        let torrent_bytes = &response[offset..offset + torrent_len];
        // Imported candidates are staged locally; normal orphan replay decides
        // later whether any of them should become canonical blocks.
        save_staged_torrent(height, torrent_bytes).await?;
        imported += 1;
        offset += torrent_len;
    }

    Ok(imported)
}