             [onyx.static.default-vals :refer [defaults]]
             [onyx.static.uuid :refer [random-uuid]]
             [onyx.extensions :as extensions]
+            [schema.core :as s]
+            [onyx.schema :as os]
             [taoensso.timbre :refer [info debug fatal]]))
+
+;;;;;;;;;;;;;;
+;;;;;;;;;;;;;;
+;; task schemas
+
+(def UserTaskMapKey
+  (os/build-allowed-key-ns :datomic))
+
+(def DatomicReadLogTaskMap
+  (s/->Both [os/TaskMap 
+             {:datomic/uri s/Str
+              (s/optional-key :datomic/log-start-tx) s/Int
+              (s/optional-key :datomic/log-end-tx) s/Int
+              :checkpoint/force-reset? s/Bool
+              (s/optional-key :onyx/max-peers) (s/enum 1)
+              (s/optional-key :onyx/n-peers) (s/enum 1)
+              UserTaskMapKey s/Any}]))
+
+(def DatomicReadDatomsTaskMap
+  (s/->Both [os/TaskMap 
+             {:datomic/uri s/Str
+              :datomic/t s/Int
+              :datomic/datoms-index s/Keyword
+              :datomic/datoms-components [s/Any]
+              :datomic/datoms-per-segment s/Int
+              (s/optional-key :onyx/max-peers) (s/enum 1)
+              (s/optional-key :onyx/n-peers) (s/enum 1)
+              UserTaskMapKey s/Any}]))
+
+
+(def DatomicReadIndexRangeTaskMap
+  (s/->Both [os/TaskMap 
+             {:datomic/uri s/Str
+              :datomic/t s/Int
+              :datomic/index-attribute s/Any
+              :datomic/index-range-start s/Any
+              :datomic/index-range-end s/Any
+              :datomic/datoms-per-segment s/Int
+              (s/optional-key :onyx/max-peers) (s/enum 1)
+              (s/optional-key :onyx/n-peers) (s/enum 1)
+              UserTaskMapKey s/Any}]))
+
+(def DatomicWriteDatomsTaskMap
+  (s/->Both [os/TaskMap 
+             {:datomic/uri db-uri
+              (s/optional-key :datomic/partition) (s/either s/Int s/Keyword)
+              (s/optional-key :onyx/max-peers) (s/enum 1)
+              (s/optional-key :onyx/n-peers) (s/enum 1)
+              UserTaskMapKey s/Any}]))
+
+;;; Helpers
 
 (defn safe-connect [task-map]
   (if-let [uri (:datomic/uri task-map)]


(defn write-bulk-datoms [pipeline-data]
  (let [task-map (:onyx.core/task-map pipeline-data)
        conn (safe-connect task-map)]
    (->DatomicWriteBulkDatoms conn)))


  (when-not (or (= 1 (:onyx/max-peers task-map))
                (= 1 (:onyx/n-peers task-map)))
    (throw (ex-info "Read datoms tasks must set :onyx/max-peers 1" task-map)))

  (let [_ (extensions/write-chunk log :chunk {:chunk-index -1 :status :incomplete} task-id)
        content (extensions/read-chunk log :chunk task-id)]
    (if (= :complete (:status content))
