Improvements to asset pulling (internal)

package.json

@@ -31,6 +31,7 @@
     "inquirer": "^6.2.2",
     "mkdirp": "^0.5.1",
     "open": "6.4.0",
-    "rimraf": "^3.0.0"
+    "rimraf": "^3.0.0",
+    "yauzl": "3.2.0"
   }
 }

src/sharetribe/flex_cli/api/client.cljs

@@ -1,12 +1,15 @@
 (ns sharetribe.flex-cli.api.client
-  (:require [cognitect.transit :as t]
-            [clojure.core.async :as async :refer [put! chan <! go]]
+  (:require ["stream" :refer [Readable]]
+            [util]
+            [cognitect.transit :as t]
+            [clojure.core.async :as async :refer [go]]
             [chalk]
             [goog.object]
             [sharetribe.flex-cli.config :as config]
             [sharetribe.flex-cli.cli-info :as cli-info]
             [sharetribe.flex-cli.exception :as exception]
-            [sharetribe.flex-cli.view :as v]))
+            [sharetribe.flex-cli.view :as v]
+            [sharetribe.flex-cli.async-util :refer [->chan <? go-try]]))
 
 (def user-agent
   "User agent string with version and platform information
@@ -38,7 +41,7 @@
 (defn default-error-format [data]
   (let [{:keys [req res]} data
         {:keys [path]} req
-        {:keys [status response]} res
+        {:keys [status response original-text]} res
         marketplace (-> req :query :marketplace)
         api-key-suffix (->> req
                             :client
@@ -66,7 +69,12 @@
        [:span "Use " (bold (str cli-info/bin " login")) " to relogin if needed." :line])
 
       :else
-      (error-page [:span "API call failed. Status: " (str status) ", reason: " (or (-> response :errors first :title) "Unspecified") :line]))))
+      (error-page (cond->
+                      [:span "API call failed. Status: " (str status)
+                       ", reason: " (or (-> response :errors first :title) original-text "Unspecified")]
+                    ;; Enable for debugging
+                    #_#_(-> response :errors first :details) (conj (str ", details: " (-> response :errors first :details)))
+                    true (conj :line))))))
 
 (defmethod exception/format-exception :api/error [_ _ data]
   (default-error-format data))
@@ -111,84 +119,105 @@
       (.set (.-searchParams url) (name k) (to-query-param v)))
     url))
 
-(defn read-transit [s]
+(defn- content-type [response]
   (try
-    {:success true
-     :data (t/read transit-reader s)}
+    (util/MIMEType. (-> response .-headers (.get "content-type")))
     (catch js/Error _e
-      {:success false})))
-
-(defn- handle-response [out-chan response req]
-  (->
-   (.text response)
-   (.then
-    (fn [body-str]
-      (let [parsed (read-transit body-str)]
-        (put! out-chan
-              (if (.-ok response)
-                (:data parsed)
-                (let [parsed (read-transit body-str)]
-                  (handle-error req
-                                (cond-> {:status (.-status response)
-                                         :status-text (.-statusText response)}
-                                  (:success parsed) (assoc :response (:data parsed))
-                                  (not (:success parsed)) (assoc :original-text body-str)))))))))))
-
-(defn do-get [client path query]
-  (let [c (chan)]
-    (-> (js/fetch
-         (construct-url (str (config/value :api-base-url) path) query)
-         (clj->js
-          {:method "GET"
-           :headers {"Authorization" (str "Apikey " (::api-key client))
-                     "User-Agent" user-agent
-                     "Accept" "application/transit+json"}}))
-        (.then (fn [response]
-                 (handle-response c response {:client client
-                                              :path path
-                                              :query query})))
-        ;; Unknown failure
-        (.catch (fn [error] (put! c (exception/exception :client/api-request-failed error)))))
-    c))
+      nil)))
+
+(defn- parse-text [response]
+  (go
+    (try
+      {:success true
+       :type "text/plain"
+       :content (<? (->chan (.text response)))}
+      (catch js/Error _e
+        {:success false}))))
+
+(defn- parse-transit [response]
+  (go
+    (try
+      {:success true
+       :type "application/transit+json"
+       :content (t/read transit-reader (<? (.text response)))}
+      (catch js/Error _e
+        {:success false}))))
+
+(defn- parse-zip [response]
+  (go
+    {:success true
+     :content (.fromWeb Readable (.-body response))}))
+
+(defn- parse-body
+  "Takes response and parses the body based on response content type header.
+
+  Returns a channel with map of:
+  - success (boolean)
+  - content (parsed content, only if success)
+  "
+  [response]
+  (let [essence (when-let [ct ^js (content-type response)]
+                  (.-essence ct))]
+    (case essence
+      ("text/plain" nil) (parse-text response) 
+      "application/transit+json" (parse-transit response)
+      "application/zip" (parse-zip response))))
+
+(defn- handle-response [accept response req]
+  (go-try
+    (let [{:keys [success content type]} (<? (parse-body response))
+          ;; Check if parsed content type matches with what we expected
+          accepted? (= accept type)]
+      (if (.-ok response)
+        ;; If response was ok, we assume that parsing was
+        ;; successful and returned content type was what we
+        ;; expected.
+        content
+        (handle-error
+         req
+         (cond-> {:status (.-status response)
+                  :status-text (.-statusText response)}
+
+           ;; Assoc :response only if parsing was
+           ;; successful and response content type
+           ;; was what we expected. If we for
+           ;; example use Accept:
+           ;; application/transit+json, but get
+           ;; back 500 with text/plain, we don't
+           ;; assoc it to response
+           (and accepted? success) (assoc :response content)
+
+           (= "text/plain" type)
+           (assoc :original-text content)))))))
+
+(defn- do-request [client path method query body opts]
+  (go-try
+     (let [accept (or (::accept opts) "application/transit+json")
+           res (<? (->chan (js/fetch
+                            (construct-url (str (config/value :api-base-url) path) query)
+                            (clj->js
+                             (cond->
+                                 {:method method
+                                  :headers (cond-> {"Authorization" (str "Apikey " (::api-key client))
+                                                    "User-Agent" user-agent
+                                                    "Accept" accept}
+                                             (::content-type opts) (assoc "Content-Type" (::content-type opts)))}
+                               body (assoc :body body))))))]
+
+       (<? (handle-response accept res {:client client
+                                        :path path
+                                        :query query})))))
+
+(defn do-get
+  ([client path query] (do-get client path query nil))
+  ([client path query opts]
+   (do-request client path "GET" query nil opts)))
 
 (defn do-post [client path query body]
-  (let [c (chan)]
-    (-> (js/fetch
-         (construct-url (str (config/value :api-base-url) path) query)
-         (clj->js
-          {:method "POST"
-           :body (t/write transit-writer body)
-           :headers {"Authorization" (str "Apikey " (::api-key client))
-                     "Content-Type" "application/transit+json"
-                     "User-Agent" user-agent
-                     "Accept" "application/transit+json"}}))
-        (.then (fn [response]
-                 (handle-response c response {:client client
-                                              :path path
-                                              :query query})))
-
-        ;; Unknown failure
-        (.catch (fn [error] (put! c (exception/exception :client/api-request-failed error)))))
-    c))
+  (do-request client path "POST" query (t/write transit-writer body) {::content-type "application/transit+json"}))
 
 (defn do-multipart-post [client path query form-data]
-  (let [c (chan)]
-    (-> (js/fetch
-         (construct-url (str (config/value :api-base-url) path) query)
-         (clj->js
-          {:method "POST"
-           :body form-data
-           :headers {"Authorization" (str "Apikey " (::api-key client))
-                     "User-Agent" user-agent
-                     "Accept" "application/transit+json"}}))
-        (.then (fn [response]
-                 (handle-response c response {:client client
-                                              :path path
-                                              :query query})))
-
-        ;; Unknown failure
-        (.catch (fn [error] (put! c (exception/exception :client/api-request-failed error)))))
-    c))
+  (do-request client path "POST" query form-data nil))
 
 (defn new-client [api-key]
   {::api-key api-key})

src/sharetribe/flex_cli/async_util.cljs

@@ -25,3 +25,12 @@
   (if (read-port? x)
     x
     (go x)))
+
+(defn ->chan [p]
+  (let [c (async/promise-chan)]
+    (-> p
+        (.then (fn [v] (if (some? v)
+                         (async/put! c v)
+                         (async/close! c))))
+        (.catch (fn [err] (async/put! c err))))
+    c))

src/sharetribe/flex_cli/commands/assets.cljs

@@ -1,6 +1,6 @@
 (ns sharetribe.flex-cli.commands.assets
   "Commands for managing assets."
-  (:require [clojure.core.async :as async :refer [go <!]]
+  (:require [clojure.core.async :as async :refer [go <! go-loop]]
             [clojure.set :as set]
             [clojure.string :as str]
             [chalk]
@@ -72,6 +72,49 @@
      :i 0}
     assets)))
 
+(defn create-temp-zipfile-path []
+  (io-util/tmp-file-path (str "assets-" (js/Date.now) ".zip")))
+
+(def metadata-filename "meta/asset-meta.edn")
+(def assets-dir "assets/")
+
+(defn remove-assets-dir [filename]
+  (subs filename (count assets-dir)))
+
+(defn try-unlink
+  "Try unlink filename. Ignore errors."
+  [filename]
+  (try
+    (io-util/unlink filename)
+    (catch js/Error _e
+      nil)))
+
+(def carriage-return
+  "Moves the cursor back to the beginning of the line. This way we can 'update in place' the text"
+  "\r")
+
+(def clear-line
+  "Clear line ANSI control. Empties the rest of the line. This way we can update
+  in place even if the previously written text was shorter than current."
+  "\033[K"
+  )
+
+(defn print-progress! [^js stream]
+  (let [downloaded (atom 0)
+        carriage-return "\r"
+        print-progress (fn []
+                         (let [mb (/ @downloaded 1024 1024)]
+                           (js/process.stderr.write (str carriage-return clear-line "Downloaded " (.toFixed mb 2) "MB"))))
+        print-interval (js/setInterval print-progress 100)]
+
+    (.on stream "data" (fn [^js chunk]
+                         (swap! downloaded + (.-length chunk))))
+    (.on stream "end" (fn []
+                        (js/clearInterval print-interval)
+                        ;; One final print
+                        (print-progress)
+                        (js/process.stderr.write "\nFinished downloading assets\n")))))
+
 (defn pull-assets [params ctx]
   (go-try
    (let [{:keys [api-client marketplace]} ctx
@@ -90,31 +133,50 @@
                        version-params
                        {:marketplace marketplace})
 
-         res (try
-               (<? (do-get api-client "/assets/pull" query-params))
-               (catch js/Error e
-                 (throw e)))
+         local-assets (when prune (io-util/list-assets path))
+
+         temp-path (create-temp-zipfile-path)
+
+         new-asset-meta
+         (let [result-stream (<? (do-get api-client "/assets/pull" query-params {::api.client/accept "application/zip"}))
+               _ (print-progress! result-stream)
+               _ (<? (io-util/pipe-stream-to-file result-stream temp-path))
+               unzipped-entries-c (io-util/unzip-entries temp-path)]
+             (<? 
+              (go-loop [asset-meta nil]
+                (if-let [{:keys [read-stream filename next]} (<? unzipped-entries-c)]
+                  (let [metadata-file? (= metadata-filename filename)
+                        meta* (if metadata-file?
+                                ;; Read new metadata
+                                (<? (io-util/parse-edn-stream read-stream))
+                                asset-meta)]
+
+                    (when-not metadata-file?
+                      ;; Write to disk only if not a metadata file
+                      (<? (io-util/pipe-stream-to-file read-stream (io-util/join path (remove-assets-dir filename)))))
+
+                    (next)
+                    (recur meta*))
+
+                  asset-meta))))
+
+         _ (try-unlink temp-path)
 
          new-version (or
-                      (-> res :meta :version)
-                      (-> res :meta :aliased-version))
-         local-assets (when prune (io-util/list-assets path))
+                      (:version new-asset-meta)
+                      (:aliased-version new-asset-meta))
          deleted-paths (when prune
                          (set/difference (into #{} (map :path local-assets))
-                                         (into #{} (map :path (:data res)))))
+                                         (into #{} (map :path (:assets new-asset-meta)))))
          updated? (not= old-version new-version)]
 
      (if (or updated? (seq deleted-paths))
        (do
-         (when updated?
-           (io-util/write-assets path (:data res)))
          (when (seq deleted-paths)
            (io-util/remove-assets path deleted-paths))
          (io-util/write-asset-meta path (assoc asset-meta
                                                :version new-version
-                                               :assets (into []
-                                                             (map #(dissoc % :data-raw))
-                                                             (:data res))))
+                                               :assets (:assets new-asset-meta)))
          (io-util/ppd [:span
                        "Version " new-version
                        " successfully pulled."]))
@@ -180,13 +242,12 @@
        (let [query-params {:marketplace marketplace}
              body-params (to-multipart-form-data
                           {:current-version (if version version "nil") ;; stringify nil as initial version
-                           :assets (concat changed-assets delete-assets)})
-
+                           :assets (concat changed-assets delete-assets)})             
              res (try
                    (<? (do-multipart-post api-client "/assets/push" query-params body-params))
                    (catch js/Error e
                      (throw e)))
-
+             
              new-version (-> res :data :version)]
 
          (if new-version