//! Ordered, bounded encode+write queue for production maps (Tim t1063u).
//!
//! The render worker resolves a view, then hands already-encoded PNG bytes to a
//! single IO thread that publishes them with [`crate::resume::RunSession`] in
//! deterministic journal order. A bounded channel provides backpressure so only
//! a few views sit in flight between GPU/CPU and disk.
use crate::resume::RunSession;
use std::{
    collections::BTreeMap,
    sync::{
        atomic::{AtomicBool, Ordering},
        mpsc, Arc,
    },
    thread::{self, JoinHandle},
};

/// Max view-bundles waiting on the IO thread (backpressure).
pub const IO_QUEUE_VIEWS: usize = 4;

/// One planned PNG: relative path + encoded file bytes.
#[derive(Debug)]
pub struct EncodedMap {
    pub rel: String,
    pub bytes: Vec<u8>,
}

/// All maps of one view, in [`crate::production::SAVED_ORDER`].
#[derive(Debug)]
pub struct ViewPublish {
    pub maps: Vec<EncodedMap>,
}

enum Msg {
    View(ViewPublish),
    /// Stop after draining; reply with the session and digests.
    Finish(mpsc::SyncSender<IoFinish>),
}

/// Result of shutting down the IO thread.
pub struct IoFinish {
    pub session: RunSession,
    pub png_sha256: BTreeMap<String, String>,
    pub error: Option<String>,
}

/// Handle for the ordered publisher.
pub struct IoQueue {
    tx: Option<mpsc::SyncSender<Msg>>,
    /// Shared cancel; IO stops accepting new publishes when set (still finishes Finish).
    cancel: Arc<AtomicBool>,
    handle: Option<JoinHandle<()>>,
    finished: bool,
}

impl IoQueue {
    /// Move `session` onto a dedicated IO thread. `seed` are digests already
    /// verified (resume skip); they are merged into the final map.
    pub fn start(
        session: RunSession,
        seed: BTreeMap<String, String>,
        cancel: Arc<AtomicBool>,
    ) -> Self {
        let (tx, rx) = mpsc::sync_channel::<Msg>(IO_QUEUE_VIEWS);
        let handle = thread::spawn(move || {
            let mut session = session;
            let mut png_sha256 = seed;
            let mut error: Option<String> = None;
            for msg in rx {
                match msg {
                    Msg::View(view) => {
                        // Once queued, always publish — cancel only blocks new
                        // submit()s so Resume keeps already-finished maps.
                        if error.is_some() {
                            continue;
                        }
                        match publish_view(&mut session, &mut png_sha256, view) {
                            Ok(()) => {}
                            Err(e) => error = Some(e),
                        }
                    }
                    Msg::Finish(reply) => {
                        let _ = reply.send(IoFinish {
                            session,
                            png_sha256,
                            error,
                        });
                        return;
                    }
                }
            }
        });
        Self {
            tx: Some(tx),
            cancel,
            handle: Some(handle),
            finished: false,
        }
    }

    /// Enqueue one view. Blocks when the queue is full (backpressure).
    /// Returns Err if the IO thread is gone or cancel was set before send.
    pub fn submit(&self, view: ViewPublish) -> Result<(), String> {
        if self.cancel.load(Ordering::Relaxed) {
            return Err("geannuleerd".into());
        }
        self.tx
            .as_ref()
            .ok_or_else(|| "IO-queue gestopt".to_string())?
            .send(Msg::View(view))
            .map_err(|_| "IO-queue gestopt".to_string())
    }

    /// Drain the queue, stop the thread, and return session + digests.
    pub fn finish(mut self) -> Result<(RunSession, BTreeMap<String, String>), String> {
        let finished = self.shutdown()?;
        self.finished = true;
        if let Some(e) = finished.error {
            return Err(e);
        }
        Ok((finished.session, finished.png_sha256))
    }

    fn shutdown(&mut self) -> Result<IoFinish, String> {
        let tx = self
            .tx
            .take()
            .ok_or_else(|| "IO-queue al gestopt".to_string())?;
        let (reply_tx, reply_rx) = mpsc::sync_channel(1);
        tx.send(Msg::Finish(reply_tx))
            .map_err(|_| "IO-queue gestopt".to_string())?;
        drop(tx);
        let finished = reply_rx
            .recv()
            .map_err(|_| "IO-queue gestopt zonder antwoord".to_string())?;
        if let Some(h) = self.handle.take() {
            let _ = h.join();
        }
        Ok(finished)
    }
}

impl Drop for IoQueue {
    fn drop(&mut self) {
        if self.finished || self.tx.is_none() {
            return;
        }
        let _ = self.shutdown();
    }
}

fn publish_view(
    session: &mut RunSession,
    png_sha256: &mut BTreeMap<String, String>,
    view: ViewPublish,
) -> Result<(), String> {
    for map in view.maps {
        let digest = session.publish(&map.rel, &map.bytes)?;
        png_sha256.insert(map.rel, digest);
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::resume::RunSession;
    use std::path::PathBuf;

    struct Temp(PathBuf);
    impl Temp {
        fn new() -> Self {
            let p = std::env::temp_dir().join(format!(
                "ppd-io-queue-{}-{}",
                std::process::id(),
                std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_nanos()
            ));
            std::fs::create_dir_all(&p).unwrap();
            Self(p)
        }
    }
    impl Drop for Temp {
        fn drop(&mut self) {
            let _ = std::fs::remove_dir_all(&self.0);
        }
    }

    #[test]
    fn publishes_in_submit_order_and_respects_backpressure_capacity() {
        let tmp = Temp::new();
        let session = RunSession::fresh(&tmp.0).unwrap();
        let cancel = Arc::new(AtomicBool::new(false));
        let q = IoQueue::start(session, BTreeMap::new(), cancel);
        for i in 0..6u32 {
            q.submit(ViewPublish {
                maps: vec![EncodedMap {
                    rel: format!("raw files/m/f{i}.png"),
                    bytes: format!("png-{i}").into_bytes(),
                }],
            })
            .unwrap();
        }
        let (session, digests) = q.finish().unwrap();
        drop(session);
        let mut keys: Vec<_> = digests.keys().cloned().collect();
        keys.sort();
        assert_eq!(
            keys,
            (0..6)
                .map(|i| format!("raw files/m/f{i}.png"))
                .collect::<Vec<_>>()
        );
        // Journal order matches submit order.
        let journal = std::fs::read_to_string(tmp.0.join(crate::resume::JOURNAL)).unwrap();
        let mut seen = Vec::new();
        for line in journal.lines() {
            if line.trim().is_empty() {
                continue;
            }
            let v: serde_json::Value = serde_json::from_str(line).unwrap();
            if let Some(p) = v.get("path").and_then(|x| x.as_str()) {
                seen.push(p.to_string());
            }
        }
        assert_eq!(
            seen,
            (0..6)
                .map(|i| format!("raw files/m/f{i}.png"))
                .collect::<Vec<_>>()
        );
    }

    #[test]
    fn cancel_before_submit_returns_error() {
        let tmp = Temp::new();
        let session = RunSession::fresh(&tmp.0).unwrap();
        let cancel = Arc::new(AtomicBool::new(true));
        let q = IoQueue::start(session, BTreeMap::new(), cancel);
        let err = q
            .submit(ViewPublish {
                maps: vec![EncodedMap {
                    rel: "a.png".into(),
                    bytes: b"x".to_vec(),
                }],
            })
            .unwrap_err();
        assert!(err.contains("geannuleerd"));
        // finish still returns (with cancel error if anything was queued — here none)
        let _ = q.finish();
    }
}