test(lib_core): sqlite return paginated tasks

refs: #5

lib_sync_core/Cargo.toml

@@ -23,3 +23,4 @@ uuid = { version = "1.16.0", features = ["serde", "v4"] }
 
 [dev-dependencies]
 fake = { version = "4.3.0", features = ["derive", "chrono", "http", "uuid"] }
+tracing-test = "0.2.5"

lib_sync_core/src/database.rs

@@ -1,12 +1,11 @@
 use crate::tasks::{Task, TaskPayload, TaskStatus};
-use futures::{Stream};
+use futures::Stream;
 mod sqlite;
 
-#[derive(Default, Clone)]
+#[derive(Default, Clone, Debug)]
 pub struct TaskPagination {
-    page_size: usize,
     limit: Option<u32>,
-    offset: Option<u32>,
+    offset: u32,
     status: Option<TaskStatus>,
 }
 
@@ -17,38 +16,37 @@ impl TaskPagination {
     
     pub fn next(&self) -> Self {
        Self {
-            page_size: self.page_size + self.page_size,
+            offset: self.offset.saturating_add(self.limit.unwrap_or(0)),
            ..self.clone()
-       } 
+       }
     }
 
     pub fn prev(&self) -> Self {
         Self {
-            page_size: self.page_size - self.page_size,
+            offset: self.offset.saturating_sub(self.limit.unwrap_or(0)),
             ..self.clone()
         }
     }
 
-    pub fn set_page_size(&mut self, page_size: usize) {
-        self.page_size = page_size;
-    }
-
-    pub fn set_limit(&mut self, limit: Option<u32>) {
+    pub fn with_limit(mut self, limit: Option<u32>) -> Self {
         self.limit = limit;
+        self
     }
 
-    pub fn set_offset(&mut self, offset: Option<u32>) {
+    pub fn with_offset(mut self, offset: u32) -> Self {
         self.offset = offset;
+        self
     }
 
-    pub fn set_status(&mut self, status: Option<TaskStatus>) {
+    pub fn with_status(mut self, status: Option<TaskStatus>) -> Self {
         self.status = status;
+        self
     }
 }
 
 pub struct TasksPage<T: TaskPayload> {
    tasks: Vec<Task<T>>, 
-    page: TaskPagination
+   page: TaskPagination
 }
 
 impl<T: TaskPayload> TasksPage<T> {
@@ -72,6 +70,6 @@ pub trait TaskStorage<T: TaskPayload> {
     async fn insert_tasks<'a, I: IntoIterator<Item=&'a Task<T>>>(&self, tasks: I) -> crate::Result<()>;
     fn get_tasks(&self, options: TaskStatus) -> impl Stream<Item = crate::Result<Task<T>>>;
 
-    async fn get_paginated_tasks(&self, page: &TaskPagination) -> crate::Result<TasksPage<T>>;
+    async fn get_paginated_tasks(&self, page: TaskPagination) -> crate::Result<TasksPage<T>>;
 }
 

lib_sync_core/src/database/sqlite.rs

@@ -5,7 +5,7 @@ use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode};
 use sqlx::{QueryBuilder, SqlitePool};
 use std::path::PathBuf;
 use tokio::fs;
-use tracing::{info, instrument};
+use tracing::{debug, info, instrument};
 
 static SQLITE_BIND_LIMIT: usize = 32766;
 static MIGRATIONS: sqlx::migrate::Migrator = sqlx::migrate!("../migrations");
@@ -108,7 +108,7 @@ impl<T: TaskPayload> TaskStorage<T> for Sqlite {
         query.fetch(&self.pool).err_into::<crate::Error>()
     }
 
-    async fn get_paginated_tasks(&self, page: &TaskPagination) -> crate::Result<TasksPage<T>> {
+    async fn get_paginated_tasks(&self, page: TaskPagination) -> crate::Result<TasksPage<T>> {
         let mut builder: QueryBuilder<'_, sqlx::Sqlite> = QueryBuilder::new(
             "select id, payload_key, payload, status_id, created_at, updated_at from tasks ",
         );
@@ -119,13 +119,13 @@ impl<T: TaskPayload> TaskStorage<T> for Sqlite {
 
         builder.push("ORDER BY created_at DESC ");
 
-        if let Some(limit) = &page.offset {
-            builder.push("OFFSET ").push_bind(limit);
+
+        if let Some(limit) = page.limit {
+            builder.push("LIMIT ").push_bind(limit);
+            builder.push(" OFFSET ").push_bind(page.offset);
         }
 
-        if let Some(limit) = &page.limit {
-            builder.push("LIMIT ").push_bind(limit);
-        }
+        debug!("SQL: \"{}\", with options: \"{:?}\"", builder.sql(), page);
 
         let tasks = builder
             .build_query_as::<Task<T>>()
@@ -139,25 +139,39 @@ impl<T: TaskPayload> TaskStorage<T> for Sqlite {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use fake::{Dummy, Fake, Faker};
+    use fake::Dummy;
     use futures::StreamExt;
     use serde::{Deserialize, Serialize};
     use sqlx::types::Uuid;
     use sqlx::Row;
+    use tracing_test::traced_test;
 
     #[derive(Dummy, Serialize, Deserialize, Debug)]
-struct DummyTaskPayload {
+    struct DummyTaskPayload {
         key: Uuid,
         _foo: String,
-        _baar: String,
+        _bar: String,
+    }
+
+    //noinspection RsUnresolvedPath
+    fn setup(pool: SqlitePool, len: usize) -> (Sqlite, Vec<Task<DummyTaskPayload>>) {
+        let owned_pool = pool.clone();
+        let sqlite = Sqlite { pool: owned_pool };
+
+        let payloads = fake::vec![DummyTaskPayload; len];
+
+        let tasks = payloads
+            .into_iter()
+            .enumerate()
+            .map(|(i, item)| Task::new((i + 1).to_string(), item, TaskStatus::Pending))
+            .collect();
+
+        (sqlite, tasks)
     }
 
     #[sqlx::test(migrator = "MIGRATIONS")]
     async fn it_save_tasks(pool: SqlitePool) -> sqlx::Result<()> {
-        let owned_pool = pool.clone();
-        let sqlite = Sqlite { pool: owned_pool };
-
-        let tasks = generate_dummy_tasks();
+        let (sqlite, tasks) = setup(pool.clone(), 100);
 
         sqlite.insert_tasks(&tasks).await.unwrap();
 
@@ -176,22 +190,35 @@ struct DummyTaskPayload {
             .await;
 
         assert_eq!(tasks.len(), saved_tasks.len());
-        
+
         let mut zip = tasks.into_iter().zip(saved_tasks.into_iter());
 
-        assert!(zip.all(|(a, b)| {
-            a.get_key() == b.get_key()            
-        }));
-        
+        assert!(zip.all(|(a, b)| { a.get_key() == b.get_key() }));
+
         Ok(())
     }
 
-    fn generate_dummy_tasks() -> Vec<Task<DummyTaskPayload>> {
-        let payloads: Vec<DummyTaskPayload> = Faker.fake();
+    #[sqlx::test(migrator = "MIGRATIONS")]
+    #[traced_test]
+    async fn it_return_paginated_tasks(pool: SqlitePool) -> sqlx::Result<()> {
+        let (sqlite, tasks) = setup(pool.clone(), 300);
 
-        payloads
-            .into_iter()
-            .map(|item| Task::new(item.key.to_string(), item, TaskStatus::Pending))
-            .collect()
+        sqlite.insert_tasks(&tasks).await.unwrap();
+
+        let page_options = TaskPagination::new().with_limit(Some(25));
+
+        let first_page: TasksPage<DummyTaskPayload> =
+            sqlite.get_paginated_tasks(page_options).await.unwrap();
+
+        assert_eq!(first_page.tasks.len(), 25);
+        assert_eq!(first_page.tasks.first().unwrap().get_key(), tasks.get(0).unwrap().get_key());
+
+        let second_page: TasksPage<DummyTaskPayload> =
+            sqlite.get_paginated_tasks(first_page.next()).await.unwrap();
+
+        assert_eq!(second_page.tasks.len(), 25);
+        assert_eq!(second_page.tasks.first().unwrap().get_key(), tasks.get(25).unwrap().get_key());
+
+        Ok(())
     }
 }

lib_sync_core/src/tasks.rs

@@ -35,10 +35,10 @@ impl Display for TaskStatus {
 }
 
 pub trait TaskPayload:
-    Serialize + DeserializeOwned + Send + Unpin + 'static
+    Serialize + DeserializeOwned + Send + Unpin + 'static + std::fmt::Debug
 {
 }
-impl<T: Serialize + DeserializeOwned + Send + Unpin + 'static>
+impl<T: Serialize + DeserializeOwned + Send + Unpin + 'static + std::fmt::Debug>
     TaskPayload for T
 {
 }