packie/src/repo/mod.rs

mod download;
mod serve;

#[doc(inline)]
pub use download::DownloadError;
#[doc(inline)]
pub use serve::RepoServeDir;

use crate::{
    Packie,
    package::{AbsPkgIdent, PkgIdent, PkgManifest, PkgSpec},
};
use download::Download;
use itertools::Itertools;
use rusqlite::{OpenFlags, Row, params};
use rust_i18n::t;
use rustc_hash::FxHashMap;
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use url::Url;

impl super::Packie {
    /// Creates a session for syncing repositories.
    pub fn sync_repo(&mut self) -> SyncSession<'_> {
        SyncSession::new(self)
    }

    /// Create a session for querying repositories.
    pub fn query_repo(&mut self) -> QuerySession {
        QuerySession::new(self)
    }

    /// Download a package.
    pub fn download_package(
        &self,
        repo: &str,
        pkg_ident: &PkgIdent,
        mut on_event: impl FnMut(DownloadPackageEvent),
    ) -> Result<PathBuf, RepoError> {
        let repo_cache_dir = self.cache.repo_dir(repo);

        let repo = self.config.repos.get(repo).ok_or(RepoError::NoSuchRepo)?;
        let repo_db = repo_cache_dir.join(RepoDb::FILENAME);
        let mut repo_db = RepoDb::open_ro(repo_db).map_err(RepoError::Database)?;
        let repo_package = repo_db
            .select_package_by_ident(pkg_ident)
            .map_err(|_| RepoError::NoSuchPackage)?;

        let standard_filename = format!("{pkg_ident}.pkg");

        let download_path = repo_cache_dir.join(&standard_filename);

        for repo_url in repo.urls() {
            let download_url = match repo_url.join(&standard_filename) {
                Ok(x) => x,
                Err(_) => {
                    on_event(DownloadPackageEvent::Error(
                        repo_url.clone(),
                        RepoError::Download(DownloadError::InvalidUrl),
                    ));
                    continue;
                }
            };
            let download_result = Download::new(download_url.to_string(), download_path.clone())
                .on_progress(|sum, total| {
                    on_event(DownloadPackageEvent::Progress(repo_url.clone(), sum, total))
                })
                .checksum_required(repo_package.checksum.clone())
                .run();
            if let Err(err) = download_result {
                on_event(DownloadPackageEvent::Error(
                    download_url.clone(),
                    RepoError::Download(err),
                ));
                continue;
            }
            return Ok(download_path);
        }

        Err(RepoError::AllTriesFailed)
    }
}

#[derive(Debug)]
pub struct RepoDb(rusqlite::Connection);
impl RepoDb {
    const FILENAME: &str = "repo.db";

    const SCHEMA: &str = r#"
CREATE TABLE IF NOT EXISTS "package"(
    "pkg_name" TEXT NOT NULL,
    "pkg_version" TEXT NOT NULL,
    "pkg_arch" TEXT NOT NULL,
    "pkg_manifest" TEXT NOT NULL,
    "download_size" INTEGER NOT NULL,
    "installed_size" INTEGER NOT NULL,
    "checksum" TEXT NOT NULL,
    PRIMARY KEY ("pkg_name", "pkg_version", "pkg_arch")
);
CREATE TABLE IF NOT EXISTS "abspkg"(
    "id" INTEGER PRIMARY KEY AUTOINCREMENT,
    "abspkg_ident" TEXT NOT NULL,
    "provider" TEXT NOT NULL
);
"#;

    pub fn open_rw<P: AsRef<Path>>(path: P) -> rusqlite::Result<Self> {
        let conn = rusqlite::Connection::open(path)?;
        conn.execute_batch(Self::SCHEMA)?;
        Ok(Self(conn))
    }

    pub fn open_ro<P: AsRef<Path>>(path: P) -> rusqlite::Result<Self> {
        rusqlite::Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY).map(Self)
    }

    pub fn select_package_by_ident(
        &mut self,
        pkg_ident: &PkgIdent,
    ) -> rusqlite::Result<RepoPackage> {
        self.0.prepare_cached(
            r#"SELECT * FROM "package" WHERE "pkg_name" = ?1 AND "pkg_version" = ?2 AND "pkg_arch" = ?3"#,
        )?.query_one(params![pkg_ident.name, pkg_ident.version.to_string(), pkg_ident.arch], map_repo_package)
    }

    pub fn select_package_by_name(&mut self, pkgname: &str) -> rusqlite::Result<Vec<RepoPackage>> {
        let mut stmt = self
            .0
            .prepare_cached(r#"SELECT * FROM "package" WHERE "pkg_name" = ?1"#)?;
        let mapped_rows = stmt.query_map(params![pkgname], map_repo_package)?;
        let mut ret = Vec::with_capacity(16);
        for mr in mapped_rows {
            ret.push(mr?);
        }
        Ok(ret)
    }

    pub fn select_abspkg_by_ident(
        &mut self,
        ident: &AbsPkgIdent,
    ) -> rusqlite::Result<Vec<(AbsPkgIdent, PkgIdent)>> {
        let mut stmt = self
            .0
            .prepare_cached(r#"SELECT * FROM "abspkg" WHERE "abspkg_ident" = ?1"#)?;
        let rows = stmt.query_map(params![ident.to_string()], map_abspkg)?;
        let mut ret = Vec::with_capacity(16);
        for row in rows {
            ret.push(row?);
        }
        Ok(ret)
    }

    pub fn for_each_package<E: From<rusqlite::Error>>(
        &mut self,
        mut f: impl FnMut(RepoPackage) -> Result<(), E>,
    ) -> Result<(), E> {
        let mut stmt = self.0.prepare_cached("SELECT * FROM \"package\"")?;
        let mapped_rows = stmt.query_map(params![], map_repo_package)?;
        for mr in mapped_rows {
            f(mr?)?;
        }
        Ok(())
    }

    pub fn insert_package(&mut self, package: &RepoPackage) -> rusqlite::Result<()> {
        let pkg_manifest = serde_json::to_string(&package.pkg_manifest)
            .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
        self.0
            .prepare_cached(
                r#"
                INSERT INTO "package"(
                    "pkg_name", "pkg_version", "pkg_arch", "pkg_manifest",
                    "download_size", "installed_size", "checksum"
                )
                VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7)
                "#,
            )?
            .execute(params![
                package.pkg_manifest.name,
                package.pkg_manifest.version.to_string(),
                package.pkg_manifest.arch,
                pkg_manifest,
                package.download_size as i64,
                package.installed_size as i64,
                package.checksum
            ])
            .map(|_| ())
    }

    pub fn insert_abspkg(&mut self, abs: &AbsPkgIdent, real: &PkgIdent) -> rusqlite::Result<()> {
        self.0
            .prepare_cached(r#"INSERT INTO "abspkg"("abspkg_ident", "provider") VALUES(?1, ?2)"#)?
            .execute(params![abs.to_string(), real.to_string()])?;
        Ok(())
    }

    pub fn delete_package_by_ident(&mut self, ident: &PkgIdent) -> rusqlite::Result<bool> {
        self.0
            .prepare_cached(
                r#"DELETE FROM "package" WHERE "pkg_name" = ?1 AND "pkg_version" = ?2 AND "pkg_arch" = ?3"#,
            )?
            .execute(params![ident.name, ident.version.to_string(), ident.arch])
            .map(|x| x > 0)
    }

    pub fn delete_abspkg_by_provider(&mut self, ident: &PkgIdent) -> rusqlite::Result<bool> {
        self.0
            .prepare_cached(r#"DELETE FROM "abspkg" WHERE "provider" = ?1"#)?
            .execute(params![ident.to_string()])
            .map(|x| x > 0)
    }
}

fn map_repo_package(row: &Row) -> rusqlite::Result<RepoPackage> {
    let pkg_manifest: String = row.get("pkg_manifest")?;
    let pkg_manifest = serde_json::from_str(&pkg_manifest).map_err(from_sql_error)?;
    Ok(RepoPackage {
        pkg_manifest,
        download_size: row.get::<_, i64>("download_size")? as u64,
        installed_size: row.get::<_, i64>("installed_size")? as u64,
        checksum: row.get("checksum")?,
    })
}

fn map_abspkg(row: &Row) -> rusqlite::Result<(AbsPkgIdent, PkgIdent)> {
    let provider = row
        .get::<_, String>("provider")?
        .parse()
        .map_err(from_sql_error)?;
    let abspkg_name = row
        .get::<_, String>("abspkg_ident")?
        .parse()
        .map_err(from_sql_error)?;
    Ok((abspkg_name, provider))
}

fn from_sql_error<E: std::error::Error + Send + Sync + 'static>(error: E) -> rusqlite::Error {
    rusqlite::Error::FromSqlConversionFailure(0, rusqlite::types::Type::Text, Box::new(error))
}

/// A record representing to a package in a repository.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RepoPackage {
    /// Package manifest.
    pub pkg_manifest: PkgManifest,

    /// Download size of the package.
    pub download_size: u64,

    /// Installed size of the package.
    pub installed_size: u64,

    /// Checksum of the package file.
    pub checksum: String,
}

/// Session of syncing repositories.
pub struct SyncSession<'a, E = fn(SyncEvent)> {
    packie: &'a mut Packie,
    repo: Option<String>,
    on_event: E,
}
impl<'a> SyncSession<'a> {
    pub fn new(packie: &'a mut Packie) -> Self {
        Self {
            packie,
            repo: None,
            on_event: |_| (),
        }
    }
}
impl<'a, E> SyncSession<'a, E> {
    pub fn repo(mut self, repo: String) -> Self {
        self.repo = Some(repo);
        self
    }

    pub fn on_event<E2>(self, on_event: E2) -> SyncSession<'a, E2> {
        SyncSession {
            packie: self.packie,
            repo: self.repo,
            on_event,
        }
    }
}
impl<E: FnMut(SyncEvent)> SyncSession<'_, E> {
    pub fn run(mut self) {
        let repos = match self.repo.clone() {
            Some(x) => vec![x],
            None => self.packie.config.repos.keys().cloned().collect(),
        };
        for repo_name in repos {
            let Some(repo) = self.packie.config.repos.get(&repo_name).cloned() else {
                self.raise_error(Some(repo_name), None, RepoError::NoSuchRepo);
                continue;
            };
            (self.on_event)(SyncEvent::BeginRepo(repo_name.clone()));
            for url in repo.urls() {
                match self.sync_url(&repo_name, &url) {
                    Ok(()) => break,
                    Err(err) => self.raise_error(Some(repo_name.clone()), Some(url.clone()), err),
                }
            }
        }
    }

    fn sync_url(&mut self, repo_name: &str, url: &Url) -> Result<(), RepoError> {
        let repo_db_url = url
            .join(RepoDb::FILENAME)
            .map_err(|_| DownloadError::InvalidUrl)
            .map_err(RepoError::Download)?;

        let repo_cache_dir = self.packie.cache.repo_dir(repo_name);
        Download::new(
            repo_db_url.to_string(),
            repo_cache_dir.join(RepoDb::FILENAME),
        )
        .no_existence_check()
        .on_progress(Box::new(|sum, total| {
            self.raise_progress(repo_name.into(), repo_db_url.clone(), sum, total)
        }))
        .run()
        .map_err(RepoError::Download)?;

        Ok(())
    }

    fn raise_error(&mut self, repo: Option<String>, url: Option<Url>, error: RepoError) {
        (self.on_event)(SyncEvent::Error(repo, url, error));
    }

    fn raise_progress(&mut self, repo: String, url: Url, downloaded_bytes: u64, total_bytes: u64) {
        (self.on_event)(SyncEvent::Progress(SyncProgress {
            repo: Some(repo),
            url: Some(url),
            downloaded_bytes,
            total_bytes,
        }));
    }
}

/// An event during running [`SyncSession`], which may report progress changes or errors.
#[derive(Debug)]
pub enum SyncEvent {
    BeginRepo(String),
    Progress(SyncProgress),
    Error(Option<String>, Option<Url>, RepoError),
}

/// A progress while syncing repositories.
#[derive(Debug)]
pub struct SyncProgress {
    pub repo: Option<String>,
    pub url: Option<Url>,
    pub downloaded_bytes: u64,
    pub total_bytes: u64,
}

/// An event occurred while downloading a package.
#[derive(Debug)]
pub enum DownloadPackageEvent {
    Progress(Url, u64, u64),
    Error(Url, RepoError),
}

#[derive(Debug)]
pub struct QuerySession {
    repos: FxHashMap<String, RepoDb>,
}
impl QuerySession {
    pub fn new(packie: &mut Packie) -> Self {
        let mut repos = FxHashMap::default();
        for name in packie.config.repos.keys() {
            let repo_db = packie.cache.repo_dir(name).join(RepoDb::FILENAME);
            let Ok(repo_db) = RepoDb::open_ro(repo_db) else {
                continue;
            };
            repos.insert(name.into(), repo_db);
        }
        Self { repos }
    }

    pub fn search_package(&mut self, pkgspec: &PkgSpec) -> Vec<(String, RepoPackage)> {
        let mut all: Vec<(String, RepoPackage)> = Vec::new();
        for (name, db) in self.repos.iter_mut() {
            for pkg in db.select_package_by_name(&pkgspec.name).unwrap_or_default() {
                if pkgspec.matches(&pkg.pkg_manifest.pkg_ident()) {
                    all.push((name.into(), pkg));
                }
            }
            let abspkg_ident = AbsPkgIdent {
                name: pkgspec.name.clone(),
                arch: pkgspec.arch.clone(),
            };
            for (_, provider) in db.select_abspkg_by_ident(&abspkg_ident).unwrap_or_default() {
                let Ok(real) = db.select_package_by_ident(&provider) else {
                    continue;
                };
                all.push((name.into(), real));
            }
        }
        all.into_iter()
            .unique_by(|(repo, pkg)| (repo.to_string(), pkg.pkg_manifest.pkg_ident()))
            .collect()
    }
}

#[derive(Debug, thiserror::Error)]
pub enum RepoError {
    #[error("{}", t!("repo.RepoError.NoSuchRepo"))]
    NoSuchRepo,

    #[error("{}", t!("repo.RepoError.NoSuchPackage"))]
    NoSuchPackage,

    #[error("{}", t!("repo.RepoError.Database", error = 0))]
    Database(rusqlite::Error),

    #[error("{0}")]
    Download(DownloadError),

    #[error("{}", t!("repo.RepoError.AllTriesFailed"))]
    AllTriesFailed,
}