<commit_before>(ns qu.cache
  "Functions to create and manipulate a cache for storing the results
of aggregations. This cache will be used to serve up aggregations to
API users without having to go through Mongo's aggregation framework.

The cache uses `clojure.core.cache`'s `CacheProtocol` so that it has a
standard interface. Unlike the caches that come with
`clojure.core.cache`, however, our `QueryCache` is stateful, as it's
backed by MongoDB. That means two instances of the cache created with
the same backing database have access to the same data."
  (:require [clj-time.core :refer [now]]
            [clojure.core.cache :as cache]
            [clojure.edn :as edn]
            [clojure.string :as str]
            [digest :refer [md5]]
            [lonocloud.synthread :as ->]
            [monger.collection :as coll]
            [monger.conversion :as conv]
            [monger.core :as mongo :refer [get-db with-db]]
            [qu.data.aggregation :as agg]
            [qu.data.result :refer [->DataResult]]
            [qu.metrics :as metrics]
            [qu.util :refer :all]
            [taoensso.timbre :as log])
  (:import (com.mongodb MongoException$DuplicateKey)))

;; Needed to interact with joda-time and Cheshire.
(require 'monger.joda-time)
(require 'monger.json)

(def ^:dynamic *wait-time* 5000)
(def ^:dynamic *work-collection* "jobs")

(defn query-to-key
  "Converts a query to a key that can be used to look up the query
  results later. The key must begin with a letter, as it will be used
  as the name of a MongoDB collection, which have to begin with a
  letter."
  [query]
  (let [squeeze #(str/replace % " " "")
        database (:database query)
        slice (:slice query)
        select (:select query)
        group (:group query)
        where (:where query)
        sqlish (-> ["SELECT" (squeeze select) "FROM" (str database "." slice)]
                   (->/when where
                     (conj "WHERE" (str/trim where)))
                   (->/when group
                     (conj "GROUP BY" (squeeze group)))
                   (->/as sqlish
                          (do (str/join " " sqlish))))]
    (str "q" (md5 sqlish))))

(defn- extract-result
  "Turn a collection + a query into results that would come from that aggregation."
  [collection query]
  (let [limit (->int (:limit query) 0)
        offset (->int (:offset query) 0)
        sort (get-in query [:mongo :sort] {})
        integerize #(if (and (float? %) (is-int? %)) (int %) %)
        integerize-row (fn [row]
                         (into {} (map (fn [[k v]] (vector k (integerize v))) row)))]
    (with-open [cursor (doto (coll/find collection {})
                         (.limit limit)
                         (.skip offset)
                         (.sort (conv/to-db-object sort)))]
      (->DataResult
       (.count cursor)
       (.size cursor)
       (map (fn [x] (-> x
                        (conv/from-db-object true)
                        (integerize-row)
                        (dissoc :_id))) cursor)))))

(defn- get-collection
  ([database query]
     (get-collection database query nil))
  ([database query not-found]
     (with-db database
       (let [collection (query-to-key query)]
		(log/info "FOUND Aggregation - ID:" collection)
         (if (coll/exists? collection)
           (do (metrics/increment "cache.hit.count")
             (extract-result collection query))
           (do (metrics/increment "cache.wait.count")
               not-found))))))

(defn add-to-cache
  "Add the specified aggregation to the cache by running it through
  MongoDB's map-reduce."
  [cache aggmap]
  (let [dataset (:dataset aggmap)<commit_msg>Moving aggregation id log inside exists check<commit_after>(ns qu.cache
  "Functions to create and manipulate a cache for storing the results
of aggregations. This cache will be used to serve up aggregations to
API users without having to go through Mongo's aggregation framework.

The cache uses `clojure.core.cache`'s `CacheProtocol` so that it has a
standard interface. Unlike the caches that come with
`clojure.core.cache`, however, our `QueryCache` is stateful, as it's
backed by MongoDB. That means two instances of the cache created with
the same backing database have access to the same data."
  (:require [clj-time.core :refer [now]]
            [clojure.core.cache :as cache]
            [clojure.edn :as edn]
            [clojure.string :as str]
            [digest :refer [md5]]
            [lonocloud.synthread :as ->]
            [monger.collection :as coll]
            [monger.conversion :as conv]
            [monger.core :as mongo :refer [get-db with-db]]
            [qu.data.aggregation :as agg]
            [qu.data.result :refer [->DataResult]]
            [qu.metrics :as metrics]
            [qu.util :refer :all]
            [taoensso.timbre :as log])
  (:import (com.mongodb MongoException$DuplicateKey)))

;; Needed to interact with joda-time and Cheshire.
(require 'monger.joda-time)
(require 'monger.json)

(def ^:dynamic *wait-time* 5000)
(def ^:dynamic *work-collection* "jobs")

(defn query-to-key
  "Converts a query to a key that can be used to look up the query
  results later. The key must begin with a letter, as it will be used
  as the name of a MongoDB collection, which have to begin with a
  letter."
  [query]
  (let [squeeze #(str/replace % " " "")
        database (:database query)
        slice (:slice query)
        select (:select query)
        group (:group query)
        where (:where query)
        sqlish (-> ["SELECT" (squeeze select) "FROM" (str database "." slice)]
                   (->/when where
                     (conj "WHERE" (str/trim where)))
                   (->/when group
                     (conj "GROUP BY" (squeeze group)))
                   (->/as sqlish
                          (do (str/join " " sqlish))))]
    (str "q" (md5 sqlish))))

(defn- extract-result
  "Turn a collection + a query into results that would come from that aggregation."
  [collection query]
  (let [limit (->int (:limit query) 0)
        offset (->int (:offset query) 0)
        sort (get-in query [:mongo :sort] {})
        integerize #(if (and (float? %) (is-int? %)) (int %) %)
        integerize-row (fn [row]
                         (into {} (map (fn [[k v]] (vector k (integerize v))) row)))]
    (with-open [cursor (doto (coll/find collection {})
                         (.limit limit)
                         (.skip offset)
                         (.sort (conv/to-db-object sort)))]
      (->DataResult
       (.count cursor)
       (.size cursor)
       (map (fn [x] (-> x
                        (conv/from-db-object true)
                        (integerize-row)
                        (dissoc :_id))) cursor)))))

(defn- get-collection
  ([database query]
     (get-collection database query nil))
  ([database query not-found]
     (with-db database
       (let [collection (query-to-key query)]
         (if (coll/exists? collection)
           (do (log/info "FOUND Aggregation - ID:" collection)
			(metrics/increment "cache.hit.count")
             (extract-result collection query))
           (do (metrics/increment "cache.wait.count")
               not-found))))))

(defn add-to-cache
  "Add the specified aggregation to the cache by running it through
  MongoDB's map-reduce."
  [cache aggmap]
  (let [dataset (:dataset aggmap)
{'fraction_non_alphanumeric': 0.12258151806250846, 'fraction_numerical': 0.0021647950209714517, 'mean_word_length': 2.1032745591939546, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7102653', 'n_tokens_mistral': 2256, 'n_tokens_neox': 2137, 'n_words': 763}