✨ Add improved activity logging.

backend/src/app/config.clj

@@ -8,6 +8,7 @@
   "A configuration management."
   (:refer-clojure :exclude [get])
   (:require
+   [app.common.data :as d]
    [app.common.spec :as us]
    [app.common.version :as v]
    [app.util.time :as dt]
@@ -16,7 +17,8 @@
    [clojure.pprint :as pprint]
    [clojure.spec.alpha :as s]
    [cuerdas.core :as str]
-   [environ.core :refer [env]]))
+   [environ.core :refer [env]]
+   [integrant.core :as ig]))
 
 (prefer-method print-method
                clojure.lang.IRecord
@@ -26,6 +28,16 @@
                clojure.lang.IPersistentMap
                clojure.lang.IDeref)
 
+(defmethod ig/init-key :default
+  [_ data]
+  (d/without-nils data))
+
+(defmethod ig/prep-key :default
+  [_ data]
+  (if (map? data)
+    (d/without-nils data)
+    data))
+
 (def defaults
   {:http-server-port 6060
    :host "devenv"

backend/src/app/http/oauth.clj

@@ -6,6 +6,7 @@
 
 (ns app.http.oauth
   (:require
+   [app.common.data :as d]
    [app.common.exceptions :as ex]
    [app.common.spec :as us]
    [app.common.uri :as u]
@@ -138,16 +139,32 @@
 
     (cond-> info
       (some? (:invitation-token state))
-      (assoc :invitation-token (:invitation-token state)))))
+      (assoc :invitation-token (:invitation-token state))
+
+      (map? (:props state))
+      (d/merge (:props state)))))
 
 ;; --- HTTP HANDLERS
 
+(defn extract-props
+  [params]
+  (reduce-kv (fn [params k v]
+               (let [sk (name k)]
+                 (cond-> params
+                   (or (str/starts-with? sk "pm_")
+                       (str/starts-with? sk "pm-"))
+                   (assoc (-> sk str/kebab keyword) v))))
+             {}
+             params))
+
 (defn- auth-handler
-  [{:keys [tokens] :as cfg} request]
+  [{:keys [tokens] :as cfg} {:keys [params] :as request}]
-  (let [invitation (get-in request [:params :invitation-token])
+  (let [invitation (:invitation-token params)
+        props      (extract-props params)
         state      (tokens :generate
                            {:iss :oauth
                             :invitation-token invitation
+                            :props props
                             :exp (dt/in-future "15m")})
         uri        (build-auth-uri cfg state)]
     {:status 200

backend/src/app/loggers/activity.clj

@@ -0,0 +1,127 @@
+;; 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) UXBOX Labs SL
+
+(ns app.loggers.activity
+  "Activity registry logger consumer."
+  (:require
+   [app.common.data :as d]
+   [app.common.spec :as us]
+   [app.config :as cf]
+   [app.util.async :as aa]
+   [app.util.http :as http]
+   [app.util.logging :as l]
+   [app.util.time :as dt]
+   [app.util.transit :as t]
+   [app.worker :as wrk]
+   [clojure.core.async :as a]
+   [clojure.spec.alpha :as s]
+   [integrant.core :as ig]
+   [lambdaisland.uri :as u]))
+
+(declare process-event)
+(declare handle-event)
+
+(s/def ::uri ::us/string)
+
+(defmethod ig/pre-init-spec ::reporter [_]
+  (s/keys :req-un [::wrk/executor]
+          :opt-un [::uri]))
+
+(defmethod ig/init-key ::reporter
+  [_ {:keys [uri] :as cfg}]
+  (if (string? uri)
+    (do
+      (l/info :msg "intializing activity reporter" :uri uri)
+      (let [xform    (comp (map process-event)
+                           (filter map?))
+            input    (a/chan (a/sliding-buffer 1024) xform)]
+        (a/go-loop []
+          (when-let [event (a/<! input)]
+            (a/<! (handle-event cfg event))
+            (recur)))
+
+        (fn [& [cmd & params]]
+          (case cmd
+            :stop (a/close! input)
+            :submit (when-not (a/offer! input (first params))
+                      (l/warn :msg "activity channel is full"))))))
+    (constantly nil)))
+
+(defmethod ig/halt-key! ::reporter
+  [_ f]
+  (f :stop))
+
+(defn- clean-params
+  "Cleans the params from complex data, only accept strings, numbers and
+  uuids and removing sensitive data such as :password and related
+  props."
+  [params]
+  (let [params (dissoc params :profile-id :session-id :password :old-password)]
+    (reduce-kv (fn [params k v]
+                 (cond-> params
+                   (or (string? v)
+                       (uuid? v)
+                       (number? v))
+                   (assoc k v)))
+               {}
+               params)))
+
+(defn- process-event
+  [{:keys [type name params result] :as event}]
+  (let [profile-id (:profile-id params)]
+    (if (uuid? profile-id)
+      {:type (str "backend:" (d/name type))
+       :name name
+       :timestamp (dt/now)
+       :profile-id profile-id
+       :props (clean-params params)}
+      (cond
+        (= "register-profile" name)
+        {:type (str "backend:" (d/name type))
+         :name name
+         :timestamp (dt/now)
+         :profile-id (:id result)
+         :props (clean-params (:props result))}
+
+        :else nil))))
+
+(defn- send-activity
+  [{:keys [uri tokens]} event i]
+  (try
+    (let [token    (tokens :generate {:iss "authentication"
+                                      :iat (dt/now)
+                                      :uid (:profile-id event)})
+          body     (t/encode {:events [event]})
+          headers  {"content-type" "application/transit+json"
+                    "origin" (cf/get :public-uri)
+                    "cookie" (u/map->query-string {:auth-token token})}
+          params   {:uri uri
+                    :timeout 6000
+                    :method :post
+                    :headers headers
+                    :body body}
+          response (http/send! params)]
+      (if (= (:status response) 204)
+        true
+        (do
+          (l/error :hint "error on sending activity"
+                   :try i
+                   :rsp (pr-str response))
+          false)))
+    (catch Exception e
+      (l/error :hint "error on sending message to loki"
+               :cause e
+               :try i)
+      false)))
+
+(defn- handle-event
+  [{:keys [executor] :as cfg} event]
+  (aa/with-thread executor
+    (loop [i 1]
+      (when (and (not (send-activity cfg event i)) (< i 20))
+        (Thread/sleep (* i 2000))
+        (recur (inc i))))))
+

backend/src/app/loggers/loki.clj

@@ -31,16 +31,16 @@
   [_ {:keys [receiver uri] :as cfg}]
   (when uri
     (l/info :msg "intializing loki reporter" :uri uri)
-    (let [output (a/chan (a/sliding-buffer 1024))]
+    (let [input (a/chan (a/sliding-buffer 1024))]
-      (receiver :sub output)
+      (receiver :sub input)
       (a/go-loop []
-        (let [msg (a/<! output)]
+        (let [msg (a/<! input)]
           (if (nil? msg)
             (l/info :msg "stoping error reporting loop")
             (do
               (a/<! (handle-event cfg msg))
               (recur)))))
-      output)))
+      input)))
 
 (defmethod ig/halt-key! ::reporter
   [_ output]

backend/src/app/main.clj

@@ -6,7 +6,6 @@
 
 (ns app.main
   (:require
-   [app.common.data :as d]
    [app.config :as cf]
    [app.util.logging :as l]
    [app.util.time :as dt]
@@ -140,7 +139,8 @@
     :storage    (ig/ref :app.storage/storage)
     :msgbus     (ig/ref :app.msgbus/msgbus)
     :rlimits    (ig/ref :app.rlimits/all)
-    :public-uri (cf/get :public-uri)}
+    :public-uri (cf/get :public-uri)
+    :activity   (ig/ref :app.loggers.activity/reporter)}
 
    :app.notifications/handler
    {:msgbus   (ig/ref :app.msgbus/msgbus)
@@ -263,6 +263,11 @@
    :app.loggers.zmq/receiver
    {:endpoint (cf/get :loggers-zmq-uri)}
 
+   :app.loggers.activity/reporter
+   {:uri      (cf/get :activity-reporter-uri)
+    :tokens   (ig/ref :app.tokens/tokens)
+    :executor (ig/ref :app.worker/executor)}
+
    :app.loggers.loki/reporter
    {:uri      (cf/get :loggers-loki-uri)
     :receiver (ig/ref :app.loggers.zmq/receiver)
@@ -299,13 +304,6 @@
    [::main :app.storage.db/backend]
    {:pool (ig/ref :app.db/pool)}})
 
-(defmethod ig/init-key :default [_ data] data)
-(defmethod ig/prep-key :default
-  [_ data]
-  (if (map? data)
-    (d/without-nils data)
-    data))
-
 (def system nil)
 
 (defn start

backend/src/app/rpc.clj

@@ -84,19 +84,31 @@
         (rlm/execute rlinst (f cfg params))))
     f))
 
+
 (defn- wrap-impl
-  [cfg f mdata]
+  [{:keys [activity] :as cfg} f mdata]
   (let [f      (wrap-with-rlimits cfg f mdata)
         f      (wrap-with-metrics cfg f mdata)
-        spec  (or (::sv/spec mdata) (s/spec any?))]
+        spec   (or (::sv/spec mdata) (s/spec any?))
+        auth?  (:auth mdata true)]
     (l/trace :action "register"
              :name (::sv/name mdata))
     (fn [params]
-      (when (and (:auth mdata true) (not (uuid? (:profile-id params))))
+      (when (and auth? (not (uuid? (:profile-id params))))
         (ex/raise :type :authentication
                   :code :authentication-required
                   :hint "authentication required for this endpoint"))
-      (f cfg (us/conform spec params)))))
+      (let [params  (us/conform spec params)
+            result  (f cfg params)
+            ;; On non authenticated handlers we check the private
+            ;; result that can be found on the metadata.
+            result* (if auth? result (:result (meta result) {}))]
+        (when (::type cfg)
+          (activity :submit {:type (::type cfg)
+                             :name (::sv/name mdata)
+                             :params params
+                             :result result*}))
+        result))))
 
 (defn- process-method
   [cfg vfn]
@@ -133,7 +145,7 @@
                :registry (get-in cfg [:metrics :registry])
                :type :histogram
                :help "Timing of mutation services."})
-        cfg  (assoc cfg ::mobj mobj)]
+        cfg  (assoc cfg ::mobj mobj ::type :mutation)]
     (->> (sv/scan-ns 'app.rpc.mutations.demo
                      'app.rpc.mutations.media
                      'app.rpc.mutations.profile
@@ -152,9 +164,11 @@
 (s/def ::storage some?)
 (s/def ::session map?)
 (s/def ::tokens fn?)
+(s/def ::activity some?)
 
 (defmethod ig/pre-init-spec ::rpc [_]
-  (s/keys :req-un [::db/pool ::storage ::session ::tokens ::mtx/metrics ::rlm/rlimits]))
+  (s/keys :req-un [::storage ::session ::tokens ::activity
+                   ::mtx/metrics ::rlm/rlimits ::db/pool]))
 
 (defmethod ig/init-key ::rpc
   [_ cfg]

backend/src/app/rpc/mutations/profile.clj

@@ -13,6 +13,7 @@
    [app.config :as cfg]
    [app.db :as db]
    [app.emails :as eml]
+   [app.http.oauth :refer [extract-props]]
    [app.media :as media]
    [app.rpc.mutations.projects :as projects]
    [app.rpc.mutations.teams :as teams]
@@ -101,7 +102,8 @@
             resp   {:invitation-token token}]
         (with-meta resp
           {:transform-response ((:create session) (:id profile))
-           :before-complete (annotate-profile-register metrics profile)}))
+           :before-complete (annotate-profile-register metrics profile)
+           :result profile}))
 
       ;; If no token is provided, send a verification email
       (let [vtoken (tokens :generate
@@ -129,7 +131,8 @@
                     :extra-data ptoken})
 
         (with-meta profile
-          {:before-complete (annotate-profile-register metrics profile)})))))
+          {:before-complete (annotate-profile-register metrics profile)
+           :result profile})))))
 
 (defn email-domain-in-whitelist?
   "Returns true if email's domain is in the given whitelist or if given
@@ -174,11 +177,12 @@
 (defn create-profile
   "Create the profile entry on the database with limited input
   filling all the other fields with defaults."
-  [conn {:keys [id fullname email password props is-active is-muted is-demo opts]
+  [conn {:keys [id fullname email password is-active is-muted is-demo opts]
-         :or {is-active false is-muted false is-demo false}}]
+         :or {is-active false is-muted false is-demo false}
+         :as params}]
   (let [id        (or id (uuid/next))
         is-active (if is-demo true is-active)
-        props     (db/tjson (or props {}))
+        props     (-> params extract-props db/tjson)
         password  (derive-password password)
         params    {:id id
                    :fullname fullname

backend/src/app/rpc/queries/files.clj

@@ -6,7 +6,6 @@
 
 (ns app.rpc.queries.files
   (:require
-   [app.common.exceptions :as ex]
    [app.common.pages.migrations :as pmg]
    [app.common.spec :as us]
    [app.db :as db]
@@ -254,7 +253,7 @@
 (defn retrieve-file-libraries
   [conn is-indirect file-id]
   (let [libraries (->> (db/exec! conn [sql:file-libraries file-id])
-                       (map #(assoc :is-indirect is-indirect))
+                       (map #(assoc % :is-indirect is-indirect))
                        (into #{} decode-row-xf))]
     (reduce #(into %1 (retrieve-file-libraries conn true %2))
             libraries

frontend/src/app/main/data/events.cljs

@@ -179,7 +179,6 @@
     (effect [_ state stream]
       (let [profile-id (:profile-id state)
             events     (into [] (take max-buffer-size) @buffer)]
-        (prn ::persistence (count events) profile-id)
         (when (seq events)
           (->> events
                (filterv #(= profile-id (:profile-id %)))

frontend/src/app/main/ui/auth.cljs

@@ -54,8 +54,7 @@
         [:& recovery-request-page {:locale locale}]
 
         :auth-recovery
-        [:& recovery-page {:locale locale
+        [:& recovery-page {:locale locale :params params}])
-                           :params (:query-params route)}])
       [:div.terms-login
        [:a {:href "https://penpot.app/terms.html" :target "_blank"} "Terms of service"]
        [:span "and"]