3pp-mirror/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2025-05-02 00:06:20 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #4795 from penpot/niwinz-bugfix-8

Browse source 
🐛 Bugfixes
This commit is contained in:
Alejandro 2024-06-26 07:41:22 +02:00 • committed by GitHub
commit 3bb2573dec
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
20 changed files with 180 additions and 168 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
backend/src/app/email.clj
View file

@ -262,13 +262,12 @@
(let [email (if factory (let [email (if factory
(factory context) (factory context)
(dissoc context ::conn))] (dissoc context ::conn))]
(wrk/submit! (merge (wrk/submit! {::wrk/task :sendmail
{::wrk/task :sendmail ::wrk/delay 0
::wrk/delay 0 ::wrk/max-retries 4
::wrk/max-retries 4 ::wrk/priority 200
::wrk/priority 200 ::db/conn conn
::wrk/conn conn} ::wrk/params email})))
email))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; SENDMAIL FN / TASK HANDLER ;; SENDMAIL FN / TASK HANDLER

22
backend/src/app/loggers/audit.clj
View file

@ -241,18 +241,16 @@
:else label) :else label)
dedupe? (boolean (and batch-key batch-timeout))] dedupe? (boolean (and batch-key batch-timeout))]
(wrk/submit! ::wrk/conn (::db/conn cfg) (wrk/submit! (-> cfg
::wrk/task :process-webhook-event (assoc ::wrk/task :process-webhook-event)
::wrk/queue :webhooks (assoc ::wrk/queue :webhooks)
::wrk/max-retries 0 (assoc ::wrk/max-retries 0)
::wrk/delay (or batch-timeout 0) (assoc ::wrk/delay (or batch-timeout 0))
::wrk/dedupe dedupe? (assoc ::wrk/dedupe dedupe?)
::wrk/label label (assoc ::wrk/label label)
(assoc ::wrk/params (-> params
::webhooks/event (dissoc :ip-addr)
(-> params (dissoc :type)))))))
(dissoc :ip-addr)
(dissoc :type)))))
params)) params))
(defn submit! (defn submit!

24
backend/src/app/loggers/webhooks.clj
View file

@ -64,22 +64,22 @@
(s/keys :req [::db/pool])) (s/keys :req [::db/pool]))
(defmethod ig/init-key ::process-event-handler (defmethod ig/init-key ::process-event-handler
[_ {:keys [::db/pool] :as cfg}] [_ cfg]
(fn [{:keys [props] :as task}] (fn [{:keys [props] :as task}]
(let [event (::event props)] (let [event (:event props)]
(l/dbg :hint "process webhook event" :name (:name event)) (l/dbg :hint "process webhook event" :name (:name event))
(when-let [items (lookup-webhooks cfg event)] (when-let [items (lookup-webhooks cfg event)]
(l/trc :hint "webhooks found for event" :total (count items)) (l/trc :hint "webhooks found for event" :total (count items))
(db/with-atomic [conn pool] (db/tx-run! cfg (fn [cfg]
(doseq [item items] (doseq [item items]
(wrk/submit! ::wrk/conn conn (wrk/submit! (-> cfg
::wrk/task :run-webhook (assoc ::wrk/task :run-webhook)
::wrk/queue :webhooks (assoc ::wrk/queue :webhooks)
::wrk/max-retries 3 (assoc ::wrk/max-retries 3)
::event event (assoc ::wrk/params {:event event
::config item))))))) :config item}))))))))))
;; --- RUN ;; --- RUN
@ -128,8 +128,8 @@
:rsp-data (db/tjson rsp)}))] :rsp-data (db/tjson rsp)}))]
(fn [{:keys [props] :as task}] (fn [{:keys [props] :as task}]
(let [event (::event props) (let [event (:event props)
whook (::config props) whook (:config props)
body (case (:mtype whook) body (case (:mtype whook)
"application/json" (json/write-str event json-write-opts) "application/json" (json/write-str event json-write-opts)

10
backend/src/app/rpc/commands/files.clj
View file

@ -927,11 +927,11 @@
{:id file-id} {:id file-id}
{::db/return-keys [:id :name :is-shared :deleted-at {::db/return-keys [:id :name :is-shared :deleted-at
:project-id :created-at :modified-at]})] :project-id :created-at :modified-at]})]
(wrk/submit! {::wrk/task :delete-object (wrk/submit! {::db/conn conn
::wrk/conn conn ::wrk/task :delete-object
:object :file ::wrk/params {:object :file
:deleted-at (:deleted-at file) :deleted-at (:deleted-at file)
:id file-id}) :id file-id}})
file)) file))
(def ^:private (def ^:private

10
backend/src/app/rpc/commands/projects.clj
View file

@ -258,11 +258,11 @@
:code :non-deletable-project :code :non-deletable-project
:hint "impossible to delete default project")) :hint "impossible to delete default project"))
(wrk/submit! {::wrk/task :delete-object (wrk/submit! {::db/conn conn
::wrk/conn conn ::wrk/task :delete-object
:object :project ::wrk/params {:object :project
:deleted-at (:deleted-at project) :deleted-at (:deleted-at project)
:id project-id}) :id project-id}})
project)) project))

10
backend/src/app/rpc/commands/teams.clj
View file

@ -527,11 +527,11 @@
:code :non-deletable-team :code :non-deletable-team
:hint "impossible to delete default team")) :hint "impossible to delete default team"))
(wrk/submit! {::wrk/task :delete-object (wrk/submit! {::db/conn conn
::wrk/conn conn ::wrk/task :delete-object
:object :team ::wrk/params {:object :team
:deleted-at deleted-at :deleted-at deleted-at
:id team-id}) :id team-id}})
team)) team))
(def ^:private schema:delete-team (def ^:private schema:delete-team

12
backend/src/app/rpc/quotes.clj
View file

@ -83,17 +83,17 @@
"- Quote ID: '~(::target params)'\n" "- Quote ID: '~(::target params)'\n"
"- Max: ~(::quote params)\n" "- Max: ~(::quote params)\n"
"- Total: ~(::total params) (INCR ~(::incr params 1))\n")] "- Total: ~(::total params) (INCR ~(::incr params 1))\n")]
(wrk/submit! {::wrk/task :sendmail (wrk/submit! {::db/conn conn
::wrk/task :sendmail
::wrk/delay (dt/duration "30s") ::wrk/delay (dt/duration "30s")
::wrk/max-retries 4 ::wrk/max-retries 4
::wrk/priority 200 ::wrk/priority 200
::wrk/conn conn
::wrk/dedupe true ::wrk/dedupe true
::wrk/label "quotes-notification" ::wrk/label "quotes-notification"
:to (vec admins) ::wrk/params {:to (vec admins)
:subject subject :subject subject
:body [{:type "text/plain" :body [{:type "text/plain"
:content content}]})))) :content content}]}}))))
(defn- generic-check! (defn- generic-check!
[{:keys [::db/conn ::incr ::quote-sql ::count-sql ::default ::target] :or {incr 1} :as params}] [{:keys [::db/conn ::incr ::quote-sql ::count-sql ::default ::target] :or {incr 1} :as params}]

55
backend/src/app/srepl/main.clj
View file

@ -59,32 +59,27 @@
([tname] ([tname]
(run-task! tname {})) (run-task! tname {}))
([tname params] ([tname params]
(let [tasks (:app.worker/registry main/system) (wrk/invoke! (-> main/system
tname (if (keyword? tname) (name tname) name)] (assoc ::wrk/task tname)
(if-let [task-fn (get tasks tname)] (assoc ::wrk/params params)))))
(task-fn params)
(println (format "no task '%s' found" tname))))))
(defn schedule-task! (defn schedule-task!
([name] ([name]
(schedule-task! name {})) (schedule-task! name {}))
([name props] ([name params]
(let [pool (:app.db/pool main/system)] (wrk/submit! (-> main/system
(wrk/submit! (assoc ::wrk/task name)
::wrk/conn pool (assoc ::wrk/params params)))))
::wrk/task name
::wrk/props props))))
(defn send-test-email! (defn send-test-email!
[destination] [destination]
(us/verify! (assert (string? destination) "destination should be provided")
:expr (string? destination) (-> main/system
:hint "destination should be provided") (assoc ::wrk/task :sendmail)
(assoc ::wrk/params {:body "test email"
(let [handler (:app.email/sendmail main/system)] :subject "test email"
(handler {:body "test email" :to [destination]})
:subject "test email" (wrk/invoke!)))
:to [destination]})))
(defn resend-email-verification-email! (defn resend-email-verification-email!
[email] [email]
@ -562,22 +557,30 @@
"Mark a team for deletion" "Mark a team for deletion"
[team-id] [team-id]
(let [team-id (h/parse-uuid team-id)] (let [team-id (h/parse-uuid team-id)]
(db/tx-run! main/system (fn [{:keys [::db/conn]}] (wrk/invoke! (-> main/system
(#'teams/delete-team conn team-id))))) (assoc ::wrk/task :delete-object)
(assoc ::wrk/params {:object :team
:deleted-at (dt/now)
:id team-id})))))
(defn delete-project! (defn delete-project!
"Mark a project for deletion" "Mark a project for deletion"
[project-id] [project-id]
(let [project-id (h/parse-uuid project-id)] (let [project-id (h/parse-uuid project-id)]
(db/tx-run! main/system (fn [{:keys [::db/conn]}] (wrk/invoke! (-> main/system
(#'projects/delete-project conn project-id))))) (assoc ::wrk/task :delete-object)
(assoc ::wrk/params {:object :project
:deleted-at (dt/now)
:id project-id})))))
(defn delete-file! (defn delete-file!
"Mark a project for deletion" "Mark a project for deletion"
[file-id] [file-id]
(let [file-id (h/parse-uuid file-id)] (let [file-id (h/parse-uuid file-id)]
(db/tx-run! main/system (fn [{:keys [::db/conn]}] (wrk/invoke! (-> main/system
(#'files/mark-file-deleted conn file-id))))) (assoc ::wrk/task :delete-object)
(assoc ::wrk/params {:object :file
:deleted-at (dt/now)
:id file-id})))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; MISC ;; MISC

4
backend/src/app/storage/gc_deleted.clj
View file

@ -110,8 +110,8 @@
(defmethod ig/init-key ::handler (defmethod ig/init-key ::handler
[_ {:keys [::min-age] :as cfg}] [_ {:keys [::min-age] :as cfg}]
(fn [params] (fn [{:keys [props] :as task}]
(let [min-age (dt/duration (or (:min-age params) min-age))] (let [min-age (dt/duration (or (:min-age props) min-age))]
(db/tx-run! cfg (fn [cfg] (db/tx-run! cfg (fn [cfg]
(let [cfg (assoc cfg ::min-age min-age) (let [cfg (assoc cfg ::min-age min-age)
total (clean-deleted! cfg)] total (clean-deleted! cfg)]

47
backend/src/app/tasks/delete_object.clj
View file

@ -20,8 +20,13 @@
(defmethod delete-object :file (defmethod delete-object :file
[{:keys [::db/conn] :as cfg} {:keys [id deleted-at]}] [{:keys [::db/conn] :as cfg} {:keys [id deleted-at]}]
(l/trc :hint "marking for deletion" :rel "file" :id (str id))
(when-let [file (db/get* conn :file {:id id} {::db/remove-deleted false})] (when-let [file (db/get* conn :file {:id id} {::db/remove-deleted false})]
(l/trc :hint "marking for deletion" :rel "file" :id (str id))
(db/update! conn :file
{:deleted-at deleted-at}
{:id id}
{::db/return-keys false})
(when (and (:is-shared file) (when (and (:is-shared file)
(not *team-deletion*)) (not *team-deletion*))
;; NOTE: we don't prevent file deletion on absorb operation failure ;; NOTE: we don't prevent file deletion on absorb operation failure
@ -49,27 +54,39 @@
(defmethod delete-object :project (defmethod delete-object :project
[{:keys [::db/conn] :as cfg} {:keys [id deleted-at]}] [{:keys [::db/conn] :as cfg} {:keys [id deleted-at]}]
(l/trc :hint "marking for deletion" :rel "project" :id (str id)) (l/trc :hint "marking for deletion" :rel "project" :id (str id))
(doseq [file (db/update! conn :file (db/update! conn :project
{:deleted-at deleted-at} {:deleted-at deleted-at}
{:project-id id} {:id id}
{::db/return-keys [:id :deleted-at] {::db/return-keys false})
::db/many true})]
(delete-object cfg (assoc file :object :file)))) (doseq [file (db/query conn :file
{:project-id id}
{::db/columns [:id :deleted-at]})]
(delete-object cfg (assoc file
:object :file
:deleted-at deleted-at))))
(defmethod delete-object :team (defmethod delete-object :team
[{:keys [::db/conn] :as cfg} {:keys [id deleted-at]}] [{:keys [::db/conn] :as cfg} {:keys [id deleted-at]}]
(l/trc :hint "marking for deletion" :rel "team" :id (str id)) (l/trc :hint "marking for deletion" :rel "team" :id (str id))
(db/update! conn :team
{:deleted-at deleted-at}
{:id id}
{::db/return-keys false})
(db/update! conn :team-font-variant (db/update! conn :team-font-variant
{:deleted-at deleted-at} {:deleted-at deleted-at}
{:team-id id}) {:team-id id}
{::db/return-keys false})
(binding [*team-deletion* true] (binding [*team-deletion* true]
(doseq [project (db/update! conn :project (doseq [project (db/query conn :project
{:deleted-at deleted-at} {:team-id id}
{:team-id id} {::db/columns [:id :deleted-at]})]
{::db/return-keys [:id :deleted-at] (delete-object cfg (assoc project
::db/many true})] :object :project
(delete-object cfg (assoc project :object :project))))) :deleted-at deleted-at)))))
(defmethod delete-object :default (defmethod delete-object :default
[_cfg props] [_cfg props]
@ -80,5 +97,5 @@
(defmethod ig/init-key ::handler (defmethod ig/init-key ::handler
[_ cfg] [_ cfg]
(fn [{:keys [props] :as params}] (fn [{:keys [props] :as task}]
(db/tx-run! cfg delete-object props))) (db/tx-run! cfg delete-object props)))

8
backend/src/app/tasks/file_gc.clj
View file

@ -299,13 +299,13 @@
(defmethod ig/init-key ::handler (defmethod ig/init-key ::handler
[_ cfg] [_ cfg]
(fn [{:keys [file-id] :as params}] (fn [{:keys [props] :as task}]
(db/tx-run! cfg (db/tx-run! cfg
(fn [{:keys [::db/conn] :as cfg}] (fn [{:keys [::db/conn] :as cfg}]
(let [min-age (dt/duration (or (:min-age params) (::min-age cfg))) (let [min-age (dt/duration (or (:min-age props) (::min-age cfg)))
cfg (-> cfg cfg (-> cfg
(update ::sto/storage media/configure-assets-storage conn) (update ::sto/storage media/configure-assets-storage conn)
(assoc ::file-id file-id) (assoc ::file-id (:file-id props))
(assoc ::min-age min-age)) (assoc ::min-age min-age))
total (reduce (fn [total file] total (reduce (fn [total file]
@ -319,7 +319,7 @@
:processed total) :processed total)
;; Allow optional rollback passed by params ;; Allow optional rollback passed by params
(when (:rollback? params) (when (:rollback? props)
(db/rollback! conn)) (db/rollback! conn))
{:processed total}))))) {:processed total})))))

6
backend/src/app/tasks/file_xlog_gc.clj
View file

@ -29,8 +29,8 @@
(defmethod ig/init-key ::handler (defmethod ig/init-key ::handler
[_ {:keys [::db/pool] :as cfg}] [_ {:keys [::db/pool] :as cfg}]
(fn [params] (fn [{:keys [props] :as task}]
(let [min-age (or (:min-age params) (::min-age cfg))] (let [min-age (or (:min-age props) (::min-age cfg))]
(db/with-atomic [conn pool] (db/with-atomic [conn pool]
(let [interval (db/interval min-age) (let [interval (db/interval min-age)
result (db/exec-one! conn [sql:delete-files-xlog interval]) result (db/exec-one! conn [sql:delete-files-xlog interval])
@ -38,7 +38,7 @@
(l/info :hint "task finished" :min-age (dt/format-duration min-age) :total result) (l/info :hint "task finished" :min-age (dt/format-duration min-age) :total result)
(when (:rollback? params) (when (:rollback? props)
(db/rollback! conn)) (db/rollback! conn))
result))))) result)))))

4
backend/src/app/tasks/objects_gc.clj
View file

@ -302,8 +302,8 @@
(defmethod ig/init-key ::handler (defmethod ig/init-key ::handler
[_ cfg] [_ cfg]
(fn [params] (fn [{:keys [props] :as task}]
(let [min-age (dt/duration (or (:min-age params) (::min-age cfg))) (let [min-age (dt/duration (or (:min-age props) (::min-age cfg)))
cfg (-> cfg cfg (-> cfg
(assoc ::min-age (db/interval min-age)) (assoc ::min-age (db/interval min-age))
(update ::sto/storage media/configure-assets-storage))] (update ::sto/storage media/configure-assets-storage))]

19
backend/src/app/tasks/orphan_teams_gc.clj
View file

@ -39,12 +39,11 @@
{:deleted-at deleted-at} {:deleted-at deleted-at}
{:id team-id}) {:id team-id})
(wrk/submit! {::wrk/task :delete-object (wrk/submit! (-> cfg
::wrk/conn conn (assoc ::wrk/task :delete-object)
:object :team (assoc ::wrk/params {:object :team
:deleted-at deleted-at :deleted-at deleted-at
:id team-id}) :id team-id})))
(inc total)) (inc total))
0)))) 0))))
@ -53,15 +52,15 @@
(defmethod ig/init-key ::handler (defmethod ig/init-key ::handler
[_ cfg] [_ cfg]
(fn [params] (fn [{:keys [props] :as task}]
(db/tx-run! cfg (fn [{:keys [::db/conn] :as cfg}] (db/tx-run! cfg (fn [{:keys [::db/conn] :as cfg}]
(l/inf :hint "gc started" :rollback? (boolean (:rollback? params))) (l/inf :hint "gc started" :rollback? (boolean (:rollback? props)))
(let [total (delete-orphan-teams cfg)] (let [total (delete-orphan-teams cfg)]
(l/inf :hint "task finished" (l/inf :hint "task finished"
:teams total :teams total
:rollback? (boolean (:rollback? params))) :rollback? (boolean (:rollback? props)))
(when (:rollback? params) (when (:rollback? props)
(db/rollback! conn)) (db/rollback! conn))
{:processed total}))))) {:processed total})))))

6
backend/src/app/tasks/tasks_gc.clj
View file

@ -27,8 +27,8 @@
(defmethod ig/init-key ::handler (defmethod ig/init-key ::handler
[_ {:keys [::db/pool ::min-age] :as cfg}] [_ {:keys [::db/pool ::min-age] :as cfg}]
(fn [params] (fn [{:keys [props] :as task}]
(let [min-age (or (:min-age params) min-age)] (let [min-age (or (:min-age props) min-age)]
(db/with-atomic [conn pool] (db/with-atomic [conn pool]
(let [interval (db/interval min-age) (let [interval (db/interval min-age)
result (db/exec-one! conn [sql:delete-completed-tasks interval]) result (db/exec-one! conn [sql:delete-completed-tasks interval])
@ -36,7 +36,7 @@
(l/debug :hint "task finished" :total result) (l/debug :hint "task finished" :total result)
(when (:rollback? params) (when (:rollback? props)
(db/rollback! conn)) (db/rollback! conn))
result))))) result)))))

12
backend/src/app/tasks/telemetry.clj
View file

@ -206,14 +206,16 @@
(defmethod ig/init-key ::handler (defmethod ig/init-key ::handler
[_ {:keys [::db/pool ::setup/props] :as cfg}] [_ {:keys [::db/pool ::setup/props] :as cfg}]
(fn [{:keys [send? enabled?] :or {send? true enabled? false}}] (fn [task]
(let [subs {:newsletter-updates (get-subscriptions-newsletter-updates pool) (let [params (:props task)
:newsletter-news (get-subscriptions-newsletter-news pool)} send? (get params :send? true)
enabled? (or (get params :enabled? false)
enabled? (or enabled?
(contains? cf/flags :telemetry) (contains? cf/flags :telemetry)
(cf/get :telemetry-enabled)) (cf/get :telemetry-enabled))
subs {:newsletter-updates (get-subscriptions-newsletter-updates pool)
:newsletter-news (get-subscriptions-newsletter-news pool)}
data {:subscriptions subs data {:subscriptions subs
:version (:full cf/version) :version (:full cf/version)
:instance-id (:instance-id props)}] :instance-id (:instance-id props)}]

29
backend/src/app/worker.clj
View file

@ -8,6 +8,7 @@
"Async tasks abstraction (impl)." "Async tasks abstraction (impl)."
(:require (:require
[app.common.data :as d] [app.common.data :as d]
[app.common.data.macros :as dm]
[app.common.logging :as l] [app.common.logging :as l]
[app.common.spec :as us] [app.common.spec :as us]
[app.common.uuid :as uuid] [app.common.uuid :as uuid]
@ -58,17 +59,6 @@
;; SUBMIT API ;; SUBMIT API
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(defn- extract-props
[options]
(let [cns (namespace ::sample)]
(persistent!
(reduce-kv (fn [res k v]
(cond-> res
(not= (namespace k) cns)
(assoc! k v)))
(transient {})
options))))
(def ^:private sql:insert-new-task (def ^:private sql:insert-new-task
"insert into task (id, name, props, queue, label, priority, max_retries, scheduled_at) "insert into task (id, name, props, queue, label, priority, max_retries, scheduled_at)
values (?, ?, ?, ?, ?, ?, ?, now() + ?) values (?, ?, ?, ?, ?, ?, ?, now() + ?)
@ -87,14 +77,13 @@
(s/def ::task (s/or :kw keyword? :str string?)) (s/def ::task (s/or :kw keyword? :str string?))
(s/def ::queue (s/or :kw keyword? :str string?)) (s/def ::queue (s/or :kw keyword? :str string?))
(s/def ::delay (s/or :int integer? :duration dt/duration?)) (s/def ::delay (s/or :int integer? :duration dt/duration?))
(s/def ::conn (s/or :pool ::db/pool :connection some?))
(s/def ::priority integer?) (s/def ::priority integer?)
(s/def ::max-retries integer?) (s/def ::max-retries integer?)
(s/def ::dedupe boolean?) (s/def ::dedupe boolean?)
(s/def ::submit-options (s/def ::submit-options
(s/and (s/and
(s/keys :req [::task ::conn] (s/keys :req [::task]
:opt [::label ::delay ::queue ::priority ::max-retries ::dedupe]) :opt [::label ::delay ::queue ::priority ::max-retries ::dedupe])
(fn [{:keys [::dedupe ::label] :or {label ""}}] (fn [{:keys [::dedupe ::label] :or {label ""}}]
(if dedupe (if dedupe
@ -102,21 +91,23 @@
true)))) true))))
(defn submit! (defn submit!
[& {:keys [::task ::delay ::queue ::priority ::max-retries ::conn ::dedupe ::label] [& {:keys [::params ::task ::delay ::queue ::priority ::max-retries ::dedupe ::label]
:or {delay 0 queue :default priority 100 max-retries 3 label ""} :or {delay 0 queue :default priority 100 max-retries 3 label ""}
:as options}] :as options}]
(us/verify! ::submit-options options) (us/verify! ::submit-options options)
(let [duration (dt/duration delay) (let [duration (dt/duration delay)
interval (db/interval duration) interval (db/interval duration)
props (-> options extract-props db/tjson) props (db/tjson params)
id (uuid/next) id (uuid/next)
tenant (cf/get :tenant) tenant (cf/get :tenant)
task (d/name task) task (d/name task)
queue (str/ffmt "%:%" tenant (d/name queue)) queue (str/ffmt "%:%" tenant (d/name queue))
conn (db/get-connectable options)
deleted (when dedupe deleted (when dedupe
(-> (db/exec-one! conn [sql:remove-not-started-tasks task queue label]) (-> (db/exec-one! conn [sql:remove-not-started-tasks task queue label])
:next.jdbc/update-count))] :next.jdbc/update-count))]
(l/trc :hint "submit task" (l/trc :hint "submit task"
:name task :name task
:task-id (str id) :task-id (str id)
@ -126,7 +117,13 @@
:delay (dt/format-duration duration) :delay (dt/format-duration duration)
:replace (or deleted 0)) :replace (or deleted 0))
(db/exec-one! conn [sql:insert-new-task id task props queue (db/exec-one! conn [sql:insert-new-task id task props queue
label priority max-retries interval]) label priority max-retries interval])
id)) id))
(defn invoke!
[{:keys [::task ::params] :as cfg}]
(assert (contains? cfg :app.worker/registry)
"missing worker registry on `cfg`")
(let [task-fn (dm/get-in cfg [:app.worker/registry (name task)])]
(task-fn {:props params})))

7
backend/test/backend_tests/helpers.clj
View file

@ -34,6 +34,7 @@
[app.util.blob :as blob] [app.util.blob :as blob]
[app.util.services :as sv] [app.util.services :as sv]
[app.util.time :as dt] [app.util.time :as dt]
[app.worker :as wrk]
[app.worker.runner] [app.worker.runner]
[clojure.java.io :as io] [clojure.java.io :as io]
[clojure.spec.alpha :as s] [clojure.spec.alpha :as s]
@ -377,9 +378,9 @@
([name] ([name]
(run-task! name {})) (run-task! name {}))
([name params] ([name params]
(let [tasks (:app.worker/registry *system*)] (wrk/invoke! (-> *system*
(let [task-fn (get tasks (d/name name))] (assoc ::wrk/task name)
(task-fn params))))) (assoc ::wrk/params params)))))
(def sql:pending-tasks (def sql:pending-tasks
"select t.* from task as t "select t.* from task as t

38
backend/test/backend_tests/loggers_webhooks_test.clj
View file

@ -21,11 +21,10 @@
(with-mocks [submit-mock {:target 'app.worker/submit! :return nil}] (with-mocks [submit-mock {:target 'app.worker/submit! :return nil}]
(let [prof (th/create-profile* 1 {:is-active true}) (let [prof (th/create-profile* 1 {:is-active true})
res (th/run-task! :process-webhook-event res (th/run-task! :process-webhook-event
{:props {:event
{:app.loggers.webhooks/event {:type "command"
{:type "command" :name "create-project"
:name "create-project" :props {:team-id (:default-team-id prof)}}})]
:props {:team-id (:default-team-id prof)}}}})]
(t/is (= 0 (:call-count @submit-mock))) (t/is (= 0 (:call-count @submit-mock)))
(t/is (nil? res))))) (t/is (nil? res)))))
@ -35,11 +34,10 @@
(let [prof (th/create-profile* 1 {:is-active true}) (let [prof (th/create-profile* 1 {:is-active true})
whk (th/create-webhook* {:team-id (:default-team-id prof)}) whk (th/create-webhook* {:team-id (:default-team-id prof)})
res (th/run-task! :process-webhook-event res (th/run-task! :process-webhook-event
{:props {:event
{:app.loggers.webhooks/event {:type "command"
{:type "command" :name "create-project"
:name "create-project" :props {:team-id (:default-team-id prof)}}})]
:props {:team-id (:default-team-id prof)}}}})]
(t/is (= 1 (:call-count @submit-mock))) (t/is (= 1 (:call-count @submit-mock)))
(t/is (nil? res))))) (t/is (nil? res)))))
@ -52,9 +50,8 @@
:name "create-project" :name "create-project"
:props {:team-id (:default-team-id prof)}} :props {:team-id (:default-team-id prof)}}
res (th/run-task! :run-webhook res (th/run-task! :run-webhook
{:props {:event evt
{:app.loggers.webhooks/event evt :config whk})]
:app.loggers.webhooks/config whk}})]
(t/is (= 1 (:call-count @http-mock))) (t/is (= 1 (:call-count @http-mock)))
@ -75,9 +72,8 @@
:name "create-project" :name "create-project"
:props {:team-id (:default-team-id prof)}} :props {:team-id (:default-team-id prof)}}
res (th/run-task! :run-webhook res (th/run-task! :run-webhook
{:props {:event evt
{:app.loggers.webhooks/event evt :config whk})]
:app.loggers.webhooks/config whk}})]
(t/is (= 1 (:call-count @http-mock))) (t/is (= 1 (:call-count @http-mock)))
@ -94,14 +90,12 @@
;; RUN 2 times more ;; RUN 2 times more
(th/run-task! :run-webhook (th/run-task! :run-webhook
{:props {:event evt
{:app.loggers.webhooks/event evt :config whk})
:app.loggers.webhooks/config whk}})
(th/run-task! :run-webhook (th/run-task! :run-webhook
{:props {:event evt
{:app.loggers.webhooks/event evt :config whk})
:app.loggers.webhooks/config whk}})
(let [rows (th/db-query :webhook-delivery {:webhook-id (:id whk)})] (let [rows (th/db-query :webhook-delivery {:webhook-id (:id whk)})]

12
frontend/src/app/main/data/workspace/libraries.cljs
View file

@ -1164,14 +1164,15 @@
changes-s changes-s
(->> stream (->> stream
(rx/filter #(or (dch/commit? %) (rx/filter dch/commit?)
(ptk/type? % ::dwn/handle-file-change))) (rx/map deref)
(rx/filter #(= :local (:source %)))
(rx/observe-on :async)) (rx/observe-on :async))
check-changes check-changes
(fn [[event [old-data _mid_data _new-data]]] (fn [[event [old-data _mid_data _new-data]]]
(when old-data (when old-data
(let [{:keys [file-id changes save-undo? undo-group]} (deref event) (let [{:keys [file-id changes save-undo? undo-group]} event
changed-components changed-components
(when (or (nil? file-id) (= file-id (:id old-data))) (when (or (nil? file-id) (= file-id (:id old-data)))
@ -1181,7 +1182,7 @@
(if (d/not-empty? changed-components) (if (d/not-empty? changed-components)
(if save-undo? (if save-undo?
(do (log/info :msg "DETECTED COMPONENTS CHANGED" (do (log/info :hint "detected component changes"
:ids (map str changed-components) :ids (map str changed-components)
:undo-group undo-group) :undo-group undo-group)
@ -1190,7 +1191,8 @@
;; even if save-undo? is false, we need to update the :modified-date of the component ;; even if save-undo? is false, we need to update the :modified-date of the component
;; (for example, for undos) ;; (for example, for undos)
(->> (rx/from changed-components) (->> (rx/from changed-components)
(rx/map #(touch-component %)))) (rx/map touch-component)))
(rx/empty))))) (rx/empty)))))
changes-s changes-s