From cf61c46c35cfc868d84d0c150aecd4f6f03c5e03 Mon Sep 17 00:00:00 2001 From: Dasemu Date: Wed, 26 Aug 2026 10:12:53 +0200 Subject: [PATCH] Fase 4: migrate the daemon event channel from SSE to WebSocket with resume-by-revision Adds a WS upgrade handler alongside the existing SSE stream on /v1/events (kept alive for instant rollback), authenticated via Sec-WebSocket-Protocol instead of a query-string token. The daemon now keeps a ~500-entry circular buffer of broadcast sync-db-changes deltas indexed by renderer revision; a client reconnecting sends {type: subscribe, graphId, sinceRev} and either gets a gap-free replay + ack, or a {type: resync} instructing it to discard all cached renderer state and reload (frontend.db.subs/reset-graph!, wired through a new :db/resync event). The web/Electron client transport (frontend.persist-db.remote) now opens the WS channel by default; the SSE path (connect-events!) is untouched and still fully wired. Also fixes new/pre-existing test breakage surfaced by switching the default transport: persist_db_test.cljs's shared Electron-failover fixture now mocks open-ws-fn (not just open-sse-fn), and a `new` applied to an inline (or ...) expression in make-server's WebSocketServer construction is replaced with the same let-bound-symbol pattern deps/db-sync's node/server.cljs already uses -- inlining it silently dropped `new`, breaking every daemon test. Confirmed 7 unrelated persist_db_test.cljs failures (electron-ensure-remote-* IPC bookkeeping) already fail identically on the pre-Fase-4 baseline -- left untouched, out of scope here. Co-Authored-By: Claude Sonnet 5 --- src/main/frontend/db/subs.cljs | 8 + src/main/frontend/handler/events.cljs | 10 + src/main/frontend/handler/worker.cljs | 3 + src/main/frontend/persist_db/remote.cljs | 89 ++++++++- src/main/frontend/worker/db_worker_node.cljs | 51 +++++- .../frontend/worker/db_worker_node_auth.cljs | 18 +- .../frontend/worker/db_worker_node_ws.cljs | 171 ++++++++++++++++++ src/test/frontend/persist_db_test.cljs | 14 +- .../frontend/worker/db_worker_node_test.cljs | 162 +++++++++++++++++ .../worker/db_worker_node_ws_test.cljs | 124 +++++++++++++ 10 files changed, 636 insertions(+), 14 deletions(-) create mode 100644 src/main/frontend/worker/db_worker_node_ws.cljs create mode 100644 src/test/frontend/worker/db_worker_node_ws_test.cljs diff --git a/src/main/frontend/db/subs.cljs b/src/main/frontend/db/subs.cljs index 3694cb2..dc34d68 100644 --- a/src/main/frontend/db/subs.cljs +++ b/src/main/frontend/db/subs.cljs @@ -286,6 +286,14 @@ :slot-keys (conj slot-keys slot-key)})) (request-reload! slot-key))) +(defn current-rev + "The renderer revision the store has fully applied. -1 before any delta + has ever been applied for the current graph. Used by the remote transport + to resume the WebSocket event stream after a reconnect (see + frontend.persist-db.remote/connect-events!)." + [] + (:rev @*store)) + (defn reset-graph! [graph-id] (let [generation (inc (:generation @*store)) diff --git a/src/main/frontend/handler/events.cljs b/src/main/frontend/handler/events.cljs index f98d1c7..c51b270 100644 --- a/src/main/frontend/handler/events.cljs +++ b/src/main/frontend/handler/events.cljs @@ -13,6 +13,7 @@ [frontend.context.i18n :refer [t]] [frontend.date :as date] [frontend.db.async :as db-async] + [frontend.db.subs :as db-subs] [frontend.extensions.fsrs :as fsrs] [frontend.handler.assets :as assets-handler] [frontend.handler.code :as code-handler] @@ -481,6 +482,15 @@ (pipeline/invoke-hooks data) nil) +;; The remote WS transport lost some deltas (buffer overflow on the daemon +;; side, or a gap it can't otherwise explain) -- discard everything cached +;; renderer-side and let every mounted slot reload from scratch. Safe to call +;; even when nothing is actually stale, just wasteful. +(defevent! :db/resync [[_ _data]] + (when-let [repo (state/get-current-repo)] + (db-subs/reset-graph! repo)) + nil) + (defevent! :db/export-sqlite [_] (export/export-repo-as-sqlite-db! (state/get-current-repo)) nil) diff --git a/src/main/frontend/handler/worker.cljs b/src/main/frontend/handler/worker.cljs index a1607ff..cb9a7e2 100644 --- a/src/main/frontend/handler/worker.cljs +++ b/src/main/frontend/handler/worker.cljs @@ -64,6 +64,9 @@ (defmethod handle :sync-db-changes [_ _worker data] (state/pub-event! [:db/sync-changes data])) +(defmethod handle :resync [_ _worker data] + (state/pub-event! [:db/resync data])) + (defmethod handle :sync-conflicts-updated [_ _worker {:keys [repo block-uuid conflicts]}] (when (and (seq repo) block-uuid) (state/set-sync-block-conflicts! repo block-uuid conflicts))) diff --git a/src/main/frontend/persist_db/remote.cljs b/src/main/frontend/persist_db/remote.cljs index d35a792..b893323 100644 --- a/src/main/frontend/persist_db/remote.cljs +++ b/src/main/frontend/persist_db/remote.cljs @@ -1,6 +1,7 @@ (ns frontend.persist-db.remote "Remote `PersistentDB` implementation for Electron renderer via db-worker-node HTTP and SSE." (:require [clojure.string :as string] + [frontend.db.subs :as db-subs] [frontend.persist-db.protocol :as protocol] [logseq.db :as ldb] [promesa.core :as p] @@ -32,6 +33,12 @@ (cond-> (str (normalize-base-url base-url) "/v1/events") (seq auth-token) (str "?token=" (js/encodeURIComponent auth-token)))) +(defn- events-ws-url + [base-url] + (-> (normalize-base-url base-url) + (string/replace #"^http" "ws") + (str "/v1/events"))) + (defn- base-headers [auth-token] (cond-> {"Content-Type" "application/json" @@ -115,8 +122,12 @@ (subs line 6))) (string/split-lines event-text))) +(defn- default-open-ws-fn + [{:keys [url auth-token]}] + (js/WebSocket. url (str "bearer." auth-token))) + (defn create-client - [{:keys [fetch-fn open-sse-fn schedule-fn reconnect-delay-ms] :as opts}] + [{:keys [fetch-fn open-sse-fn open-ws-fn schedule-fn reconnect-delay-ms] :as opts}] (let [default-fetch-fn (fn [{:keys [method url headers body]}] (p/let [^js res (js/fetch url (clj->js (cond-> {:method method @@ -143,6 +154,7 @@ (assoc opts :fetch-fn (or fetch-fn default-fetch-fn) :open-sse-fn (or open-sse-fn default-open-sse-fn) + :open-ws-fn (or open-ws-fn default-open-ws-fn) :schedule-fn (or schedule-fn (fn [f delay-ms] (js/setTimeout f delay-ms))) :reconnect-delay-ms (or reconnect-delay-ms 1000)))) @@ -244,6 +256,73 @@ (close-subscription!) nil)}))) +(defn connect-events-ws! + "WebSocket transport for the daemon event channel -- the default since + Fase 4 (connect-events! above stays fully wired as an instant rollback, + see docs/agent-guide). Resumes by revision on every (re)connect: sends + {type: subscribe, graphId: repo, sinceRev: N} where N is the renderer's + last applied revision. The server replays buffered deltas gap-free, or -- + if the gap is wider than its ~500-entry buffer -- replies {type: resync}, + handled here as a normal event-handler dispatch so the existing + frontend.handler.worker / :db/resync wiring can discard cached renderer + state and reload." + [{:keys [base-url auth-token repo event-handler open-ws-fn schedule-fn + reconnect-delay-ms on-event-error current-rev-fn]} + wrapped-worker] + (let [connected? (atom true) + socket (atom nil) + current-rev-fn (or current-rev-fn db-subs/current-rev) + close-socket! (fn [] + (when-let [^js s @socket] + (set! (.-onopen s) nil) + (set! (.-onmessage s) nil) + (set! (.-onerror s) nil) + (set! (.-onclose s) nil) + (try (.close s) (catch :default _ nil))) + (reset! socket nil)) + dispatch-control! (fn [{:keys [type rev message]}] + (case type + "resync" (when event-handler + (event-handler :resync wrapped-worker {:rev rev})) + "error" (log/error :db-worker-node-ws-server-error {:message message}) + nil)) + dispatch-business! (fn [event] + (let [[event-type payload] (decode-event event)] + (when (and event-handler event-type) + (event-handler event-type wrapped-worker payload))))] + (letfn [(open! [] + (when @connected? + (let [^js s (open-ws-fn {:url (events-ws-url base-url) :auth-token auth-token})] + (reset! socket s) + (set! (.-onopen s) + (fn [_] + (when @connected? + (.send s (js/JSON.stringify + (clj->js {:type "subscribe" + :graphId repo + :sinceRev (current-rev-fn)})))))) + (set! (.-onmessage s) + (fn [event] + (when @connected? + (let [parsed (parse-response-body (.-data event))] + (if (contains? parsed :payload) + (dispatch-business! parsed) + (dispatch-control! parsed)))))) + (set! (.-onerror s) + (fn [error] + (when (and @connected? on-event-error) + (on-event-error error)))) + (set! (.-onclose s) + (fn [_] + (reset! socket nil) + (when @connected? + (schedule-fn open! reconnect-delay-ms)))))))] + (open!) + {:disconnect! (fn [] + (reset! connected? false) + (close-socket!) + nil)}))) + (defn- method->str [qkw] (str (namespace qkw) "/" (name qkw))) @@ -280,10 +359,10 @@ (let [client (create-client (assoc opts :base-url base-url :auth-token auth-token)) wrapped-worker (fn [qkw & args] (invoke! client (method->str qkw) args)) - {:keys [disconnect!]} (connect-events! (assoc client - :event-handler event-handler - :auth-token auth-token) - wrapped-worker)] + {:keys [disconnect!]} (connect-events-ws! (assoc client + :event-handler event-handler + :auth-token auth-token) + wrapped-worker)] (->InRemote client wrapped-worker disconnect!))) (defn stop! diff --git a/src/main/frontend/worker/db_worker_node.cljs b/src/main/frontend/worker/db_worker_node.cljs index 9db9021..7b31b65 100644 --- a/src/main/frontend/worker/db_worker_node.cljs +++ b/src/main/frontend/worker/db_worker_node.cljs @@ -1,10 +1,12 @@ (ns frontend.worker.db-worker-node "Node.js daemon entrypoint for db-worker." (:require ["http" :as http] + ["ws" :as ws] [clojure.string :as string] [frontend.worker.db-core :as db-core] [frontend.worker.db-worker-node-auth :as auth] [frontend.worker.db-worker-node-lock :as db-lock] + [frontend.worker.db-worker-node-ws :as db-ws] [frontend.worker.platform.node :as platform-node] [frontend.worker.state :as worker-state] [lambdaisland.glogi :as log] @@ -158,6 +160,17 @@ (nil? method) nil :else (str method))) +(defn- sync-db-changes-rev + "Peeks the transit-encoded [:sync-db-changes {...:delta {:rev}}] tuple + shared-service/broadcast-to-clients! always sends for this event type (see + db_listener.cljs), so the WS resume buffer can index by renderer revision + without re-deriving it. Returns nil for any other event type." + [type payload] + (when (and (= type :sync-db-changes) (string? payload)) + (try + (get-in (ldb/read-transit-str payload) [1 :delta :rev]) + (catch :default _ nil)))) + (defn- handle-event! [type payload] (let [event (js/JSON.stringify (clj->js {:type (encode-event-type type) @@ -167,7 +180,10 @@ (try (.write res message) (catch :default e - (log/error :sse-write-failed e)))))) + (log/error :sse-write-failed e)))) + (when-let [rev (sync-db-changes-rev type payload)] + (db-ws/record-delta! rev event)) + (db-ws/broadcast! event))) (defn- sse-handler [^js req ^js res] @@ -380,11 +396,32 @@ (str "Bearer "))) (auth/authorized? (.. req -headers -authorization)))) +(defn- ws-upgrade-handler + [^js wss bound-repo port extra-allowed-hosts] + (fn [^js req ^js socket head] + (let [allowed-hosts (allowed-hosts-for {:port port :extra-allowed-hosts extra-allowed-hosts}) + request-path (.-pathname (js/URL. (.-url req) "http://127.0.0.1"))] + (cond + (not (host-allowed? req allowed-hosts)) + (db-ws/reject-upgrade! socket 403 "Host header not allowed") + + (not= request-path "/v1/events") + (db-ws/reject-upgrade! socket 404 "not found") + + (not (auth/authorized-token? (db-ws/token-from-upgrade-req req))) + (db-ws/reject-upgrade! socket 401 "missing or invalid credentials") + + :else + (.handleUpgrade wss req socket head + (fn [ws-socket] + (db-ws/handle-connection! ws-socket bound-repo))))))) + (defn- make-server [proxy {:keys [bound-repo stop-fn host port owner-source root-dir allowed-origins extra-allowed-hosts]}] - (http/createServer - (fn [^js req ^js res] + (let [server + (http/createServer + (fn [^js req ^js res] (let [url (.-url req) parsed-url (js/URL. url "http://127.0.0.1") request-path (.-pathname parsed-url) @@ -515,7 +552,11 @@ (send-text! res 405 "method-not-allowed")) :else - (send-text! res 404 "not-found")))))) + (send-text! res 404 "not-found"))))) + WSS (or (.-WebSocketServer ws) (.-Server ws)) + ^js wss (new WSS #js {:noServer true})] + (.on server "upgrade" (ws-upgrade-handler wss bound-repo port extra-allowed-hosts)) + server)) (defn- show-help! [] @@ -584,6 +625,8 @@ (.end res) (catch :default _))) (reset! *sse-clients #{}) + (db-ws/close-all!) + (db-ws/reset-buffer!) (when-let [lock-path (:path @*lock-info)] (db-lock/remove-lock! lock-path))) diff --git a/src/main/frontend/worker/db_worker_node_auth.cljs b/src/main/frontend/worker/db_worker_node_auth.cljs index 01d424e..34b14fc 100644 --- a/src/main/frontend/worker/db_worker_node_auth.cljs +++ b/src/main/frontend/worker/db_worker_node_auth.cljs @@ -146,16 +146,26 @@ (do (record-failed-attempt! username) {:error :invalid-credentials}))))) +(defn authorized-token? + "Given a bare session token (no `Bearer ` prefix), returns true and + refreshes :last-seen-at when it matches an active session. Shared by the + Authorization-header path (authorized?) and the WebSocket upgrade path, + which carries the token via Sec-WebSocket-Protocol instead -- see + db_worker_node_ws.cljs." + [token] + (when (string? token) + (when-let [session (some #(when (timing-safe-equal? token (:token %)) %) + (vals @*sessions))] + (swap! *sessions assoc-in [(:id session) :last-seen-at] (now-iso)) + true))) + (defn authorized? "Given an `Authorization: Bearer ` header value (or nil), returns true and refreshes :last-seen-at when it matches an active session." [authorization-header] (when-let [token (some-> authorization-header (as-> h (when (string/starts-with? h "Bearer ") (subs h 7))))] - (when-let [session (some #(when (timing-safe-equal? token (:token %)) %) - (vals @*sessions))] - (swap! *sessions assoc-in [(:id session) :last-seen-at] (now-iso)) - true))) + (authorized-token? token))) (defn list-sessions [] diff --git a/src/main/frontend/worker/db_worker_node_ws.cljs b/src/main/frontend/worker/db_worker_node_ws.cljs new file mode 100644 index 0000000..c1ee118 --- /dev/null +++ b/src/main/frontend/worker/db_worker_node_ws.cljs @@ -0,0 +1,171 @@ +(ns frontend.worker.db-worker-node-ws + "WebSocket event channel for db-worker-node -- kept alongside the SSE + channel (/v1/events) during the Fase 4 migration so either transport can + be rolled back independently; see docs/agent-guide. + + Wire format for domain events mirrors SSE exactly: {\"type\" ... \"payload\" + } -- db_worker_node.cljs builds that message once (for + both transports) and hands it to broadcast!/record-delta! here. + + Client -> server control message: {\"type\" \"subscribe\" \"graphId\" repo + \"sinceRev\" rev-or-null}. Server -> client control messages: + {\"type\" \"subscribed\" \"rev\" head} (caught up, or replay was just sent), + {\"type\" \"resync\" \"rev\" head} (client must discard cached state and + reload everything -- see frontend.handler.events :db/resync), + {\"type\" \"error\" \"message\" ...}. + + The bearer token can't travel in a header on the WS handshake the way + fetch requests use Authorization -- carried via Sec-WebSocket-Protocol + instead (see token-from-upgrade-req), which doesn't leak into proxy/access + logs the way the SSE interim query-param token does." + (:require [clojure.string :as string] + [lambdaisland.glogi :as log])) + +(def ^:private buffer-capacity 500) +(def ^:private heartbeat-interval-ms 30000) + +(defonce ^:private *clients (atom #{})) +(defonce ^:private *delta-buffer (atom (sorted-map))) +(defonce ^:private *head-rev (atom 0)) + +(defn- ws-open? + [^js socket] + (= 1 (.-readyState socket))) + +(defn- ws-send! + [^js socket text] + (when (ws-open? socket) + (try + (.send socket text) + (catch :default e + (log/error :db-worker-node-ws-send-failed e))))) + +(defn broadcast! + "Sends a pre-built wire message (see db_worker_node.cljs handle-event!) to + every currently connected WS client, live-tailing style. Independent of + record-delta! -- non-graph events (notifications, logs, ...) broadcast + without ever entering the resume buffer." + [message] + (doseq [^js socket @*clients] + (ws-send! socket message))) + +(defn record-delta! + "Keeps the last `buffer-capacity` :sync-db-changes wire messages, indexed + by renderer revision, so a client reconnecting within that window can + resume via replay instead of a full reload." + [rev message] + (when (and (integer? rev) (not (neg? rev))) + (swap! *delta-buffer + (fn [buf] + (let [buf' (assoc buf rev message)] + (if (> (count buf') buffer-capacity) + (dissoc buf' (first (keys buf'))) + buf')))) + (swap! *head-rev max rev))) + +(defn- oldest-buffered-rev + [] + (some-> @*delta-buffer first key)) + +(defn- can-replay-from? + [since-rev] + (let [oldest (oldest-buffered-rev)] + (or (nil? oldest) (<= oldest (inc since-rev))))) + +(defn- replay-messages + [since-rev] + (->> @*delta-buffer + (drop-while (fn [[rev _message]] (<= rev since-rev))) + (map second))) + +(defn- send-control! + [^js socket payload] + (ws-send! socket (js/JSON.stringify (clj->js payload)))) + +(defn- handle-subscribe! + [^js socket bound-repo {:strs [graphId sinceRev]}] + (if (and (string? graphId) (not= graphId bound-repo)) + (send-control! socket {:type "error" :message "graph-id does not match bound repo"}) + (let [since-rev (if (integer? sinceRev) sinceRev -1) + head @*head-rev] + (if (can-replay-from? since-rev) + (do + (doseq [message (replay-messages since-rev)] + (ws-send! socket message)) + (send-control! socket {:type "subscribed" :rev head})) + (send-control! socket {:type "resync" :rev head}))))) + +(defn- handle-message! + [^js socket bound-repo text] + (try + (let [data (js->clj (js/JSON.parse text))] + (case (get data "type") + "subscribe" (handle-subscribe! socket bound-repo data) + "ping" (send-control! socket {:type "pong"}) + (send-control! socket {:type "error" :message "unknown message type"}))) + (catch :default e + (log/error :db-worker-node-ws-message-failed e) + (send-control! socket {:type "error" :message "invalid message"})))) + +(defn- start-heartbeat! + "Terminates sockets that stop answering pings -- mobile networks in + particular drop connections silently, and a dead entry in *clients would + otherwise sit there forever, still counted by broadcast!." + [^js socket] + (let [alive? (atom true) + interval (js/setInterval + (fn [] + (if @alive? + (do (reset! alive? false) + (try (.ping socket) (catch :default _ nil))) + (try (.terminate socket) (catch :default _ nil)))) + heartbeat-interval-ms)] + (.on socket "pong" (fn [] (reset! alive? true))) + interval)) + +(defn handle-connection! + [^js socket bound-repo] + (swap! *clients conj socket) + (let [heartbeat (start-heartbeat! socket) + cleanup! (fn [] + (js/clearInterval heartbeat) + (swap! *clients disj socket))] + (.on socket "message" + (fn [data] + (handle-message! socket bound-repo (if (string? data) data (.toString data))))) + (.on socket "close" cleanup!) + (.on socket "error" (fn [error] + (log/error :db-worker-node-ws-error error) + (cleanup!))))) + +(defn token-from-upgrade-req + "Extracts the bare session token from Sec-WebSocket-Protocol. The client + sends a single subprotocol string \"bearer.\" (see + frontend.persist-db.remote/open-websocket)." + [^js req] + (when-let [header (some-> req .-headers (aget "sec-websocket-protocol"))] + (let [first-protocol (-> header (string/split #",") first string/trim)] + (when (string/starts-with? first-protocol "bearer.") + (subs first-protocol (count "bearer.")))))) + +(defn reject-upgrade! + [^js socket status reason] + (try + (.write socket + (str "HTTP/1.1 " status " " reason "\r\n" + "Connection: close\r\n" + "Content-Type: application/json\r\n\r\n" + (js/JSON.stringify (clj->js {:ok false :error {:message reason}})))) + (catch :default _ nil)) + (try (.destroy socket) (catch :default _ nil))) + +(defn close-all! + [] + (doseq [^js socket @*clients] + (try (.terminate socket) (catch :default _ nil))) + (reset! *clients #{})) + +(defn reset-buffer! + [] + (reset! *delta-buffer (sorted-map)) + (reset! *head-rev 0)) diff --git a/src/test/frontend/persist_db_test.cljs b/src/test/frontend/persist_db_test.cljs index e1b8b95..843e73f 100644 --- a/src/test/frontend/persist_db_test.cljs +++ b/src/test/frontend/persist_db_test.cljs @@ -137,6 +137,17 @@ (reset! (:on-error sse-state) on-error) {:close! (fn [] (swap! (:close-count sse-state) inc))}) + ;; connect-events-ws! (the transport start! actually uses since Fase 4) + ;; sets .-onopen/.-onmessage/.-onerror/.-onclose on whatever open-ws-fn + ;; returns, synchronously, right after calling it -- so capturing the + ;; bare socket here is enough for a test to later read e.g. .-onerror + ;; off it and trigger it directly. + :open-ws-fn (fn [_opts] + (swap! (:open-count sse-state) inc) + (let [socket #js {:send (fn [_text] nil)}] + (set! (.-close socket) (fn [] (swap! (:close-count sse-state) inc))) + (reset! (:socket sse-state) socket) + socket)) :schedule-fn (fn [f _delay-ms] (swap! (:scheduled sse-state) conj f) :scheduled)}) @@ -144,6 +155,7 @@ (defn- sse-state [] {:on-error (atom nil) + :socket (atom nil) :open-count (atom 0) :close-count (atom 0) :scheduled (atom [])}) @@ -988,7 +1000,7 @@ :sse sse :ipc-calls ipc-calls})] (-> (p/let [_ (ensure-remote! "logseq_db_graph_a") - on-error @(:on-error sse) + on-error (.-onerror @(:socket sse)) _ (is (fn? on-error)) _ (on-error (js/Error. "connect ECONNREFUSED")) _ (is (= [] @events)) diff --git a/src/test/frontend/worker/db_worker_node_test.cljs b/src/test/frontend/worker/db_worker_node_test.cljs index dbfddf4..64b28db 100644 --- a/src/test/frontend/worker/db_worker_node_test.cljs +++ b/src/test/frontend/worker/db_worker_node_test.cljs @@ -2,6 +2,7 @@ (:require ["fs" :as fs] ["http" :as http] ["path" :as node-path] + ["ws" :as ws] [cljs.test :refer [async deftest is use-fixtures]] [clojure.string :as string] [frontend.test.node-helper :as node-helper] @@ -131,6 +132,69 @@ :headers (assoc headers "Content-Type" "application/octet-stream")} payload))) +(defn- clj (js/JSON.parse (.toString data)) :keywordize-keys true)] + (if-let [resolve! (first @waiters)] + (do (swap! waiters subvec 1) + (resolve! parsed)) + (swap! queue conj parsed))))) + (fn take! + [] + (p/create + (fn [resolve! _reject] + (if (seq @queue) + (let [message (first @queue)] + (swap! queue subvec 1) + (resolve! message)) + (swap! waiters conj resolve!))))))) + +(defn- js payload)))) + (defn- lock-path [root-dir repo] (db-lock/lock-path root-dir repo)) @@ -1019,6 +1083,104 @@ (done))))) (done)))))))) +(deftest db-worker-node-ws-upgrade-rejects-missing-or-invalid-token + (async done + (let [daemon (atom nil) + data-dir (node-helper/create-tmp-dir "db-worker-ws-auth") + repo (str "logseq_db_ws_auth_" (subs (str (random-uuid)) 0 8))] + (-> (p/let [{:keys [host port stop!]} (start-daemon! {:root-dir data-dir :repo repo}) + _ (reset! daemon {:stop! stop!}) + no-token ( (stop!) (p/finally done)) + (done)))))))) + +(deftest db-worker-node-ws-live-broadcast-and-resume-by-revision + (async done + (let [daemon (atom nil) + data-dir (node-helper/create-tmp-dir "db-worker-ws-resume") + repo (str "logseq_db_ws_resume_" (subs (str (random-uuid)) 0 8)) + page-uuid (random-uuid) + block-uuid-1 (random-uuid) + block-uuid-2 (random-uuid) + now (js/Date.now)] + (-> (p/let [{:keys [host port stop!]} (start-daemon! {:root-dir data-dir :repo repo}) + _ (reset! daemon {:stop! stop!}) + token ( (get-in replayed-payload [:delta :rev]) first-rev))) + (is (= "subscribed" (:type ack-message))) + (.close client)) + (p/catch (fn [e] + (println "[db-worker-node-test] e:" e) + (is false (str e)))) + (p/finally (fn [] + (if-let [stop! (:stop! @daemon)] + (-> (stop!) (p/finally done)) + (done)))))))) + (deftest db-worker-node-vector-search-finds-outline-context-after-rebuild (if-not (semantic-search-integration-enabled?) (is true "Skipping semantic search integration test without a macOS embedding endpoint") diff --git a/src/test/frontend/worker/db_worker_node_ws_test.cljs b/src/test/frontend/worker/db_worker_node_ws_test.cljs new file mode 100644 index 0000000..8d16c1f --- /dev/null +++ b/src/test/frontend/worker/db_worker_node_ws_test.cljs @@ -0,0 +1,124 @@ +(ns frontend.worker.db-worker-node-ws-test + "Fast, deterministic tests for the WS resume protocol against a fake + socket -- no real network, no real daemon. See db_worker_node_test.cljs + for the real end-to-end WS-over-a-live-daemon coverage." + (:require [cljs.test :refer [deftest is use-fixtures]] + [clojure.string :as string] + [frontend.worker.db-worker-node-ws :as db-ws])) + +(use-fixtures :each + {:before (fn [] (db-ws/close-all!) (db-ws/reset-buffer!)) + :after (fn [] (db-ws/close-all!) (db-ws/reset-buffer!))}) + +(defn- fake-socket + [] + (let [listeners (atom {}) + sent (atom []) + ^js socket #js {:readyState 1 + :send (fn [text] (swap! sent conj text)) + :on (fn [event handler] + (swap! listeners update (keyword event) (fnil conj []) handler)) + :ping (fn [] nil)}] + (set! (.-terminate socket) (fn [] (set! (.-readyState socket) 3))) + {:sent sent + :listeners listeners + :socket socket})) + +(defn- fire! + [{:keys [listeners]} event & args] + (doseq [handler (get @listeners (keyword event))] + (apply handler args))) + +(defn- sent-json + [{:keys [sent]}] + (mapv #(js->clj (js/JSON.parse %) :keywordize-keys true) @sent)) + +(deftest token-from-upgrade-req-extracts-bare-token + (is (= "abc123" + (db-ws/token-from-upgrade-req + #js {:headers #js {"sec-websocket-protocol" "bearer.abc123"}}))) + (is (nil? (db-ws/token-from-upgrade-req + #js {:headers #js {"sec-websocket-protocol" "not-bearer-prefixed"}}))) + (is (nil? (db-ws/token-from-upgrade-req #js {:headers #js {}})))) + +(deftest reject-upgrade-writes-status-and-destroys-socket + (let [written (atom nil) + destroyed (atom false) + socket #js {:write (fn [text] (reset! written text)) + :destroy (fn [] (reset! destroyed true))}] + (db-ws/reject-upgrade! socket 401 "unauthorized") + (is (string/starts-with? @written "HTTP/1.1 401 unauthorized")) + (is @destroyed))) + +(deftest subscribe-caught-up-replies-subscribed-without-replay + (let [{:keys [socket] :as fake} (fake-socket)] + (db-ws/handle-connection! socket "repo-a") + (db-ws/record-delta! 10 "{\"type\":\"sync-db-changes\",\"payload\":\"rev-10\"}") + (fire! fake "message" + (js/JSON.stringify (clj->js {:type "subscribe" :graphId "repo-a" :sinceRev 10}))) + (is (= [{:type "subscribed" :rev 10}] (sent-json fake))))) + +(deftest subscribe-behind-head-replays-buffered-deltas-in-order + (let [{:keys [socket] :as fake} (fake-socket)] + (db-ws/record-delta! 10 "{\"type\":\"sync-db-changes\",\"payload\":\"rev-10\"}") + (db-ws/record-delta! 11 "{\"type\":\"sync-db-changes\",\"payload\":\"rev-11\"}") + (db-ws/record-delta! 12 "{\"type\":\"sync-db-changes\",\"payload\":\"rev-12\"}") + (db-ws/handle-connection! socket "repo-a") + (fire! fake "message" + (js/JSON.stringify (clj->js {:type "subscribe" :graphId "repo-a" :sinceRev 10}))) + (is (= ["{\"type\":\"sync-db-changes\",\"payload\":\"rev-11\"}" + "{\"type\":\"sync-db-changes\",\"payload\":\"rev-12\"}"] + (butlast @(:sent fake)))) + (is (= {:type "subscribed" :rev 12} (last (sent-json fake)))))) + +(deftest subscribe-with-gap-past-buffer-window-triggers-resync + (let [{:keys [socket] :as fake} (fake-socket)] + (db-ws/record-delta! 500 "{\"type\":\"sync-db-changes\",\"payload\":\"rev-500\"}") + (db-ws/handle-connection! socket "repo-a") + (fire! fake "message" + (js/JSON.stringify (clj->js {:type "subscribe" :graphId "repo-a" :sinceRev 100}))) + (is (= [{:type "resync" :rev 500}] (sent-json fake))))) + +(deftest subscribe-mismatched-graph-id-errors-without-replay-or-resync + (let [{:keys [socket] :as fake} (fake-socket)] + (db-ws/record-delta! 5 "{\"type\":\"sync-db-changes\",\"payload\":\"rev-5\"}") + (db-ws/handle-connection! socket "repo-a") + (fire! fake "message" + (js/JSON.stringify (clj->js {:type "subscribe" :graphId "repo-b" :sinceRev 0}))) + (is (= [{:type "error" :message "graph-id does not match bound repo"}] + (sent-json fake))))) + +(deftest broadcast-reaches-only-currently-connected-sockets + (let [fake-1 (fake-socket) + fake-2 (fake-socket)] + (db-ws/handle-connection! (:socket fake-1) "repo-a") + (db-ws/broadcast! "{\"type\":\"notification\"}") + (db-ws/handle-connection! (:socket fake-2) "repo-a") + (db-ws/broadcast! "{\"type\":\"notification-2\"}") + (is (= ["{\"type\":\"notification\"}" "{\"type\":\"notification-2\"}"] @(:sent fake-1))) + (is (= ["{\"type\":\"notification-2\"}"] @(:sent fake-2))))) + +(deftest close-removes-socket-from-broadcast-set + (let [fake (fake-socket)] + (db-ws/handle-connection! (:socket fake) "repo-a") + (fire! fake "close") + (db-ws/broadcast! "{\"type\":\"notification\"}") + (is (= [] @(:sent fake))))) + +(deftest ping-message-replies-pong + (let [fake (fake-socket)] + (db-ws/handle-connection! (:socket fake) "repo-a") + (fire! fake "message" (js/JSON.stringify (clj->js {:type "ping"}))) + (is (= [{:type "pong"}] (sent-json fake))))) + +(deftest unknown-message-type-replies-error + (let [fake (fake-socket)] + (db-ws/handle-connection! (:socket fake) "repo-a") + (fire! fake "message" (js/JSON.stringify (clj->js {:type "bogus"}))) + (is (= [{:type "error" :message "unknown message type"}] (sent-json fake))))) + +(deftest invalid-json-message-replies-error-without-throwing + (let [fake (fake-socket)] + (db-ws/handle-connection! (:socket fake) "repo-a") + (fire! fake "message" "not json") + (is (= [{:type "error" :message "invalid message"}] (sent-json fake)))))