move to bytes

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "apalis-sqlite"
-version = "1.0.0-alpha.2"
+version = "1.0.0-alpha.3"
 authors = ["Njuguna Mureithi <mureithinjuguna@gmail.com>"]
 readme = "README.md"
 edition = "2024"
@@ -15,8 +15,7 @@ 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"]
-bytes = []
-json = ["sqlx/json"]
+json = ["apalis-core/json", "sqlx/json"]
 
 [dependencies.sqlx]
 version = "0.8.6"
@@ -26,11 +25,8 @@ features = ["chrono", "sqlite"]
 [dependencies]
 serde = { version = "1", features = ["derive"] }
 serde_json = { version = "1" }
-apalis-core = { version = "1.0.0-alpha.6", default-features = false, features = [
-    "sleep",
-    "json",
-] }
-apalis-workflow = { version = "0.1.0-alpha.4" }
+apalis-core = { version = "1.0.0-alpha.7", features = ["sleep"] }
+apalis-sql = { version = "1.0.0-alpha.7" }
 log = "0.4.21"
 futures = "0.3.30"
 tokio = { version = "1", features = ["rt", "net"], optional = true }
@@ -40,17 +36,15 @@ thiserror = "2.0.0"
 pin-project = "1.1.10"
 libsqlite3-sys = "0.30.1"
 ulid = { version = "1", features = ["serde"] }
-tower-layer = "0.3.3"
-tower-service = "0.3.3"
 bytes = "1.1.0"
 
 [dev-dependencies]
 tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
-apalis-core = { version = "1.0.0-alpha.6", features = ["test-utils"] }
-apalis-workflow = { version = "0.1.0-alpha.4" }
+apalis-core = { version = "1.0.0-alpha.7", features = ["test-utils"] }
 apalis-sqlite = { path = ".", features = ["migrate", "tokio-comp"] }
+apalis-workflow = { version = "0.1.0-alpha.5" }
 
 [package.metadata.docs.rs]
 # defines the configuration attribute `docsrs`
 rustdoc-args = ["--cfg", "docsrs"]
-features = ["migrate", "tokio-comp", "json"]
+all-features = true

README.md

@@ -1,12 +1,12 @@
 # apalis-sqlite
 
-Background task processing for Rust using Apalis and SQLite.
+Background task processing for Rust using apalis and sqlite.
 
 ## Features
 
 - **Reliable job queue** using SQLite as the backend.
 - **Multiple storage types**: standard polling and event-driven (hooked) storage.
-- **Custom codecs** for serializing/deserializing job arguments as bytes and json.
+- **Custom codecs** for serializing/deserializing job arguments as bytes.
 - **Heartbeat and orphaned job re-enqueueing** for robust job processing.
 - **Integration with Apalis workers and middleware.**
 
@@ -34,7 +34,7 @@ async fn main() {
         start += 1;
         let task = Task::builder(start)
             .run_after(Duration::from_secs(1))
-            .with_ctx(SqliteContext::new().with_priority(1))
+            .with_ctx(SqlContext::new().with_priority(1))
             .build();
         Ok(task)
     })
@@ -79,7 +79,7 @@ async fn main() {
                 start += 1;
                 Task::builder(serde_json::to_value(&start).unwrap())
                     .run_after(Duration::from_secs(1))
-                    .with_ctx(SqliteContext::new().with_priority(start))
+                    .with_ctx(SqlContext::new().with_priority(start))
                     .build()
             })
             .take(20)
@@ -144,17 +144,10 @@ async fn main() {
 }
 ```
 
-## Migrations
+## Observability
 
-If the `migrate` feature is enabled, you can run built-in migrations with:
-
-```rust,no_run
-use sqlx::SqlitePool;
-#[tokio::main] async fn main() {
-    let pool = SqlitePool::connect(":memory:").await.unwrap();
-    apalis_sqlite::SqliteStorage::setup(&pool).await.unwrap();
-}
-```
+You can track your jobs using [apalis-board](https://github.com/apalis-dev/apalis-board).
+![Task](https://github.com/apalis-dev/apalis-board/raw/master/screenshots/task.png)
 
 ## License
 

migrations/bytes/20251017141210_initial_bytes_storage.sql → migrations/20251018164941_move_to_bytes.sql

@@ -1,16 +1,5 @@
-CREATE TABLE IF NOT EXISTS Workers (
-    id TEXT NOT NULL UNIQUE,
-    worker_type TEXT NOT NULL,
-    storage_name TEXT NOT NULL,
-    layers TEXT,
-    last_seen INTEGER NOT NULL DEFAULT (strftime('%s', 'now'))
-);
-
-CREATE INDEX IF NOT EXISTS Idx ON Workers(id);
-
-CREATE INDEX IF NOT EXISTS WTIdx ON Workers(worker_type);
-
-CREATE INDEX IF NOT EXISTS LSIdx ON Workers(last_seen);
+ALTER TABLE
+    Jobs RENAME TO Jobs_old;
 
 CREATE TABLE IF NOT EXISTS Jobs (
     job BLOB NOT NULL,
@@ -24,9 +13,47 @@ CREATE TABLE IF NOT EXISTS Jobs (
     lock_at INTEGER,
     lock_by TEXT,
     done_at INTEGER,
+    priority INTEGER NOT NULL DEFAULT 0,
+    metadata TEXT,
+    PRIMARY KEY(id),
     FOREIGN KEY(lock_by) REFERENCES Workers(id)
 );
 
+INSERT INTO
+    Jobs (
+        job,
+        id,
+        job_type,
+        status,
+        attempts,
+        max_attempts,
+        run_at,
+        last_result,
+        lock_at,
+        lock_by,
+        done_at,
+        priority,
+        metadata
+    )
+SELECT
+    CAST(job AS BLOB),
+    id,
+    job_type,
+    status,
+    attempts,
+    max_attempts,
+    run_at,
+    last_result,
+    lock_at,
+    lock_by,
+    done_at,
+    priority,
+    metadata
+FROM
+    Jobs_old;
+
+DROP TABLE Jobs_old;
+
 CREATE INDEX IF NOT EXISTS TIdx ON Jobs(id);
 
 CREATE INDEX IF NOT EXISTS SIdx ON Jobs(status);
@@ -35,10 +62,6 @@ CREATE INDEX IF NOT EXISTS LIdx ON Jobs(lock_by);
 
 CREATE INDEX IF NOT EXISTS JTIdx ON Jobs(job_type);
 
-
-ALTER TABLE Jobs
-ADD priority INTEGER NOT NULL DEFAULT 0;
-
 CREATE INDEX IF NOT EXISTS idx_jobs_job_type_status_run_at ON Jobs(job_type, status, run_at);
 
 CREATE INDEX IF NOT EXISTS idx_jobs_status_run_at ON Jobs(status, run_at);

src/ack.rs

@@ -1,20 +1,16 @@
-use std::any::Any;
-
 use apalis_core::{
     error::{AbortError, BoxDynError},
+    layers::{Layer, Service},
     task::{Parts, status::Status},
     worker::{context::WorkerContext, ext::ack::Acknowledge},
 };
-use apalis_workflow::StepResult;
+use apalis_sql::context::SqlContext;
 use futures::{FutureExt, future::BoxFuture};
 use serde::Serialize;
-use serde_json::Value;
 use sqlx::SqlitePool;
-use tower_layer::Layer;
-use tower_service::Service;
 use ulid::Ulid;
 
-use crate::{CompactType, SqliteTask, context::SqliteContext};
+use crate::SqliteTask;
 
 #[derive(Clone)]
 pub struct SqliteAck {
@@ -26,30 +22,19 @@ impl SqliteAck {
     }
 }
 
-impl<Res: Serialize + 'static> Acknowledge<Res, SqliteContext, Ulid> for SqliteAck {
+impl<Res: Serialize + 'static> Acknowledge<Res, SqlContext, Ulid> for SqliteAck {
     type Error = sqlx::Error;
     type Future = BoxFuture<'static, Result<(), Self::Error>>;
     fn ack(
         &mut self,
         res: &Result<Res, BoxDynError>,
-        parts: &Parts<SqliteContext, Ulid>,
+        parts: &Parts<SqlContext, Ulid>,
     ) -> Self::Future {
         let task_id = parts.task_id;
         let worker_id = parts.ctx.lock_by().clone();
 
         // Workflows need special handling to serialize the response correctly
-        let response = match res {
-            Ok(r) => {
-                if let Some(res_ref) = (r as &dyn Any).downcast_ref::<StepResult<CompactType>>() {
-                    let res_deserialized: Result<Value, serde_json::Error> =
-                        serde_json::from_str(&res_ref.0);
-                    serde_json::to_string(&res_deserialized.map_err(|e| e.to_string()))
-                } else {
-                    serde_json::to_string(&res.as_ref().map_err(|e| e.to_string()))
-                }
-            }
-            _ => serde_json::to_string(&res.as_ref().map_err(|e| e.to_string())),
-        };
+        let response = serde_json::to_string(&res.as_ref().map_err(|e| e.to_string()));
 
         let status = calculate_status(parts, res);
         parts.status.store(status.clone());
@@ -85,7 +70,7 @@ impl<Res: Serialize + 'static> Acknowledge<Res, SqliteContext, Ulid> for SqliteA
 }
 
 pub fn calculate_status<Res>(
-    parts: &Parts<SqliteContext, Ulid>,
+    parts: &Parts<SqlContext, Ulid>,
     res: &Result<Res, BoxDynError>,
 ) -> Status {
     match &res {