Move group chats to their own topic

This commit moves group chats to their own topic, based on the randomly
generated chat-id. It falls back on the discovery topic for those peers
who we can't fingerprint the version, for backward compatibility.

src/status_im/data_store/chats.cljs

@@ -51,6 +51,7 @@
   (-> chat
       (update :admins   #(into #{} %))
       (update :contacts #(into #{} %))
+      (update :members-joined #(into #{} %))
       (update :tags #(into #{} %))
       (update :membership-updates  (partial unmarshal-membership-updates chat-id))
       (update :last-clock-value utils.clocks/safe-timestamp)

src/status_im/data_store/realm/schemas/account/chat.cljs

@@ -235,7 +235,6 @@
   (update v10 :properties merge
           {:last-clock-value {:type     :int
                               :optional true}}))
-
 (def v12
   (-> v11
       (update :properties merge
@@ -243,3 +242,7 @@
                {:type     :string
                 :optional true}})
       (update :properties dissoc :last-message-type)))
+
+(def v13
+  (update v12 :properties assoc
+          :members-joined         {:type "string[]"}))

src/status_im/data_store/realm/schemas/account/core.cljs

@@ -330,6 +330,19 @@
           browser/v8
           dapp-permissions/v9])
 
+(def v31 [chat/v13
+          transport/v7
+          contact/v3
+          message/v9
+          mailserver/v11
+          mailserver-topic/v1
+          user-status/v2
+          membership-update/v1
+          installation/v2
+          local-storage/v1
+          browser/v8
+          dapp-permissions/v9])
+
 ;; put schemas ordered by version
 (def schemas [{:schema        v1
                :schemaVersion 1
@@ -420,4 +433,7 @@
                :migration     migrations/v29}
               {:schema        v30
                :schemaVersion 30
-               :migration     migrations/v30}])
+               :migration     migrations/v30}
+              {:schema        v31
+               :schemaVersion 31
+               :migration     (constantly nil)}])

src/status_im/group_chats/core.cljs

@@ -5,6 +5,7 @@
             [clojure.set :as clojure.set]
             [re-frame.core :as re-frame]
             [status-im.i18n :as i18n]
+            [status-im.constants :as constants]
             [status-im.utils.config :as config]
             [status-im.utils.clocks :as utils.clocks]
             [status-im.chat.models.message :as models.message]
@@ -15,6 +16,9 @@
             [status-im.transport.utils :as transport.utils]
             [status-im.transport.message.protocol :as protocol]
             [status-im.transport.message.group-chat :as message.group-chat]
+            [status-im.transport.message.public-chat :as transport.public-chat]
+
+            [status-im.transport.chat.core :as transport.chat]
             [status-im.utils.fx :as fx]
             [status-im.chat.models :as models.chat]
             [status-im.accounts.db :as accounts.db]
@@ -49,6 +53,21 @@
       js/JSON.parse
       (js->clj :keywordize-keys true)))
 
+(defn joined? [public-key {:keys [members-joined]}]
+  (contains? members-joined public-key))
+
+(defn invited? [my-public-key {:keys [contacts]}]
+  (contains? contacts my-public-key))
+
+(defn extract-creator
+  "Takes a chat as an input, returns the creator"
+  [{:keys [membership-updates]}]
+  (->> membership-updates
+       (filter (fn [{:keys [events]}]
+                 (some #(= "chat-created" (:type %)) events)))
+       first
+       :from))
+
 (defn signature-material
   "Transform an update into a signable string"
   [chat-id events]
@@ -103,16 +122,33 @@
   ([cofx payload chat-id]
    (send-membership-update cofx payload chat-id nil))
   ([{:keys [message-id] :as cofx} payload chat-id removed-members]
-   (let [members (clojure.set/union (get-in cofx [:db :chats chat-id :contacts])
+   (let [chat    (get-in cofx [:db :chats chat-id])
+         creator (extract-creator chat)
+         members (clojure.set/union (get-in cofx [:db :chats chat-id :contacts])
                                     removed-members)
          {:keys [web3]} (:db cofx)
-         current-public-key (accounts.db/current-public-key cofx)]
+         current-public-key (accounts.db/current-public-key cofx)
+         ;; If a member has joined is listening to the shared topic and we send there
+         ;; to ourselves we send always on contact-discovery to make sure all devices
+         ;; are informed, in case of dropped messages.
+         ;; We send on the discovery topic to the creator as it's automatically
+         ;; joined or for contact that have not joined yet,
+         ;; for backward compatibility
+         destinations (map (fn [member]
+                             (if (and (joined? member chat)
+                                      (not= creator member)
+                                      (not= current-public-key member))
+                               {:public-key member
+                                :chat chat-id}
+                               {:public-key member
+                                :chat constants/contact-discovery}))
+                           members)]
      (fx/merge
       cofx
       {:shh/send-group-message
        {:web3          web3
         :src           current-public-key
-        :dsts          members
+        :dsts          destinations
         :success-event [:transport/message-sent
                         chat-id
                         message-id
@@ -264,15 +300,6 @@
            (assoc-in [:group-chat-profile/profile :valid-name?] (valid-name? name))
            (assoc-in [:group-chat-profile/profile :name] name))})
 
-(defn extract-creator
-  "Takes a chat as an input, returns the creator"
-  [{:keys [membership-updates]}]
-  (->> membership-updates
-       (filter (fn [{:keys [events]}]
-                 (some #(= "chat-created" (:type %)) events)))
-       first
-       :from))
-
 (fx/defn handle-name-changed
   "Store name in profile scratchpad"
   [cofx new-chat-name]
@@ -438,12 +465,6 @@
      (map #(assoc % :from from) events))
    all-updates))
 
-(defn joined? [my-public-key {:keys [members-joined]}]
-  (contains? members-joined my-public-key))
-
-(defn invited? [my-public-key {:keys [contacts]}]
-  (contains? contacts my-public-key))
-
 (defn get-inviter-pk [my-public-key {:keys [membership-updates] :as chat}]
   (->> membership-updates
        unwrap-events
@@ -453,6 +474,18 @@
                  from)))
        last))
 
+(fx/defn set-up-topic [cofx chat-id previous-chat]
+  (let [my-public-key (accounts.db/current-public-key cofx)
+        new-chat (get-in cofx [:db :chats chat-id])]
+    (cond
+      (and (not (joined? my-public-key previous-chat))
+           (joined? my-public-key new-chat))
+      (transport.public-chat/join-group-chat cofx chat-id)
+
+      (and (joined? my-public-key previous-chat)
+           (not (joined? my-public-key new-chat)))
+      (transport.chat/unsubscribe-from-chat cofx chat-id))))
+
 (fx/defn handle-membership-update
   "Upsert chat and receive message if valid"
   ;; Care needs to be taken here as chat-id is not coming from a whisper filter
@@ -480,6 +513,7 @@
                                             :members-joined     (:members-joined new-group)
                                             :contacts           (:contacts new-group)})
                   (add-system-messages chat-id previous-chat new-group)
+                  (set-up-topic chat-id previous-chat)
                   #(when (and message
                               ;; don't allow anything but group messages
                               (instance? protocol/Message message)

src/status_im/transport/core.cljs

@@ -20,12 +20,20 @@
   [{:keys [db web3] :as cofx}]
   (log/debug :init-whisper)
   (when-let [public-key (get-in db [:account/account :public-key])]
-    (let [topic (transport.utils/get-topic constants/contact-discovery)]
+    (let [public-key-topics (keep (fn [[chat-id {:keys [topic sym-key]}]]
+                                    (when (and (not sym-key)
+                                               topic)
+                                      {:topic topic
+                                       :chat-id chat-id}))
+                                  (:transport/chats db))
+          discovery-topic (transport.utils/get-topic constants/contact-discovery)]
       (fx/merge cofx
-                {:shh/add-discovery-filter
+                {:shh/add-discovery-filters {:web3           web3
-                 {:web3           web3
+                                             :private-key-id public-key
-                  :private-key-id public-key
+                                             :topics (conj public-key-topics
-                  :topic          topic}
+                                                           {:topic discovery-topic
+                                                            :chat-id :discovery-topic})}
+
                  :shh/restore-sym-keys-batch
                  {:web3       web3
                   :transport  (keep (fn [[chat-id {:keys [topic sym-key]

src/status_im/transport/filters.cljs

@@ -73,20 +73,14 @@
      (add-filters! web3 filters))))
 
 (re-frame/reg-fx
- :shh/add-discovery-filter
+ :shh/add-discovery-filters
- (fn [{:keys [web3 private-key-id topic]}]
+ (fn [{:keys [web3 private-key-id topics]}]
-   (let [params   {:topics [topic]
+   (let [params   {:topics (mapv :topic topics)
                    :privateKeyID private-key-id}
          callback (fn [js-error js-message]
                     (re-frame/dispatch [:transport/messages-received js-error js-message]))]
-     (add-filter! web3 params callback :discovery-topic))))
+     (doseq [{:keys [chat-id topic]} topics]
-
+       (add-filter! web3 params callback chat-id)))))
-(defn all-filters-added?
-  [{:keys [db]}]
-  (let [filters (set (keys (get db :transport/filters)))
-        chats (into #{:discovery-topic}
-                    (keys (filter #(:topic (val %)) (get db :transport/chats))))]
-    (= chats filters)))
 
 (handlers/register-handler-fx
  :shh.callback/filter-added

src/status_im/transport/message/public_chat.cljs

@@ -10,6 +10,23 @@
 (defn- has-already-joined? [{:keys [db]} chat-id]
   (get-in db [:transport/chats chat-id]))
 
+(fx/defn join-group-chat
+  "Function producing all protocol level effects necessary for a group chat identified by chat-id"
+  [{:keys [db] :as cofx} chat-id]
+  (when-not (has-already-joined? cofx chat-id)
+    (let [public-key (get-in db [:account/account :public-key])
+          topic (transport.utils/get-topic chat-id)]
+      (fx/merge cofx
+                {:shh/add-discovery-filters {:web3 (:web3 db)
+                                             :private-key-id public-key
+                                             :topics [{:topic topic
+                                                       :chat-id chat-id}]}}
+                (protocol/init-chat {:chat-id chat-id
+                                     :topic   topic})
+                #(hash-map :data-store/tx  [(transport-store/save-transport-tx
+                                             {:chat-id chat-id
+                                              :chat    (get-in % [:db :transport/chats chat-id])})])))))
+
 (fx/defn join-public-chat
   "Function producing all protocol level effects necessary for joining public chat identified by chat-id"
   [{:keys [db] :as cofx} chat-id]

src/status_im/transport/shh.cljs

@@ -1,6 +1,7 @@
 (ns ^{:doc "Whisper API and events for managing keys and posting messages"}
  status-im.transport.shh
   (:require [re-frame.core :as re-frame]
+            [status-im.constants :as constants]
             [status-im.transport.message.transit :as transit]
             [status-im.transport.utils :as transport.utils]
             [taoensso.timbre :as log]))
@@ -73,6 +74,7 @@
             :or   {error-event :transport/send-status-message-error}} post-calls]
      (let [direct-message (clj->js {:pubKey dst
                                     :sig src
+                                    :chat constants/contact-discovery
                                     :payload (-> payload
                                                  transit/serialize
                                                  transport.utils/from-utf8)})]
@@ -88,6 +90,7 @@
    (let [{:keys [web3 payload src dsts success-event error-event]
           :or   {error-event :protocol/send-status-message-error}} params
          message (clj->js {:sig src
+                           :chat constants/contact-discovery
                            :payload (-> payload
                                         transit/serialize
                                         transport.utils/from-utf8)})]
@@ -100,18 +103,22 @@
 (re-frame/reg-fx
  :shh/send-group-message
  (fn [params]
-   (let [{:keys [web3 payload src dsts success-event error-event]
+   (let [{:keys [web3 payload chat src dsts success-event error-event]
-          :or   {error-event :transport/send-status-message-error}} params
+          :or   {error-event :transport/send-status-message-error}} params]
-         message (clj->js {:pubKeys dsts
+     (doseq [{:keys [public-key chat]} dsts]
-                           :sig src
+       (let [message
-                           :payload (-> payload
+             (clj->js {:pubKey public-key
-                                        transit/serialize
+                       :chat chat
-                                        transport.utils/from-utf8)})]
+                       :sig src
-     (.. web3
+                       :payload (-> payload
-         -shh
+                                    transit/serialize
-         (sendGroupMessage
+                                    transport.utils/from-utf8)})]
-          message
+
-          (handle-response success-event error-event))))))
+         (.. web3
+             -shh
+             (sendDirectMessage
+              message
+              (handle-response success-event error-event))))))))
 
 (re-frame/reg-fx
  :shh/send-public-message

test/cljs/status_im/test/data_store/chats.cljs

@@ -9,6 +9,7 @@
             :contacts                  #{2}
             :tags                      #{}
             :membership-updates        []
+            :members-joined #{}
             :last-message-content      {:foo "bar"}
             :last-clock-value          nil}
            (chats/normalize-chat

test/cljs/status_im/test/group_chats/core.cljs

@@ -129,6 +129,36 @@
                       "group-chat-name-changed"]
                      (map (comp :text :content) (sort-by :clock-value (vals (:messages actual-chat)))))))))))))
 
+(deftest set-up-topic
+  (with-redefs [config/group-chats-enabled? true]
+    (let [cofx {:now 0 :db {:account/account {:public-key admin}}}]
+      (testing "a brand new chat"
+        (let [actual (group-chats/handle-membership-update cofx initial-message "payload" admin)]
+          (testing "it sets up a topic"
+            (is (:shh/add-discovery-filters actual)))))
+      (testing "an existing chat"
+        (let [cofx  (assoc cofx
+                           :db
+                           (:db (group-chats/handle-membership-update cofx initial-message "payload" admin)))
+              new-message {:chat-id chat-id
+                           :membership-updates [{:from member-1
+                                                 :events [{:type "chat-created"
+                                                           :clock-value 1
+                                                           :name "group-name"}
+                                                          {:type "admins-added"
+                                                           :clock-value 10
+                                                           :members [member-2]}
+                                                          {:type "admin-removed"
+                                                           :clock-value 11
+                                                           :member member-1}]}
+                                                {:from member-1
+                                                 :events [{:type "member-removed"
+                                                           :clock-value 12
+                                                           :member member-1}]}]}
+              actual (group-chats/handle-membership-update cofx new-message "payload" admin)]
+          (testing "it removes the topic"
+            (is (:shh/remove-filter actual))))))))
+
 (deftest build-group-test
   (testing "only adds"
     (let [events [{:type    "chat-created"

test/cljs/status_im/test/transport/core.cljs

@@ -10,11 +10,16 @@
                                           :sym-key "sk1"}
                                      "2" {}
                                      "3" {:topic   "topic-3"
-                                          :sym-key "sk3"}}
+                                          :sym-key "sk3"}
+                                     "4" {:topic "topic-4"}}
                    :semaphores      #{}}}]
-    (testing "it adds the discover filter"
+    (testing "it adds the discover filters"
-      (is (= {:web3 nil :private-key-id "1" :topic "0xf8946aac"}
+      (is (= {:web3 nil :private-key-id "1" :topics [{:chat-id :discovery-topic
-             (:shh/add-discovery-filter (transport/init-whisper cofx)))))
+                                                      :topic "0xf8946aac"}
+                                                     {:chat-id "4"
+                                                      :topic "topic-4"}]}
+             (:shh/add-discovery-filters (transport/init-whisper cofx)))))
+
     (testing "it restores the sym-keys"
       (is (= [{:topic "topic-1", :sym-key "sk1", :chat-id "1"}
               {:topic "topic-3", :sym-key "sk3", :chat-id "3"}]