   (store/initialize uri))
 
 
+; TODO: rethink this
 (defmacro ^:private measure-method
   "Anophoric macro to measure a store method."
   [[method-kw args] & body]
   `(meter/measure-method*
-     ~'store ~method-kw ~args
+     ~'store ~(keyword "blocks.store" (name method-kw)) ~args
      (fn body# [] ~@body)))
 
 
+(defn list
+  "Enumerate the stored blocks, returning a stream of blocks ordered by their
+  multihash id. The store will continue reading blocks until the stream is
+  closed or there are no more matching blocks to return.
+
+  - `:algorithm`
+    Only return blocks identified by this hash algorithm.
+  - `:after`
+    Return blocks whose id (in hex) lexically follows this string. A multihash
+    may also be provided and will be coerced to hex.
+  - `:before`
+    Return blocks whose id (in hex) lexically precedes this string. A multihash
+    may also be provided and will be coerced to hex.
+  - `:limit`
+    Restrict the maximum number of blocks returned on the stream."
+  [store & opts]
+  (let [opts (args->map opts)
+        opts (merge
+               ; Validate algorithm option.
+               (when-let [algorithm (:algorithm opts)]
+                 (if (keyword? algorithm)
+                   {:algorithm algorithm}
+                   (throw (IllegalArgumentException.
+                            (str "Option :algorithm is not a keyword: "
+                                 (pr-str algorithm))))))
+               ; Validate 'after' boundary.
+               (when-let [after (:after opts)]
+                 (cond
+                   (hex-string? after)
+                   {:after (str/lower-case after)}
+
+                   (multihash? after)
+                   {:after (multihash/hex after)}
+
+                   :else
+                   (throw (IllegalArgumentException.
+                            (str "Option :after is not a hex string or multihash: "
+                                 (pr-str after))))))
+               ; Validate 'before' boundary.
+               (when-let [before (:before opts)]
+                 (cond
+                   (hex-string? before)
+                   {:before (str/lower-case before)}
+
+                   (multihash? before)
+                   {:before (multihash/hex before)}
+
+                   :else
+                   (throw (IllegalArgumentException.
+                            (str "Option :before is not a hex string or multihash: "
+                                 (pr-str before))))))
+               ; Validate query limit.
+               (when-let [limit (:limit opts)]
+                 (if (pos-int? limit)
+                   {:limit limit}
+                   (throw (IllegalArgumentException.
+                            (str "Option :limit is not a positive integer: "
+                                  (pr-str limit))))))
+               ; Ensure no other options.
+               (when-let [bad-opts (not-empty (dissoc opts :algorithm :after :before :limit))]
+                 (throw (IllegalArgumentException.
+                          (str "Unknown options passed to list: " (pr-str bad-opts))))))]
+    ; TODO: should be metering the stream of blocks somehow, not time-to-stream
+    (measure-method [:list opts]
+      (store/-list store opts))))
+
+
 (defn stat
-  "Returns a map with an `:id` and `:size` but no content. The returned map
-  may contain additional data like the date stored. Returns nil if the store
-  does not contain the identified block."
+  "Load metadata about a block if the store contains it. Returns a deferred
+  which yields a map with block information but no content, or nil if the store
+  does not contain the identified block.
+
+  The block stats include the `:id`, `:size`, and `:stored-at` fields. The
+  returned map may also have additional implementation-specific storage
+  metadata, similar to returned blocks."
   [store id]
-  (when id
-    (when-not (instance? Multihash id)
-      (throw (IllegalArgumentException.
-               (str "Id value must be a multihash, got: " (pr-str id)))))
-    (measure-method [::stat id]
-      (store/-stat store id))))
-
-
-(defn list
-  "Enumerates the stored blocks, returning a lazy sequence of block stats sorted
-  by id. Iterating over the list may result in additional operations to read
-  from the backing data store.
-
-  - `:algorithm`  only return blocks using this hash algorithm
-  - `:after`      list blocks whose id (in hex) lexically follows this string
-  - `:limit`      restrict the maximum number of results returned
-  "
-  [store & opts]
-  (let [allowed-keys #{:algorithm :after :limit}
-        opts-map (cond
-                   (empty? opts) nil
-                   (and (= 1 (count opts))
-                        (or (map? (first opts))
-                            (nil? (first opts))))
-                     (first opts)
-                   :else (apply hash-map opts))
-        bad-opts (set/difference (set (keys opts-map)) allowed-keys)]
-    (when (not-empty bad-opts)
-      (throw (IllegalArgumentException.
-               (str "Invalid options passed to list: "
-                    (str/join " " bad-opts)))))
-    (when-let [algorithm (:algorithm opts-map)]
-      (when-not (keyword? algorithm)
-        (throw (IllegalArgumentException.
-                 (str "Option :algorithm is not a keyword: "
-                      (pr-str algorithm))))))
-    (when-let [after (:after opts-map)]
-      (when-not (and (string? after) (re-matches #"^[0-9a-fA-F]*$" after))
-        (throw (IllegalArgumentException.
-                 (str "Option :after is not a hex string: "
-                      (pr-str after))))))
-    (when-let [limit (:limit opts-map)]
-      (when-not (and (integer? limit) (pos? limit))
-        (throw (IllegalArgumentException.
-                 (str "Option :limit is not a positive integer: "
-                      (pr-str limit))))))
-    (measure-method [::list opts-map]
-      (store/-list store opts-map))))
+  (when-not (multihash? id)
+    (throw (IllegalArgumentException.
+             (str "Block id must be a multihash, got: " (pr-str id)))))
+  (measure-method [:stat id]
+    (store/-stat store id)))
 
 
 (defn get
-  "Loads content for a multihash and returns a block record. Returns nil if no
-  block is stored for that id.
-
-  The returned block is checked to make sure the id matches the requested
-  multihash."
+  "Load a block from the store. Returns a deferred which yields the block if
+  the store contains it, or nil if no block is stored for that id."
   [store id]
-  (when-not (instance? Multihash id)
+  (when-not (multihash? id)
     (throw (IllegalArgumentException.
-             (str "Id value must be a multihash, got: " (pr-str id)))))
-  (let [block (measure-method [::get id]
-                (store/-get store id))]
-    (when block
-      (when-not (= id (:id block))
-        (throw (RuntimeException.
-                 (str "Asked for block " id " but got " (:id block)))))
+             (str "Block id must be a multihash, got: " (pr-str id)))))
+  (d/chain
+    (measure-method [:get id]
+      (store/-get store id))
+    (fn validate-block
+      [block]
+      (when block
+        (when-not (= id (:id block))
+          (throw (RuntimeException.
+                   (str "Asked for block " id " but got " (:id block)))))
+        (meter/metered-block store ::meter/io-read block)))))
+
+
+(defn put!
+  "Save a block into the store. Returns a deferred which yields the stored
+  block, which may have already been present in the store."
+  [store block]
+  (when-not (instance? Block block)
+    (throw (IllegalArgumentException.
+             (str "Argument must be a block, got: " (pr-str block)))))
+  (d/chain
+    (measure-method [:put! block]
+      (->> block
+           (meter/metered-block store ::meter/io-write)
+           (store/-put! store)))
+    (fn meter-block
+      [block]
       (meter/metered-block store ::meter/io-read block))))
 
 
-(defn put!
-  "Saves a block into the store. Returns the block record, updated with stat
-  metadata."
-  [store block]
-  (when block
-    (when-not (instance? Block block)
-      (throw (IllegalArgumentException.
-               (str "Argument must be a block, got: " (pr-str block)))))
-    (data/merge-blocks
-      block
-      (measure-method [::put! block]
-        (->> block
-             (meter/metered-block store ::meter/io-write)
-             (store/-put! store)
-             (meter/metered-block store ::meter/io-read))))))
-
-
 (defn store!
-  "Stores content from a byte source in a block store and returns the block
-  record.
+  "Store content from a byte source in a block store. Returns a deferred which
+  yields the stored block, or nil if the source was empty.
 
   If the source is a file, it will be streamed into the store. Otherwise, the
   content is read into memory, so this may not be suitable for large sources."
   ([store source]
    (store! store source default-algorithm))
   ([store source algorithm]
-   (let [block (if (instance? File source)
-                 (from-file source algorithm)
-                 (read! source algorithm))]
-     (when (pos? (:size block))
-       (put! store block)))))
+   ; OPTIMIZE: `BlockReceiver` protocol
+   (d/chain
+     (d/future
+       (if (instance? File source)
+         (from-file source algorithm)
+         (read! source algorithm)))
+     (fn put-block
+       [block]
+       (when (and block (pos? (:size block)))
+         (put! store block))))))
 
 
 (defn delete!
-  "Removes a block from the store. Returns true if the block was found and
-  removed."
+  "Remove a block from the store. Returns a deferred which yields true if the
+  block was found and removed."
   [store id]
-  (when id
-    (measure-method [::delete! id]
-      (store/-delete! store id))))
+  (when-not (multihash? id)
+    (throw (IllegalArgumentException.
+             (str "Block id must be a multihash, got: " (pr-str id)))))
+  (measure-method [:delete! id]
+    (store/-delete! store id)))
 
 
 
 ;; ## Batch API
 
-(defn- validate-collection-of
-  "Validates that the given argument is a collection of a certain class of
-  entries."
-  [cls xs]
-  (when-not (coll? xs)
-    (throw (IllegalArgumentException.
-             (str "Argument must be a collection: " (pr-str xs)))))
-  (when-let [bad-entries (seq (filter (complement (partial instance? cls)) xs))]
-    (throw (IllegalArgumentException.
-             (str "Collection entries must be " cls " values: "
-                  (pr-str bad-entries))))))
-
-
 (defn get-batch
-  "Retrieves a batch of blocks identified by a collection of multihashes.
-  Returns a sequence of the requested blocks in no particular order. Any blocks
-  which were not found in the store are omitted from the result."
+  "Retrieve a batch of blocks identified by a collection of multihashes.
+  Returns a deferred which yields a collection of the blocks which were found.
+
+  The blocks are returned in no particular order, and any missing blocks are
+  omitted from the result."
   [store ids]
-  (validate-collection-of Multihash ids)
-  (if (satisfies? store/BatchingStore store)
-    (measure-method [::get-batch ids]
-      (keep (partial meter/metered-block store ::meter/io-read)
-            (store/-get-batch store ids)))
-    (doall (keep (partial get store) ids))))
+  (d/chain
+    (->> (set ids)
+         (map (partial get store))
+         (apply d/zip))
+    (fn omit-missing
+      [blocks]
+      (into [] (remove nil?) blocks))))
 
 
 (defn put-batch!
-  "Saves a collection of blocks into the store. Returns a sequence of the
-  stored blocks, in no particular order.
-
-  This is not guaranteed to be an atomic operation; readers may be able to see
-  the store in a partially-updated state."
+  "Save a collection of blocks into the store. Returns a deferred which
+  yields a collection of stored blocks.
+
+  This is not guaranteed to be atomic; readers may see the store in a
+  partially updated state."
   [store blocks]
-  (validate-collection-of Block blocks)
   (if-let [blocks (seq (remove nil? blocks))]
-    (if (satisfies? store/BatchingStore store)
-      (measure-method [::put-batch! blocks]
-        (->> blocks
-             (map (partial meter/metered-block store ::meter/io-write))
-             (store/-put-batch! store)
-             (map (partial meter/metered-block store ::meter/io-read))))
-      (mapv (partial put! store) blocks))
-    []))
+    (apply d/zip (map (partial put! store) blocks))
+    (d/success-deferred [])))
 
 
 (defn delete-batch!
-  "Removes a batch of blocks from the store, identified by a collection of
-  multihashes. Returns a set of ids for the blocks which were found and
-  deleted.
-
-  This is not guaranteed to be an atomic operation; readers may be able to see
-  the store in a partially-deleted state."
+  "Remove a batch of blocks from the store, identified by a collection of
+  multihashes. Returns a deferred which yields a set of ids for the blocks
+  which were found and deleted.
+
+  This is not guaranteed to be atomic; readers may see the store in a
+  partially deleted state."
   [store ids]
-  (validate-collection-of Multihash ids)
-  (if (satisfies? store/BatchingStore store)
-    (measure-method [::delete-batch! ids]
-      (set (store/-delete-batch! store ids)))
-    (set (filter (partial delete! store) ids))))
+  (if-let [ids (not-empty (into [] (comp (remove nil?) (distinct)) ids))]
+    (d/chain
+      (apply d/zip (map (partial delete! store) ids))
+      (fn match-ids
+        [results]
+        (into #{}
+              (comp
+                (filter first)
+                (map second))
+              (map vector results ids))))
+    (d/success-deferred #{})))
 
 
 
 ;; ## Storage Utilities
 
-(defn erase!!
-  "Completely removes any data associated with the store. After this call, the
-  store should be empty. This is not guaranteed to be an atomic operation!"
+(defn erase!
+  "Completely remove all data associated with the store. After this call, the
+  store will be empty. Returns a deferred which yields true once the store has
+  been erased.
+
+  This is not guaranteed to be atomic; readers may see the store in a partially
+  erased state."
   [store]
   (if (satisfies? store/ErasableStore store)
-    (measure-method [::erase!! nil]
+    (measure-method [:erase! nil]
       (store/-erase! store))
-    (run! (comp (partial delete! store) :id)
-          (store/-list store nil))))
+    (s/consume
+      (comp (partial delete! store) :id)
+      (list store))))
 
 
 (defn scan
-  "Scans all the blocks in the store, building up a store-level summary. If
+  "Scan all the blocks in the store, building up a store-level summary. If
   given, the predicate function will be called with each block in the store.
   By default, all blocks are scanned."
   ([store]
    (scan store nil))
   ([store p]
-   (-> (store/-list store nil)
-       (cond->> p (filter p))
-       (->> (reduce sum/update (sum/init))))))
+   (->
+     (store/-list store nil)
+     (cond->>
+       p (s/filter p))
+     (->>
+       (s/reduce sum/update (sum/init))))))
 
 
 (defn sync!
   "Synchronize blocks from the `source` store to the `dest` store. Returns a
   summary of the copied blocks. Options may include:
 
-  - `:filter` a function to run on every block stats before it is copied to the
-    `dest` store. If the function returns a falsey value, the block will not be
-    copied."
-  [source dest & {:as opts}]
-  (-> (store/missing-blocks
+  - `:filter`
+    A function to run on every block before it is synchronized. The block will
+    only be copied if the filter returns a truthy value."
+  [source dest & opts]
+  (let [opts (args->map opts)]
+    (->
+      (store/missing-blocks
         (store/-list source nil)
         (store/-list dest nil))
       (cond->>
-        (:filter opts) (filter (:filter opts)))
-      (->> (reduce
-             (fn copy-block
-               [summary stat]
-               (put! dest (get source (:id stat)))
-               (sum/update summary stat))
-             (sum/init)))))
+        (:filter opts) (s/filter (:filter opts)))
+      (->>
+        (s/reduce
+          ; TODO: s/reduce expects this to be synchronous
+          (fn copy-block
+            [summary block]
+            (prn summary block)
+            (d/chain
+              (put! dest block)
+              (fn update-sum
+                [block']
+                (sum/update summary block'))))
+          (sum/init))))))



(defn from-file
  "Creates a lazy block from a local file. The file is read once to calculate
  the identifier."
  ([file]
   (from-file file default-algorithm))
  ([file algorithm]
   (let [file (io/file file)
         hash-fn (data/checked-hasher algorithm)
         reader (fn file-reader [] (FileInputStream. file))
         id (hash-fn (reader))]
     (data/lazy-block id (.length file) reader))))


(defn open
  "Opens an input stream to read the contents of the block.

  If `start` and `end` are given, the input stream will only return content
  from the starting index byte to the byte before the end index. For example,
  opening a block with size _n_ with `(open block 0 n)` would return the full
  block contents."
  (^java.io.InputStream
   [block]
   (data/content-stream block nil nil))
  (^java.io.InputStream
   [block start end]
   (when-not (and (integer? start)
                  (integer? end)
                  (not= start end)
                  (<= 0 start end (:size block)))
     (throw (IllegalArgumentException.
              (format "Range bounds must be distinct integers within block bounds: [0, %d]"
                      (:size block)))))
   (data/content-stream block start end)))


(defn read!
  "Reads data into memory from the given source and hashes it to identify the
  block."
  ([source]
   (read! source default-algorithm))

    (java.io
      File
      FileInputStream)
    java.time.Instant
    multiformats.hash.Multihash
    org.apache.commons.io.input.CountingInputStream))


;; ## Utilities

(def default-algorithm
  "The hashing algorithm used if not specified in functions which create blocks."
  :sha2-256)


(defn- args->map
  "Accept arguments and return a map corresponding to the input. Accepts either
  a single map argument or kw-args."
  [args]
  (when (seq args)
    (if (and (= 1 (count args))
             (or (map? (first args))
                 (nil? (first args))))
      (first args)
      (apply hash-map args))))


(defn- hex-string?
  "True if the value is a hexadecimal string."
  [x]
  (and (string? x) (re-matches #"[0-9a-fA-F]*" x)))


(defn- multihash?
  "True if the value is a multihash."
  [x]
  (instance? Multihash x))



