From 3b26d20e3cc5be78f0c609fe4e9034f2fda70837 Mon Sep 17 00:00:00 2001 From: incrypto32 Date: Mon, 9 Sep 2024 16:18:01 +0530 Subject: [PATCH] graph, runtime: Support subgraph entity operation detection in composed subgraphs --- graph/src/blockchain/block_stream.rs | 32 ++++++------- graph/src/data_source/subgraph.rs | 24 +++++----- graph/src/runtime/mod.rs | 5 +++ runtime/wasm/src/module/mod.rs | 2 +- runtime/wasm/src/to_from/external.rs | 45 ++++++++++++++++++- .../subgraph-data-sources/src/mapping.ts | 12 ++++- .../subgraph-data-sources/src/mapping.ts | 33 +++++++++++++- tests/src/fixture/ethereum.rs | 19 +++++--- tests/tests/runner_tests.rs | 9 +++- 9 files changed, 135 insertions(+), 46 deletions(-) diff --git a/graph/src/blockchain/block_stream.rs b/graph/src/blockchain/block_stream.rs index 9f8b6fce8a9..8bef184c88e 100644 --- a/graph/src/blockchain/block_stream.rs +++ b/graph/src/blockchain/block_stream.rs @@ -350,14 +350,13 @@ impl TriggersAdapterWrapper { fn create_subgraph_trigger_from_entities( filter: &SubgraphFilter, - entities: &Vec, + entities: Vec, ) -> Vec { entities - .iter() - .map(|e| subgraph::TriggerData { + .into_iter() + .map(|entity| subgraph::TriggerData { source: filter.subgraph.clone(), - entity: e.entity.clone(), - entity_type: e.entity_type.as_str().to_string(), + entity, }) .collect() } @@ -366,7 +365,7 @@ async fn create_subgraph_triggers( logger: Logger, blocks: Vec, filter: &SubgraphFilter, - entities: BTreeMap>, + mut entities: BTreeMap>, ) -> Result>, Error> { let logger_clone = logger.cheap_clone(); @@ -374,17 +373,12 @@ async fn create_subgraph_triggers( .into_iter() .map(|block| { let block_number = block.number(); - match entities.get(&block_number) { - Some(e) => { - let trigger_data = create_subgraph_trigger_from_entities(filter, e); - BlockWithTriggers::new_with_subgraph_triggers( - block, - trigger_data, - &logger_clone, - ) - } - None => BlockWithTriggers::new_with_subgraph_triggers(block, vec![], &logger_clone), - } + let trigger_data = entities + .remove(&block_number) + .map(|e| create_subgraph_trigger_from_entities(filter, e)) + .unwrap_or_else(Vec::new); + + BlockWithTriggers::new_with_subgraph_triggers(block, trigger_data, &logger_clone) }) .collect(); @@ -426,14 +420,14 @@ async fn scan_subgraph_triggers( } } -#[derive(Debug)] +#[derive(Debug, Clone, Eq, PartialEq)] pub enum EntitySubgraphOperation { Create, Modify, Delete, } -#[derive(Debug)] +#[derive(Debug, Clone, Eq, PartialEq)] pub struct EntityWithType { pub entity_op: EntitySubgraphOperation, pub entity_type: EntityType, diff --git a/graph/src/data_source/subgraph.rs b/graph/src/data_source/subgraph.rs index 24bc34b9b94..f7124f307c1 100644 --- a/graph/src/data_source/subgraph.rs +++ b/graph/src/data_source/subgraph.rs @@ -1,9 +1,6 @@ use crate::{ - blockchain::{Block, Blockchain}, - components::{ - link_resolver::LinkResolver, - store::{BlockNumber, Entity}, - }, + blockchain::{block_stream::EntityWithType, Block, Blockchain}, + components::{link_resolver::LinkResolver, store::BlockNumber}, data::{subgraph::SPEC_VERSION_1_3_0, value::Word}, data_source, prelude::{DataSourceContext, DeploymentHash, Link}, @@ -76,7 +73,7 @@ impl DataSource { } let trigger_ref = self.mapping.handlers.iter().find_map(|handler| { - if handler.entity != trigger.entity_type { + if handler.entity != trigger.entity_type() { return None; } @@ -281,17 +278,16 @@ impl UnresolvedDataSourceTemplate { #[derive(Clone, PartialEq, Eq)] pub struct TriggerData { pub source: DeploymentHash, - pub entity: Entity, - pub entity_type: String, + pub entity: EntityWithType, } impl TriggerData { - pub fn new(source: DeploymentHash, entity: Entity, entity_type: String) -> Self { - Self { - source, - entity, - entity_type, - } + pub fn new(source: DeploymentHash, entity: EntityWithType) -> Self { + Self { source, entity } + } + + pub fn entity_type(&self) -> &str { + self.entity.entity_type.as_str() } } diff --git a/graph/src/runtime/mod.rs b/graph/src/runtime/mod.rs index d20d1eccde3..000fcd45b2a 100644 --- a/graph/src/runtime/mod.rs +++ b/graph/src/runtime/mod.rs @@ -368,6 +368,11 @@ pub enum IndexForAscTypeId { // ... // LastStarknetType = 4499, + + // Subgraph Data Source types + AscEntityTrigger = 4500, + + // Reserved discriminant space for a future blockchain type IDs: [4,500, 5,499] // // Generated with the following shell script: diff --git a/runtime/wasm/src/module/mod.rs b/runtime/wasm/src/module/mod.rs index 532f75d2660..fa40ab3a65d 100644 --- a/runtime/wasm/src/module/mod.rs +++ b/runtime/wasm/src/module/mod.rs @@ -76,7 +76,7 @@ impl ToAscPtr for subgraph::TriggerData { heap: &mut H, gas: &GasCounter, ) -> Result, HostExportError> { - asc_new(heap, &self.entity.sorted_ref(), gas).map(|ptr| ptr.erase()) + asc_new(heap, &self.entity, gas).map(|ptr| ptr.erase()) } } diff --git a/runtime/wasm/src/to_from/external.rs b/runtime/wasm/src/to_from/external.rs index f08eacee94f..9167b87b029 100644 --- a/runtime/wasm/src/to_from/external.rs +++ b/runtime/wasm/src/to_from/external.rs @@ -1,15 +1,18 @@ use ethabi; +use graph::blockchain::block_stream::{EntitySubgraphOperation, EntityWithType}; use graph::data::store::scalar::Timestamp; use graph::data::value::Word; use graph::prelude::{BigDecimal, BigInt}; use graph::runtime::gas::GasCounter; use graph::runtime::{ - asc_get, asc_new, AscIndexId, AscPtr, AscType, AscValue, HostExportError, ToAscObj, + asc_get, asc_new, AscIndexId, AscPtr, AscType, AscValue, HostExportError, IndexForAscTypeId, + ToAscObj, }; use graph::{data::store, runtime::DeterministicHostError}; use graph::{prelude::serde_json, runtime::FromAscObj}; use graph::{prelude::web3::types as web3, runtime::AscHeap}; +use graph_runtime_derive::AscType; use crate::asc_abi::class::*; @@ -463,3 +466,43 @@ where }) } } + +#[derive(Debug, Clone, Eq, PartialEq, AscType)] +pub enum AscSubgraphEntityOp { + Create, + Modify, + Delete, +} + +#[derive(AscType)] +pub struct AscEntityTrigger { + pub entity_op: AscSubgraphEntityOp, + pub entity_type: AscPtr, + pub entity: AscPtr, + pub vid: i64, +} + +impl ToAscObj for EntityWithType { + fn to_asc_obj( + &self, + heap: &mut H, + gas: &GasCounter, + ) -> Result { + let entity_op = match self.entity_op { + EntitySubgraphOperation::Create => AscSubgraphEntityOp::Create, + EntitySubgraphOperation::Modify => AscSubgraphEntityOp::Modify, + EntitySubgraphOperation::Delete => AscSubgraphEntityOp::Delete, + }; + + Ok(AscEntityTrigger { + entity_op, + entity_type: asc_new(heap, &self.entity_type.as_str(), gas)?, + entity: asc_new(heap, &self.entity.sorted_ref(), gas)?, + vid: self.vid, + }) + } +} + +impl AscIndexId for AscEntityTrigger { + const INDEX_ASC_TYPE_ID: IndexForAscTypeId = IndexForAscTypeId::AscEntityTrigger; +} diff --git a/tests/integration-tests/subgraph-data-sources/src/mapping.ts b/tests/integration-tests/subgraph-data-sources/src/mapping.ts index 0f2df0e4783..576f49f53f5 100644 --- a/tests/integration-tests/subgraph-data-sources/src/mapping.ts +++ b/tests/integration-tests/subgraph-data-sources/src/mapping.ts @@ -1,7 +1,17 @@ import { Entity, log } from '@graphprotocol/graph-ts'; import { MirrorBlock } from '../generated/schema'; -export function handleEntity(blockEntity: Entity): void { +export class EntityTrigger { + constructor( + public entityOp: u32, + public entityType: string, + public entity: Entity, + public vid: i64, + ) {} +} + +export function handleEntity(trigger: EntityTrigger): void { + let blockEntity = trigger.entity; let blockNumber = blockEntity.getBigInt('number'); let blockHash = blockEntity.getBytes('hash'); let id = blockEntity.getString('id'); diff --git a/tests/runner-tests/subgraph-data-sources/src/mapping.ts b/tests/runner-tests/subgraph-data-sources/src/mapping.ts index 2e1a5382af3..cd5c1d4dcd1 100644 --- a/tests/runner-tests/subgraph-data-sources/src/mapping.ts +++ b/tests/runner-tests/subgraph-data-sources/src/mapping.ts @@ -1,6 +1,35 @@ import { Entity, log } from '@graphprotocol/graph-ts'; -export function handleBlock(content: Entity): void { - let stringContent = content.getString('val'); +export const SubgraphEntityOpCreate: u32 = 0; +export const SubgraphEntityOpModify: u32 = 1; +export const SubgraphEntityOpDelete: u32 = 2; + +export class EntityTrigger { + constructor( + public entityOp: u32, + public entityType: string, + public entity: Entity, + public vid: i64, + ) {} +} + +export function handleBlock(content: EntityTrigger): void { + let stringContent = content.entity.getString('val'); log.info('Content: {}', [stringContent]); + log.info('EntityOp: {}', [content.entityOp.toString()]); + + switch (content.entityOp) { + case SubgraphEntityOpCreate: { + log.info('Entity created: {}', [content.entityType]); + break + } + case SubgraphEntityOpModify: { + log.info('Entity modified: {}', [content.entityType]); + break; + } + case SubgraphEntityOpDelete: { + log.info('Entity deleted: {}', [content.entityType]); + break; + } + } } diff --git a/tests/src/fixture/ethereum.rs b/tests/src/fixture/ethereum.rs index 5381a530148..50328f89a11 100644 --- a/tests/src/fixture/ethereum.rs +++ b/tests/src/fixture/ethereum.rs @@ -6,6 +6,7 @@ use super::{ test_ptr, CommonChainConfig, MutexBlockStreamBuilder, NoopAdapterSelector, NoopRuntimeAdapterBuilder, StaticBlockRefetcher, StaticStreamBuilder, Stores, TestChain, }; +use graph::blockchain::block_stream::{EntitySubgraphOperation, EntityWithType}; use graph::blockchain::client::ChainClient; use graph::blockchain::{BlockPtr, Trigger, TriggersAdapterSelector}; use graph::cheap_clone::CheapClone; @@ -13,6 +14,7 @@ use graph::data_source::subgraph; use graph::prelude::ethabi::ethereum_types::H256; use graph::prelude::web3::types::{Address, Log, Transaction, H160}; use graph::prelude::{ethabi, tiny_keccak, DeploymentHash, Entity, LightEthereumBlock, ENV_VARS}; +use graph::schema::EntityType; use graph::{blockchain::block_stream::BlockWithTriggers, prelude::ethabi::ethereum_types::U64}; use graph_chain_ethereum::network::EthereumNetworkAdapters; use graph_chain_ethereum::trigger::LogRef; @@ -164,15 +166,20 @@ pub fn push_test_subgraph_trigger( block: &mut BlockWithTriggers, source: DeploymentHash, entity: Entity, - entity_type: &str, + entity_type: EntityType, + entity_op: EntitySubgraphOperation, + vid: i64, ) { + let entity = EntityWithType { + entity: entity, + entity_type: entity_type, + entity_op: entity_op, + vid, + }; + block .trigger_data - .push(Trigger::Subgraph(subgraph::TriggerData { - source, - entity: entity, - entity_type: entity_type.to_string(), - })); + .push(Trigger::Subgraph(subgraph::TriggerData { source, entity })); } pub fn push_test_command( diff --git a/tests/tests/runner_tests.rs b/tests/tests/runner_tests.rs index 8f01e4a98f2..1692d8cc959 100644 --- a/tests/tests/runner_tests.rs +++ b/tests/tests/runner_tests.rs @@ -6,7 +6,7 @@ use std::sync::Arc; use std::time::Duration; use assert_json_diff::assert_json_eq; -use graph::blockchain::block_stream::BlockWithTriggers; +use graph::blockchain::block_stream::{BlockWithTriggers, EntitySubgraphOperation}; use graph::blockchain::{Block, BlockPtr, Blockchain}; use graph::data::store::scalar::Bytes; use graph::data::subgraph::schema::{SubgraphError, SubgraphHealth}; @@ -1109,14 +1109,19 @@ async fn subgraph_data_sources() { ]) .unwrap(); + let entity_type = schema.entity_type("User").unwrap(); + let blocks = { let block_0 = genesis(); let mut block_1 = empty_block(block_0.ptr(), test_ptr(1)); + push_test_subgraph_trigger( &mut block_1, DeploymentHash::new("QmRFXhvyvbm4z5Lo7z2mN9Ckmo623uuB2jJYbRmAXgYKXJ").unwrap(), entity, - "User", + entity_type, + EntitySubgraphOperation::Create, + 1, ); let block_2 = empty_block(block_1.ptr(), test_ptr(2));