openquiver-runtime 0.1.0

//! What goes in the journal, and how to read it back.
//!
//! Writing and reading are one fact in two directions, so they are written
//! together. A record that encodes one way and decodes another is not a bug that
//! shows up in a test — it shows up when someone resumes a run from yesterday.
//!
//! The vocabulary lives here rather than in `openquiver-store` on purpose: the store
//! deliberately does not know what an event means, which is what lets this set
//! grow without a schema migration each time.

use std::collections::BTreeMap;

use openquiver_graph::NodeName;
use serde::{Deserialize, Serialize};

use crate::RunValue;

pub const RUN_STARTED: &str = "run.started";
pub const NODE_ENTERED: &str = "node.entered";
pub const NODE_EXITED: &str = "node.exited";
pub const PRINT: &str = "print";
pub const RUN_FINISHED: &str = "run.finished";

/// One thing a run did.
#[derive(Debug, Clone, PartialEq)]
pub(crate) enum Record {
    Started(Started),
    /// Written before the node touches anything, and paired with
    /// [`Record::Exited`] afterwards. A node with one and not the other is the
    /// only state a resumed run cannot decide by itself.
    Entered(Entered),
    Exited(Exited),
    Printed(Printed),
    Finished,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub(crate) struct Started {
    /// The fingerprint of the plan this run walks.
    ///
    /// The workflow's *name* is on the run row in the store and is not repeated
    /// here. What is recorded is the thing a name cannot tell you: whether the
    /// graph is still the one the run started on.
    pub plan: String,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub(crate) struct Entered {
    pub node: NodeName,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub(crate) struct Exited {
    pub node: NodeName,
    /// What the node put on its output pins.
    pub outputs: BTreeMap<String, RunValue>,
    /// Which execution output continues. `None` ends this line of execution.
    pub exec: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub(crate) struct Printed {
    pub text: String,
}

/// A journal entry that could not be understood.
#[derive(Debug, thiserror::Error)]
pub enum JournalError {
    #[error("this journal has a `{0}` entry, which this version does not understand")]
    UnknownKind(String),
    #[error("a `{kind}` entry could not be read: {source}")]
    Malformed {
        kind: String,
        source: serde_json::Error,
    },
}

impl Record {
    pub(crate) fn kind(&self) -> &'static str {
        match self {
            Record::Started(_) => RUN_STARTED,
            Record::Entered(_) => NODE_ENTERED,
            Record::Exited(_) => NODE_EXITED,
            Record::Printed(_) => PRINT,
            Record::Finished => RUN_FINISHED,
        }
    }

    pub(crate) fn payload(&self) -> Result<serde_json::Value, serde_json::Error> {
        match self {
            Record::Started(it) => serde_json::to_value(it),
            Record::Entered(it) => serde_json::to_value(it),
            Record::Exited(it) => serde_json::to_value(it),
            Record::Printed(it) => serde_json::to_value(it),
            Record::Finished => Ok(serde_json::Value::Object(Default::default())),
        }
    }

    /// Reads one entry back.
    ///
    /// An unrecognised kind is refused rather than skipped. Skipping would mean
    /// replaying a journal minus the part this build does not know about, and
    /// silently continuing from a state that never existed is worse than saying
    /// the journal is too new.
    pub(crate) fn decode(kind: &str, payload: &serde_json::Value) -> Result<Self, JournalError> {
        fn read<T: serde::de::DeserializeOwned>(
            kind: &str,
            payload: &serde_json::Value,
        ) -> Result<T, JournalError> {
            serde_json::from_value(payload.clone()).map_err(|source| JournalError::Malformed {
                kind: kind.to_owned(),
                source,
            })
        }

        Ok(match kind {
            RUN_STARTED => Record::Started(read(kind, payload)?),
            NODE_ENTERED => Record::Entered(read(kind, payload)?),
            NODE_EXITED => Record::Exited(read(kind, payload)?),
            PRINT => Record::Printed(read(kind, payload)?),
            RUN_FINISHED => Record::Finished,
            other => return Err(JournalError::UnknownKind(other.to_owned())),
        })
    }
}