//! Bounded transfer pipeline (port of the 0.1.0 `TransferManager._pump`): at most `PIPELINE`
//! chunks in flight, the confirmed offset only advances over contiguous acknowledged chunks,
//! stop requests stop issuing new chunks, and a stall without confirmed data fails the run.

use crate::core::error::{CoreError, CoreResult};
use futures::future::BoxFuture;
use futures::stream::{FuturesUnordered, StreamExt};
use std::collections::BTreeMap;
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::watch;

pub const CHUNK: usize = 256 * 1024;
pub const PIPELINE: usize = 8;
pub const STALL: Duration = Duration::from_secs(60);

/// Local file with positioned, blocking-thread I/O.
#[derive(Clone)]
pub struct LocalFile(pub Arc<std::fs::File>);

impl LocalFile {
    pub fn open(path: &Path, flags: &str) -> std::io::Result<LocalFile> {
        let mut options = std::fs::OpenOptions::new();
        match flags {
            "r" => {
                options.read(true);
            }
            "r+" => {
                options.read(true).write(true);
            }
            // "wx": exclusive create, private mode.
            _ => {
                options.read(true).write(true).create_new(true);
                #[cfg(unix)]
                {
                    use std::os::unix::fs::OpenOptionsExt;
                    options.mode(0o600);
                }
            }
        }
        options.open(path).map(|f| LocalFile(Arc::new(f)))
    }

    pub async fn read_at(&self, offset: u64, len: usize) -> std::io::Result<Vec<u8>> {
        let file = self.0.clone();
        tokio::task::spawn_blocking(move || {
            let mut buffer = vec![0u8; len];
            let mut total = 0;
            while total < len {
                let read = positioned_read(&file, &mut buffer[total..], offset + total as u64)?;
                if read == 0 {
                    break;
                }
                total += read;
            }
            buffer.truncate(total);
            Ok(buffer)
        })
        .await
        .map_err(|_| std::io::Error::other("local read task failed"))?
    }

    pub async fn write_at(&self, offset: u64, data: Vec<u8>) -> std::io::Result<()> {
        let file = self.0.clone();
        tokio::task::spawn_blocking(move || {
            let mut written = 0;
            while written < data.len() {
                let n = positioned_write(&file, &data[written..], offset + written as u64)?;
                if n == 0 {
                    return Err(std::io::Error::new(std::io::ErrorKind::WriteZero, "write returned zero"));
                }
                written += n;
            }
            Ok(())
        })
        .await
        .map_err(|_| std::io::Error::other("local write task failed"))?
    }

    pub async fn finish(&self, size: u64) -> std::io::Result<()> {
        let file = self.0.clone();
        tokio::task::spawn_blocking(move || {
            file.set_len(size)?;
            file.sync_all()
        })
        .await
        .map_err(|_| std::io::Error::other("local sync task failed"))?
    }

    pub fn metadata(&self) -> std::io::Result<std::fs::Metadata> {
        self.0.metadata()
    }
}

#[cfg(unix)]
fn positioned_read(file: &std::fs::File, buffer: &mut [u8], offset: u64) -> std::io::Result<usize> {
    std::os::unix::fs::FileExt::read_at(file, buffer, offset)
}
#[cfg(windows)]
fn positioned_read(file: &std::fs::File, buffer: &mut [u8], offset: u64) -> std::io::Result<usize> {
    std::os::windows::fs::FileExt::seek_read(file, buffer, offset)
}
#[cfg(unix)]
fn positioned_write(file: &std::fs::File, data: &[u8], offset: u64) -> std::io::Result<usize> {
    std::os::unix::fs::FileExt::write_at(file, data, offset)
}
#[cfg(windows)]
fn positioned_write(file: &std::fs::File, data: &[u8], offset: u64) -> std::io::Result<usize> {
    std::os::windows::fs::FileExt::seek_write(file, data, offset)
}

pub type Producer = Arc<dyn Fn(u64, usize) -> BoxFuture<'static, CoreResult<Vec<u8>>> + Send + Sync>;
pub type Consumer = Arc<dyn Fn(Vec<u8>, u64) -> BoxFuture<'static, CoreResult<()>> + Send + Sync>;

/// Why the pump ended without completing.
#[derive(Debug)]
pub enum PumpStop {
    /// A stop request (pause/cancel/shutdown) was honoured; carries the kind.
    Stopped(String),
    Failed(CoreError),
}

/// Copies `[start, size)` from `produce` to `consume`. `progress(confirmed)` is called after
/// each contiguous advance. `stop` carries the requested stop kind (empty = keep running).
pub async fn pump(start: u64, size: u64, produce: Producer, consume: Consumer, mut stop: watch::Receiver<String>, mut progress: impl FnMut(u64) + Send) -> Result<u64, PumpStop> {
    let mut next = start;
    let mut confirmed = start;
    let mut completed: BTreeMap<u64, u64> = BTreeMap::new();
    let mut inflight = FuturesUnordered::new();
    let mut failure: Option<CoreError> = None;
    let mut last_progress = Instant::now();
    let grace = Duration::from_secs(5);
    let mut stop_seen: Option<(String, Instant)> = None;
    loop {
        let stop_kind = stop.borrow().clone();
        if !stop_kind.is_empty() && stop_seen.is_none() {
            stop_seen = Some((stop_kind, Instant::now()));
        }
        while failure.is_none() && stop_seen.is_none() && inflight.len() < PIPELINE && next < size {
            let position = next;
            let length = ((size - position) as usize).min(CHUNK);
            next += length as u64;
            let (produce, consume) = (produce.clone(), consume.clone());
            inflight.push(async move {
                let buffer = produce(position, length).await?;
                consume(buffer, position).await?;
                Ok::<_, CoreError>((position, length as u64))
            });
        }
        if inflight.is_empty() {
            if let Some(error) = failure {
                return Err(PumpStop::Failed(error));
            }
            if let Some((kind, _)) = stop_seen {
                return Err(PumpStop::Stopped(kind));
            }
            if confirmed >= size {
                return Ok(confirmed);
            }
        }
        // Requests already on the wire normally settle quickly; a dead link is cut after the grace period.
        if let Some((kind, since)) = &stop_seen {
            // "shutdown" and a lost connection end the run at once.
            if since.elapsed() >= if kind == "shutdown" || kind == "lost" { Duration::ZERO } else { grace } {
                return Err(PumpStop::Stopped(kind.clone()));
            }
        }
        if last_progress.elapsed() > STALL && failure.is_none() {
            return Err(PumpStop::Failed(CoreError::new("TRANSFER_STALLED", "No data was confirmed for 60 seconds.")));
        }
        tokio::select! {
            item = inflight.next(), if !inflight.is_empty() => {
                match item {
                    Some(Ok((position, length))) => {
                        completed.insert(position, length);
                        let before = confirmed;
                        while let Some(length) = completed.remove(&confirmed) {
                            confirmed += length;
                        }
                        last_progress = Instant::now();
                        if confirmed > before {
                            progress(confirmed);
                        }
                    }
                    Some(Err(error)) => {
                        if failure.is_none() {
                            failure = Some(error);
                        }
                    }
                    None => {}
                }
            }
            _ = stop.changed() => {}
            _ = tokio::time::sleep(Duration::from_millis(500)) => {}
        }
    }
}

/// Resume check: the last confirmed bytes of the partial copy must equal the source.
pub const TAIL_CHECK: u64 = 1024 * 1024;

pub async fn verify_tail(transferred: u64, read_local: BoxFuture<'_, CoreResult<Vec<u8>>>, read_remote: BoxFuture<'_, CoreResult<Vec<u8>>>, window: u64) -> CoreResult<()> {
    if window == 0 || transferred == 0 {
        return Ok(());
    }
    let (left, right) = futures::join!(read_local, read_remote);
    let (left, right) = (left?, right?);
    if left.len() as u64 != window || right.len() as u64 != window || left != right {
        return Err(CoreError::new("FILE_RESUME_MISMATCH", "FILE_RESUME_MISMATCH: The partial data does not match the source; restart the transfer from the beginning."));
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use parking_lot::Mutex;

    #[tokio::test]
    async fn pump_copies_in_order_and_honours_stop() {
        let source: Arc<Vec<u8>> = Arc::new((0..(3 * CHUNK + 777)).map(|i| (i % 251) as u8).collect());
        let target = Arc::new(Mutex::new(vec![0u8; source.len()]));
        let src = source.clone();
        let produce: Producer = Arc::new(move |pos, len| {
            let src = src.clone();
            Box::pin(async move {
                // Out-of-order completion.
                tokio::time::sleep(Duration::from_millis((pos / CHUNK as u64 % 3) * 5)).await;
                Ok(src[pos as usize..pos as usize + len].to_vec())
            })
        });
        let dst = target.clone();
        let consume: Consumer = Arc::new(move |buf, pos| {
            let dst = dst.clone();
            Box::pin(async move {
                dst.lock()[pos as usize..pos as usize + buf.len()].copy_from_slice(&buf);
                Ok(())
            })
        });
        let (_tx, rx) = watch::channel(String::new());
        let mut seen = Vec::new();
        let done = pump(0, source.len() as u64, produce.clone(), consume.clone(), rx, |c| seen.push(c)).await.unwrap();
        assert_eq!(done, source.len() as u64);
        assert_eq!(*target.lock(), *source);
        assert!(seen.windows(2).all(|w| w[0] < w[1]));
        let (tx, rx) = watch::channel(String::new());
        tx.send("pause".into()).unwrap();
        match pump(0, source.len() as u64, produce, consume, rx, |_| {}).await {
            Err(PumpStop::Stopped(kind)) => assert_eq!(kind, "pause"),
            other => panic!("{other:?}"),
        }
    }
}