RUST: Add the database queue

src/rust/liblektor-rs/Cargo.toml

 [package]
-name    = "lektor-rs"
+name = "lektor-rs"
 version = "0.1.0"
 edition = "2021"
 
 [lib]
-crate-type = [ "staticlib" ]
+crate-type = ["staticlib"]
 
 [dependencies]
-log               = "0.4"
-libc              = "0.2.0"
+log = "0.4"
+libc = "0.2.0"
 diesel_migrations = "2"
-diesel            = { version =  "2", default-features = false, features = [ "sqlite" ] }
-serde             = { version = "^1", default-features = false, features = [ "std", "derive" ] }
+diesel = { version = "2", default-features = false, features = ["sqlite"] }
+serde = { version = "^1", default-features = false, features = [
+    "std",
+    "derive",
+] }

src/rust/liblektor-rs/src/database/connexion.rs

+use super::*;
+
+use crate::{database::models::*, kurisu_api::v1 as api_v1};
+use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness};
+
+/// The migrations!
+const MIGRATIONS: EmbeddedMigrations = embed_migrations!();
+
+/// Run migrations in a connexion!
+fn run_migration(conn: &mut SqliteConnection) -> Result<(), String> {
+    conn.run_pending_migrations(MIGRATIONS)
+        .map_err(|err| err.to_string())
+        .map(|_| ())
+}
+
+/// Create a connexion to a database and run automatically the migrations.
+fn establish_connection(path: impl AsRef<str>) -> Result<SqliteConnection, String> {
+    let mut conn = SqliteConnection::establish(path.as_ref())
+        .map_err(|err| format!("error connecting to {}: {}", path.as_ref(), err))?;
+    self::run_migration(&mut conn)?;
+    Ok(conn)
+}
+
+pub struct LktDatabaseConnection {
+    sqlite: SqliteConnection,
+    queue: queue::LktDatabaseQueue,
+}
+
+/// Load the diesel DSL for a given table. With the loaded dsl execute the
+/// expression...
+macro_rules! with_dsl {
+    ($table: ident => $expr: expr) => {{
+        use self::schema::$table::dsl::*;
+        use diesel::dsl::*;
+        $expr
+    }};
+}
+
+impl LktDatabaseConnection {
+    /// Create a new database connexion.
+    pub fn try_new(path: impl AsRef<Path>) -> LktDatabaseResult<Self> {
+        let path = path.as_ref().canonicalize().map_err(|err| {
+            LktDatabaseError::String(format!("failed to canonicalize a path: {err}"))
+        })?;
+        let sqlite =
+            establish_connection(path.to_string_lossy()).map_err(LktDatabaseError::String)?;
+        Ok(Self {
+            sqlite,
+            queue: Default::default(),
+        })
+    }
+
+    /// Get a tag id by its name.
+    pub fn get_tag_id_by_name(&mut self, tag_name: impl AsRef<str>) -> LktDatabaseResult<i32> {
+        Ok(with_dsl!(tag => tag.filter(name.is(tag_name.as_ref()))
+            .first::<Tag>(&mut self.sqlite)?.id
+        ))
+    }
+
+    /// Get a free local id for all karas. Note that using this function might
+    /// be unsafe because there is no guarenties that the returned ID will be
+    /// free by the time a kara is inserted with the said id...
+    pub fn get_kara_new_local_id(&mut self) -> LktDatabaseResult<i32> {
+        let max = with_dsl!(kara => kara.select(max(id))
+            .first::<Option<i32>>(&mut self.sqlite)?
+            .unwrap_or(0)
+        );
+        Ok(max + 1)
+    }
+
+    /// Delete a kara with its id in a repo.
+    pub fn delete_kara_by_repo(
+        &mut self,
+        arg_repo_id: u64,
+        arg_kara_id: u64,
+    ) -> LktDatabaseResult<()> {
+        let arg_repo_id = i32::try_from(arg_repo_id)?;
+        let arg_kara_id = i32::try_from(arg_kara_id)?;
+        self.sqlite.exclusive_transaction(|c| {
+            let local_id = with_dsl!(repo_kara => repo_kara
+                .filter(repo_id.eq(arg_repo_id))
+                .filter(repo_kara_id.eq(arg_kara_id))
+                .first::<KaraId>(c)?
+                .local_kara_id
+            );
+            with_dsl!(repo_kara => diesel::delete(repo_kara.filter(local_kara_id.eq(local_id))).execute(c)?);
+            with_dsl!(kara      => diesel::delete(kara.filter(id.eq(local_id))).execute(c)?);
+            Ok(())
+        })
+    }
+
+    /// Delete a kara by its local ID.
+    pub fn delete_kara_by_local_id(&mut self, kara_id: u64) -> LktDatabaseResult<()> {
+        let local_id = i32::try_from(kara_id)?;
+        self.sqlite.exclusive_transaction(|c| {
+            with_dsl!(repo_kara => diesel::delete(repo_kara.filter(local_kara_id.eq(local_id))).execute(c)?);
+            with_dsl!(kara      => diesel::delete(kara.filter(id.eq(local_id))).execute(c)?);
+            Ok(())
+        })
+    }
+
+    /// Ensure that a given language is present in the database. If it's not
+    /// insert it. Existence test is done on the code of the language.
+    pub fn ensure_language_exists(&mut self, lang: &Language) -> LktDatabaseResult<()> {
+        self.sqlite.exclusive_transaction(|c| {
+            with_dsl!(iso_639_1 => match iso_639_1.filter(code.eq(lang.code)).count().get_result(c)? {
+                1 => Ok(()),
+                0 => { diesel::insert_into(iso_639_1).values(lang).execute(c)?; Ok(()) }
+                count => Err(LktDatabaseError::String(format!("language `{lang:?}` has {count} occurences in the database..."))),
+            })
+        })
+    }
+
+    /// Add a kara with a request.
+    pub fn add_kara_from_request<'a>(&mut self, kara: NewKaraRequest<'a>) -> LktDatabaseResult<()> {
+        let (id, new_kara, lang, karamakers, tags) = kara;
+        self.ensure_language_exists(&lang)?;
+        self.sqlite.exclusive_transaction(|c| {
+            with_dsl!(kara        => diesel::insert_into(kara).values(&new_kara).execute(c)?);
+            with_dsl!(repo_kara   => diesel::insert_into(repo_kara).values(id).execute(c)?);
+            with_dsl!(kara_makers => diesel::insert_or_ignore_into(kara_makers).values(karamakers).execute(c)?);
+            with_dsl!(kara_tags   => {
+                diesel::delete(kara_tags.filter(kara_id.eq(new_kara.id))).execute(c)?;
+                diesel::insert_into(kara_tags).values(tags).execute(c)?;
+            });
+            Ok(())
+        })
+    }
+
+    /// Create a series of models from a kara signature from Kurisu's V1 API.
+    pub fn new_kara_v1<'a>(
+        &mut self,
+        repo_id: u64,
+        kara: api_v1::Kara<'a>,
+    ) -> LktDatabaseResult<NewKaraRequest<'a>> {
+        let local_id = self.get_kara_new_local_id()?;
+        let repo_id = i32::try_from(repo_id)?;
+        let id = KaraId {
+            repo_id,
+            local_kara_id: local_id,
+            repo_kara_id: i32::try_from(kara.id)?,
+        };
+        let lang = Language::from(kara.get_language());
+        let kara_makers = vec![KaraMaker {
+            id: local_id,
+            name: kara.author_name,
+        }];
+        let tags = vec![AddKaraTag {
+            kara_id: local_id,
+            tag_id: self.get_tag_id_by_name("number")?,
+            value: Some(format!("{}", kara.song_number)),
+        }];
+        let kara = NewKara {
+            id: local_id,
+            song_title: kara.song_name,
+            song_type: kara.song_type,
+            song_origin: kara.category,
+            source_name: kara.source_name,
+            language: lang.code,
+            file_hash: format!("{}", kara.unix_timestamp),
+        };
+        Ok((id, kara, lang, kara_makers, tags))
+    }
+}

src/rust/liblektor-rs/src/database/mod.rs

 //! Database implementation in rust for lektor.
 
-pub(self) use diesel::prelude::*;
-use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness};
-pub(self) use error::*;
-pub(self) use log::*;
-
+pub mod connexion;
 pub mod error;
 pub mod models;
+pub mod queue;
 pub mod schema;
 pub mod unsafe_interface;
 
-use crate::{database::models::*, kurisu_api::v1 as api_v1};
-
-/// The migrations!
-const MIGRATIONS: EmbeddedMigrations = embed_migrations!();
-
-/// Run migrations in a connexion!
-fn run_migration(conn: &mut SqliteConnection) -> Result<(), String> {
-    conn.run_pending_migrations(MIGRATIONS)
-        .map_err(|err| err.to_string())
-        .map(|_| ())
-}
-
-/// Create a connexion to a database and run automatically the migrations.
-pub fn establish_connection(path: impl AsRef<str>) -> Result<SqliteConnection, String> {
-    let mut conn = SqliteConnection::establish(path.as_ref())
-        .map_err(|err| format!("error connecting to {}: {}", path.as_ref(), err))?;
-    self::run_migration(&mut conn)?;
-    Ok(conn)
-}
+pub(self) use diesel::prelude::*;
+pub(self) use error::*;
+pub(self) use log::*;
+pub(self) use std::path::Path;
 
 /// All the information needed to add a kara recieved from a repo!
 pub type NewKaraRequest<'a> = (
@@ -38,131 +20,3 @@ pub type NewKaraRequest<'a> = (
     Vec<models::KaraMaker<'a>>,
     Vec<models::AddKaraTag>,
 );
-
-pub struct LktDatabaseConnection {
-    sqlite: SqliteConnection,
-}
-
-/// Load the diesel DSL for a given table. With the loaded dsl execute the
-/// expression...
-macro_rules! with_dsl {
-    ($table: ident => $expr: expr) => {{
-        use self::schema::$table::dsl::*;
-        use diesel::dsl::*;
-        $expr
-    }};
-}
-
-impl LktDatabaseConnection {
-    /// Get a tag id by its name.
-    pub fn get_tag_id_by_name(&mut self, tag_name: impl AsRef<str>) -> LktDatabaseResult<i32> {
-        Ok(with_dsl!(tag => tag.filter(name.is(tag_name.as_ref()))
-            .first::<Tag>(&mut self.sqlite)?.id
-        ))
-    }
-
-    /// Get a free local id for all karas. Note that using this function might
-    /// be unsafe because there is no guarenties that the returned ID will be
-    /// free by the time a kara is inserted with the said id...
-    pub fn get_kara_new_local_id(&mut self) -> LktDatabaseResult<i32> {
-        let max = with_dsl!(kara => kara.select(max(id))
-            .first::<Option<i32>>(&mut self.sqlite)?
-            .unwrap_or(0)
-        );
-        Ok(max + 1)
-    }
-
-    /// Delete a kara with its id in a repo.
-    pub fn delete_kara_by_repo(
-        &mut self,
-        arg_repo_id: u64,
-        arg_kara_id: u64,
-    ) -> LktDatabaseResult<()> {
-        let arg_repo_id = i32::try_from(arg_repo_id)?;
-        let arg_kara_id = i32::try_from(arg_kara_id)?;
-        self.sqlite.exclusive_transaction(|c| {
-            let local_id = with_dsl!(repo_kara => repo_kara
-                .filter(repo_id.eq(arg_repo_id))
-                .filter(repo_kara_id.eq(arg_kara_id))
-                .first::<KaraId>(c)?
-                .local_kara_id
-            );
-            with_dsl!(repo_kara => diesel::delete(repo_kara.filter(local_kara_id.eq(local_id))).execute(c)?);
-            with_dsl!(kara      => diesel::delete(kara.filter(id.eq(local_id))).execute(c)?);
-            Ok(())
-        })
-    }
-
-    /// Delete a kara by its local ID.
-    pub fn delete_kara_by_local_id(&mut self, kara_id: u64) -> LktDatabaseResult<()> {
-        let local_id = i32::try_from(kara_id)?;
-        self.sqlite.exclusive_transaction(|c| {
-            with_dsl!(repo_kara => diesel::delete(repo_kara.filter(local_kara_id.eq(local_id))).execute(c)?);
-            with_dsl!(kara      => diesel::delete(kara.filter(id.eq(local_id))).execute(c)?);
-            Ok(())
-        })
-    }
-
-    /// Ensure that a given language is present in the database. If it's not
-    /// insert it. Existence test is done on the code of the language.
-    pub fn ensure_language_exists(&mut self, lang: &Language) -> LktDatabaseResult<()> {
-        self.sqlite.exclusive_transaction(|c| {
-            with_dsl!(iso_639_1 => match iso_639_1.filter(code.eq(lang.code)).count().get_result(c)? {
-                1 => Ok(()),
-                0 => { diesel::insert_into(iso_639_1).values(lang).execute(c)?; Ok(()) }
-                count => Err(LktDatabaseError::String(format!("language `{lang:?}` has {count} occurences in the database..."))),
-            })
-        })
-    }
-
-    /// Add a kara with a request.
-    pub fn add_kara_from_request<'a>(&mut self, kara: NewKaraRequest<'a>) -> LktDatabaseResult<()> {
-        let (id, new_kara, lang, karamakers, tags) = kara;
-        self.ensure_language_exists(&lang)?;
-        self.sqlite.exclusive_transaction(|c| {
-            with_dsl!(kara        => diesel::insert_into(kara).values(&new_kara).execute(c)?);
-            with_dsl!(repo_kara   => diesel::insert_into(repo_kara).values(id).execute(c)?);
-            with_dsl!(kara_makers => diesel::insert_or_ignore_into(kara_makers).values(karamakers).execute(c)?);
-            with_dsl!(kara_tags   => {
-                diesel::delete(kara_tags.filter(kara_id.eq(new_kara.id))).execute(c)?;
-                diesel::insert_into(kara_tags).values(tags).execute(c)?;
-            });
-            Ok(())
-        })
-    }
-
-    /// Create a series of models from a kara signature from Kurisu's V1 API.
-    pub fn new_kara_v1<'a>(
-        &mut self,
-        repo_id: u64,
-        kara: api_v1::Kara<'a>,
-    ) -> LktDatabaseResult<NewKaraRequest<'a>> {
-        let local_id = self.get_kara_new_local_id()?;
-        let repo_id = i32::try_from(repo_id)?;
-        let id = KaraId {
-            repo_id,
-            local_kara_id: local_id,
-            repo_kara_id: i32::try_from(kara.id)?,
-        };
-        let lang = Language::from(kara.get_language());
-        let kara_makers = vec![KaraMaker {
-            id: local_id,
-            name: kara.author_name,
-        }];
-        let tags = vec![AddKaraTag {
-            kara_id: local_id,
-            tag_id: self.get_tag_id_by_name("number")?,
-            value: Some(format!("{}", kara.song_number)),
-        }];
-        let kara = NewKara {
-            id: local_id,
-            song_title: kara.song_name,
-            song_type: kara.song_type,
-            song_origin: kara.category,
-            source_name: kara.source_name,
-            language: lang.code,
-            file_hash: format!("{}", kara.unix_timestamp),
-        };
-        Ok((id, kara, lang, kara_makers, tags))
-    }
-}

src/rust/liblektor-rs/src/database/queue.rs

+use std::collections::VecDeque;
+
+/// The number of priority levels in the queues. The higher the number, the
+/// higher the priority.
+const LKT_DATABASE_QUEUES_COUNT: usize = 5;
+
+/// A type to describe a priority.
+pub struct LktDatabasePriority {
+    /// The actual priority. Alaways between `0` and
+    /// [LKT_DATABASE_QUEUES_COUNT], both included.
+    value: u8,
+}
+
+macro_rules! impl_from_for_proprity {
+    ($ty: ty) => {
+        impl From<$ty> for LktDatabasePriority {
+            fn from(value: $ty) -> Self {
+                let value: usize = value.try_into().unwrap_or(0);
+                let value = value.clamp(0, LKT_DATABASE_QUEUES_COUNT - 1) as u8;
+                Self { value }
+            }
+        }
+    };
+}
+
+impl_from_for_proprity!(u8);
+impl_from_for_proprity!(u16);
+impl_from_for_proprity!(u32);
+impl_from_for_proprity!(u64);
+
+impl_from_for_proprity!(i8);
+impl_from_for_proprity!(i16);
+impl_from_for_proprity!(i32);
+impl_from_for_proprity!(i64);
+
+impl Into<i32> for LktDatabasePriority {
+    fn into(self) -> i32 {
+        self.value as i32
+    }
+}
+
+impl Into<usize> for LktDatabasePriority {
+    fn into(self) -> usize {
+        self.value as usize
+    }
+}
+
+/// The iterator for the database queue.
+pub struct LktDatabaseQueueIter<'a> {
+    queue: &'a LktDatabaseQueue,
+    priority: Option<usize>,
+    index: usize,
+}
+
+impl<'a> Iterator for LktDatabaseQueueIter<'a> {
+    type Item = u64;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let priority = self.priority?.clamp(0, LKT_DATABASE_QUEUES_COUNT - 1);
+        let level = &self.queue.levels[priority];
+        match level.get(self.index) {
+            Some(ret) => {
+                self.index += 1;
+                Some(*ret)
+            }
+            None if priority == 0 => None,
+            None => {
+                self.priority = priority.checked_sub(1);
+                self.next()
+            }
+        }
+    }
+}
+
+/// The queue datastructure used for storing karas in the queue.
+#[derive(Debug, Default)]
+pub struct LktDatabaseQueue {
+    levels: [VecDeque<u64>; LKT_DATABASE_QUEUES_COUNT],
+}
+
+impl LktDatabaseQueue {
+    pub fn enqueue_kara(&mut self, local_id: u64) {
+        self.levels[0].push_back(local_id);
+    }
+
+    pub fn enqueue_kara_with_priority(&mut self, local_id: u64, priority: LktDatabasePriority) {
+        let priority: usize = priority.into();
+        self.levels[priority].push_back(local_id);
+    }
+
+    pub fn insert_kara(&mut self, local_id: u64) {
+        self.levels[0].push_front(local_id);
+    }
+
+    pub fn insert_kara_with_priority(&mut self, local_id: u64, priority: LktDatabasePriority) {
+        let priority: usize = priority.into();
+        self.levels[priority].push_front(local_id);
+    }
+
+    pub fn iter(&self) -> LktDatabaseQueueIter {
+        let priority = self
+            .levels
+            .iter()
+            .enumerate()
+            .rev()
+            .find_map(|(priority, content)| (!content.is_empty()).then_some(priority));
+        LktDatabaseQueueIter {
+            queue: self,
+            priority,
+            index: 0,
+        }
+    }
+
+    pub fn peek_next(&self) -> Option<u64> {
+        self.iter().next()
+    }
+
+    pub fn pop_next(&mut self) -> Option<u64> {
+        let level = self
+            .levels
+            .iter_mut()
+            .rev()
+            .find(|content| !content.is_empty())?;
+        level.pop_front()
+    }
+}

src/rust/liblektor-rs/src/database/unsafe_interface.rs

@@ -4,8 +4,8 @@
 //! Be carefull when naming things because those names might collide with things
 //! defined in lektor's C code...
 
-use super::*;
-use std::mem::ManuallyDrop;
+use super::{connexion::LktDatabaseConnection, *};
+use std::{mem::ManuallyDrop, path::PathBuf};
 
 /// Wrap the [`establish_connection`] function. On error log the message and
 /// return a [`std::ptr::null_mut`].
@@ -19,8 +19,8 @@ pub unsafe extern "C" fn lkt_database_establish_connection(
     }
     let len = path_len as usize;
     let path = ManuallyDrop::new(String::from_raw_parts(path as *mut _, len, len));
-    match establish_connection(&path[..]) {
-        Ok(conn) => Box::leak(Box::new(LktDatabaseConnection { sqlite: conn })) as *mut _,
+    match LktDatabaseConnection::try_new(PathBuf::from(&path[..])) {
+        Ok(db) => Box::leak(Box::new(db)) as *mut _,
         Err(err) => {
             error!("failed to establish connexion to {}: {err}", &path[..]);
             std::ptr::null_mut()