filespooler 1.2.4

/*! The on-disk job queue */
/*
   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::seqfile;
use anyhow::bail;
use command_group::{CommandGroup, GroupChild};
use std::collections::BTreeMap;
use std::ffi::OsString;
use std::fs::{create_dir, DirEntry, File};
use std::io::{Read, Write};
use std::os::unix::ffi::OsStrExt;
use std::path::PathBuf;
use std::process::{ChildStdout, Command, Stdio};
use tracing::*;
use uuid::Uuid;

/// Copies data from the given source to a new file in the queue.  Does
/// not need to obtain a lock to do so.
pub fn queuewrite<R: Read>(input: &mut R, queuedir: &OsString) -> Result<(), anyhow::Error> {
    let mut buf = Uuid::encode_buffer();
    let uuid = Uuid::new_v4().hyphenated().encode_lower(&mut buf);
    let fnpart: String = format!("fspl-{}.fspl", uuid);
    let tmpfnpart: String = format!("tmp-{}.tmp", fnpart);

    let tmppath = PathBuf::from(queuedir).join("jobs").join(tmpfnpart);
    debug!("Copying source to temporary file {:?}", tmppath);
    let mut tmpfile = File::create(&tmppath)?;

    let bytes = std::io::copy(input, &mut tmpfile)?;

    tmpfile.flush()?;
    tmpfile.sync_all()?;
    // Force it to close and write.
    std::mem::drop(tmpfile);

    let finalpath = PathBuf::from(queuedir).join("jobs").join(fnpart);
    debug!(
        "Wrote {} bytes.  Now renaming {:?} to {:?}",
        bytes, tmppath, finalpath
    );
    std::fs::rename(tmppath, finalpath)?;
    Ok(())
}

/// Gets the name of the seqfile, given the path to the queue.
pub fn get_seqfile(queuedir: &OsString) -> OsString {
    let mut seqfile = queuedir.clone();
    seqfile.push("/nextseq");
    seqfile
}

/// Create a queue directory structure.
pub fn queueinit(queuedir: &OsString, append_only: bool) -> Result<(), anyhow::Error> {
    let path = PathBuf::from(queuedir);
    create_dir(&path)?;
    create_dir(path.join("jobs"))?;

    if !append_only {
        let seqfn = get_seqfile(queuedir);
        let mut lock = seqfile::prepare_seqfile_lock(&seqfn, true)?;

        seqfile::SeqFile::open(&seqfn, &mut lock)?;
    }

    Ok(())
}

/// Scan a jobs directory, yielding an iterator of (filename, metadata).  Any
/// files in the directory that do not match the pattern `fspl-*.fspl` will be
/// ignored.  Any files that do match the pattern but do not possess a valid
/// [`crate::header::FSPrefix`] will be ignored.
///
/// Multiple jobs with the same ID are acceptable here.
pub fn scanqueue(
    queuedir: &OsString,
    decoder: &Option<OsString>,
) -> Result<impl Iterator<Item = Result<(OsString, FSMetaV1), anyhow::Error>>, anyhow::Error> {
    fn procdir(
        queuedir: &OsString,
        decoder: &Option<OsString>,
        entry: std::io::Result<DirEntry>,
    ) -> Result<(OsString, FSMetaV1), anyhow::Error> {
        let e = entry?;
        let filename = e.file_name();
        if !(filename.as_bytes().starts_with(b"fspl-") && filename.as_bytes().ends_with(b".fspl")) {
            debug!("Ignoring file {:?}", filename);
            bail!("File {:?} doesn't match our specs", filename);
        }
        debug!(
            "Queue scan reading header from {:?}",
            queue_genfilename(queuedir, &filename)
        );
        let mut input = queue_openjob(queuedir, &filename, decoder)?;
        let meta = jobfile::read_jobfile_header(&mut input.reader.take().unwrap().as_read())?;
        Ok((filename, meta))
    }

    let dirpath = PathBuf::from(queuedir).join("jobs");
    let dir = std::fs::read_dir(&dirpath)?;
    let queuedir = queuedir.clone();
    let decoder = decoder.clone();
    Ok(dir.map(move |e| procdir(&queuedir, &decoder, e)))
}

/// Scan a jobs directory, returning a map from job id to (filename, FSMetaV1).
/// Calls [`scanqueue`] internally, and the comments about valid files there
/// apply to this function as well.
///
/// Multiple jobs with the same sequence ID are not acceptable here and will generate
/// an Err return.
#[instrument(level = "debug")]
pub fn scanqueue_map(
    queuedir: &OsString,
    decoder: &Option<OsString>,
) -> Result<BTreeMap<u64, (OsString, FSMetaV1)>, anyhow::Error> {
    let mut retval = BTreeMap::new();
    // flatten() here removes the Err cases.
    // Called because an error here could be permission denied, file we don't know about, etc.  Just ignore.
    for (filename, meta) in scanqueue(queuedir, decoder)?.flatten() {
        if let Some(prev) = retval.insert(meta.seq, (filename.clone(), meta.clone())) {
            bail!(
                "Attempted to process {:?} with seq {}, which was already seen in {:?}",
                filename,
                meta.seq,
                prev.0
            );
        }
    }
    Ok(retval)
}

/// Given a queue directory and a filename from [`scanqueue] or [`scanqueue_map`],
/// gives the ultimate filename to use for opening.
pub fn queue_genfilename(queuedir: &OsString, filename: &OsString) -> PathBuf {
    PathBuf::from(queuedir).join("jobs").join(filename)
}

/// The decoder handle
pub enum DecoderHandle {
    DHFile(Box<File>),
    DHChildStdout(Box<ChildStdout>),
}

impl From<File> for DecoderHandle {
    fn from(f: File) -> Self {
        DecoderHandle::DHFile(Box::new(f))
    }
}

impl From<ChildStdout> for DecoderHandle {
    fn from(f: ChildStdout) -> Self {
        DecoderHandle::DHChildStdout(Box::new(f))
    }
}

impl From<DecoderHandle> for Stdio {
    fn from(dh: DecoderHandle) -> Stdio {
        match dh {
            DecoderHandle::DHFile(f) => (*f).into(),
            DecoderHandle::DHChildStdout(f) => (*f).into(),
        }
    }
}

impl DecoderHandle {
    /// Converts to a Read.
    pub fn as_read(self) -> Box<dyn Read> {
        match self {
            DecoderHandle::DHFile(f) => f,
            DecoderHandle::DHChildStdout(f) => f,
        }
    }

    /// Passes an underlying [`Read`] to a closure without consuming self.
    pub fn with_read<T>(&mut self, func: fn(&mut dyn Read) -> T) -> T {
        match self {
            DecoderHandle::DHFile(o) => func(&mut *o),
            DecoderHandle::DHChildStdout(o) => func(&mut *o),
        }
    }
}

/// A struct representing a possible decoder.  When dropped,
/// kills the decoder process group and tries to wait for it to prevent
/// zombies and runaway decoder processes.
pub struct PossibleDecoder {
    pub child: Option<GroupChild>,

    /// The reader handle.  It's an option so you can take it and move it out of here.
    pub reader: Option<DecoderHandle>,
}

impl Drop for PossibleDecoder {
    fn drop(&mut self) {
        if let Some(mut child) = self.child.take() {
            trace!("Killing decoder");
            let _ = child.kill();
            trace!("Waiting for decoder to terminate");
            let res = child.wait();
            trace!("Decoder termination status {:?}", res);
        }
    }
}

/// Given a queue directory and a filename, opens the file for reading and returns
/// the file handle.  This is the central function used by all others that
/// read or scan the queue.
pub fn queue_openjob(
    queuedir: &OsString,
    filename: &OsString,
    decoder: &Option<OsString>,
) -> Result<PossibleDecoder, anyhow::Error> {
    let genfilename = queue_genfilename(queuedir, filename);
    trace!("Opening queue file at {:?}", genfilename);

    let file = File::open(genfilename)?;
    match decoder {
        None => Ok(PossibleDecoder {
            child: None,
            reader: Some(file.into()),
        }),
        Some(decodecmd) => {
            let (child, childstdout) = with_decoder(decodecmd, file)?;
            Ok(PossibleDecoder {
                child: Some(child),
                reader: Some(childstdout.into()),
            })
        }
    }
}

/** Wrap up input in a decoder.  Returns a Child to wait on, and a ChildStdout that holds
the output. */
#[instrument(level = "debug", skip(input))]
pub fn with_decoder<T: Into<Stdio>>(
    decoder: &OsString,
    input: T,
) -> Result<(GroupChild, ChildStdout), anyhow::Error> {
    let args = [OsString::from("-c"), decoder.clone()];
    let shell = getshell();
    debug!("Preparing to invoke decoder: {:?} {:?}", shell, args);
    let mut child = Command::new(shell)
        .args(args)
        .stdin(input)
        .stdout(Stdio::piped())
        .group_spawn()?;
    debug!("Decoder PID {} started successfully", child.id());
    let stdout = child.inner().stdout.take().expect("Missing stdout in child");
    Ok((child, stdout))
}

/// Get the name of the user's shell.
pub fn getshell() -> OsString {
    match std::env::var_os("SHELL") {
        Some(x) => x,
        None => OsString::from("/bin/sh"),
    }
}