🎉 Add more reliable presence mechanism.

backend/resources/migrations/0004.tasks.sql

@@ -24,8 +24,6 @@ CREATE TRIGGER task__modified_at__tgr
 BEFORE UPDATE ON task
    FOR EACH ROW EXECUTE PROCEDURE update_modified_at();
 
-
-
 CREATE TABLE scheduled_task (
   id text PRIMARY KEY,
 

backend/resources/migrations/0006.presence.sql

@@ -0,0 +1,9 @@
+CREATE TABLE presence (
+  file_id uuid NOT NULL REFERENCES file(id) ON DELETE CASCADE,
+  profile_id uuid NOT NULL REFERENCES profile(id) ON DELETE CASCADE,
+  session_id uuid NOT NULL,
+
+  updated_at timestamptz NOT NULL DEFAULT clock_timestamp(),
+
+  PRIMARY KEY (file_id, session_id, profile_id)
+);

backend/src/uxbox/migrations.clj

@@ -29,7 +29,10 @@
      :fn (mg/resource "migrations/0004.tasks.sql")}
     {:desc "Initial libraries tables"
      :name "0005-libraries"
-     :fn (mg/resource "migrations/0005.libraries.sql")}]})
+     :fn (mg/resource "migrations/0005.libraries.sql")}
+    {:desc "Initial presence tables"
+     :name "0006-presence"
+     :fn (mg/resource "migrations/0006.presence.sql")}]})
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Entry point

backend/src/uxbox/services/notifications.clj

@@ -14,7 +14,8 @@
    [uxbox.common.exceptions :as ex]
    [uxbox.common.uuid :as uuid]
    [uxbox.redis :as redis]
-   [ring.util.codec :as codec]
+   [uxbox.db :as db]
+   [uxbox.util.time :as dt]
    [uxbox.util.transit :as t]))
 
 (defmacro go-try
@@ -31,47 +32,55 @@
        (throw r#)
        r#)))
 
-(defn- decode-message
-  [message]
-  (->> (t/str->bytes message)
-       (t/decode)))
-
-(defn- encode-message
-  [message]
-  (->> (t/encode message)
-       (t/bytes->str)))
+(defmacro thread-try
+  [& body]
+  `(a/thread
+     (try
+       ~@body
+       (catch Throwable e#
+         e#))))
 
 ;; --- Redis Interactions
 
 (defn- publish
   [channel message]
   (go-try
-   (let [message (encode-message message)]
+   (let [message (t/encode-str message)]
      (<? (redis/run :publish {:channel (str channel)
                               :message message})))))
 
+(def ^:private
+  sql:retrieve-presence
+  "select * from presence
+    where file_id=?
+      and (clock_timestamp() - updated_at) < '5 min'::interval")
+
 (defn- retrieve-presence
-  [key]
-  (go-try
-   (let [data (<? (redis/run :hgetall {:key key}))]
-     (into [] (map (fn [[k v]] [(uuid/uuid k) (uuid/uuid v)])) data))))
+  [file-id]
+  (thread-try
+   (let [rows (db/exec! db/pool [sql:retrieve-presence file-id])]
+     (mapv (juxt :session-id :profile-id) rows))))
 
-(defn- join-room
-  [file-id session-id profile-id]
-  (let [key (str file-id)
-        field (str session-id)
-        value (str profile-id)]
-    (go-try
-     (<? (redis/run :hset {:key key :field field :value value}))
-     (<? (retrieve-presence key)))))
+(def ^:private
+  sql:update-presence
+  "insert into presence (file_id, session_id, profile_id, updated_at)
+   values (?, ?, ?, clock_timestamp())
+       on conflict (file_id, session_id, profile_id)
+       do update set updated_at=clock_timestamp()")
 
-(defn- leave-room
+(defn- update-presence
   [file-id session-id profile-id]
-  (let [key (str file-id)
-        field (str session-id)]
-    (go-try
-     (<? (redis/run :hdel {:key key :field field}))
-     (<? (retrieve-presence key)))))
+  (thread-try
+   (let [now (dt/now)
+         sql [sql:update-presence file-id session-id profile-id]]
+     (db/exec-one! db/pool sql))))
+
+(defn- delete-presence
+  [file-id session-id profile-id]
+  (thread-try
+   (db/delete! db/pool :presence {:file-id file-id
+                                  :profile-id profile-id
+                                  :session-id session-id})))
 
 ;; --- WebSocket Messages Handling
 
@@ -85,29 +94,34 @@
   [{:keys [file-id profile-id session-id output] :as ws} message]
   (log/info (str "profile " profile-id " is connected to " file-id))
   (go-try
-   (let [members (<? (join-room file-id session-id profile-id))]
-     (<? (publish file-id {:type :presence :sessions  members})))))
+   (<? (update-presence file-id session-id profile-id))
+   (let [members (<? (retrieve-presence file-id))]
+     (<? (publish file-id {:type :presence :sessions members})))))
 
 (defmethod handle-message :disconnect
   [{:keys [profile-id file-id session-id] :as ws} message]
   (log/info (str "profile " profile-id " is disconnected from " file-id))
   (go-try
-   (let [members (<? (leave-room file-id session-id profile-id))]
+   (<? (delete-presence file-id session-id profile-id))
+   (let [members (<? (retrieve-presence file-id))]
      (<? (publish file-id {:type :presence :sessions members})))))
 
+(defmethod handle-message :keepalive
+  [{:keys [profile-id file-id session-id] :as ws} message]
+  (update-presence file-id session-id profile-id))
+
+(defmethod handle-message :pointer-update
+  [{:keys [profile-id file-id session-id] :as ws} message]
+  (let [message (assoc message
+                       :profile-id profile-id
+                       :session-id session-id)]
+    (publish file-id message)))
+
 (defmethod handle-message :default
   [ws message]
   (a/go
     (log/warn (str "received unexpected message: " message))))
 
-(defmethod handle-message :pointer-update
-  [{:keys [profile-id file-id session-id] :as ws} message]
-  (go-try
-   (let [message (assoc message
-                        :profile-id profile-id
-                        :session-id session-id)]
-     (<? (publish file-id message)))))
-
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; WebSocket Handler
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
@@ -176,7 +190,7 @@
   (a/go-loop []
     (let [val (a/<! out)]
       (when-not (nil? val)
-        (jetty/send! conn (encode-message val))
+        (jetty/send! conn (t/encode-str val))
         (recur)))))
 
 (defn websocket
@@ -184,7 +198,7 @@
   (let [in  (a/chan 32)
         out (a/chan 32)]
     {:on-connect (fn [conn]
-                   (let [xf  (map decode-message)
+                   (let [xf  (map t/decode-str)
                          sub (redis/subscribe (str file-id) xf)
                          ws  (WebSocket. conn in out sub nil params)]
                      (start-rcv-loop! ws)
@@ -203,7 +217,7 @@
                  (a/close! in))
 
      :on-text (fn [ws message]
-                (let [message (decode-message message)]
+                (let [message (t/decode-str message)]
                   ;; (prn "websocket" :on-text message)
                   (a/>!! in message)))
 

backend/src/uxbox/util/migrations.clj

@@ -72,7 +72,7 @@
 (defn migrate!
   "Main entry point for apply a migration."
   ([conn migrations]
-   (migrate! conn migrations nil))
+   (impl-migrate conn migrations nil))
   ([conn migrations options]
    (impl-migrate conn migrations options)))
 

backend/src/uxbox/util/time.clj

@@ -54,9 +54,16 @@
     (integer? ms-or-obj)
     (Duration/ofMillis ms-or-obj)
 
+    (string? ms-or-obj)
+    (Duration/parse ms-or-obj)
+
     :else
     (obj->duration ms-or-obj)))
 
+(defn duration-between
+  [t1 t2]
+  (Duration/between t1 t2))
+
 (defn parse-duration
   [s]
   (assert (string? s))

backend/src/uxbox/util/transit.clj

@@ -52,6 +52,9 @@
 
 ;; --- High-Level Api
 
+(declare str->bytes)
+(declare bytes->str)
+
 (defn decode
   ([data]
    (decode data nil))
@@ -68,6 +71,16 @@
        (write! w data)
        (.toByteArray out)))))
 
+(defn decode-str
+  [message]
+  (->> (str->bytes message)
+       (decode)))
+
+(defn encode-str
+  [message]
+  (->> (encode message)
+       (bytes->str)))
+
 ;; --- Helpers
 
 (defn str->bytes
@@ -83,4 +96,3 @@
    (bytes->str data "UTF-8"))
   ([^bytes data, ^String encoding]
    (String. data encoding)))
-

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

@@ -28,6 +28,7 @@
 (declare handle-pointer-update)
 (declare handle-page-change)
 (declare handle-pointer-send)
+(declare send-keepalive)
 
 (s/def ::type keyword?)
 (s/def ::message
@@ -46,8 +47,11 @@
     ptk/WatchEvent
     (watch [_ state stream]
       (let [wsession (get-in state [:ws file-id])
-            stoper (rx/filter #(= ::finalize %) stream)]
+            stoper   (rx/filter #(= ::finalize %) stream)
+            interval (* 1000 60)]
         (->> (rx/merge
+              (->> (rx/timer interval interval)
+                   (rx/map #(send-keepalive file-id)))
               (->> (ws/-stream wsession)
                    (rx/filter #(= :message (:type %)))
                    (rx/map (comp t/decode :payload))
@@ -66,6 +70,15 @@
 
              (rx/take-until stoper))))))
 
+(defn send-keepalive
+  [file-id]
+  (ptk/reify ::send-keepalive
+    ptk/EffectEvent
+    (effect [_ state stream]
+      (prn "send-keepalive" file-id)
+      (when-let [ws (get-in state [:ws file-id])]
+        (ws/-send ws (t/encode {:type :keepalive}))))))
+
 ;; --- Finalize Websocket
 
 (defn finalize