server/src/api/mod.rs

@@ -1,7 +1,5 @@
 mod pagination;
 
-use sea_orm::{*};
-
 use axum::extract::{Path, Query, State};
 use axum::routing::get;
 use axum::Json;
@@ -9,7 +7,7 @@ use axum::Router;
 
 use pagination::PaginatedResponse;
 
-use crate::db::{self, *};
+use crate::db;
 
 pub fn router() -> Router<crate::Global> {
     Router::new()
@@ -22,34 +20,26 @@ pub fn router() -> Router<crate::Global> {
 async fn get_repos(
     State(global): State<crate::Global>,
     Query(pagination): Query<pagination::Query>,
-    Query(filter): Query<db::query::repo::Filter>,
 ) -> crate::Result<Json<PaginatedResponse<db::repo::Model>>> {
-    let page = pagination.page.unwrap_or(1) - 1;
-    let per_page = pagination.per_page.unwrap_or(25);
-
-    let paginator = Repo::find()
-        .filter(filter)
-        .order_by_asc(package::Column::Id)
-        .paginate(&global.db, pagination.per_page.unwrap_or(25));
-    let items = paginator
-        .fetch_page(pagination.page.unwrap_or(1) - 1)
+    let (total_pages, repos) = global
+        .db
+        .repo
+        .page(
+            pagination.per_page.unwrap_or(25),
+            pagination.page.unwrap_or(1) - 1,
+        )
         .await?;
-    let total_pages = paginator.num_pages().await?;
-
-    Ok(Json(PaginatedResponse {
-        page,
-        per_page,
-        total_pages,
-        count: items.len(),
-        items,
-    }))
+    Ok(Json(pagination.res(total_pages, repos)))
 }
 
 async fn get_single_repo(
     State(global): State<crate::Global>,
     Path(id): Path<i32>,
 ) -> crate::Result<Json<db::repo::Model>> {
-    let repo = db::query::repo::by_id(&global.db, id)
+    let repo = global
+        .db
+        .repo
+        .by_id(id)
         .await?
         .ok_or(axum::http::StatusCode::NOT_FOUND)?;
 
@@ -59,13 +49,13 @@ async fn get_single_repo(
 async fn get_packages(
     State(global): State<crate::Global>,
     Query(pagination): Query<pagination::Query>,
-    Query(filter): Query<db::query::package::Filter>,
 ) -> crate::Result<Json<PaginatedResponse<db::package::Model>>> {
-    let (total_pages, pkgs) = db::query::package::page(
-        &global.db,
+    let (total_pages, pkgs) = global
+        .db
+        .pkg
+        .page(
             pagination.per_page.unwrap_or(25),
             pagination.page.unwrap_or(1) - 1,
-        filter,
         )
         .await?;
 
@@ -76,7 +66,10 @@ async fn get_single_package(
     State(global): State<crate::Global>,
     Path(id): Path<i32>,
 ) -> crate::Result<Json<crate::db::FullPackage>> {
-    let entry = db::query::package::full(&global.db, id)
+    let entry = global
+        .db
+        .pkg
+        .full(id)
         .await?
         .ok_or(axum::http::StatusCode::NOT_FOUND)?;
 

server/src/api/pagination.rs

@@ -1,6 +1,6 @@
 use serde::{Deserialize, Serialize};
 
-pub const DEFAULT_PAGE: u64 = 1;
+pub const DEFAULT_PAGE: u64 = 0;
 pub const DEFAULT_PER_PAGE: u64 = 25;
 
 #[derive(Deserialize)]

server/src/cli.rs

@@ -10,7 +10,6 @@ use std::sync::{Arc, RwLock};
 use tower_http::trace::TraceLayer;
 use tracing::debug;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
-use sea_orm_migration::MigratorTrait;
 
 #[derive(Parser)]
 #[command(author, version, about, long_about = None)]
@@ -76,8 +75,10 @@ impl Cli {
 
         debug!("Connecting to database with URL {}", db_url);
 
-        let db = sea_orm::Database::connect(db_url).await?;
-        crate::db::Migrator::up(&db, None).await?;
+        let db = crate::db::RieterDb::connect(db_url).await?;
+        // let db = crate::db::init("postgres://rieter:rieter@localhost:5432/rieter")
+        // .await
+        // .unwrap();
 
         let config = Config {
             data_dir: self.data_dir.clone(),

server/src/db/conn.rs

@@ -0,0 +1,61 @@
+use super::RieterDb;
+use sea_orm::{DbBackend, DbErr, ExecResult, QueryResult, Statement};
+use std::{future::Future, pin::Pin};
+
+// Allows RieterDb objects to be passed to ORM functions
+impl sea_orm::ConnectionTrait for RieterDb {
+    fn get_database_backend(&self) -> DbBackend {
+        self.conn.get_database_backend()
+    }
+    fn execute<'life0, 'async_trait>(
+        &'life0 self,
+        stmt: Statement,
+    ) -> Pin<Box<dyn Future<Output = std::result::Result<ExecResult, DbErr>> + Send + 'async_trait>>
+    where
+        Self: 'async_trait,
+        'life0: 'async_trait,
+    {
+        self.conn.execute(stmt)
+    }
+    fn execute_unprepared<'life0, 'life1, 'async_trait>(
+        &'life0 self,
+        sql: &'life1 str,
+    ) -> Pin<Box<dyn Future<Output = std::result::Result<ExecResult, DbErr>> + Send + 'async_trait>>
+    where
+        Self: 'async_trait,
+        'life0: 'async_trait,
+        'life1: 'async_trait,
+    {
+        self.conn.execute_unprepared(sql)
+    }
+    fn query_one<'life0, 'async_trait>(
+        &'life0 self,
+        stmt: Statement,
+    ) -> Pin<
+        Box<
+            dyn Future<Output = std::result::Result<Option<QueryResult>, DbErr>>
+                + Send
+                + 'async_trait,
+        >,
+    >
+    where
+        Self: 'async_trait,
+        'life0: 'async_trait,
+    {
+        self.conn.query_one(stmt)
+    }
+    fn query_all<'life0, 'async_trait>(
+        &'life0 self,
+        stmt: Statement,
+    ) -> Pin<
+        Box<
+            dyn Future<Output = std::result::Result<Vec<QueryResult>, DbErr>> + Send + 'async_trait,
+        >,
+    >
+    where
+        Self: 'async_trait,
+        'life0: 'async_trait,
+    {
+        self.conn.query_all(stmt)
+    }
+}

server/src/db/mod.rs

@@ -1,20 +1,19 @@
+mod conn;
 pub mod entities;
 mod migrator;
-pub mod query;
-
-use sea_orm::{DeriveActiveEnum, EnumIter};
+mod query;
 
+use sea_orm::{ConnectOptions, Database, DatabaseConnection, DeriveActiveEnum, EnumIter};
+use sea_orm_migration::MigratorTrait;
 use serde::{Deserialize, Serialize};
 
 pub use entities::{prelude::*, *};
-pub use migrator::Migrator;
-
+use migrator::Migrator;
 
 type Result<T> = std::result::Result<T, sea_orm::DbErr>;
 
 #[derive(EnumIter, DeriveActiveEnum, Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
 #[sea_orm(rs_type = "i32", db_type = "Integer")]
-#[serde(rename_all = "lowercase")]
 pub enum PackageRelatedEnum {
     #[sea_orm(num_value = 0)]
     Conflicts,
@@ -41,3 +40,24 @@ pub struct FullPackage {
     related: Vec<(PackageRelatedEnum, String)>,
     files: Vec<String>,
 }
+
+#[derive(Clone, Debug)]
+pub struct RieterDb {
+    conn: DatabaseConnection,
+    pub pkg: query::PackageQuery,
+    pub repo: query::RepoQuery,
+}
+
+impl RieterDb {
+    pub async fn connect<C: Into<ConnectOptions>>(opt: C) -> Result<Self> {
+        let db = Database::connect(opt).await?;
+
+        Migrator::up(&db, None).await?;
+
+        Ok(Self {
+            conn: db.clone(),
+            pkg: query::PackageQuery::new(db.clone()),
+            repo: query::RepoQuery::new(db.clone()),
+        })
+    }
+}

server/src/db/query/mod.rs

@@ -1,4 +1,7 @@
-pub mod package;
-pub mod repo;
+mod package;
+mod repo;
+
+pub use package::PackageQuery;
+pub use repo::RepoQuery;
 
 type Result<T> = std::result::Result<T, sea_orm::DbErr>;

server/src/db/query/package.rs

@@ -1,54 +1,42 @@
-use sea_orm::{sea_query::IntoCondition, *};
-use serde::Deserialize;
+use sea_orm::*;
 
 use crate::db::*;
 
-#[derive(Deserialize)]
-pub struct Filter {
-    repo: Option<i32>,
-    arch: Option<String>,
-    name: Option<String>,
+#[derive(Clone, Debug)]
+pub struct PackageQuery {
+    conn: DatabaseConnection,
 }
 
-impl IntoCondition for Filter {
-    fn into_condition(self) -> Condition {
-        Condition::all()
-            .add_option(self.repo.map(|repo| package::Column::RepoId.eq(repo)))
-            .add_option(self.arch.map(|arch| package::Column::Arch.eq(arch)))
-            .add_option(
-                self.name
-                    .map(|name| package::Column::Name.like(format!("%{}%", name))),
-            )
+impl PackageQuery {
+    pub fn new(conn: DatabaseConnection) -> Self {
+        Self { conn }
     }
-}
 
-pub async fn page(
-    conn: &DbConn,
+    pub async fn page(
+        &self,
         per_page: u64,
         page: u64,
-    filter: Filter,
-) -> super::Result<(u64, Vec<package::Model>)> {
+    ) -> super::Result<(u64, Vec<package::Model>)> {
         let paginator = Package::find()
-        .filter(filter)
             .order_by_asc(package::Column::Id)
-        .paginate(conn, per_page);
+            .paginate(&self.conn, per_page);
         let packages = paginator.fetch_page(page).await?;
         let total_pages = paginator.num_pages().await?;
 
         Ok((total_pages, packages))
-}
+    }
 
-pub async fn by_id(conn: &DbConn, id: i32) -> Result<Option<package::Model>> {
-    package::Entity::find_by_id(id).one(conn).await
-}
+    pub async fn by_id(&self, id: i32) -> Result<Option<package::Model>> {
+        package::Entity::find_by_id(id).one(&self.conn).await
+    }
 
-pub async fn by_fields(
-    conn: &DbConn,
+    pub async fn by_fields(
+        &self,
         repo_id: i32,
         name: &str,
         version: Option<&str>,
         arch: &str,
-) -> Result<Option<package::Model>> {
+    ) -> Result<Option<package::Model>> {
         let mut query = Package::find()
             .filter(package::Column::RepoId.eq(repo_id))
             .filter(package::Column::Name.eq(name))
@@ -58,18 +46,18 @@ pub async fn by_fields(
             query = query.filter(package::Column::Version.eq(version));
         }
 
-    query.one(conn).await
-}
+        query.one(&self.conn).await
+    }
 
-pub async fn delete_with_arch(conn: &DbConn, repo_id: i32, arch: &str) -> Result<DeleteResult> {
+    pub async fn delete_with_arch(&self, repo_id: i32, arch: &str) -> Result<DeleteResult> {
         Package::delete_many()
             .filter(package::Column::RepoId.eq(repo_id))
             .filter(package::Column::Arch.eq(arch))
-        .exec(conn)
+            .exec(&self.conn)
             .await
-}
+    }
 
-pub async fn insert(conn: &DbConn, repo_id: i32, pkg: crate::repo::package::Package) -> Result<()> {
+    pub async fn insert(&self, repo_id: i32, pkg: crate::repo::package::Package) -> Result<()> {
         let info = pkg.info;
 
         let model = package::ActiveModel {
@@ -90,7 +78,7 @@ pub async fn insert(conn: &DbConn, repo_id: i32, pkg: crate::repo::package::Pack
             sha256_sum: Set(info.sha256sum),
         };
 
-    let pkg_entry = model.insert(conn).await?;
+        let pkg_entry = model.insert(&self.conn).await?;
 
         // Insert all the related tables
         PackageLicense::insert_many(info.licenses.iter().map(|s| package_license::ActiveModel {
@@ -98,7 +86,7 @@ pub async fn insert(conn: &DbConn, repo_id: i32, pkg: crate::repo::package::Pack
             name: Set(s.to_string()),
         }))
         .on_empty_do_nothing()
-    .exec(conn)
+        .exec(&self.conn)
         .await?;
 
         PackageGroup::insert_many(info.groups.iter().map(|s| package_group::ActiveModel {
@@ -106,7 +94,7 @@ pub async fn insert(conn: &DbConn, repo_id: i32, pkg: crate::repo::package::Pack
             name: Set(s.to_string()),
         }))
         .on_empty_do_nothing()
-    .exec(conn)
+        .exec(&self.conn)
         .await?;
 
         let related = info
@@ -144,48 +132,45 @@ pub async fn insert(conn: &DbConn, repo_id: i32, pkg: crate::repo::package::Pack
             package_id: Set(pkg_entry.id),
             r#type: Set(t),
             name: Set(s.to_string()),
-    }))
-    .on_empty_do_nothing()
-    .exec(conn)
-    .await?;
+        }));
 
         PackageFile::insert_many(pkg.files.iter().map(|s| package_file::ActiveModel {
             package_id: Set(pkg_entry.id),
             path: Set(s.display().to_string()),
         }))
         .on_empty_do_nothing()
-    .exec(conn)
+        .exec(&self.conn)
         .await?;
 
         Ok(())
-}
+    }
 
-pub async fn full(conn: &DbConn, id: i32) -> Result<Option<FullPackage>> {
-    if let Some(entry) = by_id(conn, id).await? {
+    pub async fn full(&self, id: i32) -> Result<Option<FullPackage>> {
+        if let Some(entry) = self.by_id(id).await? {
             let licenses = entry
                 .find_related(PackageLicense)
-            .all(conn)
+                .all(&self.conn)
                 .await?
                 .into_iter()
                 .map(|e| e.name)
                 .collect();
             let groups = entry
                 .find_related(PackageGroup)
-            .all(conn)
+                .all(&self.conn)
                 .await?
                 .into_iter()
                 .map(|e| e.name)
                 .collect();
             let related = entry
                 .find_related(PackageRelated)
-            .all(conn)
+                .all(&self.conn)
                 .await?
                 .into_iter()
                 .map(|e| (e.r#type, e.name))
                 .collect();
             let files = entry
                 .find_related(PackageFile)
-            .all(conn)
+                .all(&self.conn)
                 .await?
                 .into_iter()
                 .map(|e| e.path)
@@ -201,4 +186,5 @@ pub async fn full(conn: &DbConn, id: i32) -> Result<Option<FullPackage>> {
         } else {
             Ok(None)
         }
+    }
 }

server/src/db/query/repo.rs

@@ -1,49 +1,49 @@
-use sea_orm::{sea_query::IntoCondition, *};
+use sea_orm::*;
 
 use crate::db::*;
 
-#[derive(Deserialize)]
-pub struct Filter {
-    name: Option<String>,
+#[derive(Clone, Debug)]
+pub struct RepoQuery {
+    conn: DatabaseConnection,
 }
 
-impl IntoCondition for Filter {
-    fn into_condition(self) -> Condition {
-        Condition::all().add_option(self.name.map(|name| package::Column::Name.like(name)))
+impl RepoQuery {
+    pub fn new(conn: DatabaseConnection) -> Self {
+        Self { conn }
     }
-}
 
-pub async fn page(conn: &DbConn, per_page: u64, page: u64) -> Result<(u64, Vec<repo::Model>)> {
+    pub async fn page(&self, per_page: u64, page: u64) -> Result<(u64, Vec<repo::Model>)> {
         let paginator = Repo::find()
             .order_by_asc(repo::Column::Id)
-        .paginate(conn, per_page);
+            .paginate(&self.conn, per_page);
         let repos = paginator.fetch_page(page).await?;
         let total_pages = paginator.num_pages().await?;
 
         Ok((total_pages, repos))
-}
+    }
 
-pub async fn by_id(conn: &DbConn, id: i32) -> Result<Option<repo::Model>> {
-    repo::Entity::find_by_id(id).one(conn).await
-}
+    pub async fn by_id(&self, id: i32) -> Result<Option<repo::Model>> {
+        repo::Entity::find_by_id(id).one(&self.conn).await
+    }
 
-pub async fn by_name(conn: &DbConn, name: &str) -> Result<Option<repo::Model>> {
+    pub async fn by_name(&self, name: &str) -> Result<Option<repo::Model>> {
         Repo::find()
             .filter(repo::Column::Name.eq(name))
-        .one(conn)
+            .one(&self.conn)
             .await
-}
+    }
 
-pub async fn insert(
-    conn: &DbConn,
+    pub async fn insert(
+        &self,
         name: &str,
         description: Option<&str>,
-) -> Result<InsertResult<repo::ActiveModel>> {
+    ) -> Result<InsertResult<repo::ActiveModel>> {
         let model = repo::ActiveModel {
             id: NotSet,
             name: Set(String::from(name)),
             description: Set(description.map(String::from)),
         };
 
-    Repo::insert(model).exec(conn).await
+        Repo::insert(model).exec(&self.conn).await
+    }
 }

server/src/main.rs

@@ -22,7 +22,7 @@ pub struct Config {
 pub struct Global {
     config: Config,
     repo_manager: Arc<RwLock<RepoGroupManager>>,
-    db: sea_orm::DbConn,
+    db: db::RieterDb,
 }
 
 #[tokio::main]

server/src/repo/mod.rs

@@ -5,7 +5,7 @@ pub use manager::RepoGroupManager;
 
 use std::path::PathBuf;
 
-use axum::body::{Body};
+use axum::body::{Body, BodyDataStream};
 use axum::extract::{Path, State};
 use axum::http::Request;
 use axum::http::StatusCode;
@@ -21,8 +21,6 @@ use tower_http::services::{ServeDir, ServeFile};
 use tower_http::validate_request::ValidateRequestHeaderLayer;
 use uuid::Uuid;
 
-use crate::db;
-
 const DB_FILE_EXTS: [&str; 4] = [".db", ".files", ".db.tar.gz", ".files.tar.gz"];
 
 pub fn router(api_key: &str) -> Router<crate::Global> {
@@ -130,31 +128,26 @@ async fn post_package_archive(
             tracing::info!("Added '{}' to repository '{}'", pkg.file_name(), repo);
 
             // Query the repo for its ID, or create it if it does not already exist
-            let res = db::query::repo::by_name(&global.db, &repo).await?;
+            let res = global.db.repo.by_name(&repo).await?;
 
             let repo_id = if let Some(repo_entity) = res {
                 repo_entity.id
             } else {
-                db::query::repo::insert(&global.db, &repo, None)
-                    .await?
-                    .last_insert_id
+                global.db.repo.insert(&repo, None).await?.last_insert_id
             };
 
             // If the package already exists in the database, we remove it first
-            let res = db::query::package::by_fields(
-                &global.db,
-                repo_id,
-                &pkg.info.name,
-                None,
-                &pkg.info.arch,
-            )
+            let res = global
+                .db
+                .pkg
+                .by_fields(repo_id, &pkg.info.name, None, &pkg.info.arch)
                 .await?;
 
             if let Some(entry) = res {
                 entry.delete(&global.db).await?;
             }
 
-            db::query::package::insert(&global.db, repo_id, pkg).await?;
+            global.db.pkg.insert(repo_id, pkg).await?;
 
             Ok(())
         }
@@ -179,7 +172,7 @@ async fn delete_repo(
             .await??;
 
     if repo_removed {
-        let res = db::query::repo::by_name(&global.db, &repo).await?;
+        let res = global.db.repo.by_name(&repo).await?;
 
         if let Some(repo_entry) = res {
             repo_entry.delete(&global.db).await?;
@@ -210,10 +203,10 @@ async fn delete_arch_repo(
     .await??;
 
     if repo_removed {
-        let res = db::query::repo::by_name(&global.db, &repo).await?;
+        let res = global.db.repo.by_name(&repo).await?;
 
         if let Some(repo_entry) = res {
-            db::query::package::delete_with_arch(&global.db, repo_entry.id, &arch).await?;
+            global.db.pkg.delete_with_arch(repo_entry.id, &arch).await?;
         }
         tracing::info!("Removed architecture '{}' from repository '{}'", arch, repo);
 
@@ -236,11 +229,13 @@ async fn delete_package(
     .await??;
 
     if let Some((name, version, release, arch)) = res {
-        let res = db::query::repo::by_name(&global.db, &repo).await?;
+        let res = global.db.repo.by_name(&repo).await?;
 
         if let Some(repo_entry) = res {
-            let res = db::query::package::by_fields(
-                &global.db,
+            let res = global
+                .db
+                .pkg
+                .by_fields(
                     repo_entry.id,
                     &name,
                     Some(&format!("{}-{}", version, release)),