mirror of
https://github.com/penpot/penpot.git
synced 2026-11-11 04:20:42 +00:00
* 🎉 Move backend jobs to the unified job table substrate Replace the generic 'task' table with a unified 'job' table as the durable substrate for all backend jobs, and migrate every producer and consumer to it. The legacy 'task' table stops receiving writes but stays dormant (still cleaned by tasks-gc); upload-session chunking and the other post-branch upstream changes are preserved. New 0154 migration creates the 'job' table: dispatch and lifecycle columns (name, queue, priority, scheduled_at, retries, status, timestamps, props) plus optional user-facing ledger columns (profile_id, target, progress, error, result, resource_id, expires_at). No modified_at trigger: the application updates it on every write for lease and orphan detection. Partial indexes cover the dispatcher claim query, orphan scanning, and per-profile listing. New app.jobs public API: submit!/invoke!, heartbeats and progress reporting, a precompiled job-def registry (name, schema, handler, decoder, validator), an ephemeral request/response mode, and a management API for external workers. The dispatcher claims new/retry rows (FOR UPDATE SKIP LOCKED) with a plain JSON redis payload and marks lease-expired running rows as orphaned; the runner executes per-queue with the same retry conventions. New jobs GC (daily): deletes expired rows and retains internal terminal rows per :jobs-retention, touching referenced storage objects so storage-gc-touched reclaims them. Job resources live in a dedicated job-resource bucket with a reclaim branch in storage-gc-touched. All production call-sites (sendmail, delete-object, demo-purge, file-gc, offload, webhooks, quotes, nitrate bulk delete, snapshots, telemetry) and the ten cron tasks now submit through app.jobs; cron is a pure scheduler and the legacy worker registry, task-table inserts, and per-namespace handlers are removed. Includes migrations, dispatcher, runner, jobs, request/response, jobs GC, management API, and cron test suites. AI-assisted-by: glm-5.3-flash AI-assisted-by: muse-spark-1.3-contributor AI-assisted-by: kimi-k3 AI-assisted-by: mimo-v2.5 * ✨ Address review comments on jobs substrate Resolves the four open review threads on the jobs PR. Orphaned jobs are now marked aborted, a system-side terminal state with no retry: the dispatcher sweep stores the end event with outcome aborted in a single bulk insert in the same transaction, records the terminal outcome next to the orphaned counter, and reports an error log; orphans keep alerting through the log and the counter, with no user notification. The aborted state is terminal everywhere: the status CHECK, the sweep, the lifecycle writers, the jobs GC retention, the backlog gauges and the user status mapping, with tests covering each transition. The claim comment, the 2.20 doc version and the legacy task table notes are corrected, and objects-gc heartbeats per chunk. AI-assisted-by: muse-spark-1.3-contributor
394 lines
14 KiB
Clojure
394 lines
14 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 SUBSIDIARY SL
|
|
|
|
(ns app.http.session
|
|
(:refer-clojure :exclude [read])
|
|
(:require
|
|
[app.common.data :as d]
|
|
[app.common.logging :as l]
|
|
[app.common.schema :as sm]
|
|
[app.common.time :as ct]
|
|
[app.common.uuid :as uuid]
|
|
[app.config :as cf]
|
|
[app.db :as db]
|
|
[app.db.sql :as sql]
|
|
[app.http :as-alias http]
|
|
[app.http.auth :as-alias http.auth]
|
|
[app.http.session.tasks :as-alias tasks]
|
|
[app.jobs :as jobs]
|
|
[app.main :as-alias main]
|
|
[app.setup :as-alias setup]
|
|
[app.setup.clock :as clock]
|
|
[app.tokens :as tokens]
|
|
[integrant.core :as ig]
|
|
[yetti.request :as yreq]
|
|
[yetti.response :as yres]))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; DEFAULTS
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
;; Default value for cookie max-age
|
|
(def default-cookie-max-age (ct/duration {:days 7}))
|
|
|
|
;; Default age for automatic session renewal
|
|
(def default-renewal-max-age (ct/duration {:hours 6}))
|
|
|
|
;; Default absolute maximum session duration
|
|
(def default-cookie-max-age-absolute (ct/duration {:days 30}))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; PROTOCOLS
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(defprotocol ISessionManager
|
|
(read-session [_ id])
|
|
(create-session [_ params])
|
|
(update-session [_ session])
|
|
(delete-session [_ id]))
|
|
|
|
(defn manager?
|
|
[o]
|
|
(satisfies? ISessionManager o))
|
|
|
|
(sm/register!
|
|
{:type ::manager
|
|
:pred manager?})
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; STORAGE IMPL
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(def ^:private schema:params
|
|
[:map {:title "SessionParams" :closed true}
|
|
[:profile-id ::sm/uuid]
|
|
[:user-agent {:optional true} ::sm/text]
|
|
[:sso-provider-id {:optional true} ::sm/uuid]
|
|
[:sso-session-id {:optional true} :string]])
|
|
|
|
(def ^:private valid-params?
|
|
(sm/validator schema:params))
|
|
|
|
(defn- decode-session
|
|
[session]
|
|
(cond-> session
|
|
(db/pgobject? (:props session))
|
|
(update :props db/decode-transit-pgobject)))
|
|
|
|
(defn- database-manager
|
|
[pool]
|
|
(reify ISessionManager
|
|
(read-session [_ id]
|
|
(some-> (db/exec-one! pool (sql/select :http-session-v2 {:id id}))
|
|
(decode-session)))
|
|
|
|
(create-session [_ params]
|
|
(assert (valid-params? params) "expect valid session params")
|
|
|
|
(let [now (ct/now)
|
|
params (-> params
|
|
(assoc :id (uuid/next))
|
|
(assoc :created-at now)
|
|
(assoc :modified-at now))]
|
|
(db/insert! pool :http-session-v2 params
|
|
{::db/return-keys true})))
|
|
|
|
(update-session [_ session]
|
|
(let [modified-at (ct/now)]
|
|
(db/update! pool :http-session-v2
|
|
(cond-> {:modified-at modified-at}
|
|
(some? (:props session))
|
|
(assoc :props (db/tjson (:props session))))
|
|
{:id (:id session)}
|
|
{::db/return-keys true})))
|
|
|
|
(delete-session [_ id]
|
|
(db/delete! pool :http-session-v2 {:id id} {::db/return-keys false})
|
|
nil)))
|
|
|
|
(defn inmemory-manager
|
|
[]
|
|
(let [cache (atom {})]
|
|
(reify ISessionManager
|
|
(read-session [_ id]
|
|
(get @cache id))
|
|
|
|
(create-session [_ params]
|
|
(assert (valid-params? params) "expect valid session params")
|
|
|
|
(let [now (ct/now)
|
|
session (-> params
|
|
(assoc :id (uuid/next))
|
|
(assoc :created-at now)
|
|
(assoc :modified-at now))]
|
|
(swap! cache assoc (:id session) session)
|
|
session))
|
|
|
|
(update-session [_ session]
|
|
(let [modified-at (ct/now)
|
|
session (assoc session :modified-at modified-at)]
|
|
(swap! cache assoc (:id session) session)
|
|
session))
|
|
|
|
(delete-session [_ id]
|
|
(swap! cache dissoc id)
|
|
nil))))
|
|
|
|
(defmethod ig/assert-key ::manager
|
|
[_ params]
|
|
(assert (db/pool? (::db/pool params)) "expect valid database pool"))
|
|
|
|
(defmethod ig/init-key ::manager
|
|
[_ {:keys [::db/pool]}]
|
|
(if (db/read-only? pool)
|
|
(inmemory-manager)
|
|
(database-manager pool)))
|
|
|
|
(defmethod ig/halt-key! ::manager
|
|
[_ _])
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; MANAGER IMPL
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(declare ^:private assign-session-cookie)
|
|
(declare ^:private clear-session-cookie)
|
|
|
|
(defn- assign-token
|
|
[cfg session]
|
|
(let [absolute-max-age (cf/get :auth-token-cookie-max-age-absolute default-cookie-max-age-absolute)
|
|
claims {:iss "authentication"
|
|
:aud "penpot"
|
|
:sid (:id session)
|
|
:iat (:modified-at session)
|
|
:uid (:profile-id session)
|
|
:sso-provider-id (:sso-provider-id session)
|
|
:sso-session-id (:sso-session-id session)}
|
|
claims (if (:created-at session)
|
|
(assoc claims :exp (ct/plus (:created-at session) absolute-max-age))
|
|
claims)
|
|
header {:kid 1 :ver 1}
|
|
token (tokens/generate cfg claims header)]
|
|
(assoc session :token token)))
|
|
|
|
(defn create-fn
|
|
[{:keys [::manager] :as cfg} {profile-id :id :as profile}
|
|
& {:keys [sso-provider-id sso-session-id]}]
|
|
|
|
(assert (manager? manager) "expected valid session manager")
|
|
(assert (uuid? profile-id) "expected valid uuid for profile-id")
|
|
|
|
(fn [request response]
|
|
(let [uagent (yreq/get-header request "user-agent")
|
|
session (->> {:user-agent uagent
|
|
:profile-id profile-id
|
|
:sso-provider-id sso-provider-id
|
|
:sso-session-id sso-session-id}
|
|
(d/without-nils)
|
|
(create-session manager)
|
|
(assign-token cfg))]
|
|
|
|
(l/trc :hint "create" :id (str (:id session)) :profile-id (str profile-id))
|
|
(assign-session-cookie response session))))
|
|
|
|
(defn delete-fn
|
|
[{:keys [::manager]}]
|
|
(assert (manager? manager) "expected valid session manager")
|
|
(fn [request response]
|
|
(some->> (get request ::session) :id (delete-session manager))
|
|
(clear-session-cookie response)))
|
|
|
|
(defn decode-token
|
|
[cfg token]
|
|
(try
|
|
(tokens/verify cfg {:token token :iss "authentication"})
|
|
(catch Throwable cause
|
|
(l/trc :hint "exception on decoding token"
|
|
:token token
|
|
:cause cause))))
|
|
|
|
(defn get-session
|
|
[request]
|
|
(get request ::session))
|
|
|
|
(defn invalidate-others
|
|
[cfg session]
|
|
(let [sql "delete from http_session_v2 where profile_id = ? and id != ?"]
|
|
(-> (db/exec-one! cfg [sql (:profile-id session) (:id session)])
|
|
(db/get-update-count))))
|
|
|
|
(defn invalidate-all
|
|
"Delete all sessions for a given profile. Used when a profile is deleted
|
|
to ensure immediate access revocation across all devices."
|
|
[cfg profile-id]
|
|
(let [sql "delete from http_session_v2 where profile_id = ?"]
|
|
(-> (db/exec-one! cfg [sql profile-id])
|
|
(db/get-update-count))))
|
|
|
|
(def ^:private sql:clear-organization-sso-sessions
|
|
(str "UPDATE http_session_v2 "
|
|
"SET props = props #- ARRAY['~:sso', ?]::text[] "
|
|
"WHERE props IS NOT NULL "
|
|
"AND jsonb_exists(props -> '~:sso', ?)"))
|
|
|
|
(defn clear-organization-sso-sessions!
|
|
"Remove the SSO entry for organization-id from the props of every
|
|
session that currently holds it. The key is transit-encoded as the
|
|
string '~u<uuid>' under the '~:sso' path."
|
|
[pool organization-id]
|
|
(let [organization-key (str "~u" organization-id)]
|
|
(db/exec! pool [sql:clear-organization-sso-sessions organization-key organization-key])))
|
|
|
|
(defn- renew-session?
|
|
[{:keys [modified-at]}]
|
|
(and (ct/inst? modified-at)
|
|
(let [elapsed (ct/diff modified-at (ct/now))]
|
|
(neg? (compare default-renewal-max-age elapsed)))))
|
|
|
|
(defn- session-expired?
|
|
[{:keys [modified-at]}]
|
|
(and (ct/inst? modified-at)
|
|
(let [max-age (cf/get :auth-token-cookie-max-age default-cookie-max-age)]
|
|
(pos? (compare (ct/now) (ct/plus modified-at max-age))))))
|
|
|
|
(defn- wrap-authz
|
|
[handler {:keys [::manager] :as cfg}]
|
|
(assert (manager? manager) "expected valid session manager")
|
|
(fn [request]
|
|
(let [{:keys [type claims]} (get request ::http/auth-data)]
|
|
(cond
|
|
(= type :cookie)
|
|
(let [session (some->> (:sid claims) (read-session manager))
|
|
;; Idle-expired sessions authenticate like missing ones:
|
|
;; no profile, no renewal. The GC deletes the row later.
|
|
session (when-not (session-expired? session) session)
|
|
|
|
request
|
|
(cond-> request
|
|
(some? session)
|
|
(-> (assoc ::profile-id (:profile-id session))
|
|
(assoc ::session session)))
|
|
|
|
response
|
|
(binding [ct/*clock* (clock/get-clock (:profile-id session))]
|
|
(handler request))]
|
|
|
|
(if (and session (renew-session? session))
|
|
(let [session (->> session
|
|
(update-session manager)
|
|
(assign-token cfg))]
|
|
(assign-session-cookie response session))
|
|
response))
|
|
|
|
(= type :bearer)
|
|
(let [session (some->> (:sid claims) (read-session manager))
|
|
session (when-not (session-expired? session) session)
|
|
request (cond-> request
|
|
(some? session)
|
|
(-> (assoc ::profile-id (:profile-id session))
|
|
(assoc ::session session)))]
|
|
(handler request))
|
|
|
|
:else
|
|
(handler request)))))
|
|
|
|
(def authz
|
|
{:name ::authz
|
|
:compile (constantly wrap-authz)})
|
|
|
|
;; --- IMPL
|
|
|
|
(defn- assign-session-cookie
|
|
[response {token :token modified-at :modified-at}]
|
|
(let [max-age (cf/get :auth-token-cookie-max-age default-cookie-max-age)
|
|
renewal-at modified-at
|
|
renewal (ct/plus renewal-at default-renewal-max-age)
|
|
expires (ct/plus renewal-at max-age)
|
|
secure? (contains? cf/flags :secure-session-cookies)
|
|
strict? (contains? cf/flags :strict-session-cookies)
|
|
cors? (contains? cf/flags :cors)
|
|
name (cf/get :auth-token-cookie-name)
|
|
comment (str "Renewal at: " (ct/format-inst renewal :rfc1123))
|
|
cookie {:path "/"
|
|
:http-only true
|
|
:expires expires
|
|
:value token
|
|
:comment comment
|
|
:same-site (if cors? :none (if strict? :strict :lax))
|
|
:secure secure?}]
|
|
(update response ::yres/cookies assoc name cookie)))
|
|
|
|
(defn- clear-session-cookie
|
|
[response]
|
|
(let [cname (cf/get :auth-token-cookie-name)]
|
|
(update response ::yres/cookies assoc cname {:path "/" :value "" :max-age 0})))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; TASK: SESSION GC
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(defmethod ig/assert-key ::job-def
|
|
[_ params]
|
|
(assert (db/pool? (::db/pool params)) "expected valid database pool")
|
|
(assert (ct/duration? (::tasks/max-age params)))
|
|
(assert (ct/duration? (::tasks/max-age-absolute params)))
|
|
;; NOTE: explicit throw (not `assert`) so the guard fires even when
|
|
;; the `:backend-asserts` flag leaves `*assert*` false.
|
|
(when (pos? (compare (::tasks/max-age params)
|
|
(::tasks/max-age-absolute params)))
|
|
(throw (IllegalArgumentException.
|
|
"absolute session max-age must be greater than or equal to idle max-age"))))
|
|
|
|
(defmethod ig/expand-key ::job-def
|
|
[k v]
|
|
(let [max-age (cf/get :auth-token-cookie-max-age default-cookie-max-age)
|
|
max-age-absolute (cf/get :auth-token-cookie-max-age-absolute
|
|
default-cookie-max-age-absolute)]
|
|
{k (merge {::tasks/max-age max-age
|
|
::tasks/max-age-absolute max-age-absolute}
|
|
(d/without-nils v))}))
|
|
|
|
(def ^:private
|
|
sql:delete-expired-v2
|
|
"DELETE FROM http_session_v2
|
|
WHERE modified_at < ?::timestamptz
|
|
OR created_at < ?::timestamptz")
|
|
|
|
(defn- collect-expired-tasks
|
|
[{:keys [::db/conn ::tasks/max-age ::tasks/max-age-absolute]}]
|
|
(let [idle-threshold (ct/minus (ct/now) max-age)
|
|
abs-threshold (ct/minus (ct/now) max-age-absolute)
|
|
result (-> (db/exec-one! conn [sql:delete-expired-v2
|
|
idle-threshold abs-threshold])
|
|
(db/get-update-count))]
|
|
(l/dbg :task "gc"
|
|
:hint "clean http sessions"
|
|
:deleted result)
|
|
result))
|
|
|
|
(declare execute-session-gc)
|
|
|
|
(def schema:session-gc-params
|
|
"Params map (no params needed; config-derived only)."
|
|
[:map {:closed true}])
|
|
|
|
(defmethod ig/init-key ::job-def
|
|
[_ cfg]
|
|
{::jobs/name :session-gc
|
|
::jobs/schema schema:session-gc-params
|
|
::jobs/handler
|
|
(fn [_context params]
|
|
(execute-session-gc cfg params))
|
|
::jobs/decoder (sm/decoder schema:session-gc-params sm/json-transformer)
|
|
::jobs/validator (sm/validator schema:session-gc-params)})
|
|
|
|
(defn execute-session-gc
|
|
"Plain job handler: delete expired http sessions."
|
|
([cfg] (execute-session-gc cfg nil))
|
|
([cfg _params]
|
|
(jobs/heartbeat cfg)
|
|
(db/tx-run! cfg collect-expired-tasks)))
|