From 9830c2d6f4b9cdc3b070907dc7ead53dd7762d0d Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Fri, 23 Oct 2026 17:08:47 +0000 Subject: [PATCH] :recycle: Gather the audit-log consumers in one namespace 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 --- backend/src/app/loggers/audit.clj | 2 +- .../{consumer_task.clj => consumers.clj} | 145 ++++++++++++++-- backend/src/app/main.clj | 12 +- backend/src/app/rpc/commands/audit.clj | 2 +- backend/src/app/tasks/event_consumer.clj | 130 -------------- ...t.clj => loggers_audit_consumers_test.clj} | 160 +++++++++++++++++- .../tasks_event_consumer_test.clj | 158 ----------------- .../backend_tests/tasks_team_storage_test.clj | 2 +- .../backend_tests/tasks_telemetry_test.clj | 2 +- 9 files changed, 290 insertions(+), 323 deletions(-) rename backend/src/app/loggers/audit/{consumer_task.clj => consumers.clj} (84%) delete mode 100644 backend/src/app/tasks/event_consumer.clj rename backend/test/backend_tests/{loggers_audit_consumer_task_test.clj => loggers_audit_consumers_test.clj} (85%) delete mode 100644 backend/test/backend_tests/tasks_event_consumer_test.clj diff --git a/backend/src/app/loggers/audit.clj b/backend/src/app/loggers/audit.clj index 6304ccbde7..e6816dfd81 100644 --- a/backend/src/app/loggers/audit.clj +++ b/backend/src/app/loggers/audit.clj @@ -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] diff --git a/backend/src/app/loggers/audit/consumer_task.clj b/backend/src/app/loggers/audit/consumers.clj similarity index 84% rename from backend/src/app/loggers/audit/consumer_task.clj rename to backend/src/app/loggers/audit/consumers.clj index 54d490951f..c8ed9015fe 100644 --- a/backend/src/app/loggers/audit/consumer_task.clj +++ b/backend/src/app/loggers/audit/consumers.clj @@ -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))) + diff --git a/backend/src/app/main.clj b/backend/src/app/main.clj index f7fdd47831..110ed59f1e 100644 --- a/backend/src/app/main.clj +++ b/backend/src/app/main.clj @@ -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)} diff --git a/backend/src/app/rpc/commands/audit.clj b/backend/src/app/rpc/commands/audit.clj index ec9a7497b7..6a6d6e0578 100644 --- a/backend/src/app/rpc/commands/audit.clj +++ b/backend/src/app/rpc/commands/audit.clj @@ -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] diff --git a/backend/src/app/tasks/event_consumer.clj b/backend/src/app/tasks/event_consumer.clj deleted file mode 100644 index 7a1c450de2..0000000000 --- a/backend/src/app/tasks/event_consumer.clj +++ /dev/null @@ -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)}) diff --git a/backend/test/backend_tests/loggers_audit_consumer_task_test.clj b/backend/test/backend_tests/loggers_audit_consumers_test.clj similarity index 85% rename from backend/test/backend_tests/loggers_audit_consumer_task_test.clj rename to backend/test/backend_tests/loggers_audit_consumers_test.clj index ef0e5cbf26..7ca5c65cee 100644 --- a/backend/test/backend_tests/loggers_audit_consumer_task_test.clj +++ b/backend/test/backend_tests/loggers_audit_consumers_test.clj @@ -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))))))) diff --git a/backend/test/backend_tests/tasks_event_consumer_test.clj b/backend/test/backend_tests/tasks_event_consumer_test.clj deleted file mode 100644 index c98c8087fb..0000000000 --- a/backend/test/backend_tests/tasks_event_consumer_test.clj +++ /dev/null @@ -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))))))) diff --git a/backend/test/backend_tests/tasks_team_storage_test.clj b/backend/test/backend_tests/tasks_team_storage_test.clj index a88fd00906..434069fa90 100644 --- a/backend/test/backend_tests/tasks_team_storage_test.clj +++ b/backend/test/backend_tests/tasks_team_storage_test.clj @@ -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] diff --git a/backend/test/backend_tests/tasks_telemetry_test.clj b/backend/test/backend_tests/tasks_telemetry_test.clj index 2bbdd7c371..1caa92e1c8 100644 --- a/backend/test/backend_tests/tasks_telemetry_test.clj +++ b/backend/test/backend_tests/tasks_telemetry_test.clj @@ -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]