//! Content of remakes: the new document of a `replace` or `import` remake.
//!
//! The proposal records only a reference (BLAKE3 hash and size). The coordinator keeps
//! the bytes in memory; every member fetches them before voting (in chunks, from
//! several peers at once: `swarm::blob`) and serves them in turn. A committed remake
//! therefore has its content on a majority of the electorate. Nodes that switch need
//! it to build the snapshot; nodes that rejoin later don't (they download the snapshot
//! itself).
//!
//! Content no open remake refers to is dropped an hour after it was stored.

use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;

use tracing::{info, warn};

use super::Host;
use super::records::{ContentRef, Record};

/// The content `reference` names, if this node has it; otherwise starts fetching it in
/// the background (once) and returns None.
pub(super) fn available_or_fetch<H: Host>(
    host: &H,
    reference: &ContentRef,
) -> Option<Arc<Vec<u8>>> {
    let hash = reference.hash()?;
    if let Some(bytes) = host.remake().content(&hash) {
        return Some(bytes);
    }
    if host.remake().start_content_fetch(hash) {
        let host = host.clone();
        let reference = reference.clone();
        tokio::spawn(async move {
            fetch(&host, hash, &reference).await;
            host.remake().end_content_fetch(&hash);
        });
    }
    None
}

/// The content `reference` names: from this node, or fetched now. Waits for a download
/// already running.
pub(super) async fn ensure<H: Host>(
    host: &H,
    reference: &ContentRef,
) -> Result<Arc<Vec<u8>>, String> {
    let hash = reference
        .hash()
        .ok_or_else(|| format!("unreadable content hash {:?}", reference.blake3))?;
    loop {
        if let Some(bytes) = host.remake().content(&hash) {
            return Ok(bytes);
        }
        if host.remake().start_content_fetch(hash) {
            let fetched = fetch(host, hash, reference).await;
            host.remake().end_content_fetch(&hash);
            return fetched
                .ok_or_else(|| format!("couldn't download content {}", reference.blake3));
        }
        // Someone else is downloading it.
        while host.remake().content_fetch_running(&hash) {
            tokio::time::sleep(Duration::from_millis(500)).await;
        }
    }
}

async fn fetch<H: Host>(host: &H, hash: [u8; 32], reference: &ContentRef) -> Option<Arc<Vec<u8>>> {
    match host.fetch_content(hash).await {
        Ok(bytes) if bytes.len() as u64 == reference.size => {
            info!(content = %reference.blake3, bytes = bytes.len(), "Downloaded remake content");
            let bytes = Arc::new(bytes);
            host.remake().put_content(hash, Arc::clone(&bytes));
            Some(bytes)
        }
        Ok(bytes) => {
            warn!(content = %reference.blake3, bytes = bytes.len(), expected = reference.size, "Remake content has the wrong size");
            None
        }
        Err(error) => {
            warn!(content = %reference.blake3, error = %format!("{error:#}"), "Couldn't download remake content");
            None
        }
    }
}

/// Drops content no open remake in `records` refers to (after its grace period).
pub(super) fn prune<H: Host>(host: &H, records: &[(String, Record)], now: u64, grace: u64) {
    let referenced: HashSet<[u8; 32]> = records
        .iter()
        .filter(|(_, record)| record.phase(now, grace).is_open())
        .filter_map(|(_, record)| record.proposal.content.as_ref()?.hash())
        .collect();
    host.remake().prune_contents(&referenced);
}