swapd/src/darkfi/follower_event_watcher.rs

use std::time::{SystemTime, UNIX_EPOCH};

use crate::{
    darkfi::Error,
    ethereum::swap_creator::SwapCreator,
    protocol::{
        follower::Event,
        traits::{CounterpartyKeys, FollowerEventWatcher},
    },
};
use ethers::prelude::Middleware;
use smol::{channel, stream::StreamExt as _};

pub(crate) struct Watcher;

#[darkfi_serial::async_trait]
impl FollowerEventWatcher for Watcher {
    async fn run_counterparty_funds_locked_watcher<M: Middleware>(
        event_tx: channel::Sender<Event>,
        contract: SwapCreator<M>,
        claim_commitment: [u8; 32],
        refund_commitment: [u8; 32],
        from_block: u64,
    ) -> Result<(), crate::Error> {
        // watch for a `NewSwap` event with the correct swap parameters
        // note: we still need to check for correct asset, value, and timeout,
        // which is done in the event handler [`Follower`].
        let topic2: ethers::types::U256 = claim_commitment.into();
        let topic3: ethers::types::U256 = refund_commitment.into();
        let events = contract
            .ready_filter()
            .from_block(from_block)
            .address(contract.address().into())
            .topic2(topic2) // `newSwap` event sig is topic0 and `contract_swap_id` is topic1
            .topic3(topic3);

        let mut stream = events.stream().await.unwrap().with_meta();

        // we listen for the first event, as there can only be one event
        // that matches the filter (ie. has the same swap_id)
        let Some(Ok((event, _meta))) = stream.next().await else {
            return Err(Error::ReadyEventStreamFailed.into());
        };

        let contract_swap_id = event.swap_id;
        event_tx.send(Event::CounterpartyFundsLocked(contract_swap_id)).await.unwrap();
        Ok(())
    }

    async fn run_ready_to_claim_watcher<M: Middleware>(
        event_tx: channel::Sender<Event>,
        contract: SwapCreator<M>,
        contract_swap_id: &[u8; 32],
        from_block: u64,
    ) -> Result<(), crate::Error> {
        let topic1: ethers::types::U256 = contract_swap_id.into();
        let events = contract
            .ready_filter()
            .from_block(from_block)
            .address(contract.address().into())
            .topic1(topic1); // `ready` event sig is topic0

        let mut stream = events.stream().await.unwrap().with_meta();

        // we listen for the first event, as there can only be one event
        // that matches the filter (ie. has the same swap_id)
        let Some(Ok((_, _meta))) = stream.next().await else {
            return Err(Error::ReadyEventStreamFailed.into());
        };

        event_tx.send(Event::ReadyToClaim).await.unwrap();
        Ok(())
    }

    async fn run_counterparty_funds_refunded_watcher<M: Middleware>(
        event_tx: channel::Sender<Event>,
        contract: SwapCreator<M>,
        contract_swap_id: &[u8; 32],
        from_block: u64,
    ) -> Result<(), crate::Error> {
        let topic1: ethers::types::U256 = contract_swap_id.into();
        let events = contract
            .refunded_filter()
            .from_block(from_block)
            .address(contract.address().into())
            .topic1(topic1); // `refunded` event sig is topic0

        let mut stream = events.stream().await.unwrap().with_meta();

        // we listen for the first event, as there can only be one event
        // that matches the filter (ie. has the same swap_id)
        let Some(Ok((event, _meta))) = stream.next().await else {
            return Err(Error::RefundedEventStreamFailed.into());
        };

        let counterparty_secret = event.s;
        event_tx.send(Event::CounterpartyFundsRefunded(counterparty_secret)).await.unwrap();
        Ok(())
    }
}