unify hook interface

Author: d4mr

twmq/src/lib.rs

@@ -4,6 +4,7 @@ pub mod job;
 pub mod queue;
 
 use std::marker::PhantomData;
+use std::process::Output;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
@@ -23,6 +24,20 @@ use tokio::time::sleep;
 pub use redis;
 use tracing::Instrument;
 
+pub struct SuccessHookData<'a, O> {
+    pub result: &'a O,
+}
+
+pub struct NackHookData<'a, E> {
+    pub error: &'a E,
+    pub delay: Option<Duration>,
+    pub position: RequeuePosition,
+}
+
+pub struct FailHookData<'a, E> {
+    pub error: &'a E,
+}
+
 // Main DurableExecution trait
 pub trait DurableExecution: Sized {
     type Output: Serialize + DeserializeOwned + Send + Sync;
@@ -32,33 +47,45 @@ pub trait DurableExecution: Sized {
     // Required method to process a job
     fn process(
         job: &Job<Self>,
+        c: &Self::ExecutionContext,
     ) -> impl Future<Output = JobResult<Self::Output, Self::ErrorData>> + Send + Sync;
 
     fn on_success(
         &self,
-        result: &Self::Output,
-        tx: &mut TransactionContext<'_>,
-        c: &Self::ExecutionContext,
-    ) -> impl Future<Output = ()> + Send + Sync;
+        _job: &Job<Self>,
+        _d: SuccessHookData<Self::Output>,
+        _tx: &mut TransactionContext<'_>,
+        _c: &Self::ExecutionContext,
+    ) -> impl Future<Output = ()> + Send + Sync {
+        std::future::ready(())
+    }
 
     fn on_nack(
         &self,
-        error: &Self::ErrorData,
-        delay: Option<Duration>,
-        position: RequeuePosition,
-        tx: &mut TransactionContext<'_>,
-        c: &Self::ExecutionContext,
-    ) -> impl Future<Output = ()> + Send + Sync;
+        _job: &Job<Self>,
+        _d: NackHookData<Self::ErrorData>,
+        _tx: &mut TransactionContext<'_>,
+        _c: &Self::ExecutionContext,
+    ) -> impl Future<Output = ()> + Send + Sync {
+        std::future::ready(())
+    }
 
     fn on_fail(
         &self,
-        error: &Self::ErrorData,
-        tx: &mut TransactionContext<'_>,
-        c: &Self::ExecutionContext,
-    ) -> impl Future<Output = ()> + Send + Sync;
+        _job: &Job<Self>,
+        _d: FailHookData<Self::ErrorData>,
+        _tx: &mut TransactionContext<'_>,
+        _c: &Self::ExecutionContext,
+    ) -> impl Future<Output = ()> + Send + Sync {
+        std::future::ready(())
+    }
 
-    fn on_timeout(&self, tx: &mut TransactionContext<'_>)
-    -> impl Future<Output = ()> + Send + Sync;
+    fn on_timeout(
+        &self,
+        _tx: &mut TransactionContext<'_>,
+    ) -> impl Future<Output = ()> + Send + Sync {
+        std::future::ready(())
+    }
 }
 
 // Main Queue struct
@@ -358,7 +385,7 @@ where
 
                             tokio::spawn(async move {
                                 // Process job - note we don't pass a context here
-                                let result = DurableExecution::process(&job).await;
+                                let result = DurableExecution::process(&job, &execution_context).await;
 
                                 // Mark job as complete (automatically happens in completion handlers)
 
@@ -374,8 +401,12 @@ where
                                             queue_clone.name().to_string(),
                                         );
 
+                                        let success_hook_data = SuccessHookData {
+                                            result: &output,
+                                        };
+
                                         // Call success hook to populate transaction context
-                                        job.data.on_success(&output, &mut tx_context, &execution_context).await;
+                                        job.data.on_success(&job, success_hook_data, &mut tx_context, &execution_context).await;
 
                                         // Complete job with success and execute transaction
                                         if let Err(e) = queue_clone
@@ -408,9 +439,15 @@ where
                                             queue_clone.name().to_string(),
                                         );
 
+                                        let nack_hook_data = NackHookData {
+                                            error: &error,
+                                            delay,
+                                            position,
+                                        };
+
                                         // Call nack hook to populate transaction context
                                         job.data
-                                            .on_nack(&error, delay, position, &mut tx_context,&execution_context)
+                                            .on_nack(&job, nack_hook_data, &mut tx_context,&execution_context)
                                             .await;
 
                                         // Complete job with nack and execute transaction
@@ -442,8 +479,12 @@ where
                                             queue_clone.name.clone(),
                                         );
 
+                                        let fail_hook_data = FailHookData {
+                                            error: &error
+                                        };
+
                                         // Call fail hook to populate transaction context
-                                        job.data.on_fail(&error, &mut tx_context, &execution_context).await;
+                                        job.data.on_fail(&job, fail_hook_data, &mut tx_context, &execution_context).await;
 
                                         // Complete job with fail and execute transaction
                                         if let Err(e) = queue_clone

twmq/tests/basic_hook.rs

@@ -14,9 +14,9 @@ use std::{
 
 use serde::{Deserialize, Serialize};
 use twmq::{
-    DurableExecution, Queue,
+    DurableExecution, Queue, SuccessHookData,
     hooks::TransactionContext,
-    job::{Job, JobResult, JobStatus, RequeuePosition},
+    job::{Job, JobResult, JobStatus},
     queue::QueueOptions,
 };
 
@@ -78,7 +78,10 @@ impl DurableExecution for MainJobPayload {
     type ErrorData = TestJobErrorData;
     type ExecutionContext = Arc<WebhookQueue>;
 
-    async fn process(job: &Job<Self>) -> JobResult<Self::Output, Self::ErrorData> {
+    async fn process(
+        job: &Job<Self>,
+        _: &Self::ExecutionContext,
+    ) -> JobResult<Self::Output, Self::ErrorData> {
         println!("MAIN_JOB: Processing job with id: {}", job.id);
         tokio::time::sleep(Duration::from_millis(50)).await;
 
@@ -91,15 +94,16 @@ impl DurableExecution for MainJobPayload {
 
     async fn on_success(
         &self,
-        result: &Self::Output,
+        _job: &Job<Self>,
+        d: SuccessHookData<'_, Self::Output>,
         tx: &mut TransactionContext<'_>,
         ec: &Self::ExecutionContext,
     ) {
         println!("MAIN_JOB: on_success hook - queuing webhook job");
 
         let webhook_job = WebhookJobPayload {
             url: "https://api.example.com/webhook".to_string(),
-            payload: serde_json::to_string(result).unwrap(),
+            payload: serde_json::to_string(d.result).unwrap(),
             parent_job_id: self.id_to_check.clone(),
         };
 
@@ -114,32 +118,17 @@ impl DurableExecution for MainJobPayload {
             tracing::info!("Successfully queued webhook job!");
         }
     }
-
-    async fn on_nack(
-        &self,
-        _error: &Self::ErrorData,
-        _delay: Option<Duration>,
-        _position: RequeuePosition,
-        _tx: &mut TransactionContext<'_>,
-        _: &Self::ExecutionContext,
-    ) {
-    }
-    async fn on_fail(
-        &self,
-        _error: &Self::ErrorData,
-        _tx: &mut TransactionContext<'_>,
-        _c: &Self::ExecutionContext,
-    ) {
-    }
-    async fn on_timeout(&self, _tx: &mut TransactionContext<'_>) {}
 }
 
 impl DurableExecution for WebhookJobPayload {
     type Output = WebhookJobOutput;
     type ErrorData = WebhookJobErrorData;
     type ExecutionContext = ();
 
-    async fn process(job: &Job<Self>) -> JobResult<Self::Output, Self::ErrorData> {
+    async fn process(
+        job: &Job<Self>,
+        _: &Self::ExecutionContext,
+    ) -> JobResult<Self::Output, Self::ErrorData> {
         println!("WEBHOOK_JOB: Sending webhook to: {}", job.data.url);
         println!("WEBHOOK_JOB: Payload: {}", job.data.payload);
         tokio::time::sleep(Duration::from_millis(25)).await;
@@ -155,32 +144,17 @@ impl DurableExecution for WebhookJobPayload {
 
     async fn on_success(
         &self,
-        _result: &Self::Output,
+        _job: &Job<Self>,
+        _d: SuccessHookData<'_, Self::Output>,
         _tx: &mut TransactionContext<'_>,
-        _: &Self::ExecutionContext,
+        _ec: &Self::ExecutionContext,
     ) {
-        println!(
+        tracing::info!(
             "WEBHOOK_JOB: Webhook delivered successfully for parent: {}",
             self.parent_job_id
         );
     }
 
-    async fn on_nack(
-        &self,
-        _error: &Self::ErrorData,
-        _delay: Option<Duration>,
-        _position: RequeuePosition,
-        _tx: &mut TransactionContext<'_>,
-        _: &Self::ExecutionContext,
-    ) {
-    }
-    async fn on_fail(
-        &self,
-        _error: &Self::ErrorData,
-        _tx: &mut TransactionContext<'_>,
-        _c: &Self::ExecutionContext,
-    ) {
-    }
     async fn on_timeout(&self, _tx: &mut TransactionContext<'_>) {}
 }
 

twmq/tests/fixtures.rs

@@ -6,9 +6,9 @@ use std::time::Duration;
 use serde::{Deserialize, Serialize};
 
 // Assuming your library's root is `twmq` or you use `crate::` if in lib.rs
-use twmq::DurableExecution;
 use twmq::hooks::TransactionContext;
-use twmq::job::{Job, JobResult, RequeuePosition};
+use twmq::job::{Job, JobResult};
+use twmq::{DurableExecution, SuccessHookData};
 
 // --- Test Job Definition ---
 
@@ -47,7 +47,10 @@ impl DurableExecution for TestJobPayload {
 
     // If not using async_trait, the signature is:
     // fn process(&self) -> impl std::future::Future<Output = JobResult<Self::Output, Self::ErrorData>> + Send + Sync {
-    async fn process(job: &Job<Self>) -> JobResult<Self::Output, Self::ErrorData> {
+    async fn process(
+        job: &Job<Self>,
+        _: &Self::ExecutionContext,
+    ) -> JobResult<Self::Output, Self::ErrorData> {
         println!(
             "TEST_JOB: Processing job with id_to_check: {}",
             job.data.id_to_check
@@ -60,40 +63,15 @@ impl DurableExecution for TestJobPayload {
         })
     }
 
-    async fn on_success(&self, _result: &Self::Output, _tx: &mut TransactionContext<'_>, _: &()) {
-        println!(
-            "TEST_JOB: on_success hook for id_to_check: {}",
-            self.id_to_check
-        );
-    }
-
-    async fn on_nack(
+    async fn on_success(
         &self,
-        _error: &Self::ErrorData,
-        _delay: Option<Duration>,
-        _position: RequeuePosition,
+        _job: &Job<Self>,
+        _d: SuccessHookData<'_, Self::Output>,
         _tx: &mut TransactionContext<'_>,
-        _: &(),
+        _ec: &Self::ExecutionContext,
     ) {
-        // Not expected for this test
-        println!(
-            "TEST_JOB: on_nack hook for id_to_check: {}",
-            self.id_to_check
-        );
-    }
-
-    async fn on_fail(&self, _error: &Self::ErrorData, _tx: &mut TransactionContext<'_>, _: &()) {
-        // Not expected for this test
-        println!(
-            "TEST_JOB: on_fail hook for id_to_check: {}",
-            self.id_to_check
-        );
-    }
-
-    async fn on_timeout(&self, _tx: &mut TransactionContext<'_>) {
-        // Not expected for this test
-        println!(
-            "TEST_JOB: on_timeout hook for id_to_check: {}",
+        tracing::info!(
+            "TEST_JOB: on_success hook for id_to_check: {}",
             self.id_to_check
         );
     }