📚 Describe the events consumer and its reactions

The audit-log memory still described consumer 3 as the selective
storage-usage consumer, so the next session that read it would think
the generalist did not exist.

It now describes the generalist: id 3, a queue that takes the whole
stream, the :event-consumer flag, the reactions (what they are, who
owns them, and that the producer does not know them) and the flag
table. The team-storage reaction is the only one registered today, and
the cached path of the media-storage quota needs the consumer on: with
it off nobody marks the teams dirty and the quota would read a frozen
counter.

It also fixes what the rename left behind (the job names in the drain
paragraph and in the flags table) and records that update-file now
carries the id of the change row it created.

AI-assisted-by: minimax-m3.1-flash

.serena/memories/audit-log.md

@@ -12,14 +12,15 @@ The system has:
 
 - **Three producers** (they create events): the Penpot backend (RPC layer), the Penpot frontend (browser app), and the Nitrate admin-console, which sends both server-side and client-side events.
 - **One central buffer**: the Postgres `audit_log` table in Penpot. It is temporary: it holds events for minutes/hours until they are archived, then they are deleted.
-- **Four consumers** (they read events): webhooks, error reporters, telemetry, and the external archives in Nexus and in the admin console.
+- **Five consumers** (they read events): webhooks, error reporters, telemetry, the external archives in Nexus and in the admin console, and the **events consumer**, a generalist that lets the backend react to the events as they arrive (see section 12).
 - **Two long-term archives**: **Nexus**, a separate service that stores events for the long term, and the **admin console** (Nitrate), which ingests its own allowlisted events through a second archive job (see section 12).
 
 ```
 Penpot frontend ─ POST push-audit-events (public) ─────────▶┐
 Nitrate admin-console (browser) ─ POST push-audit-events ──▶│
 Nitrate admin-console (server) ─ POST push-audit-events  ──▶│ ─▶ audit_log table (buffer) ──┬─ cron 0s every 5m ──▶ Nexus (archive)
-Penpot backend (RPC wrap-audit) ─ submit ───────────────────┘         │                      └─ cron 30s every 5m ─▶ admin console (archive)
+Penpot backend (RPC wrap-audit) ─ submit ───────────────────┘         │                     ├─ cron 30s every 5m ─▶ admin console (archive)
+                                                                      │                     └─ cron 15s every 5m ─▶ events consumer (reactions)
                                                                       │   ├──▶ webhooks
                                                                       │   ├──▶ error reporters
                                                                       └──▶ telemetry (shadow rows)
@@ -57,7 +58,7 @@ Main columns (migration `backend/src/app/migrations/sql/0054-add-audit-log-table
 - `source` text: `backend` / `frontend` for full rows, `telemetry:backend` / `telemetry:frontend` for anonymized copies (see section 10).
 - `ip_addr` inet: real IP (or `0.0.0.0` on telemetry rows).
 - `props` / `context` jsonb holding transit-encoded maps: action data and metadata (versions, browser, request-id...).
-- No `archived_at`: the `audit_log_pending` queue (migration `0157`, consumer `1` = Nexus, `2` = admin console, `3` = events — see `mem:backend/subtleties` —, ids never reused) tracks what is left. Every record is born pending for the enabled non-selective consumers (today only Nexus; the producer gates on the flag), plus one pending per selective consumer only when the event qualifies (admin-console allowlist, storage-usage event list, both ingest paths): cheaper on single-consumer hosts and true to the design doc. Consequence, accepted on purpose: rows born with no send flag on get no pendings, so enabling archiving later does not pick them up — the sweep deletes them unarchived. Same for a consumer turned selective later: only events queued after the change drain.
+- No `archived_at`: the `audit_log_pending` queue (migration `0157`, consumer `1` = Nexus, `2` = admin console, `3` = events — see `mem:backend/subtleties` —, ids never reused) tracks what is left. Every record is born pending for the enabled non-selective consumers (Nexus and the events consumer; the producer gates on the flag), plus one pending for the admin console only when the event qualifies (its allowlist, both ingest paths): cheaper on single-consumer hosts and true to the design doc. Consequence, accepted on purpose: rows born with no send flag on get no pendings, so enabling archiving later does not pick them up — the sweep deletes them unarchived.
 
 Indexes (each consumer has its own, so a slow consumer never blocks the others):
 
@@ -274,7 +275,7 @@ Nexus dispatches on `:type`: `"telemetry-events"` batches are coerced and insert
 - **No personal data is expected.** Profile ids are random UUIDs with no PII, so the telemetry consumer needs no login guard; grouping is by `instance`, not by person (see posthog-telemetry below).
 - **The endpoint always answers 200.** Failures are swallowed and only counted (`nexus_telemetry_errors_total`), so a bad batch never breaks the sending instance; successes are counted (`nexus_telemetry_events_total`, `nexus_telemetry_reports_total`). A body the parser cannot read never reaches the handler: it answers 400 and counts in `nexus_http_malformed_requests_total`. Old instances may send a fressian+zstd base64 blob instead of a plain vector; Nexus decodes both (50 MiB cap against decompression bombs).
 
-## 12. Archive to Nexus and to the admin console
+## 12. The queue consumers: Nexus, the admin console and the events consumer
 
 ### How it works today
 
@@ -288,6 +289,26 @@ Concurrent jobs are safe: every consumer locks its own pendings (`audit_log_pend
 
 Observability (`backend/src/app/loggers/audit/consumers.clj`, definitions in `backend/src/app/main.clj:default-metrics`): one `dbg` summary per job run (`:consumer` label, sent/skipped/chunks, reaped disabled pendings, swept orphans, `:elapsed` as `"1.5s"` via `ct/format-duration`, frozen in a strict binding first because log props evaluate lazily in the logger thread), one `dbg` per admin-console send call (`ready` with totals, `sent`/`failed` with counts), plus `dbg` when the queue is empty or the job's send flag is off. Queue depth is Grafana's business: it queries the database directly, so the run performs no `GROUP BY` for the logs. No `warn` before a throw: the runner reports the aborted job (refusal details travel in the error data). Metrics, all recorded with bounded labels only: `penpot_audit_consumer_events_total{ destination = nexus|admin-console|events, outcome = archived|processed|failed|skipped }` (`skipped` = filtered by the admin-console allowlist, or nobody claimed the event; the events consumer reports `processed` instead of `archived`: it dispatches the stream, it archives nothing) and `penpot_audit_consumer_discarded_total{ reason = reaped|swept }` (entries discarded without archiving: `reaped` counts pendings dropped for disabled consumers, `swept` counts rows deleted while nobody waited for them — the silent R3 loss, now dashboard-visible instead of log-only). Timing needs no custom histogram: the jobs substrate already records execution time per job (`jobs-execution-timing`, split by job name), so each archive job is timed as itself. No queue-depth gauge: pending depth is read off the replica database directly. The bulk frontend ingest counts `penpot_audit_ingest_events_total{ kind = full|shadow, outcome = inserted|failed }` (both faces share one transaction, so a failure drops the batch on both). Archive success counts are recorded by `drain-consumer` on the drain totals: a chunk that returns normally committed its transaction (a rollback always throws instead), so they cannot inflate. Logs never carry event data (`props`, `context`, `profile-id`, `ip-addr`, `team-name`, emails): only counts, statuses and the uri host.
 
+### The events consumer: a generalist with reactions
+
+The third drain (`:audit-log-consumer-events`, flag `:event-consumer`, cron `15 */5 * * * ?`) archives nothing. It takes the **whole** stream — its queue is not selective, the producer never decides which events it wants — and hands each event to the **reactions** that claim it.
+
+A reaction is a map:
+
+```clojure
+{: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
+```
+
+- 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).
+- Each domain owns its reaction: 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. The registry lives in `backend/src/app/loggers/audit/consumers.clj`; the team-storage reaction in `backend/src/app/tasks/team_storage.clj`.
+- The producer does **not** know the reactions: it queues the whole stream, so a new reaction needs no change in the producer, only a new entry in the registry.
+- The drain only decodes the props of the events some reaction claims; the rest travel untouched.
+
+There is one reaction today: `:team-storage-usage`, which marks dirty the teams whose media storage may have changed and feeds the `media-storage-bytes-per-team` quota counter (see `mem:backend/subtleties`).
+
 ### How Nexus consumes the data (and forwards to PostHog)
 
 Stored rows are not the end: a consumer pipeline reads them back in sequence order and processes them. Each consumer remembers its place (sequence number), fetches the next chunk, runs its handler, then saves the new position. Three consumers run in order:
@@ -329,6 +350,7 @@ Config (`backend/src/app/config.clj`): `:nexus-uri`, `:telemetry-uri` (default `
 - **New Penpot frontend event**: create the data-event or implement `Event`/`PerformanceEvent` in `app.main.data.event`, with `::name`/`::type`/`::origin`. Honor `skip-audit?` if the action can resume. Remember: with no props schema, **your asserts are the contract**.
 - **New admin-console event**: server action → add the backend-log call next to the mutation, best-effort so it never breaks the action; browser action → add a new event helper plus its test.
 - **Command that creates an entity or already knows its team**: put the new row's `:id` in `::audit/props` (`create-project`/`create-file`) and set `:team-id` when the team is loaded, so the consumer never has to join production tables. For entities created inside an internal call (`clone-template`, `import-binfile`, `duplicate-project`), emit one event per entity: `audit/submit-file-created-event` when only the id is known (it reads the row back), `audit/submit-create-event` when the row is already in hand.
+- **New reaction to the events consumer**: add it to the registry in `backend/src/app/loggers/audit/consumers.clj` (`reactions`) with its `:id`, `:flag`, `:events` and `:apply`. The producer does not change (the queue is not selective). The reaction belongs to the domain that owns the affected table, and it must be idempotent: the drain is at-least-once, so the same event can reach it twice.
 - **Never** put secrets in `props`; `clean-props` is a last defense, not an excuse.
 - **Do not duplicate events**: if the backend already emits an event for an action, do not add a frontend (or admin-console browser) event for the same action — the backend one wins because it is reliable. A second event for the same action is only acceptable with a strong reason (for example the backend cannot see it at all), and then it must be clearly justified.
 - **Follow the naming policy**: use kebab-case names (`"create-organization"`, not `"createOrganization"`), and before inventing prop keys, look at similar existing events and reuse their key names (`organization-id`, `team-id`, `file-id`, `event-origin`...). Same concept, same key spelling — that is what keeps the data queryable later.

.serena/memories/backend/subtleties.md

@@ -83,7 +83,7 @@
 
 ## Storage and media
 
-- The `media-storage-bytes-per-team` quota is checked on the three commands that write media bytes: `upload-file-media-object` (multipart), `assemble-file-media-object` (chunked) and `create-file-media-object-from-url`. All three read the cached `team_storage_usage` counter (fail open + mark dirty when the row is missing; the ensure reads first and only inserts when the row is absent) instead of scanning `storage_object` per upload; the live `SUM` only runs when the `:storage-usage` + `:audit-log` + `:event-consumer` flags are not all on (without the events consumer nobody marks the teams dirty, so the counter would be frozen and the quota would read a stale value), and in the background recompute job. A third audit-log consumer (id 3, the events consumer, not selective: it takes the whole stream and each reaction decides what concerns it) marks dirty teams; `:schedule-team-storage` (cron `45 */10 * * * ?`) streams every dirty team oldest-first through a plan cursor inside one read-only transaction and enqueues one `:recompute-team-storage` per team on the dedicated `:storage` queue (never the default one, which serves the delete cascades), in chunks of 50 that commit in their own transaction via `db/own-transaction`, with a per-chunk stagger (5 s step, 60 s cap) and `skip-if-active` labelled by team id. Its query also returns the rows nobody refreshed for a week (`updated_at` older than 7 days), so a mark lost on the way heals at a bounded rate. The recompute upsert only clears dirty marks older than its read, so a mark landing mid-recompute survives for the next tick. The counter lives in `team_storage_usage` (migration `0158`): `team_id` PK/FK `ON DELETE CASCADE`, `total_bytes`, `dirty_at` (set = pending recompute), `updated_at`, partial index on `dirty_at IS NOT NULL`. `file-gc` emits `file-gc` so trims mark the team dirty; `delete-object-with-cascade` carries flat `:team-ids` so the drain never joins production tables (`hard-delete-objects` carries only its entity key and is not consumed: a physical purge cannot change the live total). The `upload-sessions-per-profile` quota is unaffected.
+- The `media-storage-bytes-per-team` quota is checked on the three commands that write media bytes: `upload-file-media-object` (multipart), `assemble-file-media-object` (chunked) and `create-file-media-object-from-url`. All three read the cached `team_storage_usage` counter (fail open + mark dirty when the row is missing; the ensure reads first and only inserts when the row is absent) instead of scanning `storage_object` per upload; the live `SUM` only runs when the `:storage-usage` + `:audit-log` + `:event-consumer` flags are not all on (without the events consumer nobody marks the teams dirty, so the counter would be frozen and the quota would read a stale value), and in the background recompute job. A third audit-log consumer (id 3, the events consumer, not selective: it takes the whole stream and each reaction decides what concerns it) marks dirty teams through the only reaction registered today, `:team-storage-usage` (`app.tasks.team-storage`, flag `:storage-usage`, registered in `app.loggers.audit.consumers`); `:schedule-team-storage` (cron `45 */10 * * * ?`) streams every dirty team oldest-first through a plan cursor inside one read-only transaction and enqueues one `:recompute-team-storage` per team on the dedicated `:storage` queue (never the default one, which serves the delete cascades), in chunks of 50 that commit in their own transaction via `db/own-transaction`, with a per-chunk stagger (5 s step, 60 s cap) and `skip-if-active` labelled by team id. Its query also returns the rows nobody refreshed for a week (`updated_at` older than 7 days), so a mark lost on the way heals at a bounded rate. The recompute upsert only clears dirty marks older than its read, so a mark landing mid-recompute survives for the next tick. The counter lives in `team_storage_usage` (migration `0158`): `team_id` PK/FK `ON DELETE CASCADE`, `total_bytes`, `dirty_at` (set = pending recompute), `updated_at`, partial index on `dirty_at IS NOT NULL`. `file-gc` emits `file-gc` so trims mark the team dirty; `delete-object-with-cascade` carries flat `:team-ids` so the drain never joins production tables (`hard-delete-objects` carries only its entity key and is not consumed: a physical purge cannot change the live total). The `upload-sessions-per-profile` quota is unaffected.
 - Storage abstraction, backend configuration, logical buckets, object lifecycle, deduplication, access rules, and garbage collection: `mem:backend/storage`.
 - SVG validation strips DOCTYPE and uses secure SAX parsing. Basic SVG info falls back to 100x100 dimensions when width/height/viewBox are missing.
 - Raster metadata is shell-derived with ImageMagick `identify`, verifies detected MIME against the supplied MIME, and swaps dimensions for EXIF orientations 6/8.