Dasemu
565c2114be
Deletes frontend.worker.sync/* wholesale (apply_txs, asset_db_listener,
assets, auth, client_op, const, crypt, download, handle_message,
large_title, log_and_state, presence, temp_sqlite, transport, upload,
util), worker/sync.cljs, worker/handler/sync.cljs, and
worker/ui_request.cljs -- the remote CRDT sync protocol and its E2EE
password-request plumbing, both made obsolete by the server-client
daemon architecture built in Fases 1-7. Deviates from the plan's KEEP
list on sync/presence.cljs and sync/transport.cljs: neither has any
caller left once the rest of the cluster is gone (confirmed by grep),
because Fase 4's WS resume-by-revision work already built its own
self-contained module (db_worker_node_ws.cljs) instead of adopting
these as the plan assumed it would.
This touched far more than a mechanical deletion:
- frontend.worker.sync.client-op wasn't sync-only. Its client_ops
SQLite table is also the idempotency marker for db_core.cljs's
built-in-property repair migration and the forward/inverse-outliner-
ops relay undo/redo reads back through history-action-ops-by-tx-id.
Trimmed (not deleted) into frontend.worker.local-tx-log, keeping only
the tx-entry/local-tx functions those two callers need and dropping
the genuinely sync-only surface (pending-upload counts, checksums,
graph-uuid, sync-conflict tracking, asset-op queues).
- frontend.worker.sync.apply-txs's apply-history-action! looked
sync-specific (it lived next to the remote-tx-apply/rebase machinery)
but turned out to be the *only* implementation ever registered for
worker-undo-redo/*apply-history-action! -- undo_redo.cljs calls it on
every db-transact undo/redo, local edits included, not just
remote-origin ones. Deleting it outright would have silently broken
undo/redo app-wide the next time anyone hit Cmd-Z. Ported the
general-purpose slice (replaying a recorded entry's forward/inverse
outliner ops against the current db state, plus the raw-tx-data
fallback for edits that bypass the structured outliner-op layer) into
a new frontend.worker.undo-redo.replay, along with persist-local-tx!
(derive history ops on every local commit, stash them in
local-tx-log, hand off to gen-undo-ops!) -- dropping only the
remote-rebase-specific rest (CRDT conflict resolution, upload-queue
bookkeeping). Wired into db_listener.cljs as a renamed
:persist-local-tx handler (was :db-sync, a name that stopped meaning
anything once sync was gone).
Caught by running the full existing undo_redo_test.cljs suite (30
deftests written for the old mechanism) against the new module after
fixing its test harness to use the real db_listener.cljs dispatch
instead of a hand-rolled listener -- all 29 tests port over passing
with no behavior change, which is the actual evidence this preserves
undo/redo correctness rather than just compiling.
- db_core.cljs's own 8 :thread-api/db-sync-* handlers (remote-graph-
download import, close-db/invalidate-search-db/recreate-lock
sync-specific variants) are deleted along with their sync-crypt/
sync-download calls; :thread-api/unsafe-unlink-db, :close-db, and
:release-access-handles keep their real (non-sync) bodies.
db_worker_node.cljs's non-repo-methods set drops from 22 to 7
entries (the 15 removed were all e2ee/db-sync/ui-request keywords
backed by deleted files).
- worker/handler/export.cljs's :thread-api/recompute-checksum-diagnostics
(and worker/db/validate.cljs's backing recompute-checksum-diagnostics,
which pulled in logseq-db-sync.checksum) compared a local checksum
against a remote one that no longer exists -- deleted outright, not
ported, since there's nothing left to diff against.
- worker/handler/block.cljs's :block.temp/sync-conflicts metadata
(backed by the deleted client-op/get-all-sync-conflicts) is gone;
its client-side consumer (a retry-upload/clear-conflicts UI in
components/block.cljs) is dead code now and still needs a follow-up
pass -- flagging here since worker-side cleanup alone can't remove
it without breaking that UI's thread-api calls at runtime.
Verified: app/electron/db-worker-node/mobile all compile clean. Full
worker test suite green (db_core_test 131, db_listener_test 11,
db_worker_test 8, db_worker_node_test 59, db_worker_node_lock_test 10,
local_tx_log_test 4, undo_redo_test 29, subs_test 50, subs_loader_test
2, config_test 6 -- 0 failures across all of them). The entire
src/test/frontend/worker/sync/ test directory (7 files) and
db_sync_test.cljs/db_sync_sim_test.cljs/node_sync_test.cljs (all
testing only-deleted namespaces) are deleted wholesale rather than
ported.
Known follow-up, not yet done: the client-side (frontend.handler.*,
frontend.components.*) RTC/E2EE/db-sync UI and its dead thread-api call
sites (components/block.cljs, components/repo.cljs,
components/settings.cljs, handler/user.cljs, handler/assets.cljs,
db/restore.cljs, handler/common/developer.cljs, and the
handler/db-based/{sync,rtc-flows,rtc-background-tasks}.cljs +
handler/events/rtc.cljs + components/rtc/* + db/rtc/* files themselves)
-- a separate commit.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
946 lines
41 KiB
Clojure
946 lines
41 KiB
Clojure
(ns frontend.worker.db-worker-node
|
|
"Node.js daemon entrypoint for db-worker."
|
|
(:require ["crypto" :as crypto]
|
|
["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]
|
|
[logseq.common.graph-dir :as graph-dir]
|
|
[logseq.common.version :as build-version]
|
|
[logseq.cli.root-dir :as root-dir]
|
|
[logseq.cli.style :as style]
|
|
[logseq.db :as ldb]
|
|
[logseq.db-worker.log :as db-worker-log]
|
|
[logseq.db-worker.server-list :as server-list]
|
|
[promesa.core :as p]))
|
|
|
|
(defonce ^:private *ready? (atom false))
|
|
(defonce ^:private *sse-clients (atom #{}))
|
|
(defonce ^:private *lock-info (atom nil))
|
|
(defonce ^:private *server-list-file (atom nil))
|
|
(defonce ^:private *inflight-by-repo (atom {}))
|
|
(defonce ^:private *max-inflight-by-repo (atom {}))
|
|
|
|
(defn- invoke-started!
|
|
"Increments the /v1/invoke in-flight counter for repo and logs a new
|
|
concurrency high-water mark when observed. Dev-only instrumentation for the
|
|
server-client migration multi-client baseline (see docs/agent-guide); has no
|
|
effect on request handling."
|
|
[repo]
|
|
(let [n (get (swap! *inflight-by-repo update repo (fnil inc 0)) repo)
|
|
prev-max (get @*max-inflight-by-repo repo 0)]
|
|
(when (> n prev-max)
|
|
(swap! *max-inflight-by-repo assoc repo n)
|
|
(log/info :db-worker-node-invoke-concurrency-high-water-mark
|
|
{:repo repo :concurrent n}))))
|
|
|
|
(defn- invoke-finished!
|
|
[repo]
|
|
(swap! *inflight-by-repo update repo (fnil dec 0)))
|
|
|
|
(defn- server-list-file-path
|
|
[root-dir]
|
|
(server-list/path root-dir))
|
|
|
|
(def ^:private default-allowed-origins #{"lsp://logseq.com"})
|
|
|
|
(defn- cors-headers-for-request
|
|
"CORS is a browser-only safety layer, not access control (that's the
|
|
bearer-token + Host checks below) -- it only decides whether a browser is
|
|
allowed to read the response. Reflects the request Origin when allowed,
|
|
per the standard multi-origin CORS pattern (a static header can't list
|
|
more than one origin)."
|
|
[^js req allowed-origins]
|
|
(let [origin (some-> req .-headers .-origin)
|
|
base #js {"Access-Control-Allow-Methods" "GET,POST,DELETE,OPTIONS"
|
|
"Access-Control-Allow-Headers" "Content-Type,Authorization"
|
|
"Vary" "Origin"}]
|
|
(if (and origin (contains? allowed-origins origin))
|
|
(doto base (aset "Access-Control-Allow-Origin" origin))
|
|
base)))
|
|
|
|
(defn- set-cors-headers!
|
|
[^js res headers]
|
|
(set! (.-corsHeaders res) headers))
|
|
|
|
(defn- response-headers
|
|
[^js res headers]
|
|
(js/Object.assign #js {} (or (.-corsHeaders res) #js {}) headers))
|
|
|
|
(defn- send-no-content!
|
|
[^js res]
|
|
(.writeHead res 204 (or (.-corsHeaders res) #js {}))
|
|
(.end res))
|
|
|
|
(defn- send-json!
|
|
[^js res status payload]
|
|
(.writeHead res status (response-headers res #js {"Content-Type" "application/json"}))
|
|
(.end res (js/JSON.stringify (clj->js payload))))
|
|
|
|
(defn- send-text!
|
|
[^js res status text]
|
|
(.writeHead res status (response-headers res #js {"Content-Type" "text/plain"}))
|
|
(.end res text))
|
|
|
|
(defn- <read-body-buffer
|
|
[^js req]
|
|
(p/create
|
|
(fn [resolve reject]
|
|
(let [chunks (array)]
|
|
(.on req "data" (fn [chunk] (.push chunks chunk)))
|
|
(.on req "end" (fn []
|
|
(resolve (js/Buffer.concat chunks))))
|
|
(.on req "error" reject)))))
|
|
|
|
(defn- <read-body
|
|
[^js req]
|
|
(p/then (<read-body-buffer req)
|
|
(fn [buf]
|
|
(.toString buf "utf8"))))
|
|
|
|
(defn- parse-args
|
|
[argv]
|
|
(loop [args (vec (drop 2 argv))
|
|
opts {}]
|
|
(if (empty? args)
|
|
opts
|
|
(let [flag (first args)]
|
|
(case flag
|
|
"--root-dir" (recur (subvec args 2) (assoc opts :root-dir (second args)))
|
|
"--repo" (recur (subvec args 2) (assoc opts :repo (second args)))
|
|
"--owner-source" (recur (subvec args 2) (assoc opts :owner-source (second args)))
|
|
"--log-level" (recur (subvec args 2) (assoc opts :log-level (second args)))
|
|
"--embedding-endpoint" (recur (subvec args 2) (assoc opts :embedding-endpoint (second args)))
|
|
"--embedding-model-id" (recur (subvec args 2) (assoc opts :embedding-model-id (second args)))
|
|
"--host" (recur (subvec args 2) (assoc opts :host (second args)))
|
|
"--port" (recur (subvec args 2) (assoc opts :port (js/parseInt (second args) 10)))
|
|
"--allowed-origin" (recur (subvec args 2) (update opts :allowed-origins (fnil conj #{}) (second args)))
|
|
"--allowed-host" (recur (subvec args 2) (update opts :allowed-hosts (fnil conj #{}) (second args)))
|
|
"--create-empty-db" (recur (subvec args 1) (assoc opts :create-empty-db? true))
|
|
"--version" (recur (subvec args 1) (assoc opts :version? true))
|
|
"--help" (recur (subvec args 1) (assoc opts :help? true))
|
|
(recur (subvec args 1) opts))))))
|
|
|
|
(defn- normalize-owner-source
|
|
[owner-source]
|
|
(db-lock/normalize-owner-source owner-source))
|
|
|
|
(defn- encode-event-type
|
|
[type]
|
|
(cond
|
|
(keyword? type) (subs (str type) 1)
|
|
(string? type) type
|
|
(nil? type) nil
|
|
:else (str type)))
|
|
|
|
(defn- encode-event-payload
|
|
[payload]
|
|
(if (string? payload)
|
|
payload
|
|
(ldb/write-transit-str payload)))
|
|
|
|
(defn- normalize-method-kw
|
|
[method]
|
|
(cond
|
|
(keyword? method) method
|
|
(string? method) (keyword method)
|
|
(nil? method) nil
|
|
:else (keyword (str method))))
|
|
|
|
(defn- normalize-method-str
|
|
[method]
|
|
(cond
|
|
(keyword? method) (subs (str method) 1)
|
|
(string? method) method
|
|
(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)
|
|
:payload (encode-event-payload payload)}))
|
|
message (str "data: " event "\n\n")]
|
|
(doseq [^js res @*sse-clients]
|
|
(try
|
|
(.write res message)
|
|
(catch :default 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]
|
|
(.writeHead res 200 (response-headers res #js {"Content-Type" "text/event-stream"
|
|
"Cache-Control" "no-cache"
|
|
"Connection" "keep-alive"}))
|
|
(.write res "\n")
|
|
(swap! *sse-clients conj res)
|
|
(.on req "close" (fn []
|
|
(swap! *sse-clients disj res))))
|
|
|
|
(defn- <invoke!
|
|
[^js proxy method-str method-kw args]
|
|
(let [args-transit (if (string? args)
|
|
args
|
|
(ldb/write-transit-str args))
|
|
started-at (js/Date.now)
|
|
timeout-id (js/setTimeout
|
|
(fn []
|
|
(log/warn :db-worker-node-invoke-timeout
|
|
{:method (or method-kw method-str)
|
|
:elapsed-ms (- (js/Date.now) started-at)}))
|
|
10000)]
|
|
(-> (p/do! (.remoteInvoke proxy method-str args-transit))
|
|
(p/finally (fn []
|
|
(js/clearTimeout timeout-id))))))
|
|
|
|
(defn- <invoke-binary!
|
|
[^js proxy method-str method-kw repo payload]
|
|
(let [started-at (js/Date.now)
|
|
timeout-id (js/setTimeout
|
|
(fn []
|
|
(log/warn :db-worker-node-invoke-timeout
|
|
{:method (or method-kw method-str)
|
|
:elapsed-ms (- (js/Date.now) started-at)}))
|
|
10000)]
|
|
(-> (p/do! (.remoteInvokeBinary proxy method-str repo payload))
|
|
(p/finally (fn []
|
|
(js/clearTimeout timeout-id))))))
|
|
|
|
(defn- <init-worker!
|
|
[proxy]
|
|
(let [method-kw :thread-api/init
|
|
method-str (normalize-method-str method-kw)]
|
|
(<invoke! proxy method-str method-kw [])))
|
|
|
|
(defn- <close-bound-repo!
|
|
[proxy repo]
|
|
(if (string/blank? repo)
|
|
(p/resolved nil)
|
|
(let [method-kw :thread-api/close-db
|
|
method-str (normalize-method-str method-kw)]
|
|
(-> (<invoke! proxy method-str method-kw [repo])
|
|
(p/catch (fn [error]
|
|
(log/warn :db-worker-node-close-db-before-stop-failed
|
|
{:repo repo
|
|
:error error})
|
|
nil))))))
|
|
|
|
(def ^:private non-repo-methods
|
|
#{:thread-api/init
|
|
:thread-api/list-db
|
|
:thread-api/get-version
|
|
:thread-api/set-context
|
|
:thread-api/sync-app-state
|
|
:thread-api/update-thread-atom
|
|
:thread-api/mobile-logs})
|
|
|
|
(defn- repo-arg
|
|
[args]
|
|
(cond
|
|
(js/Array.isArray args) (aget args 0)
|
|
(sequential? args) (first args)
|
|
:else nil))
|
|
|
|
(defn- repo-error
|
|
[method args bound-repo]
|
|
(let [method-kw (normalize-method-kw method)]
|
|
(when-not (contains? non-repo-methods method-kw)
|
|
(let [repo (repo-arg args)]
|
|
(cond
|
|
(or (not (string? repo))
|
|
(string/blank? repo))
|
|
{:status 400
|
|
:error {:code :missing-repo
|
|
:message "repo is required"}}
|
|
|
|
(not (graph-dir/same-repo? repo bound-repo))
|
|
{:status 409
|
|
:error {:code :repo-mismatch
|
|
:message "repo does not match bound repo"
|
|
:repo repo
|
|
:bound-repo bound-repo}}
|
|
|
|
:else
|
|
nil)))))
|
|
|
|
(defn- set-main-thread-stub!
|
|
[]
|
|
(reset! worker-state/*main-thread
|
|
(fn [qkw & _args]
|
|
(p/rejected (ex-info "main-thread is not available in db-worker-node"
|
|
{:method qkw})))))
|
|
|
|
(defn- query-validation-error?
|
|
[data]
|
|
(= :parser/query (:error data)))
|
|
|
|
(defn- notification-validation-error?
|
|
[data]
|
|
(= :notification (:type data)))
|
|
|
|
(defn- invoke-error-status
|
|
[data]
|
|
(cond
|
|
(query-validation-error? data) 400
|
|
(notification-validation-error? data) 400
|
|
(#{:missing-repo :repo-mismatch :repo-locked} (:code data)) 409
|
|
(number? (:status data)) (:status data)
|
|
:else 500))
|
|
|
|
(defn- invoke-error-code
|
|
[data]
|
|
(cond
|
|
(query-validation-error? data) :invalid-query
|
|
(:code data) (:code data)
|
|
(notification-validation-error? data) :validation-error
|
|
:else :exception))
|
|
|
|
(defn- invoke-error-message
|
|
[error data]
|
|
(or (get-in data [:payload :message])
|
|
(.-message error)
|
|
(str error)))
|
|
|
|
(defn- health-payload
|
|
"Unauthenticated by design (see request-authorized?), so this intentionally
|
|
omits :pid and :root-dir -- filesystem/process details that shouldn't be
|
|
handed to anything reachable on the allowed network segment without a
|
|
bearer token."
|
|
[{:keys [bound-repo host port owner-source]}]
|
|
{:repo bound-repo
|
|
:status (if @*ready? "ready" "starting")
|
|
:host host
|
|
:port (if (satisfies? IDeref port) @port port)
|
|
:owner-source (name (normalize-owner-source owner-source))
|
|
:revision (build-version/revision)})
|
|
|
|
(defn- log-invoke-error!
|
|
[res error method-kw]
|
|
(let [data (ex-data error)
|
|
status (invoke-error-status data)
|
|
code (invoke-error-code data)
|
|
message (invoke-error-message error data)
|
|
payload {:ok false
|
|
:error {:code code
|
|
:message message}}]
|
|
(log/error :db-worker-node-invoke-failed
|
|
{:status status
|
|
:code code
|
|
:error error
|
|
:message message
|
|
:method method-kw})
|
|
(send-json! res status payload)))
|
|
|
|
(defn- effective-port
|
|
[port]
|
|
(if (satisfies? IDeref port) @port port))
|
|
|
|
(defn- allowed-hosts-for
|
|
"Host header allowlist, guarding against DNS rebinding: without this, any
|
|
web page the operator visits could script a request to the daemon on
|
|
loopback. Always includes loopback for the bound port; a non-default
|
|
--host (e.g. 0.0.0.0 behind Tailscale) needs its reachable name/IP added
|
|
explicitly via --allowed-host."
|
|
[{:keys [port extra-allowed-hosts]}]
|
|
(let [p (effective-port port)]
|
|
(into #{(str "localhost:" p) (str "127.0.0.1:" p) (str "[::1]:" p)}
|
|
extra-allowed-hosts)))
|
|
|
|
(defn- host-allowed?
|
|
[^js req allowed-hosts]
|
|
(contains? allowed-hosts (some-> req .-headers .-host)))
|
|
|
|
(defn- asset-get-request?
|
|
[request-path method]
|
|
(and (= method "GET") (string/starts-with? request-path "/v1/assets/")))
|
|
|
|
(defn- request-authorized?
|
|
"The Authorization header works for every route, including these two --
|
|
a plain JS fetch() can set it fine. It's only a *browser navigating
|
|
directly to a URL* that can't: EventSource (/v1/events, SSE, now a
|
|
fallback since the Fase 4 WS migration) has no header API, and an
|
|
<img src>/direct navigation to an asset GET can't set custom headers
|
|
either. Both accept ?token= as a fallback in place of Authorization, never
|
|
as a replacement for it."
|
|
[^js req parsed-url request-path method]
|
|
(or (auth/authorized? (.. req -headers -authorization))
|
|
(and (or (= request-path "/v1/events")
|
|
(asset-get-request? request-path method))
|
|
(auth/authorized? (some->> (.get (.-searchParams parsed-url) "token")
|
|
(str "Bearer "))))))
|
|
|
|
(def ^:private asset-content-types
|
|
{"png" "image/png"
|
|
"jpg" "image/jpeg"
|
|
"jpeg" "image/jpeg"
|
|
"gif" "image/gif"
|
|
"webp" "image/webp"
|
|
"svg" "image/svg+xml"
|
|
"avif" "image/avif"
|
|
"bmp" "image/bmp"
|
|
"ico" "image/x-icon"
|
|
"pdf" "application/pdf"
|
|
"mp3" "audio/mpeg"
|
|
"wav" "audio/wav"
|
|
"ogg" "audio/ogg"
|
|
"mp4" "video/mp4"
|
|
"webm" "video/webm"
|
|
"mov" "video/quicktime"
|
|
"json" "application/json"
|
|
"txt" "text/plain"})
|
|
|
|
(defn- guess-asset-content-type
|
|
[file-name]
|
|
(let [ext (some-> (re-find #"\.([A-Za-z0-9]+)$" (or file-name ""))
|
|
second
|
|
string/lower-case)]
|
|
(get asset-content-types ext "application/octet-stream")))
|
|
|
|
(defn- parse-asset-path
|
|
"/v1/assets/:repo/:name -- both segments are URI-encoded by the client
|
|
(frontend.persist-db.remote), since a repo or asset file name could
|
|
contain characters that aren't legal in a URL path segment as-is."
|
|
[request-path]
|
|
(let [rest (subs request-path (count "/v1/assets/"))
|
|
slash-idx (string/index-of rest "/")]
|
|
(when (and slash-idx (pos? slash-idx) (< (inc slash-idx) (count rest)))
|
|
(try
|
|
{:repo (js/decodeURIComponent (subs rest 0 slash-idx))
|
|
:file-name (js/decodeURIComponent (subs rest (inc slash-idx)))}
|
|
(catch :default _ nil)))))
|
|
|
|
(defn- asset-etag
|
|
[^js buf]
|
|
(str "\"" (-> (crypto/createHash "sha1") (.update buf) (.digest "hex")) "\""))
|
|
|
|
(defn- <process-batch-invoke!
|
|
"One entry of a /v1/invoke-batch request, run through the same
|
|
repo-check/lock/invoke path as /v1/invoke but never rejecting -- a failed
|
|
item becomes {:ok false :error ...} in its slot of the batch response
|
|
instead of failing every other invoke that happened to coalesce into the
|
|
same HTTP request (Fase 6)."
|
|
[proxy bound-repo {:keys [method argsTransit args]}]
|
|
(let [method-kw (normalize-method-kw method)
|
|
method-str (normalize-method-str method)
|
|
args' (or argsTransit args)]
|
|
(-> (p/let [args-for-validation (if (string? args')
|
|
(ldb/read-transit-str args')
|
|
args')]
|
|
(if-let [{:keys [error]} (repo-error method-kw args-for-validation bound-repo)]
|
|
{:ok false :error error}
|
|
(p/let [_ (when-not (contains? non-repo-methods method-kw)
|
|
(let [{:keys [path lock]} @*lock-info]
|
|
(db-lock/assert-lock-owner! path lock)))
|
|
_ (invoke-started! bound-repo)
|
|
result (-> (<invoke! proxy method-str method-kw args')
|
|
(p/finally (fn [] (invoke-finished! bound-repo))))]
|
|
{:ok true :resultTransit result})))
|
|
(p/catch (fn [error]
|
|
(let [data (ex-data error)]
|
|
{:ok false
|
|
:error {:code (invoke-error-code data)
|
|
:message (invoke-error-message error data)}}))))))
|
|
|
|
(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]}]
|
|
(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)
|
|
method (.-method req)
|
|
;; Computed per-request (not once at server start) because `port`
|
|
;; is an atom that only resolves to the real OS-assigned port
|
|
;; after `.listen` calls back -- see `start-http-server!`.
|
|
allowed-hosts (allowed-hosts-for {:port port :extra-allowed-hosts extra-allowed-hosts})]
|
|
(set-cors-headers! res (cors-headers-for-request req allowed-origins))
|
|
(cond
|
|
(= method "OPTIONS")
|
|
(send-no-content! res)
|
|
|
|
(not (host-allowed? req allowed-hosts))
|
|
(send-json! res 403 {:ok false :error {:code :host-not-allowed
|
|
:message "Host header not allowed"}})
|
|
|
|
;; Unauthenticated on purpose: readiness polling (daemon lifecycle,
|
|
;; CLI/Electron discovery) happens before any session exists, so
|
|
;; this can't require a token without a chicken-and-egg startup
|
|
;; deadlock. Still behind the Host allowlist above. health-payload
|
|
;; omits root-dir/pid so it doesn't leak filesystem/process details
|
|
;; to anything reachable on the allowed network segment.
|
|
(= request-path "/healthz")
|
|
(send-json! res (if @*ready? 200 503)
|
|
(health-payload {:bound-repo bound-repo
|
|
:host host
|
|
:port port
|
|
:owner-source owner-source}))
|
|
|
|
(and (= request-path "/v1/auth/login") (= method "POST"))
|
|
(-> (p/let [body (<read-body req)
|
|
{:keys [username password deviceLabel]}
|
|
(js->clj (js/JSON.parse body) :keywordize-keys true)
|
|
{:keys [session error]} (auth/login! root-dir
|
|
{:username username
|
|
:password password
|
|
:device-label deviceLabel})]
|
|
(if session
|
|
(send-json! res 200 {:ok true
|
|
:token (:token session)
|
|
:sessionId (:id session)
|
|
:deviceLabel (:device-label session)})
|
|
(send-json! res 401 {:ok false :error {:code error :message "invalid credentials"}})))
|
|
(p/catch (fn [_error]
|
|
(send-json! res 400 {:ok false :error {:code :invalid-request
|
|
:message "invalid login request body"}}))))
|
|
|
|
(not (request-authorized? req parsed-url request-path method))
|
|
(send-json! res 401 {:ok false :error {:code :unauthorized
|
|
:message "missing or invalid credentials"}})
|
|
|
|
(and (= request-path "/v1/auth/sessions") (= method "GET"))
|
|
(send-json! res 200 {:ok true :sessions (auth/list-sessions)})
|
|
|
|
(and (string/starts-with? request-path "/v1/auth/sessions/") (= method "DELETE"))
|
|
(send-json! res 200 {:ok true
|
|
:revoked (auth/revoke-session!
|
|
(subs request-path (count "/v1/auth/sessions/")))})
|
|
|
|
(= request-path "/v1/events")
|
|
(sse-handler req res)
|
|
|
|
(= request-path "/v1/import-db-binary")
|
|
(if (= method "POST")
|
|
(let [repo (.get (.-searchParams parsed-url) "repo")
|
|
method-kw :thread-api/import-db-binary
|
|
method-str (normalize-method-str method-kw)]
|
|
(-> (p/let [binary (<read-body-buffer req)
|
|
args-for-validation [repo binary]]
|
|
(if-let [{:keys [status error]} (repo-error method-kw args-for-validation bound-repo)]
|
|
(send-json! res status {:ok false :error error})
|
|
(p/let [_ (let [{:keys [path lock]} @*lock-info]
|
|
(db-lock/assert-lock-owner! path lock))
|
|
result (<invoke-binary! proxy method-str method-kw repo binary)]
|
|
(send-json! res 200 {:ok true :resultTransit (ldb/write-transit-str result)}))))
|
|
(p/catch (fn [error]
|
|
(let [data (ex-data error)
|
|
status (invoke-error-status data)
|
|
code (invoke-error-code data)
|
|
message (invoke-error-message error data)
|
|
payload {:ok false
|
|
:error {:code code
|
|
:message message}}]
|
|
(log/error :db-worker-node-invoke-failed
|
|
{:status status
|
|
:code code
|
|
:method method-str
|
|
:error error})
|
|
(send-json! res status payload))))))
|
|
(send-text! res 405 "method-not-allowed"))
|
|
|
|
(and (string/starts-with? request-path "/v1/assets/") (= method "GET"))
|
|
(if-let [{:keys [repo file-name]} (parse-asset-path request-path)]
|
|
(let [method-kw :thread-api/asset-read-bytes
|
|
method-str (normalize-method-str method-kw)
|
|
args [repo file-name]]
|
|
(if-let [{:keys [status error]} (repo-error method-kw args bound-repo)]
|
|
(send-json! res status {:ok false :error error})
|
|
(-> (p/let [result (<invoke! proxy method-str method-kw args)
|
|
bytes (ldb/read-transit-str result)]
|
|
(if (some? bytes)
|
|
(let [buf (js/Buffer.from bytes)
|
|
etag (asset-etag buf)]
|
|
(if (= etag (aget (.-headers req) "if-none-match"))
|
|
(do
|
|
(.writeHead res 304 (response-headers res #js {"ETag" etag
|
|
"Cache-Control" "public, max-age=31536000, immutable"}))
|
|
(.end res))
|
|
(do
|
|
(.writeHead res 200 (response-headers
|
|
res #js {"Content-Type" (guess-asset-content-type file-name)
|
|
"Content-Length" (.-length buf)
|
|
"ETag" etag
|
|
"Cache-Control" "public, max-age=31536000, immutable"}))
|
|
(.end res buf))))
|
|
(send-json! res 404 {:ok false :error {:code :not-found :message "asset not found"}})))
|
|
(p/catch (fn [error]
|
|
(log-invoke-error! res error method-kw))))))
|
|
(send-json! res 400 {:ok false :error {:code :invalid-request :message "invalid asset path"}}))
|
|
|
|
(and (string/starts-with? request-path "/v1/assets/") (= method "PUT"))
|
|
(if-let [{:keys [repo file-name]} (parse-asset-path request-path)]
|
|
(-> (p/let [binary (<read-body-buffer req)
|
|
method-kw :thread-api/asset-write-bytes
|
|
method-str (normalize-method-str method-kw)
|
|
args [repo file-name binary]]
|
|
(if-let [{:keys [status error]} (repo-error method-kw args bound-repo)]
|
|
(send-json! res status {:ok false :error error})
|
|
(p/let [_ (let [{:keys [path lock]} @*lock-info]
|
|
(db-lock/assert-lock-owner! path lock))
|
|
_ (<invoke! proxy method-str method-kw args)]
|
|
(send-json! res 200 {:ok true}))))
|
|
(p/catch (fn [error]
|
|
(log-invoke-error! res error :thread-api/asset-write-bytes))))
|
|
(send-json! res 400 {:ok false :error {:code :invalid-request :message "invalid asset path"}}))
|
|
|
|
(= request-path "/v1/invoke-batch")
|
|
(if (= method "POST")
|
|
(-> (p/let [body (<read-body req)
|
|
{:keys [invokes]} (js->clj (js/JSON.parse body) :keywordize-keys true)
|
|
results (p/all (map #(<process-batch-invoke! proxy bound-repo %) invokes))]
|
|
(send-json! res 200 {:ok true :results results}))
|
|
(p/catch (fn [error]
|
|
(log-invoke-error! res error nil))))
|
|
(send-text! res 405 "method-not-allowed"))
|
|
|
|
(= request-path "/v1/invoke")
|
|
(if (= method "POST")
|
|
(->
|
|
(p/let [body (<read-body req)
|
|
payload (js/JSON.parse body)
|
|
{:keys [method argsTransit args]} (js->clj payload :keywordize-keys true)
|
|
method-kw (normalize-method-kw method)
|
|
method-str (normalize-method-str method)]
|
|
(-> (p/let [args' (or argsTransit args)
|
|
args-for-validation (if (string? args')
|
|
(ldb/read-transit-str args')
|
|
args')]
|
|
(if-let [{:keys [status error]} (repo-error method-kw args-for-validation bound-repo)]
|
|
(send-json! res status {:ok false :error error})
|
|
(p/let [_ (when-not (contains? non-repo-methods method-kw)
|
|
(let [{:keys [path lock]} @*lock-info]
|
|
(db-lock/assert-lock-owner! path lock)))
|
|
_ (invoke-started! bound-repo)
|
|
result (-> (<invoke! proxy method-str method-kw args')
|
|
(p/finally (fn [] (invoke-finished! bound-repo))))]
|
|
(send-json! res 200 {:ok true :resultTransit result}))))
|
|
(p/catch (fn [error]
|
|
(log-invoke-error! res error method-kw)))))
|
|
(p/catch (fn [error]
|
|
(log-invoke-error! res error nil))))
|
|
(send-text! res 405 "method-not-allowed"))
|
|
|
|
(= url "/v1/shutdown")
|
|
(if (= method "POST")
|
|
(do
|
|
(send-json! res 200 {:ok true})
|
|
(js/setTimeout (fn []
|
|
(when stop-fn
|
|
(stop-fn)))
|
|
10))
|
|
(send-text! res 405 "method-not-allowed"))
|
|
|
|
:else
|
|
(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!
|
|
[]
|
|
(println (str (style/bold "db-worker-node") " " (style/bold "options") ":"))
|
|
(println (str " " (style/bold "--root-dir") " <path> (required)"))
|
|
(println (str " " (style/bold "--repo") " <name> (required)"))
|
|
(println (str " " (style/bold "--create-empty-db") " (start with empty initial datoms)"))
|
|
(println (str " " (style/bold "--embedding-endpoint") " <url>"))
|
|
(println (str " " (style/bold "--embedding-model-id") " <id>"))
|
|
(println (str " " (style/bold "--log-level") " <level> (default info)"))
|
|
(println (str " " (style/bold "--host") " <host> (default 127.0.0.1; use an explicit --allowed-host too when binding beyond loopback)"))
|
|
(println (str " " (style/bold "--port") " <port> (default OS-assigned)"))
|
|
(println (str " " (style/bold "--allowed-origin") " <origin> (repeatable; browser CORS allowlist, not access control)"))
|
|
(println (str " " (style/bold "--allowed-host") " <host:port> (repeatable; Host header allowlist beyond loopback)"))
|
|
(println (str " " (style/bold "--version") " (print build metadata and exit)"))
|
|
(println " Requires a credential: set LOGSEQ_DAEMON_USERNAME and LOGSEQ_DAEMON_PASSWORD")
|
|
(println " on first run to bootstrap <root-dir>/db-worker-auth.json, or restore that file.")
|
|
(println " logs: <root-dir>/graphs/<graph-dir>/db-worker-node-YYYYMMDD.log (retains 7)"))
|
|
|
|
(defn- startup-db-opts
|
|
[{:keys [create-empty-db?]}]
|
|
(if create-empty-db?
|
|
{:datoms []
|
|
:sync-download-graph? true}
|
|
{}))
|
|
|
|
(defn- assert-lock-owner!
|
|
[]
|
|
(let [{:keys [path lock]} @*lock-info]
|
|
(db-lock/assert-lock-owner! path lock)))
|
|
|
|
(defn- recreate-lock!
|
|
[target-repo]
|
|
(p/let [{:keys [path lock]} @*lock-info
|
|
_ (when-not (and (seq path) lock)
|
|
(throw (ex-info "lock owner missing"
|
|
{:code :repo-locked
|
|
:repo target-repo})))
|
|
_ (when (and (seq target-repo)
|
|
(not (graph-dir/same-repo? target-repo (:repo lock))))
|
|
(throw (ex-info "graph lock repo mismatch"
|
|
{:code :repo-locked
|
|
:repo target-repo
|
|
:bound-repo (:repo lock)})))
|
|
updated-lock (db-lock/update-lock! path lock)]
|
|
(swap! *lock-info assoc :lock updated-lock)
|
|
nil))
|
|
|
|
(defn- close-server!
|
|
[server]
|
|
(p/create
|
|
(fn [resolve _]
|
|
(try
|
|
(.close server (fn [] (resolve true)))
|
|
(catch :default _
|
|
(resolve true))))))
|
|
|
|
(defn- clear-runtime-state!
|
|
[actual-port]
|
|
(reset! *ready? false)
|
|
(when-let [file-path @*server-list-file]
|
|
(server-list/remove-entry! file-path {:pid (.-pid js/process)
|
|
:port actual-port}))
|
|
(doseq [^js res @*sse-clients]
|
|
(try
|
|
(.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)))
|
|
|
|
(defn- make-stop!
|
|
[{:keys [proxy repo actual-port server stopped? on-stopped!]}]
|
|
(fn []
|
|
(if @stopped?
|
|
(p/resolved true)
|
|
(do
|
|
(reset! stopped? true)
|
|
(-> (p/let [_ (<close-bound-repo! proxy repo)]
|
|
(clear-runtime-state! actual-port)
|
|
(close-server! server)
|
|
true)
|
|
(p/finally
|
|
(fn []
|
|
(when (fn? on-stopped!)
|
|
(on-stopped!))
|
|
(db-worker-log/uninstall!))))))))
|
|
|
|
(defn- resolve-listening-daemon!
|
|
[{:keys [server proxy repo host port* stop!* stopped? on-stopped!]} resolve]
|
|
(let [address (.address server)
|
|
actual-port (if (number? address) address (.-port address))
|
|
_ (reset! port* actual-port)
|
|
stop! (make-stop! {:proxy proxy
|
|
:repo repo
|
|
:actual-port actual-port
|
|
:server server
|
|
:stopped? stopped?
|
|
:on-stopped! on-stopped!})]
|
|
(reset! *ready? true)
|
|
(when-let [file-path @*server-list-file]
|
|
(server-list/append-entry! file-path {:pid (.-pid js/process)
|
|
:port actual-port}))
|
|
(reset! stop!* stop!)
|
|
(resolve {:host host
|
|
:port actual-port
|
|
:server server
|
|
:stop! stop!})))
|
|
|
|
(defn- start-http-server!
|
|
[{:keys [proxy repo host port owner-source root-dir on-stopped!
|
|
allowed-origins extra-allowed-hosts]}]
|
|
(let [stop!* (atom nil)
|
|
stopped? (atom false)
|
|
port* (atom nil)
|
|
server (make-server proxy {:bound-repo repo
|
|
:host host
|
|
:port port*
|
|
:owner-source owner-source
|
|
:root-dir root-dir
|
|
:allowed-origins allowed-origins
|
|
:extra-allowed-hosts extra-allowed-hosts
|
|
:stop-fn (fn []
|
|
(when-let [stop! @stop!*]
|
|
(stop!)))})]
|
|
(p/create
|
|
(fn [resolve reject]
|
|
(.listen server port host
|
|
(fn []
|
|
(resolve-listening-daemon! {:server server
|
|
:proxy proxy
|
|
:repo repo
|
|
:host host
|
|
:port* port*
|
|
:stop!* stop!*
|
|
:stopped? stopped?
|
|
:owner-source owner-source
|
|
:root-dir root-dir
|
|
:on-stopped! on-stopped!}
|
|
resolve)))
|
|
(.on server "error" (fn [error]
|
|
(when-let [lock-path (:path @*lock-info)]
|
|
(db-lock/remove-lock! lock-path))
|
|
(reject error)))))))
|
|
|
|
(defn start-daemon!
|
|
[{:keys [root-dir repo log-level owner-source on-stopped!
|
|
host port allowed-origins allowed-hosts] :as opts}]
|
|
(let [host (or host "127.0.0.1")
|
|
port (or port 0)
|
|
allowed-origins (into default-allowed-origins allowed-origins)
|
|
owner-source (normalize-owner-source owner-source)]
|
|
(cond
|
|
(not (seq root-dir))
|
|
(p/rejected (ex-info "root-dir is required" {:code :missing-root-dir}))
|
|
|
|
(not (seq repo))
|
|
(p/rejected (ex-info "repo is required" {:code :missing-repo}))
|
|
|
|
:else
|
|
(try
|
|
(let [root-dir (root-dir/ensure-root-dir! root-dir)
|
|
_ (auth/ensure-credential! root-dir)
|
|
server-list-file (server-list-file-path root-dir)]
|
|
(db-worker-log/install! {:root-dir root-dir
|
|
:repo repo
|
|
:log-level (keyword (or log-level "info"))})
|
|
(log/info :db-worker-node-version {:build-time (build-version/build-time)
|
|
:revision (build-version/revision)})
|
|
(reset! *ready? false)
|
|
(reset! *lock-info nil)
|
|
(reset! *server-list-file server-list-file)
|
|
(set-main-thread-stub!)
|
|
(-> (p/let [platform (platform-node/node-platform {:root-dir root-dir
|
|
:event-fn handle-event!
|
|
:write-guard-fn assert-lock-owner!
|
|
:owner-source owner-source
|
|
:recreate-lock-fn recreate-lock!
|
|
:embedding-endpoint (:embedding-endpoint opts)
|
|
:embedding-model-id (:embedding-model-id opts)})
|
|
proxy (db-core/init-core! platform)
|
|
_ (<init-worker! proxy)
|
|
{:keys [path lock]} (db-lock/ensure-lock! {:root-dir root-dir
|
|
:repo repo
|
|
:owner-source owner-source})
|
|
_ (reset! *lock-info {:path path :lock lock})
|
|
_ (let [method-kw :thread-api/create-or-open-db
|
|
method-str (normalize-method-str method-kw)]
|
|
(<invoke! proxy method-str method-kw [repo (startup-db-opts opts)]))]
|
|
(start-http-server! {:proxy proxy
|
|
:repo repo
|
|
:host host
|
|
:port port
|
|
:owner-source owner-source
|
|
:root-dir root-dir
|
|
:allowed-origins allowed-origins
|
|
:extra-allowed-hosts allowed-hosts
|
|
:on-stopped! on-stopped!}))
|
|
(p/catch (fn [e]
|
|
(when-let [lock-path (:path @*lock-info)]
|
|
(db-lock/remove-lock! lock-path))
|
|
(throw e)))))
|
|
(catch :default e
|
|
(p/rejected e))))))
|
|
|
|
(defn main
|
|
[]
|
|
(let [{:keys [root-dir repo help? version? owner-source] :as opts}
|
|
(parse-args (.-argv js/process))]
|
|
(when help?
|
|
(show-help!)
|
|
(.exit js/process 0))
|
|
(when version?
|
|
(println (build-version/format-version))
|
|
(.exit js/process 0))
|
|
(when-not (seq root-dir)
|
|
(.error js/console "root-dir is required")
|
|
(.exit js/process 1))
|
|
(when-not (seq repo)
|
|
(.error js/console "repo is required")
|
|
(.exit js/process 1))
|
|
(-> (p/let [{:keys [stop!] :as daemon}
|
|
(start-daemon! {:root-dir root-dir
|
|
:repo repo
|
|
:create-empty-db? (:create-empty-db? opts)
|
|
:owner-source owner-source
|
|
:embedding-endpoint (:embedding-endpoint opts)
|
|
:embedding-model-id (:embedding-model-id opts)
|
|
:host (:host opts)
|
|
:port (:port opts)
|
|
:allowed-origins (:allowed-origins opts)
|
|
:allowed-hosts (:allowed-hosts opts)
|
|
:on-stopped! (fn []
|
|
(log/info :db-worker-node-stopped nil)
|
|
(.exit js/process 0))
|
|
:log-level (:log-level opts)})]
|
|
(log/info :db-worker-node-ready {:host (:host daemon) :port (:port daemon)})
|
|
(let [shutdown (fn [] (stop!))]
|
|
(.on js/process "SIGINT" shutdown)
|
|
(.on js/process "SIGTERM" shutdown)))
|
|
(p/catch (fn [error]
|
|
(let [data (ex-data error)
|
|
code (:code data)
|
|
message (or (.-message error) (str error))]
|
|
(cond
|
|
(#{:missing-root-dir :root-dir-permission} code)
|
|
(.error js/console message)
|
|
|
|
(or (string/includes? message ".node")
|
|
(string/includes? message "Cannot find module")
|
|
(string/includes? message "MODULE_NOT_FOUND")
|
|
(string/includes? message "bindings file"))
|
|
(.error js/console
|
|
(str "db-worker-node failed to start: bundled runtime files are missing or incomplete. "
|
|
"Rebuild with `pnpm db-worker-node:release:bundle` and ensure "
|
|
"`dist/db-worker-node.js` exists and assets listed in "
|
|
"`dist/db-worker-node-assets.json` are next to it. "
|
|
"Root error: "
|
|
message))
|
|
|
|
:else
|
|
(.error js/console (str "db-worker-node failed to start: " message)))
|
|
(when-let [stack (.-stack error)]
|
|
(.error js/console stack))
|
|
(.exit js/process 1)))))))
|