Change creator and collection verification to use upserts to support …

…out of order (#87)

* Fix docker preparation script to build SPL

* Update owner and delegate in asset table when collection or creator verification occurs

* Modify program transformers to upsert in asset table

* This allows out-of-order Bubblegum transactions to
create and update the asset table.
* Upsert leaf schema, owner, delegate, and seq separately
since those are updated by all instructions and gated
by sequence number to ensure freshest value.
* Mint, burn, and decompress happen without regard to
sequence number because they operate on unique fields.
* Mint and burn have been updated but Decompress still
needs to be fixed to handle out of order transactions.
* Also remove unused 'filling' variable.

* Update mint and decompress to be able to upsert asset info out of order

* Add second sequence number for compression status fields

* Reduce logging in docker

* Comment out compressed_seq before regenerating Sea ORM objects

* Add migration for asset specification

* Update README

* Rename PNFT and regenerate Sea ORM types

* Apply usage of compressed_seq after regenerating Sea ORM types

* Add owner delegate sequence number for owner and delegate fields.

Also remove not null constraints for asset fields without defaults.

* Regenerating database types

* Update handling for non null constrained asset table

* Update tests to use new Sea ORM types

* Use owner_and_delegate_seq to separate upserts

Also update redeem and decompress to not use
leaf schema events.

* Adding was_decompressed flag to replace compressed_seq

compressed_seq won't work because decompression doesn't
create a cl_event.

* Regenerating Sea ORM types

* Update code to use was_decompressed flag

* Fix new boolean SQL conditions

* Update comment

* Remove column updates in asset table during mint for items not in model

* Clippy fixes in ingester main

* Cleanup debug comment

* Allow for sequence number to be NULL (needed after decompress now)

* Add leaf specific sequence number to protect that field in asset table

* Revert "Allow for sequence number to be NULL (needed after decompress now)"

This reverts commit 2713a18.

* Update nft_ingester/src/program_transformers/bubblegum/redeem.rs

Co-authored-by: Nicolas Pennie <[email protected]>

* Change creator verification to use upserts to support out of order

* Remove null constraints on asset_creators table

* Add null clause to upsert during mint

* Rename creator vecs and add comments

* Removing comment

* Fix typo in migration down function

* Fix collection verification and change to use upserts to support out of order processing (#90)

* Adding verified flag to asset_grouping table

* Regenerate Sea ORM types

* Remove null constraints on asset_grouping table

* Regenerate Sea ORM types

* Update digital asset types and ingester based on new Sea ORM objects

* Setting new verified flag in asset_grouping table to be non null with default

Also regenerating Sea ORM types

* Separate out collection insert in mintV1 into separate upserts

* Fix error message

* Separate update collection base info from collection verified

* Add group info seq to asset_grouping table

* Regenerate Sea ORM types

* Add group_info_seq checks to collection base info upsert

* Add check for verified = true in grouping for Read API

* Fix conditions for asset grouping updates

* Require grouping to verified to be returned from API in all cases

---------

Co-authored-by: Nicolas Pennie <[email protected]>

das_api/src/api/api_impl.rs

@@ -1,5 +1,3 @@
-use std::vec;
-
 use digital_asset_types::{
     dao::{
         scopes::asset::get_grouping,
@@ -356,7 +354,7 @@ impl ApiContract for DasApi {
         } = payload;
         let gs = get_grouping(&self.db_connection, group_key.clone(), group_value.clone()).await?;
         Ok(GetGroupingResponse {
-            group_key: group_key,
+            group_key,
             group_name: group_value,
             group_size: gs.size,
         })

das_api/src/api/mod.rs

@@ -1,4 +1,4 @@
-use crate::{DasApiError, RpcModule};
+use crate::DasApiError;
 use async_trait::async_trait;
 use digital_asset_types::rpc::filter::SearchConditionType;
 use digital_asset_types::rpc::response::AssetList;

das_api/src/main.rs

@@ -4,7 +4,7 @@ mod config;
 mod error;
 mod validation;
 
-use std::time::{Duration, Instant};
+use std::time::Instant;
 use {
     crate::api::DasApi,
     crate::builder::RpcApiBuilder,
@@ -17,7 +17,7 @@ use {
     std::net::UdpSocket,
 };
 
-use hyper::{http, Method};
+use hyper::Method;
 use log::{debug, info};
 use tower_http::cors::{Any, CorsLayer};
 

digital_asset_types/src/dao/generated/asset.rs

@@ -97,8 +97,8 @@ pub enum Relation {
     AssetData,
     AssetV1AccountAttachments,
     AssetCreators,
-    AssetGrouping,
     AssetAuthority,
+    AssetGrouping,
 }
 
 impl ColumnTrait for Column {
@@ -148,8 +148,8 @@ impl RelationTrait for Relation {
                 Entity::has_many(super::asset_v1_account_attachments::Entity).into()
             }
             Self::AssetCreators => Entity::has_many(super::asset_creators::Entity).into(),
-            Self::AssetGrouping => Entity::has_many(super::asset_grouping::Entity).into(),
             Self::AssetAuthority => Entity::has_many(super::asset_authority::Entity).into(),
+            Self::AssetGrouping => Entity::has_many(super::asset_grouping::Entity).into(),
         }
     }
 }
@@ -172,15 +172,15 @@ impl Related<super::asset_creators::Entity> for Entity {
     }
 }
 
-impl Related<super::asset_grouping::Entity> for Entity {
+impl Related<super::asset_authority::Entity> for Entity {
     fn to() -> RelationDef {
-        Relation::AssetGrouping.def()
+        Relation::AssetAuthority.def()
     }
 }
 
-impl Related<super::asset_authority::Entity> for Entity {
+impl Related<super::asset_grouping::Entity> for Entity {
     fn to() -> RelationDef {
-        Relation::AssetAuthority.def()
+        Relation::AssetGrouping.def()
     }
 }
 

digital_asset_types/src/dao/generated/asset_creators.rs

@@ -19,8 +19,8 @@ pub struct Model {
     pub creator: Vec<u8>,
     pub share: i32,
     pub verified: bool,
-    pub seq: i64,
-    pub slot_updated: i64,
+    pub seq: Option<i64>,
+    pub slot_updated: Option<i64>,
     pub position: i16,
 }
 
@@ -62,8 +62,8 @@ impl ColumnTrait for Column {
             Self::Creator => ColumnType::Binary.def(),
             Self::Share => ColumnType::Integer.def(),
             Self::Verified => ColumnType::Boolean.def(),
-            Self::Seq => ColumnType::BigInteger.def(),
-            Self::SlotUpdated => ColumnType::BigInteger.def(),
+            Self::Seq => ColumnType::BigInteger.def().null(),
+            Self::SlotUpdated => ColumnType::BigInteger.def().null(),
             Self::Position => ColumnType::SmallInteger.def(),
         }
     }

digital_asset_types/src/dao/generated/asset_grouping.rs

@@ -17,9 +17,11 @@ pub struct Model {
     pub id: i64,
     pub asset_id: Vec<u8>,
     pub group_key: String,
-    pub group_value: String,
-    pub seq: i64,
-    pub slot_updated: i64,
+    pub group_value: Option<String>,
+    pub seq: Option<i64>,
+    pub slot_updated: Option<i64>,
+    pub verified: bool,
+    pub group_info_seq: Option<i64>,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
@@ -30,6 +32,8 @@ pub enum Column {
     GroupValue,
     Seq,
     SlotUpdated,
+    Verified,
+    GroupInfoSeq,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DerivePrimaryKey)]
@@ -56,9 +60,11 @@ 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::Seq => ColumnType::BigInteger.def(),
-            Self::SlotUpdated => ColumnType::BigInteger.def(),
+            Self::GroupValue => ColumnType::Text.def().null(),
+            Self::Seq => ColumnType::BigInteger.def().null(),
+            Self::SlotUpdated => ColumnType::BigInteger.def().null(),
+            Self::Verified => ColumnType::Boolean.def(),
+            Self::GroupInfoSeq => ColumnType::BigInteger.def().null(),
         }
     }
 }

digital_asset_types/src/dao/generated/sea_orm_active_enums.rs

@@ -14,16 +14,6 @@ pub enum Mutability {
     Unknown,
 }
 #[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
-#[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "chain_mutability")]
-pub enum ChainMutability {
-    #[sea_orm(string_value = "immutable")]
-    Immutable,
-    #[sea_orm(string_value = "mutable")]
-    Mutable,
-    #[sea_orm(string_value = "unknown")]
-    Unknown,
-}
-#[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
 #[sea_orm(
     rs_type = "String",
     db_type = "Enum",
@@ -54,6 +44,22 @@ pub enum TaskStatus {
     Success,
 }
 #[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
+#[sea_orm(
+    rs_type = "String",
+    db_type = "Enum",
+    enum_name = "royalty_target_type"
+)]
+pub enum RoyaltyTargetType {
+    #[sea_orm(string_value = "creators")]
+    Creators,
+    #[sea_orm(string_value = "fanout")]
+    Fanout,
+    #[sea_orm(string_value = "single")]
+    Single,
+    #[sea_orm(string_value = "unknown")]
+    Unknown,
+}
+#[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
 #[sea_orm(
     rs_type = "String",
     db_type = "Enum",
@@ -82,6 +88,16 @@ pub enum SpecificationAssetClass {
     Unknown,
 }
 #[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
+#[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "chain_mutability")]
+pub enum ChainMutability {
+    #[sea_orm(string_value = "immutable")]
+    Immutable,
+    #[sea_orm(string_value = "mutable")]
+    Mutable,
+    #[sea_orm(string_value = "unknown")]
+    Unknown,
+}
+#[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
 #[sea_orm(
     rs_type = "String",
     db_type = "Enum",
@@ -98,22 +114,6 @@ pub enum SpecificationVersions {
     V2,
 }
 #[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
-#[sea_orm(
-    rs_type = "String",
-    db_type = "Enum",
-    enum_name = "royalty_target_type"
-)]
-pub enum RoyaltyTargetType {
-    #[sea_orm(string_value = "creators")]
-    Creators,
-    #[sea_orm(string_value = "fanout")]
-    Fanout,
-    #[sea_orm(string_value = "single")]
-    Single,
-    #[sea_orm(string_value = "unknown")]
-    Unknown,
-}
-#[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
 #[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "owner_type")]
 pub enum OwnerType {
     #[sea_orm(string_value = "single")]

digital_asset_types/src/dao/scopes/asset.rs

@@ -65,7 +65,8 @@ pub async fn get_grouping(
         .filter(
             Condition::all()
                 .add(asset_grouping::Column::GroupKey.eq(group_key))
-                .add(asset_grouping::Column::GroupValue.eq(group_value)),
+                .add(asset_grouping::Column::GroupValue.eq(group_value))
+                .add(asset_grouping::Column::Verified.eq(true)),
         )
         .count(conn)
         .await?;
@@ -83,7 +84,8 @@ pub async fn get_by_grouping(
 ) -> Result<Vec<FullAsset>, DbErr> {
     let condition = asset_grouping::Column::GroupKey
         .eq(group_key)
-        .and(asset_grouping::Column::GroupValue.eq(group_value));
+        .and(asset_grouping::Column::GroupValue.eq(group_value))
+        .and(asset_grouping::Column::Verified.eq(true));
     get_by_related_condition(
         conn,
         Condition::all()
@@ -192,7 +194,7 @@ pub async fn get_related_for_assets(
         {
             let id = asset.id.clone();
             let fa = FullAsset {
-                asset: asset,
+                asset,
                 data: ad.clone(),
                 authorities: vec![],
                 creators: vec![],
@@ -228,6 +230,7 @@ pub async fn get_related_for_assets(
 
     let grouping = asset_grouping::Entity::find()
         .filter(asset_grouping::Column::AssetId.is_in(ids.clone()))
+        .filter(asset_grouping::Column::Verified.eq(true))
         .order_by_asc(asset_grouping::Column::AssetId)
         .all(conn)
         .await?;
@@ -288,6 +291,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::Verified.eq(true))
         .all(conn)
         .await?;
     Ok(FullAsset {

digital_asset_types/src/dapi/common/asset.rs

@@ -2,24 +2,21 @@ use crate::dao::sea_orm_active_enums::SpecificationVersions;
 use crate::dao::FullAsset;
 use crate::dao::Pagination;
 use crate::dao::{asset, asset_authority, asset_creators, asset_data, asset_grouping};
-
 use crate::rpc::filter::{AssetSortBy, AssetSortDirection, AssetSorting};
 use crate::rpc::response::{AssetError, AssetList};
 use crate::rpc::{
     Asset as RpcAsset, Authority, Compression, Content, Creator, File, Group, Interface,
     MetadataMap, Ownership, Royalty, Scope, Supply, Uses,
 };
-
 use jsonpath_lib::JsonPathError;
+use log::warn;
 use mime_guess::Mime;
 use sea_orm::DbErr;
 use serde_json::Value;
 use std::collections::HashMap;
 use std::path::Path;
 use url::Url;
 
-use log::{debug, info, warn};
-
 pub fn to_uri(uri: String) -> Option<Url> {
     Url::parse(&*uri).ok()
 }
@@ -30,7 +27,9 @@ pub fn get_mime(url: Url) -> Option<Mime> {
 
 pub fn get_mime_type_from_uri(uri: String) -> String {
     let default_mime_type = "image/png".to_string();
-    to_uri(uri).and_then(get_mime).map_or(default_mime_type, |m| m.to_string())
+    to_uri(uri)
+        .and_then(get_mime)
+        .map_or(default_mime_type, |m| m.to_string())
 }
 
 pub fn file_from_str(str: String) -> File {
@@ -173,28 +172,25 @@ pub fn v1_content_from_json(asset_data: &asset_data::Model) -> Result<Content, D
                     match (uri, mime_type) {
                         (Some(u), Some(m)) => {
                             if let Some(str_uri) = u.as_str() {
-                                let file = 
-                                    if let Some(str_mime) = m.as_str() {
-                                        File {
-                                             uri: Some(str_uri.to_string()),
-                                             mime: Some(str_mime.to_string()),
-                                             quality: None,
-                                             contexts: None,
-                                        }
-                                    } else {
-                                        warn!("Mime is not string: {:?}", m); 
-                                        file_from_str(str_uri.to_string())
-                                    };
-                                actual_files.insert(
-                                   str_uri.to_string().clone(),
-                                   file,
-                                );
+                                let file = if let Some(str_mime) = m.as_str() {
+                                    File {
+                                        uri: Some(str_uri.to_string()),
+                                        mime: Some(str_mime.to_string()),
+                                        quality: None,
+                                        contexts: None,
+                                    }
+                                } else {
+                                    warn!("Mime is not string: {:?}", m);
+                                    file_from_str(str_uri.to_string())
+                                };
+                                actual_files.insert(str_uri.to_string().clone(), file);
                             } else {
                                 warn!("URI is not string: {:?}", u);
                             }
                         }
                         (Some(u), None) => {
-                            let str_uri = serde_json::to_string(u).unwrap_or_else(|_|String::new());
+                            let str_uri =
+                                serde_json::to_string(u).unwrap_or_else(|_| String::new());
                             actual_files.insert(str_uri.clone(), file_from_str(str_uri));
                         }
                         _ => {}
@@ -250,14 +246,18 @@ pub fn to_creators(creators: Vec<asset_creators::Model>) -> Vec<Creator> {
         .collect()
 }
 
-pub fn to_grouping(groups: Vec<asset_grouping::Model>) -> Vec<Group> {
-    groups
-        .iter()
-        .map(|a| Group {
-            group_key: a.group_key.clone(),
-            group_value: a.group_value.clone(),
+pub fn to_grouping(groups: Vec<asset_grouping::Model>) -> Result<Vec<Group>, DbErr> {
+    fn find_group(model: &asset_grouping::Model) -> Result<Group, DbErr> {
+        Ok(Group {
+            group_key: model.group_key.clone(),
+            group_value: model
+                .group_value
+                .clone()
+                .ok_or(DbErr::Custom("Group value not found".to_string()))?,
         })
-        .collect()
+    }
+
+    groups.iter().map(find_group).collect()
 }
 
 pub fn get_interface(asset: &asset::Model) -> Result<Interface, DbErr> {
@@ -286,7 +286,7 @@ pub fn asset_to_rpc(asset: FullAsset) -> Result<RpcAsset, DbErr> {
     } = asset;
     let rpc_authorities = to_authority(authorities);
     let rpc_creators = to_creators(creators);
-    let rpc_groups = to_grouping(groups);
+    let rpc_groups = to_grouping(groups)?;
     let interface = get_interface(&asset)?;
     let content = get_content(&asset, &data)?;
     let mut chain_data_selector_fn = jsonpath_lib::selector(&data.chain_data);