feat(package): handle replaced pkg_id

Author: QaidVoid

soar-cli/src/state.rs

@@ -4,17 +4,25 @@ use std::{
     sync::{Arc, Mutex, RwLockReadGuard},
 };
 
+use nu_ansi_term::Color::{Blue, Green, Magenta, Red};
 use once_cell::sync::OnceCell;
-use rusqlite::Connection;
+use rusqlite::{params, Connection};
 use soar_core::{
-    config::{get_config, Config},
+    config::{get_config, Config, Repository},
     constants::CORE_MIGRATIONS,
-    database::{connection::Database, migration::MigrationManager},
+    database::{
+        connection::Database,
+        migration::MigrationManager,
+        models::FromRow,
+        packages::{FilterCondition, PackageQueryBuilder},
+    },
     error::{ErrorContext, SoarError},
     metadata::fetch_metadata,
     SoarResult,
 };
-use tracing::error;
+use tracing::{error, info};
+
+use crate::utils::Colored;
 
 #[derive(Clone)]
 pub struct AppState {
@@ -50,20 +58,103 @@ impl AppState {
         for repo in &self.inner.config.repositories {
             let repo_clone = repo.clone();
             let task = tokio::task::spawn(async move { fetch_metadata(repo_clone, force).await });
-            tasks.push(task);
+            tasks.push((task, repo));
         }
 
-        for task in tasks {
-            if let Err(err) = task
+        for (task, repo) in tasks {
+            match task
                 .await
                 .map_err(|err| SoarError::Custom(format!("Join handle error: {}", err)))?
             {
-                if !matches!(err, SoarError::FailedToFetchRemote(_)) {
-                    return Err(err);
+                Ok(Some(etag)) => {
+                    self.validate_packages(&repo, &etag).await?;
+                    info!("[{}] Repository synced", Colored(Magenta, &repo.name));
+                }
+                Err(err) => {
+                    if !matches!(err, SoarError::FailedToFetchRemote(_)) {
+                        return Err(err);
+                    }
+                    error!("{err}");
                 }
-                error!("{err}");
+                _ => {}
             };
         }
+
+        Ok(())
+    }
+
+    async fn validate_packages(&self, repo: &Repository, etag: &str) -> SoarResult<()> {
+        let core_db = self.core_db()?;
+        let repo_name = repo.name.clone();
+
+        let repo_path = repo.get_path()?;
+        let metadata_db = repo_path.join("metadata.db");
+
+        let repo_db = Arc::new(Mutex::new(Connection::open(&metadata_db)?));
+
+        let installed_packages = PackageQueryBuilder::new(core_db.clone())
+            .where_and("repo_name", FilterCondition::Eq(repo_name.to_string()))
+            .load_installed()?;
+
+        struct RepoPackage {
+            pkg_id: String,
+        }
+
+        impl FromRow for RepoPackage {
+            fn from_row(row: &rusqlite::Row) -> rusqlite::Result<Self> {
+                Ok(Self {
+                    pkg_id: row.get("pkg_id")?,
+                })
+            }
+        }
+
+        for pkg in installed_packages.items {
+            let repo_package: Vec<RepoPackage> = PackageQueryBuilder::new(repo_db.clone())
+                .select(&["pkg_id"])
+                .where_and("pkg_id", FilterCondition::Eq(pkg.pkg_id.clone()))
+                .where_and("repo_name", FilterCondition::Eq(pkg.repo_name.clone()))
+                .load()?
+                .items;
+
+            if repo_package.is_empty() {
+                let replaced_by: Vec<RepoPackage> = PackageQueryBuilder::new(repo_db.clone())
+                    .select(&["pkg_id"])
+                    .where_and("repo_name", FilterCondition::Eq(pkg.repo_name))
+                    // there's no easy way to do this, could create scalar SQL
+                    // function, but this is enough for now
+                    .where_and(
+                        &format!("EXISTS (SELECT 1 FROM json_each(p.replaces) WHERE json_each.value = '{}')", pkg.pkg_id),
+                        FilterCondition::None,
+                    )
+                    .limit(1)
+                    .load()?
+                    .items;
+
+                if !replaced_by.is_empty() {
+                    let new_pkg_id = &replaced_by.first().unwrap().pkg_id;
+                    info!(
+                        "[{}] {} is replaced by {} in {}",
+                        Colored(Blue, "Note"),
+                        Colored(Red, &pkg.pkg_id),
+                        Colored(Green, new_pkg_id),
+                        Colored(Magenta, &repo_name)
+                    );
+
+                    let conn = core_db.lock()?;
+                    conn.execute(
+                        "UPDATE packages SET pkg_id = ? WHERE pkg_id = ? AND repo_name = ?",
+                        params![new_pkg_id, pkg.pkg_id, repo_name],
+                    )?;
+                }
+            }
+        }
+
+        let conn = repo_db.lock()?;
+        conn.execute(
+            "UPDATE repository SET name = ?, etag = ?",
+            params![repo.name, etag],
+        )?;
+
         Ok(())
     }
 

soar-core/migrations/metadata/V1_initial.sql

@@ -58,6 +58,7 @@ CREATE TABLE packages (
   provides JSONB,
   snapshots JSONB,
   repology JSONB,
+  replaces JSONB,
   download_count INTEGER,
   download_count_week INTEGER,
   download_count_month INTEGER,

soar-core/src/database/connection.rs

@@ -42,20 +42,15 @@ impl Database {
         Ok(Database { conn })
     }
 
-    pub fn from_remote_metadata(
-        &self,
-        metadata: &[RemotePackage],
-        repo_name: &str,
-        etag: &str,
-    ) -> Result<()> {
+    pub fn from_remote_metadata(&self, metadata: &[RemotePackage], repo_name: &str) -> Result<()> {
         let mut guard = self.conn.lock().unwrap();
         let _: String = guard.query_row("PRAGMA journal_mode = WAL", [], |row| row.get(0))?;
 
         let tx = guard.transaction()?;
         {
             let statements = DbStatements::new(&tx)?;
             let mut repo = PackageRepository::new(&tx, statements, repo_name);
-            repo.import_packages(&metadata, etag)?;
+            repo.import_packages(&metadata)?;
         }
         tx.commit()?;
         Ok(())

soar-core/src/database/models.rs

@@ -81,6 +81,7 @@ pub struct Package {
     pub download_count_month: Option<u64>,
     pub download_count_week: Option<u64>,
     pub maintainers: Option<Vec<Maintainer>>,
+    pub replaces: Option<Vec<String>>,
 }
 
 impl FromRow for Package {
@@ -111,6 +112,7 @@ impl FromRow for Package {
         let provides = parse_provides("provides")?;
         let snapshots = parse_json_vec("snapshots")?;
         let repology = parse_json_vec("repology")?;
+        let replaces = parse_json_vec("replaces")?;
 
         Ok(Package {
             id: row.get("id")?,
@@ -158,6 +160,7 @@ impl FromRow for Package {
             download_count_month: row.get("download_count_month")?,
             repo_name: row.get("repo_name")?,
             maintainers,
+            replaces,
         })
     }
 }
@@ -384,6 +387,7 @@ pub struct RemotePackage {
 
     pub repology: Option<Vec<String>>,
     pub snapshots: Option<Vec<String>>,
+    pub replaces: Option<Vec<String>>,
 }
 
 fn should_create_original_symlink_impl(provides: Option<&Vec<PackageProvide>>) -> bool {

soar-core/src/database/packages/query.rs

@@ -240,6 +240,7 @@ impl PackageQueryBuilder {
                         "json(provides) AS provides",
                         "json(snapshots) AS snapshots",
                         "json(repology) AS repology",
+                        "json(replaces) AS replaces",
                         "download_count",
                         "download_count_week",
                         "download_count_month",

soar-core/src/database/repository.rs

@@ -18,10 +18,12 @@ impl<'a> PackageRepository<'a> {
         }
     }
 
-    pub fn import_packages(&mut self, metadata: &[RemotePackage], etag: &str) -> Result<()> {
+    pub fn import_packages(&mut self, metadata: &[RemotePackage]) -> Result<()> {
         self.statements
             .repo_insert
-            .execute(params![self.repo_name, etag])?;
+            // to prevent incomplete sync, etag should only be updated once
+            // all checks are done
+            .execute(params![self.repo_name, ""])?;
 
         for package in metadata {
             self.insert_package(package)?;
@@ -64,6 +66,7 @@ impl<'a> PackageRepository<'a> {
         let categories = serde_json::to_string(&package.categories).unwrap();
         let snapshots = serde_json::to_string(&package.snapshots).unwrap();
         let repology = serde_json::to_string(&package.repology).unwrap();
+        let replaces = serde_json::to_string(&package.replaces).unwrap();
 
         let provides = package.provides.clone().map(|vec| {
             vec.iter()
@@ -121,6 +124,7 @@ impl<'a> PackageRepository<'a> {
             provides,
             snapshots,
             repology,
+            replaces,
             package.download_count,
             package.download_count_week,
             package.download_count_month

soar-core/src/database/statements.rs

@@ -34,15 +34,15 @@ impl<'a> DbStatements<'a> {
                     bsum, shasum, icon, desktop, appstream, homepages, notes,
                     source_urls, tags, categories, build_id, build_date,
                     build_action, build_script, build_log, provides, snapshots,
-                    repology, download_count, download_count_week,
+                    repology, replaces, download_count, download_count_week,
                     download_count_month
                 )
                 VALUES
                 (
                     ?1, jsonb(?2), ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13,
                     jsonb(?14), ?15, ?16, ?17, ?18, jsonb(?19), ?20, ?21, ?22, ?23, ?24, ?25,
                     ?26, jsonb(?27), jsonb(?28), jsonb(?29), jsonb(?30), jsonb(?31), ?32, ?33, ?34, ?35, ?36,
-                    jsonb(?37), jsonb(?38), jsonb(?39), ?40, ?41, ?42
+                    jsonb(?37), jsonb(?38), jsonb(?39), jsonb(?40), ?41, ?42, ?43
                 )
                 ON CONFLICT (pkg_id, pkg_name, version) DO NOTHING",
             )?,

soar-core/src/metadata.rs

@@ -6,7 +6,7 @@ use std::{
 
 use futures::TryStreamExt;
 use reqwest::header::{self, HeaderMap};
-use rusqlite::{params, Connection};
+use rusqlite::Connection;
 use tracing::info;
 
 use crate::{
@@ -22,7 +22,6 @@ fn handle_json_metadata<P: AsRef<Path>>(
     metadata: &[RemotePackage],
     metadata_db: P,
     repo: &Repository,
-    etag: &str,
 ) -> SoarResult<()> {
     let metadata_db = metadata_db.as_ref();
     if metadata_db.exists() {
@@ -35,7 +34,7 @@ fn handle_json_metadata<P: AsRef<Path>>(
     manager.migrate_from_dir(METADATA_MIGRATIONS)?;
 
     let db = Database::new(metadata_db)?;
-    db.from_remote_metadata(metadata.as_ref(), &repo.name, &etag)?;
+    db.from_remote_metadata(metadata.as_ref(), &repo.name)?;
 
     Ok(())
 }
@@ -69,7 +68,7 @@ pub async fn fetch_public_key<P: AsRef<Path>>(
     Ok(())
 }
 
-pub async fn fetch_metadata(repo: Repository, force: bool) -> SoarResult<()> {
+pub async fn fetch_metadata(repo: Repository, force: bool) -> SoarResult<Option<String>> {
     let repo_path = repo.get_path()?;
     let metadata_db = repo_path.join("metadata.db");
 
@@ -91,7 +90,7 @@ pub async fn fetch_metadata(repo: Repository, force: bool) -> SoarResult<()> {
             .with_context(|| format!("reading file metadata from {}", metadata_db.display()))?;
         if let Ok(created) = file_info.created() {
             if repo.sync_interval() >= created.elapsed()?.as_millis() as u128 {
-                return Ok(());
+                return Ok(None);
             }
         }
     }
@@ -118,7 +117,7 @@ pub async fn fetch_metadata(repo: Repository, force: bool) -> SoarResult<()> {
             Some(remote_etag) => {
                 let remote_etag = remote_etag.to_str().unwrap();
                 if !force && etag == remote_etag {
-                    return Ok(());
+                    return Ok(None);
                 }
                 remote_etag.to_string()
             }
@@ -153,11 +152,6 @@ pub async fn fetch_metadata(repo: Repository, force: bool) -> SoarResult<()> {
         if magic_bytes == SQLITE_MAGIC_BYTES {
             fs::rename(&tmp_path, &metadata_db)
                 .with_context(|| format!("renaming {} to {}", tmp_path, metadata_db.display()))?;
-            let conn = Connection::open(&metadata_db)?;
-            conn.execute(
-                "UPDATE repository SET name = ?, etag = ?",
-                params![repo.name, etag],
-            )?;
         } else {
             let tmp_file = File::open(&tmp_path)
                 .with_context(|| format!("opening temporary file {}", tmp_path))?;
@@ -169,7 +163,7 @@ pub async fn fetch_metadata(repo: Repository, force: bool) -> SoarResult<()> {
                 ))
             })?;
 
-            handle_json_metadata(&metadata, metadata_db, &repo, &etag)?;
+            handle_json_metadata(&metadata, metadata_db, &repo)?;
             fs::remove_file(tmp_path.clone())
                 .with_context(|| format!("removing temporary file {}", tmp_path))?;
         }
@@ -181,11 +175,6 @@ pub async fn fetch_metadata(repo: Repository, force: bool) -> SoarResult<()> {
         writer
             .write_all(&content)
             .with_context(|| format!("writing to metadata file {}", metadata_db.display()))?;
-        let conn = Connection::open(&metadata_db)?;
-        conn.execute(
-            "UPDATE repository SET name = ?, etag = ?",
-            params![repo.name, etag],
-        )?;
     } else {
         let remote_metadata: Vec<RemotePackage> =
             serde_json::from_slice(&content).map_err(|err| {
@@ -195,8 +184,8 @@ pub async fn fetch_metadata(repo: Repository, force: bool) -> SoarResult<()> {
                 ))
             })?;
 
-        handle_json_metadata(&remote_metadata, metadata_db, &repo, &etag)?;
+        handle_json_metadata(&remote_metadata, metadata_db, &repo)?;
     }
 
-    Ok(())
+    Ok(Some(etag))
 }