github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-10-29 05:50:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

♻️ Gather the audit-log consumers in one namespace

Browse Source 
The event consumer lived in app.tasks.event-consumer, but it is not a
domain task: it is the machinery of the audit log, next to the queues
it drains and next to the other two consumers. The transport was in
app.loggers.audit.consumer-task, a name that says "the task" when it
holds the three of them.

Both are now app.loggers.audit.consumers: the queue helpers, the
drain, the three job-defs and the registry of reactions live in one
namespace, and app.tasks.event-consumer is gone. The team-storage
reaction stays where it was, in app.tasks.team-storage, since it owns
the table and the SQL of its mark.

The tests follow: the consumer test is now
backend-tests.loggers-audit-consumers-test, and the tests of the
reactions live in it.

AI-assisted-by: minimax-m3.1-flash
This commit is contained in:
Andrey Antukh 2026-10-23 17:08:47 +00:00
parent 0f4c549668
commit 9830c2d6f4
9 changed files with 290 additions and 323 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
backend/src/app/loggers/audit.clj
View File

@ -21,7 +21,7 @@
[app.http :as-alias http]
[app.http.access-token :as-alias actoken]
[app.jobs :as jobs]
[app.loggers.audit.consumer-task :as consumer]
[app.loggers.audit.consumers :as consumer]
[app.loggers.audit.tasks :as-alias tasks]
[app.loggers.webhooks :as-alias webhooks]
[app.rpc :as-alias rpc]

145
backend/src/app/loggers/audit/consumer_task.clj → backend/src/app/loggers/audit/consumers.clj
View File

@ -4,7 +4,40 @@
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.loggers.audit.consumer-task
(ns app.loggers.audit.consumers
"The consumers of the audit log buffer.
Two of them archive the events in a long-term destination and the
third one dispatches them to the reactions:
- `:audit-log-consumer-nexus` sends them to Nexus.
- `:audit-log-consumer-admin-console` sends the allowlisted ones to
the admin console.
- `:audit-log-consumer-events` is the generalist: it takes the whole
stream and hands each event to the reactions that claim it.
Each job drains only its own queue and deletes its pendings on
completion; whoever deletes the last pending also deletes the record
(last one out turns off the light), so no separate GC task is
needed. Housekeeping (reaping disabled pendings, sweeping orphans)
lives in every job on purpose: only the enabled job runs, so it must
clean up for the rest.
A reaction is a map:
{:id :team-storage-usage ;; who it is
:flag :storage-usage ;; the flag that turns it on
:events #{...} ;; the event names it cares about
:apply (fn [cfg events] ...)} ;; what it does with them
Each domain contributes its own: the one that owns the table and the
SQL of its mark is the one that knows which events affect it and
where its key lives. A reaction receives the whole batch of events it
declares, once: a reaction that groups work (marking a team dirty)
must not write once per event. The dispatch does not catch
exceptions: the reaction decides which errors are data (it skips the
event) and which abort the batch (the event is retried on the next
tick)."
(:require
[app.common.exceptions :as ex]
[app.common.logging :as l]
@ -21,25 +54,10 @@
[app.metrics :as mtx]
[app.nitrate :as nitrate]
[app.setup :as-alias setup]
[app.tasks.team-storage :as team-storage]
[integrant.core :as ig]
[promesa.exec :as px]))
;; This task drains the audit log buffer: two of the jobs archive the
;; events in a long-term destination, and the third one dispatches them
;; to the reactions.
;;
;; Two separate jobs live in this namespace, one per destination:
;; `:audit-log-consumer-nexus` and
;; `:audit-log-consumer-admin-console`. The third one,
;; `:audit-log-consumer-events`, lives in `app.tasks.event-consumer`
;; (the generalist that decides what to do with each event). Each job
;; drains only its own queue and deletes its pendings on completion;
;; whoever deletes the last pending also deletes the record ("last one
;; out turns off the light"), so no separate GC task is needed.
;; Housekeeping (reaping disabled pendings, sweeping orphans) lives in
;; every job on purpose:
;; only the enabled job runs, so it must clean up for the rest.
;; Consumer ids are fixed constants, never reuse one: retired
;; pendings are reaped by id, so a reused id would inherit the
;; previous owner's queue.
@ -113,6 +131,49 @@
(when (contains? cf/flags :event-consumer)
(queue-pendings cfg ids consumer-events)))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; REACTIONS
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(def reactions
"The registered reactions, in the order they are applied."
[team-storage/team-storage-reaction])
(defn enabled-reactions
"The reactions whose flag is on."
[]
(filterv #(contains? cf/flags (:flag %)) reactions))
(defn- claimed-names
"The names the enabled reactions claim."
[]
(into #{} (mapcat :events) (enabled-reactions)))
(defn- claimed-by
"The enabled reactions that declare the name of this event."
[reactions event]
(filterv #(contains? (:events %) (:name event)) reactions))
(defn dispatch-reactions
"Applies the enabled reactions to the events. Each reaction receives
only the events it declares, and it receives them in a single batch.
Returns the number of events some reaction claimed: the ones nobody
claimed are discarded, because the queue of the consumer is not
selective and that event concerns nobody."
[cfg events]
(let [reactions (enabled-reactions)
batches (keep (fn [reaction]
(let [mine (filterv #(contains? (:events reaction) (:name %)) events)]
(when (seq mine)
[reaction mine])))
reactions)]
(doseq [[reaction mine] batches]
((:apply reaction) cfg mine))
(count (filter (fn [event]
(seq (claimed-by reactions event)))
events))))
(def ^:private consumers-by-id
(into {} (map (juxt :id identity)) consumers))
@ -500,6 +561,23 @@
:events (count decoded))]
(send-to-admin-console cfg decoded)))
(defn- process-events-chunk
"Dispatch one pending chunk to the reactions. Returns {:sent :skipped}:
the events a reaction claims count as sent, the rest as skipped.
Only the claimed events are decoded: the queue is not selective, so
most of the stream concerns nobody and opening its blob would be
work thrown away."
[cfg rows]
(let [names (claimed-names)
claimed (filterv #(contains? names (:name %)) rows)
batch (mapv row->reaction-event claimed)
sent (dispatch-reactions cfg batch)]
(l/trc :hint "consumer events chunk"
:consumer "events"
:events sent)
{:sent sent :skipped (- (count rows) sent)}))
(defn- drain-chunks
"Drain one consumer queue chunk by chunk, one transaction per chunk.
The processor receives the rows of each chunk, already locked, and
@ -581,6 +659,7 @@
(declare execute-audit-log-consumer-nexus)
(declare execute-audit-log-consumer-admin-console)
(declare execute-audit-log-consumer-events)
(def schema:audit-log-consumer-nexus-params
"Overrides for the nexus job repl invocation defaults.
@ -642,6 +721,29 @@
::jobs/decoder (sm/decoder schema:audit-log-consumer-admin-console-params sm/json-transformer)
::jobs/validator (sm/validator schema:audit-log-consumer-admin-console-params)})
(def schema:audit-log-consumer-events-params
"Params of the events job: none. Cron submits {}; system deps (pool)
live on the job-def cfg closed over by the handler, not in job params.
Closed on purpose, same as the admin-console schema."
[:map {:closed true}])
(defmethod ig/assert-key ::events-job-def
[_ cfg]
(assert (db/pool? (::db/pool cfg)) "expected a valid database pool")
(assert (mtx/metrics? (::mtx/metrics cfg)) "expected a valid metrics instance"))
(defmethod ig/init-key ::events-job-def
[_ cfg]
(l/dbg :hint "audit log consumer job initialized"
:consumer "events")
{::jobs/name :audit-log-consumer-events
::jobs/schema schema:audit-log-consumer-events-params
::jobs/handler
(fn [_context _params]
(execute-audit-log-consumer-events cfg))
::jobs/decoder (sm/decoder schema:audit-log-consumer-events-params sm/json-transformer)
::jobs/validator (sm/validator schema:audit-log-consumer-events-params)})
(defn drain-consumer
"Run one consumer: drain its queue chunk by chunk (one transaction per
chunk), then the housekeeping (reap of the disabled queues and sweep
@ -737,3 +839,12 @@
(l/dbg :hint "audit log consumer (admin-console) skipped, flag off")
(drain-consumer cfg consumer-admin-console process-admin-console-chunk)))
(defn execute-audit-log-consumer-events
"Events job handler: drain the events queue in chunks, dispatching
each event to the reactions that claim it (heartbeat per iteration:
a burst can span many chunks)."
[cfg]
(if-not (contains? cf/flags :event-consumer)
(l/dbg :hint "audit log consumer (events) skipped, flag off")
(drain-consumer cfg consumer-events process-events-chunk)))

12
backend/src/app/main.clj
View File

@ -660,9 +660,9 @@
:telemetry (ig/ref :app.tasks.telemetry/job-def)
:session-gc (ig/ref ::session/job-def)
:file-gc-scheduler (ig/ref :app.tasks.file-gc-scheduler/job-def)
:audit-log-consumer-nexus (ig/ref :app.loggers.audit.consumer-task/nexus-job-def)
:audit-log-consumer-admin-console (ig/ref :app.loggers.audit.consumer-task/admin-console-job-def)
:audit-log-consumer-events (ig/ref :app.tasks.event-consumer/events-job-def)}
:audit-log-consumer-nexus (ig/ref :app.loggers.audit.consumers/nexus-job-def)
:audit-log-consumer-admin-console (ig/ref :app.loggers.audit.consumers/admin-console-job-def)
:audit-log-consumer-events (ig/ref :app.loggers.audit.consumers/events-job-def)}
::jobs/request-pool
{::rds/client (ig/ref ::rds/client)
@ -759,18 +759,18 @@
{::db/pool (ig/ref ::db/pool)
::mtx/metrics (ig/ref ::mtx/metrics)}
:app.loggers.audit.consumer-task/nexus-job-def
:app.loggers.audit.consumers/nexus-job-def
{::db/pool (ig/ref ::db/pool)
::http.client/client (ig/ref ::http.client/client)
::setup/shared-keys (ig/ref ::setup/shared-keys)
::mtx/metrics (ig/ref ::mtx/metrics)}
:app.loggers.audit.consumer-task/admin-console-job-def
:app.loggers.audit.consumers/admin-console-job-def
{::db/pool (ig/ref ::db/pool)
:app.nitrate/client (ig/ref :app.nitrate/client)
::mtx/metrics (ig/ref ::mtx/metrics)}
:app.tasks.event-consumer/events-job-def
:app.loggers.audit.consumers/events-job-def
{::db/pool (ig/ref ::db/pool)
::mtx/metrics (ig/ref ::mtx/metrics)}

2
backend/src/app/rpc/commands/audit.clj
View File

@ -16,7 +16,7 @@
[app.db :as db]
[app.http :as-alias http]
[app.loggers.audit :as audit]
[app.loggers.audit.consumer-task :as consumer]
[app.loggers.audit.consumers :as consumer]
[app.loggers.database :as loggers.db]
[app.loggers.mattermost :as loggers.mm]
[app.metrics :as mtx]

130
backend/src/app/tasks/event_consumer.clj
View File

@ -1,130 +0,0 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns app.tasks.event-consumer
"The generalist of the audit-log event consumer: it reads the stream
of events and lets each reaction decide what concerns it.
The consumer itself (the id 3 of `audit_log_pending`) is not selective:
it queues every event and hands it to the reactions. A reaction is a
map:
{:id :team-storage-usage ;; who it is
:flag :storage-usage ;; the flag that turns it on
:events #{...} ;; the event names it cares about
:apply (fn [cfg events] ...)} ;; what it does with them
Each domain contributes its own: the one that owns the table and the
SQL of its mark is the one that knows which events affect it and
where its key lives.
A reaction receives the whole batch of events it declares, once: a
reaction that groups work (marking a team dirty) must not write once
per event. The dispatch does not catch exceptions: the reaction
decides which errors are data (it skips the event) and which abort
the batch (the event is retried on the next tick)."
(:require
[app.common.logging :as l]
[app.common.schema :as sm]
[app.config :as cf]
[app.db :as db]
[app.jobs :as jobs]
[app.loggers.audit.consumer-task :as consumer]
[app.metrics :as mtx]
[app.tasks.team-storage :as team-storage]
[integrant.core :as ig]))
(def reactions
"The registered reactions, in the order they are applied."
[team-storage/team-storage-reaction])
(defn enabled-reactions
"The reactions whose flag is on."
[]
(filterv #(contains? cf/flags (:flag %)) reactions))
(defn- claimed-names
"The names the enabled reactions claim."
[]
(into #{} (mapcat :events) (enabled-reactions)))
(defn- claimed-by
"The enabled reactions that declare the name of this event."
[reactions event]
(filterv #(contains? (:events %) (:name event)) reactions))
(defn dispatch-reactions
"Applies the enabled reactions to the events. Each reaction receives
only the events it declares, and it receives them in a single batch.
Returns the number of events some reaction claimed: the ones nobody
claimed are discarded, because the queue of the consumer is not
selective and that event concerns nobody."
[cfg events]
(let [reactions (enabled-reactions)
batches (keep (fn [reaction]
(let [mine (filterv #(contains? (:events reaction) (:name %)) events)]
(when (seq mine)
[reaction mine])))
reactions)]
(doseq [[reaction mine] batches]
((:apply reaction) cfg mine))
(count (filter (fn [event]
(seq (claimed-by reactions event)))
events))))
(defn- process-events-chunk
"Dispatch one pending chunk to the reactions. Returns {:sent :skipped}:
the events a reaction claims count as sent, the rest as skipped.
Only the claimed events are decoded: the queue is not selective, so
most of the stream concerns nobody and opening its blob would be
work thrown away."
[cfg rows]
(let [names (claimed-names)
claimed (filterv #(contains? names (:name %)) rows)
batch (mapv consumer/row->reaction-event claimed)
sent (dispatch-reactions cfg batch)]
(l/trc :hint "consumer events chunk"
:consumer "events"
:events sent)
{:sent sent :skipped (- (count rows) sent)}))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; HANDLER
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(defn execute-audit-log-consumer-events
"Events job handler: drain the events queue in chunks, dispatching
each event to the reactions that claim it (heartbeat per iteration:
a burst can span many chunks)."
[cfg]
(if-not (contains? cf/flags :event-consumer)
(l/dbg :hint "audit log consumer (events) skipped, flag off")
(consumer/drain-consumer cfg consumer/consumer-events process-events-chunk)))
(def schema:audit-log-consumer-events-params
"Params of the events job: none. Cron submits {}; system deps (pool)
live on the job-def cfg closed over by the handler, not in job params.
Closed on purpose, same as the admin-console schema."
[:map {:closed true}])
(defmethod ig/assert-key ::events-job-def
[_ cfg]
(assert (db/pool? (::db/pool cfg)) "expected a valid database pool")
(assert (mtx/metrics? (::mtx/metrics cfg)) "expected a valid metrics instance"))
(defmethod ig/init-key ::events-job-def
[_ cfg]
(l/dbg :hint "audit log consumer job initialized"
:consumer "events")
{::jobs/name :audit-log-consumer-events
::jobs/schema schema:audit-log-consumer-events-params
::jobs/handler
(fn [_context _params]
(execute-audit-log-consumer-events cfg))
::jobs/decoder (sm/decoder schema:audit-log-consumer-events-params sm/json-transformer)
::jobs/validator (sm/validator schema:audit-log-consumer-events-params)})

160
backend/test/backend_tests/loggers_audit_consumer_task_test.clj → backend/test/backend_tests/loggers_audit_consumers_test.clj
View File

@ -4,7 +4,9 @@
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns backend-tests.loggers-audit-consumer-task-test
(ns backend-tests.loggers-audit-consumers-test
"The consumers of the audit log buffer: the two that archive the
events and the generalist that dispatches them to the reactions."
(:require
[app.common.json :as json]
[app.common.schema :as sm]
@ -17,12 +19,11 @@
[app.http.client :as-alias http]
[app.jobs :as jobs]
[app.loggers.audit :as audit]
[app.loggers.audit.consumer-task :as consumer]
[app.loggers.audit.consumers :as consumer]
[app.main :as main]
[app.metrics :as mtx]
[app.metrics.definition :as-alias mdef]
[app.setup :as-alias setup]
[app.tasks.event-consumer :as-alias event-consumer]
[backend-tests.helpers :as th]
[clojure.test :as t]
[cuerdas.core :as str]
@ -236,13 +237,13 @@
;; Each consumer job is a job-def: the handler closes over the
;; cfg it was given at init and takes a nil context, because
;; an in-process invocation has no job row to describe.
nexus-job-def (ig/init-key :app.loggers.audit.consumer-task/nexus-job-def
nexus-job-def (ig/init-key :app.loggers.audit.consumers/nexus-job-def
{::db/pool (:app.db/pool th/*system*)
::setup/shared-keys shared-keys
::http/client (:app.http.client/client th/*system*)
::mtx/metrics (th/metrics-stub)
:app.nitrate/client client})
admin-console-job-def (ig/init-key :app.loggers.audit.consumer-task/admin-console-job-def
admin-console-job-def (ig/init-key :app.loggers.audit.consumers/admin-console-job-def
{::db/pool (:app.db/pool th/*system*)
::mtx/metrics (th/metrics-stub)
:app.nitrate/client client})]
@ -452,7 +453,7 @@
metrics registry. The handler takes a nil context because there is
no job row to describe (same as `th/run-task!`)."
[metrics params]
(let [job-def (ig/init-key :app.loggers.audit.consumer-task/nexus-job-def
(let [job-def (ig/init-key :app.loggers.audit.consumers/nexus-job-def
{::db/pool (:app.db/pool th/*system*)
::setup/shared-keys {:admin-console "ac-key" :nexus "nexus-key"}
::http/client (:app.http.client/client th/*system*)
@ -465,7 +466,7 @@
tests) system one: tests mock `app.nitrate/call`, so it is never
dereferenced."
[metrics]
(let [job-def (ig/init-key :app.loggers.audit.consumer-task/admin-console-job-def
(let [job-def (ig/init-key :app.loggers.audit.consumers/admin-console-job-def
{::db/pool (:app.db/pool th/*system*)
:app.nitrate/client (:app.nitrate/client th/*system*)
::mtx/metrics metrics})]
@ -483,7 +484,7 @@
metrics registry. It only needs the pool: dispatching the events to
the reactions touches no archive destination."
[metrics]
(let [job-def (ig/init-key :app.tasks.event-consumer/events-job-def
(let [job-def (ig/init-key :app.loggers.audit.consumers/events-job-def
{::db/pool (:app.db/pool th/*system*)
::mtx/metrics metrics})]
((::jobs/handler job-def) nil {})))
@ -761,3 +762,146 @@
(t/is (= 1.0 (counter-value metrics :audit-consumer-events ["admin-console" "skipped"])))
(t/is (= 0 (count (th/db-exec! ["select * from audit_log"]))))
(t/is (= 0 (pending-count)))))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; REACTIONS
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(t/use-fixtures :each th/database-reset)
(defn- capture-reaction
"A fake reaction that keeps the batch it receives."
[seen events]
{:id :test
:flag :test-flag
:events events
:apply (fn [_cfg batch] (swap! seen conj batch))})
(defn- append-row
"Append an audit-log row the way the producer does."
([name props] (append-row name props "backend"))
([name props source]
(#'audit/append-audit-entry
th/*system*
{:id (uuid/next)
:name name
:type "action"
:source source
:profile-id uuid/zero
:ip-addr "127.0.0.1"
:props props
:context {}
:tracked-at (ct/now)
:created-at (ct/now)})))
(defn- events-pendings
"The pendings of the events consumer (id 3)."
[]
(th/db-exec! ["SELECT audit_log_id FROM audit_log_pending WHERE consumer = 3"]))
(defn- usage-row
[team-id]
(th/db-exec-one! ["SELECT team_id, dirty_at FROM team_storage_usage WHERE team_id = ?"
team-id]))
(t/deftest dispatch-applies-a-reaction-to-the-events-it-declares
(let [seen (atom [])
reaction (capture-reaction seen #{"delete-file"})
event-a {:id (uuid/next) :name "delete-file" :props {}}
event-b {:id (uuid/next) :name "create-file" :props {}}]
(with-redefs [cf/flags #{:test-flag}
consumer/reactions [reaction]]
(let [claimed (consumer/dispatch-reactions
th/*system*
[event-a event-b])]
(t/is (= 1 claimed))
(t/is (= 1 (count @seen)))
(t/is (= ["delete-file"] (mapv :name (first @seen))))))))
(t/deftest dispatch-ignores-the-events-no-reaction-claims
;; The queue of the consumer is not selective: the events that nobody
;; claims are discarded.
(let [seen (atom [])
reaction (capture-reaction seen #{"delete-file"})]
(with-redefs [cf/flags #{:test-flag}
consumer/reactions [reaction]]
(let [claimed (consumer/dispatch-reactions
th/*system*
[{:id (uuid/next) :name "create-file" :props {}}])]
(t/is (= 0 claimed))
(t/is (empty? @seen))))))
(t/deftest dispatch-skips-a-reaction-whose-flag-is-off
;; The flag of the reaction is the one that turns it on: without it,
;; the event is not claimed and the reaction does not run.
(let [seen (atom [])
reaction (capture-reaction seen #{"delete-file"})]
(with-redefs [cf/flags #{}
consumer/reactions [reaction]]
(let [claimed (consumer/dispatch-reactions
th/*system*
[{:id (uuid/next) :name "delete-file" :props {}}])]
(t/is (= 0 claimed))
(t/is (empty? @seen))))))
(t/deftest dispatch-gives-each-reaction-only-its-own-events
;; Two reactions that declare the same name: each one receives its
;; batch and the event is counted once.
(let [seen-a (atom [])
seen-b (atom [])
reaction-a (capture-reaction seen-a #{"delete-file"})
reaction-b (capture-reaction seen-b #{"delete-file" "create-file"})
events [{:id (uuid/next) :name "delete-file" :props {}}
{:id (uuid/next) :name "create-file" :props {}}]]
(with-redefs [cf/flags #{:test-flag}
consumer/reactions [reaction-a reaction-b]]
(let [claimed (consumer/dispatch-reactions th/*system* events)]
(t/is (= 2 claimed))
(t/is (= ["delete-file"] (mapv :name (first @seen-a))))
(t/is (= ["delete-file" "create-file"] (mapv :name (first @seen-b))))))))
(t/deftest events-consumer-drains-the-queue-and-marks-the-team-dirty
;; The whole path: the producer queues the event, the consumer drains
;; it and the reaction marks the team dirty.
(let [prof (th/create-profile* 41 {:is-active true})
team (:default-team-id prof)]
(with-redefs [cf/flags #{:audit-log :event-consumer :storage-usage}]
(append-row "upload-file-media-object" {:team-id team})
(t/is (= 1 (count (events-pendings))))
(th/run-task! :audit-log-consumer-events {})
(t/is (= 0 (count (events-pendings))))
(t/is (some? (:dirty-at (usage-row team)))))))
(t/deftest events-consumer-discards-the-events-nobody-claims
;; An event that no reaction claims is drained and dropped: the queue
;; takes the whole stream, but nobody works on it.
(let [prof (th/create-profile* 42 {:is-active true})
team (:default-team-id prof)]
(with-redefs [cf/flags #{:audit-log :event-consumer :storage-usage}]
(append-row "navigate" {:route "dashboard"})
(th/run-task! :audit-log-consumer-events {})
(t/is (= 0 (count (events-pendings))))
(t/is (nil? (:dirty-at (usage-row team)))))))
(t/deftest events-consumer-does-nothing-without-its-flag
;; Without its flag the consumer does not even queue: the producer
;; never hands it the stream.
(let [prof (th/create-profile* 43 {:is-active true})
team (:default-team-id prof)]
(with-redefs [cf/flags #{:audit-log :storage-usage}]
(append-row "upload-file-media-object" {:team-id team})
(th/run-task! :audit-log-consumer-events {})
(t/is (= 0 (count (events-pendings))))
(t/is (nil? (:dirty-at (usage-row team)))))))
(t/deftest events-consumer-does-nothing-without-the-reaction-flag
;; The queue takes the event, but the reaction that would work on it
;; is off: the event is drained and dropped.
(let [prof (th/create-profile* 44 {:is-active true})
team (:default-team-id prof)]
(with-redefs [cf/flags #{:audit-log :event-consumer}]
(append-row "upload-file-media-object" {:team-id team})
(t/is (= 1 (count (events-pendings))))
(th/run-task! :audit-log-consumer-events {})
(t/is (= 0 (count (events-pendings))))
(t/is (nil? (:dirty-at (usage-row team)))))))

158
backend/test/backend_tests/tasks_event_consumer_test.clj
View File

@ -1,158 +0,0 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL
(ns backend-tests.tasks-event-consumer-test
"The generalist of the audit-log event consumer: the registry of
reactions, the dispatch and the drain of the queue."
(:require
[app.common.time :as ct]
[app.common.uuid :as uuid]
[app.config :as cf]
[app.db :as db]
[app.loggers.audit :as audit]
[app.tasks.event-consumer :as event-consumer]
[backend-tests.helpers :as th]
[clojure.test :as t]))
(t/use-fixtures :once th/state-init)
(t/use-fixtures :each th/database-reset)
(defn- capture-reaction
"A fake reaction that keeps the batch it receives."
[seen events]
{:id :test
:flag :test-flag
:events events
:apply (fn [_cfg batch] (swap! seen conj batch))})
(defn- append-row
"Append an audit-log row the way the producer does."
([name props] (append-row name props "backend"))
([name props source]
(#'audit/append-audit-entry
th/*system*
{:id (uuid/next)
:name name
:type "action"
:source source
:profile-id uuid/zero
:ip-addr "127.0.0.1"
:props props
:context {}
:tracked-at (ct/now)
:created-at (ct/now)})))
(defn- events-pendings
"The pendings of the events consumer (id 3)."
[]
(th/db-exec! ["SELECT audit_log_id FROM audit_log_pending WHERE consumer = 3"]))
(defn- usage-row
[team-id]
(th/db-exec-one! ["SELECT team_id, dirty_at FROM team_storage_usage WHERE team_id = ?"
team-id]))
(t/deftest dispatch-applies-a-reaction-to-the-events-it-declares
(let [seen (atom [])
reaction (capture-reaction seen #{"delete-file"})
event-a {:id (uuid/next) :name "delete-file" :props {}}
event-b {:id (uuid/next) :name "create-file" :props {}}]
(with-redefs [cf/flags #{:test-flag}
event-consumer/reactions [reaction]]
(let [claimed (event-consumer/dispatch-reactions
th/*system*
[event-a event-b])]
(t/is (= 1 claimed))
(t/is (= 1 (count @seen)))
(t/is (= ["delete-file"] (mapv :name (first @seen))))))))
(t/deftest dispatch-ignores-the-events-no-reaction-claims
;; The queue of the consumer is not selective: the events that nobody
;; claims are discarded.
(let [seen (atom [])
reaction (capture-reaction seen #{"delete-file"})]
(with-redefs [cf/flags #{:test-flag}
event-consumer/reactions [reaction]]
(let [claimed (event-consumer/dispatch-reactions
th/*system*
[{:id (uuid/next) :name "create-file" :props {}}])]
(t/is (= 0 claimed))
(t/is (empty? @seen))))))
(t/deftest dispatch-skips-a-reaction-whose-flag-is-off
;; The flag of the reaction is the one that turns it on: without it,
;; the event is not claimed and the reaction does not run.
(let [seen (atom [])
reaction (capture-reaction seen #{"delete-file"})]
(with-redefs [cf/flags #{}
event-consumer/reactions [reaction]]
(let [claimed (event-consumer/dispatch-reactions
th/*system*
[{:id (uuid/next) :name "delete-file" :props {}}])]
(t/is (= 0 claimed))
(t/is (empty? @seen))))))
(t/deftest dispatch-gives-each-reaction-only-its-own-events
;; Two reactions that declare the same name: each one receives its
;; batch and the event is counted once.
(let [seen-a (atom [])
seen-b (atom [])
reaction-a (capture-reaction seen-a #{"delete-file"})
reaction-b (capture-reaction seen-b #{"delete-file" "create-file"})
events [{:id (uuid/next) :name "delete-file" :props {}}
{:id (uuid/next) :name "create-file" :props {}}]]
(with-redefs [cf/flags #{:test-flag}
event-consumer/reactions [reaction-a reaction-b]]
(let [claimed (event-consumer/dispatch-reactions th/*system* events)]
(t/is (= 2 claimed))
(t/is (= ["delete-file"] (mapv :name (first @seen-a))))
(t/is (= ["delete-file" "create-file"] (mapv :name (first @seen-b))))))))
(t/deftest events-consumer-drains-the-queue-and-marks-the-team-dirty
;; The whole path: the producer queues the event, the consumer drains
;; it and the reaction marks the team dirty.
(let [prof (th/create-profile* 41 {:is-active true})
team (:default-team-id prof)]
(with-redefs [cf/flags #{:audit-log :event-consumer :storage-usage}]
(append-row "upload-file-media-object" {:team-id team})
(t/is (= 1 (count (events-pendings))))
(th/run-task! :audit-log-consumer-events {})
(t/is (= 0 (count (events-pendings))))
(t/is (some? (:dirty-at (usage-row team)))))))
(t/deftest events-consumer-discards-the-events-nobody-claims
;; An event that no reaction claims is drained and dropped: the queue
;; takes the whole stream, but nobody works on it.
(let [prof (th/create-profile* 42 {:is-active true})
team (:default-team-id prof)]
(with-redefs [cf/flags #{:audit-log :event-consumer :storage-usage}]
(append-row "navigate" {:route "dashboard"})
(th/run-task! :audit-log-consumer-events {})
(t/is (= 0 (count (events-pendings))))
(t/is (nil? (:dirty-at (usage-row team)))))))
(t/deftest events-consumer-does-nothing-without-its-flag
;; Without its flag the consumer does not even queue: the producer
;; never hands it the stream.
(let [prof (th/create-profile* 43 {:is-active true})
team (:default-team-id prof)]
(with-redefs [cf/flags #{:audit-log :storage-usage}]
(append-row "upload-file-media-object" {:team-id team})
(th/run-task! :audit-log-consumer-events {})
(t/is (= 0 (count (events-pendings))))
(t/is (nil? (:dirty-at (usage-row team)))))))
(t/deftest events-consumer-does-nothing-without-the-reaction-flag
;; The queue takes the event, but the reaction that would work on it
;; is off: the event is drained and dropped.
(let [prof (th/create-profile* 44 {:is-active true})
team (:default-team-id prof)]
(with-redefs [cf/flags #{:audit-log :event-consumer}]
(append-row "upload-file-media-object" {:team-id team})
(t/is (= 1 (count (events-pendings))))
(th/run-task! :audit-log-consumer-events {})
(t/is (= 0 (count (events-pendings))))
(t/is (nil? (:dirty-at (usage-row team)))))))

2
backend/test/backend_tests/tasks_team_storage_test.clj
View File

@ -16,7 +16,7 @@
[app.db :as db]
[app.jobs :as jobs]
[app.loggers.audit :as audit]
[app.loggers.audit.consumer-task :as consumer]
[app.loggers.audit.consumers :as consumer]
[app.metrics :as mtx]
[app.rpc :as-alias rpc]
[app.rpc.commands.audit :as cmd-audit]

2
backend/test/backend_tests/tasks_telemetry_test.clj
View File

@ -11,7 +11,7 @@
[app.config :as cf]
[app.db :as db]
[app.loggers.audit :as audit]
[app.loggers.audit.consumer-task :as at]
[app.loggers.audit.consumers :as at]
[app.tasks.telemetry :as telemetry]
[app.util.json :as json]
[backend-tests.helpers :as th]