                  (sort-by :time)
                  (map :metric))))
 
-(defn fetch
-  "Fetch metrics for a given resolution. Once fetched, formats
-   metrics in a way graphite can easily consume"
-  [session agg paths rollup period from to]
-  (debug "fetching paths from store: " paths rollup period from to
-         (max-points paths rollup from to))
+(defn cassandra-metric-store
+  "Connect to cassandra and start a path fetching thread.
+   The interval is fixed for now, at 1minute"
+  [{:keys [keyspace cluster hints]
+    :or   {hints {:replication {:class "SimpleStrategy"
+                                :replication_factor 1}}}}]
+  (info "creating cassandra metric store with in-memory path index")
+  (let [session (-> (alia/cluster cluster) (alia/connect))
+        index   (search/memory-index)
+        insert! (insertq session)
+        fetch!  (fetchq session)]
 
-  (if-let [data (and (seq paths)
-                     (->> (alia/execute
-                           session (fetchq session)
-                           :values [paths (int rollup) (int period) from to
-                                    (max-points paths rollup from to)])
-                          (map (partial aggregate-with (keyword agg)))
-                          (seq)))]
-    (let [min-point  (:time (first data))
-          max-point  (-> to (quot rollup) (* rollup))
-          nil-points (->> (range min-point (inc max-point) rollup)
-                          (map (fn [time] {time [{:time time}]}))
-                          (reduce merge {}))
-          by-path    (->> (group-by :path data)
-                          (map (partial fill-in nil-points))
-                          (reduce merge {}))]
-      {:from min-point
-       :to   max-point
-       :step rollup
-       :series by-path})
-    {:from from
-     :to to
-     :step rollup
-     :series {}}))
+    (try (alia/execute session (useq keyspace))
+         (catch Exception e
+           (error e "could not connect to cassandra cluster")))
+
+    (reify
+      Metricstore
+      (channel-for [this]
+        (let [ch    (channel)]
+          (receive-all
+           ch
+           (fn [payload]
+             (doseq [{:keys [metric path time rollup period ttl]} payload]
+               (alia/execute
+                session insert!
+                :values [(int ttl) [metric] (int rollup) (int period) path time]))))
+          ch))
+      (insert [this ttl data tenant rollup period path time]
+        (alia/execute
+         session
+         insert!
+         :values [ttl data tenant rollup period path time]))
+      (fetch [this agg paths tenant rollup period from to]
+        (debug "fetching paths from store: " paths rollup period from to
+               (max-points paths rollup from to))
+        (if-let [data (and (seq paths)
+                           (->> (alia/execute
+                                 session fetch!
+                                 :values [paths (int rollup) (int period) from to
+                                          (max-points paths rollup from to)])
+                                (map (partial aggregate-with (keyword agg)))
+                                (seq)))]
+          (let [min-point  (:time (first data))
+                max-point  (-> to (quot rollup) (* rollup))
+                nil-points (->> (range min-point (inc max-point) rollup)
+                                (map (fn [time] {time [{:time time}]}))
+                                (reduce merge {}))
+                by-path    (->> (group-by :path data)
+                                (map (partial fill-in nil-points))
+                                (reduce merge {}))]
+            {:from min-point
+             :to   max-point
+             :step rollup
+             :series by-path})
+          {:from from
+           :to to
+           :step rollup
+           :series {}}))
+
+
+
+      )))

    "AND time >= ? AND time <= ? ORDER BY time ASC LIMIT ?;")))


(def
  ^{:doc "This query returns only paths, to get a list of
metrics. Right now it is way too expensive and should be computed differently"}
  pathq "SELECT distinct tenant, path, rollup, period from metric;")

(defn useq
  "Yields a cassandra use statement for a keyspace"
  [keyspace]

  (:require [org.spootnik.cyanite.config :as config]
            [clojure.string              :as str]
            [qbits.alia                  :as alia]
            [clucy.core                  :as search]
            [clojure.tools.logging       :refer [error info debug]]
            [lamina.core                 :refer [channel receive-all]]))

(defprotocol Metricstore
  (insert [this ttl data tenant rollup period path time])
  (channel-for [this])
  (fetch [this agg paths tenant rollup period from to]))

;;
;; The following contains necessary cassandra queries. Since
