// SPDX-License-Identifier: AGPL-3.0-or-later

use futures::{StreamExt, stream};
use tokio::sync::mpsc::Receiver;

use crate::{cmd::Cmd, stats::Stats};

pub(crate) struct Executor {
    cmd_queue: Receiver<Cmd>,
    stats: Stats,
}
impl Executor {
    pub fn new(cmd_queue: Receiver<Cmd>) -> Self {
        Self {
            cmd_queue,
            stats: Stats::new(),
        }
    }

    pub fn with_stats(cmd_queue: Receiver<Cmd>, stats: Stats) -> Self {
        Self { cmd_queue, stats }
    }

    pub async fn run(mut self) -> Stats {
        let s = stream::unfold(self.cmd_queue, async |mut queue| {
            queue.recv().await.map(|cmd| (cmd, queue))
        })
        .map(async |cmd| cmd.run().await)
        .buffer_unordered(8);

        let mut s = std::pin::pin!(s);

        while let Some(res) = s.next().await {
            self.stats.record_cmd_result(res);
        }

        self.stats
    }
}
