feat: add buffered JobQueue and apalis enqueue

Author: 0xgleb

Cargo.lock

@@ -19,6 +19,7 @@ apalis = "1.0.0-rc.9"
 apalis-sqlite = "1.0.0-rc.8"
 apalis-core = "1.0.0-rc.9"
 apalis-codec = "0.1.0-rc.9"
+ulid = "1.2.1"
 
 [features]
 test-support = []

crates/event-sorcery/Cargo.toml

@@ -27,10 +27,13 @@ use apalis_sqlite::{CompactType, Config, SqlitePool, SqliteStorage};
 use serde::Serialize;
 use serde::de::DeserializeOwned;
 use std::future::Future;
+use std::marker::PhantomData;
 use std::sync::Arc;
 use std::time::Duration;
 use tracing::{debug, error};
 
+use crate::dependency::{Cons, Nil};
+
 /// A durable, retryable unit of side-effecting work.
 ///
 /// One `Job` type per entity (use [`Never`](crate::Never) for
@@ -74,6 +77,144 @@ pub trait Job: Serialize + DeserializeOwned + Send + 'static {
     ) -> impl Future<Output = Result<Self::Output, Self::Error>> + Send;
 }
 
+/// Type-level list of [`Job`] types an aggregate can dispatch.
+///
+/// Built from [`Cons`](crate::Cons) / [`Nil`](crate::Nil) -- the same
+/// cells [`deps!`](crate::deps) uses for reactor dependencies -- via
+/// the [`jobs!`](crate::jobs) macro: `jobs![SendEmail, ChargeCard]`.
+pub trait JobList {}
+
+impl JobList for Nil {}
+
+impl<Head: Job, Tail: JobList> JobList for Cons<Head, Tail> {}
+
+/// Membership marker: `J` is one of the job types in this [`JobList`].
+///
+/// Bounds [`JobQueue::push`] so a handler can enqueue a job only when
+/// its type is in the aggregate's declared list. The per-member impls
+/// are generated by [`register_jobs!`](crate::register_jobs), which
+/// [`jobs!`](crate::jobs) invokes.
+pub trait HasJob<J: Job>: JobList {}
+
+/// Single-element list: its one member is trivially present.
+impl<J: Job> HasJob<J> for Cons<J, Nil> {}
+
+/// Handler-facing handle for enqueuing the [`Job`]s an aggregate declares.
+///
+/// Generic over the aggregate's [`JobList`].
+/// [`push`](Self::push) / [`push_with_delay`](Self::push_with_delay)
+/// are synchronous and merely *buffer* onto the handle, and accept only
+/// job types in the list (`Jobs: HasJob<J>`). The framework drains the
+/// buffer inside the event-commit transaction, so enqueue is atomic with
+/// the triggering events and the handler stays I/O-free.
+///
+/// Constructed fresh per command by the framework and dropped at the
+/// end of the call.
+pub struct JobQueue<Jobs: JobList> {
+    buffered: Vec<PendingPush>,
+    _jobs: PhantomData<Jobs>,
+}
+
+impl<Jobs: JobList> JobQueue<Jobs> {
+    /// Buffers a job for execution as soon as the events commit.
+    ///
+    /// Compiles only when `J` is one of the aggregate's declared job
+    /// types (`Jobs: HasJob<J>`).
+    pub fn push<J: Job>(&mut self, job: J)
+    where
+        Jobs: HasJob<J>,
+    {
+        self.buffered.push(PendingPush {
+            job: Box::new(job),
+            delay: None,
+        });
+    }
+
+    /// Buffers a job to run no earlier than `delay` after commit.
+    pub fn push_with_delay<J: Job>(&mut self, job: J, delay: Duration)
+    where
+        Jobs: HasJob<J>,
+    {
+        self.buffered.push(PendingPush {
+            job: Box::new(job),
+            delay: Some(delay),
+        });
+    }
+
+    fn into_pending(self) -> Vec<PendingPush> {
+        self.buffered
+    }
+}
+
+impl<Jobs: JobList> Default for JobQueue<Jobs> {
+    fn default() -> Self {
+        Self {
+            buffered: Vec::new(),
+            _jobs: PhantomData,
+        }
+    }
+}
+
+/// A buffered push, type-erased so one queue can hold heterogeneous job
+/// types. The concrete type is recovered at flush only to write the
+/// `job_type` and JSON payload.
+struct PendingPush {
+    job: Box<dyn ErasedJob>,
+    delay: Option<Duration>,
+}
+
+/// Type-erased view of a buffered [`Job`].
+trait ErasedJob: Send {
+    fn job_type(&self) -> &'static str;
+    fn encode(&self) -> Result<Vec<u8>, serde_json::Error>;
+}
+
+impl<J: Job> ErasedJob for J {
+    fn job_type(&self) -> &'static str {
+        std::any::type_name::<J>()
+    }
+
+    fn encode(&self) -> Result<Vec<u8>, serde_json::Error> {
+        serde_json::to_vec(self)
+    }
+}
+
+/// Builds a type-level [`JobList`] from job types: `jobs![A, B]`
+/// expands to `Cons<A, Cons<B, Nil>>`, and `jobs![]` to `Nil`.
+///
+/// In statement position, declare the list and generate its
+/// [`HasJob`] impls in one call with
+/// [`register_jobs!`](crate::register_jobs).
+#[macro_export]
+macro_rules! jobs {
+    () => { $crate::Nil };
+    ($head:ty $(, $tail:ty)* $(,)?) => {
+        $crate::Cons<$head, $crate::jobs![$($tail),*]>
+    };
+}
+
+/// Generates the [`HasJob`] impls for each member of a job list, so
+/// `JobQueue::push` accepts every declared job type.
+///
+/// A single-element list is covered by a blanket impl, so this is a
+/// no-op there; multi-element lists need the generated impls.
+#[macro_export]
+macro_rules! register_jobs {
+    ($single:ty) => {};
+
+    ($($job:ty),+ $(,)?) => {
+        $crate::register_jobs!(@impls [$($job),+] [$($job),+]);
+    };
+
+    (@impls [] [$($all:ty),+]) => {};
+
+    (@impls [$current:ty $(, $rest:ty)*] [$($all:ty),+]) => {
+        impl $crate::HasJob<$current> for $crate::jobs![$($all),+] {}
+
+        $crate::register_jobs!(@impls [$($rest),*] [$($all),+]);
+    };
+}
+
 /// Worker-side handle to apalis's SQLite storage for a single job
 /// type.
 ///
@@ -130,6 +271,20 @@ pub enum JobError {
     Injected,
 }
 
+/// Error raised while flushing buffered jobs into apalis's queue.
+#[derive(Debug, thiserror::Error)]
+pub enum QueuePushError {
+    /// The job payload could not be JSON-encoded.
+    #[error("failed to encode job payload")]
+    Encode(#[from] serde_json::Error),
+    /// The job row could not be written.
+    #[error("failed to write job row")]
+    Sql(#[from] sqlx::Error),
+    /// The requested delay exceeds the representable range.
+    #[error("job delay exceeds the representable range")]
+    DelayOverflow(#[from] std::num::TryFromIntError),
+}
+
 /// Human-readable identifier for a job instance, used in logs and
 /// failure-injection targeting.
 #[derive(Debug, Clone)]
@@ -300,6 +455,42 @@ macro_rules! build_supervised_worker {
     }};
 }
 
+/// Drains a [`JobQueue`] into apalis's `Jobs` table through `connection`.
+///
+/// Called by the framework inside the event-commit transaction so a
+/// job becomes visible to the worker iff its triggering events commit.
+/// Rows are written in apalis's storage format -- JSON-encoded payload,
+/// `Pending` status, the job type as the queue name -- so the worker's
+/// fetcher picks them up.
+#[doc(hidden)]
+pub async fn flush_jobs<Jobs: JobList>(
+    connection: &mut sqlx::SqliteConnection,
+    queue: JobQueue<Jobs>,
+) -> Result<(), QueuePushError> {
+    for pending in queue.into_pending() {
+        let job_type = pending.job.job_type();
+        let payload = pending.job.encode()?;
+        let delay_secs = match pending.delay {
+            None => 0_i64,
+            Some(delay) => i64::try_from(delay.as_secs())?,
+        };
+        let id = ulid::Ulid::new().to_string();
+
+        sqlx::query(
+            "INSERT INTO Jobs (job, id, job_type, run_at) \
+             VALUES (?1, ?2, ?3, CAST(strftime('%s', 'now') AS INTEGER) + ?4)",
+        )
+        .bind(payload)
+        .bind(id)
+        .bind(job_type)
+        .bind(delay_secs)
+        .execute(&mut *connection)
+        .await?;
+    }
+
+    Ok(())
+}
+
 fn build_poll_config<J: 'static>() -> Config {
     let strategy = StrategyBuilder::new()
         .apply(
@@ -397,6 +588,30 @@ mod tests {
 
     use super::*;
 
+    async fn jobs_pool() -> sqlx::SqlitePool {
+        const JOBS_SCHEMA: &str = "CREATE TABLE Jobs ( \
+             job BLOB NOT NULL, \
+             id TEXT NOT NULL UNIQUE, \
+             job_type TEXT NOT NULL, \
+             status TEXT NOT NULL DEFAULT 'Pending', \
+             attempts INTEGER NOT NULL DEFAULT 0, \
+             max_attempts INTEGER NOT NULL DEFAULT 25, \
+             run_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now')), \
+             last_result TEXT, \
+             lock_at INTEGER, \
+             lock_by TEXT, \
+             done_at INTEGER, \
+             priority INTEGER NOT NULL DEFAULT 0, \
+             metadata TEXT, \
+             idempotency_key TEXT, \
+             PRIMARY KEY(id) \
+         )";
+
+        let pool = sqlx::SqlitePool::connect(":memory:").await.unwrap();
+        sqlx::query(JOBS_SCHEMA).execute(&pool).await.unwrap();
+        pool
+    }
+
     #[derive(Debug, Serialize, Deserialize)]
     enum SendEmail {
         Welcome { address: String },
@@ -423,6 +638,30 @@ mod tests {
         }
     }
 
+    #[derive(Debug, Serialize, Deserialize)]
+    struct ChargeCard {
+        cents: u64,
+    }
+
+    impl Job for ChargeCard {
+        type Input = ();
+        type Output = ();
+        type Error = Infallible;
+
+        const WORKER_NAME: &'static str = "charge-card";
+        const KIND: &'static str = "charge-card";
+
+        fn label(&self) -> Label {
+            Label::new(format!("charge:{}", self.cents))
+        }
+
+        async fn perform(&self, _input: &()) -> Result<(), Infallible> {
+            Ok(())
+        }
+    }
+
+    crate::register_jobs!(SendEmail, ChargeCard);
+
     #[test]
     fn label_reflects_variant_and_renders_via_display() {
         let welcome = SendEmail::Welcome {
@@ -483,4 +722,54 @@ mod tests {
             injector,
         );
     }
+
+    #[tokio::test]
+    async fn flush_writes_an_apalis_compatible_pending_row() {
+        let pool = jobs_pool().await;
+
+        let mut queue = JobQueue::<crate::jobs![SendEmail]>::default();
+        queue.push(SendEmail::Welcome {
+            address: "a@example.com".to_string(),
+        });
+
+        let mut transaction = pool.begin().await.unwrap();
+        flush_jobs(&mut transaction, queue).await.unwrap();
+        transaction.commit().await.unwrap();
+
+        let (payload, job_type, status): (Vec<u8>, String, String) =
+            sqlx::query_as("SELECT job, job_type, status FROM Jobs")
+                .fetch_one(&pool)
+                .await
+                .unwrap();
+
+        assert_eq!(job_type, std::any::type_name::<SendEmail>());
+        assert_eq!(status, "Pending");
+        let decoded: SendEmail = serde_json::from_slice(&payload).unwrap();
+        assert!(matches!(decoded, SendEmail::Welcome { address } if address == "a@example.com"));
+    }
+
+    #[tokio::test]
+    async fn flush_writes_distinct_rows_for_multiple_job_types() {
+        let pool = jobs_pool().await;
+
+        let mut queue = JobQueue::<crate::jobs![SendEmail, ChargeCard]>::default();
+        queue.push(SendEmail::Welcome {
+            address: "a@example.com".to_string(),
+        });
+        queue.push(ChargeCard { cents: 4200 });
+
+        let mut transaction = pool.begin().await.unwrap();
+        flush_jobs(&mut transaction, queue).await.unwrap();
+        transaction.commit().await.unwrap();
+
+        let rows: Vec<(String,)> = sqlx::query_as("SELECT job_type FROM Jobs ORDER BY job_type")
+            .fetch_all(&pool)
+            .await
+            .unwrap();
+
+        let job_types: Vec<String> = rows.into_iter().map(|(job_type,)| job_type).collect();
+        assert_eq!(job_types.len(), 2);
+        assert!(job_types.contains(&std::any::type_name::<ChargeCard>().to_string()));
+        assert!(job_types.contains(&std::any::type_name::<SendEmail>().to_string()));
+    }
 }

crates/event-sorcery/src/job.rs

@@ -110,10 +110,10 @@ pub use dependency::{Dependent, EntityList, Fold, HasEntity, OneOf};
 pub use job::FailureInjector;
 #[doc(hidden)]
 pub use job::{
-    ExponentialBackoff, FAIL_STOP_RECOVERY_TIMEOUT, RETRY_BACKOFF, Storage, on_terminal_failure,
-    work,
+    ExponentialBackoff, FAIL_STOP_RECOVERY_TIMEOUT, RETRY_BACKOFF, Storage, flush_jobs,
+    on_terminal_failure, work,
 };
-pub use job::{Job, JobBackend, JobError, Label};
+pub use job::{HasJob, Job, JobBackend, JobError, JobList, JobQueue, Label, QueuePushError};
 use lifecycle::Lifecycle;
 pub use lifecycle::{LifecycleError, Never};
 pub use projection::{Column, Projection, ProjectionError, Table};