Merge pull request #127 from clojureverse/arne/emoji-deploy-followup

Arne/emoji deploy followup

Author: plexus

repl/emoji_cleanup.clj

@@ -0,0 +1,42 @@
+(ns repl.emoji-cleanup
+  (:require [clojurians-log.datomic :as d]
+            [clojurians-log.repl :as repl]
+            [clojurians-log.slack-api :as slack]))
+
+;; Our :emoji/shortcode property did not originally have an index, meaning we
+;; kept creating more entities for emojis, instead of upserting them.
+;;
+;; To fix this, first retract all shortcode attributes, then transact the schema
+;; so it gets the index, then re-import
+
+;; Do the following before deploying the new schema changes:
+
+(def shortcodes (d/q '[:find ?i ?s :where [?i :emoji/shortcode ?s]] (repl/db)))
+(def tx-data (for [[i s] shortcodes]
+               [:db/retract i :emoji/shortcode s]))
+
+(run! (partial d/transact (repl/conn)) (partition-all 1000 tx-data))
+
+@(d/transact (repl/conn) [{:db/ident       :emoji/shortcode
+                           :db/valueType   :db.type/string
+                           :db/cardinality :db.cardinality/one
+                           :db/unique      :db.unique/identity}])
+
+(slack/import-emojis! (repl/conn))
+
+;; After deploying the new version, re-import the data to get the emoji reactions in there
+
+(def result (load-files! (log-files))) ;; [(volatile count) (promise)]
+
+(future
+  (while (not (realized? (second result)))
+    (println @(first result))
+    (Thread/sleep 2000))
+  (println :done))
+
+(group-by second
+          (map (juxt :reaction/type
+                     (comp :emoji/shortcode :reaction/emoji))
+               (:reaction/_message
+                (datomic.api/entity (repl/db)
+                                    [:message/key "C0GLTDB2T--1608789973.125700"]))))

src/clojurians_log/components/datomic_schema.clj

@@ -6,10 +6,10 @@
 (defrecord DatomicSchema [datomic]
   component/Lifecycle
   (start [this]
-    (d/transact (:conn datomic)
-                (if d/cloud?
-                  (map #(dissoc % :db/index) schema/full-schema)
-                  schema/full-schema)))
+    @(d/transact (:conn datomic)
+                 (if d/cloud?
+                   (map #(dissoc % :db/index) schema/full-schema)
+                   schema/full-schema)))
   (stop [this]))
 
 (defn new-datomic-schema []

src/clojurians_log/db/import.clj

@@ -94,18 +94,26 @@
   #:emoji {:shortcode (name shortcode)
            :url       url})
 
-(defmethod event->tx ["reaction_added" nil] [{:keys [user item reaction item_user event_ts ts]}]
-  {:reaction/type "reaction_added"
-   :reaction/emoji {:emoji/shortcode reaction}
+;; TODO: deal with reactions on files (I guess this will depend on us actually dealing with files in the first place :))
+;; {:reaction "joy", :event_ts "1521818444.000850", :item {:type "file", :file "F9W0B0LHM"}, :user "U06P56UUB", :item_user "U4E5W80P7", :type "reaction_added"}
+
+(defn- reaction-entity [{:keys [user item reaction ts]}]
+  {:reaction/emoji {:emoji/shortcode reaction}
    :reaction/ts ts
    :reaction/user [:user/slack-id user]
    :reaction/message {:message/key (message-key item)}})
 
-(defmethod event->tx ["reaction_removed" nil] [{:keys [user item reaction item_user event_ts ts]}]
-  ;; Placeholder just to show that we're getting some data.
-  ;; TODO: return Datomic transaction data to retract a reaction entity
-  (println "-" reaction)
-  nil)
+(defmethod event->tx ["reaction_added" nil] [{:keys [item] :as msg}]
+  (when (and (:channel item) (:ts item)) ; exclude reactions on things other than messages
+    (assoc
+      (reaction-entity msg)
+      :reaction/type "reaction_added")))
+
+(defmethod event->tx ["reaction_removed" nil] [{:keys [item] :as msg}]
+  (when (and (:channel item) (:ts item)) ; exclude reactions on things other than messages
+    (assoc
+      (reaction-entity msg)
+      :reaction/type "reaction_removed")))
 
 (defn lines-reducible [^BufferedReader rdr]
   (reify clojure.lang.IReduceInit
@@ -133,9 +141,11 @@
                     (rf acc (persistent! @part))
                     (vreset! part (transient []))
                     (vreset! part-keys (transient #{}))))
+          message-key #(or (:message/key %)
+                           (:message/key (:reaction/message %)))
           append (fn [msg]
                    (conj! @part msg)
-                   (conj! @part-keys (:message/key msg)))]
+                   (conj! @part-keys (message-key msg)))]
       (fn
         ([]
          (rf))

src/clojurians_log/db/queries.clj

@@ -52,8 +52,9 @@
          [:message/text
           :message/ts
           :message/thread-ts
-          {:reaction/_message [:reaction/user 
-                               :reaction/type {:reaction/emoji [*]}]}
+          {:reaction/_message [:reaction/user
+                               :reaction/type
+                               {:reaction/emoji [*]}]}
           {:message/user [:user/name
                           :user/slack-id
                           :user-profile/real-name

src/clojurians_log/repl.clj

@@ -19,7 +19,6 @@
    (uncaughtException [_ thread throwable]
      (println (.getMessage throwable)))))
 
-
 (defn read-edn [filepath]
   (-> filepath
       slurp
@@ -48,11 +47,11 @@
   (slack/import-emojis! (conn))
   (let [channel->db-id (q/channel-id-map (d/db (conn)))
         channels       (mapv import/channel->tx (slack/channels))]
-    (d/transact (conn) (map (fn [{slack-id :channel/slack-id :as ch}]
-                              (if-let [db-id (channel->db-id slack-id)]
-                                (assoc ch :db/id db-id)
-                                ch))
-                            channels))))
+    @(d/transact (conn) (map (fn [{slack-id :channel/slack-id :as ch}]
+                               (if-let [db-id (channel->db-id slack-id)]
+                                 (assoc ch :db/id db-id)
+                                 ch))
+                             channels))))
 
 (defn build-indexes! []
   (q/build-indexes! (d/db (conn))))
@@ -100,13 +99,27 @@
   (run! load-log-file! (log-files (java.io.File. directory "logs")))
   (build-indexes!))
 
+(defn files-from
+  "Get a sequence of log files starting ata given date"
+  [date]
+  (->> (log-files "/home/arne/github/clojurians-log/logs")
+       (drop-while #(not (clojure.string/starts-with? (.getName %) date)))))
+
 (defn load-from
   "Load log files starting from a certain date (a string like \"2019-05-20\")"
   [date]
-  (->> (log-files)
-       (drop-while #(not (clojure.string/starts-with? (.getName %) date)))
+  (->> (files-from date)
        (run! load-log-file!)))
 
+(defn wrap-catch [f]
+  (fn [& args]
+    (try
+      (apply f args)
+      (catch Throwable t
+        (let [ex-sym (gensym "ex-")]
+          (intern *ns* ex-sym t)
+          (println (str f " threw " (class t) " see " ex-sym)))))))
+
 (def file->tx
   "Transducer which consumes files and produces transaction data"
   (comp (mapcat #(import/lines-reducible (io/reader %)))
@@ -116,8 +129,7 @@
                    (println "Error decoding JSON: " %)
                    (println e)
                    nil)))
-        (filter #(= (:type %) "message"))
-        (keep import/event->tx)))
+        (keep (wrap-catch import/event->tx))))
 
 (def tx-thread-count
   "The number of threads to use for processing transactions"
@@ -135,19 +147,20 @@
     (mod-hash
      (if (vector? msg)
        (second (second msg))
-       (:message/key msg)))))
+       (or (:message/key msg)
+           (:message/key (:reaction/message msg)))))))
 
 (defn load-files!
   "Bulk import a set of files (e.g. from (log-files)), uses multiple threads to speed things up"
   [files]
-  (let [tx-chs (into [] (repeatedly tx-thread-count
-                                    #(async/chan 100 (import/partition-messages tx-size))))
-        file-ch (async/chan 100)
-        pubsub-ch (async/chan 100)
-        pubsub (async/pub pubsub-ch msg-topic)
-        conn  (conn)
-        counter (volatile! 0)
-        done? (promise)]
+  (let [make-tx-chan #(async/chan 100 (import/partition-messages tx-size))
+        tx-chs       (into [] (repeatedly tx-thread-count make-tx-chan))
+        file-ch      (async/chan 100)
+        pubsub-ch    (async/chan 100)
+        pubsub       (async/pub pubsub-ch (wrap-catch msg-topic))
+        conn         (conn)
+        counter      (volatile! 0)
+        done?        (promise)]
 
     (doseq [tx-ch tx-chs]
       (async/thread
@@ -162,14 +175,10 @@
                   (println e)))
               (recur (<!! tx-ch)))))))
 
-    (go-loop [[f & files] files]
-      (>! file-ch f)
-      (if (seq files)
-        (recur files)
-        (async/close! file-ch)))
+    (async/onto-chan file-ch files)
 
-    (doseq [i (range tx-thread-count)]
-      (async/sub pubsub i (get tx-chs i)))
+    (doseq [[i tx-ch] (map-indexed vector tx-chs)]
+      (async/sub pubsub i tx-ch))
 
     (async/pipeline-blocking 10 pubsub-ch file->tx file-ch true)
 
@@ -189,45 +198,70 @@
   (use 'clojurians-log.repl)
   (in-ns 'clojurians-log.repl)
   (load-slack-data!)
+
   (def result (load-files! (log-files)))
-  result
+  ;; or
+  (def result (load-files! (files-from "2019-01-01")))
 
-  (load-files! [f])
+  ;; see progress
+  (future
+    (while (not (realized? (second result)))
+      (println (java.util.Date.) "\t" @(first result))
+      (Thread/sleep 5000)))
 
-  (def result (load-files! (drop 1508 (log-files))))
-  (def rrr (load-files! (filter #(and (.contains (str %) "2020-08") (.contains (str %) "backfill")) (log-files))))
+  ;; After importing, or you won't see data show up
+  (build-indexes!)
 
-  (while (not (realized? (second result)))
-    (println (java.util.Date.) "\t" @(first result))
-    (Thread/sleep 5000))
+  (def result
+    (load-files!
+     (filter #(and (.contains (str %) "2020-08")
+                   (.contains (str %) "backfill"))
+             (log-files))))
 
-  ;; old way (slower)
-  (run! load-log-file! (log-files))
 
-  ;; incremental
-  (load-from "2019-08-23")
 
+  ;; old way, this does not use multi-thread core.async magic, and does not
+  ;; batch transactions, it literally transacts each slack event separately.
+  ;; Very slow but always works.
+  (run! load-log-file! (log-files))
 
+  ;; Fetch and store slack data
   (do
     (write-edn "users.edn" (map import/user->tx (slack/users)))
-    (write-edn "channels.edn" (map import/channel->tx (slack/channels))))
-
-  (time
-   (do
-     (time (clojurians-log.db.queries/channel-day-messages db "clojurescript" "2018-02-04"))
-     (time (clojurians-log.db.queries/thread-messages db '("1517722327.000023" "1517722363.000043" "1517722613.000012" "1517724278.000043" "1517724340.000044" "1517724770.000024" "1517724836.000023" "1517725105.000054")))
-     (time (ffirst (clojurians-log.db.queries/channel db "clojurescript")))
-     (time (clojurians-log.db.queries/channel-list db "2018-02-04"))
-     (time (clojurians-log.db.queries/user-names db #{"U2TUBBPNU"}))
-     (time (clojurians-log.db.queries/channel-days db "clojurescript"))
-
-     nil))
-
+    (write-edn "channels.edn" (map import/channel->tx (slack/channels)))
+    (write-edn "emoji.edn" (map import/emoji->tx (slack/emoji))))
+
+  ;; Micro-benchmark some queries, good to check if anything is unreasonably
+  ;; slow
+  (let [db (db)]
+    (time
+     (do
+       (time (clojurians-log.db.queries/channel-day-messages db "clojurescript" "2018-02-04"))
+       (time (clojurians-log.db.queries/thread-messages db '("1517722327.000023" "1517722363.000043" "1517722613.000012" "1517724278.000043" "1517724340.000044" "1517724770.000024" "1517724836.000023" "1517725105.000054")))
+       (time (ffirst (clojurians-log.db.queries/channel db "clojurescript")))
+       (time (clojurians-log.db.queries/channel-list db "2018-02-04"))
+       (time (clojurians-log.db.queries/user-names db #{"U2TUBBPNU"}))
+       (time (clojurians-log.db.queries/channel-days db "clojurescript"))
+
+       nil)))
+
+  ;; Original
   "Elapsed time: 18.166254 msecs"
-  "Elapsed time: 631.458841 msecs"
+  "Elapsed time: 631.458841 msecs" ; -> we optimized this
   "Elapsed time: 1.568807 msecs"
   "Elapsed time: 16.425878 msecs"
   "Elapsed time: 1.126005 msecs"
-  "Elapsed time: 1535.355001 msecs"
-  "Elapsed time: 2205.20762 msecs"
+  "Elapsed time: 1535.355001 msecs" ; -> and this
+  "Elapsed time: 2205.20762 msecs" ; Total
+
+  ;; Latest
+  "Elapsed time: 31.38712 msecs" ; -> seems fetching channel-day-messages is
+                                 ; slower now
+  "Elapsed time: 0.844338 msecs"
+  "Elapsed time: 1.582986 msecs"
+  "Elapsed time: 0.22545 msecs"
+  "Elapsed time: 1.120628 msecs"
+  "Elapsed time: 0.00676 msecs"
+  "Elapsed time: 37.954533 msecs" ; Total
+
   )

src/clojurians_log/views.clj

@@ -187,7 +187,9 @@
     ;; :image_512 :email :real_name_normalized :image_48 :image_192 :real_name :image_72 :image_24
     ;; :avatar_hash :title :team :image_32 :display_name :display_name_normalized
     [:div.message
-     {:id (cl.tu/format-inst-id inst) :class (when (thread-child? message) "thread-msg")}
+     {:id (cl.tu/format-inst-id inst)
+      :class (when (thread-child? message) "thread-msg")
+      :data-message-key (:message/key message)}
      [:a.message_profile-pic {:href (str "/_/_/users/" slack-id) :style (str "background-image: url(" image-48 ");")}]
      [:a.message_username {:href (str "/_/_/users/" slack-id)}
       (some #(when-not (str/blank? %) %) [display-name real-name name])]
@@ -201,12 +203,15 @@
      [:span.message_star]
      [:span.message_content [:p (slack-messages/message->hiccup text usernames emojis)]]
      [:div.message-reaction-bar
-      (let [reaction-group (group-by #(get-in % [:reaction/emoji :emoji/shortcode]) (:reaction/_message message))]
-        (for [[emoji-shortcode reactions] reaction-group]
-          [:div.message-reaction
+      (let [reaction-groups (group-by #(get-in % [:reaction/emoji :emoji/shortcode]) (:reaction/_message message))]
+        (for [[emoji-shortcode reactions] reaction-groups
+              :let [{:strs [reaction_added reaction_removed]} (group-by :reaction/type reactions)
+                    reaction-count (- (count reaction_added) (count reaction_removed))]
+              :when (< 0 reaction-count)]
+          [:div.message-reaction {:title emoji-shortcode}
            [:span.emoji (slack-messages/text->emoji emoji-shortcode emojis)]
            " "
-           (count reactions)]))]]))
+           reaction-count]))]]))
 
 (defn- message-hiccup
   "Returns either a single message hiccup, or if the given message starts a thread,