feat: handle collection unverification for reg NFTs

README.md

@@ -59,7 +59,7 @@ Because this is a multi component system the easiest way to develop or locally t
 
 Edit the init.sql, then run `docker compose up db`
 Then with a local `DATABASE_URL` var exported like this `export DATABASE_URL=postgres://solana:solana@localhost/solana` you can run
-` sea-orm-cli generate entity -o ./digital_asset_types/src/dao/generated/ --database-url $DATABASE_URL --with-serde both --expanded-format`
+`sea-orm-cli generate entity -o ./digital_asset_types/src/dao/generated/ --database-url $DATABASE_URL --with-serde both --expanded-format`
 
 If you need to install `sea-orm-cli` run `cargo install sea-orm-cli`.
 

digital_asset_types/src/dao/generated/asset_grouping.rs

@@ -1,4 +1,4 @@
-//! SeaORM Entity. Generated by sea-orm-codegen 0.9.3
+//! `SeaORM` Entity. Generated by sea-orm-codegen 0.11.3
 
 use sea_orm::entity::prelude::*;
 use serde::{Deserialize, Serialize};
@@ -12,12 +12,12 @@ impl EntityName for Entity {
     }
 }
 
-#[derive(Clone, Debug, PartialEq, DeriveModel, DeriveActiveModel, Serialize, Deserialize)]
+#[derive(Clone, Debug, PartialEq, DeriveModel, DeriveActiveModel, Eq, Serialize, Deserialize)]
 pub struct Model {
     pub id: i64,
     pub asset_id: Vec<u8>,
     pub group_key: String,
-    pub group_value: String,
+    pub group_value: Option<String>,
     pub seq: i64,
     pub slot_updated: i64,
 }
@@ -56,7 +56,7 @@ impl ColumnTrait for Column {
             Self::Id => ColumnType::BigInteger.def(),
             Self::AssetId => ColumnType::Binary.def(),
             Self::GroupKey => ColumnType::Text.def(),
-            Self::GroupValue => ColumnType::Text.def(),
+            Self::GroupValue => ColumnType::Text.def().null(),
             Self::Seq => ColumnType::BigInteger.def(),
             Self::SlotUpdated => ColumnType::BigInteger.def(),
         }

digital_asset_types/src/dao/scopes/asset.rs

@@ -290,6 +290,7 @@ pub async fn get_by_id(
         .await?;
     let grouping: Vec<asset_grouping::Model> = asset_grouping::Entity::find()
         .filter(asset_grouping::Column::AssetId.eq(asset.id.clone()))
+        .filter(asset_grouping::Column::GroupValue.is_not_null())
         .order_by_asc(asset_grouping::Column::AssetId)
         .all(conn)
         .await?;

digital_asset_types/src/rpc/asset.rs

@@ -202,7 +202,7 @@ pub type GroupValue = String;
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
 pub struct Group {
     pub group_key: String,
-    pub group_value: String,
+    pub group_value: Option<String>,
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq, JsonSchema)]

digital_asset_types/tests/common.rs

@@ -216,12 +216,12 @@ pub fn create_asset_grouping(
         asset_grouping::ActiveModel {
             asset_id: Set(asset_id.clone()),
             group_key: Set(String::from("collection")),
-            group_value: Set(bs58::encode(collection).into_string()),
+            group_value: Set(Some(bs58::encode(collection).into_string())),
             ..Default::default()
         },
         asset_grouping::Model {
             asset_id,
-            group_value: bs58::encode(collection).into_string(),
+            group_value: Some(bs58::encode(collection).into_string()),
             seq: 0,
             id: row_num,
             group_key: "collection".to_string(),

migration/src/lib.rs

@@ -21,6 +21,7 @@ mod m20230516_185005_add_reindex_to_assets;
 mod m20230525_115717_cl_audit_table;
 mod m20230528_124011_cl_audit_table_index;
 mod m20230613_114817_add_name_symbol_to_asset_data;
+mod m20230726_013107_remove_not_null_constraint_from_group_value;
 
 pub struct Migrator;
 
@@ -49,6 +50,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20230525_115717_cl_audit_table::Migration),
             Box::new(m20230528_124011_cl_audit_table_index::Migration),
             Box::new(m20230613_114817_add_name_symbol_to_asset_data::Migration),
+            Box::new(m20230726_013107_remove_not_null_constraint_from_group_value::Migration),
         ]
     }
 }

migration/src/m20230726_013107_remove_not_null_constraint_from_group_value.rs

@@ -0,0 +1,32 @@
+use digital_asset_types::dao::asset_grouping;
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .alter_table(
+                sea_query::Table::alter()
+                    .table(asset_grouping::Entity)
+                    .modify_column(ColumnDef::new(asset_grouping::Column::GroupValue).null())
+                    .to_owned(),
+            )
+            .await?;
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .alter_table(
+                sea_query::Table::alter()
+                    .table(asset_grouping::Entity)
+                    .modify_column(ColumnDef::new(asset_grouping::Column::GroupValue).not_null())
+                    .to_owned(),
+            )
+            .await?;
+        Ok(())
+    }
+}

nft_ingester/src/program_transformers/bubblegum/collection_verification.rs

@@ -40,7 +40,7 @@ where
                         let grouping = asset_grouping::ActiveModel {
                             asset_id: Set(id_bytes.clone()),
                             group_key: Set("collection".to_string()),
-                            group_value: Set(collection.to_string()),
+                            group_value: Set(Some(collection.to_string())),
                             seq: Set(seq as i64),
                             slot_updated: Set(bundle.slot as i64),
                             ..Default::default()

nft_ingester/src/program_transformers/bubblegum/mint_v1.rs

@@ -20,7 +20,6 @@ use digital_asset_types::{
     },
     json::ChainDataV1,
 };
-use log::info;
 use num_traits::FromPrimitive;
 use sea_orm::{
     entity::*, query::*, sea_query::OnConflict, ConnectionTrait, DbBackend, EntityTrait, JsonValue,
@@ -265,7 +264,7 @@ where
                         let model = asset_grouping::ActiveModel {
                             asset_id: Set(id_bytes.to_vec()),
                             group_key: Set("collection".to_string()),
-                            group_value: Set(c.key.to_string()),
+                            group_value: Set(Some(c.key.to_string())),
                             seq: Set(seq as i64), // gummyroll seq
                             slot_updated: Set(slot_i),
                             ..Default::default()

nft_ingester/src/program_transformers/token_metadata/v1_asset.rs

@@ -75,7 +75,9 @@ pub async fn save_v1_asset<T: ConnectionTrait + TransactionTrait>(
         _ => OwnerType::Single,
     };
 
-    // gets the token and token account for the mint to populate the asset. This is required when the token and token account are indexed, but not the metadata account. If the metadata account is indexed, then the token and ta ingester will update the asset with the correct data
+    // Gets the token and token account for the mint to populate the asset.
+    // This is required when the token and token account are indexed, but not the metadata account.
+    // If the metadata account is indexed, then the token and ta ingester will update the asset with the correct data.
 
     let (token, token_account): (Option<tokens::Model>, Option<token_accounts::Model>) =
         match ownership_type {
@@ -270,7 +272,7 @@ pub async fn save_v1_asset<T: ConnectionTrait + TransactionTrait>(
             let model = asset_grouping::ActiveModel {
                 asset_id: Set(id.to_vec()),
                 group_key: Set("collection".to_string()),
-                group_value: Set(c.key.to_string()),
+                group_value: Set(Some(c.key.to_string())),
                 seq: Set(0),
                 slot_updated: Set(slot_i),
                 ..Default::default()
@@ -297,6 +299,35 @@ pub async fn save_v1_asset<T: ConnectionTrait + TransactionTrait>(
             txn.execute(query)
                 .await
                 .map_err(|db_err| IngesterError::AssetIndexError(db_err.to_string()))?;
+        } else {
+            let model = asset_grouping::ActiveModel {
+                asset_id: Set(id.to_vec()),
+                group_key: Set("collection".to_string()),
+                group_value: Set(None),
+                seq: Set(0),
+                slot_updated: Set(slot_i),
+                ..Default::default()
+            };
+            let mut query = asset_grouping::Entity::insert(model)
+                .on_conflict(
+                    OnConflict::columns([
+                        asset_grouping::Column::AssetId,
+                        asset_grouping::Column::GroupKey,
+                    ])
+                    .update_columns([
+                        asset_grouping::Column::GroupValue,
+                        asset_grouping::Column::SlotUpdated,
+                    ])
+                    .to_owned(),
+                )
+                .build(DbBackend::Postgres);
+            query.sql = format!(
+                "{} WHERE excluded.slot_updated > asset_grouping.slot_updated AND excluded.seq >= asset_grouping.seq",
+                query.sql
+            );
+            txn.execute(query)
+                .await
+                .map_err(|db_err| IngesterError::AssetIndexError(db_err.to_string()))?;
         }
     }