azums_core/
model.rs

1use chrono::{DateTime, Utc};
2use serde::{Deserialize, Serialize};
3use serde_json::Value;
4use uuid::Uuid;
5
6/// Per-queue job execution ordering policy.
7#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Default)]
8pub enum QueueOrdering {
9    /// Process jobs in exact First-In, First-Out order by creation time (`created_at ASC`).
10    #[default]
11    Fifo,
12    /// Process jobs as fast as possible without strict creation order guarantees.
13    Fastest,
14}
15
16/// Configuration options for a job queue.
17#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
18pub struct QueueConfig {
19    pub ordering: QueueOrdering,
20}
21
22impl Default for QueueConfig {
23    fn default() -> Self {
24        Self {
25            ordering: QueueOrdering::Fifo,
26        }
27    }
28}
29
30impl QueueConfig {
31    pub fn new(ordering: QueueOrdering) -> Self {
32        Self { ordering }
33    }
34}
35
36/// Named queue definition plus its execution policy.
37#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
38pub struct Queue {
39    pub name: String,
40    pub config: QueueConfig,
41}
42
43impl Queue {
44    pub fn new(name: impl Into<String>, config: QueueConfig) -> Self {
45        Self {
46            name: name.into(),
47            config,
48        }
49    }
50}
51
52/// Worker identity used for leases, attempts, and execution ownership.
53#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
54pub struct Worker {
55    pub id: String,
56}
57
58impl Worker {
59    pub fn new(id: impl Into<String>) -> Self {
60        Self { id: id.into() }
61    }
62}
63
64/// Ordering strength exposed by a storage backend.
65#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
66pub enum OrderingCapability {
67    /// No meaningful ordering contract beyond at-least-once execution.
68    None,
69    /// Runnable jobs are leased in priority/schedule/FIFO order where the backend supports it.
70    FifoLeasing,
71    /// Backend supports both FIFO leasing and fastest-throughput leasing modes.
72    FifoAndFastestLeasing,
73}
74
75/// Backpressure behavior exposed by a storage backend.
76#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
77pub enum BackpressureCapability {
78    /// The backend accepts committed jobs and represents overload as queued backlog.
79    BacklogOnly,
80    /// The backend can throttle worker leasing through queue policies without dropping jobs.
81    ExecutionRateLimit,
82}
83
84/// Persistence strength provided by a backend.
85#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
86pub enum DurabilityCapability {
87    /// State is lost when the current process exits.
88    ProcessLocal,
89    /// State survives process restart when the backend is used in its persistent mode.
90    Persistent,
91    /// Durability depends on backend persistence, eviction, and deployment configuration.
92    ConfigurationDependent,
93}
94
95/// Transaction boundary in which enqueue can be atomic with application state.
96#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
97pub enum TransactionalEnqueueCapability {
98    /// Enqueue is atomic only as its own backend operation.
99    BackendOperationOnly,
100    /// Enqueue can use the caller's transaction in the same SQL database.
101    SameDatabase,
102}
103
104/// Delivery behavior of backend wake-up notifications.
105#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
106pub enum NotificationCapability {
107    /// Process-local best-effort hint; durable state must still be polled or leased.
108    ProcessLocalHint,
109    /// Best-effort backend notification; durable state remains the source of truth.
110    BestEffortHint,
111    /// Best-effort notification combined with a polling fallback.
112    BestEffortHintWithPolling,
113}
114
115/// Retention behavior exposed by a backend.
116#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
117pub enum RetentionCapability {
118    /// Retained only for the lifetime of the current process.
119    ProcessLifetime,
120    /// Retained until an explicit Azums maintenance or pruning operation removes it.
121    ExplicitPruning,
122    /// Retention also depends on backend eviction and persistence configuration.
123    BackendConfigured,
124}
125
126/// Coordination provided for consumers sharing one consumer-group name.
127#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
128pub enum ConsumerGroupCapability {
129    /// The backend persists a monotonic group offset but does not assign work to members.
130    OffsetsOnly,
131}
132
133/// Detailed semantic strength behind the compatibility-preserving feature flags.
134#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
135pub struct BackendSemanticCapabilities {
136    pub durability: DurabilityCapability,
137    pub transactional_enqueue_scope: TransactionalEnqueueCapability,
138    pub notification_delivery: NotificationCapability,
139    pub job_retention: RetentionCapability,
140    pub stream_retention: RetentionCapability,
141    pub consumer_group_coordination: ConsumerGroupCapability,
142}
143
144/// Storage backend feature and guarantee declaration.
145///
146/// Capabilities describe what a backend can honestly provide. They are not a marketing matrix:
147/// application code can inspect this value when it needs a specific storage guarantee.
148#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
149pub struct BackendCapabilities {
150    pub transactional_enqueue: bool,
151    pub durable_jobs: bool,
152    pub notifications: bool,
153    pub streams: bool,
154    pub consumer_groups: bool,
155    pub distributed_workers: bool,
156    pub ordering: OrderingCapability,
157    pub backpressure: BackpressureCapability,
158}
159
160impl BackendCapabilities {
161    pub const fn memory() -> Self {
162        Self {
163            transactional_enqueue: false,
164            durable_jobs: false,
165            notifications: true,
166            streams: true,
167            consumer_groups: true,
168            distributed_workers: false,
169            ordering: OrderingCapability::FifoAndFastestLeasing,
170            backpressure: BackpressureCapability::BacklogOnly,
171        }
172    }
173
174    pub const fn sqlite() -> Self {
175        Self {
176            transactional_enqueue: true,
177            durable_jobs: true,
178            notifications: true,
179            streams: true,
180            consumer_groups: true,
181            distributed_workers: false,
182            ordering: OrderingCapability::FifoAndFastestLeasing,
183            backpressure: BackpressureCapability::BacklogOnly,
184        }
185    }
186
187    pub const fn postgres() -> Self {
188        Self {
189            transactional_enqueue: true,
190            durable_jobs: true,
191            notifications: true,
192            streams: true,
193            consumer_groups: true,
194            distributed_workers: true,
195            ordering: OrderingCapability::FifoAndFastestLeasing,
196            backpressure: BackpressureCapability::ExecutionRateLimit,
197        }
198    }
199
200    pub const fn redis() -> Self {
201        Self {
202            transactional_enqueue: false,
203            durable_jobs: true,
204            notifications: true,
205            streams: true,
206            consumer_groups: true,
207            distributed_workers: true,
208            ordering: OrderingCapability::FifoLeasing,
209            backpressure: BackpressureCapability::BacklogOnly,
210        }
211    }
212
213    pub fn supports_portable_job_api(&self) -> bool {
214        self.durable_jobs || !self.distributed_workers
215    }
216
217    /// Returns the detailed profile for an exact built-in capability declaration.
218    ///
219    /// Unknown custom combinations return `None` instead of being guessed as a built-in backend.
220    pub const fn semantics(&self) -> Option<BackendSemanticCapabilities> {
221        match (
222            self.transactional_enqueue,
223            self.durable_jobs,
224            self.notifications,
225            self.streams,
226            self.consumer_groups,
227            self.distributed_workers,
228            self.ordering,
229            self.backpressure,
230        ) {
231            (
232                false,
233                false,
234                true,
235                true,
236                true,
237                false,
238                OrderingCapability::FifoAndFastestLeasing,
239                BackpressureCapability::BacklogOnly,
240            ) => Some(BackendSemanticCapabilities::memory()),
241            (
242                true,
243                true,
244                true,
245                true,
246                true,
247                false,
248                OrderingCapability::FifoAndFastestLeasing,
249                BackpressureCapability::BacklogOnly,
250            ) => Some(BackendSemanticCapabilities::sqlite()),
251            (
252                true,
253                true,
254                true,
255                true,
256                true,
257                true,
258                OrderingCapability::FifoAndFastestLeasing,
259                BackpressureCapability::ExecutionRateLimit,
260            ) => Some(BackendSemanticCapabilities::postgres()),
261            (
262                false,
263                true,
264                true,
265                true,
266                true,
267                true,
268                OrderingCapability::FifoLeasing,
269                BackpressureCapability::BacklogOnly,
270            ) => Some(BackendSemanticCapabilities::redis()),
271            _ => None,
272        }
273    }
274}
275
276impl BackendSemanticCapabilities {
277    pub const fn memory() -> Self {
278        Self {
279            durability: DurabilityCapability::ProcessLocal,
280            transactional_enqueue_scope: TransactionalEnqueueCapability::BackendOperationOnly,
281            notification_delivery: NotificationCapability::ProcessLocalHint,
282            job_retention: RetentionCapability::ProcessLifetime,
283            stream_retention: RetentionCapability::ProcessLifetime,
284            consumer_group_coordination: ConsumerGroupCapability::OffsetsOnly,
285        }
286    }
287
288    pub const fn sqlite() -> Self {
289        Self {
290            durability: DurabilityCapability::Persistent,
291            transactional_enqueue_scope: TransactionalEnqueueCapability::SameDatabase,
292            notification_delivery: NotificationCapability::BestEffortHintWithPolling,
293            job_retention: RetentionCapability::ExplicitPruning,
294            stream_retention: RetentionCapability::ExplicitPruning,
295            consumer_group_coordination: ConsumerGroupCapability::OffsetsOnly,
296        }
297    }
298
299    pub const fn postgres() -> Self {
300        Self {
301            durability: DurabilityCapability::Persistent,
302            transactional_enqueue_scope: TransactionalEnqueueCapability::SameDatabase,
303            notification_delivery: NotificationCapability::BestEffortHint,
304            job_retention: RetentionCapability::ExplicitPruning,
305            stream_retention: RetentionCapability::ExplicitPruning,
306            consumer_group_coordination: ConsumerGroupCapability::OffsetsOnly,
307        }
308    }
309
310    pub const fn redis() -> Self {
311        Self {
312            durability: DurabilityCapability::ConfigurationDependent,
313            transactional_enqueue_scope: TransactionalEnqueueCapability::BackendOperationOnly,
314            notification_delivery: NotificationCapability::BestEffortHintWithPolling,
315            job_retention: RetentionCapability::BackendConfigured,
316            stream_retention: RetentionCapability::BackendConfigured,
317            consumer_group_coordination: ConsumerGroupCapability::OffsetsOnly,
318        }
319    }
320}
321
322/// Lightweight job summary model returned when listing jobs in Admin UI or APIs.
323#[derive(Debug, Clone, Serialize, Deserialize)]
324#[cfg_attr(feature = "sqlx", derive(sqlx::FromRow))]
325pub struct JobListItem {
326    pub id: Uuid,
327    pub idempotency_key: Option<String>,
328    pub queue: String,
329    pub job_type: String,
330    pub status: String,
331
332    pub run_at: DateTime<Utc>,
333    #[serde(default)]
334    pub deadline_at: Option<DateTime<Utc>>,
335    #[serde(default)]
336    pub timeout_seconds: Option<i64>,
337    #[serde(default)]
338    pub recurring_interval_seconds: Option<i64>,
339    pub priority: i32,
340    pub max_attempts: i32,
341
342    pub last_error_code: Option<String>,
343    pub last_error_message: Option<String>,
344
345    pub dlq_reason_code: Option<String>,
346
347    pub created_at: DateTime<Utc>,
348    pub updated_at: DateTime<Utc>,
349}
350
351/// Primary job entity representing a unit of work stored in a storage backend.
352///
353/// # Examples
354///
355/// ```rust
356/// use azums_core::Job;
357///
358/// let job = Job::new("email_send", serde_json::json!({"to": "user@example.com"}))
359///     .queue("emails")
360///     .priority(10)
361///     .max_attempts(5);
362///
363/// assert_eq!(job.queue, "emails");
364/// assert_eq!(job.priority, 10);
365/// assert_eq!(job.max_attempts, 5);
366/// assert_eq!(job.payload["to"], "user@example.com");
367/// ```
368#[derive(Debug, Clone, Serialize, Deserialize)]
369#[cfg_attr(feature = "sqlx", derive(sqlx::FromRow))]
370pub struct Job {
371    pub dataset_id: String,
372    pub replay_of_job_id: Option<Uuid>,
373    pub idempotency_key: Option<String>,
374
375    pub id: Uuid,
376    pub queue: String,
377    pub job_type: String,
378    #[cfg_attr(feature = "sqlx", sqlx(rename = "payload_json"))]
379    pub payload: Value,
380    pub run_at: DateTime<Utc>,
381    #[serde(default)]
382    pub deadline_at: Option<DateTime<Utc>>,
383    #[serde(default)]
384    pub timeout_seconds: Option<i64>,
385    #[serde(default)]
386    pub recurring_interval_seconds: Option<i64>,
387    pub status: String,
388    pub priority: i32,
389    pub max_attempts: i32,
390
391    pub locked_at: Option<DateTime<Utc>>,
392    pub locked_by: Option<String>,
393    pub lock_expires_at: Option<DateTime<Utc>>,
394
395    pub dlq_reason_code: Option<String>,
396    pub dlq_at: Option<DateTime<Utc>>,
397
398    pub created_at: DateTime<Utc>,
399    pub updated_at: DateTime<Utc>,
400}
401
402impl Job {
403    /// Creates a new `Job` with default queue `"default"`, priority `0`, and max attempts `25`.
404    ///
405    /// # Examples
406    ///
407    /// ```rust
408    /// use azums_core::Job;
409    ///
410    /// let job = Job::new("greet", serde_json::json!({"name": "World"}));
411    /// assert_eq!(job.job_type, "greet");
412    /// assert_eq!(job.payload["name"], "World");
413    /// ```
414    pub fn new(job_type: impl Into<String>, payload: Value) -> Self {
415        let now = Utc::now();
416        Self {
417            dataset_id: "default".to_string(),
418            replay_of_job_id: None,
419            idempotency_key: None,
420            id: Uuid::new_v4(),
421            queue: "default".to_string(),
422            job_type: job_type.into(),
423            payload,
424            run_at: now,
425            deadline_at: None,
426            timeout_seconds: None,
427            recurring_interval_seconds: None,
428            status: JobStatus::Queued.as_str().to_string(),
429            priority: 0,
430            max_attempts: 25,
431            locked_at: None,
432            locked_by: None,
433            lock_expires_at: None,
434            dlq_reason_code: None,
435            dlq_at: None,
436            created_at: now,
437            updated_at: now,
438        }
439    }
440
441    /// Sets target queue name for this job.
442    pub fn queue(mut self, queue: impl Into<String>) -> Self {
443        self.queue = queue.into();
444        self
445    }
446
447    /// Sets job execution priority (higher numbers are leased first).
448    pub fn priority(mut self, priority: i32) -> Self {
449        self.priority = priority;
450        self
451    }
452
453    /// Sets maximum retry attempts before moving job to Dead-Letter Queue (DLQ).
454    pub fn max_attempts(mut self, max_attempts: i32) -> Self {
455        self.max_attempts = max_attempts;
456        self
457    }
458
459    /// Sets an application-provided enqueue idempotency key.
460    ///
461    /// Backends that support idempotent enqueue return the existing logical job ID when another
462    /// enqueue uses the same key.
463    pub fn idempotency_key(mut self, idempotency_key: impl Into<String>) -> Self {
464        self.idempotency_key = Some(idempotency_key.into());
465        self
466    }
467
468    /// Sets scheduled execution timestamp (`run_at`).
469    pub fn run_at(mut self, run_at: DateTime<Utc>) -> Self {
470        self.run_at = run_at;
471        self
472    }
473
474    /// Sets the latest timestamp at which this job may start execution.
475    ///
476    /// If the backend clock is already past this value when workers try to lease the job, Azums
477    /// moves the job to DLQ with `DEADLINE_EXCEEDED` instead of executing it late.
478    pub fn deadline_at(mut self, deadline_at: DateTime<Utc>) -> Self {
479        self.deadline_at = Some(deadline_at);
480        self
481    }
482
483    /// Sets a per-attempt handler timeout in seconds.
484    ///
485    /// Worker runtimes that execute handlers enforce this as a handler execution timeout and route
486    /// timeout failures through normal retry/DLQ classification.
487    pub fn timeout_seconds(mut self, timeout_seconds: i64) -> Self {
488        self.timeout_seconds = Some(timeout_seconds.max(0));
489        self
490    }
491
492    /// Sets fixed-interval recurring execution in seconds.
493    ///
494    /// After a successful occurrence, Azums enqueues the next occurrence as a new logical job with
495    /// `run_at = previous_run_at + recurring_interval_seconds`.
496    pub fn recurring_interval_seconds(mut self, interval_seconds: i64) -> Self {
497        self.recurring_interval_seconds = Some(interval_seconds.max(1));
498        self
499    }
500
501    /// Returns reference to job JSON payload.
502    pub fn payload_json(&self) -> &Value {
503        &self.payload
504    }
505
506    /// Derives the canonical lifecycle state from this persisted job and attempt history.
507    ///
508    /// `failed_attempts` is the number of durable failed `JobAttempt` rows for this job.
509    pub fn lifecycle_state_at(
510        &self,
511        now: DateTime<Utc>,
512        failed_attempts: usize,
513    ) -> Result<JobLifecycleState, crate::error::Error> {
514        JobLifecycleState::from_persisted(
515            JobStatus::parse(&self.status)?,
516            self.run_at,
517            now,
518            failed_attempts,
519        )
520    }
521
522    /// Deserializes the JSON payload into a concrete type `T`.
523    ///
524    /// # Examples
525    ///
526    /// ```rust
527    /// use azums_core::{Job, Error};
528    /// use serde::Deserialize;
529    ///
530    /// #[derive(Deserialize, Debug, PartialEq)]
531    /// struct EmailPayload {
532    ///     to: String,
533    /// }
534    ///
535    /// let job = Job::new("email", serde_json::json!({"to": "a@b.com"}));
536    /// let payload: EmailPayload = job.payload_typed().unwrap();
537    /// assert_eq!(payload.to, "a@b.com");
538    /// ```
539    pub fn payload_typed<T: serde::de::DeserializeOwned>(&self) -> Result<T, crate::error::Error> {
540        serde_json::from_value(self.payload.clone())
541            .map_err(crate::error::Error::PayloadDeserialization)
542    }
543}
544
545/// Trait-based job processor interface for structured background workers.
546#[async_trait::async_trait]
547pub trait JobProcessor: Send + Sync {
548    /// Processes a single background job execution attempt.
549    async fn process(&self, job: Job) -> anyhow::Result<()>;
550}
551
552/// Specification for enqueueing a new job into a storage backend.
553#[derive(Debug, Clone, Serialize, Deserialize)]
554pub struct NewJob {
555    pub queue: String,
556    pub job_type: String,
557    pub payload_json: Value,
558    pub idempotency_key: Option<String>,
559    pub run_at: DateTime<Utc>,
560    #[serde(default)]
561    pub deadline_at: Option<DateTime<Utc>>,
562    #[serde(default)]
563    pub timeout_seconds: Option<i64>,
564    #[serde(default)]
565    pub recurring_interval_seconds: Option<i64>,
566    pub priority: i32,
567    pub max_attempts: i32,
568}
569
570/// Runtime execution claim tying a job, durable attempt, worker, and lease together.
571///
572/// `JobExecution` is the in-flight view of work. The durable record of the handler run is
573/// `JobAttempt`; the durable record of the work item is `Job`.
574#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
575pub struct JobExecution {
576    pub job_id: Uuid,
577    pub attempt_id: Uuid,
578    pub attempt_no: i32,
579    pub worker_id: String,
580    pub lease_expires_at: DateTime<Utc>,
581    pub started_at: DateTime<Utc>,
582}
583
584impl From<Job> for NewJob {
585    fn from(job: Job) -> Self {
586        NewJob {
587            queue: job.queue,
588            job_type: job.job_type,
589            payload_json: job.payload,
590            idempotency_key: job.idempotency_key,
591            run_at: job.run_at,
592            deadline_at: job.deadline_at,
593            timeout_seconds: job.timeout_seconds,
594            recurring_interval_seconds: job.recurring_interval_seconds,
595            priority: job.priority,
596            max_attempts: job.max_attempts,
597        }
598    }
599}
600
601/// Stored job status values.
602///
603/// The canonical execution model is expressed by [`JobLifecycleState`]. Storage backends
604/// continue to persist compact lowercase strings for compatibility with existing schemas.
605#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
606pub enum JobStatus {
607    Queued,
608    Running,
609    /// Canonical completed terminal state.
610    Completed,
611    /// Backward-compatible alias for [`JobStatus::Completed`].
612    Succeeded,
613    /// Legacy job-level failure status. New executions should record failures on
614    /// `JobAttempt` and move the job to retry wait or DLQ instead.
615    Failed,
616    Dlq,
617    /// Canonical cancelled terminal state.
618    Cancelled,
619    /// Backward-compatible alias for [`JobStatus::Cancelled`].
620    Canceled,
621}
622
623impl JobStatus {
624    /// Returns static string representation of job status.
625    ///
626    /// # Examples
627    ///
628    /// ```rust
629    /// use azums_core::JobStatus;
630    /// assert_eq!(JobStatus::Queued.as_str(), "queued");
631    /// assert_eq!(JobStatus::Dlq.as_str(), "dlq");
632    /// ```
633    pub fn as_str(&self) -> &'static str {
634        match self {
635            JobStatus::Queued => "queued",
636            JobStatus::Running => "running",
637            JobStatus::Completed | JobStatus::Succeeded => "succeeded",
638            JobStatus::Failed => "failed",
639            JobStatus::Dlq => "dlq",
640            JobStatus::Cancelled | JobStatus::Canceled => "canceled",
641        }
642    }
643
644    /// Parses a persisted status string.
645    pub fn parse(status: &str) -> Result<Self, crate::error::Error> {
646        match status {
647            "queued" => Ok(JobStatus::Queued),
648            "running" => Ok(JobStatus::Running),
649            "succeeded" | "completed" => Ok(JobStatus::Completed),
650            "failed" => Ok(JobStatus::Failed),
651            "dlq" => Ok(JobStatus::Dlq),
652            "canceled" | "cancelled" => Ok(JobStatus::Cancelled),
653            other => Err(crate::error::Error::InvalidState(format!(
654                "unknown job status '{other}'"
655            ))),
656        }
657    }
658
659    /// Returns true when this persisted status represents a terminal job state.
660    pub fn is_terminal(&self) -> bool {
661        matches!(
662            self,
663            JobStatus::Completed
664                | JobStatus::Succeeded
665                | JobStatus::Dlq
666                | JobStatus::Cancelled
667                | JobStatus::Canceled
668        )
669    }
670}
671
672/// Canonical logical job lifecycle state.
673///
674/// `Scheduled` and `RetryWait` are derived from persisted state: both are stored as
675/// `status = "queued"` with a future `run_at`, but `RetryWait` also has prior failed
676/// attempt history. This keeps storage compact while still making lifecycle reconstruction
677/// deterministic from persisted job and attempt rows.
678#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
679pub enum JobLifecycleState {
680    Scheduled,
681    Queued,
682    Running,
683    Completed,
684    RetryWait,
685    Cancelled,
686    Dlq,
687}
688
689impl JobLifecycleState {
690    pub fn as_str(&self) -> &'static str {
691        match self {
692            JobLifecycleState::Scheduled => "scheduled",
693            JobLifecycleState::Queued => "queued",
694            JobLifecycleState::Running => "running",
695            JobLifecycleState::Completed => "completed",
696            JobLifecycleState::RetryWait => "retry_wait",
697            JobLifecycleState::Cancelled => "cancelled",
698            JobLifecycleState::Dlq => "dlq",
699        }
700    }
701
702    pub fn is_terminal(&self) -> bool {
703        matches!(
704            self,
705            JobLifecycleState::Completed | JobLifecycleState::Cancelled | JobLifecycleState::Dlq
706        )
707    }
708
709    pub fn legal_successors(&self) -> &'static [JobLifecycleState] {
710        use JobLifecycleState::*;
711        match self {
712            Scheduled => &[Queued],
713            Queued => &[Running],
714            Running => &[Completed, RetryWait, Cancelled, Dlq],
715            RetryWait => &[Queued],
716            Completed | Cancelled | Dlq => &[],
717        }
718    }
719
720    pub fn can_transition_to(&self, next: JobLifecycleState) -> bool {
721        self.legal_successors().contains(&next)
722    }
723
724    pub fn ensure_transition_to(&self, next: JobLifecycleState) -> Result<(), crate::error::Error> {
725        if self.can_transition_to(next) {
726            Ok(())
727        } else {
728            Err(crate::error::Error::InvalidState(format!(
729                "illegal job state transition: {} -> {}",
730                self.as_str(),
731                next.as_str()
732            )))
733        }
734    }
735
736    /// Derives the canonical state from persisted job state and attempt history.
737    pub fn from_persisted(
738        status: JobStatus,
739        run_at: DateTime<Utc>,
740        now: DateTime<Utc>,
741        failed_attempts: usize,
742    ) -> Result<Self, crate::error::Error> {
743        match status {
744            JobStatus::Queued if run_at > now && failed_attempts > 0 => {
745                Ok(JobLifecycleState::RetryWait)
746            }
747            JobStatus::Queued if run_at > now => Ok(JobLifecycleState::Scheduled),
748            JobStatus::Queued => Ok(JobLifecycleState::Queued),
749            JobStatus::Running => Ok(JobLifecycleState::Running),
750            JobStatus::Completed | JobStatus::Succeeded => Ok(JobLifecycleState::Completed),
751            JobStatus::Dlq => Ok(JobLifecycleState::Dlq),
752            JobStatus::Cancelled | JobStatus::Canceled => Ok(JobLifecycleState::Cancelled),
753            JobStatus::Failed => Err(crate::error::Error::InvalidState(
754                "job status 'failed' is legacy; failures belong to JobAttempt".to_string(),
755            )),
756        }
757    }
758}
759
760/// Asynchronous job handler closure type alias.
761pub type JobHandler = std::sync::Arc<
762    dyn Fn(Job) -> std::pin::Pin<Box<dyn std::future::Future<Output = anyhow::Result<()>> + Send>>
763        + Send
764        + Sync,
765>;
766
767/// Represents an immutable event stored within a durable stream log.
768#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
769#[cfg_attr(feature = "sqlx", derive(sqlx::FromRow))]
770pub struct Event {
771    /// Monotonically increasing 1-based sequence number within the stream.
772    pub sequence_no: i64,
773    /// Name of the target stream log (e.g., "orders", "audit_logs").
774    pub stream_name: String,
775    /// Domain-specific identifier for the event type (e.g., "order_created").
776    pub event_type: String,
777    /// JSON payload content of the event.
778    pub payload_json: serde_json::Value,
779    /// Timestamp when the event was appended to the stream log.
780    pub created_at: DateTime<Utc>,
781}
782
783/// Input model for publishing a new event into a stream log.
784#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
785pub struct NewEvent {
786    /// Domain-specific identifier for the event type (e.g., "order_created").
787    pub event_type: String,
788    /// JSON payload content of the event.
789    pub payload_json: serde_json::Value,
790}
791
792impl NewEvent {
793    /// Creates a new `NewEvent` with the specified event type and JSON payload.
794    pub fn new(event_type: impl Into<String>, payload_json: serde_json::Value) -> Self {
795        Self {
796            event_type: event_type.into(),
797            payload_json,
798        }
799    }
800}
801
802/// Status and offset information for a consumer group registered on a stream log.
803#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
804#[cfg_attr(feature = "sqlx", derive(sqlx::FromRow))]
805pub struct ConsumerGroupStatus {
806    /// Identifier of the consumer group (e.g., "analytics_processor").
807    pub consumer_group: String,
808    /// Name of the stream log.
809    pub stream_name: String,
810    /// Highest sequence number successfully acknowledged by this consumer group.
811    pub last_acked_seq: i64,
812    /// Timestamp when the offset was last updated.
813    pub updated_at: DateTime<Utc>,
814}
azums_core/model.rs

azums_core/
model.rs
