fix: 🚑 Handle multiple file records for same file_key (#593)

* fix: 🚑 `get_by_file_key` now returns a vector and there's a `get_latest_by_file_key`

* fix: 🚑 Propagate fix to backend functions

* fix: 🚑 ensure file operations in indexer are done by id instead of file key

* fix: 🔥 remove unused import

* fix: 🐛 update `delete_file` of backend tests with file id argument

---------

Co-authored-by: Tobi Demeco <[email protected]>

Author: ffarall

backend/lib/src/data/indexer_db/repository/mod.rs

@@ -111,6 +111,11 @@ pub trait IndexerOps: Send + Sync {
 
     /// Retrieve the file identified with the given File Key
     ///
+    /// There can be multiple file records for a given file key if there were multiple
+    /// storage requests for the same file key. We get the oldest one created, which
+    /// would be the original storage request that first created the file.
+    /// This is good enough for the purpose of this query.
+    ///
     /// # Arguments
     /// * `key` - the File Key to search
     async fn get_file_by_file_key(&self, file_key: &Hash) -> RepositoryResult<File>;

backend/lib/src/data/indexer_db/repository/postgres.rs

@@ -138,7 +138,11 @@ impl IndexerOps for Repository {
     async fn get_file_by_file_key(&self, file_key: &Hash) -> RepositoryResult<File> {
         let mut conn = self.pool.get().await?;
 
-        File::get_by_file_key(&mut conn, file_key.as_bytes())
+        // There can be multiple file records for a given file key if there were multiple
+        // storage requests for the same file key. We get the oldest one created, which
+        // would be the original storage request that first created the file.
+        // This is good enough for the purpose of this query.
+        File::get_oldest_by_file_key(&mut conn, file_key.as_bytes())
             .await
             .map_err(Into::into)
     }
@@ -307,7 +311,8 @@ impl IndexerOpsMut for Repository {
         let mut conn = self.pool.get().await?;
 
         // TODO: also clear related associations, like bsp_file
-        File::delete(&mut conn, file_key.as_bytes()).await?;
+        let file_record = File::get_latest_by_file_key(&mut conn, file_key.as_bytes()).await?;
+        File::delete(&mut conn, file_record.id).await?;
         Ok(())
     }
 

client/indexer-db/src/models/file.rs

@@ -10,7 +10,7 @@ use shc_common::types::{FileMetadata, Fingerprint};
 
 use crate::{
     models::{Bucket, MultiAddress},
-    schema::{bucket, file, file_peer_id, msp_file},
+    schema::{bucket, file, file_peer_id},
     DbConnection,
 };
 
@@ -169,22 +169,55 @@ impl File {
             .execute(conn)
             .await?;
 
-        // Update bucket total size and file count
-        Bucket::increment_file_count_and_size(conn, bucket_id, size).await?;
-
         Ok(file)
     }
 
+    /// Get all file records for a given file key.
+    ///
+    /// There can be multiple file records for a given file key if there were
+    /// multiple storage requests for the same file key.
     pub async fn get_by_file_key<'a>(
         conn: &mut DbConnection<'a>,
         file_key: impl AsRef<[u8]>,
+    ) -> Result<Vec<Self>, diesel::result::Error> {
+        let file_key = file_key.as_ref().to_vec();
+        let file_records = file::table
+            .filter(file::file_key.eq(file_key))
+            .load::<Self>(conn)
+            .await?;
+        Ok(file_records)
+    }
+
+    /// Get the most recently created file record for a given file key.
+    ///
+    /// Returns error if there are no records for the given key.
+    pub async fn get_latest_by_file_key<'a>(
+        conn: &mut DbConnection<'a>,
+        file_key: impl AsRef<[u8]>,
     ) -> Result<Self, diesel::result::Error> {
         let file_key = file_key.as_ref().to_vec();
-        let file = file::table
+        let file_record: Self = file::table
             .filter(file::file_key.eq(file_key))
-            .first::<Self>(conn)
+            .order(file::created_at.desc())
+            .first(conn)
             .await?;
-        Ok(file)
+        Ok(file_record)
+    }
+
+    /// Get the oldest file record for a given file key.
+    ///
+    /// Returns error if there are no records for the given key.
+    pub async fn get_oldest_by_file_key<'a>(
+        conn: &mut DbConnection<'a>,
+        file_key: impl AsRef<[u8]>,
+    ) -> Result<Self, diesel::result::Error> {
+        let file_key = file_key.as_ref().to_vec();
+        let file_record: Self = file::table
+            .filter(file::file_key.eq(file_key))
+            .order(file::created_at.asc())
+            .first(conn)
+            .await?;
+        Ok(file_record)
     }
 
     pub async fn update_step<'a>(
@@ -203,29 +236,14 @@ impl File {
 
     pub async fn delete<'a>(
         conn: &mut DbConnection<'a>,
-        file_key: impl AsRef<[u8]>,
+        file_id: i64,
     ) -> Result<(), diesel::result::Error> {
-        let file_key = file_key.as_ref().to_vec();
-
-        // Get file info before deletion
-        let file_info: Option<(i64, i64)> = file::table
-            .filter(file::file_key.eq(&file_key))
-            .select((file::bucket_id, file::size))
-            .first(conn)
-            .await
-            .optional()?;
-
-        // Delete the file
+        // Delete the file by its ID
         diesel::delete(file::table)
-            .filter(file::file_key.eq(file_key))
+            .filter(file::id.eq(file_id))
             .execute(conn)
             .await?;
 
-        // Update bucket counts if file was found
-        if let Some((bucket_id, file_size)) = file_info {
-            Bucket::decrement_file_count_and_size(conn, bucket_id, file_size).await?;
-        }
-
         Ok(())
     }
 
@@ -269,99 +287,68 @@ impl File {
         Ok(())
     }
 
-    /// Check if file has any MSP associations
-    pub async fn has_msp_associations<'a>(
-        conn: &mut DbConnection<'a>,
-        file_key: impl AsRef<[u8]>,
-    ) -> Result<bool, diesel::result::Error> {
-        let file_key = file_key.as_ref().to_vec();
-
-        // Get file ID
-        let file_id: Option<i64> = file::table
-            .filter(file::file_key.eq(&file_key))
-            .select(file::id)
-            .first(conn)
-            .await
-            .optional()?;
-
-        if let Some(file_id) = file_id {
-            let count: i64 = msp_file::table
-                .filter(msp_file::file_id.eq(file_id))
-                .count()
-                .get_result(conn)
-                .await?;
-            Ok(count > 0)
-        } else {
-            Ok(false)
-        }
-    }
-
     /// Check if file has any BSP associations
     pub async fn has_bsp_associations<'a>(
         conn: &mut DbConnection<'a>,
-        file_key: impl AsRef<[u8]>,
+        file_id: i64,
     ) -> Result<bool, diesel::result::Error> {
         use crate::schema::bsp_file;
 
-        let file_key = file_key.as_ref().to_vec();
-
-        // Get file ID
-        let file_id: Option<i64> = file::table
-            .filter(file::file_key.eq(&file_key))
-            .select(file::id)
-            .first(conn)
-            .await
-            .optional()?;
-
-        if let Some(file_id) = file_id {
-            let count: i64 = bsp_file::table
-                .filter(bsp_file::file_id.eq(file_id))
-                .count()
-                .get_result(conn)
-                .await?;
-            Ok(count > 0)
-        } else {
-            Ok(false)
-        }
+        let count: i64 = bsp_file::table
+            .filter(bsp_file::file_id.eq(file_id))
+            .count()
+            .get_result(conn)
+            .await?;
+        Ok(count > 0)
     }
 
     /// Delete file only if it has no BSP associations and is not in the bucket forest.
     /// The flag [`is_in_bucket`](File::is_in_bucket) is set to false or true based on the [`MutationsApplied`] event emitted by the proofs dealer pallet for catch all.
+    /// Returns true if all files associated with the file key were deleted, false if any still has associations.
     pub async fn delete_if_orphaned<'a>(
         conn: &mut DbConnection<'a>,
         file_key: impl AsRef<[u8]>,
-    ) -> Result<bool, diesel::result::Error> {
+    ) -> Result<(), diesel::result::Error> {
         let file_key = file_key.as_ref();
 
         // Check if file is still in bucket forest or has BSP associations
-        let file_record: Option<Self> = file::table
+        let file_records: Vec<Self> = file::table
             .filter(file::file_key.eq(file_key))
-            .first(conn)
-            .await
-            .optional()?;
-
-        let Some(file_record) = file_record else {
-            // File doesn't exist, nothing to delete
-            return Ok(false);
-        };
+            .load(conn)
+            .await?;
 
-        let has_bsp = Self::has_bsp_associations(conn, file_key).await?;
-        let is_in_bucket = file_record.is_in_bucket;
-
-        if !is_in_bucket && !has_bsp {
-            // File is not in bucket forest and has no BSP associations, safe to delete
-            Self::delete(conn, file_key).await?;
-            log::debug!("Deleted orphaned file key: {:?}", file_key);
-            Ok(true)
-        } else {
-            log::debug!(
-                "File with key {:?} still has storage (in_bucket: {}, BSP: {}), keeping it",
-                file_key,
-                is_in_bucket,
-                has_bsp
-            );
-            Ok(false)
+        // For each found file record, check if it has any BSP or MSP associations, and delete it if it doesn't
+        let mut deleted_all = true;
+        for file_record in file_records {
+            let has_bsp = Self::has_bsp_associations(conn, file_record.id).await?;
+            let is_in_bucket = file_record.is_in_bucket;
+
+            if !is_in_bucket && !has_bsp {
+                Self::delete(conn, file_record.id).await?;
+                log::debug!(
+                    "Deleted orphaned file key: {:?} and id: {:?}",
+                    file_record.file_key,
+                    file_record.id
+                );
+            } else {
+                log::debug!(
+                		"File with key {:?} and id {:?} still has storage (in_bucket: {}, BSP: {}), keeping it",
+                		file_record.file_key,
+                		file_record.id,
+                		is_in_bucket,
+                		has_bsp,
+            		);
+                deleted_all = false;
+            }
         }
+
+        log::debug!(
+            "Deleted all files associated with file key: {:?}: {}",
+            file_key,
+            deleted_all
+        );
+
+        Ok(())
     }
 
     pub async fn get_by_bucket_id<'a>(
@@ -649,13 +636,31 @@ impl File {
         let file_key = file_key.as_ref().to_vec();
         let onchain_bucket_id = onchain_bucket_id.as_ref().to_vec();
 
+        // Get the file info
+        let file_info: Option<(i64, i64)> = file::table
+            .filter(file::file_key.eq(&file_key))
+            .filter(file::onchain_bucket_id.eq(&onchain_bucket_id))
+            .select((file::bucket_id, file::size))
+            .first(conn)
+            .await
+            .optional()?;
+
+        // Update the file's bucket membership status
         diesel::update(file::table)
             .filter(file::file_key.eq(&file_key))
             .filter(file::onchain_bucket_id.eq(&onchain_bucket_id))
             .set(file::is_in_bucket.eq(is_in_bucket))
             .execute(conn)
             .await?;
 
+        // Update the bucket stats to reflect the change in bucket membership
+        if let Some((bucket_id, file_size)) = file_info {
+            match is_in_bucket {
+                true => Bucket::increment_file_count_and_size(conn, bucket_id, file_size).await?,
+                false => Bucket::decrement_file_count_and_size(conn, bucket_id, file_size).await?,
+            }
+        }
+
         Ok(())
     }
 }

client/indexer-db/src/models/msp_file.rs

@@ -53,7 +53,7 @@ impl MspFile {
         // Log if we found multiple files with the same key
         if file_ids.len() > 1 {
             log::warn!(
-                "Found {} files with the same file_key: {:?}. This is an inconsistent state. Will proceed to delete all associated file IDs with this key.",
+                "Found {} files with the same file_key: {:?}. This is expected only if there was more than one storage request for the same file key. Will proceed to delete all associated file IDs with this key.",
                 file_ids.len(),
                 file_key
             );