From f2c40ad35234b1a46dda8cdc8ce0f197a2e9eeae Mon Sep 17 00:00:00 2001
From: Ross Jones <ross@register-dynamics.co.uk>
Date: Fri, 12 Jan 2024 12:43:46 +0000
Subject: [PATCH 1/4] Adds bacalhau support

Adds a destination node for bacalhau that is able to submit jobs to a
local requester node, using the orchestrator API.

Each instance of the destination node is configured with the location of
the jobstore, a local directory containing .yaml files describing jobs.
It should also be provided with the job name (one of the .yaml files)
so that it might be configured with the following

```
jobstore: /tmp/jobstore
job: process
```

and will load /tmp/jobstore/process.yaml as the job specification.  Once
loaded the job specification can be templated using handlebars syntax,
so that {{key}} will be replaced with values from the 'key' field in the
input message.
---
 .gitignore                                    |   2 +
 Cargo.lock                                    | 101 +++++++++++++
 Cargo.toml                                    |   1 +
 common/src/lib.rs                             |   9 ++
 console/components/Flow/index.tsx             |  21 +--
 console/components/nodes.tsx                  | 106 +++++++++++++-
 myceliald/Cargo.toml                          |   7 +-
 myceliald/config.example.toml                 |   7 +
 myceliald/src/constructors/bacalhau.rs        |  28 ++++
 myceliald/src/constructors/mod.rs             |   1 +
 myceliald/src/runtime.rs                      |   4 +
 .../section/section_impls/bacalhau/Cargo.toml |  20 +++
 .../section/section_impls/bacalhau/src/api.rs |  49 +++++++
 .../section_impls/bacalhau/src/destination.rs | 136 ++++++++++++++++++
 .../section_impls/bacalhau/src/jobstore.rs    | 103 +++++++++++++
 .../section/section_impls/bacalhau/src/lib.rs |  69 +++++++++
 .../bacalhau/testdata/process.yaml            |  10 ++
 17 files changed, 659 insertions(+), 15 deletions(-)
 create mode 100644 myceliald/src/constructors/bacalhau.rs
 create mode 100644 pipe/section/section_impls/bacalhau/Cargo.toml
 create mode 100644 pipe/section/section_impls/bacalhau/src/api.rs
 create mode 100644 pipe/section/section_impls/bacalhau/src/destination.rs
 create mode 100644 pipe/section/section_impls/bacalhau/src/jobstore.rs
 create mode 100644 pipe/section/section_impls/bacalhau/src/lib.rs
 create mode 100644 pipe/section/section_impls/bacalhau/testdata/process.yaml

diff --git a/.gitignore b/.gitignore
index 6985cf1b..5ea5efc1 100644
--- a/.gitignore
+++ b/.gitignore
@@ -12,3 +12,5 @@ Cargo.lock
 
 # MSVC Windows builds of rustc generate these, which store debugging information
 *.pdb
+
+.vscode/
diff --git a/Cargo.lock b/Cargo.lock
index f412dbf1..371a2b89 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -702,6 +702,22 @@ dependencies = [
  "tower-service",
 ]
 
+[[package]]
+name = "bacalhau"
+version = "0.1.0"
+dependencies = [
+ "handlebars",
+ "reqwest",
+ "section",
+ "serde",
+ "serde_json",
+ "serde_yaml",
+ "stub",
+ "tokio",
+ "tokio-stream",
+ "tokio-util",
+]
+
 [[package]]
 name = "backtrace"
 version = "0.3.69"
@@ -1547,6 +1563,20 @@ dependencies = [
  "num-traits",
 ]
 
+[[package]]
+name = "handlebars"
+version = "4.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "faa67bab9ff362228eb3d00bd024a4965d8231bbb7921167f0cfa66c6626b225"
+dependencies = [
+ "log",
+ "pest",
+ "pest_derive",
+ "serde",
+ "serde_json",
+ "thiserror",
+]
+
 [[package]]
 name = "hashbrown"
 version = "0.12.3"
@@ -2175,6 +2205,7 @@ name = "myceliald"
 version = "0.2.0"
 dependencies = [
  "anyhow",
+ "bacalhau",
  "base64 0.21.5",
  "clap",
  "common",
@@ -2602,6 +2633,51 @@ version = "2.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e"
 
+[[package]]
+name = "pest"
+version = "2.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1f200d8d83c44a45b21764d1916299752ca035d15ecd46faca3e9a2a2bf6ad06"
+dependencies = [
+ "memchr",
+ "thiserror",
+ "ucd-trie",
+]
+
+[[package]]
+name = "pest_derive"
+version = "2.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bcd6ab1236bbdb3a49027e920e693192ebfe8913f6d60e294de57463a493cfde"
+dependencies = [
+ "pest",
+ "pest_generator",
+]
+
+[[package]]
+name = "pest_generator"
+version = "2.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2a31940305ffc96863a735bef7c7994a00b325a7138fdbc5bda0f1a0476d3275"
+dependencies = [
+ "pest",
+ "pest_meta",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.48",
+]
+
+[[package]]
+name = "pest_meta"
+version = "2.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a7ff62f5259e53b78d1af898941cdcdccfae7385cf7d793a6e55de5d05bb4b7d"
+dependencies = [
+ "once_cell",
+ "pest",
+ "sha2",
+]
+
 [[package]]
 name = "pin-project"
 version = "1.1.3"
@@ -3392,6 +3468,19 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "serde_yaml"
+version = "0.9.30"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b1bf28c79a99f70ee1f1d83d10c875d2e70618417fda01ad1785e027579d9d38"
+dependencies = [
+ "indexmap 2.1.0",
+ "itoa",
+ "ryu",
+ "serde",
+ "unsafe-libyaml",
+]
+
 [[package]]
 name = "server"
 version = "0.2.0"
@@ -4321,6 +4410,12 @@ version = "1.17.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "42ff0bf0c66b8238c6f3b578df37d0b7848e55df8577b3f74f92a69acceeb825"
 
+[[package]]
+name = "ucd-trie"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed646292ffc8188ef8ea4d1e0e0150fb15a5c2e12ad9b8fc191ae7a8a7f3c4b9"
+
 [[package]]
 name = "unicase"
 version = "2.7.0"
@@ -4369,6 +4464,12 @@ version = "0.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39ec24b3121d976906ece63c9daad25b85969647682eee313cb5779fdd69e14e"
 
+[[package]]
+name = "unsafe-libyaml"
+version = "0.2.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ab4c90930b95a82d00dc9e9ac071b4991924390d46cbd0dfe566148667605e4b"
+
 [[package]]
 name = "untrusted"
 version = "0.7.1"
diff --git a/Cargo.toml b/Cargo.toml
index 2d3cf265..b30f5b34 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -6,6 +6,7 @@ members = [
     "pipe/runtime",
     "pipe/arrow_msg",
     "pipe/section",
+    "pipe/section/section_impls/bacalhau",
     "pipe/section/section_impls/hello_world",
     "pipe/section/section_impls/sqlite_connector",
     "pipe/section/section_impls/excel_connector",
diff --git a/common/src/lib.rs b/common/src/lib.rs
index de7e9bfb..12192772 100644
--- a/common/src/lib.rs
+++ b/common/src/lib.rs
@@ -57,6 +57,7 @@ pub enum Destination {
     Sqlite_Connector(SqliteDestinationConfig),
     Snowflake(SnowflakeDestinationConfig),
     Sqlite_Physical_Replication(SqlitePhysicalReplicationDestinationConfig),
+    Bacalhau(BacalhauDestinationConfig),
     Hello_World(HelloWorldDestinationConfig),
     Kafka(KafkaDestinationConfig),
     Postgres_Connector(PostgresConnectorDestinationConfig),
@@ -149,6 +150,14 @@ pub struct SqlitePhysicalReplicationSourceConfig {
     // database path
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, Default)]
+pub struct BacalhauDestinationConfig {
+    #[serde(flatten)]
+    pub common_attrs: CommonAttrs,
+    pub job: String,
+    pub jobstore: String,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, Default)]
 pub struct HelloWorldSourceConfig {
     #[serde(flatten)]
diff --git a/console/components/Flow/index.tsx b/console/components/Flow/index.tsx
index 88c1f0fe..aed0a4ef 100644
--- a/console/components/Flow/index.tsx
+++ b/console/components/Flow/index.tsx
@@ -40,6 +40,7 @@ import {
   KafkaSourceNode,
   SnowflakeSourceNode,
   SnowflakeDestinationNode,
+  BacalhauDestinationNode,
   HelloWorldSourceNode,
   HelloWorldDestinationNode,
   ExcelSourceNode,
@@ -75,11 +76,10 @@ const useStyles = createStyles((theme) => ({
     marginBottom: theme.spacing.md,
 
     "&:not(:last-of-type)": {
-      borderBottom: `${rem(1)} solid ${
-        theme.colorScheme === "dark"
-          ? theme.colors.dark[4]
-          : theme.colors.gray[3]
-      }`,
+      borderBottom: `${rem(1)} solid ${theme.colorScheme === "dark"
+        ? theme.colors.dark[4]
+        : theme.colors.gray[3]
+        }`,
     },
   },
 
@@ -186,6 +186,7 @@ const nodeTypes = {
   snowflake_destination: SnowflakeDestinationNode,
   hello_world_source: HelloWorldSourceNode,
   hello_world_destination: HelloWorldDestinationNode,
+  bacalhau_destination: BacalhauDestinationNode,
   postgres_connector_source: PostgresConnectorSourceNode,
   postgres_connector_destination: PostgresConnectorDestinationNode,
   mysql_connector_destination: MysqlConnectorDestinationNode,
@@ -286,7 +287,7 @@ function NavbarSearch(props: NavbarSearchProps) {
         onDragStart={(event) => onDragStart(event, null, source, source)}
         draggable
       >
-        Mycelial Server 
+        Mycelial Server
       </div>,
     ];
   };
@@ -345,7 +346,7 @@ async function getConfigs(token: string) {
     });
     const result = await response.json();
     return result;
-  } catch (error) {}
+  } catch (error) { }
 }
 
 const dagreGraph = new dagre.graphlib.Graph();
@@ -568,7 +569,7 @@ function Flow() {
 
       if (edge.data?.id) {
         let payload = {
-          configs: [{id: edge.data.id, pipe: pipe}]
+          configs: [{ id: edge.data.id, pipe: pipe }]
         }
         try {
           const response = await fetch("/api/pipe", {
@@ -586,7 +587,7 @@ function Flow() {
         }
       } else {
         let payload = {
-          configs: [{pipe: pipe}]
+          configs: [{ pipe: pipe }]
         }
         const response = await fetch("/api/pipe", {
           method: "POST",
@@ -609,7 +610,7 @@ function Flow() {
                   id: id,
                 },
               }
-            } 
+            }
             return ed;
           });
         })
diff --git a/console/components/nodes.tsx b/console/components/nodes.tsx
index 49b91294..f7bf9c73 100644
--- a/console/components/nodes.tsx
+++ b/console/components/nodes.tsx
@@ -16,6 +16,107 @@ import styles from "@/components/Flow/Flow.module.css";
 import { ClientContext } from "./context/clientContext";
 import { ClientContextType } from "./@types/client";
 
+
+const BacalhauDestinationNode: FC<NodeProps> = memo(({ id, data, selected }) => {
+  const instance = useReactFlow();
+  const { clients } = useContext(ClientContext) as ClientContextType;
+
+  let initialValues = useMemo(() => {
+    return {
+      client: data.client ? data.client : "-",
+      job: data.job ? data.job : "job",
+      jobstore: data.jobstore ? data.jobstore : "jobstore",
+    };
+  }, []);
+
+  const handleChange = useCallback((name: string, value: string) => {
+    instance.setNodes((nodes) =>
+      nodes.map((node) => {
+        if (node.id === id) {
+          node.data = {
+            ...node.data,
+            [name]: value,
+          };
+        }
+
+        return node;
+      }),
+    );
+  }, []);
+
+  useEffect(() => {
+    handleChange("client", initialValues.client);
+    handleChange("job", initialValues.job);
+    handleChange("jobstore", initialValues.jobstore);
+  }, []);
+
+  let classNames = `${styles.customNode} `;
+  if (selected) {
+    classNames = classNames + `${styles.selected}`;
+  }
+
+  const removeNode = useCallback((id: string) => {
+    const node = instance.getNode(id);
+    if (node === undefined) {
+      return;
+    }
+    instance.deleteElements({
+      edges: getConnectedEdges([node], []),
+      nodes: [node],
+    });
+  }, []);
+
+  return (
+    <div className={classNames}>
+      <div className=" grid grid-cols-1 gap-x-6 gap-y-2">
+        <h2 className="text-slate-400 font-normal">HelloWorld Destination</h2>
+        <button
+          onClick={() => {
+            if (confirm("Are you sure you want to delete this node?")) {
+              removeNode(id);
+            }
+          }}
+          type="button"
+          className="absolute right-1 top-1 rounded bg-red-200 text-white shadow-sm hover:bg-red-600 focus-visible:outline focus-visible:outline-2 focus-visible:outline-offset-2 focus-visible:outline-red-800"
+          title="delete"
+        >
+          <XMarkIcon className="h-5 w-5" aria-hidden="true" />
+        </button>
+        <Select
+          name="client"
+          label="Client"
+          placeholder="Pick one"
+          defaultValue={initialValues.client}
+          options={(clients || []).map((c) => c.id)}
+          onChange={(value) => {
+            handleChange("client", value || "");
+          }}
+        />
+        <TextInput
+          name="jobstore"
+          label="Job Store"
+          placeholder={initialValues.jobstore}
+          defaultValue={initialValues.jobstore}
+          onChange={(event) => {
+            handleChange("jobstore", event.currentTarget.value)
+          }}
+        />
+        <TextInput
+          name="job"
+          label="Job"
+          placeholder={initialValues.job}
+          defaultValue={initialValues.job}
+          onChange={(event) => {
+            handleChange("job", event.currentTarget.value)
+          }}
+        />
+
+        <Handle type="target" position={Position.Left} id={id} />
+      </div>
+    </div>
+  );
+});
+
 const HelloWorldDestinationNode: FC<NodeProps> = memo(({ id, data, selected }) => {
   const instance = useReactFlow();
   const { clients } = useContext(ClientContext) as ClientContextType;
@@ -1401,7 +1502,7 @@ const PostgresConnectorDestinationNode: FC<NodeProps> = memo(({ id, data, select
 
   let initialValues = useMemo(() => {
     return {
-      url: data.url? data.url: "postgres://root:root@127.0.0.1:5432/test",
+      url: data.url ? data.url : "postgres://root:root@127.0.0.1:5432/test",
       client: data.client ? data.client : "-",
     };
   }, []);
@@ -1485,7 +1586,7 @@ const MysqlConnectorDestinationNode: FC<NodeProps> = memo(({ id, data, selected
 
   let initialValues = useMemo(() => {
     return {
-      url: data.url? data.url: "mysql://root:root@127.0.0.1:3306/test",
+      url: data.url ? data.url : "mysql://root:root@127.0.0.1:3306/test",
       client: data.client ? data.client : "-",
     };
   }, []);
@@ -1745,6 +1846,7 @@ export {
   KafkaDestination,
   SnowflakeSourceNode,
   SnowflakeDestinationNode,
+  BacalhauDestinationNode,
   HelloWorldSourceNode,
   HelloWorldDestinationNode,
   PostgresConnectorSourceNode,
diff --git a/myceliald/Cargo.toml b/myceliald/Cargo.toml
index 5c753ad0..56bd287f 100644
--- a/myceliald/Cargo.toml
+++ b/myceliald/Cargo.toml
@@ -11,18 +11,19 @@ clap = { version = "4", features = ["derive", "env"] }
 reqwest = { version = "0.11", features = ["json"] }
 anyhow = "1"
 base64 = { version = "0.21" }
-serde = { version = "1", features = ["derive"]}
+serde = { version = "1", features = ["derive"] }
 serde_json = "1"
-sqlx = { version = "0.7", features = ["sqlite"]}
+sqlx = { version = "0.7", features = ["sqlite"] }
 log = "0.4"
 pretty_env_logger = "0.5"
 toml = "0.7"
 common = { path = "../common" }
 thiserror = "1"
-pipe = { version = "0.3", path = "../pipe/runtime/", package="runtime" }
+pipe = { version = "0.3", path = "../pipe/runtime/", package = "runtime" }
 section = { version = "0.3", path = "../pipe/section/" }
 
 ## sections
+bacalhau = { path = "../pipe/section/section_impls/bacalhau/" }
 hello_world = { path = "../pipe/section/section_impls/hello_world/" }
 sqlite_connector = { path = "../pipe/section/section_impls/sqlite_connector/" }
 excel_connector = { path = "../pipe/section/section_impls/excel_connector/" }
diff --git a/myceliald/config.example.toml b/myceliald/config.example.toml
index c9a52ca5..91166c29 100644
--- a/myceliald/config.example.toml
+++ b/myceliald/config.example.toml
@@ -80,6 +80,13 @@ path = "Cargo.toml"
 # Define all data DESTINATIONS (data stores and directory paths)
 # accessible by the Node that should be exposed as DESTINATIONS to the Sqlite Physical Replication Server
 
+
+[[destinations]]
+display_name = "Bacalhau"
+jobstore = "c:/temp/jobstore"
+type = "bacalhau"
+job = "jobname"
+
 [[destinations]]
 type = "sqlite_physical_replication"
 display_name = "Sqlite Physical Replication Movie"
diff --git a/myceliald/src/constructors/bacalhau.rs b/myceliald/src/constructors/bacalhau.rs
new file mode 100644
index 00000000..a63f1cf8
--- /dev/null
+++ b/myceliald/src/constructors/bacalhau.rs
@@ -0,0 +1,28 @@
+use std::path::Path;
+
+use pipe::{config::Map, types::DynSection};
+use section::{command_channel::SectionChannel, SectionError};
+
+pub fn destination_ctor<S: SectionChannel>(
+    config: &Map,
+) -> Result<Box<dyn DynSection<S>>, SectionError> {
+    let job = config
+        .get("job")
+        .ok_or("bacalhau section requires 'job'")?
+        .as_str()
+        .ok_or("'job' should be a string")?;
+    let jobstore = config
+        .get("jobstore")
+        .ok_or("bacalhau section requires 'jobstore'")?
+        .as_str()
+        .ok_or("'jobstore' should be a string")?;
+
+    // Verify jobstore folder exists
+    if !Path::new(jobstore).exists() {
+        return Err(format!("'jobstore' path '{}' does not exist", jobstore).into());
+    }
+
+    Ok(Box::new(bacalhau::destination::Bacalhau::new(
+        job, jobstore,
+    )))
+}
diff --git a/myceliald/src/constructors/mod.rs b/myceliald/src/constructors/mod.rs
index b1e7d90d..c4ef9a37 100644
--- a/myceliald/src/constructors/mod.rs
+++ b/myceliald/src/constructors/mod.rs
@@ -1,3 +1,4 @@
+pub mod bacalhau;
 pub mod excel_connector;
 pub mod file;
 pub mod hello_world;
diff --git a/myceliald/src/runtime.rs b/myceliald/src/runtime.rs
index a38c74c5..a775a017 100644
--- a/myceliald/src/runtime.rs
+++ b/myceliald/src/runtime.rs
@@ -15,6 +15,10 @@ use section::command_channel::SectionChannel;
 /// Setup & populate registry
 fn setup_registry<S: SectionChannel>() -> Registry<S> {
     let arr: &[(&str, Constructor<S>)] = &[
+        (
+            "bacalhau_destination",
+            constructors::bacalhau::destination_ctor,
+        ),
         (
             "hello_world_destination",
             constructors::hello_world::destination_ctor,
diff --git a/pipe/section/section_impls/bacalhau/Cargo.toml b/pipe/section/section_impls/bacalhau/Cargo.toml
new file mode 100644
index 00000000..4ca6654f
--- /dev/null
+++ b/pipe/section/section_impls/bacalhau/Cargo.toml
@@ -0,0 +1,20 @@
+[package]
+name = "bacalhau"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+section = { version = "0.3", path = "../../../section/" }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+reqwest = { version = "0.11.22", features = ["json"] }
+tokio = { version = "1", features = ["full"] }
+tokio-stream = "0.1.14"
+tokio-util = "0.7.10"
+handlebars = "4.4"
+serde_yaml = "0.9"
+
+[dev-dependencies]
+stub = { path = "../../section_impls/stub/" }
diff --git a/pipe/section/section_impls/bacalhau/src/api.rs b/pipe/section/section_impls/bacalhau/src/api.rs
new file mode 100644
index 00000000..0f9a13c7
--- /dev/null
+++ b/pipe/section/section_impls/bacalhau/src/api.rs
@@ -0,0 +1,49 @@
+use crate::StdError;
+use reqwest;
+use serde::Deserialize;
+
+#[derive(Deserialize)]
+struct SubmitErrorResponse {
+    error: String,
+    #[allow(dead_code)]
+    message: String,
+}
+
+#[derive(Deserialize)]
+struct SubmitResponse {
+    #[serde(rename = "JobID")]
+    job_id: String,
+
+    #[serde(rename = "EvaluationID")]
+    _evaluation_id: String,
+
+    #[serde(rename = "Warnings")]
+    _warnings: Option<Vec<String>>,
+}
+
+pub async fn submit(body: &str) -> Result<String, StdError> {
+    // For now we'll assume a local instance and using the newer version
+    // of the jobs API.
+    let api_url = "http://127.0.0.1:20000/api/v1/orchestrator/jobs";
+    let client = reqwest::Client::new();
+    let response = client
+        .post(api_url)
+        .header("Content-Type", "application/json")
+        .body(body.to_owned())
+        .send()
+        .await?;
+
+    let success = &response.status().is_success();
+    let resp_body = &response.bytes().await?;
+
+    match success {
+        true => {
+            let sresp: SubmitResponse = serde_json::from_slice(resp_body.as_ref()).unwrap();
+            Ok(sresp.job_id)
+        }
+        _ => {
+            let err: SubmitErrorResponse = serde_json::from_slice(resp_body.as_ref()).unwrap();
+            Err(err.error.into())
+        }
+    }
+}
diff --git a/pipe/section/section_impls/bacalhau/src/destination.rs b/pipe/section/section_impls/bacalhau/src/destination.rs
new file mode 100644
index 00000000..d2c463eb
--- /dev/null
+++ b/pipe/section/section_impls/bacalhau/src/destination.rs
@@ -0,0 +1,136 @@
+use crate::jobstore::JobStore;
+use crate::{api::submit, BacalhauPayload};
+
+use section::message::ValueView;
+use section::{
+    command_channel::{Command, SectionChannel},
+    futures::{self, FutureExt, Sink, SinkExt, Stream, StreamExt},
+    message::Chunk,
+    pretty_print::pretty_print,
+    section::Section,
+    SectionError, SectionMessage,
+};
+use std::collections::HashMap;
+use std::pin::{pin, Pin};
+
+use std::future::Future;
+
+type StdError = Box<dyn std::error::Error + Send + Sync + 'static>;
+
+#[derive(Debug)]
+pub struct Bacalhau {
+    pub job: String,
+    pub jobstore: JobStore,
+}
+
+impl Bacalhau {
+    pub fn new(job: impl Into<String>, jobstore: impl Into<String>) -> Self {
+        Self {
+            job: job.into(),
+            jobstore: JobStore::new(jobstore).expect("should be able to create jobstore"),
+        }
+    }
+
+    pub async fn submit_job(&self, data: &HashMap<String, String>) -> Result<String, StdError> {
+        // We'll either get an Err from the call to render, or we'll get either
+        // an Ok or Err from the call to submit.
+        match self.jobstore.render(self.job.clone(), data) {
+            Ok(output) => submit(&output).await,
+            Err(m) => Err(m),
+        }
+    }
+
+    async fn enter_loop<Input, Output, SectionChan>(
+        self,
+        input: Input,
+        _output: Output,
+        mut section_channel: SectionChan,
+    ) -> Result<(), SectionError>
+    where
+        Input: Stream<Item = SectionMessage> + Send + 'static,
+        Output: Sink<SectionMessage, Error = SectionError> + Send + 'static,
+        SectionChan: SectionChannel + Send + 'static,
+    {
+        let mut input = pin!(input.fuse());
+        //let output = pin!(output);
+        loop {
+            futures::select_biased! {
+                cmd = section_channel.recv().fuse() => {
+                    if let Command::Stop = cmd? {
+                        return Ok(())
+                    }
+                }
+                stream = input.next() => {
+                    let mut msg = match stream {
+                        Some(msg) => msg,
+                        None => Err("input stream closed")?
+                    };
+
+
+                    loop {
+                        futures::select! {
+                            chunk = msg.next().fuse() => {
+                                let frame = match chunk? {
+                                    None => break,
+                                    Some(Chunk::DataFrame(df)) => df,
+                                    Some(_ch) => continue,
+                                };
+
+
+                                let columns = frame.columns();
+
+
+                                section_channel.log(format!("got dataframe chunk from {}:\n{}", msg.origin(), pretty_print(&*frame))).await?;
+
+                                let mut c = columns.into_iter().nth(0).unwrap();
+                                match c.nth(0).unwrap() {
+                                    ValueView::Str(s) => {
+                                        section_channel.log(format!("got string: {}", s)).await?;
+                                    }
+                                    _ => continue,
+                                }
+
+
+                                //ValueView::Str(child.as_string::<i32>().value(value_offset)),
+                                let args: HashMap<String, String> = HashMap::new();
+                                //args.insert("target".into(), "".into());
+                                self.submit_job(&args).await?;
+
+                                msg.ack().await;
+                            }
+                        }
+
+                    }
+
+                    // // let payload: crate::BacalhauPayload = msg.into();
+                    // // let origin = &msg.origin();
+                    // // self.submit_job(&payload.message).await?;
+
+                    // // section_channel.log(&format!("Message from '{:?}' received! {:?}", origin, payload)).await?;
+                    // output.send(msg).await?;
+                    //         }
+                    //         cmd = section_channel.recv().fuse() => {
+                    //             if let Command::Stop = cmd? {
+                    //                 return Ok(())
+                    //             }
+                    //         }
+
+                },
+            }
+        }
+    }
+}
+
+impl<Input, Output, SectionChan> Section<Input, Output, SectionChan> for Bacalhau
+where
+    Input: Stream<Item = SectionMessage> + Send + 'static,
+    Output: Sink<SectionMessage, Error = SectionError> + Send + 'static,
+    SectionChan: SectionChannel + Send + 'static,
+{
+    type Error = SectionError;
+    type Future = Pin<Box<dyn Future<Output = Result<(), Self::Error>> + Send + 'static>>;
+
+    fn start(self, input: Input, output: Output, command: SectionChan) -> Self::Future {
+        Box::pin(async move { self.enter_loop(input, output, command).await })
+    }
+}
diff --git a/pipe/section/section_impls/bacalhau/src/jobstore.rs b/pipe/section/section_impls/bacalhau/src/jobstore.rs
new file mode 100644
index 00000000..da7c0d37
--- /dev/null
+++ b/pipe/section/section_impls/bacalhau/src/jobstore.rs
@@ -0,0 +1,103 @@
+use crate::StdError;
+use std::{collections::HashMap, fs, path::PathBuf};
+
+use handlebars::Handlebars;
+use serde::Serialize;
+use serde_json::{json, Map};
+use serde_yaml::Value;
+
+pub struct JobStore {
+    templates: Handlebars<'static>,
+}
+
+impl std::fmt::Debug for JobStore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("JobStore").finish()
+    }
+}
+
+impl JobStore {
+    pub fn new(jobstore: impl Into<String>) -> Result<Self, StdError> {
+        let path = PathBuf::from(jobstore.into());
+        let entries = path.read_dir()?;
+
+        let mut handlebars = Handlebars::new();
+        for e in entries {
+            let entry = e?;
+            let meta = entry.metadata()?;
+            if !meta.is_file() {
+                continue;
+            }
+
+            let entryname = entry.file_name();
+            let fname = entryname.to_str().unwrap();
+            if fname.ends_with("yaml") || fname.ends_with("yml") {
+                let p = entry.path();
+                let name = p.file_stem().unwrap();
+
+                handlebars.register_template_string(
+                    name.to_str().unwrap(),
+                    &*fs::read_to_string(entry.path())
+                        .expect("should have been able to read the file"),
+                )?;
+            }
+        }
+
+        Ok(Self {
+            templates: handlebars,
+        })
+    }
+
+    pub fn render(&self, name: String, data: &HashMap<String, String>) -> Result<String, StdError> {
+        let text = self.templates.render(&name, &json!(data))?;
+        let job: Value = serde_yaml::from_str(&text).unwrap();
+
+        let request = PutJobRequest {
+            job,
+            idempotency_token: None,
+            namespace: None,
+            headers: None,
+        };
+
+        Ok(serde_json::to_string(&request).unwrap())
+    }
+}
+
+#[derive(Serialize)]
+struct PutJobRequest {
+    #[serde(rename = "Job")]
+    job: Value,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(rename = "IdempotencyToken")]
+    idempotency_token: Option<String>,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(rename = "Namespace")]
+    namespace: Option<String>,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(rename = "Headers")]
+    headers: Option<Map<String, serde_json::Value>>,
+}
+
+// #[cfg(test)]
+// mod tests {
+//     use super::*;
+
+//     #[tokio::test]
+//     async fn test_jobstore() -> Result<(), StdError> {
+//         let path: PathBuf = [env!("CARGO_MANIFEST_DIR"), "testdata"].iter().collect();
+
+//         let mut args: HashMap<String, String> = HashMap::new();
+//         args.insert("image".into(), "ubuntu".into());
+
+//         let j = JobStore::new(path.to_str().unwrap())?;
+//         let output = j.render(String::from("process"), &args)?;
+
+//         let r = submit(&output).await;
+//         println!("{:?}", r);
+
+//         Ok(())
+//     }
+// }
diff --git a/pipe/section/section_impls/bacalhau/src/lib.rs b/pipe/section/section_impls/bacalhau/src/lib.rs
new file mode 100644
index 00000000..f8aa98a3
--- /dev/null
+++ b/pipe/section/section_impls/bacalhau/src/lib.rs
@@ -0,0 +1,69 @@
+use std::sync::Arc;
+
+use section::message::{Ack, Chunk, Column, DataFrame, DataType, Message, ValueView};
+pub mod api;
+pub mod destination;
+pub mod jobstore;
+
+type StdError = Box<dyn std::error::Error + Send + Sync + 'static>;
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct BacalhauPayload {
+    pub filepath: String,
+}
+
+impl BacalhauPayload {
+    fn new(filepath: String) -> Self {
+        Self { filepath }
+    }
+}
+
+impl DataFrame for BacalhauPayload {
+    fn columns(&self) -> Vec<section::message::Column<'_>> {
+        vec![Column::new(
+            "filepath",
+            DataType::Str,
+            Box::new(std::iter::once(ValueView::from(&self.filepath))),
+        )]
+    }
+}
+
+pub struct BacalhauMessage {
+    origin: Arc<str>,
+    payload: Option<Box<dyn DataFrame>>,
+    ack: Option<Ack>,
+}
+
+impl BacalhauMessage {
+    fn new(origin: Arc<str>, payload: impl DataFrame, ack: Option<Ack>) -> Self {
+        Self {
+            origin,
+            payload: Some(Box::new(payload)),
+            ack,
+        }
+    }
+}
+
+impl std::fmt::Debug for BacalhauMessage {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("BacalhauMessage")
+            .field("origin", &self.origin)
+            .field("payload", &self.payload)
+            .finish()
+    }
+}
+
+impl Message for BacalhauMessage {
+    fn origin(&self) -> &str {
+        self.origin.as_ref()
+    }
+
+    fn next(&mut self) -> section::message::Next<'_> {
+        let v = self.payload.take().map(Chunk::DataFrame);
+        Box::pin(async move { Ok(v) })
+    }
+
+    fn ack(&mut self) -> Ack {
+        self.ack.take().unwrap_or(Box::pin(async {}))
+    }
+}
diff --git a/pipe/section/section_impls/bacalhau/testdata/process.yaml b/pipe/section/section_impls/bacalhau/testdata/process.yaml
new file mode 100644
index 00000000..de2ee443
--- /dev/null
+++ b/pipe/section/section_impls/bacalhau/testdata/process.yaml
@@ -0,0 +1,10 @@
+Name: Extractor
+Type: batch
+Namespace: default
+Count: 1
+Tasks:
+- Name: extract
+  Engine:
+    Type: process
+    Params:
+      Name: date

From 76f319a10a9b197190ebded06c53a49aa1e42593 Mon Sep 17 00:00:00 2001
From: Ross Jones <ross@register-dynamics.co.uk>
Date: Fri, 12 Jan 2024 15:34:27 +0000
Subject: [PATCH 2/4] Translate incoming message to hashmap

We really only expect one row in the incoming message and so we'll
convert that to a string-string hashmap to pass as the arguments for
the job submission.  We do this in a roundabout route
(dataframe->recordbatch->json->hashmap).
---
 Cargo.lock                                    |  3 +
 .../section/section_impls/bacalhau/Cargo.toml |  4 ++
 .../section_impls/bacalhau/src/destination.rs | 33 ++++------
 .../section/section_impls/bacalhau/src/lib.rs | 64 -------------------
 4 files changed, 20 insertions(+), 84 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 371a2b89..feea6d79 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -706,6 +706,9 @@ dependencies = [
 name = "bacalhau"
 version = "0.1.0"
 dependencies = [
+ "arrow 49.0.0",
+ "arrow-json 49.0.0",
+ "arrow_msg",
  "handlebars",
  "reqwest",
  "section",
diff --git a/pipe/section/section_impls/bacalhau/Cargo.toml b/pipe/section/section_impls/bacalhau/Cargo.toml
index 4ca6654f..e5914da7 100644
--- a/pipe/section/section_impls/bacalhau/Cargo.toml
+++ b/pipe/section/section_impls/bacalhau/Cargo.toml
@@ -6,6 +6,9 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+arrow = { version = "49", features = ["prettyprint"] }
+arrow-json = "49"
+arrow_msg = { path = "../../../arrow_msg" }
 section = { version = "0.3", path = "../../../section/" }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
@@ -16,5 +19,6 @@ tokio-util = "0.7.10"
 handlebars = "4.4"
 serde_yaml = "0.9"
 
+
 [dev-dependencies]
 stub = { path = "../../section_impls/stub/" }
diff --git a/pipe/section/section_impls/bacalhau/src/destination.rs b/pipe/section/section_impls/bacalhau/src/destination.rs
index d2c463eb..14e769de 100644
--- a/pipe/section/section_impls/bacalhau/src/destination.rs
+++ b/pipe/section/section_impls/bacalhau/src/destination.rs
@@ -1,19 +1,19 @@
-use crate::jobstore::JobStore;
-use crate::{api::submit, BacalhauPayload};
-
-use section::message::ValueView;
+use arrow_msg::df_to_recordbatch;
 use section::{
     command_channel::{Command, SectionChannel},
-    futures::{self, FutureExt, Sink, SinkExt, Stream, StreamExt},
+    futures::{self, FutureExt, Sink, Stream, StreamExt},
     message::Chunk,
     pretty_print::pretty_print,
     section::Section,
     SectionError, SectionMessage,
 };
+
 use std::collections::HashMap;
+use std::future::Future;
 use std::pin::{pin, Pin};
 
-use std::future::Future;
+use crate::api::submit;
+use crate::jobstore::JobStore;
 
 type StdError = Box<dyn std::error::Error + Send + Sync + 'static>;
 
@@ -76,24 +76,17 @@ impl Bacalhau {
                                     Some(_ch) => continue,
                                 };
 
-
-                                let columns = frame.columns();
-
-
                                 section_channel.log(format!("got dataframe chunk from {}:\n{}", msg.origin(), pretty_print(&*frame))).await?;
 
-                                let mut c = columns.into_iter().nth(0).unwrap();
-                                match c.nth(0).unwrap() {
-                                    ValueView::Str(s) => {
-                                        section_channel.log(format!("got string: {}", s)).await?;
-                                    }
-                                    _ => continue,
-                                }
 
+                                let batch = df_to_recordbatch(frame).unwrap();
+                                let json_rows = arrow_json::writer::record_batches_to_json_rows(&[&batch]).unwrap();
+                                let first = json_rows.first().unwrap();
+                                let args: HashMap<String, String> = first.iter().map(|(k, v)| {
+                                    (k.clone(), v.as_str().unwrap().to_string())
+                                }).collect();
 
-                                //ValueView::Str(child.as_string::<i32>().value(value_offset)),
-                                let args: HashMap<String, String> = HashMap::new();
-                                //args.insert("target".into(), "".into());
+                                section_channel.log(format!("Arguments: {:?}", &args)).await?;
                                 self.submit_job(&args).await?;
 
                                 msg.ack().await;
diff --git a/pipe/section/section_impls/bacalhau/src/lib.rs b/pipe/section/section_impls/bacalhau/src/lib.rs
index f8aa98a3..7c32ffbc 100644
--- a/pipe/section/section_impls/bacalhau/src/lib.rs
+++ b/pipe/section/section_impls/bacalhau/src/lib.rs
@@ -1,69 +1,5 @@
-use std::sync::Arc;
-
-use section::message::{Ack, Chunk, Column, DataFrame, DataType, Message, ValueView};
 pub mod api;
 pub mod destination;
 pub mod jobstore;
 
 type StdError = Box<dyn std::error::Error + Send + Sync + 'static>;
-
-#[derive(Debug, Clone, PartialEq)]
-pub struct BacalhauPayload {
-    pub filepath: String,
-}
-
-impl BacalhauPayload {
-    fn new(filepath: String) -> Self {
-        Self { filepath }
-    }
-}
-
-impl DataFrame for BacalhauPayload {
-    fn columns(&self) -> Vec<section::message::Column<'_>> {
-        vec![Column::new(
-            "filepath",
-            DataType::Str,
-            Box::new(std::iter::once(ValueView::from(&self.filepath))),
-        )]
-    }
-}
-
-pub struct BacalhauMessage {
-    origin: Arc<str>,
-    payload: Option<Box<dyn DataFrame>>,
-    ack: Option<Ack>,
-}
-
-impl BacalhauMessage {
-    fn new(origin: Arc<str>, payload: impl DataFrame, ack: Option<Ack>) -> Self {
-        Self {
-            origin,
-            payload: Some(Box::new(payload)),
-            ack,
-        }
-    }
-}
-
-impl std::fmt::Debug for BacalhauMessage {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("BacalhauMessage")
-            .field("origin", &self.origin)
-            .field("payload", &self.payload)
-            .finish()
-    }
-}
-
-impl Message for BacalhauMessage {
-    fn origin(&self) -> &str {
-        self.origin.as_ref()
-    }
-
-    fn next(&mut self) -> section::message::Next<'_> {
-        let v = self.payload.take().map(Chunk::DataFrame);
-        Box::pin(async move { Ok(v) })
-    }
-
-    fn ack(&mut self) -> Ack {
-        self.ack.take().unwrap_or(Box::pin(async {}))
-    }
-}

From f715b736e0484598d04454b1238fb9943f02b131 Mon Sep 17 00:00:00 2001
From: Ross Jones <ross@register-dynamics.co.uk>
Date: Wed, 17 Jan 2024 10:30:41 +0000
Subject: [PATCH 3/4] Update for change to df_to_recordbatch

---
 pipe/section/section_impls/bacalhau/src/destination.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pipe/section/section_impls/bacalhau/src/destination.rs b/pipe/section/section_impls/bacalhau/src/destination.rs
index 14e769de..52650aae 100644
--- a/pipe/section/section_impls/bacalhau/src/destination.rs
+++ b/pipe/section/section_impls/bacalhau/src/destination.rs
@@ -79,7 +79,7 @@ impl Bacalhau {
                                 section_channel.log(format!("got dataframe chunk from {}:\n{}", msg.origin(), pretty_print(&*frame))).await?;
 
 
-                                let batch = df_to_recordbatch(frame).unwrap();
+                                let batch = df_to_recordbatch(frame.as_ref()).unwrap();
                                 let json_rows = arrow_json::writer::record_batches_to_json_rows(&[&batch]).unwrap();
                                 let first = json_rows.first().unwrap();
                                 let args: HashMap<String, String> = first.iter().map(|(k, v)| {

From d6426c2fc99232dc841fa201a8e1d98ae2a291a7 Mon Sep 17 00:00:00 2001
From: Ross Jones <ross@register-dynamics.co.uk>
Date: Fri, 12 Apr 2024 11:12:28 +0100
Subject: [PATCH 4/4] Update arrow json version to match the arrow upgrade

---
 Cargo.lock                                     | 5 +++--
 pipe/section/section_impls/bacalhau/Cargo.toml | 4 ++--
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 1867811a..035cdf55 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -919,8 +919,8 @@ dependencies = [
 name = "bacalhau"
 version = "0.1.0"
 dependencies = [
- "arrow 49.0.0",
- "arrow-json 49.0.0",
+ "arrow 50.0.0",
+ "arrow-json 50.0.0",
  "arrow_msg",
  "handlebars",
  "reqwest",
@@ -2496,6 +2496,7 @@ dependencies = [
  "sqlite_connector",
  "sqlx",
  "tagging_transformer",
+ "thiserror",
  "tokio",
  "toml",
  "tracing",
diff --git a/pipe/section/section_impls/bacalhau/Cargo.toml b/pipe/section/section_impls/bacalhau/Cargo.toml
index e5914da7..570faac9 100644
--- a/pipe/section/section_impls/bacalhau/Cargo.toml
+++ b/pipe/section/section_impls/bacalhau/Cargo.toml
@@ -6,8 +6,8 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-arrow = { version = "49", features = ["prettyprint"] }
-arrow-json = "49"
+arrow = { version = "50", features = ["prettyprint"] }
+arrow-json = "50"
 arrow_msg = { path = "../../../arrow_msg" }
 section = { version = "0.3", path = "../../../section/" }
 serde = { version = "1", features = ["derive"] }