[MET-46] feat: dynamic data slot for each field

digital_asset_types/src/dao/converters.rs

@@ -23,7 +23,7 @@ impl TryFrom<SearchAssetsQuery> for postgre_client::model::SearchAssetsFilter {
             .grouping
             .map(|(key, val)| {
                 if key != "collection" {
-                    return Err(ConversionError::IncompatibleGroupingKey(key))
+                    return Err(ConversionError::IncompatibleGroupingKey(key));
                 }
                 Ok(val)
             })

digital_asset_types/src/dao/scopes/asset.rs

@@ -477,6 +477,7 @@ fn convert_rocks_offchain_data(
     let ch_data: serde_json::Value = serde_json::from_str(
         dynamic_data
             .onchain_data
+            .1
             .clone()
             .unwrap_or_default()
             .as_ref(),
@@ -490,7 +491,7 @@ fn convert_rocks_offchain_data(
         metadata_url: offchain_data.url.clone(),
         metadata_mutability: Mutability::Immutable,
         metadata: Json::from_str(metadata.as_str()).map_err(|e| DbErr::Custom(e.to_string()))?,
-        slot_updated: dynamic_data.slot_updated as i64,
+        slot_updated: 0, // TODO
         reindex: None,
     })
 }
@@ -523,14 +524,10 @@ fn convert_rocks_asset_model(
     } else {
         Some(leaf.tree_id.to_bytes().to_vec())
     };
-    let slot_updated = vec![
-        dynamic_data.slot_updated,
-        owner.slot_updated,
-        leaf.slot_updated,
-    ]
-    .into_iter()
-    .max()
-    .unwrap(); // unwrap here is save, because vec is not empty
+    let slot_updated = vec![owner.slot_updated, leaf.slot_updated]
+        .into_iter()
+        .max()
+        .unwrap(); // unwrap here is save, because vec is not empty
 
     Ok(asset::Model {
         id: static_data.pubkey.to_bytes().to_vec(),
@@ -540,29 +537,30 @@ fn convert_rocks_asset_model(
         owner: Some(owner.owner.to_bytes().to_vec()),
         owner_type: owner.owner_type.clone().into(),
         delegate: owner.delegate.map(|pk| pk.to_bytes().to_vec()),
-        frozen: dynamic_data.is_frozen,
+        frozen: dynamic_data.is_frozen.1,
         supply: dynamic_data
             .supply
+            .1
             .map(|supply| supply as i64)
             .unwrap_or_default(),
         supply_mint: Some(static_data.pubkey.to_bytes().to_vec()),
-        compressed: dynamic_data.is_compressed,
-        compressible: dynamic_data.is_compressible,
-        seq: dynamic_data.seq.map(|u| u.try_into().ok()).flatten(),
+        compressed: dynamic_data.is_compressed.1,
+        compressible: dynamic_data.is_compressible.1,
+        seq: dynamic_data.seq.1.map(|u| u.try_into().ok()).flatten(),
         tree_id,
         leaf: leaf.leaf.clone(),
         nonce: leaf.nonce.map(|nonce| nonce as i64),
         royalty_target_type: static_data.royalty_target_type.clone().into(),
         royalty_target: None, // TODO
-        royalty_amount: dynamic_data.royalty_amount as i32,
+        royalty_amount: dynamic_data.royalty_amount.1 as i32,
         asset_data: Some(static_data.pubkey.to_bytes().to_vec()),
-        burnt: dynamic_data.is_burnt,
+        burnt: dynamic_data.is_burnt.1,
         created_at: Some(static_data.created_at),
         slot_updated: Some(slot_updated as i64),
         data_hash: leaf.data_hash.map(|h| h.to_string()),
         creator_hash: leaf.creator_hash.map(|h| h.to_string()),
         owner_delegate_seq: owner.owner_delegate_seq.map(|seq| seq as i64),
-        was_decompressed: dynamic_data.was_decompressed,
+        was_decompressed: dynamic_data.was_decompressed.1,
         leaf_seq: leaf.leaf_seq.map(|seq| seq as i64),
     })
 }
@@ -666,6 +664,7 @@ fn convert_rocks_creators_model(
 
     dynamic_data
         .creators
+        .1
         .iter()
         .enumerate()
         .map(|(position, creator)| asset_creators::Model {
@@ -674,8 +673,8 @@ fn convert_rocks_creators_model(
             creator: creator.creator.to_bytes().to_vec(),
             share: creator.creator_share as i32,
             verified: creator.creator_verified,
-            seq: Some(dynamic_data.slot_updated as i64),
-            slot_updated: Some(dynamic_data.slot_updated as i64),
+            seq: Some(0),          // TODO
+            slot_updated: Some(0), // TODO
             position: position as i16,
         })
         .collect::<Vec<_>>()

digital_asset_types/src/dapi/mod.rs

@@ -14,4 +14,4 @@ pub use assets_by_owner::*;
 pub use change_logs::*;
 pub use get_asset::*;
 pub use get_asset_batch::*;
-pub use search_assets::*;
+pub use search_assets::*;

digital_asset_types/src/dapi/search_assets.rs

@@ -20,7 +20,8 @@ pub async fn search_assets(
     before: Option<String>,
     after: Option<String>,
 ) -> Result<AssetList, DbErr> {
-    let filter_result: &Result<postgre_client::model::SearchAssetsFilter, ConversionError> = &filter.try_into();
+    let filter_result: &Result<postgre_client::model::SearchAssetsFilter, ConversionError> =
+        &filter.try_into();
     if let Err(ConversionError::IncompatibleGroupingKey(_)) = filter_result {
         // If the error is IncompatibleGroupingKey, return an empty response
         return Ok(AssetList {
@@ -29,7 +30,9 @@ pub async fn search_assets(
             ..AssetList::default()
         });
     }
-    let filter = filter_result.as_ref().map_err(|e| DbErr::Custom(e.to_string()))?;
+    let filter = filter_result
+        .as_ref()
+        .map_err(|e| DbErr::Custom(e.to_string()))?;
     let keys = index_client
         .get_asset_pubkeys_filtered(filter, &sort_by.into(), limit, page, before, after)
         .await

nft_ingester/src/api/builder.rs

@@ -124,7 +124,6 @@ impl RpcApiBuilder {
         });
         module.add_alias("searchAssets", "search_assets");
 
-
         Ok(module)
     }
 }

nft_ingester/src/api/mod.rs

@@ -127,7 +127,7 @@ pub struct SearchAssets {
     pub limit: Option<u32>,
     pub page: Option<u32>,
     // before and after are base64 encoded sorting keys, which is NOT a pubkey, passing any non empty base64 encoded string will not cause an error
-    pub before: Option<String>, 
+    pub before: Option<String>,
     pub after: Option<String>,
     #[serde(default)]
     pub json_uri: Option<String>,
@@ -154,7 +154,7 @@ impl TryFrom<SearchAssets> for digital_asset_types::dao::SearchAssetsQuery {
             .transpose()?;
         Ok(digital_asset_types::dao::SearchAssetsQuery {
             negate: search_assets.negate,
-            condition_type: search_assets.condition_type.map(|s| s.into()), 
+            condition_type: search_assets.condition_type.map(|s| s.into()),
             owner_address: validate_opt_pubkey(&search_assets.owner_address)?,
             owner_type: search_assets.owner_type.map(|s| s.into()),
             creator_address: validate_opt_pubkey(&search_assets.creator_address)?,

nft_ingester/src/api/validation.rs

@@ -15,4 +15,4 @@ pub fn validate_opt_pubkey(pubkey: &Option<String>) -> Result<Option<Vec<u8>>, D
         None
     };
     Ok(opt_bytes)
-}
+}

nft_ingester/src/bubblegum_updates_processor.rs

@@ -370,12 +370,11 @@ impl BubblegumTxProcessor {
             let asset_data = self.rocks_client.asset_dynamic_data.get(asset_id).unwrap();
             if let Some(current_asset_data) = asset_data {
                 let mut new_asset_data = current_asset_data.clone();
-                new_asset_data.is_burnt = true;
-                new_asset_data.supply = Some(0);
-                new_asset_data.seq = Some(cl.seq);
-                new_asset_data.slot_updated = bundle.slot;
+                new_asset_data.is_burnt = (bundle.slot, true);
+                new_asset_data.supply = (bundle.slot, Some(0));
+                new_asset_data.seq = (bundle.slot, Some(cl.seq));
 
-                if let Some(current_seq) = current_asset_data.seq {
+                if let Some(current_seq) = current_asset_data.seq.1 {
                     if current_seq < cl.seq {
                         if let Err(e) = self
                             .rocks_client
@@ -397,17 +396,16 @@ impl BubblegumTxProcessor {
             } else {
                 let new_asset_data = AssetDynamicDetails {
                     pubkey: asset_id,
-                    is_compressible: false,
-                    is_compressed: true,
-                    is_frozen: false,
-                    supply: Some(0),
-                    seq: Some(cl.seq),
-                    is_burnt: true,
-                    was_decompressed: false,
-                    onchain_data: None,
-                    creators: vec![],
-                    royalty_amount: 0,
-                    slot_updated: bundle.slot,
+                    is_compressible: (bundle.slot, false),
+                    is_compressed: (bundle.slot, true),
+                    is_frozen: (bundle.slot, false),
+                    supply: (bundle.slot, Some(0)),
+                    seq: (bundle.slot, Some(cl.seq)),
+                    is_burnt: (bundle.slot, true),
+                    was_decompressed: (bundle.slot, false),
+                    onchain_data: (bundle.slot, None),
+                    creators: (bundle.slot, vec![]),
+                    royalty_amount: (bundle.slot, 0),
                 };
                 if let Err(e) = self
                     .rocks_client
@@ -550,23 +548,22 @@ impl BubblegumTxProcessor {
 
                     let asset_dynamic_details = AssetDynamicDetails {
                         pubkey: id,
-                        is_compressible: false,
-                        is_compressed: true,
-                        is_frozen: false,
-                        supply: Some(1),
-                        seq: Some(cl.seq),
-                        is_burnt: false,
-                        was_decompressed: false,
-                        onchain_data: Some(chain_data.to_string()),
-                        creators,
-                        royalty_amount: args.seller_fee_basis_points,
-                        slot_updated: bundle.slot,
+                        is_compressible: (bundle.slot, false),
+                        is_compressed: (bundle.slot, true),
+                        is_frozen: (bundle.slot, false),
+                        supply: (bundle.slot, Some(1)),
+                        seq: (bundle.slot, Some(cl.seq)),
+                        is_burnt: (bundle.slot, false),
+                        was_decompressed: (bundle.slot, false),
+                        onchain_data: (bundle.slot, Some(chain_data.to_string())),
+                        creators: (bundle.slot, creators),
+                        royalty_amount: (bundle.slot, args.seller_fee_basis_points),
                     };
 
                     if let Err(e) = self
                         .rocks_client
                         .asset_dynamic_data
-                        .put(id, &asset_dynamic_details)
+                        .merge(id, &asset_dynamic_details)
                     {
                         error!("Error while saving dynamic data for cNFT: {}", e);
                     };
@@ -766,9 +763,8 @@ impl BubblegumTxProcessor {
                     let asset_data = self.rocks_client.asset_dynamic_data.get(id).unwrap();
                     if let Some(current_asset_data) = asset_data {
                         let mut new_asset_data = current_asset_data.clone();
-                        new_asset_data.seq = None;
-                        new_asset_data.was_decompressed = true;
-                        new_asset_data.slot_updated = bundle.slot;
+                        new_asset_data.seq = (bundle.slot, None);
+                        new_asset_data.was_decompressed = (bundle.slot, true);
 
                         if let Err(e) = self
                             .rocks_client
@@ -780,17 +776,16 @@ impl BubblegumTxProcessor {
                     } else {
                         let new_asset_data = AssetDynamicDetails {
                             pubkey: id,
-                            is_compressible: true,
-                            is_compressed: false,
-                            is_frozen: false,
-                            supply: Some(1),
-                            seq: None,
-                            is_burnt: false,
-                            was_decompressed: true,
-                            onchain_data: None,
-                            creators: vec![],
-                            royalty_amount: 0,
-                            slot_updated: bundle.slot,
+                            is_compressible: (bundle.slot, true),
+                            is_compressed: (bundle.slot, false),
+                            is_frozen: (bundle.slot, false),
+                            supply: (bundle.slot, Some(1)),
+                            seq: (bundle.slot, None),
+                            is_burnt: (bundle.slot, false),
+                            was_decompressed: (bundle.slot, true),
+                            onchain_data: (bundle.slot, None),
+                            creators: (bundle.slot, vec![]),
+                            royalty_amount: (bundle.slot, 0),
                         };
                         if let Err(e) = self
                             .rocks_client
@@ -858,10 +853,9 @@ impl BubblegumTxProcessor {
                     let asset_data = self.rocks_client.asset_dynamic_data.get(id).unwrap();
                     if let Some(current_asset_data) = asset_data {
                         let mut new_asset_data = current_asset_data.clone();
-                        new_asset_data.seq = Some(cl.seq);
-                        new_asset_data.slot_updated = bundle.slot;
+                        new_asset_data.seq = (bundle.slot, Some(cl.seq));
 
-                        for crt in new_asset_data.creators.iter_mut() {
+                        for crt in new_asset_data.creators.1.iter_mut() {
                             if crt.creator == *creator {
                                 crt.creator_verified = *verify;
                             }
@@ -883,17 +877,16 @@ impl BubblegumTxProcessor {
 
                         let new_asset_data = AssetDynamicDetails {
                             pubkey: id,
-                            is_compressible: false,
-                            is_compressed: true,
-                            is_frozen: false,
-                            supply: Some(1),
-                            seq: Some(cl.seq),
-                            is_burnt: false,
-                            was_decompressed: false,
-                            onchain_data: None,
-                            creators: vec![creator],
-                            royalty_amount: 0,
-                            slot_updated: bundle.slot,
+                            is_compressible: (bundle.slot, false),
+                            is_compressed: (bundle.slot, true),
+                            is_frozen: (bundle.slot, false),
+                            supply: (bundle.slot, Some(1)),
+                            seq: (bundle.slot, Some(cl.seq)),
+                            is_burnt: (bundle.slot, false),
+                            was_decompressed: (bundle.slot, false),
+                            onchain_data: (bundle.slot, None),
+                            creators: (bundle.slot, vec![creator]),
+                            royalty_amount: (bundle.slot, 0),
                         };
                         if let Err(e) = self
                             .rocks_client