json-job-dispatch 2.0.1

// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
// option. This file may not be copied, modified, or distributed
// except according to those terms.

use std::borrow::Cow;
use std::collections::hash_map::{Entry, HashMap};
use std::error;
use std::ffi::OsStr;
use std::fmt::{self, Debug, Display};
use std::fs::{self, File};
use std::io::{self, Write};
use std::path::{Path, PathBuf};

use crates::chrono::Utc;
use crates::inotify::EventMask;
use crates::itertools::Itertools;
use crates::rand::{self, Rng};
use crates::serde_json::{self, Map, Value};
use crates::thiserror::Error;

use handler::{Handler, HandlerResult};
use watcher::Watcher;

/// Dispatch jobs to registered handlers.
///
/// Jobs are sorted into `accept`, `reject`, and `fail` directories based on whether they were
/// accepted by the relevant handler. Once handled, a `.stamp` file containing the timestamp of
/// when the job was completed is created beside the final location. In addition, rejected and
/// failed jobs have a `.reason` file describing what happened.
pub struct Director<'a> {
    /// The path to the directory for accepted jobs.
    accept: PathBuf,
    /// The path to the directory for rejected jobs.
    reject: PathBuf,
    /// The path to the directory for failed jobs.
    fail: PathBuf,

    /// The set of handlers to run.
    handlers: HashMap<String, &'a dyn Handler>,
}

/// The filename of a path for logging purposes.
fn log_file_name<P>(path: &P) -> Cow<str>
where
    P: AsRef<Path>,
{
    path.as_ref()
        .file_name()
        .map_or_else(|| Cow::Borrowed("."), OsStr::to_string_lossy)
}

impl<'a> Debug for Director<'a> {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.debug_struct("Director")
            .field("accept", &self.accept)
            .field("reject", &self.reject)
            .field("fail", &self.fail)
            .field("handlers", &self.handlers.keys().collect::<Vec<_>>())
            .finish()
    }
}

/// The result of running jobs.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum RunResult {
    /// The director should continue running.
    Continue,
    /// The director should be restarted from the beginning.
    Restart,
    /// The director should be terminated.
    Done,
}

impl RunResult {
    /// Whether the result indicates that the director should exit.
    fn should_exit(self) -> bool {
        if let RunResult::Continue = self {
            false
        } else {
            true
        }
    }

    /// Whether the result indicates that the director is done.
    fn is_done(self) -> bool {
        if let RunResult::Done = self {
            true
        } else {
            false
        }
    }
}

/// Outbox labels.
///
/// Jobs, once processed, are assigned one of three labels. Each label is its own directory where
/// completed jobs are placed once they have been handled.
///
/// Once placed in an outbox, the job is accompanied by a `.stamp` file beside it to indicate when
/// the job was handled.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Outbox {
    /// Jobs which have been accepted and acted upon without error.
    Accept,
    /// Jobs for which there is nothing to do.
    ///
    /// Each job contains a `.reason` file beside it indicating the reason for rejection.
    Reject,
    /// Jobs which caused an error to occur.
    ///
    /// As with `Reject`, but the `.reason` contains error information.
    Fail,
}

impl Outbox {
    /// The name of the directory to use for the outbox.
    pub fn name(self) -> &'static str {
        match self {
            Outbox::Accept => "accept",
            Outbox::Reject => "reject",
            Outbox::Fail => "fail",
        }
    }
}

impl Display for Outbox {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(f, "{}", self.name())
    }
}

/// Errors which may occur when running a director.
#[derive(Debug, Error)]
// TODO: #[non_exhaustive]
pub enum DirectorError {
    /// The director was given a non-directory as its input queue path.
    #[error("not a directory: {}", path.display())]
    NotADirectory {
        /// The path given to the director.
        path: PathBuf,
    },
    /// Two handlers for the same `kind` were added.
    #[error("duplicate handler: {}", kind)]
    DuplicateHandler {
        /// The duplicate `kind`.
        kind: String,
    },
    /// Failure to create outbox directory.
    #[error("failed to create {} directory: {}", outbox, source)]
    CreateDirectory {
        /// The outbox that could not be created.
        outbox: Outbox,
        /// The source of the error.
        #[source]
        source: io::Error,
    },
    /// Failure to create a stamp or reason file.
    #[error("failed to create file {}: {}", filename.display(), source)]
    CreateFile {
        /// The file that could not be created.
        filename: PathBuf,
        /// The source of the error.
        #[source]
        source: io::Error,
    },
    /// Failure to write a stamp or reason file.
    #[error("failed to write file {}: {}", filename.display(), source)]
    WriteFile {
        /// The file that could not be written.
        filename: PathBuf,
        /// The source of the error.
        #[source]
        source: io::Error,
    },
    /// Failure to move a job to an outbox.
    #[error("failed to move job {} into {}: {}", filepath.display(), outbox, source)]
    MoveJob {
        /// The outbox that the job was to be moved into.
        outbox: Outbox,
        /// The job file that could not be moved.
        filepath: PathBuf,
        /// The source of the error.
        #[source]
        source: io::Error,
    },
    /// Failure to list the input queue's directory.
    #[error("failed to list the queue {}: {}", path.display(), source)]
    ListQueue {
        /// The path to the queue.
        path: PathBuf,
        /// The source of the error.
        #[source]
        source: io::Error,
    },
    /// Failure to watch a queue directory for new files.
    #[error("failed to watch the queue {}: {}", path.display(), source)]
    WatchQueue {
        /// The path to the queue.
        path: PathBuf,
        /// The source of the error.
        #[source]
        source: io::Error,
    },
    /// Failure to open a job file.
    #[error("failed to open job {}: {}", filepath.display(), source)]
    OpenJob {
        /// The path to the job.
        filepath: PathBuf,
        /// The source of the error.
        #[source]
        source: io::Error,
    },
    /// Failure to create a defer job.
    #[error("failed to defer job {}: {}", filepath.display(), source)]
    CreateDeferJob {
        /// The path to the defer job.
        filepath: PathBuf,
        /// The source of the error.
        #[source]
        source: io::Error,
    },
    /// Failure to write a defer job.
    #[error("failed to defer job {}: {}", filepath.display(), source)]
    WriteDeferJob {
        /// The path to the defer job.
        filepath: PathBuf,
        /// The source of the error.
        #[source]
        source: serde_json::Error,
    },
    /// A handler returned an error.
    #[error("handler error: {}", source)]
    Handler {
        /// The source of the error.
        #[source]
        source: Box<dyn error::Error + Send + Sync>,
    },
    /// This is here to force `_` matching right now.
    ///
    /// **DO NOT USE**
    #[doc(hidden)]
    #[error("unreachable...")]
    _NonExhaustive,
}

impl DirectorError {
    fn not_a_directory(path: &Path) -> Self {
        DirectorError::NotADirectory {
            path: path.into(),
        }
    }

    fn duplicate_handler(kind: &str) -> Self {
        DirectorError::DuplicateHandler {
            kind: kind.into(),
        }
    }

    fn create_directory(outbox: Outbox, source: io::Error) -> Self {
        DirectorError::CreateDirectory {
            outbox,
            source,
        }
    }

    fn file_with_ext(path: &Path, ext: &'static str) -> PathBuf {
        path.file_name()
            .map_or_else(|| ".".into(), |name| Path::new(name).with_extension(ext))
    }

    fn create_file(filepath: &Path, ext: &'static str, source: io::Error) -> Self {
        DirectorError::CreateFile {
            filename: Self::file_with_ext(filepath, ext),
            source,
        }
    }

    fn write_file(filepath: &Path, ext: &'static str, source: io::Error) -> Self {
        DirectorError::WriteFile {
            filename: Self::file_with_ext(filepath, ext),
            source,
        }
    }

    fn move_job(filepath: PathBuf, outbox: Outbox, source: io::Error) -> Self {
        DirectorError::MoveJob {
            outbox,
            filepath,
            source,
        }
    }

    fn list_queue(path: PathBuf, source: io::Error) -> Self {
        DirectorError::ListQueue {
            path,
            source,
        }
    }

    fn watch_queue(path: PathBuf, source: io::Error) -> Self {
        DirectorError::WatchQueue {
            path,
            source,
        }
    }

    fn open_job(filepath: PathBuf, source: io::Error) -> Self {
        DirectorError::OpenJob {
            filepath,
            source,
        }
    }

    fn create_defer_job(filepath: PathBuf, source: io::Error) -> Self {
        DirectorError::CreateDeferJob {
            filepath,
            source,
        }
    }

    fn write_defer_job(filepath: PathBuf, source: serde_json::Error) -> Self {
        DirectorError::WriteDeferJob {
            filepath,
            source,
        }
    }

    fn handler(source: Box<dyn error::Error + Send + Sync>) -> Self {
        DirectorError::Handler {
            source,
        }
    }
}

type DirectorResult<T> = Result<T, DirectorError>;

impl<'a> Director<'a> {
    /// Creates a new `Director`.
    ///
    /// The `accept` and `reject` directories are created as children of the given directory.
    pub fn new(root: &Path) -> DirectorResult<Self> {
        if !root.is_dir() {
            return Err(DirectorError::not_a_directory(root));
        }

        info!(target: "director", "setting up a director in {}", root.display());

        let accept_dir = root.join("accept");
        let reject_dir = root.join("reject");
        let fail_dir = root.join("fail");
        fs::create_dir_all(&accept_dir)
            .map_err(|err| DirectorError::create_directory(Outbox::Accept, err))?;
        fs::create_dir_all(&reject_dir)
            .map_err(|err| DirectorError::create_directory(Outbox::Reject, err))?;
        fs::create_dir_all(&fail_dir)
            .map_err(|err| DirectorError::create_directory(Outbox::Fail, err))?;

        Ok(Director {
            accept: accept_dir,
            reject: reject_dir,
            fail: fail_dir,

            handlers: HashMap::new(),
        })
    }

    /// Add a handler for jobs labeled as `kind`.
    pub fn add_handler<K>(&mut self, kind: K, handler: &'a dyn Handler) -> DirectorResult<()>
    where
        K: Into<String>,
    {
        match self.handlers.entry(kind.into()) {
            Entry::Occupied(o) => Err(DirectorError::duplicate_handler(o.key())),
            Entry::Vacant(v) => {
                debug!(target: "director", "adding handler '{}'", v.key());

                v.insert(handler);
                Ok(())
            },
        }
    }

    /// Move the job file into the appropriate directory.
    fn tag(&self, outbox: Outbox, file: &Path) -> DirectorResult<PathBuf> {
        let mut target_path = match outbox {
            Outbox::Accept => &self.accept,
            Outbox::Reject => &self.reject,
            Outbox::Fail => &self.fail,
        }
        .to_path_buf();
        target_path.push(
            file.file_name()
                .expect("expected the input file to have a file name"),
        );

        // Write the stamp file.
        let mut stamp_file = File::create(target_path.with_extension("stamp"))
            .map_err(|err| DirectorError::create_file(&target_path, "stamp", err))?;
        let time = Utc::now();
        writeln!(stamp_file, "{}", time)
            .map_err(|err| DirectorError::write_file(&target_path, "stamp", err))?;

        // Rename the file into the target path.
        fs::rename(file, &target_path)
            .map_err(|err| DirectorError::move_job(file.into(), outbox, err))?;

        Ok(target_path)
    }

    /// Tag a job file into a directory with a given reason.
    fn tag_with_reason<R>(&self, outbox: Outbox, file: &Path, reason: R) -> DirectorResult<()>
    where
        R: fmt::Display,
    {
        let target_file = self.tag(outbox, file)?;
        self.write_reason(&target_file, reason)
    }

    /// Write a reason file.
    fn write_reason<R>(&self, target_file: &Path, reason: R) -> DirectorResult<()>
    where
        R: fmt::Display,
    {
        let mut reason_file = File::create(target_file.with_extension("reason"))
            .map_err(|err| DirectorError::create_file(target_file, "reason", err))?;
        writeln!(reason_file, "{}", reason)
            .map_err(|err| DirectorError::write_file(target_file, "reason", err))?;

        Ok(())
    }

    /// Handle a single file.
    ///
    /// Returns `true` if the loop should terminate.
    pub fn run_one<P>(&self, path: P) -> RunResult
    where
        P: AsRef<Path>,
    {
        match self.dispatch(path.as_ref()) {
            Ok(res) => res,
            Err(err) => {
                error!("error when handling {}: {}", log_file_name(&path), err);
                RunResult::Done
            },
        }
    }

    /// Handle a batch of files at once.
    ///
    /// Returns `true` if the loop should terminate.
    pub fn run<I, P>(&self, paths: I) -> RunResult
    where
        I: IntoIterator<Item = P>,
        P: AsRef<Path>,
    {
        let mut result = RunResult::Continue;

        for path in paths {
            let one_result = self.run_one(path);
            if one_result > result {
                result = one_result;
            }

            if result.is_done() {
                break;
            }
        }

        result
    }

    /// Watch a directory for new job files and act upon them.
    ///
    /// Upon startup, this searches the directory for existing files. These files are processed in
    /// lexigraphic order.
    pub fn watch_directory(&self, path: &Path) -> DirectorResult<RunResult> {
        let mut watcher = {
            let paths = path
                .read_dir()
                .map_err(|err| DirectorError::list_queue(path.into(), err))?;
            let watcher =
                Watcher::new(path).map_err(|err| DirectorError::watch_queue(path.into(), err))?;

            let path_entries = paths
                .filter_map(|e| e.ok())
                .filter_map(|e| {
                    if e.path().is_dir() {
                        None
                    } else {
                        Some(e.path())
                    }
                })
                .sorted();
            let result = self.run(path_entries);
            if result.should_exit() {
                return Ok(result);
            }

            watcher
        };

        let loop_result;

        loop {
            let events = watcher
                .events()
                .map_err(|err| DirectorError::watch_queue(path.into(), err))?;

            let path_entries = events
                .filter_map(|event| {
                    if event.mask.contains(EventMask::ISDIR) {
                        None
                    } else {
                        event.name.map(|name| path.join(name))
                    }
                })
                .sorted();
            let result = self.run(path_entries);
            if result.should_exit() {
                loop_result = result;
                break;
            }
        }

        Ok(loop_result)
    }

    /// Takes the file and directs it to the right location.
    fn dispatch(&self, file: &Path) -> DirectorResult<RunResult> {
        trace!(target: "director", "handling file {}", log_file_name(&file));

        let ext = file.extension().map(OsStr::to_string_lossy);
        let ext_str = ext.as_ref().map(AsRef::as_ref);

        if let Some("json") = ext_str {
        } else {
            return Ok(RunResult::Continue);
        }

        let event_file =
            File::open(file).map_err(|err| DirectorError::open_job(file.into(), err))?;
        let mut payload: Value = match serde_json::from_reader(event_file) {
            Ok(payload) => payload,
            Err(err) => {
                info!("failed to read JSON from {}: {}", file.display(), err);

                return Ok(RunResult::Continue);
            },
        };

        if !payload.is_object() {
            return Ok(RunResult::Continue);
        }

        let ret = match self.handle(&payload)? {
            HandlerResult::Accept => {
                debug!("accepted {}", log_file_name(&file));
                self.tag(Outbox::Accept, file)?;
                RunResult::Continue
            },
            HandlerResult::Defer(ref reason) => {
                debug!("deferring {}: {}", log_file_name(&file), reason);
                let rndpart = rand::thread_rng()
                    .sample_iter(&rand::distributions::Alphanumeric)
                    .take(12)
                    .collect::<String>();
                let defer_job_file =
                    file.with_file_name(format!("{}-{}.json", Utc::now().to_rfc3339(), rndpart));
                Self::add_reason_to_object(&mut payload, file.to_string_lossy(), reason);
                let mut defer_file = File::create(&defer_job_file).map_err(|err| {
                    // XXX(nightly): Result::unwrap_or_else API (rust-lang/rust#53268)
                    DirectorError::create_defer_job(defer_job_file.clone(), err)
                })?;
                serde_json::to_writer(&mut defer_file, &payload)
                    .map_err(|err| DirectorError::write_defer_job(defer_job_file, err))?;
                self.tag_with_reason(Outbox::Reject, file, reason)?;
                RunResult::Continue
            },
            HandlerResult::Reject(ref reason) => {
                debug!("rejecting {}: {}", file.display(), reason);
                self.tag_with_reason(Outbox::Reject, file, reason)?;
                RunResult::Continue
            },
            HandlerResult::Fail(ref reason) => {
                debug!("failed {}: {:?}", file.display(), reason);
                self.tag_with_reason(Outbox::Fail, file, format!("{:?}", reason))?;
                RunResult::Continue
            },
            HandlerResult::Restart => {
                info!(target: "director", "restarting via {}", log_file_name(&file));

                self.tag(Outbox::Accept, file)?;
                RunResult::Restart
            },
            HandlerResult::Done => {
                info!(target: "director", "completed via {}", log_file_name(&file));

                self.tag(Outbox::Accept, file)?;
                RunResult::Done
            },
        };

        trace!(target: "director", "handled file {}", log_file_name(&file));

        Ok(ret)
    }

    /// Add a reason to a JSON job object.
    fn add_reason_to_object<N, R>(object: &mut Value, name: N, reason: R)
    where
        N: Into<String>,
        R: Into<String>,
    {
        let retry_map = object
            .as_object_mut()
            .expect("expected an object; internal logic failure")
            .entry("retry")
            .or_insert_with(|| Value::Object(Map::new()));

        if !retry_map.is_object() {
            *retry_map = Value::Object(Map::new());
        }

        retry_map
            .as_object_mut()
            .expect("expected an object; internal logic failure") // Verified above.
            .insert(name.into(), Value::String(reason.into()));
    }

    /// Dispatch the contents of the file to the right handler.
    fn handle(&self, payload: &Value) -> DirectorResult<HandlerResult> {
        let kind = match payload.pointer("/kind") {
            Some(&Value::String(ref kind)) => kind,
            Some(_) => return Ok(HandlerResult::reject("'kind' is not a string")),
            None => return Ok(HandlerResult::reject("no 'kind'")),
        };
        let data = match payload.pointer("/data") {
            Some(data) => data,
            None => return Ok(HandlerResult::reject("no 'data'")),
        };
        let retry_reasons = match payload.pointer("/retry") {
            Some(&Value::Object(ref reasons)) => {
                let retry_reasons = reasons
                    .iter()
                    .map(|(_, reason)| {
                        reason
                            .as_str()
                            .map(Into::into)
                            .ok_or_else(|| HandlerResult::reject("retry reason is not a string"))
                    })
                    .collect::<::std::result::Result<Vec<_>, HandlerResult>>();

                match retry_reasons {
                    Ok(reasons) => reasons,
                    Err(reject) => return Ok(reject),
                }
            },
            Some(_) => return Ok(HandlerResult::reject("'retry' is not an object")),
            None => vec![],
        };

        match self.handlers.get(kind) {
            Some(handler) => {
                handler
                    .handle_retry(kind, data, retry_reasons)
                    .map_err(DirectorError::handler)
            },
            None => {
                Ok(HandlerResult::Reject(format!(
                    "no handler for kind '{}'",
                    kind,
                )))
            },
        }
    }
}