         (>! ch event)))
     ch))
 
-(defn cold-stream [a-stream date-string & args]
-  (let [m-args (apply hash-map args)
-        s (riak/lazy-events (:db a-stream) date-string)]
-    (if (true? (:channel m-args))
-      (seq->channel s)
-      s)))
-
-(defmulti m-stream (fn [params _] (get params "stream-type" "hot")))
-
-(defmethod m-stream "cold" [params a-stream]
+(defprotocol ColdStream
+  (clean! [this])
+  (data-from [this date-string]))
+
+(defprotocol HotStream
+  (next! [this]))
+
+(defprotocol EventProcessor
+  (register-query! [this query-name f init])
+  (current-query-value [this query-name])
+  (process-event! [this ev]))
+
+(defn parse-old-date [date-string]
+  (cc/to-long
+    (clj-time.format/parse
+      (clj-time.format/formatter
+        "EEE MMM dd yyyy HH:mm:ss zZ '(GMT)'")
+      (clojure.string/replace date-string #"BST|UTC" "GMT"))))
+
+
+
+(defmulti stream (fn [_ params]
+                   (log/info (pr-str params))
+                   (get params "stream-type" "hot")))
+
+(def queries (ref {}))
+
+(defrecord DummyStream [db channel mult-channel] 
+  ColdStream
+  (clean! [this]
+    (m/with-mongo db
+      (m/destroy! :events {})))
+  (data-from [this date]
+    (let [records (m/with-mongo db
+                    (m/fetch :events :where {}))
+          query-date (if (string? date) (read-string date) date)]
+      (filter #(and
+                 (not (nil? (:server_timestamp %)))
+                 (not (empty? (:server_timestamp %)))
+                 (let [db-date (parse-old-date (:server_timestamp %))]
+                   (>= db-date query-date)))
+              records)))
+  HotStream
+  (next! [this] (go (<! channel)))
+  EventProcessor
+  (register-query! [this query-name f init]
+    (let [s (stream this {"from" "0" "stream-type" "hot-cold"})
+          running-query (ref {:query-name query-name
+                              :fn f
+                              :current-value init
+                              :processed 0
+                              :last-event nil
+                              :last-error nil
+                              :status :running})]
+      (dosync (alter queries assoc query-name running-query))
+      (go
+        (loop [current-value init current-event (<! s)]
+          (if (nil? current-event)
+            (dosync (alter running-query assoc :status :finished))
+            (let [new-value (try
+                              (f current-value current-event)
+                              (catch Exception e e))]
+              (if (instance? Exception new-value)
+                (dosync
+                  (alter running-query
+                         merge {:last-event current-event
+                                :processed (inc (:processed @running-query))
+                                :last-error new-value 
+                                :status :failed}))                
+                (do
+                  (dosync
+                    (alter running-query
+                           merge {:last-event current-event
+                                  :current-value new-value
+                                  :processed (inc (:processed @running-query))}))     
+                  (recur new-value (<! s))))))))))
+  (process-event! [this ev]
+    (go (>! channel ev))
+    (m/with-mongo db
+      (m/insert! :events ev))))
+
+(defrecord AsyncStream [db channel mult-channel]
+  ColdStream
+  (clean! [this] (riak/delete-all! db))
+  (data-from [this date-string]
+    (riak/lazy-events db date-string))
+  HotStream
+  (next! [this] (go (<! channel)))
+  EventProcessor
+  (register-query! [this query-name f init]
+    (let [s (stream this {"from" "0" "stream-type" "hot-cold"})
+          running-query (ref {:query-name query-name
+                              :fn f
+                              :current-value init
+                              :processed 0
+                              :last-event nil
+                              :last-error nil
+                              :status :running})]
+      (dosync (alter queries assoc query-name running-query))
+      (go
+        (loop [current-value init current-event (<! s)]
+          (if (nil? current-event)
+            (dosync (alter running-query assoc :status :finished))
+            (let [new-value (try
+                              (f current-value current-event)
+                              (catch Exception e e))]
+              (if (instance? Exception new-value)
+                (dosync
+                  (alter running-query
+                         merge {:last-event current-event
+                                :processed (inc (:processed @running-query))
+                                :last-error new-value 
+                                :status :failed}))                
+                (do
+                  (dosync
+                    (alter running-query
+                           merge {:last-event current-event
+                                  :current-value new-value
+                                  :processed (inc (:processed @running-query))}))
+                  (recur new-value (<! s))))))))))
+  (process-event! [this ev]
+    (go (>! channel ev))
+    (riak/store db "events" (riak/uuid) ev)))
+
+(defn async-stream [db]
+  (let [tube (chan 1)
+        tube-m (mult tube)]
+    (->AsyncStream db tube tube-m)))
+
+
+(defmethod stream "cold" [a-stream params]
   (let [ch (chan (buffer 1))
-        full-s (cold-stream a-stream (get params "from" 0))]
+        full-s (data-from a-stream (get params "from" 0))]
     (go
+      (log/info "Opening stream...")
       (loop [e (first full-s) s (rest full-s) closed? false]
         (if (nil? e)
           (do

                (recur (first s) (rest s) closed?)))))))
    ch))

(defmethod m-stream "hot-cold" [params a-stream]
  (let [ch (chan (buffer 1))
        full-s (cold-stream a-stream (get params "from" 0))]
    (go
      (loop [e (first full-s)
             s (rest full-s)
             closed? false
             last-t (System/currentTimeMillis)]
        (let [rest-s (rest s)
              new-s (if (empty? rest-s) (cold-stream a-stream last-t) s)
              last-t (if (empty? rest-s) (System/currentTimeMillis) last-t)]
          (if (nil? e)
            (do
              #_(close! ch)
              (log/info ":::::::::::::::::::: Stream depleted, switching to hot stream"))
            (if closed?
              (log/info ":::::::::::::::::::: Stream closed by peer, switching to hot stream")
              (do
                (let [closed? (not (>! ch e))]
                  (recur (first new-s) (rest new-s) closed? last-t)))))))
      (tap (:mult-channel a-stream) ch))
    ch))

(defmethod m-stream "hot" [params a-stream]
  (let [ch (chan 1)]
    (tap (:tube-m a-stream) ch)
    ch))

(defprotocol Stream
  (clean! [this])
  (stream [this params]))

(defprotocol EventProcessor
  (process-event! [this ev]))

(defrecord AsyncStream [db channel mult-channel]
  Stream
  (clean! [this] (riak/delete-all! db))
  (stream [this params] (m-stream params this))
  EventProcessor
  (process-event! [this ev]
    (go (>! channel ev))
    (riak/store db "events" (riak/uuid) ev)))

(defn async-stream [db]
  (let [tube (chan 1)
        tube-m (mult tube)]
    (->AsyncStream db tube tube-m)))


#_(def test-s (stream {"stream-type" "cold"}))

(ns eventstore.streams
  (:require [clojure.core.async :refer [go-loop go <!! <! >! chan buffer
                                        sliding-buffer mult tap close!]
             :as async]
            [clojure.tools.logging :as log]
            [somnium.congomongo :as m]
            [clj-time.coerce :as cc]
            [eventstore.riak :as riak]))

(def cold-latency 5000)
