add partial account filters for near (#3669)

chain/near/src/adapter.rs

@@ -1,9 +1,10 @@
 use std::collections::HashSet;
 
 use crate::capabilities::NodeCapabilities;
+use crate::data_source::PartialAccounts;
 use crate::{data_source::DataSource, Chain};
 use graph::blockchain as bc;
-use graph::firehose::BasicReceiptFilter;
+use graph::firehose::{BasicReceiptFilter, PrefixSuffixPair};
 use graph::prelude::*;
 use prost::Message;
 use prost_types::Any;
@@ -54,6 +55,14 @@ impl bc::TriggerFilter<Chain> for TriggerFilter {
 
         let filter = BasicReceiptFilter {
             accounts: receipt.accounts.into_iter().collect(),
+            prefix_and_suffix_pairs: receipt
+                .partial_accounts
+                .iter()
+                .map(|(prefix, suffix)| PrefixSuffixPair {
+                    prefix: prefix.clone().unwrap_or("".to_string()),
+                    suffix: suffix.clone().unwrap_or("".to_string()),
+                })
+                .collect(),
         };
 
         vec![Any {
@@ -70,36 +79,125 @@ pub(crate) type Account = String;
 #[derive(Clone, Debug, Default)]
 pub(crate) struct NearReceiptFilter {
     pub accounts: HashSet<Account>,
+    pub partial_accounts: HashSet<(Option<String>, Option<String>)>,
 }
 
 impl NearReceiptFilter {
     pub fn matches(&self, account: &String) -> bool {
-        self.accounts.contains(account)
+        let NearReceiptFilter {
+            accounts,
+            partial_accounts,
+        } = self;
+
+        if accounts.contains(account) {
+            return true;
+        }
+
+        partial_accounts.iter().any(|partial| match partial {
+            (Some(prefix), Some(suffix)) => {
+                account.starts_with(prefix) && account.ends_with(suffix)
+            }
+            (Some(prefix), None) => account.starts_with(prefix),
+            (None, Some(suffix)) => account.ends_with(suffix),
+            (None, None) => unreachable!(),
+        })
     }
 
     pub fn is_empty(&self) -> bool {
-        let NearReceiptFilter { accounts } = self;
+        let NearReceiptFilter {
+            accounts,
+            partial_accounts,
+        } = self;
 
-        accounts.is_empty()
+        accounts.is_empty() && partial_accounts.is_empty()
     }
 
     pub fn from_data_sources<'a>(iter: impl IntoIterator<Item = &'a DataSource>) -> Self {
-        let accounts: Vec<String> = iter
+        struct Source {
+            account: Option<String>,
+            partial_accounts: Option<PartialAccounts>,
+        }
+
+        // Select any ds with either partial or exact accounts.
+        let sources: Vec<Source> = iter
             .into_iter()
             .filter(|data_source| {
-                data_source.source.account.is_some()
+                (data_source.source.account.is_some() || data_source.source.accounts.is_some())
                     && !data_source.mapping.receipt_handlers.is_empty()
             })
-            .map(|ds| ds.source.account.as_ref().unwrap().clone())
+            .map(|ds| Source {
+                account: ds.source.account.clone(),
+                partial_accounts: ds.source.accounts.clone(),
+            })
+            .collect();
+
+        // Handle exact matches
+        let accounts: Vec<String> = sources
+            .iter()
+            .filter(|s| s.account.is_some())
+            .map(|s| s.account.as_ref().cloned().unwrap())
+            .collect();
+
+        // Parse all the partial accounts, produces all possible combinations of the values
+        // eg:
+        // prefix [a,b] and suffix [d] would produce [a,d], [b,d]
+        // prefix [a] and suffix [c,d] would produce [a,c], [a,d]
+        // prefix [] and suffix [c, d] would produce [None, c], [None, d]
+        // prefix [a,b] and suffix [] would produce [a, None], [b, None]
+        let partial_accounts: Vec<(Option<String>, Option<String>)> = sources
+            .iter()
+            .filter(|s| s.partial_accounts.is_some())
+            .map(|s| {
+                let partials = s.partial_accounts.as_ref().unwrap();
+
+                let mut pairs: Vec<(Option<String>, Option<String>)> = vec![];
+                let prefixes: Vec<Option<String>> = if partials.prefixes.is_empty() {
+                    vec![None]
+                } else {
+                    partials
+                        .prefixes
+                        .iter()
+                        .filter(|s| !s.is_empty())
+                        .map(|s| Some(s.clone()))
+                        .collect()
+                };
+
+                let suffixes: Vec<Option<String>> = if partials.suffixes.is_empty() {
+                    vec![None]
+                } else {
+                    partials
+                        .suffixes
+                        .iter()
+                        .filter(|s| !s.is_empty())
+                        .map(|s| Some(s.clone()))
+                        .collect()
+                };
+
+                for prefix in prefixes.into_iter() {
+                    for suffix in suffixes.iter() {
+                        pairs.push((prefix.clone(), suffix.clone()))
+                    }
+                }
+
+                pairs
+            })
+            .flatten()
             .collect();
 
         Self {
             accounts: HashSet::from_iter(accounts),
+            partial_accounts: HashSet::from_iter(partial_accounts),
         }
     }
 
     pub fn extend(&mut self, other: NearReceiptFilter) {
-        self.accounts.extend(other.accounts);
+        let NearReceiptFilter {
+            accounts,
+            partial_accounts,
+        } = self;
+
+        accounts.extend(other.accounts);
+        partial_accounts.extend(other.partial_accounts);
     }
 }
 
@@ -130,7 +228,10 @@ mod test {
 
     use super::NearBlockFilter;
     use crate::adapter::{TriggerFilter, BASIC_RECEIPT_FILTER_TYPE_URL};
-    use graph::{blockchain::TriggerFilter as _, firehose::BasicReceiptFilter};
+    use graph::{
+        blockchain::TriggerFilter as _,
+        firehose::{BasicReceiptFilter, PrefixSuffixPair},
+    };
     use prost::Message;
     use prost_types::Any;
 
@@ -142,6 +243,7 @@ mod test {
             },
             receipt_filter: super::NearReceiptFilter {
                 accounts: HashSet::new(),
+                partial_accounts: HashSet::new(),
             },
         };
         assert_eq!(filter.to_firehose_filter(), vec![]);
@@ -155,6 +257,7 @@ mod test {
             },
             receipt_filter: super::NearReceiptFilter {
                 accounts: HashSet::from_iter(vec!["acc1".into(), "acc2".into(), "acc3".into()]),
+                partial_accounts: HashSet::new(),
             },
         };
 
@@ -170,6 +273,7 @@ mod test {
             },
             receipt_filter: super::NearReceiptFilter {
                 accounts: HashSet::from_iter(vec!["acc1".into(), "acc2".into(), "acc3".into()]),
+                partial_accounts: HashSet::new(),
             },
         };
 
@@ -188,6 +292,53 @@ mod test {
         );
     }
 
+    #[test]
+    fn near_trigger_partial_filter() {
+        let filter = TriggerFilter {
+            block_filter: NearBlockFilter {
+                trigger_every_block: false,
+            },
+            receipt_filter: super::NearReceiptFilter {
+                accounts: HashSet::from_iter(vec!["acc1".into()]),
+                partial_accounts: HashSet::from_iter(vec![
+                    (Some("acc1".into()), None),
+                    (None, Some("acc2".into())),
+                    (Some("acc3".into()), Some("acc4".into())),
+                ]),
+            },
+        };
+
+        let filter = filter.to_firehose_filter();
+        assert_eq!(filter.len(), 1);
+
+        let firehose_filter = decode_filter(filter);
+        assert_eq!(firehose_filter.accounts, vec![String::from("acc1"),],);
+
+        let expected_pairs = vec![
+            PrefixSuffixPair {
+                prefix: "acc3".to_string(),
+                suffix: "acc4".to_string(),
+            },
+            PrefixSuffixPair {
+                prefix: "".to_string(),
+                suffix: "acc2".to_string(),
+            },
+            PrefixSuffixPair {
+                prefix: "acc1".to_string(),
+                suffix: "".to_string(),
+            },
+        ];
+
+        let pairs = firehose_filter.prefix_and_suffix_pairs;
+        assert_eq!(pairs.len(), 3);
+        assert_eq!(
+            true,
+            expected_pairs.iter().all(|x| pairs.contains(x)),
+            "{:?}",
+            pairs
+        );
+    }
+
     fn decode_filter(firehose_filter: Vec<Any>) -> BasicReceiptFilter {
         let firehose_filter = firehose_filter[0].clone();
         assert_eq!(