From 03518a8da198d5ca2c19dd71fed8709c9dabf832 Mon Sep 17 00:00:00 2001
From: Andrey Antukh <niwi@niwi.nz>
Date: Fri, 1 Dec 2023 16:18:39 +0100
Subject: [PATCH] :sparkles: Add the ability to stream events on rpc methods

---
 .../app/templates/api-doc-entry.tmpl          |   6 ++
 backend/resources/app/templates/api-doc.tmpl  |  11 ++
 backend/src/app/features/components_v2.clj    |  12 ++-
 backend/src/app/http/errors.clj               |   4 +
 backend/src/app/http/sse.clj                  |  86 +++++++++++++++
 backend/src/app/rpc.clj                       |  18 ++--
 backend/src/app/rpc/commands/binfile.clj      | 101 +++++++++++-------
 backend/src/app/rpc/commands/management.clj   |  65 ++++++-----
 backend/src/app/rpc/doc.clj                   |   2 +
 backend/test/backend_tests/helpers.clj        |  32 +++++-
 .../backend_tests/rpc_management_test.clj     |   9 +-
 common/deps.edn                               |   2 +-
 docker/devenv/files/nginx.conf                |   2 +
 frontend/package.json                         |   1 +
 frontend/src/app/main/data/dashboard.cljs     |  16 ++-
 frontend/src/app/main/repo.cljs               |  32 ++++--
 .../src/app/main/ui/dashboard/projects.cljs   |   6 +-
 frontend/src/app/util/http.cljs               |  27 +++--
 frontend/src/app/util/sse.cljs                |  54 ++++++++++
 frontend/src/app/worker/import.cljs           |  20 ++--
 frontend/yarn.lock                            |   8 ++
 21 files changed, 408 insertions(+), 106 deletions(-)
 create mode 100644 backend/src/app/http/sse.clj
 create mode 100644 frontend/src/app/util/sse.cljs

diff --git a/backend/resources/app/templates/api-doc-entry.tmpl b/backend/resources/app/templates/api-doc-entry.tmpl
index c61157ec2..31c48deeb 100644
--- a/backend/resources/app/templates/api-doc-entry.tmpl
+++ b/backend/resources/app/templates/api-doc-entry.tmpl
@@ -25,6 +25,12 @@
           <span>SCHEMA</span>
         </span>
       {% endif %}
+
+      {% if item.sse %}
+        <span class="tag">
+          <span>SSE</span>
+        </span>
+      {% endif %}
     </div>
   </div>
   <div class="rpc-row-detail hidden">
diff --git a/backend/resources/app/templates/api-doc.tmpl b/backend/resources/app/templates/api-doc.tmpl
index 4a4082537..aa0b2a7d9 100644
--- a/backend/resources/app/templates/api-doc.tmpl
+++ b/backend/resources/app/templates/api-doc.tmpl
@@ -56,6 +56,17 @@
         desired content-type on the <b>`Accept`</b> header, the transit encoding is used
         by default.</p>
 
+        <h3>SSE (Server-Sent Events)</h3>
+        <p>The methods marked with <b>SSE</b> returns
+        a <a href="https://html.spec.whatwg.org/multipage/server-sent-events.html"> SSE
+        formatted</a> stream on the response body, always with status 200. The events are
+        always encoded using `application/transit+json` encoding (for now no content
+        negotiation is possible on methods that return SSE streams). </p>
+        <p>On the javascript side you can use
+        the <a href="https://github.com/rexxars/eventsource-parser">eventsoure-parser</a>
+        library for propertly parsing the response body using the
+        standard <a href="https://developer.mozilla.org/en-US/docs/Web/API/Fetch_API">Fetch
+        API</a></p>
 
         <h3>Limits</h3>
         <p>The rate limit work per user basis (this means that different api keys share
diff --git a/backend/src/app/features/components_v2.clj b/backend/src/app/features/components_v2.clj
index 3a41d818a..d1c828232 100644
--- a/backend/src/app/features/components_v2.clj
+++ b/backend/src/app/features/components_v2.clj
@@ -32,6 +32,7 @@
    [app.common.types.shape-tree :as ctst]
    [app.common.uuid :as uuid]
    [app.db :as db]
+   [app.http.sse :as sse]
    [app.media :as media]
    [app.rpc.commands.files :as files]
    [app.rpc.commands.files-snapshot :as fsnap]
@@ -362,6 +363,8 @@
   shapes from library components.  Mark the file with
   the :components-v2 option."
   [file-data libraries]
+  (sse/tap {:type :migration-progress
+            :section :components})
   (let [components (ctkl/components-seq file-data)]
     (if (empty? components)
       (assoc-in file-data [:options :components-v2] true)
@@ -435,6 +438,9 @@
             add-instance-grid
             (fn [fdata frame-id grid assets]
               (reduce (fn [result [component position]]
+                        (sse/tap {:type :migration-progress
+                                  :section :components
+                                  :name (:name component)})
                         (add-main-instance result component frame-id (gpt/add position
                                                                               (gpt/point grid-gap grid-gap))))
                       fdata
@@ -701,6 +707,9 @@
       (->> (d/zip media-group grid)
            (map (fn [[mobj position]]
                   (l/trc :hint "submit graphic processing" :file-id (str (:id fdata)) :id (str (:id mobj)))
+                  (sse/tap {:type :migration-progress
+                            :section :graphics
+                            :name (:name mobj)})
                   (px/submit! executor (partial process mobj position))))
            (reduce (fn [fdata promise]
                      (if-let [changes (deref promise)]
@@ -713,6 +722,8 @@
 
 (defn- migrate-graphics
   [fdata]
+  (sse/tap {:type :migration-progress
+            :section :graphics})
   (if (empty? (:media fdata))
     fdata
     (let [[fdata page-id start-pos]
@@ -812,7 +823,6 @@
 
 (defn migrate-file!
   [system file-id & {:keys [validate? throw-on-validate?]}]
-
   (let [tpoint  (dt/tpoint)
         file-id (if (string? file-id)
                   (parse-uuid file-id)
diff --git a/backend/src/app/http/errors.clj b/backend/src/app/http/errors.clj
index 93c67845b..0e8e065ad 100644
--- a/backend/src/app/http/errors.clj
+++ b/backend/src/app/http/errors.clj
@@ -232,3 +232,7 @@
   (if (ex/error? cause)
     (handle-error cause request nil)
     (handle-exception cause request nil)))
+
+(defn handle'
+  [cause request]
+  (::rres/body (handle cause request)))
diff --git a/backend/src/app/http/sse.clj b/backend/src/app/http/sse.clj
new file mode 100644
index 000000000..c89c91b61
--- /dev/null
+++ b/backend/src/app/http/sse.clj
@@ -0,0 +1,86 @@
+;; This Source Code Form is subject to the terms of the Mozilla Public
+;; License, v. 2.0. If a copy of the MPL was not distributed with this
+;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
+;;
+;; Copyright (c) KALEIDOS INC
+
+(ns app.http.sse
+  "SSE (server sent events) helpers"
+  (:refer-clojure :exclude [tap])
+  (:require
+   [app.common.data :as d]
+   [app.common.exceptions :as ex]
+   [app.common.logging :as l]
+   [app.common.transit :as t]
+   [app.http.errors :as errors]
+   [promesa.core :as p]
+   [promesa.exec :as px]
+   [promesa.exec.csp :as sp]
+   [promesa.util :as pu]
+   [ring.response :as rres])
+  (:import
+   java.io.OutputStream))
+
+(def ^:dynamic *channel* nil)
+
+(defn- write!
+  [^OutputStream output ^bytes  data]
+  (l/trc :hint "writting data" :data data :length (alength data))
+  (.write output data)
+  (.flush output))
+
+(defn- create-writer-loop
+  [^OutputStream output]
+  (try
+    (loop []
+      (when-let [event (sp/take! *channel*)]
+        (let [result (ex/try! (write! output event))]
+          (if (ex/exception? result)
+            (l/wrn :hint "unexpected exception on sse writer" :cause result)
+            (recur)))))
+    (finally
+      (pu/close! output))))
+
+(defn- encode
+  [[name data]]
+  (try
+    (let [data (with-out-str
+                 (println "event:" (d/name name))
+                 (println "data:" (t/encode-str data {:type :json-verbose}))
+                 (println))]
+      (.getBytes data "UTF-8"))
+    (catch Throwable cause
+      (l/err :hint "unexpected error on encoding value on sse stream"
+             :cause cause)
+      nil)))
+
+;; ---- PUBLIC API
+
+(def default-headers
+  {"Content-Type" "text/event-stream;charset=UTF-8"
+   "Cache-Control" "no-cache, no-store, max-age=0, must-revalidate"
+   "Pragma" "no-cache"})
+
+(defn tap
+  ([data] (tap "event" data))
+  ([name data]
+   (when-let [channel *channel*]
+     (sp/put! channel [name data])
+     nil)))
+
+(defn response
+  [handler & {:keys [buf] :or {buf 32} :as opts}]
+  (fn [request]
+    {::rres/headers default-headers
+     ::rres/status 200
+     ::rres/body (reify rres/StreamableResponseBody
+                   (-write-body-to-stream [_ _ output]
+                     (binding [*channel* (sp/chan :buf buf :xf (keep encode))]
+                       (let [writer (px/run! :virtual (partial create-writer-loop output))]
+                         (try
+                           (tap "end" (handler))
+                           (catch Throwable cause
+                             (tap "error" (errors/handle' cause request)))
+                           (finally
+                             (sp/close! *channel*)
+                             (p/await! writer)))))))}))
diff --git a/backend/src/app/rpc.clj b/backend/src/app/rpc.clj
index d4b3a4baa..69f9d84fb 100644
--- a/backend/src/app/rpc.clj
+++ b/backend/src/app/rpc.clj
@@ -57,14 +57,16 @@
 
 (defn- handle-response
   [request result]
-  (if (fn? result)
-    (result request)
-    (let [mdata (meta result)]
-      (-> {::rres/status  (::http/status mdata 200)
-           ::rres/headers (::http/headers mdata {})
-           ::rres/body    (rph/unwrap result)}
-          (handle-response-transformation request mdata)
-          (handle-before-comple-hook mdata)))))
+  (let [mdata    (meta result)
+        response (if (fn? result)
+                   (result request)
+                   (let [result (rph/unwrap result)]
+                     {::rres/status  (::http/status mdata 200)
+                      ::rres/headers (::http/headers mdata {})
+                      ::rres/body    result}))]
+    (-> response
+        (handle-response-transformation request mdata)
+        (handle-before-comple-hook mdata))))
 
 (defn- rpc-handler
   "Ring handler that dispatches cmd requests and convert between
diff --git a/backend/src/app/rpc/commands/binfile.clj b/backend/src/app/rpc/commands/binfile.clj
index ba00b26fa..a347c893f 100644
--- a/backend/src/app/rpc/commands/binfile.clj
+++ b/backend/src/app/rpc/commands/binfile.clj
@@ -15,6 +15,7 @@
    [app.common.files.validate :as fval]
    [app.common.fressian :as fres]
    [app.common.logging :as l]
+   [app.common.schema :as sm]
    [app.common.spec :as us]
    [app.common.types.file :as ctf]
    [app.common.uuid :as uuid]
@@ -22,6 +23,7 @@
    [app.db :as db]
    [app.features.components-v2 :as features.components-v2]
    [app.features.fdata :as features.fdata]
+   [app.http.sse :as sse]
    [app.loggers.audit :as-alias audit]
    [app.loggers.webhooks :as-alias webhooks]
    [app.media :as media]
@@ -30,7 +32,6 @@
    [app.rpc.commands.projects :as projects]
    [app.rpc.commands.teams :as teams]
    [app.rpc.doc :as-alias doc]
-   [app.rpc.helpers :as rph]
    [app.storage :as sto]
    [app.storage.tmp :as tmp]
    [app.tasks.file-gc]
@@ -38,11 +39,13 @@
    [app.util.pointer-map :as pmap]
    [app.util.services :as sv]
    [app.util.time :as dt]
+   [app.worker :as-alias wrk]
    [clojure.set :as set]
    [clojure.spec.alpha :as s]
    [clojure.walk :as walk]
    [cuerdas.core :as str]
    [datoteka.io :as io]
+   [promesa.core :as p]
    [promesa.util :as pu]
    [ring.response :as rres]
    [yetti.adapter :as yt])
@@ -642,6 +645,9 @@
             validate? (contains? cf/flags :file-validation)
             features  (cfeat/get-team-enabled-features cf/flags team)]
 
+        (sse/tap {:type :import-progress
+                  :section :read-import})
+
         ;; Process all sections
         (run! (fn [section]
                 (l/dbg :hint "reading section" :section section ::l/sync? true)
@@ -651,6 +657,8 @@
                                   (assoc ::section section)
                                   (assoc ::input input))]
                   (binding [*options* options]
+                    (sse/tap {:type :import-progress
+                              :section section})
                     (read-section options))))
               [:v1/metadata :v1/files :v1/rels :v1/sobjects])
 
@@ -1056,54 +1064,71 @@
 
 ;; --- Command: export-binfile
 
-(s/def ::file-id ::us/uuid)
-(s/def ::include-libraries? ::us/boolean)
-(s/def ::embed-assets? ::us/boolean)
-
-(s/def ::export-binfile
-  (s/keys :req [::rpc/profile-id]
-          :req-un [::file-id ::include-libraries? ::embed-assets?]))
+(def ^:private
+  schema:export-binfile
+  (sm/define
+    [:map {:title "export-binfile"}
+     [:file-id ::sm/uuid]
+     [:include-libraries? :boolean]
+     [:embed-assets? :boolean]]))
 
 (sv/defmethod ::export-binfile
   "Export a penpot file in a binary format."
   {::doc/added "1.15"
-   ::webhooks/event? true}
+   ::webhooks/event? true
+   ::sm/result schema:export-binfile}
   [{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id file-id include-libraries? embed-assets?] :as params}]
   (files/check-read-permissions! pool profile-id file-id)
-  (let [body (reify rres/StreamableResponseBody
-               (-write-body-to-stream [_ _ output-stream]
-                 (-> cfg
-                     (assoc ::file-ids [file-id])
-                     (assoc ::embed-assets? embed-assets?)
-                     (assoc ::include-libraries? include-libraries?)
-                     (export! output-stream))))]
+  (fn [_]
+    {::rres/status 200
+     ::rres/headers {"content-type" "application/octet-stream"}
+     ::rres/body (reify rres/StreamableResponseBody
+                   (-write-body-to-stream [_ _ output-stream]
+                     (-> cfg
+                         (assoc ::file-ids [file-id])
+                         (assoc ::embed-assets? embed-assets?)
+                         (assoc ::include-libraries? include-libraries?)
+                         (export! output-stream))))}))
 
-    (fn [_]
-      {::rres/status 200
-       ::rres/body body
-       ::rres/headers {"content-type" "application/octet-stream"}})))
 
-(s/def ::file ::media/upload)
-(s/def ::import-binfile
-  (s/keys :req [::rpc/profile-id]
-          :req-un [::project-id ::file]))
+;; --- Command: import-binfile
+
+(def ^:private
+  schema:import-binfile
+  (sm/define
+    [:map {:title "import-binfile"}
+     [:project-id ::sm/uuid]
+     [:file ::media/upload]]))
+
+(declare ^:private import-binfile)
 
 (sv/defmethod ::import-binfile
   "Import a penpot file in a binary format."
   {::doc/added "1.15"
-   ::webhooks/event? true}
+   ::webhooks/event? true
+   ::sse/stream? true
+   ::sm/params schema:import-binfile}
   [{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id project-id file] :as params}]
-  (db/with-atomic [conn pool]
-    (projects/check-read-permissions! conn profile-id project-id)
-    (let [ids (import! (assoc cfg
-                              ::input (:path file)
-                              ::project-id project-id
-                              ::profile-id profile-id
-                              ::ignore-index-errors? true))]
+  (projects/check-read-permissions! pool profile-id project-id)
+  (let [params (-> cfg
+                   (assoc ::input (:path file))
+                   (assoc ::project-id project-id)
+                   (assoc ::profile-id profile-id)
+                   (assoc ::ignore-index-errors? true))]
+    (with-meta
+      (sse/response #(import-binfile params))
+      {::audit/props {:file nil}})))
 
-      (db/update! conn :project
-                  {:modified-at (dt/now)}
-                  {:id project-id})
-
-      (rph/with-meta ids
-        {::audit/props {:file nil :file-ids ids}}))))
+(defn- import-binfile
+  [{:keys [::wrk/executor ::project-id] :as params}]
+  (db/tx-run! params
+              (fn [{:keys [::db/conn] :as params}]
+                ;; NOTE: the importation process performs some operations that
+                ;; are not very friendly with virtual threads, and for avoid
+                ;; unexpected blocking of other concurrent operations we
+                ;; dispatch that operation to a dedicated executor.
+                (let [result (p/thread-call executor (partial import! params))]
+                  (db/update! conn :project
+                              {:modified-at (dt/now)}
+                              {:id project-id})
+                  (deref result)))))
diff --git a/backend/src/app/rpc/commands/management.clj b/backend/src/app/rpc/commands/management.clj
index 1c0090e69..00195d17e 100644
--- a/backend/src/app/rpc/commands/management.clj
+++ b/backend/src/app/rpc/commands/management.clj
@@ -14,6 +14,7 @@
    [app.common.schema :as sm]
    [app.common.uuid :as uuid]
    [app.db :as db]
+   [app.http.sse :as sse]
    [app.loggers.webhooks :as-alias webhooks]
    [app.rpc :as-alias rpc]
    [app.rpc.commands.binfile :as binfile]
@@ -27,7 +28,9 @@
    [app.util.pointer-map :as pmap]
    [app.util.services :as sv]
    [app.util.time :as dt]
+   [app.worker :as-alias wrk]
    [clojure.walk :as walk]
+   [promesa.core :as p]
    [promesa.exec :as px]))
 
 ;; --- COMMAND: Duplicate File
@@ -405,29 +408,6 @@
 
 ;; --- COMMAND: Clone Template
 
-(defn- clone-template!
-  [{:keys [::db/conn] :as cfg} {:keys [profile-id template-id project-id]}]
-  (let [template (tmpl/get-template-stream cfg template-id)
-        project  (db/get-by-id conn :project project-id {:columns [:id :team-id]})]
-
-    (when-not template
-      (ex/raise :type :not-found
-                :code :template-not-found
-                :hint "template not found"))
-
-    (teams/check-edition-permissions! conn profile-id (:team-id project))
-
-    (-> cfg
-        ;; FIXME: maybe reuse the conn instead of creating more
-        ;; connections in the import process?
-        (dissoc ::db/conn)
-        (assoc ::binfile/input template)
-        (assoc ::binfile/project-id (:id project))
-        (assoc ::binfile/profile-id profile-id)
-        (assoc ::binfile/ignore-index-errors? true)
-        (assoc ::binfile/migrate? true)
-        (binfile/import!))))
-
 (def ^:private
   schema:clone-template
   (sm/define
@@ -435,15 +415,46 @@
      [:project-id ::sm/uuid]
      [:template-id ::sm/word-string]]))
 
+(declare ^:private clone-template)
+
 (sv/defmethod ::clone-template
   "Clone into the specified project the template by its id."
   {::doc/added "1.16"
+   ::sse/stream? true
    ::webhooks/event? true
    ::sm/params schema:clone-template}
-  [{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id] :as params}]
-  (db/with-atomic [conn pool]
-    (-> (assoc cfg ::db/conn conn)
-        (clone-template! (assoc params :profile-id profile-id)))))
+  [{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id project-id template-id] :as params}]
+  (let [project   (db/get-by-id pool :project project-id {:columns [:id :team-id]})
+        _         (teams/check-edition-permissions! pool profile-id (:team-id project))
+        template  (tmpl/get-template-stream cfg template-id)
+        params    (-> cfg
+                      (assoc ::binfile/input template)
+                      (assoc ::binfile/project-id (:id project))
+                      (assoc ::binfile/profile-id profile-id)
+                      (assoc ::binfile/ignore-index-errors? true)
+                      (assoc ::binfile/migrate? true))]
+
+    (when-not template
+      (ex/raise :type :not-found
+                :code :template-not-found
+                :hint "template not found"))
+
+    (sse/response #(clone-template params))))
+
+(defn- clone-template
+  [{:keys [::wrk/executor ::binfile/project-id] :as params}]
+  (db/tx-run! params
+              (fn [{:keys [::db/conn] :as params}]
+                ;; NOTE: the importation process performs some operations that
+                ;; are not very friendly with virtual threads, and for avoid
+                ;; unexpected blocking of other concurrent operations we
+                ;; dispatch that operation to a dedicated executor.
+                (let [result (p/thread-call executor (partial binfile/import! params))]
+                  (db/update! conn :project
+                              {:modified-at (dt/now)}
+                              {:id project-id})
+
+                  (deref result)))))
 
 ;; --- COMMAND: Get list of builtin templates
 
diff --git a/backend/src/app/rpc/doc.clj b/backend/src/app/rpc/doc.clj
index bcfcd96c8..185f3fc4c 100644
--- a/backend/src/app/rpc/doc.clj
+++ b/backend/src/app/rpc/doc.clj
@@ -16,6 +16,7 @@
    [app.common.schema.openapi :as oapi]
    [app.common.schema.registry :as sr]
    [app.config :as cf]
+   [app.http.sse :as-alias sse]
    [app.loggers.webhooks :as-alias webhooks]
    [app.rpc :as-alias rpc]
    [app.util.json :as json]
@@ -55,6 +56,7 @@
              :module (or (some-> (::module mdata) d/name)
                          (-> (:ns mdata) (str/split ".") last))
              :auth (::rpc/auth mdata true)
+             :sse (::sse/stream? mdata false)
              :webhook (::webhooks/event? mdata false)
              :docs (::sv/docstring mdata)
              :deprecated (::deprecated mdata)
diff --git a/backend/test/backend_tests/helpers.clj b/backend/test/backend_tests/helpers.clj
index 6b8ef788a..3fad16114 100644
--- a/backend/test/backend_tests/helpers.clj
+++ b/backend/test/backend_tests/helpers.clj
@@ -15,12 +15,13 @@
    [app.common.pprint :as pp]
    [app.common.schema :as sm]
    [app.common.spec :as us]
+   [app.common.transit :as tr]
    [app.common.uuid :as uuid]
    [app.config :as cf]
    [app.db :as db]
    [app.main :as main]
-   [app.media]
    [app.media :as-alias mtx]
+   [app.media]
    [app.migrations]
    [app.msgbus :as-alias mbus]
    [app.rpc :as-alias rpc]
@@ -43,8 +44,12 @@
    [integrant.core :as ig]
    [mockery.core :as mk]
    [promesa.core :as p]
+   [promesa.exec :as px]
+   [ring.response :as rres]
    [yetti.request :as yrq])
   (:import
+   java.io.PipedInputStream
+   java.io.PipedOutputStream
    java.util.UUID
    org.postgresql.ds.PGSimpleDataSource))
 
@@ -553,3 +558,28 @@
                  (assoc :return-list [])
                  (assoc :call-args nil)
                  (assoc :call-args-list [])))))
+
+(defn- slurp'
+  [input & opts]
+  (let [sw (java.io.StringWriter.)]
+    (with-open [^java.io.Reader r (java.io.InputStreamReader. input "UTF-8")]
+      (io/copy r sw)
+      (.toString sw))))
+
+(defn consume-sse
+  [callback]
+  (let [{:keys [::rres/status ::rres/body ::rres/headers] :as response} (callback {})
+        output (PipedOutputStream.)
+        input  (PipedInputStream. output)]
+
+    (try
+      (px/exec! :virtual #(rres/-write-body-to-stream body nil output))
+      (into []
+            (map (fn [event]
+                   (let [[item1 item2] (re-seq #"(.*): (.*)\n?" event)]
+                     [(keyword (nth item1 2))
+                      (tr/decode-str (nth item2 2))])))
+            (-> (slurp' input)
+                (str/split "\n\n")))
+      (finally
+        (.close input)))))
diff --git a/backend/test/backend_tests/rpc_management_test.clj b/backend/test/backend_tests/rpc_management_test.clj
index e56109d33..2b5f182e8 100644
--- a/backend/test/backend_tests/rpc_management_test.clj
+++ b/backend/test/backend_tests/rpc_management_test.clj
@@ -6,6 +6,7 @@
 
 (ns backend-tests.rpc-management-test
   (:require
+   [app.common.pprint :as pp]
    [app.common.uuid :as uuid]
    [app.db :as db]
    [app.http :as http]
@@ -604,9 +605,11 @@
 
     (t/is (nil? (:error out)))
     (let [result (:result out)]
-      (t/is (set? result))
-      (t/is (uuid? (first result)))
-      (t/is (= 1 (count result))))))
+      (t/is (fn? result))
+
+      (let [events (th/consume-sse result)]
+        (t/is (= 8 (count events)))
+        (t/is (= :end (first (last events))))))))
 
 (t/deftest get-list-of-buitin-templates
   (let [prof (th/create-profile* 1 {:is-active true})
diff --git a/common/deps.edn b/common/deps.edn
index 4bb93a5a8..4dcd1987d 100644
--- a/common/deps.edn
+++ b/common/deps.edn
@@ -32,7 +32,7 @@
 
   funcool/tubax {:mvn/version "2021.05.20-0"}
   funcool/cuerdas {:mvn/version "2022.06.16-403"}
-  funcool/promesa {:git/sha "658c429c56c11c33da7594fa2ef53f4e6afedac4"
+  funcool/promesa {:git/sha "484b7f5c0d08d817746caa685ed9ac5583eb37fa"
                    :git/url "https://github.com/funcool/promesa"}
 
   funcool/datoteka {:mvn/version "3.0.66"
diff --git a/docker/devenv/files/nginx.conf b/docker/devenv/files/nginx.conf
index 175f7f2cf..ab2c5ea6e 100644
--- a/docker/devenv/files/nginx.conf
+++ b/docker/devenv/files/nginx.conf
@@ -105,6 +105,8 @@ http {
 
         location /api {
             proxy_pass http://127.0.0.1:6060/api;
+            proxy_buffering off;
+            proxy_http_version 1.1;
         }
 
         location /admin {
diff --git a/frontend/package.json b/frontend/package.json
index d2ac50f09..e33d2a692 100644
--- a/frontend/package.json
+++ b/frontend/package.json
@@ -78,6 +78,7 @@
   "dependencies": {
     "date-fns": "^2.30.0",
     "draft-js": "^0.11.7",
+    "eventsource-parser": "^1.1.1",
     "highlight.js": "^11.8.0",
     "js-beautify": "^1.14.9",
     "jszip": "^3.10.1",
diff --git a/frontend/src/app/main/data/dashboard.cljs b/frontend/src/app/main/data/dashboard.cljs
index fbe417428..9e723ca20 100644
--- a/frontend/src/app/main/data/dashboard.cljs
+++ b/frontend/src/app/main/data/dashboard.cljs
@@ -10,6 +10,7 @@
    [app.common.data.macros :as dm]
    [app.common.features :as cfeat]
    [app.common.files.helpers :as cfh]
+   [app.common.logging :as log]
    [app.common.schema :as sm]
    [app.common.uri :as u]
    [app.common.uuid :as uuid]
@@ -25,6 +26,7 @@
    [app.util.dom :as dom]
    [app.util.i18n :as i18n :refer [tr]]
    [app.util.router :as rt]
+   [app.util.sse :as sse]
    [app.util.time :as dt]
    [app.util.timers :as tm]
    [app.util.webapi :as wapi]
@@ -32,6 +34,8 @@
    [clojure.set :as set]
    [potok.core :as ptk]))
 
+(log/set-level! :warn)
+
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Initialization
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
@@ -949,7 +953,17 @@
       (let [{:keys [on-success on-error]
              :or {on-success identity
                   on-error rx/throw}} (meta params)]
-        (->> (rp/cmd! :clone-template {:project-id project-id :template-id template-id})
+        (->> (rp/cmd! ::sse/clone-template {:project-id project-id
+                                            :template-id template-id})
+             (rx/tap (fn [event]
+                       (let [payload (sse/get-payload event)
+                             type    (sse/get-type event)]
+                         (if (= type "event")
+                           (log/dbg :hint "clone-template: progress" :section (:section payload) :name (:name payload))
+                           (log/dbg :hint "clone-template: end")))))
+
+             (rx/filter sse/end-of-stream?)
+             (rx/map sse/get-payload)
              (rx/tap on-success)
              (rx/catch on-error))))))
 
diff --git a/frontend/src/app/main/repo.cljs b/frontend/src/app/main/repo.cljs
index e8b5f8f01..e99f9a461 100644
--- a/frontend/src/app/main/repo.cljs
+++ b/frontend/src/app/main/repo.cljs
@@ -7,9 +7,11 @@
 (ns app.main.repo
   (:require
    [app.common.data :as d]
+   [app.common.transit :as t]
    [app.common.uri :as u]
    [app.config :as cf]
    [app.util.http :as http]
+   [app.util.sse :as sse]
    [beicon.core :as rx]
    [cuerdas.core :as str]))
 
@@ -56,8 +58,14 @@
    {:query-params [:file-id :revn]
     :form-data? true}
 
+   ::sse/clone-template
+   {:response-type ::sse/stream}
+
+   ::sse/import-binfile
+   {:response-type ::sse/stream
+    :form-data? true}
+
    :export-binfile {:response-type :blob}
-   :import-binfile {:form-data? true}
    :retrieve-list-of-builtin-templates {:query-params :all}
    })
 
@@ -85,9 +93,9 @@
                     :else :post)
 
         request   {:method method
-                   :uri (u/join cf/public-uri "api/rpc/command/" (name id))
+                   :uri (u/join cf/public-uri "api/rpc/command/" nid)
                    :credentials "include"
-                   :headers {"accept" "application/transit+json"}
+                   :headers {"accept" "application/transit+json,text/event-stream,*/*"}
                    :body (when (= method :post)
                            (if form-data?
                              (http/form-data params)
@@ -97,11 +105,21 @@
                             (if query-params
                               (select-keys params query-params)
                               nil))
-                   :response-type (or response-type :text)}]
 
-    (->> (http/send! request)
-         (rx/map decode-fn)
-         (rx/mapcat handle-response))))
+                   :response-type
+                   (if (= response-type ::sse/stream)
+                     :stream
+                     (or response-type :text))}
+
+        result    (->> (http/send! request)
+                       (rx/map decode-fn)
+                       (rx/mapcat handle-response))]
+
+    (cond->> result
+      (= ::sse/stream response-type)
+      (rx/mapcat (fn [body]
+                   (-> (sse/create-stream body)
+                       (sse/read-stream t/decode-str)))))))
 
 (defmulti cmd! (fn [id _] id))
 
diff --git a/frontend/src/app/main/ui/dashboard/projects.cljs b/frontend/src/app/main/ui/dashboard/projects.cljs
index b017c2020..545d9c0aa 100644
--- a/frontend/src/app/main/ui/dashboard/projects.cljs
+++ b/frontend/src/app/main/ui/dashboard/projects.cljs
@@ -147,8 +147,10 @@
         (mf/use-fn
          (mf/deps template default-project-id)
          (fn []
-           (let [mdata  {:on-success on-template-cloned-success :on-error on-template-cloned-error}
-                 params {:project-id default-project-id :template-id (:id template)}]
+           (let [mdata  {:on-success on-template-cloned-success
+                         :on-error on-template-cloned-error}
+                 params {:project-id default-project-id
+                         :template-id (:id template)}]
              (swap! state #(assoc % :status :importing))
              (st/emit! (with-meta (dd/clone-template (with-meta params mdata))
                          {::ev/origin "get-started-hero-block"})))))]
diff --git a/frontend/src/app/util/http.cljs b/frontend/src/app/util/http.cljs
index 60fc675f6..35c8ae485 100644
--- a/frontend/src/app/util/http.cljs
+++ b/frontend/src/app/util/http.cljs
@@ -105,17 +105,22 @@
 
 (defn send!
   [{:keys [response-type] :or {response-type :text} :as params}]
-  (letfn [(on-response [response]
-            (let [body (case response-type
-                         :json (.json ^js response)
-                         :text (.text ^js response)
-                         :blob (.blob ^js response))]
-              (->> (rx/from body)
-                   (rx/map (fn [body]
-                             {::response response
-                              :status    (.-status ^js response)
-                              :headers   (parse-headers (.-headers ^js response))
-                              :body      body})))))]
+  (letfn [(on-response [^js response]
+            (if (= :stream response-type)
+              (rx/of {:status (.-status response)
+                      :headers (parse-headers (.-headers response))
+                      :body (.-body response)
+                      ::response response})
+              (let [body (case response-type
+                           :json   (.json ^js response)
+                           :text   (.text ^js response)
+                           :blob   (.blob ^js response))]
+                (->> (rx/from body)
+                     (rx/map (fn [body]
+                               {::response response
+                                :status    (.-status ^js response)
+                                :headers   (parse-headers (.-headers ^js response))
+                                :body      body}))))))]
     (->> (fetch params)
          (rx/mapcat on-response))))
 
diff --git a/frontend/src/app/util/sse.cljs b/frontend/src/app/util/sse.cljs
new file mode 100644
index 000000000..222e1bda7
--- /dev/null
+++ b/frontend/src/app/util/sse.cljs
@@ -0,0 +1,54 @@
+;; This Source Code Form is subject to the terms of the Mozilla Public
+;; License, v. 2.0. If a copy of the MPL was not distributed with this
+;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
+;;
+;; Copyright (c) KALEIDOS INC
+
+(ns app.util.sse
+  (:require
+   ["eventsource-parser/stream" :as sse]
+   [beicon.core :as rx]))
+
+(defn create-stream
+  [^js/ReadableStream stream]
+  (.. stream
+      (pipeThrough (js/TextDecoderStream.))
+      (pipeThrough (sse/EventSourceParserStream.))))
+
+(defn read-stream
+  [^js/ReadableStream stream decode-fn]
+  (letfn [(read-items [^js reader]
+            (->> (rx/from (.read reader))
+                 (rx/mapcat (fn [result]
+                              (if (.-done result)
+                                (rx/empty)
+                                (rx/concat
+                                 (rx/of (.-value result))
+                                 (read-items reader)))))))]
+    (->> (read-items (.getReader stream))
+         (rx/mapcat (fn [^js event]
+                      (let [type (.-event event)
+                            data (.-data event)
+                            data (decode-fn data)]
+                        (if (= "error" type)
+                          (rx/throw (ex-info "stream exception" data))
+                          (rx/of #js {:type type :data data}))))))))
+
+(defn get-type
+  [event]
+  (unchecked-get event "type"))
+
+(defn get-payload
+  [event]
+  (unchecked-get event "data"))
+
+(defn end-of-stream?
+  [event]
+  (= "end" (get-type event)))
+
+(defn event?
+  [event]
+  (= "event" (get-type event)))
+
+
+
diff --git a/frontend/src/app/worker/import.cljs b/frontend/src/app/worker/import.cljs
index 9895aaf47..bccb975ea 100644
--- a/frontend/src/app/worker/import.cljs
+++ b/frontend/src/app/worker/import.cljs
@@ -22,6 +22,7 @@
    [app.util.i18n :as i18n :refer [tr]]
    [app.util.import.parser :as cip]
    [app.util.json :as json]
+   [app.util.sse :as sse]
    [app.util.webapi :as wapi]
    [app.util.zip :as uz]
    [app.worker.impl :as impl]
@@ -329,7 +330,7 @@
                                   (map #(assoc % :type :fill)))
           stroke-images-data (->> (cip/get-stroke-images-data node)
                                   (map #(assoc % :type :stroke)))
-          
+
           images-data        (concat
                                fill-images-data
                                stroke-images-data
@@ -709,15 +710,22 @@
                     :response-type :blob
                     :method :get})
                   (rx/map :body)
-                  (rx/mapcat #(rp/cmd! :import-binfile {:file % :project-id project-id}))
-                  (rx/map (fn [_]
-                            {:status :import-finish
-                             :file-id (:file-id data)}))
+                  (rx/mapcat (fn [file]
+                               (->> (rp/cmd! ::sse/import-binfile {:file file :project-id project-id})
+                                    (rx/tap (fn [event]
+                                              (let [payload (sse/get-payload event)
+                                                    type    (sse/get-type event)]
+                                                (if (= type "event")
+                                                  (log/dbg :hint "import-binfile: progress" :section (:section payload) :name (:name payload))
+                                                  (log/dbg :hint "import-binfile: end")))))
+                                    (rx/filter sse/end-of-stream?)
+                                    (rx/map (fn [_]
+                                              {:status :import-finish
+                                               :file-id (:file-id data)})))))
                   (rx/catch (fn [cause]
                               (log/error :hint "unexpected error on import process"
                                          :project-id project-id
                                          ::log/sync? true)
-                              ;; TODO: consider do thi son logging directly ?
 
                               (when (map? cause)
                                 (println "Error data:")
diff --git a/frontend/yarn.lock b/frontend/yarn.lock
index fc9af1bd1..0cd007f2f 100644
--- a/frontend/yarn.lock
+++ b/frontend/yarn.lock
@@ -7028,6 +7028,13 @@ __metadata:
   languageName: node
   linkType: hard
 
+"eventsource-parser@npm:^1.1.1":
+  version: 1.1.1
+  resolution: "eventsource-parser@npm:1.1.1"
+  checksum: 6eae5e8300dc5d4dcd29d09d037a43b9954077fbcc936904065d1abf5ec503f1fa56a942ad4f796b7dcc3bf64224440917ea635ef4a75f5522e9951df4b9aadf
+  languageName: node
+  linkType: hard
+
 "evp_bytestokey@npm:^1.0.0, evp_bytestokey@npm:^1.0.3":
   version: 1.0.3
   resolution: "evp_bytestokey@npm:1.0.3"
@@ -7578,6 +7585,7 @@ __metadata:
     concurrently: "npm:^8.2.2"
     date-fns: "npm:^2.30.0"
     draft-js: "npm:^0.11.7"
+    eventsource-parser: "npm:^1.1.1"
     gettext-parser: "npm:^7.0.1"
     gulp: "npm:4.0.2"
     gulp-cached: "npm:^1.1.1"