feat: add optional time crate support

Cargo.toml

@@ -14,21 +14,24 @@ categories = ["asynchronous", "database", "network-programming"]
 publish = true
 
 [features]
-default = ["migrate", "tokio-comp"]
+default = ["migrate", "tokio-comp", "chrono"]
 migrate = ["sqlx/migrate", "sqlx/macros"]
 async-std-comp = ["async-std", "sqlx/runtime-async-std-rustls"]
 async-std-comp-native-tls = ["async-std", "sqlx/runtime-async-std-native-tls"]
 tokio-comp = ["tokio", "sqlx/runtime-tokio-rustls"]
 tokio-comp-native-tls = ["tokio", "sqlx/runtime-tokio-native-tls"]
+chrono = ["dep:chrono", "sqlx/chrono"]
+time = ["dep:time", "sqlx/time"]
 
 [dependencies]
-apalis-core = { version = "1.0.0-rc.1", default-features = false, features = [
+apalis-core = { path = "../apalis/apalis-core", default-features = false, features = [
   "sleep",
 ] }
-apalis-sql = { version = "1.0.0-rc.1", default-features = false }
+apalis-sql = { path = "../apalis/apalis-sql", default-features = false }
 apalis-codec = { version = "0.1.0-rc.1", features = ["json"] }
 serde = { version = "1", features = ["derive"], default-features = false }
-chrono = { version = "0.4", features = ["serde"], default-features = false }
+chrono = { version = "0.4", features = ["serde"], default-features = false, optional = true }
+time = { version = "0.3", features = ["serde"], optional = true }
 pin-project = "1.1.10"
 serde_json = "1"
 futures = "0.3.30"
@@ -44,11 +47,17 @@ ulid = { version = "1", features = ["serde"] }
 [dependencies.sqlx]
 version = "0.8.1"
 default-features = false
-features = ["chrono", "postgres", "json"]
+features = ["postgres", "json"]
+
+[package.metadata.cargo-udeps.ignore]
+# When both chrono and time features are enabled, time takes precedence
+# so chrono appears unused. This is expected behavior for --all-features.
+normal = ["chrono"]
 
 [dev-dependencies]
 tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
 once_cell = "1.19.0"
 apalis = { version = "1.0.0-rc.1" }
 apalis-workflow = { version = "0.1.0-rc.1" }
 futures-util = "0.3.30"
+

src/fetcher.rs

@@ -8,6 +8,8 @@ use std::{
 
 use apalis_core::{task::Task, timer::Delay, worker::context::WorkerContext};
 use apalis_sql::from_row::TaskRow;
+
+use crate::timestamp::PgDateTime;
 use futures::{Future, FutureExt, future::BoxFuture, stream::Stream};
 use pin_project::pin_project;
 
@@ -35,7 +37,7 @@ async fn fetch_next(
     .await?
     .into_iter()
     .map(|r| {
-        let row: TaskRow = r.try_into()?;
+        let row: TaskRow<PgDateTime> = r.try_into()?;
         row.try_into_task_compact()
             .map_err(|e| sqlx::Error::Protocol(e.to_string()))
     })

src/from_row.rs

@@ -1,4 +1,5 @@
-use chrono::{DateTime, Utc};
+use crate::{PgDateTime, RawDateTime};
+
 #[derive(Debug)]
 pub struct PgTaskRow {
     pub job: Option<Vec<u8>>,
@@ -7,18 +8,19 @@ pub struct PgTaskRow {
     pub status: Option<String>,
     pub attempts: Option<i32>,
     pub max_attempts: Option<i32>,
-    pub run_at: Option<DateTime<Utc>>,
+    pub run_at: Option<RawDateTime>,
     pub last_result: Option<serde_json::Value>,
-    pub lock_at: Option<DateTime<Utc>>,
+    pub lock_at: Option<RawDateTime>,
     pub lock_by: Option<String>,
-    pub done_at: Option<DateTime<Utc>>,
+    pub done_at: Option<RawDateTime>,
     pub priority: Option<i32>,
     pub metadata: Option<serde_json::Value>,
 }
-impl TryInto<apalis_sql::from_row::TaskRow> for PgTaskRow {
+
+impl TryInto<apalis_sql::from_row::TaskRow<PgDateTime>> for PgTaskRow {
     type Error = sqlx::Error;
 
-    fn try_into(self) -> Result<apalis_sql::from_row::TaskRow, Self::Error> {
+    fn try_into(self) -> Result<apalis_sql::from_row::TaskRow<PgDateTime>, Self::Error> {
         Ok(apalis_sql::from_row::TaskRow {
             job: self.job.unwrap_or_default(),
             id: self
@@ -35,11 +37,11 @@ impl TryInto<apalis_sql::from_row::TaskRow> for PgTaskRow {
                 .ok_or_else(|| sqlx::Error::Protocol("Missing attempts".into()))?
                 as usize,
             max_attempts: self.max_attempts.map(|v| v as usize),
-            run_at: self.run_at,
+            run_at: self.run_at.map(PgDateTime::from),
             last_result: self.last_result,
-            lock_at: self.lock_at,
+            lock_at: self.lock_at.map(PgDateTime::from),
             lock_by: self.lock_by,
-            done_at: self.done_at,
+            done_at: self.done_at.map(PgDateTime::from),
             priority: self.priority.map(|v| v as usize),
             metadata: self.metadata,
         })

src/lib.rs

@@ -6,7 +6,7 @@ use std::{fmt::Debug, marker::PhantomData};
 
 use apalis_codec::json::JsonCodec;
 use apalis_core::{
-    backend::{Backend, BackendExt, TaskStream, codec::Codec, queue::Queue},
+    backend::{Backend, BackendExt, TaskStream, codec::Codec},
     features_table,
     layers::Stack,
     task::{Task, task_id::TaskId},
@@ -35,8 +35,12 @@ pub use crate::{
 mod ack;
 mod fetcher;
 mod from_row;
+mod timestamp;
 
-pub type PgContext = apalis_sql::context::SqlContext<PgPool>;
+pub use timestamp::PgDateTime;
+pub(crate) use timestamp::{RawDateTime, now_raw, timestamp_from_unix};
+
+pub type PgContext = apalis_sql::context::SqlContext;
 mod queries;
 pub mod shared;
 pub mod sink;
@@ -251,9 +255,6 @@ where
     type Codec = Decode;
     type CompactStream = TaskStream<PgTask<CompactType>, Self::Error>;
 
-    fn get_queue(&self) -> Queue {
-        self.config.queue().clone()
-    }
     fn poll_compact(self, worker: &WorkerContext) -> Self::CompactStream {
         self.poll_basic(worker).boxed()
     }
@@ -348,10 +349,6 @@ where
     type Codec = Decode;
     type CompactStream = TaskStream<PgTask<CompactType>, Self::Error>;
 
-    fn get_queue(&self) -> Queue {
-        self.config.queue().clone()
-    }
-
     fn poll_compact(self, worker: &WorkerContext) -> Self::CompactStream {
         self.poll_with_notify(worker).boxed()
     }
@@ -413,7 +410,7 @@ impl<Args, Decode> PostgresStorage<Args, CompactType, Decode, PgNotify> {
                     )
                     .fetch(&mut *tx)
                     .map(|r| {
-                        let row: TaskRow = r?.try_into()?;
+                        let row: TaskRow<PgDateTime> = r?.try_into()?;
                         Ok(Some(
                             row.try_into_task_compact()
                                 .map_err(|e| sqlx::Error::Protocol(e.to_string()))?,

src/queries/fetch_by_id.rs

@@ -24,7 +24,7 @@ where
                 .fetch_optional(&pool)
                 .await?
                 .map(|r: PgTaskRow| {
-                    let row: TaskRow = r.try_into()?;
+                    let row: TaskRow<crate::PgDateTime> = r.try_into()?;
                     row.try_into_task_compact()
                         .and_then(|a| {
                             a.try_map(|t| D::decode(&t))

src/queries/keep_alive.rs

@@ -1,10 +1,9 @@
 use apalis_core::worker::context::WorkerContext;
-use chrono::Utc;
 use futures::{FutureExt, Stream, stream};
 use sqlx::PgPool;
 
 use crate::{
-    Config,
+    Config, now_raw,
     queries::{
         reenqueue_orphaned::reenqueue_orphaned, register_worker::register as register_worker,
     },
@@ -36,7 +35,7 @@ pub async fn initial_heartbeat(
     storage_type: &str,
 ) -> Result<(), sqlx::Error> {
     reenqueue_orphaned(pool.clone(), config.clone()).await?;
-    let last_seen = Utc::now();
+    let last_seen = now_raw();
     register_worker(
         pool,
         config.queue().to_string(),