filespooler 1.2.4

/*! Tools for managing the queues */

/*
    Copyright (C) 2022 John Goerzen <jgoerzen@complete.org>

    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

use crate::header::FSMetaV1;
use crate::jobfile;
use crate::jobqueue;
use crate::seqfile;
use anyhow::bail;
use clap::Args;
use std::ffi::OsString;
use std::io::Write;
use std::os::unix::ffi::OsStringExt;

/// Generic CLI options for all operations on queues
#[derive(Debug, Args)]
pub struct QueueOpts {
    /// The path to the queue directory
    #[arg(short, long, value_name = "DIR")]
    pub queuedir: OsString,
}

/// Options for creating a queue
#[derive(Debug, Args)]
pub struct QueueInitOpts {
    #[clap(flatten)]
    pub qopts: QueueOpts,

    /// If yes, create the queue in write-only mode.
    #[arg(long)]
    pub append_only: bool,
}

/// Generic CLI options for all operations on queues that read packets
#[derive(Debug, Args)]
pub struct QueueOptsWithDecoder {
    #[clap(flatten)]
    pub qopts: QueueOpts,

    /// Command to decode the files.  Unlike other commands, this one is executed by a shell ($SHELL -c)
    #[arg(long, short, value_name = "DECODECMD")]
    pub decoder: Option<OsString>,
}

#[derive(Debug, Args)]
pub struct QueueJobOpts {
    #[clap(flatten)]
    pub qdopts: QueueOptsWithDecoder,

    /// The ID of the job to process.
    #[arg(short, long, value_name = "ID")]
    pub job: u64,
}

#[derive(Debug, Args)]
pub struct QueueSetSeqOpts {
    #[clap(flatten)]
    pub qopts: QueueOpts,

    /// The ID to set it to.
    #[arg(value_name = "ID")]
    pub nextid: u64,
}

pub fn cmd_queueinit(opts: &QueueInitOpts) -> Result<(), anyhow::Error> {
    jobqueue::queueinit(&opts.qopts.queuedir, opts.append_only)
}

/// Writes a packet in stdin to the queue.  Does not need to obtain a lock
/// to do so.
pub fn cmd_queuewrite(opts: &QueueOpts) -> Result<(), anyhow::Error> {
    let mut stdin_fd = std::io::stdin();

    jobqueue::queuewrite(&mut stdin_fd, &opts.queuedir)
}

/// Prints a listing of items in the queue.
pub fn cmd_queuels(opts: &QueueOptsWithDecoder) -> Result<(), anyhow::Error> {
    let mut entries: Vec<(OsString, FSMetaV1)> =
        jobqueue::scanqueue(&opts.qopts.queuedir, &opts.decoder)?
            .flatten()
            .collect();
    entries.sort_by(|(_, ma), (_, mb)| ma.seq.cmp(&mb.seq));
    let mut out = std::io::stdout();
    println!("{:20} {:27} filename", "ID", "creation timestamp");
    for (filename, meta) in entries.into_iter() {
        print!(
            "{:<20} {:27} ",
            meta.seq,
            meta.get_datestring_rfc3339_local(),
        );
        out.write_all(&filename.into_vec())?;
        out.write_all(b"\n")?;
    }
    Ok(())
}

/// Prints info about a specific job.
pub fn cmd_queueinfo(opts: &QueueJobOpts) -> Result<(), anyhow::Error> {
    let qmap = jobqueue::scanqueue_map(&opts.qdopts.qopts.queuedir, &opts.qdopts.decoder)?;
    match qmap.get(&opts.job) {
        Some((filename, meta)) => {
            meta.render(
                Some(filename),
                Some(&opts.qdopts.qopts.queuedir),
                &mut std::io::stdout(),
            )?;
        }
        None => bail!(
            "Job {} not found in directory {:?}",
            opts.job,
            opts.qdopts.qopts.queuedir
        ),
    }
    Ok(())
}

/// Render the payload of a specific job.
pub fn cmd_queuepayload(opts: &QueueJobOpts) -> Result<(), anyhow::Error> {
    let qmap = jobqueue::scanqueue_map(&opts.qdopts.qopts.queuedir, &opts.qdopts.decoder)?;
    match qmap.get(&opts.job) {
        Some((filename, _meta)) => {
            let mut inputinfo = jobqueue::queue_openjob(
                &opts.qdopts.qopts.queuedir,
                filename,
                &opts.qdopts.decoder,
            )?;
            let mut inhandle = inputinfo.reader.take().unwrap().as_read();
            let _meta = jobfile::read_jobfile_header(&mut inhandle)?;
            std::io::copy(&mut inhandle, &mut std::io::stdout())?;
        }
        None => bail!(
            "Job {} not found in directory {:?}",
            opts.job,
            opts.qdopts.qopts.queuedir
        ),
    }
    Ok(())
}

/// Print the next sequence number
pub fn cmd_queuegetnext(opts: &QueueOpts) -> Result<(), anyhow::Error> {
    let seqfn = jobqueue::get_seqfile(&opts.queuedir);
    let mut lock = seqfile::prepare_seqfile_lock(&seqfn, false)?;

    let seqf = seqfile::SeqFile::open(&seqfn, &mut lock)?;
    println!("{}", seqf.get_next());
    Ok(())
}

/// Set the next sequence number
pub fn cmd_queuesetnext(opts: &QueueSetSeqOpts) -> Result<(), anyhow::Error> {
    let seqfn = jobqueue::get_seqfile(&opts.qopts.queuedir);
    let mut lock = seqfile::prepare_seqfile_lock(&seqfn, false)?;

    let mut seqf = seqfile::SeqFile::open(&seqfn, &mut lock)?;
    seqf.set(opts.nextid)?;
    Ok(())
}