Simple Job Runner

runner.clj

(ns healthunlocked.runner
  (:require [clojure.walk :as walk]
            [com.stuartsierra.component :as component]
            [schema.core :as s]
            [taoensso.timbre :as log]))

(s/defschema Params
  {s/Any s/Any})

(s/defschema Job
  {:id         s/Keyword
   :has-run?   (s/=> s/Bool component/Lifecycle Params)
   :run        (s/=> s/Any  component/Lifecycle Params)
   :params     (s/protocol s/Schema)
   :depends-on [s/Keyword]})

(defonce jobs
  (atom {}))

(defn defjob
  [id & {:as job}]
  (-> (assoc job :id id)
      (update :params assoc s/Any s/Any)
      (->> (s/validate Job)
           (swap! jobs assoc id))))

(defn get-job
  [id]
  (or (get @jobs id)
      (throw (ex-info "Could not find job for ID" {:id id :jobs @jobs}))))

(defn execution-plan
  [id]
  (->> (get-job id)
       (tree-seq map? #(map get-job (:depends-on %)))
       (reverse)
       (distinct)))

(defn execute!
  [system id params]
  (let [system (component/start system)
        plan   (execution-plan id)]
    (log/info "Executing" id "and its dependencies")
    (log/info "Plan:" (mapv :id plan))
    (try
      (doseq [{:keys [id run has-run?] :as job} plan]
        (log/info "Running" id)
        (s/validate (:params job) params)
        (if (has-run? system params)
          (log/info "Skipping" id "as it has already run")
          (run system params)))
      (log/info "Execution complete")
      (catch Exception e
        (log/error e "Failed running jobs"))
      (finally
        (component/stop system)))))