         (partition batch-size batch-size [])
         (map (fn [batch]
                (with-meta batch {:order order-info
+                                 :action action
                                  :size (count batch)
                                  :start (first batch)
                                  :end (last batch)}))))))

                         db)]
            (let [jobs (->> (map (fn [[v g]]
                                   [v gene/->Variation g]) v-g)
                            (make-batches 1000 :gene->variation))]
              (doseq [job jobs]
                (>!! scheduler job)))
            (let [jobs (->> (map (fn [[v g]]
                                   [g variation/->Gene v]) v-g)
                            (make-batches 1000 :variation->gene))]
              (doseq [job jobs]
                (>!! scheduler job)))

            )

          ;; close the channel to indicate no more input
          ;; existing jobs on the channel will remain available for consumers

  (let [url-prefix (if index
                     (format "%s/%s" es-base-url index)
                     es-base-url)]
    (http/post (format "%s/_bulk?refresh=%s" url-prefix (or refresh "false"))
               {:headers {:content-type "application/x-ndjson"}
                :body formatted-docs})))

