   (def hist
     (r/commit-history (get-in @client-stage [user cdvcs-id :state :commit-graph])
                       (first (get-in @client-stage [user cdvcs-id :state :heads]))))
-
+ 
   (require '[clojure.core.async :as async]
            '[konserve.core :as k])
 
   (def commits
-    (->> hist
-         (map (fn [id] (k/get-in mem-store [id])))
+    (->> (take 10 hist)
+         (map #(k/get-in client-store [%]))
          async/merge
          (async/into [])
-         <??))
+         <??
+         ))
 
-  (take 10 commits)
+  (filter nil? commits)
+
+  (map (comp keys second) commits)
+  (map (fn [{[[_ v]] :transactions :as elem}] (if-not v elem)) commits)
+
+  (->> commits
+       (map (fn [{[[_ v]] :transactions :as elem}] v))
+       (filter identity)
+       (map #(go-try [% (<? (k/get-in client-store [%]))]))
+       async/merge
+       (async/into [])
+       <??
+       (map first))
 
   (<?? (k/get-in mem-store [(first hist)]))
 

(defn store-tweets [stage pending]
  (go-try
   (let [st (.getTime (java.util.Date.))
         tweet-txs [['add-tweets (vec (first (swap! pending (fn [[prev cur]] [cur '()]))))]]]
     (when-not (empty? tweet-txs)
       (<? (cs/transact! stage [user cdvcs-id] tweet-txs)))
     (when (< (rand) 0.01)
       (println "Date: " (java.util.Date.))


  (count (second @pending))

  (require '[replikativ.p2p.fetch :refer [fetch]])

  (def client-store (<?? (new-fs-store "/media/void/1e843516-40ec-4b2b-83d2-c796ee312a59/twitter/")))

  (def client (<?? (client-peer client-store :middleware fetch)))

  (def client-stage (<?? (create-stage! user client)))
  (<?? (cs/create-cdvcs! client-stage :id cdvcs-id))

  (<?? (cs/merge! client-stage [user cdvcs-id]
                  (seq (get-in @client-stage [user cdvcs-id :state :heads]))))

  (<?? (connect! client-stage "ws://127.0.0.1:9095"))

  (<?? (connect! client-stage "ws://topiq.es:9095"))

  (keys @(:state (get-in @peer [:volatile])))

