<commit_before>(ns swarmpit.event.channel
  (:refer-clojure :exclude [list])
  (:require [org.httpkit.server :refer [send!]]
            [clojure.core.async :refer [go <! timeout]]
            [clojure.core.memoize :as memo]
            [clojure.edn :as edn]
            [clojure.walk :refer [keywordize-keys]]
            [cheshire.core :refer [generate-string]]
            [swarmpit.base64 :as base64]
            [swarmpit.event.rule :as rule]))

(def hub (atom {}))

(defn- message
  [data]
  (str "data: " (generate-string data) "\n\n"))

(defn- subscription
  [channel]
  (-> (val channel)
      (get-in [:query-params "subscription"])
      (base64/decode)
      (edn/read-string)))

(defn- subscribers
  [channel-subscription]
  (->> @hub
       (filter #(= channel-subscription (subscription %)))
       (keys)))

(defn list
  ([event]
   "Get subscribed channel list based on given event"
   (->> (filter #(rule/match? % event) rule/list)
        (map #(rule/subscription % event))
        (map #(subscribers %))
        (flatten)
        (filter #(some? %))
        (map #(str %))))
  ([]
   "Get susbcribed channel list"
   (->> (keys @hub)
        (map #(str %)))))

(defn broadcast
  [event]
  "Broadcast data to subscribers based on event subscription.<commit_msg>fixed typos<commit_after>(ns swarmpit.event.channel
  (:refer-clojure :exclude [list])
  (:require [org.httpkit.server :refer [send!]]
            [clojure.core.async :refer [go <! timeout]]
            [clojure.core.memoize :as memo]
            [clojure.edn :as edn]
            [clojure.walk :refer [keywordize-keys]]
            [cheshire.core :refer [generate-string]]
            [swarmpit.base64 :as base64]
            [swarmpit.event.rule :as rule]))

(def hub (atom {}))

(defn- message
  [data]
  (str "data: " (generate-string data) "\n\n"))

(defn- subscription
  [channel]
  (-> (val channel)
      (get-in [:query-params "subscription"])
      (base64/decode)
      (edn/read-string)))

(defn- subscribers
  [channel-subscription]
  (->> @hub
       (filter #(= channel-subscription (subscription %)))
       (keys)))

(defn list
  ([event]
   "Get subscribed channels based on given event"
   (->> (filter #(rule/match? % event) rule/list)
        (map #(rule/subscription % event))
        (map #(subscribers %))
        (flatten)
        (filter #(some? %))
        (map #(str %))))
  ([]
   "Get subscribed channels"
   (->> (keys @hub)
        (map #(str %)))))

(defn broadcast
  [event]
  "Broadcast data to subscribers based on event subscription.
{'fraction_non_alphanumeric': 0.16862745098039217, 'fraction_numerical': 0.004705882352941176, 'mean_word_length': 2.5137741046831956, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 17, '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': '3292057', 'n_tokens_mistral': 895, 'n_tokens_neox': 880, 'n_words': 206}