Site transitions processing

; priority-map-by from [org.clojure/data.priority-map "0.0.10"]
; see https://github.com/clojure/data.priority-map

(defonce *state (atom {:prioritized-sessions (priority-map-by <)
                       :urls {}}))

(defn timestamp-to-epoch [ts]
    ; ts string -> unix time
    )

(defn get-now
    ; current moment in unix time
    )

(defn send-to-external-sys [sid-2-urls]
    (defn send-urls [sid urls]
        ; create HTTP-request to the external system with (sid, urls) in body
        )
    ; return vector of results of sendings
    (map send-urls sid-2-urls))

(defn accept-site-transition! [site-trans]
    (let [[sid url ts] site-trans
          old-sessions (:prioritized-sessions @*state)
          old-urls (:urls @*state)
          new-state {:prioritized-sessions (assoc old-sessions sid (timestamp-to-epoch ts))
                     :urls (merge-with into old-urls {sid [url]})}]
        (reset! *state new-state)))

; if it turns out that access to *state slows down seriously speed of the processing of an incoming request
; then we will have to group incoming site transitions into bulks by m elements (could be got empirically).
(defn map-sid-to-latest-ts [trans-by-sid]
    ; {sid1 [[sid1 url1 ts1]
    ;        [sid1 url2 ts2]],
    ;  sid2 [[sid2 url3 ts3]]} -> {sid1 ts2, sid2 ts3}
    (reduce merge {}
        (map (fn [pair]
                  (let [[sid transitions] pair
                        timestamps (map #(timestamp-to-epoch (last %))
                                        transitions)
                        latest-ts (reduce max timestamps)]
                    {sid latest-ts}))
             trans-by-sid)))

(defn map-sid-to-urls [trans-by-sid]
    ; {sid1 [[sid1 url1 ts1]
    ;        [sid1 url2 ts2]],
    ;  sid2 [[sid2 url3 ts3]]} -> {sid1 [url1 url2], sid2 [url3]}
    (reduce merge {}
            (map (fn [pair]
                    (let [[sid transitions] pair
                          urls (vec (map (fn [_ url _] url) transitions))]
                        {sid urls}))
                 trans-by-sid)))

(defn accept-site-transition-chunk! [chunk]
    (let [old-sessions (:prioritized-sessions @*state)
          old-urls (:urls @*state)
          trans-by-sid (group-by first chunk)
          sids-to-latest-ts (map-sid-to-latest-ts trans-by-sid)
          sids-to-urls (map-sid-to-urls trans-by-sid)
          new-state {:prioritized-sessions (merge old-sessions sids-to-latest-ts)
                     :urls (merge-with into old-urls sids-to-urls)}]
        (reset! *state new-state)))


(defn find-urls-to-send
    (let [sessions (:prioritized-sessions @*state)
          urls (:urls @*state)
          now (get-now)
          sids (->> sessions
                    (take-while (fn [[sid ts]] (>= (- now ts))))
                    (map first))]
        {:used-sids sids
         :send-data (select-keys urls sids)}))

(defn wipe-used-sids! [sids]
    (let [old-sessions (:prioritized-sessions @*state)
          old-urls (:urls @*state)
          new-state {:prioritized-sessions (dissoc old-sessions sids)
                     :urls (dissoc old-urls sids)}]
        (reset! *state new-state)))

(defn periodic-task
    (let [search-res (find-urls-to-send)]
        (send-to-external-sys (:send-data search-res))
        (wipe-used-sids! (:used-data search-res))))

; this could be optimized by executing send-to-external-sys in a separate asynchronous task, as well as wipe-used-sids!
; asynchronous calls could be organized with a help of https://github.com/clojure/core.async