 ;;
 ;; The next section contains a series of path matching functions
 
-
-(defn prepare-path-elem-re
-  "Each graphite path elem may contain '*' wildcards, this
-   functions yields a regexp pattern fro this format"
-  [e]
-  (re-pattern (format "^%s$" (str/replace e "*" ".*"))))
-
-(defn prepare-path-query
-  "For a complete path query, yield a list of regexp pattern
-   for each element"
-  [q]
-  (map prepare-path-elem-re (str/split q #"\.")))
-
-(defn path-matches?
-  "Predicate testing a path against a query. partial? determines
-   whether the query should be treated as a prefix query or an
-   absolute query"
-  [partial? query path]
-  (let [path-elems (str/split path #"\.")]
-    (and (or partial? (= (count query) (count path-elems)))
-         (every? seq (map re-matches query path-elems)))))
-
-(defn truncate-path
-  "When doing prefix searches, we're only interested in returning
-   prefixes, this function yields a map of two keys:
-
-   * leaf: indicates whether this is a prefix or an actual point
-   * path: the prefix"
-  [query path]
-  (let [depth     (count (str/split query #"\."))
-        truncated (str/join "." (take depth (str/split path #"\.")))]
-    {:leaf (= truncated path) :path truncated}))
-
-(defn find-paths
-  "Find either prefix or absolute matches for a query"
-  [partial? query]
-  (let [depth    (count (str/split query #"\."))]
-    (filter (partial path-matches? partial? (prepare-path-query query))
-            @path-db)))
-
-(defn find-prefixes
-  "Find prefix matches for a query and then format for graphite output"
-  [query]
-  (->> (find-paths true query)
-       (map (partial truncate-path query))
-       (set)
-       (sort-by :path)))
-
-(defn update-path-db-every
-  "At each interval, fetch all known paths, and store the
-   resulting set in path-db"
-  [session interval]
-  (while true
-    (try
-      (->> (alia/execute session pathq)
-           (map :path)
-           (set)
-           (reset! path-db))
-      (catch Exception e
-        (error e "could not update path database")))
-    (Thread/sleep (* interval 1000))))
-
-(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 "connecting to cassandra cluster")
-  (let [session (-> (alia/cluster cluster) (alia/connect))]
-    (try (alia/execute session (useq keyspace))
-         (future (update-path-db-every session 60))
-         session
-         (catch Exception e
-           (error e "could not connect to cassandra cluster")))))
-
-(defn channel-for
-  "Yields a lamina channel which expects lists of metrics and will
-   insert each for every resolution"
-  [session]
-  (let [ch    (channel)
-        query (insertq session)]
-    (receive-all
-     ch
-     (fn [payload]
-       (doseq [{:keys [metric path time rollup period ttl]} payload]
-         (alia/execute
-          session query
-          :values [(int ttl) [metric] (int rollup) (int period) path time]))))
-    ch))
 
 (defmulti aggregate-with
   "This transforms a raw list of points according to the provided aggregation

    "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]

                 (sort-by :time)
                 (map :metric))))

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

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



      )))
