✨ Add the worker config and the management API client

The worker side of the exporter gets its config: `PENPOT_EXPORTER_ROLES`
(`http` by default, `http,worker` during the transition) and
`PENPOT_EXPORTER_WORKER_CONCURRENCY` (K pollers, the concurrency of the
process, floored at one), both read from the live config, never at
import time; every accessor is a plain function so the tests do not
journey through globals. The queue key it polls is
`penpot.worker.queue:<tenant>:exporter`, the same one the dispatcher
pushes.

The management client (`app.consumer.api`) is the only place of the
worker that speaks HTTP: claim, progress, session, complete and fail,
all POSTs with `X-Shared-Key: exporter <key>`, transit+json bodies and
transit answers. The artifact form of `complete-job` rides multipart
(the undici FormData the uploads already use); a non-200 answer arrives
as the failure of the call, carrying the status and the error body.

AI-assisted-by: glm-5.3-flash

exporter/src/app/config.cljs

@@ -33,6 +33,10 @@
    :exporter-max-jobs-per-profile 2
    :exporter-queue-max 64
    :exporter-job-ttl 3600
+   ;; the roles this instance serves: by default only the http service;
+   ;; `worker` turns the process into a consumer of the backend jobs
+   :exporter-roles "http"
+   :exporter-worker-concurrency 2
    :wasm-worker-pool-max 2
    :wasm-worker-pool-min 1
    :wasm-worker-idle-timeout 300
@@ -55,6 +59,8 @@
    [:exporter-max-jobs-per-profile {:optional true} ::sm/int]
    [:exporter-queue-max {:optional true} ::sm/int]
    [:exporter-job-ttl {:optional true} ::sm/int]
+   [:exporter-roles {:optional true} :string]
+   [:exporter-worker-concurrency {:optional true} ::sm/int]
    [:wasm-worker-pool-max {:optional true} ::sm/int]
    [:wasm-worker-pool-min {:optional true} ::sm/int]
    [:wasm-worker-idle-timeout {:optional true} ::sm/int]

exporter/src/app/consumer/api.cljs

@@ -0,0 +1,153 @@
+;; 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.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."
+  (: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]
+   [promesa.core :as p]))
+
+(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"})
+
+(def ^:private 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."
+  (fn [response]
+    (let [status (.-status response)]
+      (->> (.text response)
+           (p/fmap (fn [text]
+                     (let [body (let [decoded (ex/try! (t/decode-str text))]
+                                  (if-not (ex/exception? decoded) decoded text))]
+                       (if (= status 200)
+                         body
+                         (ex/raise :type :internal
+                                   :code :failed-management-request
+                                   :status status
+                                   :response body)))))))))
+
+(defn- request
+  "POST `params` to a management method as transit+json."
+  [method params]
+  (->> (fetch (method-uri method)
+              #js {:method     "POST"
+                   :headers    (shared-key)
+                   :body       (t/encode-str params)
+                   :dispatcher (dispatcher)})
+       (p/mcat parse-response)))
+
+(defn- 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]
+  (->> (fetch (method-uri method)
+              #js {:method     "POST"
+                   :headers    (shared-key)
+                   :body       form-data
+                   :dispatcher (dispatcher)})
+       (p/mcat parse-response)))
+
+;; ---- THE METHODS
+
+(defn 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]
+  (request :claim-job {:job-id job-id :scheduled-at scheduled-at}))
+
+(defn 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]
+  (request :report-job-progress {:job-id job-id :progress progress}))
+
+(defn 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]
+  (request :create-job-session {:job-id job-id}))
+
+(defn 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]}]
+   (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 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]}]
+  (->> (fsp/readFile (str path))
+       (p/fmap (fn [buffer]
+                 (let [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)))
+                   fdata)))
+       (p/mcat (fn [fdata]
+                 (request-multipart :complete-job fdata)))))
+
+(defn 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]}]
+  (request :fail-job
+           (cond-> {:job-id job-id :error error}
+             (some? session-id) (assoc :session-id session-id))))

exporter/src/app/consumer/config.cljs

@@ -0,0 +1,68 @@
+;; 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.consumer.config
+  "The worker role of the exporter configuration.
+
+  The exporter process may serve its http surfaces, consume the
+  backend `:exporter` queue, or both: `PENPOT_EXPORTER_ROLES` names the
+  roles (`http` by default, `http,worker` during the transition). The
+  worker pieces never read these values at import time: each accessor
+  reads the live configuration, so a long-lived process can be
+  restarted with another role without code changes."
+  (:require
+   [app.config :as cf]
+   [cljs.core :as c]
+   [cuerdas.core :as str]))
+
+(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- scan-roles
+  [spec]
+  (->> (if (string? spec)
+         (str/split spec ",")
+         spec)
+       (map str/trim)
+       (remove str/blank?)
+       (map keyword)
+       (into #{})))
+
+(defn roles
+  "The set of roles this instance serves, as keywords."
+  ([] (roles (cf/get :exporter-roles)))
+  ([spec]
+   (if (nil? spec)
+     #{:http}
+     (scan-roles spec))))
+
+(defn worker-enabled?
+  "True when this instance also consumes the exporter queue."
+  ([] (contains? (roles) :worker))
+  ([roles]
+   (contains? roles :worker)))
+
+(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/test/exporter_tests/api_test.cljs

@@ -0,0 +1,180 @@
+;; 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.api-test
+  "The management client of the worker 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.exceptions :as ex]
+   [app.common.transit :as transit]
+   [app.common.uuid :as uuid]
+   [app.config :as cf]
+   [app.consumer.api :as api]
+   [app.util.shell :as sh]
+   [cljs.core :as c]
+   [cljs.test :as t :include-macros true]
+   [cuerdas.core :as str]
+   [promesa.core :as p]))
+
+(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])
+  (p/resolved #js {:status 200
+                   :text   (constantly (p/resolved (transit/encode-str {:action :run})))}))
+
+(defn- failure-fetch
+  "A fake that answers like a management endpoint that refuses the key."
+  [_uri _request]
+  (p/resolved #js {:status 403
+                   :text   (constantly (p/resolved "{}"))}))
+
+(defn- install-fetch!
+  "Put a fake in place and return the thunk that puts the original
+  back. The thunk belongs at the end of the promise chain's `p/do`, so
+  the fake covers the whole async run."
+  ([] (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)))
+
+(defn- with-restore-and-done
+  "Run `thunk`, then put the original fetch back, then say the test is
+  finished: the rest never waits on the assertions, so a broken chain
+  reports a failure, never a hang."
+  [thunk restore! done]
+  (p/do thunk (restore!) (done)))
+
+(t/deftest claim-job-posts-the-claim-of-one-poller
+  (t/async done
+    (let [job-id   (uuid/next)
+          restore! (install-fetch!)]
+      (with-restore-and-done
+        (p/fmap (fn [_]
+                  (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))))
+                (api/claim-job job-id "2026-10-07T00:00:00Z"))
+        restore!
+        done))))
+
+(t/deftest report-job-progress-posts-the-beat
+  (t/async done
+    (let [job-id   (uuid/next)
+          progress {:stage :rendering
+                    :counters {:objects {:current 1 :total 2}}}
+          restore! (install-fetch!)]
+      (with-restore-and-done
+        (p/fmap (fn [_]
+                  (t/is (str/ends-with? (last-uri) "/api/management/methods/report-job-progress"))
+                  (t/is (= {:job-id job-id :progress progress} (decode-body))))
+                (api/report-job-progress job-id progress))
+        restore!
+        done))))
+
+(t/deftest fail-job-carries-the-error-and-the-session
+  (t/async done
+    (let [job-id   (uuid/next)
+          sid      (uuid/next)
+          error    {:type :internal :code :processing-error :hint "boom"}
+          restore! (install-fetch!)]
+      (with-restore-and-done
+        (p/mcat (fn [_]
+                  (p/fmap (fn [_]
+                            (t/is (= {:job-id job-id :error error :session-id sid}
+                                     (decode-body))))
+                          (api/fail-job job-id error :session-id sid)))
+                (api/fail-job job-id error))
+        restore!
+        done))))
+
+(t/deftest complete-job-posts-the-plain-result
+  (t/async done
+    (let [job-id   (uuid/next)
+          sid      (uuid/next)
+          restore! (install-fetch!)]
+      (with-restore-and-done
+        (p/mcat (fn [_]
+                  (p/fmap (fn [_]
+                            (t/is (= {:job-id job-id :result {:total 1} :session-id sid}
+                                     (decode-body))))
+                          (api/complete-job job-id {:total 1} {:session-id sid})))
+                (api/complete-job job-id {:total 1}))
+        restore!
+        done))))
+
+(t/deftest complete-job-with-artifact-rides-multipart
+  (t/async done
+    (let [job-id   (uuid/next)
+          sid      (uuid/next)
+          tmpfile  (path/join sh/tmpdir (str "test-artifact." (uuid/next)))
+          restore! (install-fetch!)]
+      (with-restore-and-done
+        (p/mcat (fn [_]
+                  (p/fmap (fn [_]
+                            (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"))))
+                          (api/complete-job-with-artifact
+                           {:job-id job-id :session-id sid}
+                           {:path     tmpfile
+                            :filename "export.zip"
+                            :mtype    "application/zip"})))
+                (->> (fsp/writeFile tmpfile "the export bytes")
+                     (p/fmap (fn [_] nil))))
+        restore!
+        done))))
+
+(t/deftest a-failed-management-answer-is-the-failure-of-the-call
+  (t/async done
+    (let [restore! (install-fetch! failure-fetch)]
+      (with-restore-and-done
+        (p/catch (fn [error]
+                   (let [data (ex-data error)]
+                     (t/is (= :internal (:type data)))
+                     (t/is (= :failed-management-request (:code data)))
+                     (t/is (= 403 (:status data)))))
+                 (api/claim-job (uuid/next) "2026-10-07T00:00:00Z"))
+        restore!
+        done))))

exporter/test/exporter_tests/consumer_config_test.cljs

@@ -0,0 +1,41 @@
+;; 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-config-test
+  "The worker side of the config: the roles a process serves and how
+  many pollers a worker owns."
+  (:require
+   [app.consumer.config :as ccfg]
+   [cljs.test :as t :include-macros true]
+   [cuerdas.core :as str]))
+
+(t/deftest roles-parse-the-env-list
+  (t/testing "the default keeps the http surfaces only"
+    (t/is (= #{:http} (ccfg/roles "http"))))
+  (t/is (= #{:http :worker} (ccfg/roles "http,worker")))
+  (t/is (= #{:http :worker} (ccfg/roles "http, worker")))
+  (t/testing "the parser does not invent roles from noise"
+    (t/is (= #{} (ccfg/roles "")))
+    (t/is (= #{:worker} (ccfg/roles ",worker")))))
+
+(t/deftest worker-enabled-reads-the-roles
+  (t/is (true? (ccfg/worker-enabled? #{:http :worker})))
+  (t/is (true? (ccfg/worker-enabled? #{:worker})))
+  (t/is (false? (ccfg/worker-enabled? #{:http}))))
+
+(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/runner.cljs

@@ -10,6 +10,8 @@
    [cljs.test :as t]
    [clojure.string :as str]
    [clojure.tools.cli :refer [parse-opts]]
+   [exporter-tests.api-test]
+   [exporter-tests.consumer-config-test]
    [exporter-tests.export-shapes-test]
    [exporter-tests.jobs-test]
    [exporter-tests.renderer-svg-test]
@@ -23,6 +25,8 @@
 (def test-namespaces
   ['exporter-tests.export-shapes-test
    'exporter-tests.jobs-test
+   'exporter-tests.api-test
+   'exporter-tests.consumer-config-test
    'exporter-tests.renderer-svg-test
    'exporter-tests.scheduler-test
    'exporter-tests.shell-test