refactor(indexer): event filtering and related substate storing (#1043)

Description
---
* New configuration in the indexer for event filtering
* Event filtering logic in the `event_scanner` module
* Removal of all the code related to substate address monitoring
* If an event is filtered to be stored in the database, the related
substate (if any) will also be stored in the database automatically. The
related substate is extracted from the corresponding transaction result.
* Refactor of the `get_non_fungibles` functionality to fetch all non
fungibles of a collection from the network on the fly
* Refactored the `event_manager` and `substate_manager` to simplify and
properly split them, as a lot of functionality was duplicated and
overlapped between the two modules.
* Updated TypeScript bindings

Motivation and Context
---
Up until now, the behaviour of the indexer regarding events and
substates was:
* Storing in the database _all_ events in the network, with no
possibility of filtering.
* Allowing to store substates in the database via the _user manually
adding addresses to a watchlist_ using JRPC methods.

This PR refactors the indexer behaviour to be more useful:
* Allowing the user to **filter which events** in the network we want to
store in the database.
* Automatically **storing the related substate of each event** in the
database. Thus removing all the address watchlist functionality from the
indexer, as we will be using the new event filters to decide what to
store.

Regarding the event filtering:
* The list of event filters is static, it can be edited in the **indexer
config file**. Any change in event filtering requires a restart of the
indexer.
* It allows to specify a list of **event filters.** If a new event
matches _at least_ one filter, it will be stored in the database.
* Each filter allows to specify an AND condition of values for `topic`,
`substate_id`, `template_address` and `entity_id`. All filter fields are
optional. An event will match a filter if all of its fields match the
values specified in the filter.
* The `entity_id` one can be used when we want to store everything that
happens in a particular EntityId no matter the type of substate
(component, vault or resource).
* The default behaviour is an empty filter which matches all events,
making it store all events that happen in the network.

Regarding the substate storing:
* We can fetch the substate data from the transaction result
(`up_substates` field) corresponding to the event we are storing. This
is a handy optimization because it avoids a query to the network to
fetch the related substate.
* Because we are constantly fetching new events and extracting the
related substate from the transaction results, it means we don't need to
keep monitoring for the latest versions of each substate.
* In the database we only store the latest version of each substate.

This PR **removes functionality regarding substate monitoring and
address watchlist**, as now the substate storing is determined by the
events.

This PR keeps existing functionality regarding:
* Arbitrary substate fetching via the JRPC `get_substate` method. Even
if the substate is not stored in the database, the indexer will try to
fetch it from the network.
* Listing all NFTs from a collection via the `get_non_fungibles` method.
This functionality requires to fetch the non fungibles from the network
on the fly, as currently we don't emit the proper events regarding NFT
minting. Even if we do in a future, we would need to account for NFT
burning (thus deleting the substate from the database in that case)

How Has This Been Tested?
---
Manually by:
* Spawning a local network using `tari_swarm`.
* Using different `event_filters` configurations in the indexer.
* Doing transactions (e.g. transfers) and inspecting which events and
substates where stored in the indexer database.

What process can a PR reviewer use to test or verify this change?
---
See previous section.

Breaking Changes
---

- [x] None
- [ ] Requires data directory to be deleted
- [ ] Other - Please specify

applications/tari_dan_app_utilities/config_presets/d_indexer.toml

@@ -35,11 +35,30 @@
 # HTTP UI listener address (default = "127.0.0.1:15000")
 #http_ui_address = "127.0.0.1:15000"
 
-# Substate ids to keep watching
-#address_watchlist=[]
-
 # How often do we want to scan the dan layer for change. (default = 10)
 #dan_layer_scanning_internal=10
 
 [indexer.p2p]
 #transport = "tor"
+
+
+# List of filters for events that we want to persist in the indexer database
+# If an event matches ANY of the filters, it will be persisted
+# We can have as many "[[indexer.event_filters]]" sections as we want, each one will be a filter
+
+# By default there is an empty filter that will match ALL events, making the indexer persist the entire network
+[[indexer.event_filters]]
+
+# Fields are "topic", "entity_id", "substate_id" and "template_address"
+#[[indexer.event_filters]]
+#topic = "another topic"
+#entity_id = "0000000000000000000000000000000000000000"
+#substate_id = "component_00000000000000000000000000000000000000000000000000000000"
+#template_address = "0000000000000000000000000000000000000000000000000000000000000000"
+
+# Each individual filter is an "AND" condition, meaning all the specified fields in the filter must match
+# But we can specify only a subset of fields
+#[[indexer.event_filters]]
+#topic = "another topic"
+#substate_id = "component_00000000000000000000000000000000000000000000000000000000"
+

applications/tari_indexer/Cargo.toml

@@ -56,7 +56,6 @@ log4rs = { workspace = true, features = [
     "fixed_window_roller",
 ] }
 mime_guess = { workspace = true }
-rand = { workspace = true }
 reqwest = { workspace = true }
 serde = { workspace = true, features = ["default", "derive"] }
 serde_json = { workspace = true }

applications/tari_indexer/src/config.rs

@@ -39,7 +39,6 @@ use tari_dan_app_utilities::{
     p2p_config::{P2pConfig, PeerSeedsConfig},
     template_manager::implementation::TemplateConfig,
 };
-use tari_engine_types::substate::SubstateId;
 
 #[derive(Debug, Clone)]
 pub struct ApplicationConfig {
@@ -89,8 +88,6 @@ pub struct IndexerConfig {
     /// The jrpc address where the UI should connect (it can be the same as the json_rpc_address, but doesn't have to
     /// be), if this will be None, then the listen_addr will be used.
     pub ui_connect_address: Option<String>,
-    /// Substate ids to keep watching
-    pub address_watchlist: Vec<SubstateId>,
     /// How often do we want to scan the second layer for new versions
     #[serde(with = "serializers::seconds")]
     pub dan_layer_scanning_internal: Duration,
@@ -100,8 +97,10 @@ pub struct IndexerConfig {
     pub sidechain_id: Option<RistrettoPublicKey>,
     /// The templates sidechain id
     pub templates_sidechain_id: Option<RistrettoPublicKey>,
-    /// the burnt utxos sidechain id
+    /// The burnt utxos sidechain id
     pub burnt_utxo_sidechain_id: Option<RistrettoPublicKey>,
+    /// The event filtering configuration
+    pub event_filters: Vec<EventFilterConfig>,
 }
 
 impl IndexerConfig {
@@ -136,12 +135,12 @@ impl Default for IndexerConfig {
             graphql_address: Some("127.0.0.1:18301".parse().unwrap()),
             http_ui_address: Some("127.0.0.1:15000".parse().unwrap()),
             ui_connect_address: None,
-            address_watchlist: vec![],
             dan_layer_scanning_internal: Duration::from_secs(10),
             templates: TemplateConfig::default(),
             sidechain_id: None,
             templates_sidechain_id: None,
             burnt_utxo_sidechain_id: None,
+            event_filters: vec![],
         }
     }
 }
@@ -151,3 +150,11 @@ impl SubConfigPath for IndexerConfig {
         "indexer"
     }
 }
+
+#[derive(Default, Debug, Serialize, Deserialize, Clone)]
+pub struct EventFilterConfig {
+    pub topic: Option<String>,
+    pub entity_id: Option<String>,
+    pub substate_id: Option<String>,
+    pub template_address: Option<String>,
+}

applications/tari_indexer/src/event_data.rs

@@ -0,0 +1,30 @@
+//   Copyright 2024. The Tari Project
+//
+//   Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+//   following conditions are met:
+//
+//   1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+//   disclaimer.
+//
+//   2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
+//   following disclaimer in the documentation and/or other materials provided with the distribution.
+//
+//   3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
+//   products derived from this software without specific prior written permission.
+//
+//   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+//   INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+//   DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+//   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+//   SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+//   WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+//   USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+use serde::{Deserialize, Serialize};
+use tari_engine_types::{events::Event, substate::Substate};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct EventData {
+    pub event: Event,
+    pub substate: Option<Substate>,
+}