enhance(db-sync): update doc and align with codebase

2026-04-24 22:25:01 +00:00 · 2026-01-11 18:02:53 +08:00
parent 611a52443e
commit 2bc5783b78
4 changed files with 66 additions and 34 deletions
--- a/deps/db-sync/src/logseq/db_sync/worker.cljs
+++ b/deps/db-sync/src/logseq/db_sync/worker.cljs
@@ -377,7 +377,7 @@
 (defn- handle-tx-batch! [^js self sender txs t-before]
  (let [current-t (t-now self)]
    (cond
-      (not (number? t-before))
+      (or (not (number? t-before)) (neg? t-before))
      {:type "tx/reject"
       :reason "invalid t_before"}

@@ -400,7 +400,7 @@
 (defn- handle-ws-message! [^js self ^js ws raw]
  (let [message (-> raw protocol/parse-message coerce-ws-client-message)]
    (if-not (map? message)
-      (fail-fast :db-sync/request-parse-failed {:raw raw})
+      (send! ws {:type "error" :message "invalid request"})
      (case (:type message)
        "hello"
        (send! ws {:type "hello" :t (t-now self)})
@@ -409,8 +409,11 @@
        (send! ws {:type "pong"})

        "pull"
-        (let [since (or (:since message) 0)]
-          (send! ws (pull-response self since)))
+        (let [raw-since (:since message)
+              since (if (some? raw-since) (parse-int raw-since) 0)]
+          (if (or (and (some? raw-since) (not (number? since))) (neg? since))
+            (send! ws {:type "error" :message "invalid since"})
+            (send! ws (pull-response self since))))

        ;; "snapshot"
        ;; (send! ws (snapshot-response self))
@@ -462,8 +465,11 @@
            (json-response :sync/health {:ok true})

            (and (= method "GET") (= path "/pull"))
-            (let [since (or (parse-int (.get (.-searchParams url) "since")) 0)]
-              (json-response :sync/pull (pull-response self since)))
+            (let [raw-since (.get (.-searchParams url) "since")
+                  since (if (some? raw-since) (parse-int raw-since) 0)]
+              (if (or (and (some? raw-since) (not (number? since))) (neg? since))
+                (bad-request "invalid since")
+                (json-response :sync/pull (pull-response self since))))

            ;; (and (= method "GET") (= path "/snapshot"))
            ;; (common/json-response (snapshot-response self))
--- a/docs/agent-guide/db-sync/db-sync-guide.md
+++ b/docs/agent-guide/db-sync/db-sync-guide.md
@@ -16,6 +16,7 @@ This guide helps AI agents implement and review db-sync features consistently ac
 3) **Server changes**: update `worker.cljs`, `worker_core.cljs`, `storage.cljs`, or `cycle.cljs`.
 4) **Client changes**: update `db_sync.cljs` and thread APIs in `db_worker.cljs`.
 5) **Handler glue**: add or adjust the entry points in `handler/db_based/db_sync.cljs`.
+6) **Function ordering**: keep related ClojureScript fns together and ordered to minimize `declare` usage.

 ## Data, Keywords, and Schema
 - Use existing `:db-sync/*` keywords; add new keywords via `logseq.common.defkeywords/defkeyword`.
@@ -23,17 +24,18 @@ This guide helps AI agents implement and review db-sync features consistently ac

 ## Fail-Fast Error Policy
 - db-sync core code must fail fast on bugs: log an error (`log/error`) and throw immediately.
- Treat these as bugs:
+- Treat these as bugs (internal invariants or bad server responses on the client):
  - db connection missing when required (client or worker).
  - WS/HTTP response fields missing (e.g., `:t`, `:txs`, `:reason`, `:data`).
  - Response parse/coercion failures (e.g., transit decode, malli coercion).
-  - Unexpected WS/HTTP message type or reason value.
+  - Unexpected WS/HTTP message type or reason value on the client.
+  - Asset operations missing required fields when processing client-side metadata.
+  - Invariant violations in tx apply (e.g., tx-data empty after normalization).
+- Server-side validation of client input should not throw. Respond with `tx/reject` or `400` errors for:
  - tx payload type mismatch (e.g., `:txs` not a sequence of strings).
  - Invalid graph identity (missing/empty graph id or uuid in sync path).
  - Invalid or negative `t`/`t_before` values.
-  - Asset operations missing required fields (checksum/type/uuid).
-  - Invariant violations in tx apply (e.g., tx-data empty after normalization).
- Do not silently recover or drop messages for these cases; surface them via exceptions.
+- Do not silently recover or drop messages for bug cases; surface them via exceptions.

 ## HTTP API (Bootstrap + Assets)
 - HTTP endpoints backfill initial graph data, snapshots, and assets.
--- a/docs/agent-guide/db-sync/protocol.md
+++ b/docs/agent-guide/db-sync/protocol.md
@@ -34,27 +34,33 @@
 - `{"type":"pong"}`
  - Keepalive response.
 - `{"type":"error","message":"..."}`
-  - Invalid/unknown message.
+  - Invalid/unknown message. Current messages: `"unknown type"`, `"invalid request"`, `"server error"`, `"invalid since"`.

 ## HTTP API
 - Auth: Bearer token via `Authorization: Bearer <token>` or `?token=...`.
 - JSON body/response unless noted.
+- Auth required for `/graphs`, `/sync/:graph-id/*`, and `/assets/*`. Expect `401` (unauthorized) or `403` (forbidden) on access failure.
+
+### Worker Health
+- `GET /health`
+  - Worker health check. Response: `{"ok":true}`.

 ### Graphs (index DO)
 - `GET /graphs`
-  - List graphs the user owns. Response: `{"graphs":[{graph_id, graph_name, schema_version, created_at, updated_at}...]}`.
+  - List graphs the user owns. Response: `{"graphs":[{graph_id, graph_name, schema_version?, created_at, updated_at}...]}`.
 - `POST /graphs`
-  - Create graph. Body: `{"graph_name":"...","schema_version":"<major>"}`. Response: `{"graph_id":"..."}`.
+  - Create graph. Body: `{"graph_name":"...","schema_version":"<major>"}` (schema_version optional). Response: `{"graph_id":"..."}`.
 - `GET /graphs/:graph-id/access`
-  - Access check. Response: `{"ok":true}` or `403`.
+  - Access check. Response: `{"ok":true}`, `401` (unauthorized), `403` (forbidden), or `404` (not found).
 - `DELETE /graphs/:graph-id`
-  - Delete graph and reset data. Response: `{"graph_id":"...","deleted":true}`.
+  - Delete graph and reset data. Response: `{"graph_id":"...","deleted":true}` or `400` (missing graph id).

 ### Sync (per-graph DO, via `/sync/:graph-id/...`)
 - `GET /sync/:graph-id/health`
  - Health check. Response: `{"ok":true}`.
 - `GET /sync/:graph-id/pull?since=<t>`
  - Same as WS pull. Response: `{"type":"pull/ok","t":<t>,"txs":[{"t":<t>,"tx":"<tx-transit>"}...]}`.
+  - Error response (400): `{"error":"invalid since"}`.
 - `POST /sync/:graph-id/tx/batch`
  - Same as WS tx/batch. Body: `{"t_before":<t>,"txs":["<tx-transit>", ...]}`.
  - Response: `{"type":"tx/batch/ok","t":<t>}` or `{"type":"tx/reject","reason":...}`.
@@ -70,8 +76,9 @@

 ### Assets
 - `GET /assets/:graph-id/:asset-uuid.:ext`
-  - Download asset (binary response, `content-type` set).
+  - Download asset (binary response, `content-type` set, `x-asset-type` header included).
 - `PUT /assets/:graph-id/:asset-uuid.:ext`
  - Upload asset (binary body). Size limit ~100MB. Response: `{"ok":true}`.
 - `DELETE /assets/:graph-id/:asset-uuid.:ext`
  - Delete asset. Response: `{"ok":true}`.
+- Asset error responses: `{"error":"invalid asset path"}` (400), `{"error":"not found"}` (404), `{"error":"asset too large"}` (413), `{"error":"method not allowed"}` (405), `{"error":"missing assets bucket"}` (500).
--- a/src/main/frontend/worker/db_sync.cljs
+++ b/src/main/frontend/worker/db_sync.cljs
@@ -120,6 +120,11 @@
  (when-not (number? value)
    (fail-fast :db-sync/invalid-field (assoc context :value value))))

+(defn- require-non-negative [value context]
+  (require-number value context)
+  (when (neg? value)
+    (fail-fast :db-sync/invalid-field (assoc context :value value))))
+
 (defn- require-seq [value context]
  (when-not (sequential? value)
    (fail-fast :db-sync/invalid-field (assoc context :value value))))
@@ -209,6 +214,12 @@
    :logseq.property.asset/type
    :logseq.property.asset/checksum})

+(defn- require-asset-field
+  [repo field value context]
+  (when (or (nil? value) (and (string? value) (string/blank? value)))
+    (fail-fast :db-sync/missing-field
+               (merge {:repo repo :field field :value value} context))))
+
 (defn- asset-uuids-from-tx [db tx-data]
  (->> tx-data
       (keep (fn [datom]
@@ -404,7 +415,7 @@
          remote-tx (:t message)]
      (case (:type message)
        "hello" (do
-                  (require-number remote-tx {:repo repo :type "hello"})
+                  (require-non-negative remote-tx {:repo repo :type "hello"})
                  (when (> remote-tx local-tx)
                    (send! (:ws client) {:type "pull" :since local-tx}))
                  (enqueue-asset-sync! repo client)
@@ -412,7 +423,7 @@
                  (flush-pending! repo client))
        ;; Upload response
        "tx/batch/ok" (do
-                        (require-number remote-tx {:repo repo :type "tx/batch/ok"})
+                        (require-non-negative remote-tx {:repo repo :type "tx/batch/ok"})
                        (client-op/update-local-tx repo remote-tx)
                        (remove-pending-txs! repo @(:inflight client))
                        (reset! (:inflight client) [])
@@ -420,7 +431,7 @@
        ;; Download response
        ;; Merge batch txs to one tx, does it really work? We'll see
        "pull/ok" (let [txs (:txs message)
-                        _ (require-number remote-tx {:repo repo :type "pull/ok"})
+                        _ (require-non-negative remote-tx {:repo repo :type "pull/ok"})
                        _ (require-seq txs {:repo repo :type "pull/ok" :field :txs})
                        tx (mapcat (fn [data]
                                     (parse-transit (:tx data) {:repo repo :type "pull/ok"}))
@@ -430,13 +441,15 @@
                      (client-op/update-local-tx repo remote-tx)
                      (flush-pending! repo client)))
        "changed" (do
-                    (require-number remote-tx {:repo repo :type "changed"})
+                    (require-non-negative remote-tx {:repo repo :type "changed"})
                    (when (< local-tx remote-tx)
                      (send! (:ws client) {:type "pull" :since local-tx})))
        "tx/reject" (let [reason (:reason message)]
                      (when (nil? reason)
                        (fail-fast :db-sync/missing-field
                                   {:repo repo :type "tx/reject" :field :reason}))
+                      (when (contains? message :t)
+                        (require-non-negative remote-tx {:repo repo :type "tx/reject"}))
                      (case reason
                        "stale"
                        (send! (:ws client) {:type "pull" :since local-tx})
@@ -551,7 +564,12 @@

 (defn- process-asset-op!
  [repo graph-id asset-op]
-  (let [asset-uuid (:block/uuid asset-op)]
+  (let [asset-uuid (:block/uuid asset-op)
+        op-type (cond
+                  (contains? asset-op :update-asset) :update-asset
+                  (contains? asset-op :remove-asset) :remove-asset
+                  :else :unknown)]
+    (require-asset-field repo :asset-uuid asset-uuid {:op op-type})
    (cond
      (contains? asset-op :update-asset)
      (if-let [conn (worker-state/get-datascript-conn repo)]
@@ -559,12 +577,11 @@
              asset-type (:logseq.property.asset/type ent)
              checksum (:logseq.property.asset/checksum ent)
              size (:logseq.property.asset/size ent 0)]
+          (require-asset-field repo :asset-type asset-type {:op :update-asset :asset-uuid asset-uuid})
+          (require-asset-field repo :checksum checksum {:op :update-asset
+                                                        :asset-uuid asset-uuid
+                                                        :asset-type asset-type})
          (cond
-            (or (nil? ent) (nil? asset-type) (nil? checksum))
-            (do
-              (client-op/remove-asset-op repo asset-uuid)
-              (p/resolved nil))
-
            (> size max-asset-size)
            (do
              (log/info :db-sync/asset-too-large {:repo repo
@@ -591,11 +608,11 @@
                             :rtc.exception/upload-asset-failed
                             nil

-                             (log/error :db-sync/asset-upload-failed
-                                        {:repo repo
-                                         :asset-uuid asset-uuid
-                                         :error e})))))))
-        (p/resolved nil))
+                           (log/error :db-sync/asset-upload-failed
+                                      {:repo repo
+                                       :asset-uuid asset-uuid
+                                       :error e})))))))
+        (fail-fast :db-sync/missing-db {:repo repo :op :process-asset-op}))

      (contains? asset-op :remove-asset)
      (-> (p/let [conn (worker-state/get-datascript-conn repo)
@@ -603,9 +620,9 @@
                  asset-type (if (seq (:logseq.property.asset/type ent))
                               (:logseq.property.asset/type ent)
                               (asset-type-from-files repo asset-uuid))]
+            (require-asset-field repo :asset-type asset-type {:op :remove-asset :asset-uuid asset-uuid})
            (p/do!
-             (when (seq asset-type)
-               (delete-remote-asset! repo graph-id asset-uuid asset-type))
+             (delete-remote-asset! repo graph-id asset-uuid asset-type)
             (client-op/remove-asset-op repo asset-uuid)))
          (p/catch (fn [e]
                     (log/error :db-sync/asset-delete-failed