github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-10-28 21:40:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

✨ Cache per-team media storage usage from the audit log

Browse Source 
The media-storage-bytes-per-team quota ran a full SUM over
storage_object on every upload, taking seconds on large teams. A new
team_storage_usage table (migration 0158) caches the counter: a third
audit-log consumer (storage-usage) selectively queues the events that
change media storage, marks their teams dirty, and a scheduler turns
the marks into per-team recompute jobs on a dedicated :storage queue.
The quota reads the counter, always answers one row, and fails open
when the row is missing; the three commands that write media bytes pay
it.

The dirty-team stream runs in one read-only transaction and submits
each 50-team chunk through its own writable one (db/own-transaction),
so a killed run keeps what it already enqueued and the next tick
resumes the rest. The recompute only clears marks older than its read,
and the query also returns the rows nobody refreshed for a week, so a
lost mark heals at a bounded rate instead of never.

Producers resolve team ids at emission with one batched lookup, so the
drain never joins production tables: file-gc carries its team, the
delete cascade reports flat and nested ids, moves carry explicit
source/destination teams (mixed origins rejected), restores carry
theirs, and purges carry only their entity key. Selective queueing is
gated per consumer flag on both ingest paths, and telemetry shadows
never queue the admin-console consumer. An audit event that cannot be
stored now fails loudly instead of leaving a consumer stale.

The delete cascade owns library relations: the RPC only marks the file
deleted, and the job captures dependents, removes the relations and
absorbs them under a savepoint, only for survivors. Jobs queued by the
previous release still absorb through a deprecated :file-ids snapshot.

Closes #12462

AI-assisted-by: muse-spark-1.3-contributor
This commit is contained in:
Andrey Antukh 2026-10-20 09:03:26 +00:00
parent 74add05226
commit 42f6c952e1
35 changed files with 2289 additions and 246 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
.serena/memories/audit-log.md
View File

@ -57,7 +57,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, ids never reused) tracks what is left. Every record is born pending only for the consumers enabled at insert time (never a fixed both): 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.
- No `archived_at`: the `audit_log_pending` queue (migration `0157`, consumer `1` = Nexus, `2` = admin console, `3` = storage-usage — 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.
Indexes (each consumer has its own, so a slow consumer never blocks the others):
@ -129,10 +129,10 @@ The event stream is meant to let a consumer (Nexus) rebuild what it needs **with
- **Created entity ids.** `create-project` and `create-file` add the new row's `:id` (and `:project-id`/`:team-id`) to `::audit/props`, so a projection can key it.
- **Bulk operations.** `clone-template`, `import-binfile` and `duplicate-project` create entities through internal calls that emit nothing on their own; they emit a `create-file` (and `create-project`) per entity (only with `:audit-log`). The import paths go through `audit/submit-file-created-event`, which reads the light columns (`id`, `name`, `project-id`, timestamps, `revn`) back from the row — the file must really exist, a phantom id emits nothing. `duplicate-project` already holds the rows and uses `audit/submit-create-event` directly. Both binfile versions return the same shape from `import-files!` (`{:file-ids [...]}`).
- **Deletes.** `delete-team` still returns nil but carries the deleted row (`:id`, `:name`, `:created-at`, `:modified-at`, `:deleted-at`) in `::audit/props`, like `delete-project`. `delete-profile` carries `:id`, `:email`, `:fullname`, `:created-at`, `:modified-at` and `:deleted-at` (without this its props are empty: every param is namespaced and `clean-props` drops them). `create-team` adds `:created-at` and `:modified-at`. `delete-team-member` needs nothing: the delete is physical and the params already carry `:team-id`/`:member-id`.
- **Hard deletes.** `objects-gc` emits one `hard-delete-objects` event per purged chunk, carrying only that chunk's entity key (`:team-ids`, `:project-ids` or `:file-ids`; one chunk transaction purges a single entity, so one event can never span sections) with context `{:triggered-by "objects-gc"}`, only with `:audit-log` on. No other GC path emits (profiles, fonts, thumbnails); those purges stay silent.
- **Hard deletes.** `objects-gc` emits one `hard-delete-objects` event per purged chunk, carrying only that chunk's entity key (`:team-ids`, `:project-ids` or `:file-ids`; one chunk transaction purges a single entity, so one event can never span sections), with context `{:triggered-by "objects-gc"}`, only with `:audit-log` on. No team-id resolution: the live total only counts rows without `deleted_at`, which a physical purge cannot change, so no team-keyed consumer reads these events. No other GC path emits (profiles, fonts, thumbnails); those purges stay silent. `file-gc` is the exception: it emits `file-gc` (`action`, `triggered-by: cron-task`) with the cleaned `:file-id` and its `:team-id`, so the storage-usage consumer learns about trimmed media.
- **SREPL deletes.** `delete-profile!` carries the subject id in props (it used to emit nothing usable); the event profile stays zero because SREPL has no actor. `delete-team!` props already carried the id. Both travel as actions with a `triggered-by: srepl` context.
- **SREPL restores.** `restore-profile!` carries the full row (with id) in props with a zero event profile; `restore-team!` and `restore-project!` already carried their ids. The consumer resolves the subject from props by a uniform rule, with no SREPL special case.
- **Cascades.** The `:delete-object` worker task emits one `delete-object-with-cascade` event per execution, type `action`, with `:object`/`:id`/`:deleted-at` plus the affected ids nested under `:cascade` (`:team-ids`, `:project-ids`, `:file-ids`), in the same transaction. The handler submits unconditionally (`submit` fans out per flag, so no `:audit-log` re-check); the `audit_log` row itself is still only written with `:audit-log` on. Direct calls (objects-gc) and snapshot cascades stay silent on purpose.
- **Cascades.** The `:delete-object` worker task emits one `delete-object-with-cascade` event per execution, type `action`, with `:object`/`:id`/`:deleted-at`, flat `:team-ids` resolved at emission plus the affected ids nested under `:cascade` (`:team-ids`, `:project-ids`, `:file-ids`), in the same transaction. The handler submits unconditionally (`submit` fans out per flag, so no `:audit-log` re-check); the `audit_log` row itself is still only written with `:audit-log` on. Direct calls (objects-gc) and snapshot cascades stay silent on purpose.
- **Ownership.** `create-team-with-invitations` carries the new team `:id` like `create-team`; `update-team-member-role` names the demoted member in `:prev-owner-id` on owner transfers.
Entry points:
@ -277,15 +277,15 @@ Nexus dispatches on `:type`: `"telemetry-events"` batches are coerced and insert
### How it works today
Every 5 min each archive job drains its own queue (`:audit-log-consumer-nexus` on flag `:nexus`, cron `0 */5 * * * ?`; `:audit-log-consumer-admin-console` on flag `:admin-console`, cron `30 */5 * * * ?`, 30s offset so the two drains never start on the same second), in 128-row chunks taken oldest-first (`ORDER BY created_at ASC, id ASC`) with `FOR UPDATE OF audit_log` (`SET LOCAL lock_timeout = '30s'`) inside the same transaction that sends and completes them, with 100 ms pauses between chunks. Batches arrive at Nexus in chronological order (it numbers by arrival for order-sensitive consumers). Nexus receives transit `{:events [...]}` with `x-shared-key: "nexus <key>"` (`:nexus-shared-key`, or derived from the instance secret) and `origin` = public-uri; flag on but no URI raises `:task-not-configured`. The admin console receives only its allowlisted event names (enriched with team names and emails); anything filtered out counts as done for the admin-console queue. A chunk whose destination refuses it (Nexus non-204 raises `:consumer-chunk-refused`) or throws (transport failure, admin-console error) rolls back and stays for the next tick, and the job aborts as failed: it archives what it can, and the cron retries the rest. Committed chunks stay sent. The send fns return counts or throw carrying the attempted count as `:events`; `run-single-consumer` records the `failed` metric from it, outside every chunk transaction (a local DB failure carries no `:events` and stays uncounted — it is not a destination failure). No `warn` before the throw: the runner reports the aborted job. Each job touches only its own queue, so a fallen Nexus never starves the admin console (nor vice versa).
Every 5 min each archive job drains its own queue (`:audit-log-consumer-nexus` on flag `:nexus`, cron `0 */5 * * * ?`; `:audit-log-consumer-admin-console` on flag `:admin-console`, cron `30 */5 * * * ?`; `:audit-log-consumer-storage-usage` on flag `:storage-usage`, cron `15 */5 * * * ?`; the offsets keep the three drains from starting on the same second), in 128-row chunks taken oldest-first (`ORDER BY created_at ASC, id ASC`) with `FOR UPDATE OF audit_log` (`SET LOCAL lock_timeout = '30s'`) inside the same transaction that sends and completes them, with 100 ms pauses between chunks. Batches arrive at Nexus in chronological order (it numbers by arrival for order-sensitive consumers). Nexus receives transit `{:events [...]}` with `x-shared-key: "nexus <key>"` (`:nexus-shared-key`, or derived from the instance secret) and `origin` = public-uri; flag on but no URI raises `:task-not-configured`. The admin console receives only its allowlisted event names (enriched with team names and emails); anything filtered out counts as done for the admin-console queue. A chunk whose destination refuses it (Nexus non-204 raises `:consumer-chunk-refused`) or throws (transport failure, admin-console error) rolls back and stays for the next tick, and the job aborts as failed: it archives what it can, and the cron retries the rest. Committed chunks stay sent. The send fns return counts or throw carrying the attempted count as `:events`; `run-single-consumer` records the `failed` metric from it, outside every chunk transaction (a local DB failure carries no `:events` and stays uncounted — it is not a destination failure). No `warn` before the throw: the runner reports the aborted job. Each job touches only its own queue, so a fallen Nexus never starves the admin console (nor vice versa).
A chunk is **one POST** and the task advances only on 204, so a single oversized event wedges the whole queue: a crash report whose message embeds a whole payload (a serialized file, for example) can push the request past the proxy body limit and the rejection blocks every later event until it is pruned by hand. Frontend crash reports are therefore size-bounded at the source (`mem:frontend/routing-app-shell-subtleties`): `app.common.exceptions/get-hint` and the CLJS `format-throwable` clamp exception messages (`max-message-length`).
On the other side, Nexus checks `origin` against its allowlist (403 if not allowed), coerces the events, and inserts them into its own `audit_log` (partitioned by `received_at`, with an incrementing sequence number and `source`). Watch the rename: Penpot's `created-at` is stored as **`received-at`** in Nexus; Nexus's `created-at` is when it was created there. `tracked-at` keeps the origin time. Telemetry batches arriving at `telemetry.penpot.app` land in the same table through the telemetry endpoint above.
Concurrent jobs are safe: both lock rows in `created_at ASC, id ASC` order (one total order, deliberately blocking — no `SKIP LOCKED`, so archive arrival stays chronological), so the second waits for the first to commit and then sees whether it was last. Only acknowledged work deletes pendings; an admin-console failure never blocks the Nexus send of the same rows (each job owns its queue).
Concurrent jobs are safe: the two archive consumers lock rows in `created_at ASC, id ASC` order (one total order, deliberately blocking — no `SKIP LOCKED`, so archive arrival stays chronological), so the second waits for the first to commit and then sees whether it was last. The storage-usage consumer is the exception: it archives nothing and its dirty mark is idempotent, so it takes its chunk with `FOR UPDATE … SKIP LOCKED` and never waits on an archive job holding the oldest rows (a wait could hit `lock_timeout` and abort the run). Only acknowledged work deletes pendings; an admin-console failure never blocks the Nexus send of the same rows (each job owns its queue).
Observability (`backend/src/app/loggers/audit/consumer_task.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, outcome = archived|failed|skipped }` (`skipped` = filtered by the admin-console allowlist) 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 `run-single-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.
Observability (`backend/src/app/loggers/audit/consumer_task.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|storage-usage, outcome = archived|processed|failed|skipped }` (`skipped` = filtered by the admin-console allowlist; storage-usage reports `processed` instead of `archived`: it marks teams dirty, 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 `run-single-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.
### How Nexus consumes the data (and forwards to PostHog)
@ -299,7 +299,7 @@ So the full journey is: Penpot buffer → Nexus `audit_log` (via either archive
### Admin console: the second consumer
Penpot → admin-console archiving lives in its own job (`:audit-log-consumer-admin-console`, consumer `2`, flag `:admin-console`, params closed empty): it drains its own pending queue and ingests allowlisted events through `nitrate/call :ingest-audit-log`, enriched as described above. What also exists (do not confuse with archiving):
Penpot → admin-console archiving lives in its own job (`:audit-log-consumer-admin-console`, consumer `2`, flag `:admin-console`, params closed empty): it drains its own pending queue and ingests allowlisted events through `nitrate/call :ingest-audit-log`, enriched as described above. The queue itself is selective too: both ingest paths (`append-audit-entry` and bulk `insert-rows!`) only queue allowlisted names, so the drain filter is the last line of defense (legacy pendings, producer bugs), not the normal path — `skipped` is normally 0. What also exists (do not confuse with archiving):
- **Nitrate → Penpot**: `POST /api/management/methods/push-audit-events` (`backend/src/app/rpc/management/nitrate.clj`) lets the trusted Nitrate backend inject events into the Penpot audit log (no RPC auth, `::audit/skip`, request context + `submit`). Section 7a is the main user.
- **Admin-console browser → collector**: usage events from the admin-console UI traveling through the normal public `push-audit-events` (section 7b).
@ -314,6 +314,7 @@ Flags (`common/src/app/common/flags.cljc`, varia; enabled with `PENPOT_FLAGS=ena
| `:audit-log` | Stores full rows; turns on `wrap-audit` and frontend ingest |
| `:nexus` | Send flag of the Nexus archive job `:audit-log-consumer-nexus` (cron `0 */5 * * * ?`; needs `:nexus-uri` (ingest path `api/audit` appended by the job)) |
| `:admin-console` | Send flag of the admin-console archive job `:audit-log-consumer-admin-console` (cron `30 */5 * * * ?`, 30s after Nexus); also the module flag for the admin-console integration |
| `:storage-usage` | Send flag of the storage-usage consumer `:audit-log-consumer-storage-usage` (cron `15 */5 * * * ?`) and of its scheduler `:schedule-team-storage` (cron `45 */10 * * * ?`, queue `:storage`); also gates the cached path of the `media-storage-bytes-per-team` quota |
| `:audit-log-logger` | Structured `app.audit` log per event |
| `:error-reporting` | Starts the error reporters (off by default); database reporter on flag alone, Mattermost on flag plus webhook URL |
| `:telemetry` | Stores shadow rows and turns on shipping (auto with `:telemetry-enabled`) |
@ -337,6 +338,6 @@ Config (`backend/src/app/config.clj`): `:nexus-uri`, `:telemetry-uri` (default `
- Penpot frontend producer: `frontend/src/app/main/data/event.cljs` (collector, buffer, send), `frontend/src/app/main/repo.cljs` + `frontend/src/app/util/http.cljs` (headers), `frontend/src/app/main/data/nitrate.cljs` (Nitrate membership events).
- Admin-console producer (private repo, no paths cited here): server-side organization/subscription events through the management `push-audit-events`, browser-side UI events through the public `push-audit-events`.
- Consumers: `backend/src/app/loggers/webhooks.clj`, `backend/src/app/loggers/database.clj`, `backend/src/app/loggers/mattermost.clj`, `backend/src/app/tasks/telemetry.clj`.
- Archive (two jobs sharing one namespace + pending queue, no GC): `backend/src/app/loggers/audit/consumer_task.clj` (`:audit-log-consumer-nexus` / `:audit-log-consumer-admin-console` job-defs, shared drain/reap/sweep helpers), producer hook in `backend/src/app/loggers/audit.clj` (`append-audit-entry`, single `queue-pendings` helper) plus the bulk ingest in `backend/src/app/rpc/commands/audit.clj` (`handle-events`), metrics/cron wiring in `backend/src/app/main.clj` (one cron entry per job, Nexus at second 0 and admin console at second 30, each gated by its own send flag), single migration `0157` (queue, drop of `archived_at`, covering `(created_at, id)` index for the chronological scan; no backfill — pre-existing rows are swept unsent, an accepted loss).
- Archive (three jobs sharing one namespace + pending queue, no GC): `backend/src/app/loggers/audit/consumer_task.clj` (`:audit-log-consumer-nexus` / `:audit-log-consumer-admin-console` / `:audit-log-consumer-storage-usage` job-defs, shared drain/reap/sweep helpers); the storage-usage counter itself is recomputed by `backend/src/app/tasks/team_storage.clj` (`:schedule-team-storage` on cron `45 */10 * * * ?` + `:recompute-team-storage` on the `:storage` queue) and read by `app.rpc.quotes`; its table is `team_storage_usage` (`team_id` PK/FK `ON DELETE CASCADE`, `total_bytes`, `dirty_at`, `updated_at`, partial index on `dirty_at IS NOT NULL`), created by migration `0158`, producer hook in `backend/src/app/loggers/audit.clj` (`append-audit-entry`, `queue-pendings` with explicit consumers per selectivity: Nexus default-gated on its flag, storage-usage and admin-console only for qualifying events) plus the bulk ingest in `backend/src/app/rpc/commands/audit.clj` (`handle-events`, same selectivity), metrics/cron wiring in `backend/src/app/main.clj` (one cron entry per job, Nexus at second 0, admin console at second 30, storage drain at second 15 and its scheduler every 10 min at second 45, each gated by its own send flag), single migration `0157` (queue, drop of `archived_at`, covering `(created_at, id)` index for the chronological scan; no backfill — pre-existing rows are swept unsent, an accepted loss).
- Nexus (private repo, no paths cited here): archive ingest, ordered consumers (local profile mirror, PostHog SaaS events, PostHog telemetry), telemetry endpoint, analytics views.
- Tests: `backend_tests/rpc-audit-test.clj`, `rpc-team-test/accept-organization-invitation-audit-event`.

6
.serena/memories/backend/core.md
View File

@ -84,7 +84,7 @@ For worker dispatch, cron, retry semantics (`ex/raise :type ::wrk/retry` with `:
- The canonical job name is a keyword (`app.jobs/schema:job-name`): `submit` and `invoke` reject strings. The row stores text, so `jobs/get-job-def` still coerces (it is the runner's boundary), cron entries carry the job under `:job` with `d/name` only at the SQL/log boundary, and test helpers keyword-ize row names before invoking.
- The durable lifecycle metrics are submitted, dispatched, completed, retries, orphaned and rescheduled counters; queue-wait, execution and total-time histograms; dispatcher and backlog state; GC, cron and ephemeral request metrics.
- `app.metrics/run!` is safe for recording failures, but metrics components still require a valid metrics instance. Durable submit call sites carry `::mtx/metrics` through RPC or job-def configuration. The jobs substrate takes the **cfg**, not the instance, and resolves it with `app.metrics/instance`, which raises `:missing-metrics` when absent: `app.jobs`, `app.jobs.metrics`, the runner, the dispatcher, cron and the jobs GC all follow that. `app.http`, `app.rpc.climit` and `app.storage.s3.metrics` still take the instance and were left out of scope on purpose, so do not read the codebase as uniform. The six job writers resolve at the top of their body, before writing, because their metric lands in a `db/after-commit` callback that swallows exceptions. The GC, the dispatcher and cron resolve inside that callback instead, so their guarantee rests on the component schema instead. A job that calls `heartbeat` from a component cfg must carry `::mtx/metrics` on it, and the event helper records the counter from the cfg the transaction hands it.
- SQL-backed job metrics are registered with `app.db/after-commit` from INSIDE the transaction that changed the row, so a rollback cannot inflate submitted, terminal, retry, event or dispatcher counters. Every lifecycle writer owns that transaction (`app.jobs/own-transaction` drops `::db/conn`), and `db/transact!` gives a transaction that joins no other its own callback context, so the write and its metric commit or vanish together. `submit` is the exception and joins the caller on purpose: creating a job is part of the caller's business work.
- SQL-backed job metrics are registered with `app.db/after-commit` from INSIDE the transaction that changed the row, so a rollback cannot inflate submitted, terminal, retry, event or dispatcher counters. Every lifecycle writer owns that transaction (`app.db/own-transaction` (drops `::db/conn` and `::db/read-only`, so a write can own its transaction even inside a read-only caller) drops `::db/conn`), and `db/transact!` gives a transaction that joins no other its own callback context, so the write and its metric commit or vanish together. `submit` is the exception and joins the caller on purpose: creating a job is part of the caller's business work.
- `penpot_jobs_events_total` counts stored job events with no labels: the interesting split is `kind`, and grouping a growing append-only table to build a metric is not worth the query on the write path.
- The backlog sampler is `::jobs-metrics/sampler`; it runs every 30 seconds on worker-enabled, writable systems, groups by status and publishes the oldest pending age, and does not start on a read-only database.
- `penpot_tasks_timing` remains exported for compatibility while the jobs-specific histograms are adopted.
@ -125,8 +125,8 @@ Fixtures can populate local data for manual testing/perf work. From the backend
* **Type Hinting:** Use explicit JVM type hints (e.g. `^String`, `^long`) in performance-critical paths to avoid reflection overhead.
* **Batch inserts:** Use `db/insert-many!` for bulk row inserts — generates a single SQL with multiple parameter tuples. Avoid on very large datasets (SQL length / parameter count limits).
* **Server-side cursors:** Use `db/plan` (fetch-size 1000, forward-only, read-only) for large result sets. Never fetch large collections into memory at once. `db/plan` is an `IReduceInit`: `reduce`/`transduce` over it (`sequence` does not work on it), and with `partition-all` the transducer's completing arity is what flushes the last partial chunk. A plan row is a **view over the open result set**, valid only inside the reducing function: read the columns you need right there — buffering rows to read them later (a `partition-all` buffer, a lazy seq) silently yields nil columns. `db/cursor` (manual `DECLARE`/`FETCH`) is a leftover with no callers left.
* **Transaction discipline:** Use `tx-run!` for writes (opens a transaction), `run!` for reads (single connection, no transaction). Set `:read-only` on `tx-run!` when applicable to let PostgreSQL optimize.
* **Server-side cursors:** Use `db/plan` (fetch-size 1000, forward-only, read-only) for large result sets, always inside `tx-run!` (prefer `::db/read-only true`): the pool runs with auto-commit on, and the PostgreSQL JDBC driver only honors `fetch-size` (real streaming cursor) with auto-commit off, so a `plan` outside a transaction silently materializes the whole `ResultSet` in memory. Never fetch large collections into memory at once. `db/plan` is an `IReduceInit`: `reduce`/`transduce` over it (`sequence` does not work on it), and with `partition-all` the transducer's completing arity is what flushes the last partial chunk; the `reduce` holds one connection (and, inside `tx-run!`, the whole transaction) open until it finishes, so keep chunk transactions separate from the plan one. A plan row is a **view over the open result set**, valid only inside the reducing function: read the columns you need right there — buffering rows to read them later (a `partition-all` buffer, a lazy seq) silently yields nil columns. `db/cursor` (manual `DECLARE`/`FETCH`) is a leftover with no callers left.
* **Transaction discipline:** Use `tx-run!` for writes (opens a transaction), `run!` for reads (single connection, no transaction) — except `db/plan`, which always needs `tx-run!` (see above). Set `:read-only` on `tx-run!` when applicable to let PostgreSQL optimize.
## Lint and Format

3
.serena/memories/backend/subtleties.md
View File

@ -29,6 +29,7 @@
- Whether a call is nested is decided by the CONNECTION, not by an ambient flag: `db/transact!` binds `*after-commit-conn*` around the body, and a transactable that is not identical to that connection is an independent unit of work that owns its own after-commit context. A `tx-run!` that passes a pool while a transaction is open therefore commits on its own and drains its own callbacks, instead of leaving them in a context that may roll back.
- `after-commit` callbacks run on commit only. A transaction marked `::db/rollback` (used by `srepl` and `app.tasks/offload-file-data`) commits nothing, so it runs none of them.
- `db/run!` opens/reuses one connection but does not create a transaction.
- `db/own-transaction` returns a cfg that owns its next transaction: it drops `::db/conn` (so `tx-run!` opens its own) and `::db/read-only` (so a caller running read-only, like the storage scheduler streaming a plan cursor, cannot make the nested write read-only too). Lifecycle writes use it so a rollback of the surrounding work can undo neither the write nor its metric.
- `db/tjson` is Transit JSON for jsonb storage; `db/json` is plain JSON. Job params (`job.params`) are plain JSON, decoded with the job-def decoder (`app.jobs/decode-params`); only the legacy `task` props were Transit (`decode-transit-pgobject`). Job results (`job.result`) are plain JSON too: they are read with the result decoder of the job-def (`app.jobs/decode-result`), and a result map keyed by uuid must declare its schema (`[:map-of ::sm/uuid ...]`, the import resolution does) or every key comes back as a keyword and every uuid as the text it is on disk.
- Advisory transaction locks accept UUIDs or ints. UUID locks are hashed using a zero-UUID seeded siphash.
@ -80,7 +81,7 @@
## Storage and media
- The `media-storage-bytes-per-team` quota check on `upload-file-media-object` is TEMPORARILY DISABLED (2026-10-19, kept commented in `app.rpc.commands.media` with its `sql:get-team-id-for-file` lookup still active for the audit props): the quote performs a full `SUM` over `storage_object` per upload and was too slow in production. The `upload-sessions-per-profile` quota is unaffected and stays active. Re-enable only with a cheaper approximation.
- 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` flags are off, and in the background recompute job. A third audit-log consumer (id 3, selective pendings) 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.
- 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.

2
.serena/memories/prod-infra/core.md
View File

@ -14,7 +14,7 @@ Backend (`app.config`, `PENPOT_*` env vars) is parameterized; deployments choose
Async jobs are enqueued via `jobs/submit` (`app.jobs`), which inserts a row into the shared Postgres `job` table tagged with `queue = "<tenant>:<queue-name>"`. Submission is **fire-and-forget** — RPC handlers never poll and never wait. The completion signal is the `job` row's `status` / `completed_at` columns, which the creation commands read back to answer with the state of the job they just created. A job submitted with `::jobs/profile-id` also publishes a `:job-event` message on that profile's msgbus topic after each transaction commits, and the binfile commands (`create-export-binfile-job`, `create-import-binfile-job`) are the callers that pass that owner, so the progress of their jobs does reach the msgbus. Soft-delete RPCs return immediately after marking the top-level row, leaving the cascade and reaping to workers.
Workers run on backends with `enable-backend-worker` in `PENPOT_FLAGS`. Each worker-enabled backend has a `dispatcher` (polls `job` with `FOR UPDATE SKIP LOCKED`, marks status='scheduled', RPUSHes the JSON `[id scheduled-at]` payloads into **its own** Redis list) and one or more `runner`s per queue (BLPOP from that same local list, execute, update the Postgres row). The Redis hand-off list is purely intra-backend — cross-backend coordination happens at the Postgres row level. The queues are `default`, `webhooks`, `cron` and `binfile`: the last one carries the heavy work of a user job (`:worker-binfile-parallelism`, default 1), so a long export or import does not block the default runner.
Workers run on backends with `enable-backend-worker` in `PENPOT_FLAGS`. Each worker-enabled backend has a `dispatcher` (polls `job` with `FOR UPDATE SKIP LOCKED`, marks status='scheduled', RPUSHes the JSON `[id scheduled-at]` payloads into **its own** Redis list) and one or more `runner`s per queue (BLPOP from that same local list, execute, update the Postgres row). The Redis hand-off list is purely intra-backend — cross-backend coordination happens at the Postgres row level. The queues are `default`, `webhooks`, `cron`, `binfile` and `storage`: `binfile` carries the heavy work of a user job (`:worker-binfile-parallelism`, default 1) and `storage` the per-team media SUM of the storage-usage counter (`:worker-storage-parallelism`, default 2), so neither a long export/import nor the recompute backlog blocks the default runner that serves the delete cascades.
## Cross-backend safety

1
backend/src/app/config.clj
View File

@ -219,6 +219,7 @@
[:worker-webhook-parallelism {:optional true} ::sm/int]
[:worker-cron-parallelism {:optional true} ::sm/int]
[:worker-binfile-parallelism {:optional true} ::sm/int]
[:worker-storage-parallelism {:optional true} ::sm/int]
[:database-password {:optional true} [:maybe :string]]
[:database-uri {:optional true} ::sm/uri]

13
backend/src/app/db.clj
View File

@ -648,6 +648,19 @@
(finally
(reset! context [])))))))
(defn own-transaction
"A cfg for a write that must own its transaction: drops `::db/conn`
so the next `tx-run!` opens its own transaction on its own
connection, and `::db/read-only` so that transaction is writable
even when the caller runs read-only (as the storage scheduler does
for its plan cursor). It never joins the transaction of whoever
called it; because that transaction joins no other, it owns its
after-commit context too, so the write and its metric commit or
vanish together and a rollback of the surrounding work can undo
neither."
[cfg]
(dissoc cfg ::conn ::read-only))
(defn tx-run!
"Run a function in a transaction."
[system f & params]

31
backend/src/app/jobs.clj
View File

@ -716,21 +716,6 @@
:created-at created-at})))
{:event-id id :created-at created-at :kind kind :payload payload})))
(defn- own-transaction
"A cfg for a job write that must own its transaction.
A lifecycle write reports on work that already happened elsewhere, so
it never joins the transaction of whoever called it: the `::db/conn` is
dropped and `db/tx-run!` opens its own transaction on its own
connection. Because that transaction joins no other, it owns its
after-commit context too, so the write and its metric commit or vanish
together and a rollback of the surrounding work can undo neither.
`submit` is the deliberate exception: creating a job is part of the
business unit of work of whoever asks for it."
[cfg]
(dissoc cfg ::db/conn))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; HEARTBEAT / PROGRESS
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
@ -825,10 +810,10 @@
here, so a progress event can never land on an already terminal job.
Returns 1 when the event was stored.
Like every lifecycle write it owns its transaction (see `own-transaction`),
Like every lifecycle write it owns its transaction (see `db/own-transaction`),
so a beat survives the rollback of the work around it."
[cfg job-id progress]
(db/tx-run! (own-transaction cfg)
(db/tx-run! (db/own-transaction cfg)
(fn [{:keys [::db/conn] :as tx-cfg}]
(if (db/exec-one! conn [sql:lock-active-job job-id])
(do
@ -940,7 +925,7 @@
;; Every metric is registered with `db/after-commit` from inside the
;; transaction, next to the event, so it lands on the outermost commit
;; whatever that is: a rolled back transition is never counted. Each
;; writer owns that transaction (see `own-transaction`), so the commit
;; writer owns that transaction (see `db/own-transaction`), so the commit
;; in question is always its own.
;;
;; That callback swallows exceptions, which is why every writer below
@ -1042,7 +1027,7 @@
attempt this run consumes from the retry budget."
[cfg job-id scheduled-at]
(assert (mtx/instance cfg) "missing ::mtx/metrics on the jobs cfg")
(db/tx-run! (own-transaction cfg)
(db/tx-run! (db/own-transaction cfg)
(fn [{:keys [::db/conn] :as tx-cfg}]
(if-let [{:keys [retry-num]}
(db/exec-one! conn [sql:claim-job job-id scheduled-at])]
@ -1074,7 +1059,7 @@
(assert (mtx/instance cfg) "missing ::mtx/metrics on the jobs cfg")
(let [job (get-job cfg job-id)
encoded (db/json (check-job-error error))]
(db/tx-run! (own-transaction cfg)
(db/tx-run! (db/own-transaction cfg)
(fn [{:keys [::db/conn] :as tx-cfg}]
(if-let [{:keys [retry-num]}
(db/exec-one! conn [sql:retry-job (ct/now) scheduled-at
@ -1149,7 +1134,7 @@
[cfg & {:keys [job-id result resource-id]}]
(assert (mtx/instance cfg) "missing ::mtx/metrics on the jobs cfg")
(check-resource-id resource-id)
(let [n (db/tx-run! (own-transaction cfg)
(let [n (db/tx-run! (db/own-transaction cfg)
(fn [{:keys [::db/conn] :as tx-cfg}]
(let [job (some-> (db/exec-one! conn [sql:lock-job job-id])
decode-row)
@ -1204,7 +1189,7 @@
(assert (mtx/instance cfg) "missing ::mtx/metrics on the jobs cfg")
(let [job (get-job cfg job-id)
encoded (db/json (check-job-error error))
n (db/tx-run! (own-transaction cfg)
n (db/tx-run! (db/own-transaction cfg)
(fn [{:keys [::db/conn] :as tx-cfg}]
(if (db/exec-one! conn [sql:fail-job (ct/now) encoded job-id])
(do
@ -1237,7 +1222,7 @@
[cfg job-id]
(assert (mtx/instance cfg) "missing ::mtx/metrics on the jobs cfg")
(let [job (get-job cfg job-id)
n (db/tx-run! (own-transaction cfg)
n (db/tx-run! (db/own-transaction cfg)
(fn [{:keys [::db/conn] :as tx-cfg}]
(if (db/exec-one! conn [sql:cancel-job (ct/now) job-id])
(do

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

@ -48,6 +48,17 @@
"incomplete-user-info"
"unexpected-error"})
(def ^:private telemetry-sources
#{"telemetry:backend" "telemetry:frontend"})
(defn telemetry-shadow-source?
"True for the anonymized shadow sources of the telemetry consumer. A
selective consumer must skip them: the full row already queues, and
the shadow keeps the event name, so queueing it would ingest the
same event twice (once full, once anonymized)."
[source]
(contains? telemetry-sources (str source)))
(def ^:private safe-backend-context-keys
#{:version
:initiator
@ -244,10 +255,27 @@
(update :ip-addr db/inet))
params (select-keys params event-keys)
row (db/insert! cfg :audit-log params)]
;; The consumer queue follows the row in the same transaction: one
;; pending per enabled consumer. A rollback takes both on its way
;; out, so a row never outlives its queue.
(consumer/queue-pendings cfg [(:id row)])
;; The consumer queue follows the row in the same transaction: a
;; rollback takes both on its way out, so a row never outlives its
;; queue. Nexus takes every row; the two `when`s below add the
;; pendings of the selective consumers (storage-usage,
;; admin-console), each gated on its own flag.
(consumer/queue-nexus-pendings cfg [(:id row)])
;; The storage-usage consumer is selective: only the events that
;; can change per-team media storage queue its pending, and never
;; for telemetry shadows (the full row already queues).
(when (and (contains? cf/flags :storage-usage)
(contains? consumer/storage-usage-event-names (:name params))
(not (telemetry-shadow-source? (:source params))))
(consumer/queue-pendings cfg [(:id row)] consumer/consumer-storage-usage))
;; The admin-console consumer is selective too: only allowlisted
;; names queue its pending, and never a telemetry shadow.
(when (and (contains? cf/flags :admin-console)
(contains? consumer/event-names-for-admin-console (:name params))
(not (telemetry-shadow-source? (:source params))))
(consumer/queue-pendings cfg [(:id row)] consumer/consumer-admin-console))
row))
(def ^:private xf:filter-telemetry-props
@ -333,13 +361,13 @@
event))
(defn submit*
"A public API, lower-level than submit, assumes all required fields are filled"
"A public API, lower-level than submit, assumes all required fields are
filled. A failure propagates: an event that cannot be stored is lost,
and a lost event is a stale consumer (the media-storage counter would
never learn about it). Swallowing it here made that invisible."
[cfg event]
(try
(let [event (check-event event)]
(db/tx-run! cfg process-event event))
(catch Throwable cause
(l/error :hint "unexpected error processing event" :cause cause))))
(let [event (check-event event)]
(db/tx-run! cfg process-event event)))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; PUBLIC API
@ -418,15 +446,57 @@
coerced))
(def ^:private sql:team-id-by-project
"SELECT id FROM team
WHERE id = (SELECT team_id FROM project WHERE id = ?)")
(def ^:private sql:team-ids-by-projects
"SELECT team_id
FROM project
WHERE id = ANY(?::uuid[])")
(def ^:private sql:team-id-by-file
"SELECT t.id FROM team AS t
JOIN project AS p ON (p.team_id = t.id)
JOIN file AS f ON (f.project_id = p.id)
WHERE f.id = ?")
(def ^:private sql:team-ids-by-files
"SELECT p.team_id AS team_id
FROM file AS f
JOIN project AS p ON (p.id = f.project_id)
WHERE f.id = ANY(?::uuid[])")
(def ^:private xf:team-ids
"Transducer: pull the team id out of a team/project/file resolution
row. Allocated once so hot paths (every audited RPC) do not rebuild
it per call."
(map :team-id))
(defn- resolve-team-ids
"Resolve one bag of project or file ids to team ids with one ANY(?)
query on a plain connection. The result holds one tiny row per id,
so neither a cursor nor a transaction is needed."
[cfg kind ids]
;; Fetch inside the connection, reduce outside it: the rows are
;; already materialized, so the pool connection is held as briefly
;; as possible.
(->> (db/run! cfg
(fn [{:keys [::db/conn]}]
(db/exec! conn [(case kind
:project sql:team-ids-by-projects
:file sql:team-ids-by-files)
(db/create-array conn "uuid" ids)])))
(into #{} xf:team-ids)))
(defn get-team-ids
"Resolve a mixed bag of team, project and file ids to team ids with
two batched lookups. Producers of cascade-style events call it so
consumers never join production tables. The pure part runs without
touching the database; each lookup resolves its connection lazily,
only when it has ids (reusing the caller transaction). Every id
reaching this function is a UUID already (RPC params and frontend
ingest travel as transit, jobs read DB rows): anything else is a
bug at the source, never coerced here."
[cfg {:keys [team-ids project-ids file-ids]}]
(let [direct (into #{} team-ids)]
(-> direct
(into (if (seq project-ids)
(resolve-team-ids cfg :project project-ids)
#{}))
(into (if (seq file-ids)
(resolve-team-ids cfg :file file-ids)
#{})))))
(def ^:private read-event-names
"Read commands excluded from team-id resolution on top of the `get-*`
@ -453,9 +523,9 @@
(let [project-id (:project-id props)
file-id (:file-id props)
team-id (or (when (some? project-id)
(:id (db/exec-one! cfg [sql:team-id-by-project project-id])))
(first (get-team-ids cfg {:project-ids [project-id]})))
(when (some? file-id)
(:id (db/exec-one! cfg [sql:team-id-by-file file-id]))))]
(first (get-team-ids cfg {:file-ids [file-id]}))))]
(cond-> props
(some? team-id) (assoc :team-id team-id)))))

325
backend/src/app/loggers/audit/consumer_task.clj
View File

@ -26,13 +26,14 @@
;; This task drains the audit log buffer into the long-term archives.
;;
;; Two separate jobs share this namespace and the queue helpers, one
;; per destination: `:audit-log-consumer-nexus` and
;; `:audit-log-consumer-admin-console`. Each job drains only its own queue
;; Three separate jobs share this namespace and the queue helpers, one
;; per destination: `:audit-log-consumer-nexus`,
;; `:audit-log-consumer-admin-console` and
;; `:audit-log-consumer-storage-usage`. 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 both jobs on purpose:
;; 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
@ -40,10 +41,17 @@
;; previous owner's queue.
(def consumer-nexus 1)
(def consumer-admin-console 2)
(def consumer-storage-usage 3)
(def ^:private consumers
[{:id consumer-nexus :flag :nexus :label "nexus"}
{:id consumer-admin-console :flag :admin-console :label "admin-console"}])
[{:id consumer-nexus :flag :nexus :label "nexus"
:outcome "archived"}
{:id consumer-admin-console :flag :admin-console :label "admin-console"
:outcome "archived"}
;; marks teams dirty, it archives nothing
{:id consumer-storage-usage :flag :storage-usage :label "storage-usage"
:outcome "processed"
:summary "teams marked dirty"}])
(def ^:private xf:enabled-consumers
"Static transducer: keep the consumers whose send flag is on.
@ -53,33 +61,116 @@
(map :id)))
(defn enabled-consumers
"Consumers allowed to work right now, from the send flags. Public:
producers queue every record for exactly these consumers."
"Consumers allowed to work right now, from the send flags. A test
helper: producers name their consumer explicitly
(`queue-nexus-pendings`, or `queue-pendings` with an id at the
selective call sites) and housekeeping walks the private table."
[]
(into [] xf:enabled-consumers consumers))
(defn queue-pendings
"Queue records for every enabled consumer, nothing when there is
nothing to queue or no consumer is on. One multi-row statement for
all ids instead of one per consumer: this runs inside every audited
RPC and the bulk frontend ingest. Runs inside the row transaction:
a rollback takes both."
[cfg ids]
(let [consumers (enabled-consumers)]
(when (and (seq ids) (seq consumers))
(db/insert-many! cfg :audit-log-pending
[:audit-log-id :consumer]
(for [id ids
consumer consumers]
[id consumer])
{::sql/on-conflict-do-nothing true}))))
(def ^:private single-insert-max-ids
"Below this many pendings, a single `insert-many!` stays far below
the PostgreSQL parameter limit; above it, chunk. An if is cheaper
than the chunking overhead on the hot path (every audited RPC)."
1000)
(def ^:private consumer-labels
(into {} (map (juxt :id :label)) consumers))
(defn queue-pendings
"Queue records for one consumer, nothing when there is nothing to
queue. One multi-row statement for all ids, chunked past
`single-insert-max-ids`: this runs inside every audited RPC and the
bulk frontend ingest. Runs inside the row transaction: a rollback
takes both. Callers name their consumer explicitly: nexus through
`queue-nexus-pendings`, the selective ones at their own call sites."
([cfg ids consumer-id]
(when (seq ids)
(let [rows (mapv (fn [id] [id consumer-id]) ids)]
(if (< (count rows) single-insert-max-ids)
(db/insert-many! cfg :audit-log-pending
[:audit-log-id :consumer]
rows
{::sql/on-conflict-do-nothing true})
(db/insert-many-chunked! cfg :audit-log-pending
[:audit-log-id :consumer]
rows
{::sql/on-conflict-do-nothing true}))))))
(defn queue-nexus-pendings
"Queue records for nexus, the only non-selective consumer. Gated on
its flag here so callers never rely on what the `queue-pendings`
default happens to exclude today."
[cfg ids]
(when (contains? cf/flags :nexus)
(queue-pendings cfg ids consumer-nexus)))
(def ^:private default-team-keys
"Key map of the events that carry a single :team-id in props."
{:paths [[:team-id]]
:keys [:team-id]})
(def ^:private team-keys-by-name
"Single source of the storage-usage contract: event name -> where its
team ids live. `:paths` are the prop paths `team-ids-of-event` reads;
`:keys` are the top-level props `carries-team-keys?` checks with
`contains?`, so an explicit empty collection counts as a fact and not
as a producer that lost its keys. Adding a name here also adds it to
`storage-usage-event-names`, so the two cannot drift."
{"upload-file-media-object" default-team-keys
"create-file-media-object-from-url" default-team-keys
"clone-file-media-object" default-team-keys
"assemble-file-media-object" default-team-keys
;; Files created in bulk (import, clone, duplicate) may carry media,
;; so they can change the total even though create-file itself adds
;; no bytes: do not drop it, and do not copy it to cheaper events.
"create-file" default-team-keys
"create-font-variant" default-team-keys
"delete-font" default-team-keys
"delete-font-variant" default-team-keys
"delete-file" default-team-keys
"delete-object-with-cascade" {:paths [[:team-ids] [:cascade :team-ids]]
:keys [:team-ids :cascade]}
"move-files" {:paths [[:source-team-id] [:dest-team-id]]
:keys [:source-team-id :dest-team-id]}
"move-project" {:paths [[:source-team-id] [:dest-team-id]]
:keys [:source-team-id :dest-team-id]}
"permanently-delete-team-files" default-team-keys
"restore-deleted-team-files" default-team-keys
"restore-file" default-team-keys
"restore-profile" {:paths [[:team-ids] [:cascade :team-ids]]
:keys [:team-ids :cascade]}
"restore-project" default-team-keys
"restore-team" default-team-keys
"file-gc" default-team-keys})
(def storage-usage-event-names
"Audit event names that can change per-team media storage. Only these
queue a pending for the storage-usage consumer; everything else skips
it, so unrelated traffic pays no extra row and the drain stays lean.
Derived from `team-keys-by-name` so the allowlist and the key map
cannot disagree. Purges (`hard-delete-objects`) are excluded on
purpose: the physical delete only removes rows the live SUM already
ignores (everything with `deleted_at` set), so queueing them would
recompute the same total — and purged teams would violate the usage
table foreign key."
(set (keys team-keys-by-name)))
(def ^:private consumers-by-id
(into {} (map (juxt :id identity)) consumers))
(defn- consumer-label
[consumer]
(get consumer-labels consumer))
(get-in consumers-by-id [consumer :label]))
(defn- consumer-outcome
"Success outcome of a consumer in the events metric: storage-usage
marks teams dirty, it archives nothing."
[consumer]
(get-in consumers-by-id [consumer :outcome]))
(defn- consumer-summary
"Run-summary hint of a consumer: the default ones archive, so only
the exception names its own."
[consumer]
(get-in consumers-by-id [consumer :summary] "events archived"))
(defn- decode-row
[{:keys [props ip-addr context] :as row}]
@ -135,6 +226,20 @@
LIMIT 128
FOR UPDATE OF a")
(def ^:private sql:pending-chunk-unordered
"The same chunk for a consumer that does not archive and does not
care about order (storage-usage only marks teams dirty, and the mark
is idempotent). SKIP LOCKED keeps it from waiting on an archive job
holding the oldest rows: that wait could hit lock_timeout and abort
the run."
"SELECT a.*
FROM audit_log_pending p
JOIN audit_log a ON a.id = p.audit_log_id
WHERE p.consumer = ?::smallint
ORDER BY a.created_at ASC, a.id ASC
LIMIT 128
FOR UPDATE OF a SKIP LOCKED")
(def ^:private sql:delete-pendings
"DELETE FROM audit_log_pending
WHERE consumer = ?::smallint
@ -150,8 +255,11 @@
(defn- get-pending-rows
[{:keys [::db/conn] :as cfg} consumer]
(->> (db/exec! conn [sql:pending-chunk consumer])
(not-empty)))
(let [sql (if (= consumer consumer-storage-usage)
sql:pending-chunk-unordered
sql:pending-chunk)]
(->> (db/exec! conn [sql consumer])
(not-empty))))
(defn- complete-batch
"Delete our pendings for the batch, then the records nobody is
@ -206,7 +314,7 @@
:resp-body (truncate-body (:body resp) 500)
:events n)))))
(def ^:private event-names-for-admin-console
(def event-names-for-admin-console
#{"accept-organization-invitation"
"accept-organization-invitation-from"
"accept-team-invitation"
@ -396,7 +504,7 @@
[cfg]
(db/exec! cfg ["SET LOCAL lock_timeout = '30s'"])
(when-let [rows (get-pending-rows cfg consumer-nexus)]
(let [events (mapv row->event (mapv decode-row rows))
(let [events (mapv (comp row->event decode-row) rows)
;; send-to-nexus returns the count or throws: a refusal
;; aborts the job, the rows stay for the next tick.
n (send-to-nexus cfg events)]
@ -427,6 +535,114 @@
(complete-batch cfg consumer-admin-console rows)
admin-console-res)))
(def ^:private sql:mark-storage-usage-dirty
"INSERT INTO team_storage_usage AS tsu (team_id, dirty_at)
SELECT id, now() FROM unnest(?::uuid[]) AS u(id)
WHERE EXISTS (SELECT 1 FROM team AS t WHERE t.id = u.id)
ON CONFLICT (team_id) DO UPDATE
SET dirty_at = EXCLUDED.dirty_at
WHERE tsu.dirty_at IS NULL")
(defn- team-ids-of-event
"Team ids of one storage-usage event, read through its key map (see
team-keys-by-name). Throws on a present non-UUID: a producer bug,
never coerced; the caller logs the row and skips that event."
[{:keys [name props]}]
(let [{:keys [paths]} (get team-keys-by-name name default-team-keys)
;; A path holds either one id (:team-id) or a collection
;; (:team-ids): normalize both to a seq.
ids (mapcat (fn [path]
(let [v (get-in props path)]
(cond (nil? v) nil
(coll? v) v
:else [v])))
paths)]
(doseq [id ids :when (some? id)]
(when-not (uuid? id)
(ex/raise :type :validation
:code :non-uuid-team-id
:hint "non-uuid team id in storage-usage event"
:event name)))
ids))
(defn- carries-team-keys?
"True when the event carries any of the team keys its key map names,
even an empty collection: an explicit empty list (a profile that owns
no team) is a fact, not a producer bug."
[{:keys [name props]}]
(let [{:keys [keys]} (get team-keys-by-name name default-team-keys)]
;; `some?` on the value, not `contains?`: an explicit empty
;; collection counts (a fact), a nil value does not (the producer
;; failed to resolve its team).
(boolean (some #(some? (get props %)) keys))))
(defn- collect-storage-usage-team-ids
"Team ids whose media storage may have changed in this chunk, read
straight from props: producers key their events by team so the drain
never joins production tables. Each event name maps to its keys
explicitly; anything else falls back to the single :team-id in
props. A present non-UUID is a producer bug (see get-team-ids, never
coerced): that event is logged with its row id and skipped, so one
bad row cannot wedge the queue. The key map lives in
team-keys-by-name; storage-usage-queueing-covers-every-listed-name
pins the real prop shape of every listed name."
[entries]
(into #{}
(comp (map (fn [[id event]]
(try
(team-ids-of-event event)
(catch Throwable cause
(l/warn :hint "skipping storage-usage event with invalid team ids"
:audit-log-id (str id)
:event (:name event)
:cause cause)
nil))))
(mapcat (fn [ids] (or ids [])))
(filter some?))
entries))
(defn- mark-storage-usage-dirty
[conn team-ids]
(when (seq team-ids)
;; Keeps the oldest dirty mark: a team marked twice in one window
;; is recomputed once.
(db/exec-one! conn [sql:mark-storage-usage-dirty
(db/create-array conn "uuid" team-ids)])))
(defn- consumer-storage-usage-chunk
"Mark dirty the teams changed in one pending chunk and complete it.
Returns {:sent :skipped} like the other chunk fns: relevant events
count as sent, anything else (defense in depth, the queue is
selective) as skipped. Takes a transactional cfg, no metrics here:
same contract as the other chunk fns."
[cfg]
(db/exec! cfg ["SET LOCAL lock_timeout = '30s'"])
(when-let [rows (get-pending-rows cfg consumer-storage-usage)]
(let [pairs (mapv (fn [row]
[(:id row) ((comp row->event decode-row) row)])
rows)
relevant (into []
(filter (fn [[_ event]]
(contains? storage-usage-event-names (:name event))))
pairs)
skipped (- (count pairs) (count relevant))
teams (collect-storage-usage-team-ids relevant)
_ (mark-storage-usage-dirty (::db/conn cfg) teams)
;; No relevant event carries any team key: a producer
;; stopped keying its events. An explicit empty collection
;; does not count (see carries-team-keys?).
_ (when (and (seq relevant)
(not-any? (comp carries-team-keys? second) relevant))
(l/warn :hint "storage-usage chunk resolved no team ids"
:events (count relevant)
:ids (mapv (comp str first) relevant)))]
(l/trc :hint "consumer events chunk"
:consumer "storage-usage"
:events (count relevant)
:teams (count teams))
(complete-batch cfg consumer-storage-usage rows)
{:sent (count relevant) :skipped skipped})))
(defn- drain-consumer
"Drain one consumer queue chunk by chunk, one transaction per chunk.
Returns totals; a chunk the destination refuses (or that throws)
@ -501,6 +717,7 @@
(declare execute-audit-log-consumer-nexus)
(declare execute-audit-log-consumer-admin-console)
(declare execute-audit-log-consumer-storage-usage)
(def schema:audit-log-consumer-nexus-params
"Overrides for the nexus job repl invocation defaults.
@ -562,19 +779,43 @@
::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-storage-usage-params
"Params of the storage-usage 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 ::storage-usage-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 ::storage-usage-job-def
[_ cfg]
(l/dbg :hint "audit log consumer job initialized"
:consumer "storage-usage")
{::jobs/name :audit-log-consumer-storage-usage
::jobs/schema schema:audit-log-consumer-storage-usage-params
::jobs/handler
(fn [_context params]
(execute-audit-log-consumer-storage-usage cfg params))
::jobs/decoder (sm/decoder schema:audit-log-consumer-storage-usage-params sm/json-transformer)
::jobs/validator (sm/validator schema:audit-log-consumer-storage-usage-params)})
(defn- run-single-consumer
"Drain one consumer queue, then housekeeping (reap disabled
pendings, sweep orphans) and the per-run summary. Housekeeping lives
in both jobs: only the enabled job runs, so it must reap the queues
in every job: only the enabled job runs, so it must reap the queues
of the consumers that are off; both halves are idempotent deletes.
A refused chunk (or a throw) aborts the job: the committed chunks
stay sent, the rest waits for the next tick, and the cron retries.
Every effect is rows and logs (drain metrics land here, outside
every chunk transaction)."
[cfg consumer send-chunk tpoint]
(let [label (consumer-label consumer)
metrics (mtx/instance cfg)
reaped (reap-disabled cfg)
(let [label (consumer-label consumer)
ok-outcome (consumer-outcome consumer)
metrics (mtx/instance cfg)
reaped (reap-disabled cfg)
;; All drain metrics land here, outside every chunk
;; transaction. Success is recorded on the totals: a normal
;; chunk return means its transaction committed (a rollback
@ -592,7 +833,7 @@
(throw cause)))
_ (let [{:keys [sent skipped]} drained]
(when (pos? sent)
(record-count metrics :audit-consumer-events [label "archived"] sent))
(record-count metrics :audit-consumer-events [label ok-outcome] sent))
(when (pos? skipped)
(record-count metrics :audit-consumer-events [label "skipped"] skipped)))
swept (sweep-orphans cfg)
@ -608,7 +849,7 @@
(zero? swept))
(l/dbg :hint "no pending audit events"
:consumer label)
(l/dbg :hint "events archived"
(l/dbg :hint (consumer-summary consumer)
:consumer label
:elapsed elapsed
:sent sent
@ -638,9 +879,17 @@
(defn execute-audit-log-consumer-admin-console
"Admin-console job handler: drain the admin-console queue in chunks,
then clean up (heartbeat per iteration: the sent chunk batches can
be long)."
then clean up (heartbeat per iteration: the batches can be long)."
[cfg _params]
(if-not (contains? cf/flags :admin-console)
(l/dbg :hint "audit log consumer (admin-console) skipped, flag off")
(run-single-consumer cfg consumer-admin-console consumer-admin-console-chunk (ct/tpoint))))
(defn execute-audit-log-consumer-storage-usage
"Storage-usage job handler: drain the storage-usage queue in chunks,
marking dirty the teams whose media storage may have changed
(heartbeat per iteration: a burst can span many chunks)."
[cfg _params]
(if-not (contains? cf/flags :storage-usage)
(l/dbg :hint "audit log consumer (storage-usage) skipped, flag off")
(run-single-consumer cfg consumer-storage-usage consumer-storage-usage-chunk (ct/tpoint))))

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

@ -647,6 +647,8 @@
:run-webhook (ig/ref ::webhooks/run-webhook-job-def)
:process-webhook-event (ig/ref ::webhooks/process-webhook-event-job-def)
:file-gc (ig/ref :app.tasks.file-gc/job-def)
:schedule-team-storage (ig/ref :app.tasks.team-storage/job-def)
:recompute-team-storage (ig/ref :app.tasks.team-storage/worker-job-def)
:offload-file-data (ig/ref :app.tasks.offload-file-data/job-def)
:objects-gc (ig/ref :app.tasks.objects-gc/job-def)
:export-binfile (ig/ref :app.tasks.export-binfile/job-def)
@ -659,7 +661,8 @@
: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-admin-console (ig/ref :app.loggers.audit.consumer-task/admin-console-job-def)
:audit-log-consumer-storage-usage (ig/ref :app.loggers.audit.consumer-task/storage-usage-job-def)}
::jobs/request-pool
{::rds/client (ig/ref ::rds/client)
@ -693,6 +696,14 @@
::sto/storage (ig/ref ::sto/storage)
::mtx/metrics (ig/ref ::mtx/metrics)}
:app.tasks.team-storage/job-def
{::db/pool (ig/ref ::db/pool)
::mtx/metrics (ig/ref ::mtx/metrics)}
:app.tasks.team-storage/worker-job-def
{::db/pool (ig/ref ::db/pool)
::mtx/metrics (ig/ref ::mtx/metrics)}
:app.tasks.offload-file-data/job-def
{::db/pool (ig/ref ::db/pool)
::sto/storage (ig/ref ::sto/storage)
@ -759,6 +770,10 @@
:app.nitrate/client (ig/ref :app.nitrate/client)
::mtx/metrics (ig/ref ::mtx/metrics)}
:app.loggers.audit.consumer-task/storage-usage-job-def
{::db/pool (ig/ref ::db/pool)
::mtx/metrics (ig/ref ::mtx/metrics)}
::email/blacklist
{}
@ -854,37 +869,46 @@
::db/pool (ig/ref ::db/pool)
::mtx/metrics (ig/ref ::mtx/metrics)
::wrk/entries
[{:cron #penpot/cron "0 0 0 * * ?" ;; daily
:job :session-gc}
;; a flag may contribute several entries (storage-usage does): each
;; element is one entry or a seq of them
(into [] (mapcat #(if (map? %) [%] %))
[{:cron #penpot/cron "0 0 0 * * ?" ;; daily
:job :session-gc}
{:cron #penpot/cron "0 0 0 * * ?" ;; daily
:job :objects-gc}
{:cron #penpot/cron "0 0 0 * * ?" ;; daily
:job :objects-gc}
{:cron #penpot/cron "0 0 0 * * ?" ;; daily
:job :storage-gc-deleted}
{:cron #penpot/cron "0 0 0 * * ?" ;; daily
:job :storage-gc-deleted}
{:cron #penpot/cron "0 0 0 * * ?" ;; daily
:job :storage-gc-touched}
{:cron #penpot/cron "0 0 0 * * ?" ;; daily
:job :storage-gc-touched}
{:cron #penpot/cron "0 0 0 * * ?" ;; daily
:job :storage-pending-gc}
{:cron #penpot/cron "0 0 0 * * ?" ;; daily
:job :storage-pending-gc}
{:cron #penpot/cron "0 0 0 * * ?" ;; daily
:job :jobs-gc}
{:cron #penpot/cron "0 0 0 * * ?" ;; daily
:job :jobs-gc}
{:cron #penpot/cron "0 0 2 * * ?" ;; daily
:job :file-gc-scheduler}
{:cron #penpot/cron "0 0 2 * * ?" ;; daily
:job :file-gc-scheduler}
{:cron #penpot/cron "0 30 */3,23 * * ?"
:job :telemetry}
{:cron #penpot/cron "0 30 */3,23 * * ?"
:job :telemetry}
(when (contains? cf/flags :nexus)
{:cron #penpot/cron "0 */5 * * * ?" ;; every 5m
:job :audit-log-consumer-nexus})
(when (contains? cf/flags :nexus)
{:cron #penpot/cron "0 */5 * * * ?" ;; every 5m
:job :audit-log-consumer-nexus})
(when (contains? cf/flags :admin-console)
{:cron #penpot/cron "30 */5 * * * ?" ;; every 5m, offset 30s from nexus
:job :audit-log-consumer-admin-console})]}
(when (contains? cf/flags :admin-console)
{:cron #penpot/cron "30 */5 * * * ?" ;; every 5m, offset 30s from nexus
:job :audit-log-consumer-admin-console})
(when (contains? cf/flags :storage-usage)
[{:cron #penpot/cron "15 */5 * * * ?" ;; every 5m, offset 15s
:job :audit-log-consumer-storage-usage}
{:cron #penpot/cron "45 */10 * * * ?" ;; every 10m, after the drain
:job :schedule-team-storage}])])}
::wrk/dispatcher
{::rds/client (ig/ref ::rds/client)
@ -927,6 +951,18 @@
::db/pool (ig/ref ::db/pool)
::mbus/msgbus (ig/ref ::mbus/msgbus)}
;; the storage recompute runs the slow per-team SUM: it must not
;; share the runner of the deletion cascades
[::storage ::wrk/runner]
{::wrk/parallelism (cf/get :worker-storage-parallelism 2)
::wrk/queue :storage
::wrk/tenant (cf/get :tenant)
::rds/client (ig/ref ::rds/client)
::jobs/defs (ig/ref ::jobs/defs)
::mtx/metrics (ig/ref ::mtx/metrics)
::db/pool (ig/ref ::db/pool)
::mbus/msgbus (ig/ref ::mbus/msgbus)}
[::cron ::wrk/runner]
{::wrk/parallelism (cf/get :worker-cron-parallelism 2)
::wrk/queue :cron

5
backend/src/app/migrations.clj
View File

@ -532,7 +532,10 @@
:fn (mg/resource "app/migrations/sql/0156-reset-pg-storage-defaults.sql")}
{:name "0157-add-audit-log-pending-table"
:fn (mg/resource "app/migrations/sql/0157-add-audit-log-pending-table.sql")}])
:fn (mg/resource "app/migrations/sql/0157-add-audit-log-pending-table.sql")}
{:name "0158-add-team-storage-usage"
:fn (mg/resource "app/migrations/sql/0158-add-team-storage-usage.sql")}])
(defn apply-migrations!
[pool name migrations]

17
backend/src/app/migrations/sql/0158-add-team-storage-usage.sql Normal file
View File

@ -0,0 +1,17 @@
-- Cached per-team media storage usage for the
-- media-storage-bytes-per-team quota. A background audit-log consumer
-- marks dirty teams and a bounded job recomputes their totals with the
-- quota query, so uploads read one narrow row instead of scanning
-- storage_object. `dirty_at` set = pending recompute; null = fresh.
-- A missing row means never computed (fail open + mark dirty).
CREATE TABLE team_storage_usage (
team_id uuid PRIMARY KEY REFERENCES team(id) ON DELETE CASCADE,
total_bytes bigint NOT NULL DEFAULT 0,
dirty_at timestamptz NULL DEFAULT NULL,
updated_at timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX team_storage_usage__dirty_at__idx
ON team_storage_usage (dirty_at ASC, team_id ASC)
WHERE dirty_at IS NOT NULL;

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

@ -122,13 +122,36 @@
(when (pos? amount)
(mtx/run! metrics :id :audit-ingest-events :labels [kind outcome] :inc amount)))
(defn- admin-console-eligible?
"Rows are positional, in `event-columns` order: [:id :name :source ...]."
[[_ name source]]
(and (contains? consumer/event-names-for-admin-console name)
(not (audit/telemetry-shadow-source? source))))
(defn- insert-rows!
[tx rows]
(when (seq rows)
(db/insert-many! tx :audit-log event-columns rows)
;; Same transaction as the rows: every ingested record is born
;; pending for every enabled consumer (see consumer-task).
(consumer/queue-pendings tx (map first rows))))
;; pending for every enabled non-selective consumer (see
;; consumer-task). The storage-usage consumer is fed only by
;; backend events (append-audit-entry): no frontend event name can
;; change per-team media storage, so bulk ingest queues nothing
;; for it.
;; The nexus consumer takes every row (gated on its flag like
;; the backend producer does in append-audit-entry).
(consumer/queue-nexus-pendings tx (map first rows))
;; The admin-console consumer is selective: only allowlisted names
;; queue its pending (this keeps its browser events working), and
;; never a telemetry shadow (the full row already queues; the
;; shadow keeps the name, so it would be ingested twice).
(when (contains? cf/flags :admin-console)
(consumer/queue-pendings tx
(into []
(comp (filter admin-console-eligible?)
(map first))
rows)
consumer/consumer-admin-console))))
(defn- handle-events
[{:keys [::db/pool] :as cfg} params]

29
backend/src/app/rpc/commands/files.clj
View File

@ -1015,7 +1015,10 @@
:modified-at (ct/now)
:has-media-trimmed false}))))
(defn- get-referenced-files
(defn get-referenced-files
"Ids of the live files using the given file as a library, oldest
first. Public so the delete cascade can capture the dependents
before removing the relations."
[cfg library-id]
(->> (db/exec! cfg [sql:get-referenced-files library-id])
(into [] bfc/xf-map-id)))
@ -1129,23 +1132,17 @@
file (db/update! conn :file
{:deleted-at (ct/in-future delay)}
{:id file-id}
{::db/return-keys [:id :name :is-shared :deleted-at
:project-id :created-at :modified-at]})
;; Files using the library, captured before their relations are removed
file-ids (when (:is-shared file)
(get-referenced-files conn file-id))]
;; Remove all possible relations for that file
(db/delete! conn :file-library-rel
{:library-file-id file-id})
{::db/return-keys [:name :deleted-at
:project-id :created-at :modified-at]})]
;; Only the mark happens here, so the file disappears at once. The
;; cascade job captures the files using this one as a library,
;; removes the relations and absorbs them: the relations must still
;; exist when the job runs.
(jobs/submit cfg
{::jobs/name :delete-object
::jobs/params (cond-> {:object :file
:deleted-at (:deleted-at file)
:id file-id}
(some? file-ids)
(assoc :file-ids file-ids))})
::jobs/params {:object :file
:deleted-at (:deleted-at file)
:id file-id}})
file))
(def ^:private

62
backend/src/app/rpc/commands/management.clj
View File

@ -25,6 +25,7 @@
[app.rpc.commands.projects :as proj]
[app.rpc.commands.teams :as teams]
[app.rpc.doc :as-alias doc]
[app.rpc.helpers :as rph]
[app.setup :as-alias setup]
[app.setup.templates :as tmpl]
[app.storage.tmp :as tmp]
@ -271,7 +272,10 @@
;; --- COMMAND: Move file
(def sql:get-files
"select id, features, project_id from file where id = ANY(?)")
"select f.id, f.features, f.project_id, p.team_id
from file as f
join project as p on (p.id = f.project_id)
where f.id = ANY(?)")
(def sql:move-files
"update file set project_id = ? where id = ANY(?)")
@ -299,7 +303,15 @@
(map files/decode-row))
source (into #{} (map :project-id) files)
pids (->> (conj source project-id)
(db/create-array conn "uuid"))]
(db/create-array conn "uuid"))
;; A move carries a single origin team in its props: refuse
;; mixed-origin requests up front instead of guessing.
source-teams (into #{} (keep :team-id) files)]
(when (> (count source-teams) 1)
(ex/raise :type :validation
:code :cant-move-from-multiple-teams
:hint "Unable to move files from different teams at once"))
(when (contains? source project-id)
(ex/raise :type :validation
@ -313,7 +325,9 @@
(doseq [project-id source]
(proj/check-edition-permissions! conn profile-id project-id))
;; Check the team compatibility
;; Check the team compatibility, then move: the destination team
;; is already loaded here, so the event props below need no extra
;; query.
(let [orig-team (teams/get-team conn :profile-id profile-id :project-id (first source))
dest-team (teams/get-team conn :profile-id profile-id :project-id project-id)]
(cfeat/check-teams-compatibility! orig-team dest-team)
@ -321,26 +335,30 @@
;; Check if all pending to move files are compaib
(let [features (cfeat/get-team-enabled-features cf/flags dest-team)]
(doseq [file files]
(cfeat/check-file-features! features (:features file)))))
(cfeat/check-file-features! features (:features file))))
;; move all files to the project
(db/exec-one! conn [sql:move-files project-id fids])
;; move all files to the project
(db/exec-one! conn [sql:move-files project-id fids])
;; delete possible broken relations on moved files
(db/exec-one! conn [sql:delete-broken-relations pids])
;; delete possible broken relations on moved files
(db/exec-one! conn [sql:delete-broken-relations pids])
;; Update the modification date of the all affected projects
;; ensuring that the destination project is the most recent one.
(loop [project-ids (into (list project-id) source)
modified-at (ct/now)]
(when-let [project-id (first project-ids)]
(db/update! conn :project
{:modified-at modified-at}
{:id project-id})
(recur (rest project-ids)
(ct/plus modified-at 10))))
;; Update the modification date of the all affected projects
;; ensuring that the destination project is the most recent one.
(loop [project-ids (into (list project-id) source)
modified-at (ct/now)]
(when-let [project-id (first project-ids)]
(db/update! conn :project
{:modified-at modified-at}
{:id project-id})
(recur (rest project-ids)
(ct/plus modified-at 10))))
nil))
;; Both teams travel explicitly in the event props so the
;; storage-usage drain never joins production tables.
(rph/with-meta (rph/wrap)
{::audit/props {:source-team-id (first source-teams)
:dest-team-id (:id dest-team)}}))))
(def ^:private
schema:move-files
@ -409,7 +427,11 @@
;; delete possible broken relations on moved files
(db/exec-one! conn [sql:delete-broken-relations pids])
nil))
;; Both teams travel explicitly in the event props so the
;; storage-usage drain never joins production tables.
(rph/with-meta (rph/wrap)
{::audit/props {:source-team-id (:team-id project)
:dest-team-id team-id}})))
(def ^:private
schema:move-project

35
backend/src/app/rpc/commands/media.clj
View File

@ -40,15 +40,28 @@
(declare create-file-media-object)
;; TEMPORARILY DISABLED (2026-10-19): the media-storage-bytes-per-team
;; quote check that used this lookup is disabled below. Kept active
;; because the audit props of the upload still report the team id.
;; Resolves the team of the target file for the quota check and the
;; audit props of the upload.
(def ^:private sql:get-team-id-for-file
"SELECT p.team_id
FROM file AS f
JOIN project AS p ON (p.id = f.project_id)
WHERE f.id = ?")
(defn- check-media-storage-quote!
"Apply the media-storage-bytes-per-team quota to a file that is about
to receive `size` bytes, and answer its team id (the audit props need
it). The counter path is cheap: the live SUM only runs with the
storage-usage flags off."
[cfg file-id profile-id size]
(let [team-id (:team-id (db/exec-one! (db/get-connectable cfg)
[sql:get-team-id-for-file file-id]))]
(quotes/check! cfg {::quotes/id ::quotes/media-storage-bytes-per-team
::quotes/profile-id profile-id
::quotes/team-id team-id
::quotes/incr size})
team-id))
(def ^:private schema:upload-file-media-object
[:map {:title "upload-file-media-object" :closed true}
[:file-id ::sm/uuid]
@ -68,16 +81,7 @@
(media.v/validate-media-type! content)
(media.v/validate-media-size! content)
(let [team-id (:team-id (db/exec-one! pool [sql:get-team-id-for-file file-id]))]
;; TEMPORARILY DISABLED (2026-10-19): media-storage-bytes-per-team
;; quote performs a full SUM over storage_object per upload (3-12s in
;; prod on a 3.5TB DB, ~34 slow executions/hour). Disabled until we
;; land a cheaper approximation (cached counter / rewritten query).
;; TODO: re-enable with the new implementation.
;; (quotes/check! cfg {::quotes/id ::quotes/media-storage-bytes-per-team
;; ::quotes/profile-id profile-id
;; ::quotes/team-id team-id
;; ::quotes/incr (:size content)})
(let [team-id (check-media-storage-quote! cfg file-id profile-id (:size content))]
(db/run! cfg (fn [{:keys [::db/conn] :as cfg}]
;; We get the minimal file for proper checking if
;; file is not already deleted
@ -263,6 +267,10 @@
(assoc :content content)
(assoc :name (d/nilv name "unknown")))]
;; The URL path writes bytes too: it must pay the same quota as a
;; plain upload, not bypass it.
(check-media-storage-quote! cfg (:file-id params) (:profile-id params) (:size content))
;; NOTE: we use the climit here in a dynamic invocation because we
;; don't want saturate the process-image limit with IO (download
;; of external image)
@ -585,6 +593,7 @@
(assoc :mtype mtype)
(media.v/validate-media-type!)
(media.v/validate-media-size!))
_ (check-media-storage-quote! cfg file-id profile-id (:size content))
mobj (create-file-media-object cfg (assoc params
:from-chunks? true
:content content))]

13
backend/src/app/rpc/commands/profile.clj
View File

@ -642,6 +642,10 @@
;; --- HELPERS
;; Owned teams with their participant count. The second parameter
;; decides whether soft-deleted teams count: the restore path needs
;; them (the delete cascade took the teams with the profile), the read
;; path never does.
(def sql:owned-teams
"WITH owner_teams AS (
SELECT tpr.team_id AS id
@ -649,7 +653,7 @@
JOIN team AS t ON (t.id = tpr.team_id)
WHERE tpr.is_owner IS TRUE
AND tpr.profile_id = ?
AND t.deleted_at IS NULL
AND (?::boolean OR t.deleted_at IS NULL)
)
SELECT tpr.team_id AS id,
count(tpr.profile_id) - 1 AS participants
@ -658,8 +662,11 @@
GROUP BY 1")
(defn get-owned-teams
[conn profile-id]
(db/exec! conn [sql:owned-teams profile-id]))
"Teams the profile owns. `:include-deleted?` also returns the ones
the delete cascade soft-deleted: the restore path needs them, the
read path does not."
[conn profile-id & {:keys [include-deleted?]}]
(db/exec! conn [sql:owned-teams profile-id (boolean include-deleted?)]))
(def ^:private sql:profile-existence
"select exists (select * from profile

63
backend/src/app/rpc/quotes.clj
View File

@ -607,16 +607,63 @@
JOIN storage_object AS so ON (so.id = refs.so_id)
WHERE so.deleted_at IS NULL")
(defn get-live-media-storage-bytes
"Total live media storage bytes of a team, computed straight from
the domain tables. Slow on large teams (full scan); only the
background recompute job calls it, never the upload path."
[conn team-id]
(:total (db/exec-one! conn [sql:get-media-storage-bytes-per-team
team-id team-id team-id team-id team-id team-id])))
(def ^:private sql:ensure-team-storage-usage
"INSERT INTO team_storage_usage (team_id, dirty_at)
SELECT ?, now()
WHERE EXISTS (SELECT 1 FROM team AS t WHERE t.id = ?)
ON CONFLICT (team_id) DO NOTHING")
(def ^:private sql:get-team-storage-usage-row
"SELECT team_id FROM team_storage_usage WHERE team_id = ?")
(defn- ensure-team-storage-usage
"Create the counter row (dirty) when missing. Reads first so the
common case (the row already exists) pays no write; concurrent
creates for a new team race safely into ON CONFLICT DO NOTHING."
[conn team-id]
(when-not (db/exec-one! conn [sql:get-team-storage-usage-row team-id])
;; WHERE EXISTS like its siblings: a team purged between the read
;; and the insert must not turn the fail-open path into an error.
(db/exec-one! conn [sql:ensure-team-storage-usage team-id team-id])))
(def ^:private sql:get-cached-media-storage-bytes-per-team
"Always one row: `generic-check!` assumes the count query never
comes back empty, and a team with no usage row (never computed, or
purged before the ensure could create it) must fail open, not
crash."
"SELECT COALESCE((SELECT total_bytes
FROM team_storage_usage
WHERE team_id = ?), 0) AS total")
(defmethod check-quote ::media-storage-bytes-per-team
[{:keys [::profile-id ::team-id ::target] :as cfg}]
[{:keys [::db/conn ::profile-id ::team-id ::target] :as cfg}]
(assert (valid-media-storage-bytes-per-team-quote? cfg) "invalid quote parameters")
(-> cfg
(assoc ::default (cf/get :quotes-media-storage-bytes-per-team
(* 20 1024 1024 1024)))
(assoc ::quote-sql [sql:get-quotes-2 target team-id profile-id profile-id])
(assoc ::count-sql [sql:get-media-storage-bytes-per-team
team-id team-id team-id team-id team-id team-id])
(generic-check!)))
(let [count-sql (if (and (contains? cf/flags :storage-usage)
(contains? cf/flags :audit-log))
;; Cached path: the background recompute job keeps the
;; counter. Fail open: a team with no row yet is allowed
;; through and marked dirty for the next recompute pass.
(do
(ensure-team-storage-usage conn team-id)
[sql:get-cached-media-storage-bytes-per-team team-id])
;; Live path: full scan, only when the background
;; counter is off.
[sql:get-media-storage-bytes-per-team
team-id team-id team-id team-id team-id team-id])]
(-> cfg
(assoc ::default (cf/get :quotes-media-storage-bytes-per-team
(* 20 1024 1024 1024)))
(assoc ::quote-sql [sql:get-quotes-2 target team-id profile-id profile-id])
(assoc ::count-sql count-sql)
(generic-check!))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; QUOTE: BINFILES JOBS PER PROFILE

44
backend/src/app/srepl/main.clj
View File

@ -684,12 +684,17 @@
"Mark a project for deletion"
[file-id]
(let [file-id (h/parse-uuid file-id)
tnow (ct/now)]
tnow (ct/now)
;; The event carries its team: the storage-usage drain never
;; joins production tables, and a keyless event would trip its
;; "producer stopped keying" warn.
team-id (first (audit/get-team-ids sys/system {:file-ids [file-id]}))]
(audit/insert sys/system
{:name "delete-file"
:type "action"
:props {:id file-id}
:props (cond-> {:id file-id}
(some? team-id) (assoc :team-id team-id))
:context {:triggered-by "srepl"
:cause "explicit call to delete-file!"}
:tracked-at tnow})
@ -709,11 +714,14 @@
(when-let [file (db/get* system :file
{:id file-id}
{::db/remove-deleted false
::sql/columns [:id :name]})]
::sql/columns [:id :name :project-id]})]
(audit/insert system
{:name "restore-file"
:type "action"
:props file
:props (assoc file
:team-id (first (audit/get-team-ids
system
{:project-ids [(:project-id file)]})))
:context {:triggered-by "srepl"
:cause "explicit call to restore-file!"}})
@ -837,7 +845,7 @@
(audit/insert system
{:name "restore-team"
:type "action"
:props team
:props (assoc team :team-id (:id team))
:context {:triggered-by "srepl"
:cause "explicit call to restore-team!"}})
@ -874,20 +882,28 @@
{:id profile-id}
{::db/remove-deleted false})
(profile/decode-row))]
(audit/insert system
{:name "restore-profile"
:type "action"
:props (audit/profile->props profile)
:context {:triggered-by "srepl"
:cause "explicit call to restore-profile!"}})
(db/update! system :profile
{:deleted-at nil}
{:id profile-id}
{::db/return-keys false})
(doseq [{:keys [id]} (profile/get-owned-teams system profile-id)]
(restore-team* system id))
;; Owned teams are still soft-deleted here, so resolve
;; them through the membership table: the read
;; path only sees live ones.
(let [team-ids (mapv :id (profile/get-owned-teams system (:id profile)
:include-deleted? true))]
(doseq [id team-ids]
(restore-team* system id))
;; The drain never joins tables, so the event
;; carries the restored teams explicitly.
(audit/insert system
{:name "restore-profile"
:type "action"
:props (assoc (audit/profile->props profile)
:team-ids team-ids)
:context {:triggered-by "srepl"
:cause "explicit call to restore-profile!"}}))
:restored)))))

116
backend/src/app/tasks/delete_object.clj
View File

@ -7,6 +7,7 @@
(ns app.tasks.delete-object
"A generic task for object deletion cascade handling"
(:require
[app.common.data :as d]
[app.common.logging :as l]
[app.common.schema :as sm]
[app.common.time :as ct]
@ -21,6 +22,14 @@
(def ^:dynamic *team-deletion* false)
(def ^:dynamic *deleting-file-ids*
"Ids of the files the running cascade is going to delete. Bound by
the `:project` method with its own files so `:file` only absorbs the
dependents that survive: absorbing into files that die in the same
run is wasted work. Team deletions need no such set: they bind
`*team-deletion*` instead, which skips absorb entirely."
nil)
(def ^:dynamic *cascade*
"Accumulates the ids marked deleted by the current cascade execution,
grouped by entity. Bound by the task entry point; nil on direct calls
@ -78,14 +87,43 @@
{::db/return-keys false})
(when (and (:is-shared file)
;; Team deletions skip absorb entirely: every file dies
;; in the same cascade, so no survivor can absorb.
(not *team-deletion*))
;; NOTE: we don't prevent file deletion on absorb operation failure
(try
(db/tx-run! cfg files/absorb-library id :file-ids file-ids)
(catch Throwable cause
(l/warn :hint "error on absorbing library"
:file-id id
:cause cause))))
;; NOTE: we don't prevent file deletion on absorb operation failure.
;; Only live dependents outside this cascade matter: if the whole
;; closure dies together there is nothing to absorb and the
;; relations die with their files (both FKs are ON DELETE CASCADE).
;; DEPRECATED `:file-ids`: jobs queued by the previous release
;; already deleted the relations in the RPC, so the live lookup
;; finds nothing. Trust their snapshot for one release; new jobs
;; never carry the key.
(let [doomed (or *deleting-file-ids* #{})
dependents (if (seq file-ids)
(into [] (remove doomed) file-ids)
(into [] (remove doomed)
(files/get-referenced-files conn id)))]
(when (seq dependents)
;; Runs in the caller transaction: a nested transaction here
;; would join it anyway, so the call goes directly. The
;; savepoint contains an absorb failure: it rolls back the
;; absorb and the relations removal and the cascade goes on.
;; Nothing retries the absorb itself: the job succeeds, the
;; relations stay, and the later hard delete cascades them
;; away with their files.
(let [sp (db/savepoint conn "absorb-library")]
(try
;; The relations must still exist at this point: removing
;; them first is what lets a later hard delete proceed.
(db/delete! conn :file-library-rel {:library-file-id id})
(files/absorb-library cfg id :file-ids dependents)
(catch Throwable cause
(db/rollback! conn sp)
(l/warn :hint "error on absorbing library"
:file-id id
:cause cause))
(finally
(db/release! conn sp)))))))
;; Mark file change to be deleted, but only the rows that do not carry a
;; date yet. A revision is born already scheduled for deletion one hour
@ -132,12 +170,16 @@
{:id id}
{::db/return-keys false})
(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)))
(let [files (db/query conn :file
{:project-id id}
{::db/columns [:id :deleted-at]})]
;; Every file below dies in this run: absorbing a library into
;; them is wasted work, so the files only absorb survivors.
(binding [*deleting-file-ids* (into #{} d/xf:map-id files)]
(doseq [file files]
(delete-object cfg (assoc file
:object :file
:deleted-at deleted-at)))))
(record :projects id))
@ -191,29 +233,39 @@
[:deleted-at ::ct/inst]
[:id ::sm/uuid]
[:file-id {:optional true} ::sm/uuid]
;; DEPRECATED: only jobs queued before the cascade owned the
;; relations carry it (their RPC already deleted them). Remove once
;; no such rows can be left.
[:file-ids {:optional true} [:vector ::sm/uuid]]])
(defn execute-delete-object
"Plain job handler: run the delete-object multimethod on the provided
params inside a single transaction, tracking the cascade and reporting
it as an audit event when anything was marked."
params, tracking the cascade and reporting it as an audit event when
anything was marked. Runs inside the caller transaction (the job-def
handler opens it with `tx-run!`)."
[cfg params]
(db/tx-run! cfg
(fn [cfg]
(when (top-exists? cfg params)
(binding [*cascade* (atom {:teams [] :projects [] :files []})]
(delete-object cfg params)
(let [{:keys [teams projects files]} @*cascade*]
(when (or (seq teams) (seq projects) (seq files))
(audit/submit cfg {:name "delete-object-with-cascade"
:type "action"
:props {:object (:object params)
:id (:id params)
:deleted-at (:deleted-at params)
:cascade {:team-ids teams
:project-ids projects
:file-ids files}}
:context {:triggered-by "delete-object"}}))))))))
(when (top-exists? cfg params)
(binding [*cascade* (atom {:teams [] :projects [] :files []})]
(delete-object cfg params)
(let [{:keys [teams projects files]} @*cascade*]
(when (or (seq teams) (seq projects) (seq files))
;; Flat team ids for consumers that key on them
;; (the storage-usage drain), so they never join
;; production tables.
(let [team-ids (audit/get-team-ids cfg
{:team-ids teams
:project-ids projects
:file-ids files})]
(audit/submit cfg {:name "delete-object-with-cascade"
:type "action"
:props {:object (:object params)
:id (:id params)
:deleted-at (:deleted-at params)
:team-ids (vec team-ids)
:cascade {:team-ids teams
:project-ids projects
:file-ids files}}
:context {:triggered-by "delete-object"}})))))))
(defmethod ig/assert-key ::job-def
[_ params]
@ -226,6 +278,6 @@
::jobs/schema schema:delete-object-params
::jobs/handler
(fn [_context params]
(execute-delete-object cfg params))
(db/tx-run! cfg execute-delete-object params))
::jobs/decoder (sm/decoder schema:delete-object-params sm/json-transformer)
::jobs/validator (sm/validator schema:delete-object-params)})

12
backend/src/app/tasks/file_gc.clj
View File

@ -26,6 +26,7 @@
[app.features.fdata :as feat.fdata]
[app.features.file-snapshots :as fsnap]
[app.jobs :as jobs]
[app.loggers.audit :as audit]
[app.storage :as sto]
[integrant.core :as ig]))
@ -64,6 +65,17 @@
:file-id (str id)
:id (str id)))
;; The storage-usage consumer learns about this shrink path only
;; through this event: no RPC runs here. The team id travels in
;; props so the drain never joins production tables.
(when (and (seq unused-media) (contains? cf/flags :audit-log))
(let [team-id (first (audit/get-team-ids cfg {:file-ids [id]}))]
(audit/submit cfg {:name "file-gc"
:type "action"
:props (cond-> {:file-id id}
(some? team-id) (assoc :team-id team-id))
:context {:triggered-by "cron-task"}})))
file))
(def ^:private sql:mark-file-object-thumbnails-deleted

5
backend/src/app/tasks/objects_gc.clj
View File

@ -42,7 +42,10 @@
chunk carries only its own entity key (`:team-ids`, `:project-ids`
or `:file-ids`); a single event spanning sections could never share
their transactions. Runs inside the caller transaction, next to the
deletes: if the purge aborts, the event aborts with it."
deletes: if the purge aborts, the event aborts with it. No team-id
resolution: the live quota total only counts rows without `deleted_at`,
which a physical purge cannot change, so no team-keyed consumer needs
these events."
[cfg ids-key ids]
(when (contains? cf/flags :audit-log)
(audit/submit cfg

181
backend/src/app/tasks/team_storage.clj Normal file
View File

@ -0,0 +1,181 @@
;; 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.team-storage
"Background recompute of the cached per-team media storage usage
(`team_storage_usage`) that the media-storage-bytes-per-team quota
reads instead of scanning `storage_object` on every upload.
The audit-log storage-usage consumer marks dirty the teams whose
media may have changed. Every few minutes the `:schedule-team-storage`
scheduler streams all dirty teams oldest-first and enqueues one
`:recompute-team-storage` job per team (skipped while one is already live),
so a failure is contained in its own job and the next tick
reschedules it. Teams without a row are born on first quota check
(fail open + dirty), never swept."
(:require
[app.common.logging :as l]
[app.common.schema :as sm]
[app.db :as db]
[app.jobs :as jobs]
[app.metrics :as mtx]
[app.rpc.quotes :as quotes]
[integrant.core :as ig]))
(def ^:private sql:get-dirty-teams
"Dirty teams first, then the ones nobody refreshed for a week. The
second branch is the self-healing path: a mark lost on the way (an
audit insert that failed, a flag that was off) would otherwise keep
the counter low for ever. The table holds one row per team, so the
stale branch needs no index."
"SELECT team_id
FROM team_storage_usage
WHERE dirty_at IS NOT NULL
OR updated_at < now() - interval '7 days'
ORDER BY dirty_at ASC NULLS LAST, updated_at ASC")
(def ^:private sql:upsert-team-storage-usage
"INSERT INTO team_storage_usage AS tsu (team_id, total_bytes, dirty_at, updated_at)
SELECT ?, ?, NULL, now()
WHERE EXISTS (SELECT 1 FROM team AS t WHERE t.id = ?)
ON CONFLICT (team_id) DO UPDATE
SET total_bytes = EXCLUDED.total_bytes,
dirty_at = CASE WHEN tsu.dirty_at > ?::timestamptz THEN tsu.dirty_at ELSE NULL END,
updated_at = now()")
(defn- recompute-team-storage
"Recompute one team: slow read, fast write. A mark written by a
transaction that started after this recompute began survives (its
dirty_at is newer than the read) and the next tick recomputes it. A
marker transaction that overlaps this one can still be cleared
stale: the CASE is a cheap safety net, not a proof. A stale read can
over-allow for one window (accepted: abuse prevention, not a hard
limit)."
[{:keys [::db/conn]} team-id]
;; Read the clock before the slow SUM: now() is stable per
;; transaction, and this whole worker runs in one, so a mark newer
;; than the read outlives the upsert.
(let [read-at (:now (db/exec-one! conn ["SELECT now() AS now"]))
total (quotes/get-live-media-storage-bytes conn team-id)]
(db/exec-one! conn [sql:upsert-team-storage-usage team-id total team-id read-at])
total))
(defn execute-recompute-team-storage
"Plain job handler: recompute a single team. A failure fails this
job (the runner retries it); the team stays dirty, so the next
scheduler tick enqueues it again."
[cfg {:keys [team-id]}]
(let [bytes (recompute-team-storage cfg team-id)]
(l/trc :hint "team storage recomputed"
:team-id (str team-id)
:bytes bytes)
bytes))
(def ^:private scheduler-chunk-size
"Teams per scheduler transaction: each chunk commits on its own so
killing the job mid-run keeps the chunks already enqueued."
50)
(def ^:private scheduler-delay-step-ms 5000)
(def ^:private scheduler-delay-cap-ms 60000)
(defn- chunk-delay
"Stagger delay of a scheduler chunk: per chunk, not per team, capped
so huge backlogs do not push work too far out."
[chunk-idx]
(min (* chunk-idx scheduler-delay-step-ms)
scheduler-delay-cap-ms))
(defn- submit-team-chunk
"Enqueue one recompute job per team id in the caller transaction and
answer how many were enqueued: call it through `tx-run!` on a cfg
without `::db/conn` (see `db/own-transaction`) so the chunk commits
(or rolls back) whole. A nil submit means skip-if-active found a live
job for that team: nothing enqueued, not counted."
[cfg team-ids delay]
(reduce (fn [n team-id]
(if (jobs/submit cfg
{::jobs/name :recompute-team-storage
::jobs/queue :storage
::jobs/params {:team-id team-id}
::jobs/label (str team-id)
::jobs/skip-if-active true
::jobs/priority 10
::jobs/delay delay})
(inc n)
n))
0
team-ids))
(defn execute-schedule-team-storage
"Stream every dirty team oldest-first and enqueue one recompute job
per team, staggered by chunk so the queue does not spike. Runs
inside the caller read-only transaction (the job-def handler opens
it: pgjdbc only streams the plan cursor with auto-commit off): each
chunk is submitted through its own writable transaction (progress is
durable: killing the run keeps the chunks already enqueued and the
next tick picks up the rest). Teams with a live job are skipped
(`skip-if-active` on the team id as label); their mark survives and a
later tick picks them up. The rows are only read as values (`map
:team-id` before `partition-all`): a plan row is a view over the open
result set, so buffering rows would read nil columns."
[{:keys [::db/conn] :as cfg}]
(transduce
(comp (map :team-id)
(partition-all scheduler-chunk-size)
(map-indexed vector))
(completing
(fn [total [chunk-idx team-ids]]
(let [submitted (-> (db/own-transaction cfg)
(db/tx-run! submit-team-chunk team-ids (chunk-delay chunk-idx)))]
(jobs/heartbeat cfg)
(+ total ^long submitted))))
0
(db/plan conn [sql:get-dirty-teams] {:fetch-size 100})))
(def schema:schedule-team-storage-params
"The scheduler takes no params."
[:map {:closed true}])
(def schema:recompute-team-storage-params
"A single team to recompute."
[:map {:closed true}
[:team-id ::sm/uuid]])
(defmethod ig/assert-key ::job-def
[_ params]
(assert (db/pool? (::db/pool params)) "expected a valid database pool")
(assert (mtx/metrics? (::mtx/metrics params)) "expected valid metrics instance"))
(defmethod ig/assert-key ::worker-job-def
[_ params]
(assert (db/pool? (::db/pool params)) "expected a valid database pool")
(assert (mtx/metrics? (::mtx/metrics params)) "expected valid metrics instance"))
(defmethod ig/init-key ::job-def
[_ cfg]
{::jobs/name :schedule-team-storage
::jobs/schema schema:schedule-team-storage-params
::jobs/decoder (sm/decoder schema:schedule-team-storage-params sm/json-transformer)
::jobs/validator (sm/validator schema:schedule-team-storage-params)
::jobs/handler
(fn [_context _params]
;; Read-only: pgjdbc only honors the plan fetch-size (a real
;; streaming cursor) with auto-commit off, so the whole stream
;; runs in one read transaction. The chunk submits open their
;; own writable transactions (see db/own-transaction).
(-> (assoc cfg ::db/read-only true)
(db/tx-run! execute-schedule-team-storage)))})
(defmethod ig/init-key ::worker-job-def
[_ cfg]
{::jobs/name :recompute-team-storage
::jobs/schema schema:recompute-team-storage-params
::jobs/decoder (sm/decoder schema:recompute-team-storage-params sm/json-transformer)
::jobs/validator (sm/validator schema:recompute-team-storage-params)
::jobs/handler
(fn [_context params]
(db/tx-run! cfg execute-recompute-team-storage params))})

9
backend/test/backend_tests/delete_object_test.clj
View File

@ -56,9 +56,9 @@
(t/is (= "team" (name (:object props))))
(t/is (= (:id team) (:id props)))
(t/is (= deleted-at (:deleted-at props)))
(t/is (not (contains? props :team-ids)))
(t/is (not (contains? props :project-ids)))
(t/is (not (contains? props :file-ids)))
;; Flat team ids for team-keyed consumers, next to the
;; per-entity cascade breakdown.
(t/is (= #{(:id team)} (set (:team-ids props))))
(t/is (= #{(:id team)} (set (get-in props [:cascade :team-ids]))))
(t/is (= #{(:default-project-id team) (:id project)}
(set (get-in props [:cascade :project-ids]))))
@ -85,7 +85,8 @@
(t/is (= (:id file) (:id props)))
(t/is (= [] (get-in props [:cascade :team-ids])))
(t/is (= [] (get-in props [:cascade :project-ids])))
(t/is (= [(:id file)] (get-in props [:cascade :file-ids])))))))))
(t/is (= [(:id file)] (get-in props [:cascade :file-ids])))
(t/is (= #{(:id team)} (set (:team-ids props))))))))))
(t/deftest delete-object-missing-row-emits-nothing
;; A cascade over an unknown id marks nothing, so there is nothing to report.

29
backend/test/backend_tests/loggers_audit_consumer_task_test.clj
View File

@ -37,10 +37,12 @@
(def ^:private archive-uri "http://nexus.example/archive")
(defn- insert-pendings!
"Queue the row with the real producer helper: pendings for the
enabled consumers, like append-audit-entry does."
"Queue the row with the real producer helper: pendings for every
enabled consumer, like production before selective queueing narrows
it (selectivity itself is covered in tasks-team-storage-test)."
[row-id]
(consumer/queue-pendings (:app.db/pool th/*system*) [row-id]))
(doseq [consumer-id (consumer/enabled-consumers)]
(consumer/queue-pendings (:app.db/pool th/*system*) [row-id] consumer-id)))
(defn- pending-count
[]
@ -475,6 +477,16 @@
(invoke-nexus! metrics params)
(invoke-admin-console! metrics))
(defn- invoke-storage-usage!
"Invoke the storage-usage consumer handler in-process with an
isolated metrics registry. It only needs the pool: marking teams
dirty touches no archive destination."
[metrics]
(let [job-def (ig/init-key :app.loggers.audit.consumer-task/storage-usage-job-def
{::db/pool (:app.db/pool th/*system*)
::mtx/metrics metrics})]
((::jobs/handler job-def) nil {})))
(defn- counter-value
[metrics id labels]
(let [collector (mtx/get-collector metrics id)
@ -503,6 +515,17 @@
(let [rows (th/db-exec! ["select * from audit_log"])]
(t/is (= 0 (count rows))))))))
(t/deftest storage-usage-consumer-records-processed-outcome
;; The storage-usage consumer marks teams dirty, it archives
;; nothing: its success outcome reads "processed", not "archived".
(let [prof (th/create-profile* 1 {:is-active true})
metrics (make-consumer-metrics)]
(with-redefs [cf/flags #{:audit-log :storage-usage}]
(insert-audit-row! (:id prof) "upload-file-media-object")
(invoke-storage-usage! metrics)
(t/is (= 1.0 (counter-value metrics :audit-consumer-events ["storage-usage" "processed"])))
(t/is (= 0 (pending-count))))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; PENDING-QUEUE TESTS
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

6
backend/test/backend_tests/objects_gc_test.clj
View File

@ -153,8 +153,10 @@
(th/reset-mock! audit-mock)
(th/run-task! :objects-gc {:skip-delay true})
(t/is (nil? (th/db-get :team {:id (:id team)} {::db/remove-deleted false})))
(let [events (filter #(contains? (:props %) :team-ids)
(let [events (filter #(= #{:team-ids} (set (keys (:props %))))
(events-named audit-mock "hard-delete-objects"))]
;; Every chunk carries exactly its own entity key and nothing
;; else: no flat team-ids are resolved after the delete.
(t/is (= 1 (count events)))
(let [event (first events)]
(t/is (= "command" (:type event)))
@ -180,6 +182,7 @@
(t/is (= 1 (count events)))
(let [event (first events)]
(t/is (= "command" (:type event)))
(t/is (= #{:project-ids} (set (keys (:props event)))))
(t/is (= [(:id project)] (get-in event [:props :project-ids])))
(t/is (= "objects-gc" (get-in event [:context :triggered-by])))))))))
@ -204,6 +207,7 @@
(t/is (= 1 (count events)))
(let [event (first events)]
(t/is (= "command" (:type event)))
(t/is (= #{:file-ids} (set (keys (:props event)))))
(t/is (= [(:id file)] (get-in event [:props :file-ids])))
(t/is (= "objects-gc" (get-in event [:context :triggered-by])))))))))

8
backend/test/backend_tests/rpc_audit_test.clj
View File

@ -274,8 +274,9 @@
(let [day-now (ct/truncate (ct/now) :days)]
(t/is (= day-now (:created-at row2)))
(t/is (= day-now (:tracked-at row2))))
;; Both rows are born pending for the enabled consumers.
(t/is (= 4 (count (th/db-exec! ["select * from audit_log_pending"])))))))))
;; Both rows are born pending for Nexus; the admin-console
;; consumer is selective and "navigate" is not allowlisted.
(t/is (= 2 (count (th/db-exec! ["select * from audit_log_pending"])))))))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; BACKEND PROCESS-EVENT PATH (RPC commands)
@ -932,7 +933,8 @@
(t/is (= 1.0 (ingest-counter metrics "full" "inserted")))
(t/is (= 1.0 (ingest-counter metrics "shadow" "inserted")))
(t/is (= 2 (count (th/db-exec! ["select * from audit_log"]))))
(t/is (= 4 (count (th/db-exec! ["select * from audit_log_pending"])))))))
;; Nexus only: "navigate" is not in the admin-console allowlist.
(t/is (= 2 (count (th/db-exec! ["select * from audit_log_pending"])))))))
(t/deftest push-handle-events-records-failed-on-error
(let [prof (th/create-profile* 1 {:is-active true})

154
backend/test/backend_tests/rpc_file_test.clj
View File

@ -22,6 +22,7 @@
[app.db.sql :as sql]
[app.features.fdata :as fdata]
[app.http :as http]
[app.jobs :as-alias jobs]
[app.loggers.audit :as-alias audit]
[app.rpc :as-alias rpc]
[app.rpc.commands.files :as files]
@ -2327,15 +2328,162 @@
:id (:id library)})]
(t/is (nil? (:error out)))
;; The relations are removed at once, the absorption is done by the
;; deletion task.
(t/is (empty? (th/db-query :file-library-rel {:library-file-id (:id library)})))
;; The relations stay until the deletion task runs: the job
;; captures the dependents, removes the relations and absorbs
;; them there.
(t/is (not (empty? (th/db-query :file-library-rel {:library-file-id (:id library)}))))
(t/is (= (:id library)
(cfo/get-tokens-source (:data (bfc/get-file th/*system* (:id file))))))
(th/run-pending-jobs)
(t/is (empty? (th/db-query :file-library-rel {:library-file-id (:id library)})))
(check-tokens-absorbed prepared)))
(t/deftest delete-shared-library-absorb-failure-keeps-cascade-going
;; If the absorb blows up, the savepoint rolls it back (relations
;; included) and the cascade goes on: the file is still marked deleted,
;; the relations are intact for the next retry, and the job succeeds.
(let [profile (th/create-profile* 1)
{:keys [library]} (prepare-tokens-library profile)
out (th/command! {::th/type :delete-file
::rpc/profile-id (:id profile)
:id (:id library)})]
(t/is (nil? (:error out)))
(with-redefs [files/absorb-library (fn [& _] (throw (ex-info "boom" {})))]
(th/run-pending-jobs))
(t/is (not (empty? (th/db-query :file-library-rel {:library-file-id (:id library)}))))
(t/is (some? (:deleted-at (th/db-get :file {:id (:id library)} {::db/remove-deleted false}))))))
(t/deftest delete-shared-library-queues-job-without-file-ids
;; The cascade captures the dependents itself: the RPC no longer
;; snapshots them into the job params.
(let [profile (th/create-profile* 1)
{:keys [library]} (prepare-tokens-library profile)
submitted (atom nil)]
(with-redefs [jobs/submit (fn [_ params] (reset! submitted params) (uuid/next))]
(let [out (th/command! {::th/type :delete-file
::rpc/profile-id (:id profile)
:id (:id library)})]
(t/is (nil? (:error out)))))
(t/is (= :delete-object (::jobs/name @submitted)))
(t/is (not (contains? (::jobs/params @submitted) :file-ids)))))
(t/deftest delete-project-skips-absorb-for-files-dying-together
;; When the whole project dies, absorbing the shared library into
;; its own dying files is wasted work: the cascade cleans the
;; relations but leaves the dependents untouched.
(let [profile (th/create-profile* 1)
team-id (:default-team-id profile)
proj (th/create-project* 1 {:profile-id (:id profile) :team-id team-id})
library (th/create-file* 1 {:profile-id (:id profile)
:project-id (:id proj)
:is-shared true})
file (th/create-file* 2 {:profile-id (:id profile)
:project-id (:id proj)})
_ (th/link-file-to-library* {:file-id (:id file)
:library-id (:id library)})
revn (:revn (th/db-get :file {:id (:id file)}))]
(let [out (th/command! {::th/type :delete-project
::rpc/profile-id (:id profile)
:id (:id proj)})]
(t/is (nil? (:error out))))
(th/run-pending-jobs)
;; Nothing outside the cascade uses the library, so the relations
;; stay until the hard delete removes them with their files.
(t/is (not (empty? (th/db-query :file-library-rel {:library-file-id (:id library)}))))
(t/is (= revn (:revn (th/db-get :file {:id (:id file)} {::db/remove-deleted false}))))
;; The hard delete cascades the relations away with their files.
(th/run-task! :objects-gc {:skip-delay true})
(t/is (nil? (th/db-get :file {:id (:id file)} {::db/remove-deleted false})))
(t/is (empty? (th/db-query :file-library-rel {:library-file-id (:id library)})))))
(t/deftest delete-object-absorbs-with-deprecated-file-ids
;; Deploy window: a job queued by the previous release carries the
;; dependents in :file-ids and its RPC already deleted the relations.
;; The live lookup finds nothing, so the worker must trust the
;; snapshot instead of skipping the absorb silently.
(let [profile (th/create-profile* 6 {:is-active true})
team-id (:default-team-id profile)
proj (th/create-project* 6 {:profile-id (:id profile) :team-id team-id})
library (th/create-file* 8 {:profile-id (:id profile)
:project-id (:id proj)
:is-shared true})
dependent (th/create-file* 9 {:profile-id (:id profile)
:project-id (:id proj)})
_ (th/link-file-to-library* {:file-id (:id dependent)
:library-id (:id library)})
revn (:revn (th/db-get :file {:id (:id dependent)}))]
;; What the old RPC did: drop the relations before queueing.
(th/db-delete! :file-library-rel {:library-file-id (:id library)})
(th/run-task! :delete-object {:object :file
:id (:id library)
:deleted-at (ct/now)
:file-ids [(:id dependent)]})
(t/is (> (:revn (th/db-get :file {:id (:id dependent)} {::db/remove-deleted false}))
revn))))
(t/deftest delete-project-absorbs-into-surviving-dependent
;; The doomed-set only skips files that die in the same cascade: a
;; dependent in another project of the team survives, so the library
;; must be absorbed into it (and only into it).
(let [profile (th/create-profile* 5 {:is-active true})
team-id (:default-team-id profile)
proj-a (th/create-project* 5 {:profile-id (:id profile) :team-id team-id})
proj-b (th/create-project* 6 {:profile-id (:id profile) :team-id team-id})
library (th/create-file* 5 {:profile-id (:id profile)
:project-id (:id proj-a)
:is-shared true})
doomed (th/create-file* 6 {:profile-id (:id profile)
:project-id (:id proj-a)})
survivor (th/create-file* 7 {:profile-id (:id profile)
:project-id (:id proj-b)})
_ (th/link-file-to-library* {:file-id (:id survivor)
:library-id (:id library)})
revn-sur (:revn (th/db-get :file {:id (:id survivor)}))
revn-doom (:revn (th/db-get :file {:id (:id doomed)}))]
(let [out (th/command! {::th/type :delete-project
::rpc/profile-id (:id profile)
:id (:id proj-a)})]
(t/is (nil? (:error out))))
(th/run-pending-jobs)
;; The survivor absorbed the library; its relation is gone.
(t/is (> (:revn (th/db-get :file {:id (:id survivor)} {::db/remove-deleted false}))
revn-sur))
(t/is (empty? (th/db-query :file-library-rel {:file-id (:id survivor)})))
;; The doomed file did not absorb anything.
(t/is (= revn-doom (:revn (th/db-get :file {:id (:id doomed)} {::db/remove-deleted false}))))))
(t/deftest delete-team-skips-absorb-for-files-dying-together
;; Team deletions skip absorb entirely (*team-deletion*): a library
;; in project A used from project B dies with the team, so absorb
;; never runs and the relations stay until the hard delete.
(let [profile (th/create-profile* 4 {:is-active true})
team (th/create-team* 4 {:profile-id (:id profile)})
proj-a (th/create-project* 4 {:profile-id (:id profile) :team-id (:id team)})
proj-b (th/create-project* 5 {:profile-id (:id profile) :team-id (:id team)})
library (th/create-file* 6 {:profile-id (:id profile)
:project-id (:id proj-a)
:is-shared true})
file (th/create-file* 7 {:profile-id (:id profile)
:project-id (:id proj-b)})
_ (th/link-file-to-library* {:file-id (:id file)
:library-id (:id library)})
revn (:revn (th/db-get :file {:id (:id file)}))
calls (atom 0)]
(let [out (th/command! {::th/type :delete-team
::rpc/profile-id (:id profile)
:id (:id team)})]
(t/is (nil? (:error out))))
(with-redefs [files/absorb-library (fn [& _] (swap! calls inc) nil)]
(th/run-pending-jobs))
(t/is (zero? @calls))
(t/is (not (empty? (th/db-query :file-library-rel {:library-file-id (:id library)}))))
(t/is (= revn (:revn (th/db-get :file {:id (:id file)} {::db/remove-deleted false}))))
;; The hard delete cascades the relations away with their files.
(th/run-task! :objects-gc {:skip-delay true})
(t/is (nil? (th/db-get :file {:id (:id file)} {::db/remove-deleted false})))
(t/is (empty? (th/db-query :file-library-rel {:library-file-id (:id library)})))))
(t/deftest unpublish-shared-library-absorbs-its-tokens-lib
(let [profile (th/create-profile* 1)
{:keys [library] :as prepared} (prepare-tokens-library profile)

25
backend/test/backend_tests/rpc_management_test.clj
View File

@ -482,6 +482,31 @@
(t/is (= (:library-file-id item) (:id file3)))))))
(t/deftest move-files-from-multiple-teams-is-rejected
;; A single move carries one origin team in its props: mixed origins
;; are refused before anything moves.
(let [profile (th/create-profile* 1 {:is-active true})
team (th/create-team* 1 {:profile-id (:id profile)})
project1 (th/create-project* 1 {:team-id (:default-team-id profile)
:profile-id (:id profile)})
project2 (th/create-project* 2 {:team-id (:id team)
:profile-id (:id profile)})
file1 (th/create-file* 1 {:profile-id (:id profile)
:project-id (:id project1)})
file3 (th/create-file* 3 {:profile-id (:id profile)
:project-id (:id project2)})]
(let [out (th/command! {::th/type :move-files
::rpc/profile-id (:id profile)
:project-id (:id project2)
:ids #{(:id file1) (:id file3)}})
error (:error out)]
(t/is (th/ex-info? error))
(t/is (th/ex-of-type? error :validation))
(t/is (th/ex-of-code? error :cant-move-from-multiple-teams)))
(t/is (= 1 (count (th/db-query :file {:project-id (:id project1)}))))
(t/is (= 1 (count (th/db-query :file {:project-id (:id project2)}))))))
(t/deftest move-library-to-other-team
(let [profile (th/create-profile* 1 {:is-active true})
team (th/create-team* 1 {:profile-id (:id profile)})

51
backend/test/backend_tests/rpc_media_test.clj
View File

@ -11,6 +11,7 @@
[app.http.client :as http]
[app.media :as media]
[app.rpc :as-alias rpc]
[app.rpc.commands.media :as cmd-media]
[app.storage :as sto]
[backend-tests.helpers :as th]
[clojure.test :as t]
@ -23,6 +24,56 @@
(t/use-fixtures :once th/state-init)
(t/use-fixtures :each th/database-reset)
(t/deftest media-object-assemble-pays-the-storage-quota
;; The chunked upload path writes bytes too: it must pay the same
;; quota as the multipart upload.
(let [prof (th/create-profile* 3)
file (th/create-file* 3 {:profile-id (:id prof)
:project-id (:default-project-id prof)
:is-shared false})
path (th/tempfile "backend_tests/test_files/sample.jpg")]
(with-redefs [cmd-media/assemble-chunks
(fn [_ _ _]
{:filename "assembled.jpg"
:path path
:mtype "image/jpeg"
:size 312043})]
(with-mocks [mock {:target 'app.config/get
:return (th/config-get-mock {:quotes-media-storage-bytes-per-team 100})}]
(let [out (th/command! {::th/type :assemble-file-media-object
::rpc/profile-id (:id prof)
:session-id (uuid/next)
:file-id (:id file)
:is-local true
:name "assembled"
:mtype "image/jpeg"})]
(t/is (some? (:error out)))
(t/is (= :max-quote-reached (:code (ex-data (:error out))))))))))
(t/deftest media-object-from-url-pays-the-storage-quota
;; The URL path writes bytes like a plain upload: with a low limit it
;; must be refused instead of bypassing the quota.
(let [prof (th/create-profile* 2)
file (th/create-file* 2 {:profile-id (:id prof)
:project-id (:default-project-id prof)
:is-shared false})
path (th/tempfile "backend_tests/test_files/sample.jpg")]
(with-redefs [media/download-image
(fn [_ _]
{:filename "remote.jpg"
:path path
:mtype "image/jpeg"
:size 312043})]
(with-mocks [mock {:target 'app.config/get
:return (th/config-get-mock {:quotes-media-storage-bytes-per-team 100})}]
(let [out (th/command! {::th/type :create-file-media-object-from-url
::rpc/profile-id (:id prof)
:file-id (:id file)
:is-local true
:url "https://example.com/x.jpg"})]
(t/is (some? (:error out)))
(t/is (= :max-quote-reached (:code (ex-data (:error out))))))))))
(t/deftest media-object-from-url
(let [prof (th/create-profile* 1)
proj (th/create-project* 1 {:profile-id (:id prof)

11
backend/test/backend_tests/rpc_quotes_test.clj
View File

@ -445,7 +445,6 @@
(t/is (true? true) "deduped storage counted once, under quota"))))
(t/deftest media-upload-enforces-storage-quote
;; TEMPORARILY DISABLED (2026-10-19), see body.
(with-mocks [mock {:target 'app.config/get
:return (th/config-get-mock
{:quotes-media-storage-bytes-per-team 100})}]
@ -470,10 +469,12 @@
out (th/multipart-command! params)]
;; TEMPORARILY DISABLED (2026-10-19): upload path no longer checks
;; media-storage-bytes-per-team, so even over quota the upload
;; succeeds. TODO: restore rejection with cheaper implementation.
(t/is (th/success? out)))))
;; 312043 bytes > 100 byte limit → should be rejected
(t/is (not (th/success? out)))
(let [error (:error out)]
(t/is (= :restriction (th/ex-type error)))
(t/is (= :max-quote-reached (th/ex-code error)))
(t/is (= "media-storage-bytes-per-team" (:target (ex-data error))))))))
;; --- SNAPSHOTS (VERSIONS) QUOTES

16
backend/test/backend_tests/srepl_main_test.clj
View File

@ -171,6 +171,22 @@
(t/is (nil? (:profile-id event)))
(t/is (= (:id profile) (get-in event [:props :id]))))))))
(t/deftest delete-file-carries-its-team
;; The storage-usage drain keys on props, so the SREPL event must
;; carry the team: a keyless one trips the "producer stopped keying"
;; warn.
(with-mocks [audit-mock {:target 'app.loggers.audit/insert :return nil}
wrk-mock {:target 'app.jobs/invoke :return nil}]
(let [profile (th/create-profile* 1 {:is-active true})
team-id (:default-team-id profile)
proj (th/create-project* 1 {:profile-id (:id profile) :team-id team-id})
file (th/create-file* 1 {:profile-id (:id profile) :project-id (:id proj)})]
(with-redefs [sys/system th/*system*]
(srepl/delete-file! (str (:id file))))
(let [event (-> @audit-mock :call-args-list first second)]
(t/is (= "delete-file" (:name event)))
(t/is (= team-id (get-in event [:props :team-id])))))))
(t/deftest delete-team-emits-identifiable-audit-event
(with-mocks [audit-mock {:target 'app.loggers.audit/insert :return nil}
wrk-mock {:target 'app.jobs/invoke :return nil}]

1010
backend/test/backend_tests/tasks_team_storage_test.clj Normal file
View File

File diff suppressed because it is too large Load Diff

14
backend/test/backend_tests/worker_test.clj
View File

@ -50,6 +50,20 @@
@#'wrkr/schema:params]]
(t/is (sm/schema? (sm/schema s))))))
(t/deftest the-storage-runner-is-wired
(let [cfg (get main/worker-config [:app.main/storage ::wrk/runner])]
(t/is (some? cfg))
(t/testing "the slow per-team SUM does not share the default runner"
(t/is (= :storage (::wrk/queue cfg)))
(t/is (= (cf/get :worker-storage-parallelism 2) (::wrk/parallelism cfg))))
(t/testing "and the runner resolves jobs, metrics and the pool"
(t/is (contains? cfg ::jobs/defs))
(t/is (contains? cfg ::mtx/metrics))
(t/is (contains? cfg ::db/pool)))))
(t/deftest the-binfile-runner-is-wired
(let [cfg (get main/worker-config [:app.main/binfile ::wrk/runner])]

3
common/src/app/common/flags.cljc
View File

@ -121,6 +121,9 @@
:onboarding
:quotes
:soft-quotes
;; Background per-team media storage usage counter fed by the
;; audit-log consumer. Needs :audit-log to receive events.
:storage-usage
;; Concurrency limit.
:rpc-climit
;; Rate limit.