     (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)
 
 
-
-;; ## Stat Metadata
-
-(defn with-stats
-  "Returns the given block with updated stat metadata."
-  [block stats]
-  (vary-meta block assoc :block/stats stats))
-
-
-(defn meta-stats
-  "Returns stat information from a block's metadata, if present."
-  [block]
-  (:block/stats (meta block)))
+(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))
 
 
 

  (store/initialize uri))


(defmacro ^:private measure-method
  "Anophoric macro to measure a store method."
  [[method-kw args] & body]
  `(meter/measure-method*
     ~'store ~method-kw ~args
     (fn body# [] ~@body)))


(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."
  [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))))


(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."
  [store id]
  (when-not (instance? 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)))))
      (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.

  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)))))


(defn delete!
  "Removes a block from the store. Returns true if the block was found and
  removed."
  [store id]
  (when 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."
  [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))))


(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."
  [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))
    []))


(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."
  [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))))



;; ## 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!"
  [store]
  (if (satisfies? store/ErasableStore store)
    (measure-method [::erase!! nil]
      (store/-erase! store))
    (run! (comp (partial delete! store) :id)
          (store/-list store nil))))


(defn scan
  "Scans 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))))))


(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
        (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)))))

    [clojure.java.io :as io]
    [clojure.set :as set]
    [clojure.string :as str]
    [manifold.deferred :as d]
    [manifold.stream :as s]
    [multiformats.hash :as multihash])
  (:import
    (blocks.data
      Block
