 (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)
+         tweets (vec (first (swap! pending (fn [[prev cur]] [cur '()]))))
+         tweet-txs [['add-tweets tweets]]]
+     (when-not (empty? tweets)
        (<? (cs/transact! stage [user cdvcs-id] tweet-txs)))
      (when (< (rand) 0.01)
        (println "Date: " (java.util.Date.))

  (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])))
         async/merge
         (async/into [])
         <??))

  (take 10 commits)

  (<?? (k/get-in mem-store [(first hist)]))



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

