document.clj

;; Copyright (c) 2011-2019 Michael S. Klishin, Alex Petrov, and the ClojureWerkz Team
;;
;; Licensed under the Apache License, Version 2.0 (the "License");
;; you may not use this file except in compliance with the License.
;; You may obtain a copy of the License at
;;
;;     http://www.apache.org/licenses/LICENSE-2.0
;;
;; Unless required by applicable law or agreed to in writing, software
;; distributed under the License is distributed on an "AS IS" BASIS,
;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
;; See the License for the specific language governing permissions and
;; limitations under the License.

(ns clojurewerkz.elastisch.native.document
  (:refer-clojure :exclude [get replace count sort])
  (:require [clojurewerkz.elastisch.native :as es]
            [clojurewerkz.elastisch.native.conversion :as cnv]
            [clojurewerkz.elastisch.query  :as q]
            [clojurewerkz.elastisch.native.response :as r])
  (:import clojure.lang.IPersistentMap
           org.elasticsearch.client.Client
           (org.elasticsearch.action.get GetResponse MultiGetResponse)
           org.elasticsearch.action.count.CountResponse
           org.elasticsearch.action.delete.DeleteResponse
           org.elasticsearch.action.search.SearchResponse
           java.util.Map))

(defn ^IPersistentMap create
  "Adds document to the search index and waits for the response.
  If not given as an option, document id will be generated automatically.

  Options:

    * `:id` (string): unique document id. If not provided, it will be generated by Elasticsearch
    * `:timestamp` (string): document timestamp either as millis since the epoch,
                                         or, in the configured date format
    * `:ttl` (long): document TTL in milliseconds. Must be > 0
    * `:refresh` (boolean, default: `false`): should a refresh be executed post this index operation?
    * `:version` (long): document version
    * `:version-type` (string, default: `\"internal\"`): `\"internal\"` or `\"external\"`
    * `:content-type` (string): document content type
    * `:routing` (string): controls the shard routing of the request. Using this value to hash the shard
                         and not the id
    * `:percolate` (string): the percolate query to use to reduce the percolated queries that are going to run against this doc.
                          Can be set to `\"*\"` which means \"all queries\"
    * `:parent` (string): parent document id

  Examples:

  ```clojure
  (require '[clojurewerkz.elastisch.native.document :as doc])

  (doc/create conn \"people\" \"person\" {:first-name \"John\" :last-name \"Appleseed\" :age 28})

  (doc/create conn \"people\" \"person\" {:first-name \"John\" :last-name \"Appleseed\" :age 28} {:id \"1825c5432775b8d1a477acfae57e91ac8c767aed\"})
  ```"
  ([^Client conn index mapping-type document]
     (let [res (es/index conn (cnv/->index-request index
                                                          mapping-type
                                                          document
                                                          {:op-type "create"}))]
       (cnv/index-response->map (.actionGet res))))
  ([^Client conn index mapping-type document opts]
     (let [res (es/index conn (cnv/->index-request index
                                                          mapping-type
                                                          document
                                                          (merge opts {:op-type "create"})))]
       (cnv/index-response->map (.actionGet res)))))


(defn create-search-template
  "Adds a search template to the .scripts index the template should be
  a map of the form:
  `{:template {:filter {:term {:name \"{{name}}\"}}}}`
  templates can be referenced at search time using their given id"
  ([^Client conn ^String id ^Map document]
     (create-search-template conn "mustache" id document))
  ([^Client conn ^String language ^String id ^Map document]
     (create conn ".scripts" language document {:id id})))

(defn async-create
  "Adds document to the search index and returns a future without waiting
  for the response. Takes exactly the same arguments as [[create]]."
  ([^Client conn index mapping-type document]
     (future (create conn index mapping-type document)))
  ([^Client conn index mapping-type document opts]
     (future (create conn index mapping-type document opts))))

(defn put
  "Creates or updates a document in the search index using the provided document id
  and waits for the response."
  ([^Client conn index mapping-type id document]
     (let [res (es/index conn (cnv/->index-request index
                                              mapping-type
                                              document
                                              {:id id :op-type "index"}))]
       (cnv/index-response->map (.actionGet res))))
  ([^Client conn index mapping-type id document opts]
     (let [res  (es/index conn (cnv/->index-request index
                                               mapping-type
                                               document
                                               (merge opts {:id id :op-type "index"})))]
       (cnv/index-response->map (.actionGet res)))))

(defn put-search-template
  "Updates a search template in the .scripts index. Templates are expressed
  as maps similar to:
  `{:template {:filter {:term {:name \"{{name}}\"}}}}`"
  ([^Client connid ^String id ^Map document]
    (put-search-template connid "mustache" id document))
  ([^Client connid ^String language  ^String id ^Map document]
    (put connid ".scripts" language id document)))

(defn async-put
  "Creates or updates a document in the search index using the provided document id
  and returns a future without waiting for the response. Takes exactly the same arguments as [[put]]."
  ([^Client conn index mapping-type id document]
     (future (put conn index mapping-type id document)))
  ([^Client conn index mapping-type id document opts]
     (future (put conn index mapping-type id document opts))))

(defn upsert
  "Updates or creates a document using provided data"
  ([^Client conn index mapping-type ^String id ^Map doc]
     (upsert conn index mapping-type id doc {}))
  ([^Client conn index mapping-type ^String id ^Map doc ^Map opts]
     (let [res (es/update conn (cnv/->upsert-request index
                                                mapping-type
                                                id
                                                doc
                                                opts))]
       (cnv/update-response->map (.actionGet res)))))

(defn update-with-partial-doc
  "Updates an existing document in the search index with given partial document"
  ([^Client conn index mapping-type ^String id ^Map partial-doc]
     (update-with-partial-doc conn index mapping-type id partial-doc {}))
  ([^Client conn index mapping-type ^String id ^Map partial-doc ^Map opts]
     (let [res (es/update conn (cnv/->partial-update-request index
                                                             mapping-type
                                                             id
                                                             partial-doc
                                                             opts))]
       (cnv/update-response->map (.actionGet res)))))

(defn update-with-script
  "Updates a document using a script"
  ([^Client conn index mapping-type ^String id ^String script]
     (update-with-script conn index mapping-type id script nil nil))
  ([^Client conn index mapping-type ^String id ^String script ^Map params]
     (update-with-script conn index mapping-type id script params nil))
  ([^Client conn index mapping-type ^String id ^String script ^Map params
    {upsert :upsert :as optional-params}]
   (let [res (es/update conn
                        (cnv/->update-request index
                                              mapping-type
                                              id
                                              upsert
                                              (assoc optional-params
                                                     :script_params params
                                                     :script script)))]
     (cnv/update-response->map (.actionGet res)))))



(defn get
  "Fetches and returns a document by id or `nil` if it does not exist.
  Waits for response.

  Examples:

  ```clojure
  (require '[clojurewerkz.elastisch.native.document :as doc])

  (doc/get conn \"people\" \"person\" \"1825c5432775b8d1a477acfae57e91ac8c767aed\")
  ```"
  ([^Client conn index mapping-type id]
     (let [ft               (es/get conn (cnv/->get-request index
                                                       mapping-type
                                                       id))
           ^GetResponse res (.actionGet ft)]
       (when (.isExists res)
         (cnv/get-response->map res))))
  ([^Client conn index mapping-type id opts]
     (let [ft               (es/get conn (cnv/->get-request index
                                                       mapping-type
                                                       id
                                                       opts))
           ^GetResponse res (.actionGet ft)]
       (when (.isExists res)
         (cnv/get-response->map (.actionGet ft))))))

(defn get-search-template
  ([^Client conn ^String id]
     (get-search-template conn "mustache" id))
  ([^Client conn ^String language ^String id]
     (get conn ".scripts" language id)))

(defn async-get
  "Fetches and returns a document by id or `nil` if it does not exist.
  Returns a future without waiting."
  ([^Client conn index mapping-type id]
     (future (get conn index mapping-type id)))
  ([^Client conn index mapping-type id opts]
     (future (get conn index mapping-type id opts))))

(defn present?
  "Returns true if a document with the given id is present in the provided index
  with the given mapping type."
  [^Client conn index mapping-type id]
  (not (nil? (get conn index mapping-type id))))

(defn multi-get
  "Multi get returns only documents that are found (exist).

  Queries can passed as a collection of maps with three keys: `:_index`,
  `:_type` and `:_id`:

  ```clojure
  (doc/multi-get conn [{:_index index-name :_type mapping-type :_id \"1\"}
                       {:_index index-name :_type mapping-type :_id \"2\"}])
  ```

  2-argument version accepts an index name that eliminates the need to include
  `:_index` in every query map:

  ```clojure
  (doc/multi-get conn index-name [{:_type mapping-type :_id \"1\"}
                                  {:_type mapping-type :_id \"2\"}])
  ```

  3-argument version also accepts a mapping type that eliminates the need to include
  `:_type` in every query map:

  ```clojure
  (doc/multi-get conn index-name mapping-type [{:_id \"1\"}
                                               {:_id \"2\"}])
  ```"
  ([^Client conn queries]
     ;; example response from the REST API:
     ;; ({:_index people, :_type person, :_id 1, :_version 1, :exists true, :_source {...}})
     (let [ft                    (es/multi-get conn (cnv/->multi-get-request queries))
           ^MultiGetResponse res (.actionGet ft)
           results               (cnv/multi-get-response->seq res)]
       (filter :exists results)))
  ([^Client conn index queries]
     (let [qs                    (map #(assoc % :_index index) queries)
           ft                    (es/multi-get conn (cnv/->multi-get-request qs))
           ^MultiGetResponse res (.actionGet ft)
           results               (cnv/multi-get-response->seq res)]
       (filter :exists results)))
  ([^Client conn index mapping-type queries]
     (let [qs                    (map #(assoc % :_index index :_type mapping-type) queries)
           ft                    (es/multi-get conn (cnv/->multi-get-request qs))
           ^MultiGetResponse res (.actionGet ft)
           results               (cnv/multi-get-response->seq res)]
       (filter :exists results))))

(defn delete
  "Deletes document from the index.

  Examples:

  ```clojure
  (require '[clojurewerkz.elastisch.native.document :as doc])

  (doc/delete conn \"people\" \"person\" \"1825c5432775b8d1a477acfae57e91ac8c767aed\")
  ```"
  ([^Client conn index mapping-type id]
     (let [ft                  (es/delete conn (cnv/->delete-request index mapping-type id))
           ^DeleteResponse res (.actionGet ft)]
       (cnv/delete-response->map res)))
  ([^Client conn index mapping-type id opts]
     (let [ft                  (es/delete conn (cnv/->delete-request index mapping-type id opts))
           ^DeleteResponse res (.actionGet ft)]
       (cnv/delete-response->map res))))

(defn delete-search-template
  "Removes a search template from .scripts index"
  ([^Client conn ^String id]
     (delete-search-template conn "mustache" id))
  ([^Client conn ^String language ^String id]
     (delete conn ".scripts" language id)))

(defn count
  "Performs a count query.

  Examples:

  ```clojure
  (require '[clojurewerkz.elastisch.native.document :as doc])
  (require '[clojurewerkz.elastisch.query :as q])

  (doc/count conn \"people\" \"person\")
  (doc/count conn \"people\" \"person\" (q/prefix {:username \"appl\"}))
  ```"
  ([^Client conn index mapping-type]
     (count conn index mapping-type (q/match-all)))
  ([^Client conn index mapping-type query]
     (let [ft (es/count conn (cnv/->count-request index mapping-type {:source query}))
           ^CountResponse res (.actionGet ft)]
       (merge {:count (.getCount res)}
              (cnv/broadcast-operation-response->map res))))
  ([^Client conn index mapping-type query opts]
     (let [ft (es/count conn (cnv/->count-request index mapping-type (merge opts
                                                                       {:source query})))
           ^CountResponse res (.actionGet ft)]
       (merge {:count (.getCount res)}
              (cnv/broadcast-operation-response->map res)))))

(defn search
  "Performs a search query across one or more indexes and one or more mapping types.

  Multiple indexes or types can be passed in as a seq of strings.

  Examples:

  ```clojure
  (require '[clojurewerkz.elastisch.native.document :as doc])
  (require '[clojurewerkz.elastisch.query :as q])

  (doc/search conn \"people\" \"person\" {:query (q/prefix {:username \"appl\"})})
  ```"
  ([^Client conn index mapping-type] (search conn index mapping-type nil))
  ([^Client conn index mapping-type opts]
   (let [ft (es/search conn
                       (cnv/->search-request index mapping-type opts))
         ^SearchResponse res (.actionGet ft)]
     (cnv/search-response->seq res))))

(defn search-all-types
  "Performs a search query across one or more indexes and all mapping types.

  Multiple indexes can be passed in as a seq of strings."
  ([^Client conn index] (search-all-types conn index nil))
  ([^Client conn index opts]
   (let [ft                  (es/search conn (cnv/->search-request index nil opts))
         ^SearchResponse res (.actionGet ft)]
     (cnv/search-response->seq res))))

(defn search-all-indexes-and-types
  "Performs a search query across all indexes and all mapping types. This may put very high load on your
  Elasticsearch cluster so use this function with care."
  ([^Client conn] (search-all-indexes-and-types conn nil))
  ([^Client conn opts]
   (let [ft                  (es/search conn (cnv/->search-request [] nil opts))
         ^SearchResponse res (.actionGet ft)]
     (cnv/search-response->seq res))))

(defn scroll
  "Performs a scroll query, fetching the next page of results from a
  query given a scroll id"
  ([^Client conn scroll-id] (scroll conn scroll-id nil))
  ([^Client conn scroll-id opts]
   (let [ft                  (es/search-scroll conn (cnv/->search-scroll-request scroll-id opts))
         ^SearchResponse res (.actionGet ft)]
     (cnv/search-response->seq res))))

(defn scroll-seq
  "Returns a lazy sequence of all documents for a given scroll query"
  ([^Client conn prev-resp {:keys [search_type]}]
   (let [hits (r/hits-from prev-resp)
         scroll-id (:_scroll_id prev-resp)]
     (if (or (seq hits) (= search_type "scan"))
       (concat hits (lazy-seq (scroll-seq conn (scroll conn scroll-id {:scroll "1m"}))))
       hits)))
  ([^Client conn prev-resp]
    (scroll-seq conn prev-resp nil)))

(defn replace
  "Replaces document with given id with a new one"
  [^Client conn index mapping-type id document]
  (delete conn index mapping-type id)
  (put conn index mapping-type id document))

(defn suggest
  "Suggests similar looking terms based on a provided text by using a suggester.
  Usage:
  `(suggest es-conn \"locations\" :completion \"Stockh\" {:field \"suggest\" :size 5})`"
  [^Client conn indices ^clojure.lang.Keyword suggest-type ^String term ^IPersistentMap opts]
  (let [q (cnv/->suggest-query suggest-type term opts)
        req (-> conn
                (.prepareSuggest (cnv/->string-array indices))
                (.addSuggestion q))
        res (.. req (execute) (actionGet))]
    (cnv/suggest-response->seq res)))