fix(db-sync): stream snapshot downloads directly from sync do

Author: deephbz

deps/db-sync/src/logseq/db_sync/worker/handler/sync.cljs

@@ -129,6 +129,10 @@
   (let [url (js/URL. (.-url request))]
     (str (.-origin url) "/assets/" graph-id "/" snapshot-id ".snapshot")))
 
+(defn- snapshot-stream-url [request graph-id]
+  (let [url (js/URL. (.-url request))]
+    (str (.-origin url) "/sync/" graph-id "/snapshot/stream")))
+
 (defn- maybe-decompress-stream [stream encoding]
   (if (and (= encoding snapshot-content-encoding) (exists? js/DecompressionStream))
     (.pipeThrough stream (js/DecompressionStream. "gzip"))
@@ -419,35 +423,17 @@
 
 (defn- handle-sync-snapshot-download
   [^js self request]
-  (let [graph-id (graph-id-from-request request)
-        ^js bucket (.-LOGSEQ_SYNC_ASSETS (.-env self))]
+  (let [graph-id (graph-id-from-request request)]
     (cond
       (not (seq graph-id))
       (http/bad-request "missing graph id")
 
-      (nil? bucket)
-      (http/error-response "missing assets bucket" 500)
-
       :else
       (p/let [ready-for-sync? (<ready-for-sync? self graph-id)]
         (if-not ready-for-sync?
           (http/error-response "graph not ready" 409)
-          (p/let [snapshot-id (str (random-uuid))
-                  key (snapshot-key graph-id snapshot-id)
-                  stream (-> (snapshot-export-stream self)
-                             (maybe-compress-stream))
-                  multipart? (and (some? (.-createMultipartUpload bucket))
-                                  (fn? (.-createMultipartUpload bucket)))
-                  opts #js {:httpMetadata #js {:contentType snapshot-content-type
-                                               :contentEncoding snapshot-content-encoding
-                                               :cacheControl snapshot-cache-control}
-                            :customMetadata #js {:purpose "snapshot"
-                                                 :created-at (str (common/now-ms))}}
-                  _ (if multipart?
-                      (upload-multipart! bucket key stream opts)
-                      (p/let [body (<buffer-stream stream)]
-                        (.put bucket key body opts)))
-                  url (snapshot-url request graph-id snapshot-id)]
+          (let [key (str "stream/" graph-id ".snapshot")
+                url (snapshot-stream-url request graph-id)]
             (http/json-response :sync/snapshot-download {:ok true
                                                          :key key
                                                          :url url

deps/db-sync/test/logseq/db_sync/worker_handler_sync_test.cljs

@@ -29,55 +29,29 @@
 
 (deftest snapshot-download-uses-gzip-encoding-when-compression-supported-test
   (async done
-         (let [put-call (atom nil)
-               rows [[1 "row-1" nil]
-                     [2 "row-2" "{\"a\":1}"]]
-               bucket #js {:put (fn [key body opts]
-                                  (reset! put-call {:key key :body body :opts opts})
-                                  (js/Promise.resolve #js {:ok true}))}
-               sql (empty-sql)
+         (let [sql (empty-sql)
                conn (d/create-conn db-schema/schema)
-               self #js {:env #js {:LOGSEQ_SYNC_ASSETS bucket}
+               self #js {:env #js {}
                          :conn conn
                          :schema-ready true
                          :sql sql}
                {:keys [request url]} (request-url)
-               original-compression-stream (.-CompressionStream js/globalThis)
-               restore! #(aset js/globalThis "CompressionStream" original-compression-stream)]
-           (aset js/globalThis
-                 "CompressionStream"
-                 (passthrough-compression-stream-constructor))
-           (-> (p/with-redefs [sync-handler/fetch-snapshot-kvs-rows (fn [_sql last-addr _limit]
-                                                                      (if (neg? last-addr) rows []))
-                               sync-handler/snapshot-row-count (fn [_sql] (count rows))]
-                 (p/let [resp (sync-handler/handle {:self self
-                                                    :request request
-                                                    :url url
-                                                    :route {:handler :sync/snapshot-download}})
+               expected-url "http://localhost/sync/graph-1/snapshot/stream"]
+           (-> (p/let [resp (sync-handler/handle {:self self
+                                                  :request request
+                                                  :url url
+                                                  :route {:handler :sync/snapshot-download}})
                        text (.text resp)
-                       body (js->clj (js/JSON.parse text) :keywordize-keys true)
-                       http-metadata (aget (:opts @put-call) "httpMetadata")
-                       payload (js/Uint8Array. (:body @put-call))
-                       rows (snapshot/finalize-framed-buffer payload)
-                       addrs (mapv first rows)]
+                       body (js->clj (js/JSON.parse text) :keywordize-keys true)]
                  (is (= 200 (.-status resp)))
+                 (is (= true (:ok body)))
+                 (is (= "stream/graph-1.snapshot" (:key body)))
+                 (is (= expected-url (:url body)))
                  (is (= "gzip" (:content-encoding body)))
-                 (is (= "gzip" (aget http-metadata "contentEncoding")))
-                 (is (= "application/transit+json" (aget http-metadata "contentType")))
-                 (is (= 2 (count rows)))
-                 (is (= (sort addrs) addrs))
-                 (is (every? (fn [[addr content _addresses]]
-                               (and (int? addr)
-                                    (string? content)))
-                             rows))
-                 (is (= [[1 "row-1" nil]
-                         [2 "row-2" "{\"a\":1}"]]
-                        rows))))
+                 (done))
                (p/then (fn []
-                         (restore!)
-                         (done)))
+                         nil))
                (p/catch (fn [error]
-                          (restore!)
                           (is false (str error))
                           (done)))))))