ethereum: Re-work matching transaction logic

Currently the logic for matching transactions is super generic.  We
would like an API that accepts only what is needed to match a
transaction and returns what we need from the matched transaction.  From
the COMIT protocol level i.e., rfc003 this is a 'per action' API i.e.,
funded/deployed/redeem/refunded.

Pleas note; this patch removes all the integration and unit tests, these
will be added back in in coming patches.

The API is now

    matching_create_contract_transaction(...) -> Result<Transaction>
    matching_events_transaction(...) -> Result<Transaction, TranscationReceipt>

All the matching logic is contained within the `btsieve/ethereum`
module.

cnd/src/btsieve/ethereum/mod.rs

@@ -1,20 +1,16 @@
 mod cache;
-mod transaction_pattern;
 mod web3_connector;
 
-pub use self::{
-    cache::Cache,
-    transaction_pattern::{Event, Topic, TransactionPattern, TRANSACTION_STATUS_OK},
-    web3_connector::Web3Connector,
-};
+pub use self::{cache::Cache, web3_connector::Web3Connector};
 use crate::{
     btsieve::{BlockByHash, LatestBlock, Predates, ReceiptByHash},
-    ethereum::{Transaction, TransactionAndReceipt, TransactionReceipt, H256, U256},
+    ethereum::{Address, Bytes, Transaction, TransactionReceipt, H256, U256},
     Never,
 };
 use anyhow;
 use chrono::NaiveDateTime;
-use futures_core::compat::Future01CompatExt;
+use ethbloom::Input;
+use futures_core::{compat::Future01CompatExt, Future, FutureExt};
 use genawaiter::{
     sync::{Co, Gen},
     GeneratorState,
@@ -24,16 +20,134 @@ use std::collections::HashSet;
 type Hash = H256;
 type Block = crate::ethereum::Block<Transaction>;
 
-pub async fn matching_transaction<C>(
+pub const TRANSACTION_STATUS_OK: u32 = 1;
+
+pub async fn matching_create_contract_transaction<C>(
+    blockchain_connector: C,
+    start_of_swap: NaiveDateTime,
+    bytecode: Bytes,
+) -> anyhow::Result<Transaction>
+where
+    C: LatestBlock<Block = Option<Block>>
+        + BlockByHash<Block = Option<Block>, BlockHash = Hash>
+        + ReceiptByHash<Receipt = Option<TransactionReceipt>, TransactionHash = Hash>
+        + Clone,
+{
+    matching_transaction(blockchain_connector.clone(), start_of_swap, |transaction| {
+        Box::new({
+            let blockchain_connector = blockchain_connector.clone();
+            let bytecode = bytecode.clone();
+            async move {
+                // transaction.to address is None if, and only if, the transaction
+                // creates a contract.
+                if transaction.to.is_none() && transaction.input == bytecode {
+                    // We can get a transaction that matches even for failed attempts
+                    // at contract creation e.g., low gas.
+                    let receipt = fetch_receipt(blockchain_connector, transaction.hash).await?;
+                    if receipt.status == Some(TRANSACTION_STATUS_OK.into()) {
+                        return Ok(true);
+                    }
+                }
+                Ok(false)
+            }
+            .boxed()
+        })
+    })
+    .await
+}
+
+pub async fn matching_event_transaction<C>(
+    blockchain_connector: C,
+    start_of_swap: NaiveDateTime,
+    event: Event,
+    action: &'static str,
+) -> anyhow::Result<(Transaction, TransactionReceipt)>
+where
+    C: LatestBlock<Block = Option<Block>>
+        + BlockByHash<Block = Option<Block>, BlockHash = Hash>
+        + ReceiptByHash<Receipt = Option<TransactionReceipt>, TransactionHash = Hash>
+        + Clone,
+{
+    matching_transaction_and_receipt(
+        blockchain_connector.clone(),
+        start_of_swap,
+        event.topics.clone(),
+        action,
+        |transaction| {
+            Box::new({
+                let blockchain_connector = blockchain_connector.clone();
+                let event = event.clone();
+                async move {
+                    let receipt = fetch_receipt(blockchain_connector, transaction.hash).await?;
+                    if event_exists_in_receipt(&event, &receipt) {
+                        if receipt.status != Some(TRANSACTION_STATUS_OK.into()) {
+                            return Ok(None);
+                        }
+                        tracing::trace!("{} transaction matched {:x}", action, transaction.hash,);
+                        return Ok(Some((transaction, receipt)));
+                    }
+
+                    Ok(None)
+                }
+                .boxed()
+            })
+        },
+    )
+    .await
+}
+
+/// Fetch receipt from connector using transaction hash.
+async fn fetch_receipt<C>(blockchain_connector: C, hash: Hash) -> anyhow::Result<TransactionReceipt>
+where
+    C: ReceiptByHash<Receipt = Option<TransactionReceipt>, TransactionHash = Hash>,
+{
+    let receipt = blockchain_connector
+        .receipt_by_hash(hash)
+        .compat()
+        .await?
+        .ok_or_else(|| {
+            anyhow::anyhow!(
+                "Could not get transaction receipt for transaction {:x}",
+                hash
+            )
+        })?;
+    Ok(receipt)
+}
+
+fn event_exists_in_receipt(event: &Event, receipt: &TransactionReceipt) -> bool {
+    match event {
+        Event { topics, .. } if topics.is_empty() => false,
+        Event { address, topics } => receipt.logs.iter().any(|tx_log| {
+            if address
+                .as_ref()
+                .map_or(false, |address| address != &tx_log.address)
+            {
+                return false;
+            }
+
+            if tx_log.topics.len() == topics.len() {
+                tx_log.topics.iter().enumerate().all(|(index, tx_topic)| {
+                    let topic = &topics[index];
+                    topic.as_ref().map_or(true, |topic| tx_topic == &topic.0)
+                })
+            } else {
+                false
+            }
+        }),
+    }
+}
+
+async fn matching_transaction<C, F>(
     connector: C,
-    pattern: TransactionPattern,
     start_of_swap: NaiveDateTime,
-) -> anyhow::Result<TransactionAndReceipt>
+    matcher: F,
+) -> anyhow::Result<Transaction>
 where
     C: LatestBlock<Block = Option<Block>>
         + BlockByHash<Block = Option<Block>, BlockHash = Hash>
         + ReceiptByHash<Receipt = Option<TransactionReceipt>, TransactionHash = Hash>
         + Clone,
+    F: Fn(Transaction) -> Box<dyn Future<Output = anyhow::Result<bool>> + Unpin + Send>,
 {
     let mut block_generator = Gen::new({
         let connector = connector.clone();
@@ -43,13 +157,81 @@ where
     loop {
         match block_generator.async_resume().await {
             GeneratorState::Yielded(block) => {
-                if let Some(transaction_and_receipt) =
-                    check_block_against_pattern(connector.clone(), block, pattern.clone()).await?
-                {
-                    return Ok(transaction_and_receipt);
-                } else {
+                for transaction in block.transactions.into_iter() {
+                    if matcher(transaction.clone()).await? {
+                        return Ok(transaction.clone());
+                    }
+                }
+            }
+            GeneratorState::Complete(Err(e)) => return Err(e),
+            // By matching against the never type explicitly, we assert that the `Ok` value of the
+            // result is actually the never type and has not been changed since this
+            // line was written. The never type can never be constructed, so we cannot
+            // reach this line never anyway.
+            GeneratorState::Complete(Ok(never)) => match never {},
+        }
+    }
+}
+
+async fn matching_transaction_and_receipt<C, F>(
+    connector: C,
+    start_of_swap: NaiveDateTime,
+    topics: Vec<Option<Topic>>,
+    action: &str,
+    matcher: F,
+) -> anyhow::Result<(Transaction, TransactionReceipt)>
+where
+    C: LatestBlock<Block = Option<Block>>
+        + BlockByHash<Block = Option<Block>, BlockHash = Hash>
+        + ReceiptByHash<Receipt = Option<TransactionReceipt>, TransactionHash = Hash>
+        + Clone,
+    F: Fn(
+        Transaction,
+    ) -> Box<
+        dyn Future<Output = anyhow::Result<Option<(Transaction, TransactionReceipt)>>>
+            + Unpin
+            + Send,
+    >,
+{
+    let mut block_generator = Gen::new({
+        let connector = connector.clone();
+        |co| async move { find_relevant_blocks(connector, &co, start_of_swap).await }
+    });
+
+    loop {
+        match block_generator.async_resume().await {
+            GeneratorState::Yielded(block) => {
+                let block_hash = block
+                    .hash
+                    .ok_or_else(|| anyhow::anyhow!("block without hash"))?;
+
+                let maybe_contains_transaction = topics.iter().all(|topic| {
+                    topic.as_ref().map_or(true, |topic| {
+                        block
+                            .logs_bloom
+                            .contains_input(Input::Raw(topic.0.as_ref()))
+                    })
+                });
+                if !maybe_contains_transaction {
+                    tracing::trace!(
+                        "bloom filter indicates that block does not contain {} transaction:
+                {:x}",
+                        action,
+                        block_hash,
+                    );
                     continue;
                 }
+
+                tracing::trace!(
+                    "bloom filter indicates that we should check the block for {} transactions: {:x}",
+                    action,
+                    block_hash,
+                );
+                for transaction in block.transactions.into_iter() {
+                    if let Some(transaction_and_receipt) = matcher(transaction.clone()).await? {
+                        return Ok(transaction_and_receipt);
+                    }
+                }
             }
             GeneratorState::Complete(Err(e)) => return Err(e),
             // By matching against the never type explicitly, we assert that the `Ok` value of the
@@ -192,71 +374,47 @@ fn seen_block_or_predates_start_of_swap(
     }
 }
 
-async fn check_block_against_pattern<C>(
-    connector: C,
-    block: Block,
-    pattern: TransactionPattern,
-) -> anyhow::Result<Option<TransactionAndReceipt>>
-where
-    C: ReceiptByHash<Receipt = Option<TransactionReceipt>, TransactionHash = Hash>,
-{
-    let needs_receipt = pattern.needs_receipts(&block);
-    let block_hash = block
-        .hash
-        .ok_or_else(|| anyhow::anyhow!("block without hash"))?;
-
-    if needs_receipt {
-        tracing::debug!(
-            "bloom-filter of block {:x} suggests to fetch receipts for {:?}",
-            block_hash,
-            pattern
-        );
-    } else {
-        tracing::debug!(
-            "bloom-filter of block {:x} suggests to not fetch receipts for {:?}",
-            block_hash,
-            pattern
-        );
-    }
-
-    for transaction in block.transactions.into_iter() {
-        let tx_hash = transaction.hash;
-
-        let receipt = connector
-            .receipt_by_hash(tx_hash)
-            .compat()
-            .await?
-            .ok_or_else(|| {
-                anyhow::anyhow!(
-                    "Could not get transaction receipt for transaction {:x}",
-                    tx_hash
-                )
-            })?;
-
-        let result = pattern.matches(&transaction, &receipt);
-
-        tracing::debug!(
-            "matching {:?} against transaction {:x} yielded {}",
-            pattern,
-            tx_hash,
-            result
-        );
-
-        if result {
-            return Ok(Some(TransactionAndReceipt {
-                transaction,
-                receipt,
-            }));
-        }
-    }
-
-    Ok(None)
-}
-
 impl Predates for Block {
     fn predates(&self, timestamp: NaiveDateTime) -> bool {
         let unix_timestamp = timestamp.timestamp();
 
         self.timestamp < U256::from(unix_timestamp)
     }
 }
+
+#[derive(Clone, Copy, Default, Eq, PartialEq, serde::Serialize, serdebug::SerDebug)]
+#[serde(transparent)]
+pub struct Topic(pub H256);
+
+/// Event works similar to web3 filters:
+/// https://web3js.readthedocs.io/en/1.0/web3-eth-subscribe.html?highlight=filter#subscribe-logs
+/// E.g. this `Event` would match this `Log`:
+/// ```rust, ignore
+/// Event {
+/// address: "0xe46FB33e4DB653De84cB0E0E8b810A6c4cD39d59",
+/// topics: [
+/// None,
+/// 0x000000000000000000000000e46fb33e4db653de84cb0e0e8b810a6c4cd39d59,
+/// None,
+/// ],
+/// ```
+/// ```rust, ignore
+/// Log:
+/// [ { address: "0xe46FB33e4DB653De84cB0E0E8b810A6c4cD39d59",
+/// data: "0x123",
+/// ..
+/// topics:
+/// [ "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",
+/// "0x000000000000000000000000e46fb33e4db653de84cb0e0e8b810a6c4cd39d59",
+/// "0x000000000000000000000000d51ecee7414c4445534f74208538683702cbb3e4" ],
+/// },
+/// .. ] //Other data omitted
+/// }
+/// ```
+#[derive(Clone, Default, Eq, PartialEq, serde::Serialize, serdebug::SerDebug)]
+pub struct Event {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub address: Option<Address>,
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    pub topics: Vec<Option<Topic>>,
+}