feat: use SqlDateTime abstraction from apalis-sql

Cargo.toml

@@ -14,21 +14,22 @@ 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 = ["apalis-sql/chrono", "sqlx/chrono"]
+time = ["apalis-sql/time", "sqlx/time"]
 
 [dependencies]
-apalis-core = { version = "1.0.0-rc.1", default-features = false, features = [
+apalis-core = { path = "../apalis/apalis-core", version = "1.0.0-beta.2", default-features = false, features = [
   "sleep",
 ] }
-apalis-sql = { version = "1.0.0-rc.1", default-features = false }
+apalis-sql = { path = "../apalis/apalis-sql", version = "1.0.0-beta.2", 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 }
 pin-project = "1.1.10"
 serde_json = "1"
 futures = "0.3.30"
@@ -44,7 +45,7 @@ ulid = { version = "1", features = ["serde"] }
 [dependencies.sqlx]
 version = "0.8.1"
 default-features = false
-features = ["chrono", "postgres", "json"]
+features = ["postgres", "json"]
 
 [dev-dependencies]
 tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

src/from_row.rs

@@ -1,4 +1,5 @@
-use chrono::{DateTime, Utc};
+use apalis_sql::{SqlDateTime, TaskRow};
+
 #[derive(Debug)]
 pub struct PgTaskRow {
     pub job: Option<Vec<u8>>,
@@ -7,19 +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<SqlDateTime>,
     pub last_result: Option<serde_json::Value>,
-    pub lock_at: Option<DateTime<Utc>>,
+    pub lock_at: Option<SqlDateTime>,
     pub lock_by: Option<String>,
-    pub done_at: Option<DateTime<Utc>>,
+    pub done_at: Option<SqlDateTime>,
     pub priority: Option<i32>,
     pub metadata: Option<serde_json::Value>,
 }
-impl TryInto<apalis_sql::from_row::TaskRow> for PgTaskRow {
+impl TryInto<TaskRow> for PgTaskRow {
     type Error = sqlx::Error;
 
-    fn try_into(self) -> Result<apalis_sql::from_row::TaskRow, Self::Error> {
-        Ok(apalis_sql::from_row::TaskRow {
+    fn try_into(self) -> Result<TaskRow, Self::Error> {
+        Ok(TaskRow {
             job: self.job.unwrap_or_default(),
             id: self
                 .id

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},
@@ -36,7 +36,7 @@ mod ack;
 mod fetcher;
 mod from_row;
 
-pub type PgContext = apalis_sql::context::SqlContext<PgPool>;
+pub type PgContext = apalis_sql::context::SqlContext;
 mod queries;
 pub mod shared;
 pub mod sink;
@@ -251,9 +251,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 +345,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()
     }

src/queries/keep_alive.rs

@@ -1,5 +1,5 @@
 use apalis_core::worker::context::WorkerContext;
-use chrono::Utc;
+use apalis_sql::{SqlDateTime, SqlDateTimeExt};
 use futures::{FutureExt, Stream, stream};
 use sqlx::PgPool;
 
@@ -36,7 +36,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 = SqlDateTime::now();
     register_worker(
         pool,
         config.queue().to_string(),

src/queries/list_workers.rs

@@ -1,5 +1,5 @@
 use apalis_core::backend::{BackendExt, ListWorkers, RunningWorker};
-use chrono::{DateTime, Utc};
+use apalis_sql::{SqlDateTime, SqlDateTimeExt};
 use futures::TryFutureExt;
 use ulid::Ulid;
 
@@ -9,8 +9,8 @@ pub struct WorkerRow {
     pub worker_type: String,
     pub storage_name: String,
     pub layers: Option<String>,
-    pub last_seen: DateTime<Utc>,
-    pub started_at: Option<DateTime<Utc>>,
+    pub last_seen: SqlDateTime,
+    pub started_at: Option<SqlDateTime>,
 }
 
 use crate::{CompactType, PgContext, PostgresStorage};
@@ -42,8 +42,8 @@ where
                     .map(|w| RunningWorker {
                         id: w.id,
                         backend: w.storage_name,
-                        started_at: w.started_at.map(|t| t.timestamp()).unwrap_or_default() as u64,
-                        last_heartbeat: w.last_seen.timestamp() as u64,
+                        started_at: w.started_at.map(|t| t.to_unix_timestamp()).unwrap_or_default() as u64,
+                        last_heartbeat: w.last_seen.to_unix_timestamp() as u64,
                         layers: w.layers.unwrap_or_default(),
                         queue: w.worker_type,
                     })
@@ -73,8 +73,8 @@ where
                     .map(|w| RunningWorker {
                         id: w.id,
                         backend: w.storage_name,
-                        started_at: w.started_at.map(|t| t.timestamp()).unwrap_or_default() as u64,
-                        last_heartbeat: w.last_seen.timestamp() as u64,
+                        started_at: w.started_at.map(|t| t.to_unix_timestamp()).unwrap_or_default() as u64,
+                        last_heartbeat: w.last_seen.to_unix_timestamp() as u64,
                         layers: w.layers.unwrap_or_default(),
                         queue: w.worker_type,
                     })

src/queries/register_worker.rs

@@ -1,12 +1,12 @@
 use apalis_core::worker::context::WorkerContext;
-use chrono::{DateTime, Utc};
+use apalis_sql::SqlDateTime;
 use sqlx::PgPool;
 
 pub async fn register(
     pool: PgPool,
     worker_type: String,
     worker: WorkerContext,
-    last_seen: DateTime<Utc>,
+    last_seen: SqlDateTime,
     backend_type: &str,
 ) -> Result<(), sqlx::Error> {
     let res = sqlx::query_file!(

src/queries/wait_for.rs

@@ -24,7 +24,7 @@ where
         BackendExt<Context = PgContext, Compact = CompactType, IdType = Ulid, Error = sqlx::Error>,
     Result<O, String>: DeserializeOwned,
 {
-    type ResultStream = BoxStream<'static, Result<TaskResult<O, Ulid>, Self::Error>>;
+    type ResultStream = BoxStream<'static, Result<TaskResult<O>, Self::Error>>;
     fn wait_for(
         &self,
         task_ids: impl IntoIterator<Item = TaskId<Self::IdType>>,
@@ -80,7 +80,7 @@ where
     fn check_status(
         &self,
         task_ids: impl IntoIterator<Item = TaskId<Self::IdType>> + Send,
-    ) -> impl Future<Output = Result<Vec<TaskResult<O, Ulid>>, Self::Error>> + Send {
+    ) -> impl Future<Output = Result<Vec<TaskResult<O>>, Self::Error>> + Send {
         let pool = self.pool.clone();
         let ids: Vec<String> = task_ids.into_iter().map(|id| id.to_string()).collect();