refactor(lib_core): introduce database module

refs: #5

lib_sync_core/src/database.rs

@@ -0,0 +1,197 @@
+use crate::task_manager::{Task, TaskPayload, TaskStatus};
+use futures::stream::BoxStream;
+use futures::{Stream, StreamExt, TryStreamExt};
+use serde::Serialize;
+use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode};
+use sqlx::{Error, QueryBuilder, Sqlite, SqlitePool};
+use std::path::PathBuf;
+use tokio::fs;
+use tracing::{info, instrument};
+
+static SQLITE_BIND_LIMIT: usize = 32766;
+
+#[derive(Default, Clone)]
+pub struct TaskPagination {
+    page_size: usize,
+    limit: Option<u32>,
+    offset: Option<u32>,
+    status: Option<TaskStatus>,
+}
+
+impl TaskPagination {
+    pub fn new() -> Self {
+        Self::default()
+    }
+    
+    pub fn next(&self) -> Self {
+       Self {
+            page_size: self.page_size + self.page_size,
+           ..self.clone()
+       } 
+    }
+
+    pub fn prev(&self) -> Self {
+        Self {
+            page_size: self.page_size - self.page_size,
+            ..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>) {
+        self.limit = limit;
+    }
+
+    pub fn set_offset(&mut self, offset: Option<u32>) {
+        self.offset = offset;
+    }
+
+    pub fn set_status(&mut self, status: Option<TaskStatus>) {
+        self.status = status;
+    }
+}
+
+pub struct TasksPage<T: TaskPayload> {
+   tasks: Vec<Task<T>>, 
+    page: TaskPagination
+}
+
+impl<T: TaskPayload> TasksPage<T> {
+    fn new(tasks: Vec<Task<T>>, page: TaskPagination) -> Self {
+       Self {
+           tasks,
+           page
+       } 
+    }
+
+    pub fn next(&self) -> TaskPagination {
+       self.page.next() 
+    }
+
+    pub fn prev(&self) -> TaskPagination {
+        self.page.prev()
+    }
+}
+
+pub trait TaskStorage<T: TaskPayload> {
+    fn insert_tasks(&self, tasks: Vec<Task<T>>) -> 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>>;
+}
+
+#[derive(Debug)]
+pub struct Database {
+    pool: SqlitePool,
+}
+
+impl Database {
+    pub async fn new<P: Into<PathBuf>>(base_path: P) -> crate::Result<Self> {
+        Ok(Self {
+            pool: Self::connect_database(base_path).await?,
+        })
+    }
+
+    async fn connect_database<P: Into<PathBuf>>(base_path: P) -> crate::Result<SqlitePool> {
+        let base_path = base_path.into();
+
+        let database_file_path = base_path.join("db.sql");
+
+        fs::create_dir_all(base_path).await?;
+
+        let opts = SqliteConnectOptions::new()
+            .filename(database_file_path)
+            .create_if_missing(true)
+            .journal_mode(SqliteJournalMode::Wal);
+
+        let pool = SqlitePool::connect_with(opts).await?;
+
+        sqlx::migrate!("../migrations").run(&pool).await?;
+
+        Ok(pool)
+    }
+
+    #[instrument(skip(self, values))]
+    pub async fn load_tasks<T>(&self, values: Vec<T>) -> crate::Result<()>
+    where
+        T: TaskPayload + Serialize + std::fmt::Debug,
+    {
+        let mut tx = self.pool.begin().await?;
+        let mut builder: QueryBuilder<'_, Sqlite> =
+            QueryBuilder::new("insert into tasks(payload_key, payload, status_id)");
+
+        let args: crate::Result<Vec<(String, String)>> = values
+            .iter()
+            .map(|value| Ok((value.get_key(), serde_json::to_string(value)?)))
+            .collect();
+
+        let mut affected_rows = 0;
+        // Chunk the query by the size limit of bind params
+        for chunk in args?.chunks(SQLITE_BIND_LIMIT / 3) {
+            builder.push_values(chunk, |mut builder, item| {
+                builder
+                    .push_bind(&item.0)
+                    .push_bind(&item.1)
+                    .push_bind(TaskStatus::Pending);
+            });
+            builder.push("ON conflict (payload_key) DO NOTHING");
+
+            let query = builder.build();
+
+            affected_rows += query.execute(&mut *tx).await?.rows_affected();
+            builder.reset();
+        }
+
+        tx.commit().await?;
+
+        info!("{} rows inserted.", affected_rows);
+
+        Ok(())
+    }
+}
+
+impl<T: TaskPayload> TaskStorage<T> for Database {
+    fn insert_tasks(&self, tasks: Vec<Task<T>>) -> crate::error::Result<()> {
+        todo!()
+    }
+
+    fn get_tasks(&self, task_status: TaskStatus) -> impl Stream<Item = crate::Result<Task<T>>> {
+        let query= sqlx::query_as::<_, Task<T>>(
+            "
+            SELECT id, payload_key, payload, status_id, created_at, updated_at
+            FROM tasks
+            WHERE status_id = ?
+            ORDER BY created_at DESC 
+            ",
+        ).bind(task_status);
+
+        query.fetch(&self.pool).err_into::<crate::Error>()
+    }
+
+    async fn get_paginated_tasks(&self, page: &TaskPagination) -> crate::Result<TasksPage<T>> {
+        let mut builder: QueryBuilder<'_, Sqlite> = QueryBuilder::new(
+            "select id, payload_key, payload, status_id, created_at, updated_at from tasks ",
+        );
+
+        if let Some(status) = &page.status {
+            builder.push("where status_id = ").push_bind(status);
+        }
+
+        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);
+        }
+
+        let tasks = builder.build_query_as::<Task<T>>().fetch_all(&self.pool).await?;
+        
+        Ok(TasksPage::new(tasks, page.clone()))
+    }
+}