runledger_postgres/jobs/types/
events.rs

1use chrono::{DateTime, Utc};
2use runledger_core::jobs::{JobEventType, JobStage};
3use serde_json::Value;
4use sqlx::types::Uuid;
5
6use super::enqueue::JobRequeueStatePolicy;
7
8pub(crate) const BASIC_REQUEUE_KIND: &str = "BASIC";
9pub(crate) const COMPARE_AND_REQUEUE_KIND: &str = "COMPARE_AND_REQUEUE";
10pub(crate) const HANDLER_CONTINUATION_KIND: &str = "HANDLER_CONTINUATION";
11pub(crate) const HANDLER_CONTINUATION_REASON: &str = HANDLER_CONTINUATION_KIND;
12
13#[derive(Clone, Debug)]
14pub struct JobEventRecord {
15    pub id: i64,
16    pub job_id: Uuid,
17    pub run_number: i32,
18    pub attempt: Option<i32>,
19    pub event_type: JobEventType,
20    pub stage: Option<JobStage>,
21    pub progress_done: Option<i64>,
22    pub progress_total: Option<i64>,
23    pub payload: Value,
24    pub occurred_at: DateTime<Utc>,
25}
26
27impl JobEventRecord {
28    /// Decodes payload variants whose JSON schema is owned by
29    /// `runledger-postgres`.
30    ///
31    /// This accessor is deliberately infallible. Unknown event types,
32    /// malformed JSON fields, and future discriminators remain available via
33    /// [`Self::payload`] and decode to a compatibility fallback.
34    #[must_use]
35    pub fn decoded_payload(&self) -> DecodedJobEventPayload<'_> {
36        match self.event_type {
37            JobEventType::Requeued => {
38                DecodedJobEventPayload::Requeued(decode_requeued_event_payload(&self.payload))
39            }
40            JobEventType::Enqueued => decode_successful_replay_enqueued_payload(&self.payload)
41                .map(DecodedJobEventPayload::SuccessfulReplayEnqueued)
42                .unwrap_or(DecodedJobEventPayload::Other),
43            _ => DecodedJobEventPayload::Other,
44        }
45    }
46}
47
48/// Payload shapes authored by `runledger-postgres` that can be decoded without
49/// exposing their JSON representation to consumers.
50///
51/// [`JobEventRecord::payload`] remains available so older, custom, malformed,
52/// and future event payloads can still be inspected. Such payloads decode to an
53/// `Unknown` or `Other` variant rather than failing the event-list query.
54#[derive(Clone, Debug, Eq, PartialEq)]
55#[non_exhaustive]
56pub enum DecodedJobEventPayload<'a> {
57    /// An ordinary, compare-and-requeue, continuation, or unrecognized
58    /// `REQUEUED` payload.
59    Requeued(DecodedRequeuedEventPayload<'a>),
60    /// Provenance attached to the `ENQUEUED` event for a successful-job replay.
61    SuccessfulReplayEnqueued(SuccessfulReplayEnqueuedEventPayload<'a>),
62    /// An event type or payload shape not decoded by this version of the
63    /// persistence driver.
64    Other,
65}
66
67/// Decoded payload for a `REQUEUED` event.
68///
69/// The decoder recognizes both current payloads with a `requeue_kind`
70/// discriminator and historical kindless payloads written before that field
71/// was introduced.
72#[derive(Clone, Debug, Eq, PartialEq)]
73#[non_exhaustive]
74pub enum DecodedRequeuedEventPayload<'a> {
75    /// A release or legacy administrative requeue.
76    #[non_exhaustive]
77    Basic { reason: &'a str },
78    /// An optimistic compare-and-requeue recovery.
79    #[non_exhaustive]
80    CompareAndRequeue {
81        reason: &'a str,
82        state_policy: JobRequeueStatePolicy,
83    },
84    /// A successful handler continuation into a newly pending run.
85    #[non_exhaustive]
86    HandlerContinuation {
87        reason: &'a str,
88        next_run_number: i32,
89        next_run_at: DateTime<Utc>,
90        delay_microseconds: i64,
91    },
92    /// A malformed or future `REQUEUED` payload. The reason is retained when
93    /// it is a JSON string so generic operator surfaces can still display it.
94    #[non_exhaustive]
95    Unknown { reason: Option<&'a str> },
96}
97
98/// Successful-job replay provenance decoded from an `ENQUEUED` event.
99#[derive(Clone, Debug, Eq, PartialEq)]
100#[non_exhaustive]
101pub struct SuccessfulReplayEnqueuedEventPayload<'a> {
102    pub replayed_from_job_id: Uuid,
103    pub replayed_from_run_number: i32,
104    pub replay_request_key: &'a str,
105    pub reason: &'a str,
106}
107
108fn decode_requeued_event_payload(payload: &Value) -> DecodedRequeuedEventPayload<'_> {
109    let reason = payload.get("reason").and_then(Value::as_str);
110    let unknown = || DecodedRequeuedEventPayload::Unknown { reason };
111    let decode_basic = || {
112        reason
113            .map(|reason| DecodedRequeuedEventPayload::Basic { reason })
114            .unwrap_or_else(unknown)
115    };
116    let decode_compare_and_requeue = || {
117        let Some(reason) = reason else {
118            return unknown();
119        };
120        let Some(state_policy) = payload
121            .get("state_policy")
122            .and_then(Value::as_str)
123            .and_then(JobRequeueStatePolicy::from_event_value)
124        else {
125            return unknown();
126        };
127        DecodedRequeuedEventPayload::CompareAndRequeue {
128            reason,
129            state_policy,
130        }
131    };
132    let decode_handler_continuation = || {
133        let Some(reason) = reason else {
134            return unknown();
135        };
136        let Some(next_run_number) = payload
137            .get("next_run_number")
138            .and_then(Value::as_i64)
139            .and_then(|value| i32::try_from(value).ok())
140        else {
141            return unknown();
142        };
143        let Some(next_run_at) = payload
144            .get("next_run_at")
145            .and_then(Value::as_str)
146            .and_then(|value| DateTime::parse_from_rfc3339(value).ok())
147            .map(|value| value.with_timezone(&Utc))
148        else {
149            return unknown();
150        };
151        let Some(delay_microseconds) = payload.get("delay_microseconds").and_then(Value::as_i64)
152        else {
153            return unknown();
154        };
155        DecodedRequeuedEventPayload::HandlerContinuation {
156            reason,
157            next_run_number,
158            next_run_at,
159            delay_microseconds,
160        }
161    };
162
163    let handler_schedule_keys = ["next_run_number", "next_run_at", "delay_microseconds"];
164    let has_complete_handler_schedule = handler_schedule_keys
165        .iter()
166        .all(|key| payload.get(*key).is_some());
167    let has_any_handler_schedule_field = handler_schedule_keys
168        .iter()
169        .any(|key| payload.get(*key).is_some());
170
171    match payload.get("requeue_kind") {
172        Some(Value::String(kind)) => match kind.as_str() {
173            BASIC_REQUEUE_KIND => decode_basic(),
174            COMPARE_AND_REQUEUE_KIND => decode_compare_and_requeue(),
175            HANDLER_CONTINUATION_KIND => decode_handler_continuation(),
176            _ => unknown(),
177        },
178        Some(_) => unknown(),
179        None if reason == Some(HANDLER_CONTINUATION_REASON) && has_complete_handler_schedule => {
180            decode_handler_continuation()
181        }
182        None if payload.get("state_policy").is_some() => decode_compare_and_requeue(),
183        None if has_any_handler_schedule_field => unknown(),
184        None => decode_basic(),
185    }
186}
187
188fn decode_successful_replay_enqueued_payload(
189    payload: &Value,
190) -> Option<SuccessfulReplayEnqueuedEventPayload<'_>> {
191    let replayed_from_job_id = payload
192        .get("replayed_from_job_id")?
193        .as_str()?
194        .parse()
195        .ok()?;
196    let replayed_from_run_number =
197        i32::try_from(payload.get("replayed_from_run_number")?.as_i64()?).ok()?;
198    let replay_request_key = payload.get("replay_request_key")?.as_str()?;
199    let reason = payload.get("reason")?.as_str()?;
200
201    Some(SuccessfulReplayEnqueuedEventPayload {
202        replayed_from_job_id,
203        replayed_from_run_number,
204        replay_request_key,
205        reason,
206    })
207}
208
209#[cfg(test)]
210mod decoded_event_payload_tests {
211    use serde_json::json;
212
213    use super::*;
214
215    fn event_record(event_type: JobEventType, payload: Value) -> JobEventRecord {
216        JobEventRecord {
217            id: 1,
218            job_id: Uuid::nil(),
219            run_number: 1,
220            attempt: None,
221            event_type,
222            stage: None,
223            progress_done: None,
224            progress_total: None,
225            payload,
226            occurred_at: Utc::now(),
227        }
228    }
229
230    #[test]
231    fn kindless_requeue_payloads_preserve_legacy_decoding() {
232        let basic = event_record(JobEventType::Requeued, json!({"reason": "released"}));
233        assert!(matches!(
234            basic.decoded_payload(),
235            DecodedJobEventPayload::Requeued(DecodedRequeuedEventPayload::Basic {
236                reason: "released",
237                ..
238            })
239        ));
240
241        let compare = event_record(
242            JobEventType::Requeued,
243            json!({
244                "reason": "operator recovery",
245                "state_policy": "reset_progress_and_checkpoint"
246            }),
247        );
248        assert!(matches!(
249            compare.decoded_payload(),
250            DecodedJobEventPayload::Requeued(DecodedRequeuedEventPayload::CompareAndRequeue {
251                reason: "operator recovery",
252                state_policy: JobRequeueStatePolicy::ResetProgressAndCheckpoint,
253                ..
254            })
255        ));
256
257        let continuation = event_record(
258            JobEventType::Requeued,
259            json!({
260                "reason": "HANDLER_CONTINUATION",
261                "next_run_number": 2,
262                "next_run_at": "2026-07-19T12:34:56.123456Z",
263                "delay_microseconds": 250_000
264            }),
265        );
266        assert!(matches!(
267            continuation.decoded_payload(),
268            DecodedJobEventPayload::Requeued(DecodedRequeuedEventPayload::HandlerContinuation {
269                reason: "HANDLER_CONTINUATION",
270                next_run_number: 2,
271                delay_microseconds: 250_000,
272                ..
273            })
274        ));
275    }
276
277    #[test]
278    fn present_unknown_or_non_string_discriminators_are_not_legacy_payloads() {
279        for requeue_kind in [json!("FUTURE_REQUEUE_KIND"), json!(42), json!(null)] {
280            let payload = json!({
281                "requeue_kind": requeue_kind,
282                "reason": "future recovery"
283            });
284            let event = event_record(JobEventType::Requeued, payload.clone());
285
286            assert!(matches!(
287                event.decoded_payload(),
288                DecodedJobEventPayload::Requeued(DecodedRequeuedEventPayload::Unknown {
289                    reason: Some("future recovery"),
290                    ..
291                })
292            ));
293            assert_eq!(event.payload, payload);
294        }
295    }
296
297    #[test]
298    fn malformed_known_payloads_fall_back_without_losing_the_raw_payload() {
299        let payload = json!({
300            "requeue_kind": "HANDLER_CONTINUATION",
301            "reason": "HANDLER_CONTINUATION",
302            "next_run_number": "2",
303            "next_run_at": "not-a-timestamp",
304            "delay_microseconds": 250_000
305        });
306        let event = event_record(JobEventType::Requeued, payload.clone());
307
308        assert!(matches!(
309            event.decoded_payload(),
310            DecodedJobEventPayload::Requeued(DecodedRequeuedEventPayload::Unknown {
311                reason: Some("HANDLER_CONTINUATION"),
312                ..
313            })
314        ));
315        assert_eq!(event.payload, payload);
316    }
317}
runledger_postgres/jobs/types/events.rs

runledger_postgres/jobs/types/
events.rs
