<commit_before>(ns org.teamweek.slackbot
  (:require [clojure.core.async :as async]
            [clojure.string :as str]
            [datomic.api :as d]
            [aleph.http :as http]
            [manifold.stream :as stream]
            [cheshire.core :as json]
            [clojurewerkz.quartzite.scheduler :as scheduler]
            [clojurewerkz.quartzite.jobs :as jobs :refer (defjob)]
            [clojurewerkz.quartzite.triggers :as triggers]
            [clojurewerkz.quartzite.schedule.cron :as cron]
            [clojurewerkz.quartzite.conversion :as conversion]
            [taoensso.timbre :as log]
            [byte-streams])
  (:gen-class))

(def config (atom {}))

(defn connect-url [token]
  (str "https://slack.com/api/rtm.start?token=" token))

(defn keep-alive [ws-conn shutdown?]
  (future (while (not @shutdown?)
            (Thread/sleep 5000)
            @(stream/put! ws-conn
                          (json/generate-string {:id (rand-int 1e6)
                                                 :type "ping"})))))

(defn find-username [conn user-id]
  (if-let [username (some #(if (= (:id %) user-id)
                             (:name %))
                          (:users conn))]
    username
    (throw (ex-info "No such user" {:user-id user-id}))))

(defn direct-message? [msg]
  (.startsWith (:channel msg) "D"))

(defn route-messages [conn incoming-messages shutdown?]
  (future (while (not @shutdown?)
            (let [msg (json/parse-string @(stream/take! (:ws-conn conn)) true)]
              (condp = (:type msg)
                "message" (when (and (not= (:user msg) (:id (:self conn)))
                                     (direct-message? msg))
                            (async/put! incoming-messages
                                        (assoc msg
                                               :username (find-username conn (:user msg)))))
                (do #_ignore))))))

(defn websocket-client [url shutdown?]
  (let [conn @(http/websocket-client url)
        first-msg (json/parse-string @(stream/take! conn))]
    (when-not (= first-msg {"type" "hello"})
      (throw (ex-info "Unable to connect" first-msg)))
    (keep-alive conn shutdown?)
    conn))

(defn connect [token]
  (try
    (log/info "Attempting to connect to" token)
    (let [conn-data (-> (str "https://slack.com/api/rtm.start?token=" token)
                      (http/get)
                      deref
                      :body
                      byte-streams/to-string
                      (json/parse-string true))
          shutdown? (atom false)
          incoming-messages (async/chan)
          dm-pub (async/pub incoming-messages
                            (fn topic-fn [msg]
                              (:username msg)))
          ws-conn (websocket-client (:url conn-data)
                                    shutdown?)
          shutdown-fn (fn []
                        (log/info "Shutting down" (:url conn-data))
                        (reset! shutdown? true)
                        (async/close! incoming-messages)
                        (stream/close! ws-conn)
                        (log/info "Successfully shut down" (:url conn-data)))
          conn (assoc conn-data
                      :ws-conn ws-conn
                      :incoming-messages incoming-messages
                      :dm-pub dm-pub
                      :shutdown-fn shutdown-fn)]
      (route-messages conn incoming-messages shutdown?)
      (log/info "Successfully connected to" token)
      conn)
    (catch Exception e
      (log/error e "Failed to connect to" token)))
  )

(defn find-user-dm-id [conn user]
  (let [user-id (some #(if (= (:name %) user)
                         (:id %))
                      (:users conn))
        dm-id (some #(if (= (:user %) user-id)
                       (:id %))
                    (:ims conn))]
    (or dm-id
        (throw (ex-info "No such user" {:user user})))))

(defn send-to-user! [conn user msg]
  @(stream/put! (:ws-conn conn)
                (json/generate-string {:id (rand-int 1e6)
                                       :type "message"
                                       :channel (find-user-dm-id conn user)
                                       :text msg})))

(defn find-channel-id [conn channel]
  (if-let [id (some #(if (= (:name %) channel)
                       (:id %))
                    (:channels conn))]
    id
    (throw (ex-info "No such channel" {:channel channel}))))

(defn send-to-channel! [conn channel msg]
  @(stream/put! (:ws-conn conn)
                (json/generate-string {:id (rand-int 1e6)
                                       :type "message"
                                       :channel (find-channel-id conn channel)
                                       :text msg})))

(defn send-questionnaire
  [conn token username questions]
  (try
    (send-to-user! conn username
                   (format "Greetings %s. I have a few questions for you." username))
    (let [dm-pub (:dm-pub conn)
          answer-chan (async/chan)
          dm-sub (async/sub dm-pub username answer-chan)
          qas (reduce
               (fn [question-answers question]
                 (log/infof "Sending \"%s\" to %s" (:question/text question) username)
                 (send-to-user! conn username (:question/text question))
                 (conj question-answers
                       {:ts (java.util.Date.)
                        :team (:name (:team conn))
                        :domain (:domain (:team conn))
                        :user username
                        :question-id (:db/id question)
                        :answer (:text (async/<!! answer-chan))}))
               []
               questions)]
      (send-to-user! conn username
                     (format "Thank you for your answers. You can view your teams data now or later at https://teamweek.org/search?t=%s"
                             token))
      (log/infof "Questionnaire completed for %s. %s questions answered" username (count qas))
      (async/unsub dm-pub username answer-chan)
      (async/close! answer-chan)
      qas)
    (catch Exception e
      (log/errorf e "Failed to send questions %s to %s" (pr-str questions) username)
      (log/error "Connection: %s" (pr-str conn))
      (throw e))))


(defn submit-answers [db-conn domain user answers]
  (try
    (log/infof "Submitting %s answers to %s for %s" (count answers) domain user)
    (let [db (d/db db-conn)
          user-eid (d/q '[:find ?member .
                          :in $ ?domain ?user
                          :where
                          [?team :team/domain ?domain]
                          [?team :team/members ?member]
                          [?member :member/name ?user]]
                        db domain user)
          tx (for [answer answers]
               {:db/id (d/tempid :db.part/user)
                :answer/author user-eid<commit_msg>Add a ten minute timeout<commit_after>(ns org.teamweek.slackbot
  (:require [clojure.core.async :as async]
            [clojure.string :as str]
            [datomic.api :as d]
            [aleph.http :as http]
            [manifold.stream :as stream]
            [cheshire.core :as json]
            [clojurewerkz.quartzite.scheduler :as scheduler]
            [clojurewerkz.quartzite.jobs :as jobs :refer (defjob)]
            [clojurewerkz.quartzite.triggers :as triggers]
            [clojurewerkz.quartzite.schedule.cron :as cron]
            [clojurewerkz.quartzite.conversion :as conversion]
            [taoensso.timbre :as log]
            [byte-streams])
  (:gen-class))

(def config (atom {}))

(defn connect-url [token]
  (str "https://slack.com/api/rtm.start?token=" token))

(defn keep-alive [ws-conn shutdown?]
  (future (while (not @shutdown?)
            (Thread/sleep 5000)
            @(stream/put! ws-conn
                          (json/generate-string {:id (rand-int 1e6)
                                                 :type "ping"})))))

(defn find-username [conn user-id]
  (if-let [username (some #(if (= (:id %) user-id)
                             (:name %))
                          (:users conn))]
    username
    (throw (ex-info "No such user" {:user-id user-id}))))

(defn direct-message? [msg]
  (.startsWith (:channel msg) "D"))

(defn route-messages [conn incoming-messages shutdown?]
  (future (while (not @shutdown?)
            (let [msg (json/parse-string @(stream/take! (:ws-conn conn)) true)]
              (condp = (:type msg)
                "message" (when (and (not= (:user msg) (:id (:self conn)))
                                     (direct-message? msg))
                            (async/put! incoming-messages
                                        (assoc msg
                                               :username (find-username conn (:user msg)))))
                (do #_ignore))))))

(defn websocket-client [url shutdown?]
  (let [conn @(http/websocket-client url)
        first-msg (json/parse-string @(stream/take! conn))]
    (when-not (= first-msg {"type" "hello"})
      (throw (ex-info "Unable to connect" first-msg)))
    (keep-alive conn shutdown?)
    conn))

(defn connect [token]
  (try
    (log/info "Attempting to connect to" token)
    (let [conn-data (-> (str "https://slack.com/api/rtm.start?token=" token)
                      (http/get)
                      deref
                      :body
                      byte-streams/to-string
                      (json/parse-string true))
          shutdown? (atom false)
          incoming-messages (async/chan)
          dm-pub (async/pub incoming-messages
                            (fn topic-fn [msg]
                              (:username msg)))
          ws-conn (websocket-client (:url conn-data)
                                    shutdown?)
          shutdown-fn (fn []
                        (log/info "Shutting down" (:url conn-data))
                        (reset! shutdown? true)
                        (async/close! incoming-messages)
                        (stream/close! ws-conn)
                        (log/info "Successfully shut down" (:url conn-data)))
          conn (assoc conn-data
                      :ws-conn ws-conn
                      :incoming-messages incoming-messages
                      :dm-pub dm-pub
                      :shutdown-fn shutdown-fn)]
      (route-messages conn incoming-messages shutdown?)
      (log/info "Successfully connected to" token)
      conn)
    (catch Exception e
      (log/error e "Failed to connect to" token)))
  )

(defn find-user-dm-id [conn user]
  (let [user-id (some #(if (= (:name %) user)
                         (:id %))
                      (:users conn))
        dm-id (some #(if (= (:user %) user-id)
                       (:id %))
                    (:ims conn))]
    (or dm-id
        (throw (ex-info "No such user" {:user user})))))

(defn send-to-user! [conn user msg]
  @(stream/put! (:ws-conn conn)
                (json/generate-string {:id (rand-int 1e6)
                                       :type "message"
                                       :channel (find-user-dm-id conn user)
                                       :text msg})))

(defn find-channel-id [conn channel]
  (if-let [id (some #(if (= (:name %) channel)
                       (:id %))
                    (:channels conn))]
    id
    (throw (ex-info "No such channel" {:channel channel}))))

(defn send-to-channel! [conn channel msg]
  @(stream/put! (:ws-conn conn)
                (json/generate-string {:id (rand-int 1e6)
                                       :type "message"
                                       :channel (find-channel-id conn channel)
                                       :text msg})))
(def ten-minutes 600000)

(defn send-questionnaire
  [conn token username questions]
  (try
    (send-to-user! conn username
                   (format "Greetings %s. I have a few questions for you." username))
    (let [dm-pub (:dm-pub conn)
          answer-chan (async/chan)
          dm-sub (async/sub dm-pub username answer-chan)
          qas (reduce
               (fn [question-answers question]
                 (log/infof "Sending \"%s\" to %s" (:question/text question) username)
                 (send-to-user! conn username (:question/text question))
                 (conj question-answers
                       {:ts (java.util.Date.)
                        :team (:name (:team conn))
                        :domain (:domain (:team conn))
                        :user username
                        :question-id (:db/id question)
                        :answer (:text (async/alt!!
                                         (async/<!! answer-chan) ([text] text)
                                         (async/timeout ten-minutes) ([_] "")))}))
               []
               questions)]
      (send-to-user! conn username
                     (format "Thank you for your answers. You can view your teams data now or later at https://teamweek.org/search?t=%s"
                             token))
      (log/infof "Questionnaire completed for %s. %s questions answered" username (count qas))
      (async/unsub dm-pub username answer-chan)
      (async/close! answer-chan)
      qas)
    (catch Exception e
      (log/errorf e "Failed to send questions %s to %s" (pr-str questions) username)
      (log/error "Connection: %s" (pr-str conn))
      (throw e))))


(defn submit-answers [db-conn domain user answers]
  (try
    (log/infof "Submitting %s answers to %s for %s" (count answers) domain user)
    (let [db (d/db db-conn)
          user-eid (d/q '[:find ?member .
                          :in $ ?domain ?user
                          :where
                          [?team :team/domain ?domain]
                          [?team :team/members ?member]
                          [?member :member/name ?user]]
                        db domain user)
          tx (for [answer answers]
               {:db/id (d/tempid :db.part/user)
                :answer/author user-eid
{'fraction_non_alphanumeric': 0.12746564124474827, 'fraction_numerical': 0.0018514562415438297, 'mean_word_length': 1.381549940647787, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 6, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 30, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2477888', 'n_tokens_mistral': 4148, 'n_tokens_neox': 3875, 'n_words': 1073}