Merge pull request #214 from uxbox/other/backend-improvements

Other/backend improvements

backend/bin/start-prod

@@ -2,4 +2,4 @@
 
 set -e
 
-clojure ${CLOJURE_OPTIONS} -m uxbox.main
+clojure -O:jmx-remote -A:dev -J-Xms100m -J-Xmx100m -J-XX:+AlwaysPreTouch -J-XX:+UseBiasedLocking -J-Duxbox.enable-asserts=false -J-Dclojure.compiler.direct-linking=true -J-Dclojure.server.repl='{:port 5555 :accept clojure.core.server/repl}' -m uxbox.main

backend/deps.edn

@@ -9,11 +9,15 @@
 
   ;; Logging
   org.clojure/tools.logging {:mvn/version "1.1.0"}
-  org.apache.logging.log4j/log4j-api {:mvn/version "2.13.2"}
+  org.apache.logging.log4j/log4j-api {:mvn/version "2.13.3"}
-  org.apache.logging.log4j/log4j-core {:mvn/version "2.13.2"}
+  org.apache.logging.log4j/log4j-core {:mvn/version "2.13.3"}
-  org.apache.logging.log4j/log4j-web {:mvn/version "2.13.2"}
+  org.apache.logging.log4j/log4j-web {:mvn/version "2.13.3"}
-  org.apache.logging.log4j/log4j-jul {:mvn/version "2.13.2"}
+  org.apache.logging.log4j/log4j-jul {:mvn/version "2.13.3"}
-  org.apache.logging.log4j/log4j-slf4j-impl {:mvn/version "2.13.2"}
+  org.apache.logging.log4j/log4j-slf4j-impl {:mvn/version "2.13.3"}
+
+  io.prometheus/simpleclient {:mvn/version "0.9.0"}
+  io.prometheus/simpleclient_hotspot {:mvn/version "0.9.0"}
+  io.prometheus/simpleclient_httpserver {:mvn/version "0.9.0"}
 
   expound/expound {:mvn/version "0.8.4"}
   instaparse/instaparse {:mvn/version "1.4.10"}
@@ -26,7 +30,7 @@
   seancorfield/next.jdbc {:mvn/version "1.0.424"}
   metosin/reitit-ring {:mvn/version "0.4.2"}
   org.postgresql/postgresql {:mvn/version "42.2.12"}
-  com.zaxxer/HikariCP {:mvn/version "3.4.3"}
+  com.zaxxer/HikariCP {:mvn/version "3.4.5"}
 
   funcool/datoteka {:mvn/version "1.2.0"}
   funcool/promesa {:mvn/version "5.1.0"}
@@ -51,7 +55,7 @@
   io.aviso/pretty {:mvn/version "0.1.37"}
 
   mount/mount {:mvn/version "0.1.16"}
-  environ/environ {:mvn/version "1.1.0"}}
+  environ/environ {:mvn/version "1.2.0"}}
  :paths ["src" "resources" "../common" "common"]
  :aliases
  {:dev

backend/resources/log4j2.xml

@@ -1,3 +1,4 @@
+<?xml version="1.0" encoding="UTF-8"?>
 <Configuration status="info" monitorInterval="60">
   <Appenders>
     <Console name="console" target="SYSTEM_OUT">

backend/src/uxbox/db.clj

@@ -6,6 +6,7 @@
 
 (ns uxbox.db
   (:require
+   [clojure.data.json :as json]
    [clojure.string :as str]
    [clojure.tools.logging :as log]
    [lambdaisland.uri :refer [uri]]
@@ -16,10 +17,13 @@
    [next.jdbc.result-set :as jdbc-rs]
    [next.jdbc.sql :as jdbc-sql]
    [next.jdbc.sql.builder :as jdbc-bld]
+   [uxbox.metrics :as mtx]
    [uxbox.common.exceptions :as ex]
    [uxbox.config :as cfg]
    [uxbox.util.data :as data])
   (:import
+   org.postgresql.util.PGobject
+   com.zaxxer.hikari.metrics.prometheus.PrometheusMetricsTrackerFactory
    com.zaxxer.hikari.HikariConfig
    com.zaxxer.hikari.HikariDataSource))
 
@@ -28,17 +32,20 @@
   (let [dburi    (:database-uri cfg)
         username (:database-username cfg)
         password (:database-password cfg)
-        config (HikariConfig.)]
+        config (HikariConfig.)
+        mfactory (PrometheusMetricsTrackerFactory. mtx/registry)]
     (doto config
       (.setJdbcUrl (str "jdbc:" dburi))
+      (.setPoolName "main")
       (.setAutoCommit true)
       (.setReadOnly false)
-      (.setConnectionTimeout 30000)
+      (.setConnectionTimeout 30000) ;; 30seg
-      (.setValidationTimeout 5000)
+      (.setValidationTimeout 5000)  ;; 5seg
-      (.setIdleTimeout 600000)
+      (.setIdleTimeout 900000)      ;; 15min
-      (.setMaxLifetime 1800000)
+      (.setMaxLifetime 900000)      ;; 15min
-      (.setMinimumIdle 10)
+      (.setMinimumIdle 5)
-      (.setMaximumPoolSize 20))
+      (.setMaximumPoolSize 10)
+      (.setMetricsTrackerFactory mfactory))
     (when username (.setUsername config username))
     (when password (.setPassword config password))
     config))
@@ -112,3 +119,24 @@
    (get-by-params ds table {:id id} nil))
   ([ds table id opts]
    (get-by-params ds table {:id id} opts)))
+
+(defn pgobject?
+  [v]
+  (instance? PGobject v))
+
+(defn decode-pgobject
+  [^PGobject obj]
+  (let [typ (.getType obj)
+        val (.getValue obj)]
+    (if (or (= typ "json")
+            (= typ "jsonb"))
+      (json/read-str val)
+      val)))
+
+;; Instrumentation
+
+(mtx/instrument-with-counter!
+ {:var [#'jdbc/execute-one!
+        #'jdbc/execute!]
+  :id "database__query_counter"
+  :help "An absolute counter of database queries."})

backend/src/uxbox/emails.clj

@@ -41,9 +41,9 @@
                    :reply-to (:sendmail-reply-to cfg/config)}
          data (merge defaults context)
          email (email-factory data)]
-     (tasks/schedule! conn {:name "sendmail"
+     (tasks/submit! conn {:name "sendmail"
-                            :delay 0
+                          :delay 0
-                            :props email}))))
+                          :props email}))))
 
 ;; --- Emails
 

backend/src/uxbox/http.clj

@@ -17,12 +17,14 @@
    [uxbox.http.middleware :as middleware]
    [uxbox.http.session :as session]
    [uxbox.http.ws :as ws]
+   [uxbox.metrics :as mtx]
    [uxbox.services.notifications :as usn]))
 
 (defn- create-router
   []
   (rring/router
-   [["/api" {:middleware [[middleware/format-response-body]
+   [["/metrics" {:get mtx/dump}]
+    ["/api" {:middleware [[middleware/format-response-body]
                           [middleware/errors errors/handle]
                           [middleware/parse-request-body]
                           [middleware/params]
@@ -37,7 +39,6 @@
      ["/logout" {:handler handlers/logout-handler
                  :method :post}]
 
-
      ["/w" {:middleware [session/auth]}
       ["/query/:type" {:get handlers/query-handler}]
       ["/mutation/:type" {:post handlers/mutation-handler}]]]]))
@@ -46,8 +47,9 @@
   :start (rring/ring-handler
           (create-router)
           (constantly {:status 404, :body ""})
-          {:middleware [middleware/development-resources
+          {:middleware [[middleware/development-resources]
-                        middleware/development-cors]}))
+                        [middleware/development-cors]
+                        [middleware/metrics]]}))
 
 (defn start-server
   [cfg app]

backend/src/uxbox/http/errors.clj

@@ -9,6 +9,7 @@
   (:require
    [clojure.tools.logging :as log]
    [cuerdas.core :as str]
+   [uxbox.metrics :as mtx]
    [io.aviso.exception :as e]))
 
 (defmulti handle-exception

backend/src/uxbox/http/middleware.clj

@@ -15,6 +15,7 @@
    [ring.middleware.multipart-params :refer [wrap-multipart-params]]
    [ring.middleware.params :refer [wrap-params]]
    [ring.middleware.resource :refer [wrap-resource]]
+   [uxbox.metrics :as mtx]
    [uxbox.common.exceptions :as ex]
    [uxbox.config :as cfg]
    [uxbox.util.transit :as t]))
@@ -83,6 +84,12 @@
   {:name ::errors
    :compile (constantly wrap-errors)})
 
+(def metrics
+  {:name ::metrics
+   :wrap (fn [handler]
+           (mtx/wrap-counter handler {:id "http__requests_counter"
+                                      :help "Absolute http requests counter."}))})
+
 (def cookies
   {:name ::cookies
    :compile (constantly wrap-cookies)})

backend/src/uxbox/media_loader.clj

@@ -49,7 +49,8 @@
 (s/def ::path ::us/string)
 (s/def ::regex #(instance? java.util.regex.Pattern %))
 
-(s/def ::colors (s/every ::us/color :kind set?))
+(s/def ::colors
+  (s/* (s/cat :name ::us/string :color ::us/color)))
 
 (s/def ::import-item-media
   (s/keys :req-un [::name ::path ::regex]))
@@ -238,22 +239,23 @@
     id))
 
 (defn- create-color
-  [conn library-id content]
+  [conn library-id name content]
   (s/assert ::us/uuid library-id)
   (s/assert ::us/color content)
   (let [color-id (uuid/namespaced +colors-uuid-ns+ (str library-id content))]
-    (log/info "Creating color" content color-id)
+    (log/info "Creating color" color-id "-" name content)
     (colors/create-color conn {:id color-id
                                :library-id library-id
-                               :name content
+                               :name name
                                :content content})
     color-id))
 
 (defn- import-colors
   [conn library-id {:keys [colors] :as item}]
-  (us/verify ::import-item-color item)
   (db/delete! conn :color {:library-id library-id})
-  (run! #(create-color conn library-id %) colors))
+  (run! (fn [[name content]]
+          (create-color conn library-id name content))
+        (partition-all 2 colors)))
 
 (defn- process-colors-library
   [conn {:keys [name id colors] :as item}]

backend/src/uxbox/metrics.clj

@@ -0,0 +1,181 @@
+;; 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/.
+;;
+;; This Source Code Form is "Incompatible With Secondary Licenses", as
+;; defined by the Mozilla Public License, v. 2.0.
+;;
+;; Copyright (c) 2020 UXBOX Labs SL
+
+(ns uxbox.metrics
+  (:require
+   [clojure.tools.logging :as log]
+   [cuerdas.core :as str])
+  (:import
+   io.prometheus.client.CollectorRegistry
+   io.prometheus.client.Counter
+   io.prometheus.client.Gauge
+   io.prometheus.client.Summary
+   io.prometheus.client.exporter.common.TextFormat
+   io.prometheus.client.hotspot.DefaultExports
+   java.io.StringWriter))
+
+(defn- create-registry
+  []
+  (let [registry (CollectorRegistry.)]
+    (DefaultExports/register registry)
+    registry))
+
+(defonce registry (create-registry))
+(defonce cache (atom {}))
+
+(defmacro with-measure
+  [sym expr teardown]
+  `(let [~sym (System/nanoTime)]
+     (try
+       ~expr
+       (finally
+         (let [~sym (/ (- (System/nanoTime) ~sym) 1000000)]
+           ~teardown)))))
+
+(defn make-counter
+  [{:keys [id help] :as props}]
+  (let [instance (doto (Counter/build)
+                   (.name id)
+                   (.help help))
+        instance (.register instance registry)]
+    (reify
+      clojure.lang.IDeref
+      (deref [_] instance)
+
+      clojure.lang.IFn
+      (invoke [_ cmd]
+        (.inc ^Counter instance))
+
+      (invoke [_ cmd val]
+        (case cmd
+          :wrap (fn
+                  ([a]
+                   (.inc ^Counter instance)
+                   (val a))
+                  ([a b]
+                   (.inc ^Counter instance)
+                   (val a b))
+                  ([a b c]
+                   (.inc ^Counter instance)
+                   (val a b c)))
+
+          (throw (IllegalArgumentException. "invalid arguments")))))))
+
+(defn counter
+  [{:keys [id] :as props}]
+  (or (get @cache id)
+      (let [v (make-counter props)]
+        (swap! cache assoc id v)
+        v)))
+
+(defn make-gauge
+  [{:keys [id help] :as props}]
+  (let [instance (doto (Gauge/build)
+                   (.name id)
+                   (.help help))
+        instance (.register instance registry)]
+    (reify
+      clojure.lang.IDeref
+      (deref [_] instance)
+
+      clojure.lang.IFn
+      (invoke [_ cmd]
+        (case cmd
+          :inc (.inc ^Gauge instance)
+          :dec (.dec ^Gauge instance))))))
+
+(defn gauge
+  [{:keys [id] :as props}]
+  (or (get @cache id)
+      (let [v (make-gauge props)]
+        (swap! cache assoc id v)
+        v)))
+
+(defn make-summary
+  [{:keys [id help] :as props}]
+  (let [instance (doto (Summary/build)
+                   (.name id)
+                   (.help help)
+                   (.quantile 0.5 0.05)
+                   (.quantile 0.9 0.01)
+                   (.quantile 0.99 0.001))
+        instance  (.register instance registry)]
+    (reify
+      clojure.lang.IDeref
+      (deref [_] instance)
+
+      clojure.lang.IFn
+      (invoke [_ val]
+        (.observe ^Summary instance val))
+
+      (invoke [_ cmd val]
+        (case cmd
+          :wrap (fn
+                  ([a]
+                   (with-measure $$
+                     (val a)
+                     (.observe ^Summary instance $$)))
+                  ([a b]
+                   (with-measure $$
+                     (val a b)
+                     (.observe ^Summary instance $$)))
+                  ([a b c]
+                   (with-measure $$
+                     (val a b c)
+                     (.observe ^Summary instance $$))))
+
+          (throw (IllegalArgumentException. "invalid arguments")))))))
+
+(defn summary
+  [{:keys [id] :as props}]
+  (or (get @cache id)
+      (let [v (make-summary props)]
+        (swap! cache assoc id v)
+        v)))
+
+(defn wrap-summary
+  [f props]
+  (let [sm (summary props)]
+    (sm :wrap f)))
+
+(defn wrap-counter
+  [f props]
+  (let [cnt (counter props)]
+    (cnt :wrap f)))
+
+(defn instrument-with-counter!
+  [{:keys [var] :as props}]
+  (let [cnt  (counter props)
+        vars (if (var? var) [var] var)]
+    (doseq [var vars]
+      (alter-var-root var (fn [root]
+                            (let [mdata (meta root)
+                                  original (::counter-original mdata root)]
+                              (with-meta
+                                (cnt :wrap original)
+                                (assoc mdata ::counter-original original))))))))
+
+(defn instrument-with-summary!
+  [{:keys [var] :as props}]
+  (let [sm (summary props)]
+    (alter-var-root var (fn [root]
+                          (let [mdata (meta root)
+                                original (::summary-original mdata root)]
+                            (with-meta
+                              (sm :wrap original)
+                              (assoc mdata ::summary-original original)))))))
+
+(defn dump
+  [& args]
+  (let [samples (.metricFamilySamples ^CollectorRegistry registry)
+        writer  (StringWriter.)]
+    (TextFormat/write004 writer samples)
+    {:headers {"content-type" TextFormat/CONTENT_TYPE_004}
+     :body (.toString writer)}))
+

backend/src/uxbox/services/middleware.clj

@@ -0,0 +1,73 @@
+;; 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/.
+;;
+;; This Source Code Form is "Incompatible With Secondary Licenses", as
+;; defined by the Mozilla Public License, v. 2.0.
+;;
+;; Copyright (c) 2020 UXBOX Labs SL
+
+(ns uxbox.services.middleware
+  "Common middleware for services."
+  (:require
+   [clojure.tools.logging :as log]
+   [clojure.spec.alpha :as s]
+   [cuerdas.core :as str]
+   [expound.alpha :as expound]
+   [uxbox.common.exceptions :as ex]
+   [uxbox.common.spec :as us]
+   [uxbox.metrics :as mtx]))
+
+(defn wrap-spec
+  [handler]
+  (let [mdata (meta handler)
+        spec (s/get-spec (:spec mdata))]
+    (if (nil? spec)
+      handler
+      (with-meta
+        (fn [params]
+          (let [result (us/conform spec params)]
+            (handler result)))
+        (assoc mdata ::wrap-spec true)))))
+
+(defn wrap-error
+  [handler]
+  (let [mdata (meta handler)]
+    (with-meta
+      (fn [params]
+        (try
+          (handler params)
+          (catch Throwable error
+            (ex/raise :type :service-error
+                      :name (:spec mdata)
+                      :cause error))))
+      (assoc mdata ::wrap-error true))))
+
+(defn- get-prefix
+  [nsname]
+  (let [[a b c] (str/split nsname ".")]
+    c))
+
+(defn wrap-metrics
+  [handler]
+  (let [mdata  (meta handler)
+        nsname (namespace (:spec mdata))
+        smname (name (:spec mdata))
+        prefix (get-prefix nsname)
+
+        sname  (str prefix "/" smname)
+
+        props  {:id (str/join "__" [prefix
+                                    (str/snake smname)
+                                    "response_time"])
+                :help (str "Service timing measures for: " sname ".")}]
+    (with-meta
+      (mtx/wrap-summary handler props)
+      (assoc mdata ::wrap-metrics true))))
+
+(defn wrap
+  [handler]
+  (-> handler
+      (wrap-spec)
+      (wrap-error)
+      (wrap-metrics)))

backend/src/uxbox/services/mutations.clj

@@ -5,16 +5,16 @@
 ;; This Source Code Form is "Incompatible With Secondary Licenses", as
 ;; defined by the Mozilla Public License, v. 2.0.
 ;;
-;; Copyright (c) 2019-2020 Andrey Antukh <niwi@niwi.nz>
+;; Copyright (c) 2020 UXBOX Labs SL
 
 (ns uxbox.services.mutations
   (:require
+   [uxbox.services.middleware :as middleware]
    [uxbox.util.dispatcher :as uds]))
 
 (uds/defservice handle
   :dispatch-by ::type
-  :wrap [uds/wrap-spec
+  :wrap middleware/wrap)
-         uds/wrap-error])
 
 (defmacro defmutation
   [key & rest]

backend/src/uxbox/services/mutations/colors.clj

@@ -104,9 +104,9 @@
       (teams/check-edition-permissions! conn profile-id (:team-id lib))
 
       ;; Schedule object deletion
-      (tasks/schedule! conn {:name "delete-object"
+      (tasks/submit! conn {:name "delete-object"
-                             :delay cfg/default-deletion-delay
+                           :delay cfg/default-deletion-delay
-                             :props {:id id :type :color-library}})
+                           :props {:id id :type :color-library}})
 
       (db/update! conn :color-library
                   {:deleted-at (dt/now)}
@@ -188,9 +188,9 @@
       (teams/check-edition-permissions! conn profile-id (:team-id clr))
 
       ;; Schedule object deletion
-      (tasks/schedule! conn {:name "delete-object"
+      (tasks/submit! conn {:name "delete-object"
-                             :delay cfg/default-deletion-delay
+                           :delay cfg/default-deletion-delay
-                             :props {:id id :type :color}})
+                           :props {:id id :type :color}})
 
       (db/update! conn :color
                   {:deleted-at (dt/now)}

backend/src/uxbox/services/mutations/demo.clj

@@ -38,8 +38,8 @@
                                         :password password})
 
       ;; Schedule deletion of the demo profile
-      (tasks/schedule! conn {:name "delete-profile"
+      (tasks/submit! conn {:name "delete-profile"
-                             :delay cfg/default-deletion-delay
+                           :delay cfg/default-deletion-delay
-                             :props {:profile-id id}})
+                           :props {:profile-id id}})
       {:email email
        :password password})))

backend/src/uxbox/services/mutations/files.clj

@@ -113,9 +113,9 @@
     (files/check-edition-permissions! conn profile-id id)
 
     ;; Schedule object deletion
-    (tasks/schedule! conn {:name "delete-object"
+    (tasks/submit! conn {:name "delete-object"
-                           :delay cfg/default-deletion-delay
+                         :delay cfg/default-deletion-delay
-                           :props {:id id :type :file}})
+                         :props {:id id :type :file}})
 
     (mark-file-deleted conn params)))
 

backend/src/uxbox/services/mutations/icons.clj

@@ -111,9 +111,9 @@
       (teams/check-edition-permissions! conn profile-id (:team-id lib))
 
       ;; Schedule object deletion
-      (tasks/schedule! conn {:name "delete-object"
+      (tasks/submit! conn {:name "delete-object"
-                             :delay cfg/default-deletion-delay
+                           :delay cfg/default-deletion-delay
-                             :props {:id id :type :icon-library}})
+                           :props {:id id :type :icon-library}})
 
       (db/update! conn :icon-library
                   {:deleted-at (dt/now)}
@@ -196,9 +196,9 @@
       (teams/check-edition-permissions! conn profile-id (:team-id icn))
 
       ;; Schedule object deletion
-      (tasks/schedule! conn {:name "delete-object"
+      (tasks/submit! conn {:name "delete-object"
-                             :delay cfg/default-deletion-delay
+                           :delay cfg/default-deletion-delay
-                             :props {:id id :type :icon}})
+                           :props {:id id :type :icon}})
 
       (db/update! conn :icon
                   {:deleted-at (dt/now)}

backend/src/uxbox/services/mutations/images.clj

@@ -96,9 +96,9 @@
       (teams/check-edition-permissions! conn profile-id (:team-id lib))
 
       ;; Schedule object deletion
-      (tasks/schedule! conn {:name "delete-object"
+      (tasks/submit! conn {:name "delete-object"
-                             :delay cfg/default-deletion-delay
+                           :delay cfg/default-deletion-delay
-                             :props {:id id :type :image-library}})
+                           :props {:id id :type :image-library}})
 
       (db/update! conn :image-library
                   {:deleted-at (dt/now)}
@@ -226,9 +226,9 @@
       (teams/check-edition-permissions! conn profile-id (:team-id img))
 
       ;; Schedule object deletion
-      (tasks/schedule! conn {:name "delete-object"
+      (tasks/submit! conn {:name "delete-object"
-                             :delay cfg/default-deletion-delay
+                           :delay cfg/default-deletion-delay
-                             :props {:id id :type :image}})
+                           :props {:id id :type :image}})
 
       (db/update! conn :image
                   {:deleted-at (dt/now)}

backend/src/uxbox/services/mutations/pages.clj

@@ -21,8 +21,10 @@
    [uxbox.services.queries.files :as files]
    [uxbox.services.queries.pages :refer [decode-row]]
    [uxbox.tasks :as tasks]
+   [uxbox.redis :as redis]
    [uxbox.util.blob :as blob]
-   [uxbox.util.time :as dt]))
+   [uxbox.util.time :as dt]
+   [uxbox.util.transit :as t]))
 
 ;; --- Helpers & Specs
 
@@ -148,9 +150,10 @@
 (s/def ::changes
   (s/coll-of map? :kind vector?))
 
+(s/def ::session-id ::us/uuid)
 (s/def ::revn ::us/integer)
 (s/def ::update-page
-  (s/keys :req-un [::id ::profile-id ::revn ::changes]))
+  (s/keys :req-un [::id ::session-id ::profile-id ::revn ::changes]))
 
 (declare update-page)
 (declare retrieve-lagged-changes)
@@ -172,7 +175,9 @@
               :hint "The incoming revision number is greater that stored version."
               :context {:incoming-revn (:revn params)
                         :stored-revn (:revn page)}))
-  (let [changes  (:changes params)
+  (let [sid      (:session-id params)
+        changes  (->> (:changes params)
+                      (mapv #(assoc % :session-id sid)))
         data (-> (:data page)
                  (blob/decode)
                  (cp/process-changes changes)
@@ -183,7 +188,16 @@
                     :revn (inc (:revn page))
                     :changes (blob/encode changes))
 
-        chng (insert-page-change! conn page)]
+        chng (insert-page-change! conn page)
+        msg  {:type :page-change
+              :profile-id (:profile-id params)
+              :page-id (:id page)
+              :session-id sid
+              :revn (:revn page)
+              :changes changes}]
+
+    @(redis/run! :publish {:channel (str (:file-id page))
+                           :message (t/encode-str msg)})
 
     (db/update! conn :page
                 {:revn (:revn page)
@@ -192,13 +206,6 @@
 
     (retrieve-lagged-changes conn chng params)))
 
-;; (p/do! (ve/publish! uxbox.core/system topic
-;;                     {:type :page-change
-;;                      :profile-id (:profile-id params)
-;;                      :page-id (:page-id s)
-;;                      :revn (:revn s)
-;;                      :changes changes})
-
 (defn- insert-page-change!
   [conn {:keys [revn data changes] :as page}]
   (let [id (uuid/next)
@@ -242,9 +249,9 @@
       (files/check-edition-permissions! conn profile-id (:file-id page))
 
       ;; Schedule object deletion
-      (tasks/schedule! conn {:name "delete-object"
+      (tasks/submit! conn {:name "delete-object"
-                             :delay cfg/default-deletion-delay
+                           :delay cfg/default-deletion-delay
-                             :props {:id id :type :page}})
+                           :props {:id id :type :page}})
 
       (db/update! conn :page
                   {:deleted-at (dt/now)}

backend/src/uxbox/services/mutations/profile.clj

@@ -155,8 +155,8 @@
       ;; Schedule deletion of old photo
       (when (and (string? (:photo profile))
                  (not (str/blank? (:photo profile))))
-        (tasks/schedule! conn {:name "remove-media"
+        (tasks/submit! conn {:name "remove-media"
-                               :props {:path (:photo profile)}}))
+                             :props {:path (:photo profile)}}))
       ;; Save new photo
       (update-profile-photo conn profile-id photo))))
 
@@ -363,9 +363,9 @@
     (check-teams-ownership! conn profile-id)
 
     ;; Schedule a complete deletion of profile
-    (tasks/schedule! conn {:name "delete-profile"
+    (tasks/submit! conn {:name "delete-profile"
-                           :delay (dt/duration {:hours 48})
+                         :delay (dt/duration {:hours 48})
-                           :props {:profile-id profile-id}})
+                         :props {:profile-id profile-id}})
 
     (db/update! conn :profile
                 {:deleted-at (dt/now)}

backend/src/uxbox/services/mutations/projects.clj

@@ -124,9 +124,9 @@
     (check-edition-permissions! conn profile-id id)
 
     ;; Schedule object deletion
-    (tasks/schedule! conn {:name "delete-object"
+    (tasks/submit! conn {:name "delete-object"
-                           :delay cfg/default-deletion-delay
+                         :delay cfg/default-deletion-delay
-                           :props {:id id :type :project}})
+                         :props {:id id :type :project}})
 
     (mark-project-deleted conn params)))
 

backend/src/uxbox/services/notifications.clj

@@ -13,8 +13,9 @@
    [ring.adapter.jetty9 :as jetty]
    [uxbox.common.exceptions :as ex]
    [uxbox.common.uuid :as uuid]
-   [uxbox.redis :as redis]
    [uxbox.db :as db]
+   [uxbox.redis :as redis]
+   [uxbox.metrics :as mtx]
    [uxbox.util.time :as dt]
    [uxbox.util.transit :as t]))
 
@@ -193,11 +194,20 @@
         (jetty/send! conn (t/encode-str val))
         (recur)))))
 
+(defonce metrics-active-connections
+  (mtx/gauge {:id "notificatons__active_connections"
+              :help "Active connections to the notifications service."}))
+
+(defonce metrics-message-counter
+  (mtx/counter {:id "notificatons__messages_counter"
+                :help "A total number of messages handled by the notifications service."}))
+
 (defn websocket
   [{:keys [file-id] :as params}]
   (let [in  (a/chan 32)
         out (a/chan 32)]
     {:on-connect (fn [conn]
+                   (metrics-active-connections :inc)
                    (let [xf  (map t/decode-str)
                          sub (redis/subscribe (str file-id) xf)
                          ws  (WebSocket. conn in out sub nil params)]
@@ -207,21 +217,19 @@
                        (a/close! sub))))
 
      :on-error (fn [conn e]
-                 ;; (prn "websocket" :on-error e)
                  (a/close! out)
                  (a/close! in))
 
      :on-close (fn [conn status-code reason]
-                 ;; (prn "websocket" :on-close status-code reason)
+                 (metrics-active-connections :dec)
                  (a/close! out)
                  (a/close! in))
 
      :on-text (fn [ws message]
+                (metrics-message-counter :inc)
                 (let [message (t/decode-str message)]
-                  ;; (prn "websocket" :on-text message)
                   (a/>!! in message)))
 
-     :on-bytes (fn [ws bytes offset len]
+     :on-bytes (constantly nil)}))
-                 #_(prn "websocket" :on-bytes bytes))}))
 
 

backend/src/uxbox/services/queries.clj

@@ -2,16 +2,19 @@
 ;; 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) 2019 Andrey Antukh <niwi@niwi.nz>
+;; This Source Code Form is "Incompatible With Secondary Licenses", as
+;; defined by the Mozilla Public License, v. 2.0.
+;;
+;; Copyright (c) 2020 UXBOX Labs SL
 
 (ns uxbox.services.queries
   (:require
+   [uxbox.services.middleware :as middleware]
    [uxbox.util.dispatcher :as uds]))
 
 (uds/defservice handle
   :dispatch-by ::type
-  :wrap [uds/wrap-spec
+  :wrap middleware/wrap)
-         uds/wrap-error])
 
 (defmacro defquery
   [key & rest]

backend/src/uxbox/services/queries/icons.clj

@@ -10,8 +10,6 @@
 (ns uxbox.services.queries.icons
   (:require
    [clojure.spec.alpha :as s]
-   [promesa.core :as p]
-   [promesa.exec :as px]
    [uxbox.common.exceptions :as ex]
    [uxbox.common.spec :as us]
    [uxbox.common.uuid :as uuid]

backend/src/uxbox/services/queries/images.clj

@@ -10,13 +10,12 @@
 (ns uxbox.services.queries.images
   (:require
    [clojure.spec.alpha :as s]
-   [promesa.core :as p]
    [uxbox.common.exceptions :as ex]
    [uxbox.common.spec :as us]
    [uxbox.db :as db]
    [uxbox.images :as images]
-   [uxbox.services.queries.teams :as teams]
+   [uxbox.services.queries :as sq]
-   [uxbox.services.queries :as sq]))
+   [uxbox.services.queries.teams :as teams]))
 
 (s/def ::id ::us/uuid)
 (s/def ::name ::us/string)

backend/src/uxbox/tasks.clj

@@ -16,12 +16,23 @@
    [uxbox.common.spec :as us]
    [uxbox.config :as cfg]
    [uxbox.db :as db]
+   [uxbox.metrics :as mtx]
    [uxbox.tasks.sendmail]
+   [uxbox.tasks.gc]
    [uxbox.tasks.remove-media]
    [uxbox.tasks.delete-profile]
    [uxbox.tasks.delete-object]
    [uxbox.tasks.impl :as impl]
-   [uxbox.util.time :as dt]))
+   [uxbox.util.time :as dt])
+  (:import
+   java.util.concurrent.ScheduledExecutorService
+   java.util.concurrent.Executors))
+
+;; --- Scheduler Executor Initialization
+
+(defstate scheduler
+  :start (Executors/newScheduledThreadPool (int 1))
+  :stop (.shutdownNow ^ScheduledExecutorService scheduler))
 
 ;; --- State initialization
 
@@ -36,33 +47,30 @@
    "remove-media" #'uxbox.tasks.remove-media/handler
    "sendmail" #'uxbox.tasks.sendmail/handler})
 
+(def ^:private schedule
+  [{:id "remove-deleted-media"
+    :cron (dt/cron "1 1 */1 * * ? *")
+    :fn #'uxbox.tasks.gc/remove-media}])
+
 (defstate worker
-  :start (impl/start-worker! {:tasks tasks})
+  :start (impl/start-worker! {:tasks tasks
+                              :xtor scheduler})
+  :stop (impl/stop! worker))
+
+(defstate scheduler-worker
+  :start (impl/start-scheduler-worker! {:schedule schedule
+                                        :xtor scheduler})
   :stop (impl/stop! worker))
 
 ;; --- Public API
 
-(defn schedule!
+(defn submit!
-  ([opts] (schedule! db/pool opts))
+  ([opts] (submit! db/pool opts))
   ([conn opts]
    (s/assert ::impl/task-options opts)
-   (impl/schedule! conn opts)))
+   (impl/submit! conn opts)))
 
-;; (defstate scheduler
+(mtx/instrument-with-counter!
-;;   :start (impl/start-scheduler! tasks)
+ {:var #'submit!
-;;   :stop (impl/stop! tasks-worker))
+  :id "tasks__submit_counter"
-
+  :help "Absolute task submit counter."})
-;; :start (as-> (impl/worker-verticle {:tasks tasks}) $$
-;;          (vc/deploy! system $$ {:instances 1})
-;;          (deref $$)))
-
-;; (def ^:private schedule
-;;   [{:id "every 1 hour"
-;;     :cron (dt/cron "1 1 */1 * * ? *")
-;;     :fn #'uxbox.tasks.gc/handler
-;;     :props {:foo 1}}])
-
-;; (defstate scheduler
-;;   :start (as-> (impl/scheduler-verticle {:schedule schedule}) $$
-;;            (vc/deploy! system $$ {:instances 1 :worker true})
-;;            (deref $$)))

backend/src/uxbox/tasks/delete_object.clj

@@ -15,7 +15,7 @@
    [uxbox.common.exceptions :as ex]
    [uxbox.common.spec :as us]
    [uxbox.db :as db]
-   [uxbox.media :as media]
+   [uxbox.metrics :as mtx]
    [uxbox.util.storage :as ust]))
 
 (s/def ::type keyword?)
@@ -36,6 +36,11 @@
   (db/with-atomic [conn db/pool]
     (handle-deletion conn props)))
 
+(mtx/instrument-with-summary!
+ {:var #'handler
+  :id "tasks__delete_object"
+  :help "Timing of remove-object task."})
+
 (defmethod handle-deletion :image
   [conn {:keys [id] :as props}]
   (let [sql "delete from image where id=? and deleted_at is not null"]

backend/src/uxbox/tasks/delete_profile.clj

@@ -15,7 +15,7 @@
    [uxbox.common.exceptions :as ex]
    [uxbox.common.spec :as us]
    [uxbox.db :as db]
-   [uxbox.media :as media]
+   [uxbox.metrics :as mtx]
    [uxbox.util.storage :as ust]))
 
 (declare delete-profile-data)
@@ -39,6 +39,11 @@
         (log/warn "Profile " (:id profile)
                   "does not match constraints for deletion")))))
 
+(mtx/instrument-with-summary!
+ {:var #'handler
+  :id "tasks__delete_profile"
+  :help "Timing of delete-profile task."})
+
 (defn- delete-profile-data
   [conn profile-id]
   (log/info "Proceding to delete all data related to profile" profile-id)

backend/src/uxbox/tasks/gc.clj

@@ -9,8 +9,8 @@
 
 (ns uxbox.tasks.gc
   (:require
-   [clojure.tools.logging :as log]
    [clojure.spec.alpha :as s]
+   [clojure.tools.logging :as log]
    [cuerdas.core :as str]
    [postal.core :as postal]
    [promesa.core :as p]
@@ -18,36 +18,47 @@
    [uxbox.common.spec :as us]
    [uxbox.config :as cfg]
    [uxbox.db :as db]
-   [uxbox.util.blob :as blob]))
+   [uxbox.media :as media]
+   [uxbox.util.blob :as blob]
+   [uxbox.util.storage :as ust]))
 
-;; TODO: delete media referenced in pendint_to_delete table
+(def ^:private sql:delete-items
+  "with items_part as (
+     select i.id
+       from pending_to_delete as i
+      order by i.created_at
+      limit ?
+      for update skip locked
+   )
+   delete from pending_to_delete
+    where id in (select id from items_part)
+   returning *")
 
-;; (def ^:private sql:delete-item
+(defn- impl-remove-media
-;;   "with items_part as (
+  [result]
-;;      select i.id
+  (run! (fn [item]
-;;        from pending_to_delete as i
+          (let [path1 (get item "path")
-;;       order by i.created_at
+                path2 (get item "thumb_path")]
-;;       limit 1
+            (ust/delete! media/media-storage path1)
-;;       for update skip locked
+            (ust/delete! media/media-storage path2)))
-;;    )
+        result))
-;;    delete from pending_to_delete
-;;     where id in (select id from items_part)
-;;    returning *")
 
-;; (defn- remove-items
+(defn- decode-row
-;;   []
+  [{:keys [data] :as row}]
-;;   (vu/loop []
+  (cond-> row
-;;     (db/with-atomic [conn db/pool]
+    (db/pgobject? data) (assoc :data (db/decode-pgobject data))))
-;;       (-> (db/query-one conn sql:delete-item)
+
-;;           (p/then decode-row)
+(defn- get-items
-;;           (p/then (vu/wrap-blocking remove-media))
+  [conn]
-;;           (p/then (fn [item]
+  (->> (db/exec! conn [sql:delete-items 10])
-;;                     (when (not (empty? items))
+       (map decode-row)
-;;                       (p/recur))))))))
+       (map :data)))
+
+(defn remove-media
+  [{:keys [props] :as task}]
+  (db/with-atomic [conn db/pool]
+    (loop [result (get-items conn)]
+      (when-not (empty? result)
+        (impl-remove-media result)
+        (recur (get-items conn))))))
 
-;; (defn- remove-media
-;;   [{:keys
-;;   (doseq [item files]
-;;     (ust/delete! media/media-storage (:path item))
-;;     (ust/delete! media/media-storage (:thumb-path item)))
-;;   files)

backend/src/uxbox/tasks/impl.clj

@@ -13,6 +13,7 @@
    [clojure.core.async :as a]
    [clojure.spec.alpha :as s]
    [clojure.tools.logging :as log]
+   [promesa.exec :as px]
    [uxbox.common.spec :as us]
    [uxbox.common.uuid :as uuid]
    [uxbox.config :as cfg]
@@ -20,14 +21,12 @@
    [uxbox.util.blob :as blob]
    [uxbox.util.time :as dt])
   (:import
+   java.util.concurrent.ScheduledExecutorService
+   java.util.concurrent.Executors
    java.time.Duration
    java.time.Instant
    java.util.Date))
 
-(defrecord Worker [stop]
-  java.lang.AutoCloseable
-  (close [_] (a/close! stop)))
-
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Tasks
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
@@ -37,7 +36,8 @@
   (with-out-str
     (.printStackTrace err (java.io.PrintWriter. *out*))))
 
-(def ^:private sql:mark-as-retry
+(def ^:private
+  sql:mark-as-retry
   "update task
       set scheduled_at = clock_timestamp() + '5 seconds'::interval,
           error = ?,
@@ -45,48 +45,32 @@
           retry_num = retry_num + 1
     where id = ?")
 
-(defn- reschedule
+(defn- mark-as-retry
   [conn task error]
   (let [explain (ex-message error)
         sqlv [sql:mark-as-retry explain (:id task)]]
     (db/exec-one! conn sqlv)
     nil))
 
-(def ^:private sql:mark-as-failed
-  "update task
-      set scheduled_at = clock_timestamp() + '5 seconds'::interval,
-          error = ?,
-          status = 'failed'
-    where id = ?;")
-
 (defn- mark-as-failed
   [conn task error]
-  (let [explain (ex-message error)
+  (let [explain (ex-message error)]
-        sqlv [sql:mark-as-failed explain (:id task)]]
+    (db/update! conn :task
-    (db/exec-one! conn sqlv)
+                {:error explain
+                 :status "failed"}
+                {:id (:id task)})
     nil))
 
-(def ^:private sql:mark-as-completed
-  "update task
-      set completed_at = clock_timestamp(),
-          status = 'completed'
-    where id = ?")
-
 (defn- mark-as-completed
   [conn task]
-  (db/exec-one! conn [sql:mark-as-completed (:id task)])
+  (db/update! conn :task
+              {:completed-at (dt/now)
+               :status "completed"}
+              {:id (:id task)})
   nil)
 
-(defn- handle-task
+(def ^:private
-  [tasks {:keys [name] :as item}]
+  sql:select-next-task
-  (let [task-fn (get tasks name)]
-    (if task-fn
-      (task-fn item)
-      (do
-        (log/warn "no task handler found for" (pr-str name))
-        nil))))
-
-(def ^:private sql:select-next-task
   "select * from task as t
     where t.scheduled_at <= now()
       and t.queue = ?
@@ -108,6 +92,15 @@
              (with-out-str
                (.printStackTrace ^Throwable err (java.io.PrintWriter. *out*)))))
 
+(defn- handle-task
+  [tasks {:keys [name] :as item}]
+  (let [task-fn (get tasks name)]
+    (if task-fn
+      (task-fn item)
+      (do
+        (log/warn "no task handler found for" (pr-str name))
+        nil))))
+
 (defn- event-loop-fn
   [{:keys [tasks] :as options}]
   (let [queue (:queue options "default")
@@ -125,156 +118,140 @@
               (log-task-error item e)
               (if (>= (:retry-num item) max-retries)
                 (mark-as-failed conn item e)
-                (reschedule conn item e)))))))))
+                (mark-as-retry conn item e)))))))))
 
-(defn- start-worker-eventloop!
+(defn- execute-worker-task
-  [options]
+  [{:keys [::stop ::xtor poll-interval]
-  (let [stop (::stop options)
+    :or {poll-interval 5000}
-        mbs  (:max-batch-size options 10)]
+    :as opts}]
-    (a/go-loop []
+  (try
-      (let [timeout    (a/timeout 5000)
+    (when-not @stop
-            [val port] (a/alts! [stop timeout])]
+      (let [res (event-loop-fn opts)]
-        (when (= port timeout)
+        (if (= res ::handled)
-          (a/<! (a/thread
+          (px/schedule! xtor 0 (partial execute-worker-task opts))
-                  ;; Tasks batching in one event loop execution.
+          (px/schedule! xtor poll-interval (partial execute-worker-task opts)))))
-                  (loop [cnt 1
+    (catch Throwable e
-                         res (event-loop-fn options)]
+      (log/error "unexpected exception:" e)
-                    (when (and (= res ::handled)
+      (px/schedule! xtor poll-interval (partial execute-worker-task opts)))))
-                               (> mbs cnt))
-                      (recur (inc 1)
-                             (event-loop-fn options))))))
-          (recur))))))
-
-(defn- duration->pginterval
-  [^Duration d]
-  (->> (/ (.toMillis d) 1000.0)
-       (format "%s seconds")))
-
-(defn start-worker!
-  [options]
-  (let [stop (a/chan)]
-    (start-worker-eventloop! (assoc options ::stop stop))
-    (->Worker stop)))
-
-(defn stop!
-  [worker]
-  (.close ^java.lang.AutoCloseable worker))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Scheduled Tasks
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
-;; (def ^:privatr sql:upsert-scheduled-task
+(def ^:private
-;;   "insert into scheduled_task (id, cron_expr)
+  sql:upsert-scheduled-task
-;;    values ($1, $2)
+  "insert into scheduled_task (id, cron_expr)
-;;        on conflict (id)
+   values (?, ?)
-;;        do update set cron_expr=$2")
+       on conflict (id)
+       do update set cron_expr=?")
 
-;; (defn- synchronize-schedule-item
+(defn- synchronize-schedule-item
-;;   [conn {:keys [id cron]}]
+  [conn {:keys [id cron] :as item}]
-;;   (-> (db/query-one conn [sql:upsert-scheduled-task id (str cron)])
+  (let [cron (str cron)]
-;;       (p/then' (constantly nil))))
+    (db/exec-one! conn [sql:upsert-scheduled-task id cron cron])))
 
-;; (defn- synchronize-schedule
+(defn- synchronize-schedule!
-;;   [schedule]
+  [schedule]
-;;   (db/with-atomic [conn db/pool]
+  (db/with-atomic [conn db/pool]
-;;     (p/run! (partial synchronize-schedule-item conn) schedule)))
+    (run! (partial synchronize-schedule-item conn) schedule)))
 
-;; (def ^:private sql:lock-scheduled-task
+(def ^:private sql:lock-scheduled-task
-;;   "select id from scheduled_task where id=$1 for update skip locked")
+  "select id from scheduled_task where id=? for update skip locked")
 
-;; (declare schedule-task)
+(declare schedule-task!)
 
-;; (defn- log-scheduled-task-error
+(defn- log-scheduled-task-error
-;;   [item err]
+  [item err]
-;;   (log/error "Unhandled exception on scheduled task '" (:id item) "' \n"
+  (log/error "Unhandled exception on scheduled task '" (:id item) "' \n"
-;;              (with-out-str
+             (with-out-str
-;;                (.printStackTrace ^Throwable err (java.io.PrintWriter. *out*)))))
+               (.printStackTrace ^Throwable err (java.io.PrintWriter. *out*)))))
 
-;; (defn- execute-scheduled-task
+(defn- execute-scheduled-task
-;;   [{:keys [id cron] :as stask}]
+  [{:keys [id cron ::xtor] :as task}]
-;;   (db/with-atomic [conn db/pool]
+  (try
-;;     ;; First we try to lock the task in the database, if locking us
+    (db/with-atomic [conn db/pool]
-;;     ;; successful, then we execute the scheduled task; if locking is
+      ;; First we try to lock the task in the database, if locking is
-;;     ;; not possible (because other instance is already locked id) we
+      ;; successful, then we execute the scheduled task; if locking is
-;;     ;; just skip it and schedule to be executed in the next slot.
+      ;; not possible (because other instance is already locked id) we
-;;     (-> (db/query-one conn [sql:lock-scheduled-task id])
+      ;; just skip it and schedule to be executed in the next slot.
-;;         (p/then (fn [result]
+      (when (db/exec-one! conn [sql:lock-scheduled-task id])
-;;                   (when result
+        (log/info "Executing scheduled task" id)
-;;                     (-> (p/do! ((:fn stask) stask))
+        ((:fn task) task)))
-;;                         (p/catch (fn [e]
-;;                                    (log-scheduled-task-error stask e)
-;;                                    nil))))))
-;;         (p/finally (fn [v e]
-;;                      (-> (vu/current-context)
-;;                          (schedule-task stask)))))))
-;; (defn ms-until-valid
-;;   [cron]
-;;   (s/assert dt/cron? cron)
-;;   (let [^Instant now (dt/now)
-;;         ^Instant next (dt/next-valid-instant-from cron now)
-;;         ^Duration duration (Duration/between now next)]
-;;     (.toMillis duration)))
 
-;; (defn- schedule-task
+    (catch Throwable e
-;;   [ctx {:keys [cron] :as stask}]
+      (log-scheduled-task-error task e))
-;;   (let [ms (ms-until-valid cron)]
+    (finally
-;;     (vt/schedule! ctx (assoc stask
+      (schedule-task! xtor task))))
-;;                              :ctx ctx
-;;                              ::vt/once true
-;;                              ::vt/delay ms
-;;                              ::vt/fn execute-scheduled-task))))
 
-;; (defn- on-scheduler-start
+(defn ms-until-valid
-;;   [ctx {:keys [schedule] :as options}]
+  [cron]
-;;   (-> (synchronize-schedule schedule)
+  (s/assert dt/cron? cron)
-;;       (p/then' (fn [_]
+  (let [^Instant now  (dt/now)
-;;                  (run! #(schedule-task ctx %) schedule)))))
+        ^Instant next (dt/next-valid-instant-from cron now)]
+    (inst-ms (dt/duration-between now next))))
+
+(defn- schedule-task!
+  [xtor {:keys [cron] :as task}]
+  (let [ms    (ms-until-valid cron)
+        task  (assoc task ::xtor xtor)]
+    (px/schedule! xtor ms (partial execute-scheduled-task task))))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Public API
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
-;; --- Worker Verticle
+(s/def ::id string?)
+(s/def ::name string?)
+(s/def ::cron dt/cron?)
+(s/def ::fn (s/or :var var? :fn fn?))
+(s/def ::props (s/nilable map?))
+(s/def ::xtor #(instance? ScheduledExecutorService %))
 
-;; (s/def ::callable (s/or :fn fn? :var var?))
+(s/def ::scheduled-task
-;; (s/def ::max-batch-size ::us/integer)
+  (s/keys :req-un [::id ::cron ::fn]
-;; (s/def ::max-retries ::us/integer)
+          :opt-un [::props]))
-;; (s/def ::tasks (s/map-of string? ::callable))
 
-;; (s/def ::worker-verticle-options
+(s/def ::tasks (s/map-of string? ::fn))
-;;   (s/keys :req-un [::tasks]
+(s/def ::schedule (s/coll-of ::scheduled-task))
-;;           :opt-un [::queue ::max-batch-size]))
 
-;; (defn worker-verticle
+(defn start-scheduler-worker!
-;;   [options]
+  [{:keys [schedule xtor] :as opts}]
-;;   (s/assert ::worker-verticle-options options)
+  (us/assert ::xtor xtor)
-;;   (let [on-start #(on-worker-start % options)]
+  (us/assert ::schedule schedule)
-;;     (vc/verticle {:on-start on-start})))
+  (let [stop (atom false)]
+    (synchronize-schedule! schedule)
+    (run! (partial schedule-task! xtor) schedule)
+    (reify
+      java.lang.AutoCloseable
+      (close [_]
+        (reset! stop true)))))
 
-;; --- Scheduler Verticle
+(defn start-worker!
+  [{:keys [tasks xtor poll-interval]
+    :or {poll-interval 5000}
+    :as opts}]
+  (us/assert ::tasks tasks)
+  (us/assert ::xtor xtor)
+  (us/assert number? poll-interval)
+  (let [stop (atom false)
+        opts (assoc opts
+                    ::xtor xtor
+                    ::stop stop)]
+    (px/schedule! xtor poll-interval (partial execute-worker-task opts))
+    (reify
+      java.lang.AutoCloseable
+      (close [_]
+        (reset! stop true)))))
 
-;; (s/def ::id string?)
+(defn stop!
-;; (s/def ::cron dt/cron?)
+  [worker]
-;; (s/def ::fn ::callable)
+  (.close ^java.lang.AutoCloseable worker))
-;; (s/def ::props (s/nilable map?))
 
-;; (s/def ::scheduled-task
-;;   (s/keys :req-un [::id ::cron ::fn]
-;;           :opt-un [::props]))
 
-;; (s/def ::schedule (s/coll-of ::scheduled-task))
 
-;; (s/def ::scheduler-verticle-options
-;;   (s/keys :opt-un [::schedule]))
 
-;; (defn scheduler-verticle
-;;   [options]
-;;   (s/assert ::scheduler-verticle-options options)
-;;   (let [on-start #(on-scheduler-start % options)]
-;;     (vc/verticle {:on-start on-start})))
 
-;; --- Schedule API
+;; --- Submit API
 
 (s/def ::name ::us/string)
 (s/def ::delay
@@ -290,7 +267,12 @@
    values (?, ?, ?, ?, clock_timestamp()+cast(?::text as interval))
    returning id")
 
-(defn schedule!
+(defn- duration->pginterval
+  [^Duration d]
+  (->> (/ (.toMillis d) 1000.0)
+       (format "%s seconds")))
+
+(defn submit!
   [conn {:keys [name delay props queue key]
          :or {delay 0 props {} queue "default"}
          :as options}]
@@ -299,9 +281,7 @@
         pginterval (duration->pginterval duration)
         props      (blob/encode props)
         id         (uuid/next)]
-    (log/info "Schedule task" name
+    (log/info "Submit task" name "to be executed in" (str duration))
-              ;; "with props" (pr-str props)
-              "to be executed in" (str duration))
     (db/exec-one! conn [sql:insert-new-task
                         id name props queue pginterval])
     id))

backend/src/uxbox/tasks/remove_media.clj

@@ -15,6 +15,7 @@
    [uxbox.common.exceptions :as ex]
    [uxbox.common.spec :as us]
    [uxbox.media :as media]
+   [uxbox.metrics :as mtx]
    [uxbox.util.storage :as ust]))
 
 (s/def ::path ::us/not-empty-string)
@@ -28,3 +29,7 @@
     (ust/delete! media/media-storage (:path props))
     (log/debug "Media " (:path props) " removed.")))
 
+(mtx/instrument-with-summary!
+ {:var #'handler
+  :id "tasks__remove_media"
+  :help "Timing of remove-media task."})

backend/src/uxbox/tasks/sendmail.clj

@@ -15,6 +15,7 @@
    [uxbox.common.data :as d]
    [uxbox.common.exceptions :as ex]
    [uxbox.config :as cfg]
+   [uxbox.metrics :as mtx]
    [uxbox.util.http :as http]))
 
 (defmulti sendmail (fn [config email] (:sendmail-backend config)))
@@ -94,3 +95,7 @@
   [{:keys [props] :as task}]
   (sendmail cfg/config props))
 
+(mtx/instrument-with-summary!
+ {:var #'handler
+  :id "tasks__sendmail"
+  :help "Timing of sendmail task."})

backend/src/uxbox/util/dispatcher.clj

@@ -20,22 +20,18 @@
 (definterface IDispatcher
   (^void add [key f]))
 
-(defn- wrap-handler
+(deftype Dispatcher [reg attr wrap]
-  [items handler]
-  (reduce #(%2 %1) handler items))
-
-(deftype Dispatcher [reg attr wrap-fns]
   IDispatcher
   (add [this key f]
-    (let [f (wrap-handler wrap-fns f)]
+    (.put ^Map reg key (wrap f))
-      (.put ^Map reg key f)
+    this)
-      this))
+
 
   clojure.lang.IDeref
   (deref [_]
     {:registry reg
      :attr attr
-     :wrap-fns wrap-fns})
+     :wrap wrap})
 
   clojure.lang.IFn
   (invoke [_ params]
@@ -100,36 +96,3 @@
     `(do
        (s/assert dispatcher? ~sym)
        (add-method ~sym ~key ~f ~meta))))
-
-(defn wrap-spec
-  [handler]
-  (let [mdata (meta handler)
-        spec (s/get-spec (:spec mdata))]
-    (if (nil? spec)
-      handler
-      (with-meta
-        (fn [params]
-          (let [result (s/conform spec params)]
-            (if (not= result ::s/invalid)
-              (handler result)
-              (let [data (s/explain-data spec params)]
-                (ex/raise :type :validation
-                          :code :spec-validation
-                          :explain (with-out-str
-                                     (expound/printer data))
-                          :data (::s/problems data))))))
-        (assoc mdata ::wrap-spec true)))))
-
-(defn wrap-error
-  [handler]
-  (let [mdata (meta handler)]
-    (with-meta
-      (fn [params]
-        (try
-          (handler params)
-          (catch Throwable error
-            (ex/raise :type :service-error
-                      :name (:spec mdata)
-                      :cause error))))
-      (assoc mdata ::wrap-error true))))
-

backend/tests/user.clj

@@ -19,11 +19,9 @@
    [clojure.repl :refer :all]
    [criterium.core :refer [quick-bench bench with-progress-reporting]]
    [clj-kondo.core :as kondo]
-   [promesa.core :as p]
-   [promesa.exec :as px]
    [uxbox.migrations]
    [uxbox.db :as db]
-   ;; [uxbox.redis :as rd]
+   [uxbox.metrics :as mtx]
    [uxbox.util.storage :as st]
    [uxbox.util.time :as tm]
    [uxbox.util.blob :as blob]

docker/devenv/docker-compose.yaml

@@ -38,7 +38,6 @@ services:
       - 9090:9090
 
     environment:
-      - CLOJURE_OPTS=-J-XX:-OmitStackTraceInFastThrow
       - UXBOX_DATABASE_URI=postgresql://postgres/uxbox
       - UXBOX_DATABASE_USERNAME=uxbox
       - UXBOX_DATABASE_PASSWORD=uxbox

frontend/src/uxbox/main/data/workspace.cljs

@@ -197,7 +197,6 @@
 
 (defn initialize-viewport
   [{:keys [width height] :as size}]
-  (js/console.log "initialize-viewport" size)
   (ptk/reify ::initialize-viewport
     ptk/UpdateEvent
     (update [_ state]
@@ -208,11 +207,7 @@
                     (update :vbox (fn [vbox]
                                     (if (nil? vbox)
                                       (assoc size :x 0 :y 0)
-                                      vbox)))))))
+                                      vbox)))))))))
-
-    ptk/WatchEvent
-    (watch [_ state stream]
-      #_(rx/of zoom-to-fit-all))))
 
 (defn update-viewport-position
   [{:keys [x y] :or {x identity y identity}}]
@@ -852,7 +847,8 @@
 
             shapes (map lookup selected)
             shape? #(not= (:type %) :frame)]
-        (rx/of (delete-shapes selected))))))
+        (rx/of (delete-shapes selected)
+               deselect-all)))))
 
 
 ;; --- Rename Shape

frontend/src/uxbox/main/data/workspace/notifications.cljs

@@ -18,6 +18,8 @@
    [uxbox.main.repo :as rp]
    [uxbox.main.store :as st]
    [uxbox.main.streams :as ms]
+   [uxbox.main.data.workspace.common :as dwc]
+   [uxbox.main.data.workspace.persistence :as dwp]
    [uxbox.util.avatars :as avatars]
    [uxbox.util.geom.point :as gpt]
    [uxbox.util.time :as dt]
@@ -75,7 +77,6 @@
   (ptk/reify ::send-keepalive
     ptk/EffectEvent
     (effect [_ state stream]
-      (prn "send-keepalive" file-id)
       (when-let [ws (get-in state [:ws file-id])]
         (ws/-send ws (t/encode {:type :keepalive}))))))
 
@@ -165,13 +166,11 @@
         (ws/-send ws (t/encode msg))))))
 
 (defn handle-page-change
-  [{:keys [profile-id page-id revn operations] :as msg}]
+  [msg]
   (ptk/reify ::handle-page-change
     ptk/WatchEvent
     (watch [_ state stream]
-      #_(let [page-id' (get-in state [:workspace-page :id])]
+      (rx/of (dwp/shapes-changes-persisted msg)
-        (when (= page-id page-id')
+             (dwc/update-page-indices (:page-id msg))))))
-          (rx/of (shapes-changes-commited msg)))))))
-
 
 

frontend/src/uxbox/main/data/workspace/persistence.cljs

@@ -42,7 +42,7 @@
       (let [stoper (rx/filter #(= ::finalize %) stream)
             notifier (->> stream
                           (rx/filter (ptk/type? ::dwc/commit-changes))
-                          (rx/debounce 2000)
+                          (rx/debounce 200)
                           (rx/merge stoper))]
         (rx/merge
          (->> stream
@@ -64,15 +64,13 @@
   (ptk/reify ::persist-changes
     ptk/WatchEvent
     (watch [_ state stream]
-      (let [session-id (:session-id state)
+      (let [sid     (:session-id state)
-            page (get-in state [:workspace-pages page-id])
+            page    (get-in state [:workspace-pages page-id])
-            changes (->> changes
+            changes (into [] (mapcat identity) changes)
-                         (mapcat identity)
+            params  {:id (:id page)
-                         (map #(assoc % :session-id session-id))
+                     :revn (:revn page)
-                         (vec))
+                     :session-id sid
-            params {:id (:id page)
+                     :changes changes}]
-                    :revn (:revn page)
-                    :changes changes}]
         (->> (rp/mutation :update-page params)
              (rx/map shapes-changes-persisted))))))
 

frontend/src/uxbox/main/ui/workspace/shapes/text.cljs

@@ -260,15 +260,16 @@
         (fn [event]
           (mf/set-ref-val! selecting-ref false))
 
-        on-keyup
+        on-key-up
         (fn [event]
+          (dom/stop-propagation event)
           (when (= (.-keyCode event) 27) ; ESC
             (on-close)))
 
         on-mount
         (fn []
           (let [lkey1 (events/listen js/document EventType.CLICK on-click)
-                lkey2 (events/listen js/document EventType.KEYUP on-keyup)]
+                lkey2 (events/listen js/document EventType.KEYUP on-key-up)]
             (st/emit! (dwt/assign-editor id editor))
             #(do
                (st/emit! (dwt/assign-editor id nil))

frontend/src/uxbox/main/ui/workspace/viewport.cljs

@@ -33,6 +33,7 @@
    [uxbox.main.ui.workspace.snap-feedback :refer [snap-feedback]]
    [uxbox.util.math :as mth]
    [uxbox.util.dom :as dom]
+   [uxbox.util.object :as obj]
    [uxbox.util.geom.point :as gpt]
    [uxbox.util.perf :as perf]
    [uxbox.common.uuid :as uuid])
@@ -162,10 +163,7 @@
 
                (and (not edition)
                     (= 2 (.-which event)))
-               (handle-viewport-positioning viewport-ref)
+               (handle-viewport-positioning viewport-ref)))))
-
-               :else
-               (js/console.log "on-mouse-down" event)))))
 
         on-context-menu
         (mf/use-callback
@@ -234,10 +232,13 @@
                  shift? (kbd/shift? event)
                  opts {:key key
                        :shift? shift?
-                       :ctrl? ctrl?}]
+                       :ctrl? ctrl?}
+                 target (dom/get-target event)]
+
              (when-not (.-repeat bevent)
                (st/emit! (ms/->KeyboardEvent :down key ctrl? shift?))
-               (when (kbd/space? event)
+               (when (and (kbd/space? event)
+                          (not= "rich-text" (obj/get target "className")))
                  (handle-viewport-positioning viewport-ref))))))
 
         on-key-up