 (defn- read-body [message]
   (read-string (message :body)))
 
-(defn run-job-from-message
-  "Process a queue item from the worker queue. Runs script, stores output.
+(defn- read-body-keys [message & ks]
+  (-> (message :body)
+      read-string
+      (select-keys ks)))
+
+(defn message-job-runner
+  "Returns a fn that processes a message from the worker queue.
+  Runs script, stores output.
   Currently outputs to both a big log and separate job logs.
   TODO: Remove big log."
-  [workspaces-path big-log-key
-   credentials bucket ; TODO: make all these args a map
-   message]
+  [{:keys [workspaces-path log-key
+           aws-credentials bucket]}]
   (println "--------------------------------------")
-  (println "run-job-from-message")
-  (let [{uuid     :uuid
-         job-name :name
-         git-url  :repo
-         git-ref  :ref
-         script   :script}  (read-body message)
+  (println "message-job-runner")
+  (fn [message]
+    (let [job                 (read-body-keys message :uuid :name :repo :ref :script)
 
-        workspace-path      (join "/" [workspaces-path job-name "workspace"])
+          workspace-path      (path workspaces-path (job :name) "workspace")
 
-        existing-big-log    (io/get-key credentials bucket big-log-key)
+          existing-big-log    (io/get-key aws-credentials bucket log-key)
 
-        repo                (git/fresh-clone git-url workspace-path git-ref)
-        build               (runner/run! workspace-path script)
-        ]
+          repo                (git/fresh-clone (job :repo) workspace-path (job :ref))
+          build               (runner/run! workspace-path (job :script))
+          ]
 
-    (println "for" job-name)
-    (io/put credentials bucket
-            (results/log-key uuid job-name (build :exit))
-            (build :out))
-    (io/put credentials bucket
-            big-log-key
-            (str existing-big-log (build :out)))
-    (build :out)))
+      (println "for" (job :name))
+      (io/put aws-credentials bucket
+              (results/log-key (job :uuid) (job :name) (build :exit))
+              (build :out))
+      (io/put aws-credentials bucket
+              log-key
+              (str existing-big-log (build :out)))
+      (build :out))))
 
 (defn- no-pending-upstream-jobs?
   "Predicate used to filter out messages to leave in queue
   i.e. don't process and don't delete."
   [credentials bucket
    message]
-  (let [ls (partial io/ls
-                    credentials
-                    bucket)
-        {job-name             :name
-         uuid                 :uuid
-         upstream-job-names   :upstream} (read-body message)
-        all-results           (ls uuid)
+  (let [ls                    (partial io/ls
+                                       credentials
+                                       bucket)
+        job                   (read-body-keys message :name :uuid :upstream)
+        all-results           (ls (job :uuid))
         completed-job-names   (results/job-names all-results)]
     (println "--------------------------------------")
-    (println "no-pending-upstream-jobs? for" job-name)
+    (println "no-pending-upstream-jobs? for" (job :name))
     (println "raw list:" all-results)
-    (println "superset?" completed-job-names upstream-job-names)
-    (doto (superset? completed-job-names upstream-job-names) println)))
+    (println "superset?" completed-job-names (job :upstream))
+    (doto (superset? completed-job-names (job :upstream)) println)))
 
 (defn parser [config]
   (let [client (sqs-client (vals (config :aws-credentials)))

                                :limit 10)))))

(defn worker [config]
  (let [credentials (config :aws-credentials)
        client (sqs-client (vals credentials))
        q (sqs-queue client (config :build-queue))
        ls (partial io/ls credentials (config :bucket))
        upstream-all-passed? (fn [message]
                               (let [{job-name             :name
                                      uuid                 :uuid
                                      upstream-job-names   :upstream} (read-body message)
                                     passed-job-names      (results/passed-job-names (ls uuid))]
                                 (println "--------------------------------------")
                                 (println "upstream-all-passed? for" job-name)
                                 (doto (or (empty? upstream-job-names)
                                           (superset? passed-job-names upstream-job-names)) println)))]
    (println "Starting worker")
    (dorun
      (map (sqs/deleting-consumer client
                                  (runner/decide-whether-to
                                    (partial run-job-from-message
                                             (config :workspaces-path)
                                             (config :log-key)
                                             (config :aws-credentials)
                                             (config :bucket))
                                    upstream-all-passed?))
           (filter (partial no-pending-upstream-jobs?
                            (config :aws-credentials)

(defn- sqs-client [credentials] (apply sqs/create-client credentials))
(defn- sqs-queue [client queue-name] (sqs/create-queue client queue-name))

(defn- path [& parts] (join "/" parts))

(defn- workspace-dir [basedir url]
  (path basedir (last (split url #"/")) "workspace"))

(defn process-push-message [client workspaces-path build-queue-name {payload :body}]
  (when-let [url (payload/clone-target-url payload)]
