(ns filter.core
  (:require [clojure.string :as s]))


; Написать HTTP сервис позволяющий фильтровать сообщения из Kafka. Пользователь может добавить фильтр и система начинает запоминать все новые сообщения, которые подходят под фильтр (регистронезависимое вхождение строки) и потом отдавать по запросу.

; API должно поддерживать такие операции:

; Добавление нового фильтра для топика
; POST /filter
; {"topic": "books",
;  "q": "sicp"}
; Получение списка фильтров
; GET /filter
; Получение сообщений, накопленных фильтром со времени его создания
; GET /filter?id=1
; Удаление фильтра
; DELETE /filter
; {"id": 1}
; Для упрощения локального тестирования и разработки можно запускать зукипер и кафку прямо в процессе сервиса. Это можно сделать так.

(def datas (atom ()))

(defn filter-find [id]
  (fn [row] (= (:id row) id)))

(defn add-filter [id filter]
  (swap! datas conj {:id id :data filter}))

(defn remove-filter [id]
  (swap! datas (fn [fs] (remove (filter-find id) fs))))

(defn filter-by [id]
  (->> @datas
       (filter (filter-find id))
       (first)
       :data))

(defn all-filters []
  @datas)

(add-filter 1 {:topic "books"})
(add-filter 2 {:topic "shops"})

(remove-filter 2)

(filter-by 2)

(def kafka-msg-1 {:topic "books", :name "Clojure", :author "Rich Hickey"})
(def kafka-msg-2 {:topic "books", :name "F#", :author "Done Syme"})

(def filter-sample {:topic "books" :name "b" :author "rich"})

(into [] kafka-msg-1)
(into [] filter-sample)

(defn pair-check [msg-pair filter-pair]
  (let [[_ msg-value] msg-pair
        [_ filter-value] filter-pair]
    (and (and (string? msg-value) (string? filter-value)) (s/includes? msg-value filter-value))))

(defn find [item coll]
  (->> coll
       (filter #(= (first %) (first item)))
       (first)))

(defn left-join-map [left-map right-map]
  (let [left-vector (into [] left-map)
        right-vector (into [] right-map)]
    (map #(->> right-vector
               (find %)
               (pair-check %)) left-vector)))

(defn check-filter [msg filter is-strict]
  (let [results (left-join-map msg filter)]
    (if (true? is-strict)
      (every? true? results)
      (some true? results))))

(check-filter kafka-msg-1 filter-sample false)