workspace.clj

(ns automat-play.core
  (:require [schema.core :as s]
            [automat.core :as a]))

(def ContractOutput
  {:witan/schema       s/Any
   :witan/key          s/Keyword
   :witan/display-name s/Str})

(def ContractInput
  {:witan/schema       s/Any
   :witan/key          s/Keyword
   :witan/display-name s/Str})

(def Contract
  {:witan/name s/Keyword
   :witan/fn   s/Keyword
   :witan/version s/Str ;; TODO check semver
   :witan/outputs [ContractOutput]
   (s/optional-key :witan/inputs) [ContractInput]
   (s/optional-key :witan/params-schema) (s/maybe s/Any)})

(def Input
  {:witan/input-src-fn s/Keyword
   :witan/input-src-key s/Keyword
   :witan/input-dest-key s/Keyword})

(def CatalogEntry
  {:witan/name s/Keyword
   :witan/version s/Str ;; TODO check semver
   (s/optional-key :witan/params) s/Any
   (s/optional-key :witan/inputs) [Input]})

(def Workspace
  {:workflow  [s/Keyword]
   :contracts [Contract]
   :catalog   [CatalogEntry]})

;;;;

(def FooNumber
  s/Num)

(def MulX
  {:x s/Num})

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(defn get-data
  [key]
  (get {:a 1
        :b 3
        :c 5} key))

(defn inc*
  [{:keys [number]} _]
  {:number (inc number)})

(defn mul2
  [{:keys [number]} _]
  {:number (* number 2)})

(defn mulX
  [{:keys [number]} {:keys [x]}]
  {:number (* number x)})

(def contracts
  [{:witan/name          :foo/inc
    :witan/fn            :automat-play.core/inc*
    :witan/version       "1.0"
    :witan/params-schema nil
    :witan/inputs [{:witan/schema       FooNumber
                    :witan/key          :number
                    :witan/display-name "Number"}]
    :witan/outputs [{:witan/schema       FooNumber
                     :witan/key          :number
                     :witan/display-name "Number"}]}
   {:witan/name          :foo/mul2
    :witan/fn            :automat-play.core/mul2
    :witan/version       "1.0"
    :witan/params-schema nil
    :witan/inputs [{:witan/schema       FooNumber
                    :witan/key          :number
                    :witan/display-name "Number"}]
    :witan/outputs [{:witan/schema       FooNumber
                     :witan/key          :number
                     :witan/display-name "Number"}]}
   {:witan/name          :foo/mulX
    :witan/fn            :automat-play.core/mulX
    :witan/version       "1.0"
    :witan/params-schema MulX
    :witan/inputs [{:witan/schema       FooNumber
                    :witan/key          :number
                    :witan/display-name "Number"}]
    :witan/outputs [{:witan/schema       FooNumber
                     :witan/key          :number
                     :witan/display-name "Number"}]}])

(def catalog
  [{:witan/name :foo/inc
    :witan/version "1.0"
    :witan/inputs [{:witan/input-src-fn   :automat-play.core/get-data
                    :witan/input-src-key  :a
                    :witan/input-dest-key :number}]}
   {:witan/name :foo/mul2
    :witan/version "1.0"}
   {:witan/name :foo/mulX
    :witan/version "1.0"
    :witan/params {:x 3}}])

(def workflow
  [:foo/inc :foo/mul2 :foo/mulX])

(def workspace
  {:workflow workflow
   :contracts contracts
   :catalog catalog})

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(defn- fetch-data
  [{:keys [witan/input-src-fn witan/input-src-key witan/input-dest-key]}]
  (let [fnc (resolve (symbol (subs (str input-src-fn) 1)))]
    (if fnc
      [input-dest-key (fnc input-src-key)]
      (throw (Exception. (str input-src-fn " could not be resolved."))))))

(s/defn execute
  [{:keys [workflow contracts catalog]} :- Workspace]
  (when (or
         (not-empty (clojure.set/difference (set workflow) (->> contracts (map :witan/name) (set))))
         (not-empty (clojure.set/difference (set workflow) (->> catalog   (map :witan/name) (set)))))
    (throw (IllegalArgumentException. (str "A workflow entry is unrepresented in catalog and/or contracts: " workflow))))
  (let [workflow* (interleave workflow (repeat (count workflow) (a/$ :_process)))
        reducers {:_process (fn [state id]
                              (let [catalog-entry (some #(when (= (:witan/name %) id) %) catalog)
                                    version       (:witan/version catalog-entry)
                                    contract      (some #(when (and (= (:witan/name %) id) (= (:witan/version %) version)) %) contracts)
                                    inputs        (reduce (fn [a c] (apply assoc a (fetch-data c))) state (:witan/inputs catalog-entry))
                                    input-errors  (->> (:witan/inputs contract)
                                                       (reduce (fn [a {:keys [witan/schema witan/key]}]
                                                                 (assoc a key (s/check schema (get inputs key)))) {})
                                                       (remove (comp nil? second))
                                                       (not-empty))]
                                (if input-errors
                                  (assoc state
                                         :error (str "Failed to procure appropriate inputs: " (vec input-errors))
                                         :error-state id)
                                  (let [params (:witan/params catalog-entry)
                                        param-errors (when-let [schema (:witan/params-schema contract)] (s/check schema params))]
                                    (if param-errors
                                      (assoc state
                                             :error (str "One ore more parameters failed to validate: " param-errors)
                                             :error-state id)
                                      (let [state (merge state inputs)
                                            fnc (resolve (symbol (subs (str (:witan/fn contract)) 1)))]
                                        (if-not fnc
                                          (assoc state
                                                 :error (str "Function failed to resolve:" (:witan/fn contract))
                                                 :error-state id)
                                          (let [result (fnc state params)
                                                result-errors (->> result
                                                                   (reduce (fn [a [k v]]
                                                                             (let [output (some #(when (= (:witan/key %) k) %) (:witan/outputs contract))]
                                                                               (assoc a k (if-not output
                                                                                            "Output not defined"
                                                                                            (s/check (:witan/schema output) v))))) {})
                                                                   (remove (comp nil? second))
                                                                   (not-empty))]
                                            (if result-errors
                                              (assoc state
                                                     :error (str "One or more outputs failed to validate: " (vec result-errors))
                                                     :error-state id)
                                              (merge state result))))))))))}
        compiled (a/compile workflow* {:reducers reducers})]
    (a/find compiled {} workflow)))

;; => (execute workspace)