✨ Add initial impl for migrate-components-v2 manage.py command

backend/scripts/manage.py

@@ -44,11 +44,16 @@ def send_eval(expr):
         s.send(b":repl/quit\n\n")
 
         with s.makefile() as f:
-            result = json.load(f)
-            tag = result.get("tag", None)
-            if tag != "ret":
-                raise RuntimeError("unexpected response from PREPL")
-            return result.get("val", None), result.get("exception", None)
+            while True:
+                line = f.readline()
+                result = json.loads(line)
+                tag = result.get("tag", None)
+                if tag == "ret":
+                    return result.get("val", None), result.get("exception", None)
+                elif tag == "out":
+                    print(result.get("val"), end="")
+                else:
+                    raise RuntimeError("unexpected response from PREPL")
 
 def encode(val):
     return json.dumps(json.dumps(val))
@@ -60,7 +65,7 @@ def print_error(res):
 
 def run_cmd(params):
     try:
-        expr = "(app.srepl.ext/run-json-cmd {})".format(encode(params))
+        expr = "(app.srepl.cli/exec {})".format(encode(params))
         res, failed = send_eval(expr)
         if failed:
             print_error(res)
@@ -140,12 +145,22 @@ def derive_password(password):
     res = run_cmd(params)
     print(f"Derived password: \"{res}\"")
 
+
+def migrate_components_v2():
+    params = {
+        "cmd": "migrate-v2",
+        "params": {}
+    }
+
+    run_cmd(params)
+
 available_commands = (
     "create-profile",
     "update-profile",
     "delete-profile",
     "search-profile",
     "derive-password",
+    "migrate-components-v2",
 )
 
 parser = argparse.ArgumentParser(
@@ -217,3 +232,8 @@ elif args.action == "search-profile":
         email = input("Email: ")
 
     search_profile(email)
+
+elif args.action == "migrate-components-v2":
+    migrate_components_v2()
+
+

backend/src/app/srepl.clj

@@ -10,7 +10,7 @@
    [app.common.logging :as l]
    [app.common.spec :as us]
    [app.config :as cf]
-   [app.srepl.ext]
+   [app.srepl.cli]
    [app.srepl.main]
    [app.util.json :as json]
    [app.util.locks :as locks]

backend/src/app/srepl/cli.clj

@@ -4,14 +4,16 @@
 ;;
 ;; Copyright (c) KALEIDOS INC
 
-(ns app.srepl.ext
+(ns app.srepl.cli
   "PREPL API for external usage (CLI or ADMIN)"
   (:require
    [app.auth :as auth]
    [app.common.exceptions :as ex]
    [app.common.uuid :as uuid]
    [app.db :as db]
+   [app.main :as main]
    [app.rpc.commands.auth :as cmd.auth]
+   [app.srepl.components-v2]
    [app.util.json :as json]
    [app.util.time :as dt]
    [cuerdas.core :as str]))
@@ -21,18 +23,18 @@
   (or (deref (requiring-resolve 'app.main/system))
       (deref (requiring-resolve 'user/system))))
 
-(defmulti ^:private run-json-cmd* ::cmd)
+(defmulti ^:private exec-command ::cmd)
 
-(defn run-json-cmd
+(defn exec
   "Entry point with external tools integrations that uses PREPL
   interface for interacting with running penpot backend."
   [data]
-  (let [data (json/decode data)
-        params (merge {::cmd (keyword (:cmd data "default"))}
-                      (:params data))]
-    (run-json-cmd* params)))
+  (let [data (json/decode data)]
+    (-> {::cmd (keyword (:cmd data "default"))}
+        (merge (:params data))
+        (exec-command))))
 
-(defmethod run-json-cmd* :create-profile
+(defmethod exec-command :create-profile
   [{:keys [fullname email password is-active]
     :or {is-active true}}]
   (when-let [system (get-current-system)]
@@ -46,7 +48,7 @@
         (->> (cmd.auth/create-profile! conn params)
              (cmd.auth/create-profile-rels! conn))))))
 
-(defmethod run-json-cmd* :update-profile
+(defmethod exec-command :update-profile
   [{:keys [fullname email password is-active]}]
   (when-let [system (get-current-system)]
     (db/with-atomic [conn (:app.db/pool system)]
@@ -67,7 +69,7 @@
                                 {::db/return-keys? false})]
             (pos? (:next.jdbc/update-count res))))))))
 
-(defmethod run-json-cmd* :delete-profile
+(defmethod exec-command :delete-profile
   [{:keys [email soft]}]
   (when-not email
     (ex/raise :type :assertion
@@ -87,7 +89,7 @@
                               {::db/return-keys? false}))]
         (pos? (:next.jdbc/update-count res))))))
 
-(defmethod run-json-cmd* :search-profile
+(defmethod exec-command :search-profile
   [{:keys [email]}]
   (when-not email
     (ex/raise :type :assertion
@@ -101,11 +103,33 @@
                      " where email similar to ? order by created_at desc limit 100")]
         (db/exec! conn [sql email])))))
 
-(defmethod run-json-cmd* :derive-password
+(defmethod exec-command :derive-password
   [{:keys [password]}]
   (auth/derive-password password))
 
-(defmethod run-json-cmd* :default
+(defmethod exec-command :migrate-v2
+  [_]
+  (letfn [(on-start [{:keys [total rollback]}]
+            (println
+             (str/ffmt "The components/v2 migration started (rollback:%, teams:%)"
+                       (if rollback "on" "off")
+                       total)))
+
+          (on-progress [{:keys [total elapsed progress completed]}]
+            (println (str/ffmt "Progress % (total: %, completed: %, elapsed: %)"
+                               progress total completed elapsed)))
+          (on-error [cause]
+            (println "ERR:" (ex-message cause)))
+
+          (on-end [_]
+            (println "Migration finished"))]
+    (app.srepl.components-v2/migrate-teams! main/system
+                                            :on-start on-start
+                                            :on-error on-error
+                                            :on-progress on-progress
+                                            :on-end on-end)))
+
+(defmethod exec-command :default
   [{:keys [::cmd]}]
   (ex/raise :type :internal
             :code :not-implemented

backend/src/app/srepl/components_v2.clj

@@ -41,21 +41,26 @@
                :elapsed   (dt/format-duration elapsed))))))
 
 (defn- report-progress-teams
-  [tpoint]
+  [tpoint on-progress]
   (fn [_ _ oldv newv]
     (when (not= (:processed/teams oldv)
                 (:processed/teams newv))
       (let [total     (:total/teams newv)
             completed (:processed/teams newv)
             progress  (/ (* completed 100.0) total)
-            elapsed   (tpoint)]
+            progress  (str (int progress) "%")
+            elapsed   (dt/format-duration (tpoint))]
+
+        (when (fn? on-progress)
+          (on-progress {:total total
+                        :elapsed elapsed
+                        :completed completed
+                        :progress progress}))
+
         (l/dbg :hint "progress"
-               :completed-teams (:processed/teams newv)
-               :completed-files (:processed/files newv)
-               :completed-graphics (:processed/graphics newv)
-               :completed-components (:processed/components newv)
-               :progress  (str (int progress) "%")
-               :elapsed   (dt/format-duration elapsed))))))
+               :completed completed
+               :progress progress
+               :elapsed elapsed)))))
 
 (defn- get-total-files
   [pool & {:keys [team-id]}]
@@ -191,13 +196,23 @@
         (let [elapsed (dt/format-duration (tpoint))]
           (l/dbg :hint "migrate:end" :elapsed elapsed))))))
 
+(defn default-on-end
+  [stats]
+  (print-stats!
+   (-> stats
+       (update :elapsed/total dt/format-duration)
+       (dissoc :total/teams))))
+
 (defn migrate-teams!
   [{:keys [::db/pool] :as system}
-   & {:keys [chunk-size max-jobs max-items start-at rollback? preset skip-on-error max-time validate?]
+   & {:keys [chunk-size max-jobs max-items start-at
+             rollback? validate? preset skip-on-error
+             max-time on-start on-progress on-error on-end]
       :or {chunk-size 10000
            validate? false
            rollback? true
            skip-on-error true
+           on-end default-on-end
            preset :shutdown-on-failure
            max-jobs Integer/MAX_VALUE
            max-items Long/MAX_VALUE}}]
@@ -242,7 +257,10 @@
           tpoint (dt/tpoint)
           mtime  (some-> max-time dt/duration)]
 
-      (add-watch stats :progress-report (report-progress-teams tpoint))
+      (when (fn? on-start)
+        (on-start {:total total :rollback rollback?}))
+
+      (add-watch stats :progress-report (report-progress-teams tpoint on-progress))
 
       (binding [feat/*stats* stats
                 feat/*semaphore* sem
@@ -257,13 +275,15 @@
 
             (p/await! scope))
 
-          (print-stats!
-           (-> (deref feat/*stats*)
-               (assoc :elapsed/total (dt/format-duration (tpoint)))
-               (dissoc :total/teams)))
+          (when (fn? on-end)
+            (-> (deref stats)
+                (assoc :elapsed/total (tpoint))
+                (on-end)))
 
           (catch Throwable cause
-            (l/dbg :hint "migrate:error" :cause cause))
+            (l/dbg :hint "migrate:error" :cause cause)
+            (when (fn? on-error)
+              (on-error cause)))
 
           (finally
             (let [elapsed (dt/format-duration (tpoint))]

backend/src/app/srepl/main.clj

@@ -22,6 +22,7 @@
    [app.rpc.commands.auth :as auth]
    [app.rpc.commands.files-snapshot :as fsnap]
    [app.rpc.commands.profile :as profile]
+   [app.srepl.cli :as cli]
    [app.srepl.fixes :as f]
    [app.srepl.helpers :as h]
    [app.storage :as sto]