Move all sql queries to storage module

src/bin/scan-flake.rs

@@ -15,6 +15,8 @@ use flake_tracker::{
         FlakeUri,
     },
     storage::{
+        InputRow,
+        RevisionRow,
         Storage,
     },
 };
@@ -48,22 +50,17 @@ async fn main() -> Result<()> {
     let flake_metadata: FlakeMetadata = serde_json::from_slice(&flake_metadata_raw.stdout)
         .context("Failed to parse flake metadata")?;
 
-    sqlx::query("INSERT INTO revisions (revision_uri, flake_uri, nix_store_path, nar_hash, last_modified, tracker_last_scanned)
+    let revision_row = RevisionRow {
-        VALUES (?, ?, ?, ?, ?, ?)
+        revision_uri: flake_metadata.locked.flake_uri()?.clone(),
-        ON CONFLICT(revision_uri) DO UPDATE SET
+        flake_uri: Some(flake_metadata.resolved.flake_uri()?.clone()),
-            flake_uri=excluded.flake_uri,
+        nix_store_path: Some(flake_metadata.path.clone()),
-            nix_store_path=excluded.nix_store_path,
+        nar_hash: Some(flake_metadata.locked.narHash.clone()),
-            nar_hash=excluded.nar_hash,
+        last_modified: Some(flake_metadata.locked.lastModified.clone()),
-            last_modified=excluded.last_modified,
+        tracker_last_scanned: Some(scan_time.clone()),
-            tracker_last_scanned=excluded.tracker_last_scanned
+    };
-        ")
+
-        .bind(&flake_metadata.locked.flake_uri()?)
+    storage.set_revision(revision_row)
-        .bind(&flake_metadata.resolved.flake_uri()?)
+        .await?;
-        .bind(&flake_metadata.path)
-        .bind(&flake_metadata.locked.narHash)
-        .bind(&flake_metadata.locked.lastModified)
-        .bind(&scan_time)
-        .execute(&storage.db).await?;
 
     let locks_root_name = &flake_metadata.locks.root;
     let locks_root_node = flake_metadata.locks.nodes.get(locks_root_name)
@@ -75,29 +72,27 @@ async fn main() -> Result<()> {
             let locks_input_node = flake_metadata.locks.nodes.get(&locks_input_name)
                 .context("Failed to find lock of input")?;
 
-            sqlx::query("INSERT INTO inputs (revision_uri, input_name, locked_revision_uri, locked_flake_uri, locked_nar_hash, last_modified)
+            let input_row = InputRow {
-                VALUES (?, ?, ?, ?, ?, ?)
+                revision_uri: flake_metadata.locked.flake_uri()?.clone(),
-                ON CONFLICT(revision_uri, input_name) DO UPDATE SET
+                input_name: input_name.clone(),
-                    locked_revision_uri=excluded.locked_revision_uri,
+                locked_revision_uri: Some(locks_input_node.locked.clone().context("Unexpected missing lock")?.flake_uri()?),
-                    locked_flake_uri=excluded.locked_flake_uri,
+                locked_flake_uri: Some(locks_input_node.original.clone().context("Unexpected missing lock")?.flake_uri()?),
-                    locked_nar_hash=excluded.locked_nar_hash,
+                locked_nar_hash: Some(locks_input_node.locked.clone().context("Unexpected missing lock")?.narHash),
-                    last_modified=excluded.last_modified
+                last_modified: Some(locks_input_node.locked.clone().context("Unexpected missing lock")?.lastModified),
-                ")
+            };
-                .bind(flake_metadata.locked.flake_uri()?)
+            storage.set_input(input_row)
-                .bind(input_name)
+                .await?;
-                .bind(locks_input_node.locked.clone().context("Unexpected missing lock")?.flake_uri()?)
-                .bind(locks_input_node.original.clone().context("Unexpected missing lock")?.flake_uri()?)
-                .bind(locks_input_node.locked.clone().context("Unexpected missing lock")?.narHash)
-                .bind(locks_input_node.locked.clone().context("Unexpected missing lock")?.lastModified)
-                .execute(&storage.db).await?;
 
-            sqlx::query("INSERT INTO revisions (revision_uri, flake_uri)
+            let revision_row = RevisionRow {
-                VALUES (?, ?)
+                revision_uri: locks_input_node.locked.clone().context("Unexpected missing lock")?.flake_uri()?.clone(),
-                ON CONFLICT(revision_uri) DO NOTHING
+                flake_uri: Some(locks_input_node.original.clone().context("Unexpected missing lock")?.flake_uri()?.clone()),
-                ")
+                nix_store_path: None,
-                .bind(locks_input_node.locked.clone().context("Unexpected missing lock")?.flake_uri()?)
+                nar_hash: None,
-                .bind(locks_input_node.original.clone().context("Unexpected missing lock")?.flake_uri()?)
+                last_modified: None,
-                .execute(&storage.db).await?;
+                tracker_last_scanned: None,
+            };
+            storage.set_revision_exist(revision_row)
+                .await?;
         }
     }
 

src/storage.rs

@@ -12,7 +12,10 @@ use crate::utils::{
 use sqlx::{
     FromRow,
     SqlitePool,
-    sqlite::SqlitePoolOptions,
+    sqlite::{
+        SqlitePoolOptions,
+        SqliteQueryResult,
+    },
 };
 
 #[derive(Clone)]
@@ -44,6 +47,39 @@ impl Storage {
             .context("Failed to fetch data from database")
     }
 
+    pub async fn set_revision(&self, revision_row: RevisionRow) -> Result<SqliteQueryResult> {
+        sqlx::query("INSERT INTO revisions (revision_uri, flake_uri, nix_store_path, nar_hash, last_modified, tracker_last_scanned)
+            VALUES (?, ?, ?, ?, ?, ?)
+            ON CONFLICT(revision_uri) DO UPDATE SET
+                flake_uri=excluded.flake_uri,
+                nix_store_path=excluded.nix_store_path,
+                nar_hash=excluded.nar_hash,
+                last_modified=excluded.last_modified,
+                tracker_last_scanned=excluded.tracker_last_scanned
+            ")
+            .bind(&revision_row.revision_uri)
+            .bind(&revision_row.flake_uri)
+            .bind(&revision_row.nix_store_path)
+            .bind(&revision_row.nar_hash)
+            .bind(&revision_row.last_modified)
+            .bind(&revision_row.tracker_last_scanned)
+            .execute(&self.db)
+            .await
+            .context("Failed to execute database query")
+    }
+
+    pub async fn set_revision_exist(&self, revision_row: RevisionRow) -> Result<SqliteQueryResult> {
+        sqlx::query("INSERT INTO revisions (revision_uri, flake_uri)
+            VALUES (?, ?)
+            ON CONFLICT(revision_uri) DO NOTHING
+            ")
+            .bind(&revision_row.revision_uri)
+            .bind(&revision_row.flake_uri)
+            .execute(&self.db)
+            .await
+            .context("Failed to execute database query")
+    }
+
     pub async fn revisions_from_flake(&self, uri: &str) -> Result<Vec<RevisionRow>> {
         sqlx::query_as("
             SELECT
@@ -63,6 +99,27 @@ impl Storage {
             .context("Failed to fetch data from database")
     }
 
+    pub async fn set_input(&self, input_row: InputRow) -> Result<SqliteQueryResult> {
+        sqlx::query("INSERT INTO inputs (revision_uri, input_name, locked_revision_uri, locked_flake_uri, locked_nar_hash, last_modified)
+            VALUES (?, ?, ?, ?, ?, ?)
+            ON CONFLICT(revision_uri, input_name) DO UPDATE SET
+                locked_revision_uri=excluded.locked_revision_uri,
+                locked_flake_uri=excluded.locked_flake_uri,
+                locked_nar_hash=excluded.locked_nar_hash,
+                last_modified=excluded.last_modified
+            ")
+            .bind(input_row.revision_uri)
+            .bind(input_row.input_name)
+            .bind(input_row.locked_revision_uri)
+            .bind(input_row.locked_flake_uri)
+            .bind(input_row.locked_nar_hash)
+            .bind(input_row.last_modified)
+            .execute(&self.db)
+            .await
+            .context("Failed to execute database query")
+
+    }
+
     pub async fn inputs_for_revision(&self, revision_uri: &str) -> Result<Vec<InputRow>> {
         sqlx::query_as("
             SELECT