✨ Port the management client to the new exporter tree

exporter.consumer.api mirrors the legacy client call for call
(claim, progress, session, multipart settle) with ^:async and await
instead of promesa chains, keeping the fetch def seam; and
exporter.consumer.config carries the queue knobs. Runner, worker
and main read the new namespaces, so nothing under exporter.*
depends on app.consumer.* anymore.

AI-assisted-by: muse-spark-1.3-contributor

.serena/memories/exporter/core.md

@@ -50,6 +50,8 @@ on this side: the row of the job, its cancel and its result are the backend's.
 - `exporter.jobs` is the local cancel arm, smaller than the legacy `app.jobs`: one flag per claimed job (`register`/`mark-cancelled` returning true/nil/`cancelled?`/`release`) and one zero-arg check constructor per job (`check-cancelled`, raising `:job-cancelled`). No terminate callbacks and no shared-array signal travel through here — the abort belongs to whoever holds the render (the per-lease interval of `exporter.wasm.scope` terminates its own worker). The runner releases the registry entry on every settle; the legacy runner never did (one leaked entry per job).
 - `exporter.consumer.plan` is the render plan (`prepare-exports`, `count-objects`, `sanitize-file-regex`), dispatching on `exporter.renderer/wasm?` instead of the legacy `headless?`.
 - `exporter.consumer` is the runner: `run-export [cfg claim params]` keeps the legacy beats/watchdog/settle (first breath, render session, multipart complete or `fail-job`, temp release) but renders through the facade (`render [cfg & task]`) with one check from `exporter.jobs` shared between units of work and drivers. `cfg` is the render view; the claim names the job, `params` carries the frozen job params. The settle never rejects; only bugs of this process do.
+- `exporter.consumer.api` is the backend management client (claim/progress/session/settle, transit bodies, multipart artifact), a faithful `^:async` port with no promesa chains and the same `fetch` def seam for hermetic tests.
+- `exporter.consumer.config` holds the consumer knobs (`concurrency` floored at 1, `queue-key`, `queue-name`): the runner, the worker and `exporter.main` read it, so no namespace under `exporter.*` depends on `app.consumer.*` anymore.
 - `exporter.consumer.worker` is the queue poller (`read-payload`, `process`, `poll-once`, `start`/`stop` with a running flag, no promesa): K connections over the `:exporter` queue, claim → `run-export` through the `:render` view → settle before the next pop. `stop` quits every connection and ends every loop (the legacy loops kept failing turns after the stop); boot cleans orphaned temp files first.
 - `exporter.main` wires both pools (`:exporter.browser/pool` and `:exporter.wasm.pool/pool`, each sized to the worker concurrency) plus `:exporter.consumer/worker` (`:concurrency`, `:queue-key`, the `:render` view with both pools as `ref`s and the static uris/flags). No test boots the production wiring: the wasm pool warms one worker eagerly and the worker script in a test process would be the test bundle itself; the wiring is asserted as data and the lifecycle rides fake configs.
 - Async test vars prove their chain only when really awaited: a test passing with 0 assertions means the chain broke upstream.

exporter/src/exporter/consumer.cljs

@@ -33,11 +33,11 @@
    [app.common.exceptions :as ex]
    [app.common.logging :as l]
    [app.common.uuid :as uuid]
-   [app.consumer.api :as api]
    [app.handlers.resources :as rsc]
    [app.jobs.utils :as job.utils]
    [app.util.shell :as sh]
    [cuerdas.core :as str]
+   [exporter.consumer.api :as api]
    [exporter.consumer.plan :as plan]
    [exporter.jobs :as jobs]
    [exporter.renderer :as renderer]))

exporter/src/exporter/consumer/api.cljs

@@ -0,0 +1,152 @@
+;; 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 exporter.consumer.api
+  "The backend management API client.
+
+  The only place of the worker that speaks HTTP: every method of the
+  backend ledger (claim, progress, session, settle) travels through
+  here, with the shared key the exporter was configured with and, when
+  a call renders as a user, the Bearer token of its session. Request
+  bodies travel as transit+json (except the artifact form of
+  `complete-job`, which is multipart, like every other upload) and the
+  answers come back as transit too. There are no promesa chains in
+  this namespace."
+  (:require
+   ["node:fs/promises" :as fsp]
+   ["undici" :as http]
+   [app.common.exceptions :as ex]
+   [app.common.transit :as t]
+   [app.common.uri :as u]
+   [app.config :as cf]
+   [cljs.core :as c]))
+
+(def fetch
+  "The network entry point of the client: a plain def, redefined by the
+  tests of the client, which never need a network."
+  http/fetch)
+
+;; ---- THE REQUEST
+
+(defn- dispatcher
+  "The agent of one request: a self-signed certificate happens in
+  deployments the same way the existing uploads already accept."
+  []
+  (new http/Agent #js {:connect #js {:rejectUnauthorized false}}))
+
+(defn- method-uri
+  [method]
+  (-> (cf/get-internal-uri)
+      (u/ensure-path-slash)
+      (u/join (str "api/management/methods/" (c/name method)))
+      (str)))
+
+(defn- shared-key
+  "What every management call carries: the key id of this exporter plus
+  the key, and the wish to read transit back."
+  []
+  #js {"X-Shared-Key" (str "exporter " cf/management-key)
+       "Accept" "application/transit+json"})
+
+(defn- ^:async parse-response
+  "The answer of a management call: transit when it lands. A non-200 is
+  the failure of the call, with the error body the backend answered;
+  a body that does not decode stays as the plain text it came as."
+  [response]
+  (let [status (.-status response)
+        text   (await (.text response))
+        body   (let [decoded (ex/try! (t/decode-str text))]
+                 (if-not (ex/exception? decoded) decoded text))]
+    (if (= status 200)
+      body
+      (throw (ex/error :type :internal
+                       :code :failed-management-request
+                       :status status
+                       :response body)))))
+
+(defn- ^:async request
+  "POST `params` to a management method as transit+json. The content
+  type travels on every call: without it the backend never parses the
+  body as transit and validates an empty map."
+  [method params]
+  (let [response (await (fetch (method-uri method)
+                               #js {:method     "POST"
+                                    :headers    (doto (shared-key)
+                                                  (unchecked-set "Content-Type" "application/transit+json"))
+                                    :body       (t/encode-str params)
+                                    :dispatcher (dispatcher)}))]
+    (await (parse-response response))))
+
+(defn- ^:async request-multipart
+  "POST a FormData body to a management method. The content type is
+  settled by the FormData itself (boundary included), so only the
+  shared key goes in the headers."
+  [method form-data]
+  (let [response (await (fetch (method-uri method)
+                               #js {:method     "POST"
+                                    :headers    (shared-key)
+                                    :body       form-data
+                                    :dispatcher (dispatcher)}))]
+    (await (parse-response response))))
+
+;; ---- THE METHODS
+
+(defn ^:async claim-job
+  "Request a queued job from the backend. The answer is either
+  `{:action :run :name ... :params ...}` (this poller owns the job) or
+  `{:action :skip ...}` (something else won it, or it is gone)."
+  [job-id scheduled-at]
+  (await (request :claim-job {:job-id job-id :scheduled-at scheduled-at})))
+
+(defn ^:async report-job-progress
+  "The heartbeat of a running job: the stage hitos of the progress
+  contract plus the counters of the current work. A `:skip` says the
+  job can no longer hear the worker (cancelled, aborted, settled)."
+  [job-id progress]
+  (await (request :report-job-progress {:job-id job-id :progress progress})))
+
+(defn ^:async create-job-session
+  "Mint an authentication session of the owner of a running job, so the
+  render views authenticate as the user, not as a machine."
+  [job-id]
+  (await (request :create-job-session {:job-id job-id})))
+
+(defn ^:async complete-job
+  "Close a job whose result is the plain value `result`, with an
+  optional `:resource-id` this worker already owns (the internal path)
+  and an optional `:session-id` to close."
+  ([job-id result]
+   (complete-job job-id result nil))
+  ([job-id result {:keys [resource-id session-id]}]
+   (await (request :complete-job
+                   (cond-> {:job-id job-id :result result}
+                     (some? resource-id) (assoc :resource-id resource-id)
+                     (some? session-id)  (assoc :session-id session-id))))))
+
+(defn ^:async complete-job-with-artifact
+  "Close a job whose result is the artifact itself: the backend stores
+  the bytes and derives the result descriptor. `artifact` says what it
+  is (`:path`, `:filename`, `:mtype`); `ctx` names the job and, when
+  the worker rendered with a session, the one to close."
+  [{:keys [job-id session-id]} {:keys [path filename mtype]}]
+  (let [buffer (await (fsp/readFile (str path)))
+        blob   (new js/Blob #js [buffer] #js {:type mtype})
+        fdata  (new http/FormData)]
+    (.append fdata "job-id" (str job-id))
+    (.append fdata "content" blob filename)
+    (.append fdata "filename" filename)
+    (.append fdata "mtype" (str mtype))
+    (when session-id
+      (.append fdata "session-id" (str session-id)))
+    (await (request-multipart :complete-job fdata))))
+
+(defn ^:async fail-job
+  "Close a job as failed with the rich error the backend stores for
+  `job.error`. `:session-id` closes the render session the same way."
+  [job-id error & {:keys [session-id]}]
+  (await (request :fail-job
+                  (cond-> {:job-id job-id :error error}
+                    (some? session-id) (assoc :session-id session-id)))))

exporter/src/exporter/consumer/config.cljs

@@ -0,0 +1,38 @@
+;; 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 exporter.consumer.config
+  "The knobs of the consumer: how many jobs run at once and which
+  queue they come from. Read live, so a long-lived process obeys a
+  configuration change on the next poll."
+  (:require
+   [app.config :as cf]
+   [cljs.core :as c]))
+
+(def queue-name
+  "The queue this worker consumes: the one the `:export-assets` job-def
+  routes its jobs to through `::jobs/queue-name`."
+  "exporter")
+
+(defn concurrency
+  "K pollers: each one owns its Redis connection, and the blocking
+  connection is the semaphore of the slot, so K is the whole concurrency
+  of the process. At least one: a worker role with zero pollers is a
+  misconfiguration, not a rest.
+
+  Takes the configured value as an argument, so the tests do not
+  journey through the global config; the zero-arity reads it live."
+  ([]
+   (concurrency (c/get cf/config :exporter-worker-concurrency)))
+  ([value]
+   (max 1 (if (number? value) value 2))))
+
+(defn queue-key
+  "The Redis list the dispatcher pushes this queue's payloads through:
+  `penpot.worker.queue:<tenant>:<queue>`."
+  ([] (queue-key (cf/get :tenant)))
+  ([tenant]
+   (str "penpot.worker.queue:" tenant ":" queue-name)))

exporter/src/exporter/consumer/worker.cljs

@@ -32,10 +32,10 @@
    ["ioredis" :as redis]
    [app.common.logging :as l]
    [app.config :as cf]
-   [app.consumer.api :as api]
-   [app.consumer.config :as ccfg]
    [app.jobs.utils :as job.utils]
    [exporter.consumer :as consumer]
+   [exporter.consumer.api :as api]
+   [exporter.consumer.config :as ccfg]
    [exporter.utils.system :as system]))
 
 (def ^:private poll-timeout-s 5)

exporter/src/exporter/main.cljs

@@ -17,10 +17,10 @@
    ["node:worker_threads" :as wt]
    [app.common.logging :as l]
    [app.config :as cf]
-   [app.consumer.config :as ccfg]
    ;; Loaded for their init-key/halt-key methods, which is what makes
    ;; the pools and the queue consumer part of the system below.
    [exporter.browser]
+   [exporter.consumer.config :as ccfg]
    [exporter.consumer.worker]
    [exporter.utils.system :as system]
    [exporter.wasm.pool]))

exporter/test/exporter_tests/consumer_api_test.cljs

@@ -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 exporter-tests.consumer-api-test
+  "The management client of the new tree against a fake http: what it
+  sends (method uri, shared key, shape of the body) is the whole
+  contract this client owns."
+  (:require
+   ["node:fs/promises" :as fsp]
+   ["node:path" :as path]
+   ["undici" :as http]
+   [app.common.transit :as transit]
+   [app.common.uuid :as uuid]
+   [app.config :as cf]
+   [app.util.shell :as sh]
+   [cljs.test :as t :include-macros true]
+   [cuerdas.core :as str]
+   [exporter.consumer.api :as api]))
+
+(def ^:private original-fetch api/fetch)
+
+(def ^:private calls
+  "Every fetch the client made, in order."
+  (atom []))
+
+(defn- fake-fetch
+  [uri request]
+  (swap! calls conj [uri request])
+  (js/Promise.resolve #js {:status 200
+                           :text   (constantly
+                                    (js/Promise.resolve
+                                     (transit/encode-str {:action :run})))}))
+
+(defn- failure-fetch
+  "A fake that answers like a management endpoint that refuses the key."
+  [_uri _request]
+  (js/Promise.resolve #js {:status 403
+                           :text   (constantly (js/Promise.resolve "{}"))}))
+
+(defn- install-fetch
+  ([] (install-fetch fake-fetch))
+  ([the-fake]
+   (set! api/fetch the-fake)
+   (reset! calls [])
+   (fn [] (set! api/fetch original-fetch))))
+
+(defn- last-request
+  []
+  (second (last @calls)))
+
+(defn- fetch-header
+  "One header of the last request, exactly as it would travel."
+  [name]
+  (unchecked-get (.-headers (second (last @calls))) name))
+
+(defn- body-value
+  "One field of the last FormData request."
+  [name]
+  (.get (.-body (last-request)) name))
+
+(defn- decode-body
+  "The body of the last transit request, decoded."
+  []
+  (transit/decode-str (.-body (last-request))))
+
+(defn- last-uri
+  []
+  (first (last @calls)))
+
+(t/deftest ^:async claim-job-posts-the-claim-of-one-poller
+  (let [restore (install-fetch)]
+    (try
+      (let [job-id (uuid/next)]
+        (await (api/claim-job job-id "2026-10-07T00:00:00Z"))
+        (t/is (str/ends-with? (last-uri) "/api/management/methods/claim-job"))
+        (t/is (= (str "exporter " cf/management-key) (fetch-header "X-Shared-Key")))
+        (t/is (= {:job-id job-id :scheduled-at "2026-10-07T00:00:00Z"}
+                 (decode-body))))
+      (catch :default cause
+        (t/is false (str "unexpected failure: " (ex-message cause))))
+      (finally
+        (restore)))))
+
+(t/deftest ^:async json-calls-declare-the-transit-content-type
+  ;; without it the backend never parses the body as transit and
+  ;; validates an empty map (every key a `missing-key`)
+  (let [restore (install-fetch)]
+    (try
+      (await (api/claim-job (uuid/next) "2026-10-07T00:00:00Z"))
+      (t/is (= "application/transit+json"
+               (fetch-header "Content-Type")))
+      (catch :default cause
+        (t/is false (str "unexpected failure: " (ex-message cause))))
+      (finally
+        (restore)))))
+
+(t/deftest ^:async report-job-progress-posts-the-beat
+  (let [restore (install-fetch)]
+    (try
+      (let [job-id   (uuid/next)
+            progress {:stage    :rendering
+                      :counters {:objects {:current 1 :total 2}}}]
+        (await (api/report-job-progress job-id progress))
+        (t/is (str/ends-with? (last-uri) "/api/management/methods/report-job-progress"))
+        (t/is (= {:job-id job-id :progress progress} (decode-body))))
+      (catch :default cause
+        (t/is false (str "unexpected failure: " (ex-message cause))))
+      (finally
+        (restore)))))
+
+(t/deftest ^:async fail-job-carries-the-error-and-the-session
+  (let [restore (install-fetch)]
+    (try
+      (let [job-id (uuid/next)
+            sid    (uuid/next)
+            error  {:type :internal :code :processing-error :hint "boom"}]
+        (await (api/fail-job job-id error))
+        (await (api/fail-job job-id error :session-id sid))
+        (t/is (= {:job-id job-id :error error :session-id sid}
+                 (decode-body))))
+      (catch :default cause
+        (t/is false (str "unexpected failure: " (ex-message cause))))
+      (finally
+        (restore)))))
+
+(t/deftest ^:async complete-job-posts-the-plain-result
+  (let [restore (install-fetch)]
+    (try
+      (let [job-id (uuid/next)
+            sid    (uuid/next)]
+        (await (api/complete-job job-id {:total 1}))
+        (await (api/complete-job job-id {:total 1} {:session-id sid}))
+        (t/is (= {:job-id job-id :result {:total 1} :session-id sid}
+                 (decode-body))))
+      (catch :default cause
+        (t/is false (str "unexpected failure: " (ex-message cause))))
+      (finally
+        (restore)))))
+
+(t/deftest ^:async complete-job-with-artifact-rides-multipart
+  (let [restore (install-fetch)]
+    (try
+      (let [job-id  (uuid/next)
+            sid     (uuid/next)
+            tmpfile (path/join sh/tmpdir (str "test-artifact." (uuid/next)))]
+        (await (fsp/writeFile tmpfile "the export bytes"))
+        (await (api/complete-job-with-artifact
+                {:job-id job-id :session-id sid}
+                {:path     tmpfile
+                 :filename "export.zip"
+                 :mtype    "application/zip"}))
+        (t/is (instance? http/FormData (.-body (last-request))))
+        (t/is (= (str job-id) (body-value "job-id")))
+        (t/is (= "export.zip" (body-value "filename")))
+        (t/is (= "application/zip" (body-value "mtype")))
+        (t/is (= (str sid) (body-value "session-id")))
+        (t/is (= (str "exporter " cf/management-key)
+                 (fetch-header "X-Shared-Key"))))
+      (catch :default cause
+        (t/is false (str "unexpected failure: " (ex-message cause))))
+      (finally
+        (restore)))))
+
+(t/deftest ^:async a-failed-management-answer-is-the-failure-of-the-call
+  (let [restore (install-fetch failure-fetch)]
+    (try
+      (try
+        (await (api/claim-job (uuid/next) "2026-10-07T00:00:00Z"))
+        (t/is false "the call should have rejected on a 403")
+        (catch :default cause
+          (let [data (ex-data cause)]
+            (t/is (= :internal (:type data)))
+            (t/is (= :failed-management-request (:code data)))
+            (t/is (= 403 (:status data))))))
+      (catch :default cause
+        (t/is false (str "unexpected failure: " (ex-message cause))))
+      (finally
+        (restore)))))

exporter/test/exporter_tests/consumer_test.cljs

@@ -14,11 +14,11 @@
    ["undici" :as http]
    [app.common.transit :as transit]
    [app.common.uuid :as uuid]
-   [app.consumer.api :as api]
    [app.util.shell :as sh]
    [cljs.test :as t :include-macros true]
    [cuerdas.core :as cstr]
    [exporter.consumer :as consumer]
+   [exporter.consumer.api :as api]
    [exporter.jobs :as jobs]
    [exporter.renderer :as renderer]))
 

exporter/test/exporter_tests/consumer_worker_test.cljs

@@ -11,10 +11,10 @@
   (:require
    [app.common.transit :as transit]
    [app.common.uuid :as uuid]
-   [app.consumer.api :as api]
    [cljs.test :as t :include-macros true]
    [cuerdas.core :as cstr]
    [exporter.consumer :as consumer]
+   [exporter.consumer.api :as api]
    [exporter.consumer.worker :as worker]))
 
 ;; ---- THE MANAGEMENT FAKE

exporter/test/exporter_tests/exporter_consumer_config_test.cljs

@@ -0,0 +1,26 @@
+;; 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 exporter-tests.exporter-consumer-config-test
+  "The consumer knobs of the new tree: how many jobs run at once and
+  which queue they come from."
+  (:require
+   [cljs.test :as t :include-macros true]
+   [exporter.consumer.config :as ccfg]))
+
+(t/deftest concurrency-is-at-least-one
+  (t/testing "the configured value is taken as given"
+    (t/is (= 2 (ccfg/concurrency 2)))
+    (t/is (= 4 (ccfg/concurrency 4))))
+  (t/testing "zero or less is a misconfiguration, floored to one"
+    (t/is (= 1 (ccfg/concurrency 0)))
+    (t/is (= 1 (ccfg/concurrency -3))))
+  (t/testing "an unset value falls back to the default K"
+    (t/is (= 2 (ccfg/concurrency nil)))))
+
+(t/deftest queue-key-names-tenant-and-queue
+  (t/is (= "penpot.worker.queue:main:exporter"
+           (ccfg/queue-key "main"))))

exporter/test/exporter_tests/main_test.cljs

@@ -10,9 +10,9 @@
   (:require
    ["node:process" :as proc]
    [app.common.logging :as l]
-   [app.consumer.config :as ccfg]
    [cljs.test :as t :include-macros true]
    [exporter.browser :as browser]
+   [exporter.consumer.config :as ccfg]
    [exporter.main :as main]
    [exporter.utils.system :as system]))
 

exporter/test/exporter_tests/runner.cljs

@@ -13,10 +13,12 @@
    [exporter-tests.api-test]
    [exporter-tests.browser-scope-test]
    [exporter-tests.browser-test]
+   [exporter-tests.consumer-api-test]
    [exporter-tests.consumer-config-test]
    [exporter-tests.consumer-plan-test]
    [exporter-tests.consumer-test]
    [exporter-tests.consumer-worker-test]
+   [exporter-tests.exporter-consumer-config-test]
    [exporter-tests.exporter-consumer-plan-test]
    [exporter-tests.exporter-jobs-test]
    [exporter-tests.jobs-test]
@@ -45,9 +47,11 @@
 (def test-namespaces
   ['exporter-tests.browser-scope-test
    'exporter-tests.browser-test
+   'exporter-tests.consumer-api-test
    'exporter-tests.consumer-plan-test
    'exporter-tests.consumer-test
    'exporter-tests.consumer-worker-test
+   'exporter-tests.exporter-consumer-config-test
    'exporter-tests.exporter-consumer-plan-test
    'exporter-tests.exporter-jobs-test
    'exporter-tests.jobs-test