🐛 Keep a file reload from doubling the workspace streams

Reloading the open file (a peer's restore, the fallback of a rollback)
runs initialize-workspace again without leaving the workspace. The
streams the first run started kept running next to the new ones, so a
commit added one undo entry per reload and each notification was
handled twice. The old subscription also sent its unsubscribe after the
new subscribe, which ended the new one on the server.

A new initialize-workspace, text reflow, shape layout or notification
subscription stops the one before it, and a subscription replaced by
one to the same file sends no unsubscribe.

frontend/src/app/main/data/workspace.cljs

@@ -416,7 +416,10 @@
 
     ptk/WatchEvent
     (watch [_ state stream]
-      (let [stoper-s     (rx/filter (ptk/type? ::finalize-workspace) stream)
+      (let [;; A reload starts a new `initialize-workspace`, which replaces these streams
+            stoper-s     (rx/filter #(or (ptk/type? ::finalize-workspace %)
+                                         (ptk/type? ::initialize-workspace %))
+                                    stream)
             rparams      (rt/get-params state)
             features     (features/get-enabled-features state team-id)
             render-wasm-enabled? (features/active-feature? state "render-wasm/v1")

frontend/src/app/main/data/workspace/notifications.cljs

@@ -48,12 +48,29 @@
 (declare handle-pointer-send)
 (declare handle-export-update)
 
+(defn- resubscribed?
+  "True when `event` is a new subscription to `file-id`, which a reload of
+  the open file starts."
+  [file-id event]
+  (and (ptk/type? ::initialize event)
+       (= file-id (:file-id (deref event)))))
+
 (defn initialize
+  "Subscribes to the notifications of `file-id` and its team until the file
+  is left or a new subscription starts. A new subscription to the same file
+  replaces this one, so a reload never doubles the handled messages."
   [team-id file-id]
   (ptk/reify ::initialize
+    IDeref
+    (-deref [_] {:file-id file-id})
+
     ptk/WatchEvent
     (watch [_ state stream]
-      (let [stopper     (rx/filter (ptk/type? ::finalize) stream)
+      (let [stopped-by  (volatile! nil)
+            stopper     (->> stream
+                             (rx/filter #(or (ptk/type? ::finalize %)
+                                             (ptk/type? ::initialize %)))
+                             (rx/tap #(vreset! stopped-by %)))
             profile-id (:profile-id state)
             organization-id (dm/get-in state [:teams team-id :organization :id])
 
@@ -107,7 +124,13 @@
 
                             (rx/take-until stopper))]
 
-        (rx/concat stream (rx/of (dws/send endmsg)))))))
+        (rx/concat
+         stream
+         ;; The server keeps one file subscription per connection: an
+         ;; unsubscribe sent after the new subscribe would end it
+         (rx/defer #(if (resubscribed? file-id @stopped-by)
+                      (rx/empty)
+                      (rx/of (dws/send endmsg)))))))))
 
 (defn- handle-change-team-role
   [{:keys [role] :as msg}]

frontend/src/app/main/data/workspace/shape_layout.cljs

@@ -150,7 +150,10 @@
   (ptk/reify ::initialize-shape-layout
     ptk/WatchEvent
     (watch [_ _ stream]
-      (let [stopper (rx/filter (ptk/type? ::finalize-shape-layout) stream)]
+      (let [;; A workspace reload starts a new one, which replaces this one
+            stopper (rx/filter #(or (ptk/type? ::finalize-shape-layout %)
+                                    (ptk/type? ::initialize-shape-layout %))
+                               stream)]
         (->> stream
              (rx/filter (ptk/type? :layout/update))
              ;; Keeps each update's page and the shapes that can be laid out;

frontend/src/app/main/data/workspace/texts.cljs

@@ -96,7 +96,10 @@
   (ptk/reify ::initialize-text-reflow
     ptk/WatchEvent
     (watch [_ _ stream]
-      (let [stopper      (rx/filter (ptk/type? ::finalize-text-reflow) stream)
+      (let [;; A workspace reload starts a new one, which replaces this one
+            stopper      (rx/filter #(or (ptk/type? ::finalize-text-reflow %)
+                                         (ptk/type? ::initialize-text-reflow %))
+                                    stream)
             page-stopper (rx/filter page-finalize? stream)]
         (->> stream
              (rx/filter (ptk/type? :text/reflow))

frontend/test/frontend_tests/data/workspace_notifications_test.cljs

@@ -12,12 +12,14 @@
    [app.main.data.changes :as dch]
    [app.main.data.modal :as modal]
    [app.main.data.notifications :as ntf]
+   [app.main.data.websocket :as dws]
    [app.main.data.workspace :as-alias dw]
    [app.main.data.workspace.notifications :as dwntf]
    [app.util.i18n :as i18n]
    [beicon.v2.core :as rx]
    [cljs.test :as t :include-macros true]
    [frontend-tests.helpers.async :as async]
+   [frontend-tests.helpers.mock :as mock]
    [potok.v2.core :as ptk]))
 
 (defn- file-change-msg
@@ -143,3 +145,95 @@
                                                :session-id (uuid/next)
                                                :vern 2}))]
     (t/is (empty? events))))
+
+(defn- ^:async subscription-messages
+  "Runs `scenario` with a store over `team-id`, `dws/send` and the pointer
+  handler stubbed. `scenario` gets the store and a function that delivers
+  a peer's pointer message for a file. Resolves to the messages sent over
+  the websocket and the number of handled pointer messages."
+  [team-id scenario]
+  (let [sent    (atom [])
+        handled (atom 0)
+        result  (atom nil)
+        store   (ptk/store {:state {:profile-id (uuid/next)
+                                    :session-id (uuid/next)
+                                    :teams {team-id {:id team-id}}}
+                            :on-error #(t/is false (str %))})
+        pointer (fn [file-id]
+                  (ptk/emit! store (ptk/data-event ::dws/message
+                                                   {:type :pointer-update
+                                                    :topic file-id
+                                                    :file-id file-id
+                                                    :session-id (uuid/next)})))]
+    (await
+     (mock/with-mocks*
+       {dws/send                    (mock/stub (fn [msg]
+                                                 (swap! sent conj msg)
+                                                 (ptk/data-event ::sent msg)))
+        dwntf/handle-pointer-update (mock/stub (fn [_]
+                                                 (swap! handled inc)
+                                                 (ptk/data-event ::handled {})))}
+       (try
+         (await (scenario store pointer))
+         ;; Taken before the store ends, which ends the subscriptions too
+         (reset! result {:sent (mapv #(select-keys % [:type :file-id]) @sent)
+                         :handled @handled})
+         (finally
+           (rx/dispose! store)))))
+    @result))
+
+;; Scenario: the open file reloads three times (a peer's restore, the
+;; restorer's own version reload, the rollback fallback), each starting a
+;; new subscription, and then a peer's message arrives. Proves: one live
+;; subscription per open file, so each message is handled once, and no
+;; unsubscribe ends the server's subscription of the reloaded file.
+(t/deftest ^:async a-reload-replaces-the-file-subscription
+  (let [team-id (uuid/next)
+        file-id (uuid/next)
+        result  (await
+                 (subscription-messages
+                  team-id
+                  (^:async fn [store pointer]
+                    (ptk/emit! store
+                               (dwntf/initialize team-id file-id)
+                               (dwntf/initialize team-id file-id)
+                               (dwntf/initialize team-id file-id))
+                    (await (async/settle))
+                    (pointer file-id)
+                    (await (async/settle)))))]
+    (t/is (= 1 (:handled result)) "the message is handled once")
+    (t/is (= 3 (count (filter #(= :subscribe-file (:type %)) (:sent result)))))
+    (t/is (not-any? #(= :unsubscribe-file (:type %)) (:sent result)))))
+
+;; Scenario: the user leaves a file, and in another case opens another file
+;; without leaving the first one. Proves: the subscription ends with an
+;; unsubscribe of the file it was for, and the old file's messages are no
+;; longer handled.
+(t/deftest ^:async leaving-or-switching-files-ends-the-file-subscription
+  (let [team-id (uuid/next)
+        file-a  (uuid/next)
+        file-b  (uuid/next)
+        left    (await
+                 (subscription-messages
+                  team-id
+                  (^:async fn [store _]
+                    (ptk/emit! store
+                               (dwntf/initialize team-id file-a)
+                               (dwntf/finalize file-a))
+                    (await (async/settle)))))
+        switch  (await
+                 (subscription-messages
+                  team-id
+                  (^:async fn [store pointer]
+                    (ptk/emit! store
+                               (dwntf/initialize team-id file-a)
+                               (dwntf/initialize team-id file-b))
+                    (await (async/settle))
+                    (pointer file-a)
+                    (pointer file-b)
+                    (await (async/settle)))))]
+    (t/is (= [{:type :unsubscribe-file :file-id file-a}]
+             (filterv #(= :unsubscribe-file (:type %)) (:sent left))))
+    (t/is (= [{:type :unsubscribe-file :file-id file-a}]
+             (filterv #(= :unsubscribe-file (:type %)) (:sent switch))))
+    (t/is (= 1 (:handled switch)) "only the new file's message is handled")))

frontend/test/frontend_tests/data/workspace_undo_test.cljs

@@ -12,9 +12,14 @@
    [app.common.test-helpers.ids-map :as cthi]
    [app.common.uuid :as uuid]
    [app.main.data.changes :as dch]
+   [app.main.data.comments :as dcmt]
+   [app.main.data.fonts :as df]
+   [app.main.data.workspace :as dw]
    [app.main.data.workspace.undo :as dwu]
    [beicon.v2.core :as rx]
    [cljs.test :as t :include-macros true]
+   [frontend-tests.helpers.async :as async]
+   [frontend-tests.helpers.mock :as mock]
    [frontend-tests.helpers.state :as ths]
    [potok.v2.core :as ptk]))
 
@@ -236,3 +241,34 @@
     (t/is (= 2 (:index (forget undo-state #{(uuid/next)} 2))))
     (t/is (= 0 (:index (forget undo-state #{g2} 2)))
           "a cut history cannot point past its end")))
+
+;; Scenario: the workspace opens and then reloads twice (a peer's restore,
+;; the rollback fallback), each reload running `initialize-workspace` again
+;; without leaving the workspace; then one commit is made. Proves: each
+;; reload replaces the workspace streams, so the commit adds one undo entry,
+;; not one per reload.
+(t/deftest ^:async a-workspace-reload-adds-one-undo-entry-per-commit
+  (let [file    (cthf/sample-file :file1 :page-label :page1)
+        team-id (uuid/next)
+        store   (ths/setup-store file {:renderer :svg})
+        loaded  (fn [type] (mock/stub (fn [& _] (ptk/data-event type {}))))]
+    (await
+     (mock/with-mocks*
+       {dw/initialize-file             (loaded ::file-requested)
+        dcmt/retrieve-comment-threads  (loaded ::threads-requested)
+        dcmt/fetch-profiles            (loaded ::profiles-requested)
+        df/fetch-fonts                 (loaded ::fonts-requested)}
+       (try
+         (ptk/emit! store
+                    #(assoc % :current-team-id team-id)
+                    (dw/initialize-workspace team-id (:id file))
+                    (ptk/event ::dw/reload-current-file)
+                    (ptk/event ::dw/reload-current-file))
+         (await (async/settle))
+         (ptk/emit! store (rename-page (cthf/current-page-id file) "Renamed"))
+         (await (async/wait-for #(seq (get-in @store [:workspace-undo :items]))
+                                "the commit reaches the undo history"))
+         (await (async/settle))
+         (t/is (= 1 (count (get-in @store [:workspace-undo :items]))))
+         (finally
+           (rx/dispose! store)))))))