mirror of
https://github.com/penpot/penpot.git
synced 2026-10-30 22:40:33 +00:00
bb76b29fe2
Teach the file-gc task to heal files whose stored data fails schema or semantic validation instead of failing the nightly run: read once (sanitize, migrate in memory, check the scheduled revn), run the shared best-effort repair loop, then take the row lock and persist the snapshot, marking the file trimmed. A new revision orphaned the current file thumbnail (the list joins on revn), so a run that changes bytes touches modified-at instead (the ETag covers it) and no-op runs touch nothing; the change is detected against the decoded pre-sanitize data, which get-file now returns as :orig-data. The handler reports a result map with a keyword reason (:cleaned, :gone, :stale-revn, :unfixable, :error) and the tiered-storage offload submit runs outside the result handling, so a queue failure after a good write stays truthful and only logs its own error. The library-user scan moved out of the write transaction, so long scans no longer trip idle-in-transaction timeouts. Repair convergence lives in one shared loop used by file-gc, the debug endpoint and the srepl helper: validate, apply the computed repair changes and repeat up to ten times, returning the input untouched when it throws or does not converge. file-gc revalidates after the loop and raises :repair-did-not-converge, so a partial cleanup is never stored as clean: the classifier reports :unfixable, the file stays unmarked and the scheduler retries it. Validation skips shapes without :shape-ref (copy roots legitimately carry none), so the repair converges instead of crashing on the get-shape precondition; a dedicated repair for copies that genuinely lost their ref is future work. Libraries load through the weak-map loader, and that loader was broken: the threading macro passed :data to update as the map itself, so every library load threw and the repair loop swallowed the exception, silently skipping every repair for files with linked libraries. The pipeline is now explicit: load, realize pointers (they take [cfg file], so the reversed -> threading was returning the Integrant system map) and then sanitize. Unconverted types broke ref resolution (detaching copies) and hid real uses from instance-of?, so libraries are sanitized on load, user files are sanitized inside the library-user scan, and the srepl helper and the debug endpoint sanitize too. Only migrate when migrations pend. sanitize-file-data now returns the input instance unchanged when neither the nil-entry drop nor the schema decode repairs anything: lazy containers (objects-map) keep their identity and file-gc can tell a real fix from a no-op by comparing with the pre-sanitize data. The walk still happens on every call, and that cost is documented on the loader. Migration 0027 heals stored data durably with the same recipe the read path applies: schema-decode unconverted types (precompiled decoders, explicit component-objects pass) plus the legacy-57 nil page cleanup that reappeared through a write path. clean-file keeps only its post-decode fixes and documents the single sanitize pass. Change-level guards reject :add-page without a valid page id (structured error, not assert, so release CLJS still enforces it) and conflicting id/page pairs. Tests pin the contract with production-shaped corruption builders: file-gc repair and the outcome matrix, loop convergence and give-up branches, the :shape-ref guard both ways, page guards, sanitize idempotence and identity (plain and objects-map), the 0027 migration, a remote copy resolved through the loaded library and the non-convergence report; existing file suites move to the result map through the run-file-gc! helper. Closes #12433 Follow-up: https://github.com/penpot/penpot/issues/12509 AI-assisted-by: muse-spark-1.3-contributor AI-assisted-by: glm-5.3-flash
735 lines
24 KiB
Clojure
735 lines
24 KiB
Clojure
;; 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 INC Sucursal en España SL
|
|
|
|
(ns backend-tests.helpers
|
|
(:require
|
|
[app.auth]
|
|
[app.common.data :as d]
|
|
[app.common.data.macros :as dm]
|
|
[app.common.exceptions :as ex]
|
|
[app.common.features :as cfeat]
|
|
[app.common.flags :as flags]
|
|
[app.common.pprint :as pp]
|
|
[app.common.schema :as sm]
|
|
[app.common.spec :as us]
|
|
[app.common.time :as ct]
|
|
[app.common.transit :as tr]
|
|
[app.common.uuid :as uuid]
|
|
[app.config :as cf]
|
|
[app.db :as db]
|
|
[app.main :as main]
|
|
[app.media]
|
|
[app.media :as-alias mtx]
|
|
[app.metrics :as-alias metrics]
|
|
[app.migrations]
|
|
[app.msgbus :as-alias mbus]
|
|
[app.rpc :as-alias rpc]
|
|
[app.rpc.climit :as-alias climit]
|
|
[app.rpc.commands.auth :as cmd.auth]
|
|
[app.rpc.commands.files :as files]
|
|
[app.rpc.commands.files-create :as files.create]
|
|
[app.rpc.commands.files-update :as files.update]
|
|
[app.rpc.commands.projects :as projects]
|
|
[app.rpc.commands.teams :as teams]
|
|
[app.rpc.helpers :as rph]
|
|
[app.util.blob :as blob]
|
|
[app.util.services :as sv]
|
|
[app.worker :as wrk]
|
|
[app.worker.runner]
|
|
[clojure.java.io :as io]
|
|
[clojure.spec.alpha :as s]
|
|
[clojure.test :as t]
|
|
[cuerdas.core :as str]
|
|
[datoteka.fs :as fs]
|
|
[environ.core :refer [env]]
|
|
[expound.alpha :as expound]
|
|
[integrant.core :as ig]
|
|
[mockery.core :as mk]
|
|
[promesa.core :as p]
|
|
[promesa.exec :as px]
|
|
[ring.core.protocols :as rcp]
|
|
[yetti.request :as yrq]
|
|
[yetti.response :as yres])
|
|
(:import
|
|
java.io.PipedInputStream
|
|
java.io.PipedOutputStream
|
|
java.util.UUID
|
|
org.postgresql.ds.PGSimpleDataSource))
|
|
|
|
(def ^:dynamic *system* nil)
|
|
(def ^:dynamic *pool* nil)
|
|
|
|
(def default
|
|
{:database-uri "postgresql://postgres/penpot_test"
|
|
:redis-uri "redis://valkey/1"
|
|
:auto-file-snapshot-every 1
|
|
:file-data-backend "db"})
|
|
|
|
(def config
|
|
(cf/read-config :prefix "penpot-test"
|
|
:default (merge cf/default default)))
|
|
|
|
(def default-flags
|
|
[:enable-secure-session-cookies
|
|
:enable-email-verification
|
|
:enable-smtp
|
|
:enable-quotes
|
|
:enable-rpc-climit
|
|
:enable-auto-file-snapshot
|
|
:disable-file-validation])
|
|
|
|
(defn init-config
|
|
([next]
|
|
(init-config nil next))
|
|
([extra-flags next]
|
|
(let [flags (into default-flags extra-flags)]
|
|
(with-redefs [app.config/flags (flags/parse flags/default flags)
|
|
app.config/config config
|
|
app.loggers.audit/submit (constantly nil)
|
|
app.auth/derive-password identity
|
|
app.auth/verify-password (fn [a b] {:valid (= a b)})
|
|
app.common.features/get-enabled-features
|
|
(fn [& _] app.common.features/supported-features)]
|
|
(cf/validate! :exit-on-error? false)
|
|
(fs/create-dir "/tmp/penpot")
|
|
(next)))))
|
|
|
|
(defn init-system
|
|
[next]
|
|
(let [templates [{:id "test"
|
|
:name "test"
|
|
:file-uri "test"
|
|
:thumbnail-uri "test"
|
|
:path (-> "backend_tests/test_files/template.penpot" io/resource fs/path)}]
|
|
system (-> (merge main/system-config main/worker-config)
|
|
(assoc-in [:app.redis/client :app.redis/uri] (:redis-uri config))
|
|
(assoc-in [::db/pool ::db/uri] (:database-uri config))
|
|
(assoc-in [::db/pool ::db/username] (:database-username config))
|
|
(assoc-in [::db/pool ::db/password] (:database-password config))
|
|
(assoc-in [:app.rpc/methods :app.setup/templates] templates)
|
|
(assoc-in [:app.rpc/methods :app.setup/templates] templates)
|
|
(update :app.rpc/rlimit assoc
|
|
:app.loggers.mattermost/reporter nil
|
|
:app.loggers.database/reporter nil)
|
|
(update :app.rpc/methods assoc
|
|
:app.setup/templates templates
|
|
:app.loggers.mattermost/reporter nil
|
|
:app.loggers.database/reporter nil)
|
|
(dissoc :app.srepl/server
|
|
:app.http/server
|
|
:app.http/route
|
|
:app.setup/templates
|
|
:app.http.oauth/handler
|
|
:app.notifications/handler
|
|
:app.loggers.mattermost/reporter
|
|
:app.loggers.database/reporter
|
|
:app.loggers.metrics/reporter
|
|
:app.worker/cron
|
|
:app.worker/dispatcher
|
|
[:app.main/default :app.worker/runner]
|
|
[:app.main/webhook :app.worker/runner]))
|
|
_ (ig/load-namespaces system)
|
|
system (-> (ig/expand system) (ig/init))]
|
|
(try
|
|
(binding [*system* system
|
|
*pool* (:app.db/pool system)]
|
|
(next))
|
|
(finally
|
|
(ig/halt! system)))))
|
|
|
|
(def state-init
|
|
(t/compose-fixtures init-config init-system))
|
|
|
|
(defn database-reset
|
|
[next]
|
|
(let [sql (str "SELECT table_name "
|
|
" FROM information_schema.tables "
|
|
" WHERE table_schema = 'public' "
|
|
" AND table_name != 'migrations';")]
|
|
(db/transact! *pool* (fn [conn]
|
|
(db/exec-one! conn ["SET CONSTRAINTS ALL DEFERRED"])
|
|
(db/exec-one! conn ["SET LOCAL rules.deletion_protection TO off"])
|
|
(let [result (->> (db/exec! conn [sql])
|
|
(map :table-name))]
|
|
(doseq [table result]
|
|
(db/exec! conn [(str "delete from " table ";")])))))
|
|
(next)))
|
|
|
|
(defn clean-storage
|
|
[next]
|
|
(let [path (fs/path "/tmp/penpot")]
|
|
(when (fs/exists? path)
|
|
(fs/delete (fs/path "/tmp/penpot")))
|
|
(fs/create-dir "/tmp/penpot")
|
|
(next)))
|
|
|
|
(defn serial
|
|
[& funcs]
|
|
(fn [next]
|
|
(loop [f (first funcs)
|
|
fs (rest funcs)]
|
|
(when f
|
|
(let [prm (promise)]
|
|
(f #(deliver prm true))
|
|
(deref prm)
|
|
(recur (first fs)
|
|
(rest fs)))))
|
|
(next)))
|
|
|
|
(defn mk-uuid
|
|
[prefix & args]
|
|
(UUID/nameUUIDFromBytes (-> (apply str prefix args)
|
|
(.getBytes "UTF-8"))))
|
|
;; --- FACTORIES
|
|
|
|
(defn create-profile*
|
|
([i] (create-profile* *system* i {}))
|
|
([i params] (create-profile* *system* i params))
|
|
([system i params]
|
|
(let [params (merge {:id (mk-uuid "profile" i)
|
|
:fullname (str "Profile " i)
|
|
:email (str "profile" i ".test@nodomain.com")
|
|
:password "Test123!"
|
|
:is-demo false}
|
|
params)]
|
|
(db/run! system
|
|
(fn [cfg]
|
|
(->> params
|
|
(cmd.auth/create-profile cfg)
|
|
(cmd.auth/create-profile-rels cfg)))))))
|
|
|
|
(defn create-project*
|
|
([i params] (create-project* *system* i params))
|
|
([system i {:keys [profile-id team-id] :as params}]
|
|
|
|
(assert (uuid? profile-id))
|
|
(assert (uuid? team-id))
|
|
(let [timestamp (ct/now)]
|
|
(db/run! system
|
|
(fn [cfg]
|
|
(->> (merge {:id (mk-uuid "project" i)
|
|
:name (str "project" i)}
|
|
params
|
|
{::rpc/request-at timestamp})
|
|
(#'projects/create-project cfg)))))))
|
|
|
|
(defn create-file*
|
|
([i params]
|
|
(create-file* *system* i params))
|
|
([system i {:keys [profile-id project-id] :as params}]
|
|
(dm/assert! "expected uuid" (uuid? profile-id))
|
|
(dm/assert! "expected uuid" (uuid? project-id))
|
|
(db/tx-run! system
|
|
(fn [system]
|
|
(let [features (cfeat/get-enabled-features cf/flags)]
|
|
(files.create/create-file system
|
|
(merge {:id (mk-uuid "file" i)
|
|
:name (str "file" i)
|
|
:features features}
|
|
params)))))))
|
|
|
|
(defn mark-file-deleted*
|
|
([params]
|
|
(mark-file-deleted* *system* params))
|
|
([conn {:keys [id] :as params}]
|
|
(#'files/mark-file-deleted conn {} id)))
|
|
|
|
(defn create-team*
|
|
([i params] (create-team* *system* i params))
|
|
([system i {:keys [profile-id] :as params}]
|
|
(us/assert uuid? profile-id)
|
|
(dm/with-open [conn (db/open system)]
|
|
(let [id (mk-uuid "team" i)
|
|
features (cfeat/get-enabled-features cf/flags)]
|
|
(teams/create-team {::db/conn conn} {:id id
|
|
:profile-id profile-id
|
|
:features features
|
|
:name (str "team" i)})))))
|
|
|
|
(defn create-file-media-object*
|
|
([params] (create-file-media-object* *system* params))
|
|
([system {:keys [name width height mtype file-id is-local media-id]
|
|
:or {name "sample" width 100 height 100 mtype "image/svg+xml" is-local true}}]
|
|
|
|
(dm/with-open [conn (db/open system)]
|
|
(db/insert! conn :file-media-object
|
|
{:id (uuid/next)
|
|
:file-id file-id
|
|
:is-local is-local
|
|
:name name
|
|
:media-id media-id
|
|
:width width
|
|
:height height
|
|
:mtype mtype}))))
|
|
|
|
(defn link-file-to-library*
|
|
([params] (link-file-to-library* *system* params))
|
|
([system {:keys [file-id library-id] :as params}]
|
|
(dm/with-open [conn (db/open system)]
|
|
(#'files/link-file-to-library conn {:file-id file-id :library-id library-id}))))
|
|
|
|
(defn create-complaint-for
|
|
[system {:keys [id created-at type]}]
|
|
(dm/with-open [conn (db/open system)]
|
|
(db/insert! conn :profile-complaint-report
|
|
{:profile-id id
|
|
:created-at (or created-at (ct/now))
|
|
:type (name type)
|
|
:content (db/tjson {})})))
|
|
|
|
(defn create-global-complaint-for
|
|
[system {:keys [email type created-at]}]
|
|
(dm/with-open [conn (db/open system)]
|
|
(db/insert! conn :global-complaint-report
|
|
{:email email
|
|
:type (name type)
|
|
:created-at (or created-at (ct/now))
|
|
:content (db/tjson {})})))
|
|
|
|
(defn create-team-role*
|
|
([params] (create-team-role* *system* params))
|
|
([system {:keys [team-id profile-id role] :or {role :owner}}]
|
|
(dm/with-open [conn (db/open system)]
|
|
(#'teams/create-team-role {::db/conn conn}
|
|
{:team-id team-id
|
|
:profile-id profile-id
|
|
:role role}))))
|
|
|
|
(defn create-project-role*
|
|
([params] (create-project-role* *system* params))
|
|
([system {:keys [project-id profile-id role] :or {role :owner}}]
|
|
(dm/with-open [conn (db/open system)]
|
|
(#'teams/create-project-role conn profile-id project-id role))))
|
|
|
|
(defn create-file-role*
|
|
([params] (create-file-role* *system* params))
|
|
([system {:keys [file-id profile-id role] :or {role :owner}}]
|
|
(dm/with-open [conn (db/open system)]
|
|
(files.create/create-file-role! conn {:file-id file-id
|
|
:profile-id profile-id
|
|
:role role}))))
|
|
|
|
(defn update-file*
|
|
([params] (update-file* *system* params))
|
|
([system {:keys [file-id changes session-id profile-id revn]
|
|
:or {session-id (uuid/next) revn 0}}]
|
|
(-> system
|
|
(assoc ::files.update/timestamp (ct/now))
|
|
(db/tx-run! (fn [{:keys [::db/conn] :as system}]
|
|
(let [file (files.update/get-file conn file-id)]
|
|
(#'files.update/update-file* system
|
|
{:id file-id
|
|
:revn revn
|
|
:vern 0
|
|
:file file
|
|
:features (:features file)
|
|
:changes changes
|
|
:session-id session-id
|
|
:profile-id profile-id})))))))
|
|
|
|
(declare command!)
|
|
|
|
(defn update-file! [& {:keys [profile-id file-id changes revn] :or {revn 0}}]
|
|
(let [features (cfeat/get-enabled-features cf/flags)
|
|
params {::type :update-file
|
|
::rpc/profile-id profile-id
|
|
:id file-id
|
|
:session-id (uuid/random)
|
|
:revn revn
|
|
:vern 0
|
|
:features features
|
|
:changes changes}
|
|
out (command! params)]
|
|
(t/is (nil? (:error out)))
|
|
(:result out)))
|
|
|
|
|
|
(defn create-webhook*
|
|
([params] (create-webhook* *system* params))
|
|
([system {:keys [team-id id uri mtype is-active]
|
|
:or {is-active true
|
|
mtype "application/json"
|
|
uri "http://example.com/webhook"}}]
|
|
(db/run! system (fn [{:keys [::db/conn]}]
|
|
(db/insert! conn :webhook
|
|
{:id (or id (uuid/next))
|
|
:team-id team-id
|
|
:uri uri
|
|
:is-active is-active
|
|
:mtype mtype})))))
|
|
|
|
;; --- RPC HELPERS
|
|
|
|
(defn handle-error
|
|
[^Throwable err]
|
|
(if (instance? java.util.concurrent.ExecutionException err)
|
|
(handle-error (.getCause err))
|
|
err))
|
|
|
|
(defmacro try-on!
|
|
[expr]
|
|
`(try
|
|
(let [result# ~expr
|
|
result# (cond-> result# (rph/wrapped? result#) deref)]
|
|
{:error nil
|
|
:result result#})
|
|
(catch Exception e#
|
|
{:error (handle-error e#)
|
|
:result nil})))
|
|
|
|
(defn command!
|
|
[{:keys [::type] :as data}]
|
|
(let [[mdata method-fn] (get-in *system* [:app.rpc/methods type])]
|
|
(when-not method-fn
|
|
(ex/raise :type :assertion
|
|
:code :rpc-method-not-found
|
|
:hint (str/ffmt "rpc method '%' not found" (name type))))
|
|
|
|
;; (app.common.pprint/pprint (:app.rpc/methods *system*))
|
|
(try-on! (method-fn (-> data
|
|
(dissoc ::type)
|
|
(assoc :app.rpc/request-at (ct/now)))))))
|
|
|
|
(defn management-command!
|
|
[{:keys [::type] :as data}]
|
|
(let [[_ method-fn] (get-in *system* [:app.rpc/management-methods type])]
|
|
(when-not method-fn
|
|
(ex/raise :type :assertion
|
|
:code :rpc-method-not-found
|
|
:hint (str/ffmt "management rpc method '%' not found" (name type))))
|
|
(try-on! (method-fn (-> data
|
|
(dissoc ::type)
|
|
(assoc :app.rpc/request-at (ct/now)))))))
|
|
|
|
(defn run-task!
|
|
([name]
|
|
(run-task! name {}))
|
|
([name params]
|
|
(wrk/invoke! (-> *system*
|
|
(assoc ::wrk/task name)
|
|
(assoc ::wrk/params params)))))
|
|
|
|
(def sql:pending-tasks
|
|
"select t.* from task as t
|
|
where t.status = 'new'
|
|
order by t.priority desc, t.scheduled_at")
|
|
|
|
(defn run-pending-tasks!
|
|
[]
|
|
(db/tx-run! *system* (fn [{:keys [::db/conn] :as cfg}]
|
|
(let [tasks (->> (db/exec! conn [sql:pending-tasks])
|
|
(map #'app.worker.runner/decode-task-row))]
|
|
(doseq [task tasks]
|
|
(let [cfg (-> cfg
|
|
(assoc :app.worker.runner/queue (:queue task))
|
|
(assoc :app.worker.runner/id 0))]
|
|
(#'app.worker.runner/run-task cfg task)))))))
|
|
|
|
(defn metrics-stub
|
|
"A metrics instance with no collectors, where every `run!` is a no-op."
|
|
[]
|
|
(reify metrics/IMetrics
|
|
(get-handler [_] nil)
|
|
(get-collector [_ _] nil)
|
|
(get-registry [_] nil)))
|
|
|
|
(defn init-climit
|
|
"Build a concurrency limiter component from an inline config map. The
|
|
optional second argument is the metrics instance to publish to (a
|
|
`metrics-stub` by default, where every `run!` is a no-op)."
|
|
([config]
|
|
(init-climit config (metrics-stub)))
|
|
([config metrics]
|
|
(let [file (doto (java.io.File/createTempFile "climit" ".edn")
|
|
(.deleteOnExit))]
|
|
(spit file (pr-str config))
|
|
(ig/init-key ::rpc/climit
|
|
{::climit/enabled true
|
|
::climit/config (.getAbsolutePath file)
|
|
::metrics/metrics metrics}))))
|
|
|
|
(defmacro with-climit
|
|
"Build a concurrency limiter from an inline config map, run the body and
|
|
halt it, so the republisher thread it starts never outlives the test. An
|
|
optional third binding element is the metrics instance to publish to."
|
|
[[sym config & [metrics]] & body]
|
|
`(let [~sym (init-climit ~config ~@(when metrics [metrics]))]
|
|
(try
|
|
~@body
|
|
(finally
|
|
(ig/halt-key! ::rpc/climit ~sym)))))
|
|
|
|
;; --- UTILS
|
|
|
|
(defn print-error!
|
|
[error]
|
|
(let [data (ex-data error)]
|
|
(cond
|
|
(= :spec-validation (:code data))
|
|
(println
|
|
(us/pretty-explain data))
|
|
|
|
(= :params-validation (:code data))
|
|
(println
|
|
(sm/humanize-explain (::sm/explain data)))
|
|
|
|
(= :data-validation (:code data))
|
|
(println
|
|
(sm/humanize-explain (::sm/explain data)))
|
|
|
|
(= :service-error (:type data))
|
|
(print-error! (.getCause ^Throwable error))
|
|
|
|
:else
|
|
(.printStackTrace ^Throwable error))))
|
|
|
|
(defn print-result!
|
|
[{:keys [error result]}]
|
|
(if error
|
|
(do
|
|
(println "====> START ERROR")
|
|
(print-error! error)
|
|
(println "====> END ERROR"))
|
|
(do
|
|
(println "====> START RESPONSE")
|
|
(pp/pprint result)
|
|
(println "====> END RESPONSE"))))
|
|
|
|
(defn exception?
|
|
[v]
|
|
(instance? Throwable v))
|
|
|
|
(defn ex-info?
|
|
[v]
|
|
(ex/error? v))
|
|
|
|
(defn ex-type
|
|
[e]
|
|
(:type (ex-data e)))
|
|
|
|
(defn ex-code
|
|
[e]
|
|
(:code (ex-data e)))
|
|
|
|
(defn ex-of-type?
|
|
[e type]
|
|
(let [data (ex-data e)]
|
|
(= type (:type data))))
|
|
|
|
(defn ex-of-code?
|
|
[e code]
|
|
(let [data (ex-data e)]
|
|
(= code (:code data))))
|
|
|
|
(defn ex-with-code?
|
|
[e code]
|
|
(let [data (ex-data e)]
|
|
(= code (:code data))))
|
|
|
|
(defn success?
|
|
[{:keys [result error]}]
|
|
(nil? error))
|
|
|
|
(defn tempfile
|
|
[source]
|
|
(let [rsc (io/resource source)
|
|
tmp (fs/create-tempfile :dir "/tmp/penpot" :prefix "test-")]
|
|
(io/copy (io/file rsc)
|
|
(io/file tmp))
|
|
tmp))
|
|
|
|
(defn pause
|
|
[]
|
|
(let [^java.io.Console cnsl (System/console)]
|
|
(println "[waiting RETURN]")
|
|
(.readLine cnsl)
|
|
nil))
|
|
|
|
(defn db-exec!
|
|
[sql]
|
|
(db/exec! *pool* sql))
|
|
|
|
(defn db-exec-one!
|
|
[sql]
|
|
(db/exec-one! *pool* sql))
|
|
|
|
(defn db-delete!
|
|
[& params]
|
|
(apply db/delete! *pool* params))
|
|
|
|
(defn db-update!
|
|
[& params]
|
|
(apply db/update! *pool* params))
|
|
|
|
(defn db-insert!
|
|
[& params]
|
|
(apply db/insert! *pool* params))
|
|
|
|
(defn db-delete!
|
|
[& params]
|
|
(apply db/delete! *pool* params))
|
|
|
|
(defn db-query
|
|
[& params]
|
|
(apply db/query *pool* params))
|
|
|
|
(defn db-get
|
|
[& params]
|
|
(apply db/get* *pool* params))
|
|
|
|
(defn run-file-gc!
|
|
"Run the file-gc task the way the scheduler does: props always carry
|
|
the current revn, so the read-time freshness check has a baseline."
|
|
[file-id]
|
|
(run-task! :file-gc {:file-id file-id
|
|
:revn (:revn (db-get :file {:id file-id}))}))
|
|
|
|
(defn sleep
|
|
[ms-or-duration]
|
|
(Thread/sleep (inst-ms (ct/duration ms-or-duration))))
|
|
|
|
(defn config-get-mock
|
|
[data]
|
|
(fn
|
|
([key]
|
|
(get data key (get cf/config key)))
|
|
([key default]
|
|
(get data key (get cf/config key default)))))
|
|
|
|
(defn reset-mock!
|
|
[m]
|
|
(swap! m (fn [m]
|
|
(-> m
|
|
(assoc :called? false)
|
|
(assoc :call-count 0)
|
|
(assoc :return-list [])
|
|
(assoc :call-args nil)
|
|
(assoc :call-args-list [])))))
|
|
|
|
(defn- slurp'
|
|
[input & opts]
|
|
(let [sw (java.io.StringWriter.)]
|
|
(with-open [^java.io.Reader r (java.io.InputStreamReader. input "UTF-8")]
|
|
(io/copy r sw)
|
|
(.toString sw))))
|
|
|
|
(defn parse-sse
|
|
[content]
|
|
(let [state
|
|
(reduce (fn [{:keys [events data event id] :as state} line]
|
|
(cond
|
|
;; empty line → dispatch event if we have data
|
|
(str/blank? line)
|
|
(if (seq data)
|
|
(-> state
|
|
(update :events conj {:event (or event "message")
|
|
:data (-> (str/join "\n" data))})
|
|
(assoc :data [] :event nil))
|
|
state)
|
|
|
|
;; comment line (starts with :)
|
|
(str/starts-with? line ":")
|
|
state
|
|
|
|
:else
|
|
(let [[field raw-value] (str/split line #":" 2)
|
|
value (some-> raw-value (str/replace #"^ " ""))]
|
|
(case field
|
|
"data" (update state :data conj (or value ""))
|
|
"event" (assoc state :event value)
|
|
;; ignore retry and unknown fields
|
|
state))))
|
|
{:events [] :data [] :event nil}
|
|
(str/split content #"\r?\n"))
|
|
|
|
;; handle unterminated last event (no trailing blank line)
|
|
state (if (seq (:data state))
|
|
(update state :events conj
|
|
{:event (or (:event state) "message")
|
|
:data (str/join "\n" (:data state))})
|
|
state)]
|
|
|
|
(:events state)))
|
|
|
|
(defn consume-sse
|
|
[callback]
|
|
(let [{:keys [::yres/status ::yres/body ::yres/headers] :as response} (callback {})
|
|
output (PipedOutputStream.)
|
|
input (PipedInputStream. output)]
|
|
|
|
(try
|
|
(px/exec! :virtual #(rcp/write-body-to-stream body nil output))
|
|
(into []
|
|
(map (fn [{:keys [event data]}]
|
|
(d/vec2 (keyword event)
|
|
(tr/decode-str data))))
|
|
(parse-sse (slurp' input)))
|
|
(finally
|
|
(.close input)))))
|
|
|
|
;; ---- Dummy Request Helpers
|
|
|
|
(defrecord DummyRequest [headers cookies method body-stream
|
|
remote-addr server-name server-port
|
|
scheme protocol path query ssl-client-cert]
|
|
yrq/IRequestCookies
|
|
(get-cookie [_ name]
|
|
{:value (get cookies name)})
|
|
|
|
yrq/IRequest
|
|
(get-header [_ name]
|
|
(get headers name))
|
|
(method [_] method)
|
|
(body [_] body-stream)
|
|
(path [_] path)
|
|
(query [_] query)
|
|
(server-port [_] server-port)
|
|
(server-name [_] server-name)
|
|
(remote-addr [_] remote-addr)
|
|
(ssl-client-cert [_] ssl-client-cert)
|
|
(scheme [_] scheme)
|
|
(protocol [_] protocol))
|
|
|
|
(defn make-dummy-request
|
|
"Constructs a DummyRequest from an options map. Every key is
|
|
optional; missing values fall back to sensible defaults. New
|
|
fields added to DummyRequest won't break existing call sites
|
|
as long as this constructor keeps its `:or` defaults in sync.
|
|
|
|
Recognized keys:
|
|
:headers — map of header name → value
|
|
:cookies — map of cookie name → value
|
|
:method — HTTP method keyword (default :get)
|
|
:body-stream — InputStream for the body (used directly)
|
|
:body-bytes — bytes or string for the body; wrapped in a
|
|
ByteArrayInputStream if :body-stream is not
|
|
given
|
|
:remote-addr — string (default \"127.0.0.1\")
|
|
:server-name — string (default \"test\")
|
|
:server-port — long (default 0)
|
|
:scheme — keyword (default :http)
|
|
:protocol — string (default \"HTTP/1.1\")
|
|
:path — string (default \"/test\")
|
|
:query — string or nil (default nil)
|
|
:ssl-client-cert — X509Certificate or nil (default nil)"
|
|
[{:keys [headers cookies method body-stream body-bytes
|
|
remote-addr server-name server-port scheme protocol
|
|
path query ssl-client-cert]
|
|
:or {headers {} cookies {} method :get
|
|
body-stream nil
|
|
remote-addr "127.0.0.1" server-name "test" server-port 0
|
|
scheme :http protocol "HTTP/1.1" path "/test" query nil
|
|
ssl-client-cert nil}}]
|
|
(let [body-stream (or body-stream
|
|
(when body-bytes
|
|
(java.io.ByteArrayInputStream.
|
|
(if (string? body-bytes)
|
|
(.getBytes ^String body-bytes "UTF-8")
|
|
body-bytes))))]
|
|
(->DummyRequest headers cookies method body-stream
|
|
remote-addr server-name server-port
|
|
scheme protocol path query ssl-client-cert)))
|