begin swap implemenatation

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "swapd"
-version = "0.4.1"
+version = "0.1.0"
 homepage = "https://dark.fi"
 description = "Atomic Swap Daemon"
 authors = ["Dyne.org foundation <foundation@dyne.org>"]
@@ -9,8 +9,8 @@ license = "AGPL-3.0-only"
 edition = "2021"
 
 [dependencies]
-darkfi = {git = "https://codeberg.org/darkrenaissance/darkfi", features = ["async-daemonize", "async-serial", "system", "util", "net", "rpc", "sled"]}
-darkfi-serial = {git = "https://codeberg.org/darkrenaissance/darkfi", features = ["async"]}
+darkfi = { git = "https://github.com/darkrenaissance/darkfi", features = ["async-daemonize", "async-serial", "system", "util", "net", "rpc", "sled"] }
+darkfi-serial = { git = "https://github.com/darkrenaissance/darkfi", features = ["async"] }
 
 # Misc
 log = "0.4.21"
@@ -32,3 +32,15 @@ smol = "1.3.0"
 serde = {version = "1.0.197", features = ["derive"]}
 structopt = "0.3.26"
 structopt-toml = "0.5.1"
+
+async-watch = "0.3.1"
+ethers = { version = "2.0", features = ["ethers-solc", "ws"] }
+thiserror = "1.0.57"
+
+[dev-dependencies]
+swapd = { path = ".", features = ["test-utils"] }
+async-std = {version = "1.12.0", features = ["attributes", "tokio1"]}
+
+[features]
+default = []
+test-utils = []

ethereum/.github/workflows/test.yml

@@ -0,0 +1,34 @@
+name: test
+
+on: workflow_dispatch
+
+env:
+  FOUNDRY_PROFILE: ci
+
+jobs:
+  check:
+    strategy:
+      fail-fast: true
+
+    name: Foundry project
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          submodules: recursive
+
+      - name: Install Foundry
+        uses: foundry-rs/foundry-toolchain@v1
+        with:
+          version: nightly
+
+      - name: Run Forge build
+        run: |
+          forge --version
+          forge build --sizes
+        id: build
+
+      - name: Run Forge tests
+        run: |
+          forge test -vvv
+        id: test

ethereum/.gitignore

@@ -0,0 +1,14 @@
+# Compiler files
+cache/
+out/
+
+# Ignores development broadcast logs
+!/broadcast
+/broadcast/*/31337/
+/broadcast/**/dry-run/
+
+# Docs
+docs/
+
+# Dotenv file
+.env

ethereum/README.md

@@ -0,0 +1,66 @@
+## Foundry
+
+**Foundry is a blazing fast, portable and modular toolkit for Ethereum application development written in Rust.**
+
+Foundry consists of:
+
+-   **Forge**: Ethereum testing framework (like Truffle, Hardhat and DappTools).
+-   **Cast**: Swiss army knife for interacting with EVM smart contracts, sending transactions and getting chain data.
+-   **Anvil**: Local Ethereum node, akin to Ganache, Hardhat Network.
+-   **Chisel**: Fast, utilitarian, and verbose solidity REPL.
+
+## Documentation
+
+https://book.getfoundry.sh/
+
+## Usage
+
+### Build
+
+```shell
+$ forge build
+```
+
+### Test
+
+```shell
+$ forge test
+```
+
+### Format
+
+```shell
+$ forge fmt
+```
+
+### Gas Snapshots
+
+```shell
+$ forge snapshot
+```
+
+### Anvil
+
+```shell
+$ anvil
+```
+
+### Deploy
+
+```shell
+$ forge script script/Counter.s.sol:CounterScript --rpc-url <your_rpc_url> --private-key <your_private_key>
+```
+
+### Cast
+
+```shell
+$ cast <subcommand>
+```
+
+### Help
+
+```shell
+$ forge --help
+$ anvil --help
+$ cast --help
+```

ethereum/foundry.toml

@@ -0,0 +1,6 @@
+[profile.default]
+src = "src"
+out = "out"
+libs = ["lib"]
+
+# See more config options https://github.com/foundry-rs/foundry/blob/master/crates/config/README.md#all-options

ethereum/src/Secp256k1.sol

@@ -0,0 +1,20 @@
+// SPDX-License-Identifier: LGPLv3
+// Implemention based on Vitalik's idea:
+// https://ethresear.ch/t/you-can-kinda-abuse-ecrecover-to-do-ecmul-in-secp256k1-today
+
+pragma solidity ^0.8.20;
+
+contract Secp256k1 {
+    // solhint-disable-next-line
+    uint256 private constant gx =
+        0x79BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798;
+    // solhint-disable-next-line
+    uint256 private constant m = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141;
+
+    // mulVerify returns true if `Q = s * G` on the secp256k1 curve
+    // qKeccak is defined as uint256(keccak256(abi.encodePacked(qx, qy))
+    function mulVerify(uint256 scalar, uint256 qKeccak) public pure returns (bool) {
+        address qRes = ecrecover(0, 27, bytes32(gx), bytes32(mulmod(scalar, gx, m)));
+        return uint160(qKeccak) == uint160(qRes);
+    }
+}

ethereum/src/SwapCreator.sol

@@ -0,0 +1,333 @@
+// SPDX-License-Identifier: LGPLv3
+pragma solidity ^0.8.20;
+
+import {IERC20} from "../lib/openzeppelin-contracts/contracts/token/ERC20/IERC20.sol";
+import {SafeERC20} from "../lib/openzeppelin-contracts/contracts/token/ERC20/utils/SafeERC20.sol";
+import {Secp256k1} from "./Secp256k1.sol";
+
+// SwapCreator facilitates swapping between Alice, a party that has an EVM
+// native currency or a token (ERC-20 or compatible API) that she wants to
+// exchange cross-chain for a different currency, and Bob, a party that has the
+// other chain's currency and wishes to exchange it for Alice's currency.
+contract SwapCreator is Secp256k1 {
+    using SafeERC20 for IERC20;
+
+    // Stage represents the swap state. It is PENDING when `newSwap` is called
+    // to create and fund the swap. Alice sets Stage to READY, via `setReady`,
+    // after verifying that funds are locked on the other chain. Bob cannot
+    // claim the swap funds until Alice sets the swap Stage to READY. The Stage
+    // is set to COMPLETED when Bob claims directly via `claim` or indirectly
+    // via `claimRelayer`, or by Alice calling `refund`.
+    enum Stage {
+        INVALID,
+        PENDING,
+        READY,
+        COMPLETED
+    }
+
+    // swaps maps from a swap ID to the swap's current Stage
+    mapping(bytes32 => Stage) public swaps;
+
+    // Swap stores the swap parameters, the hash of which forms the swap ID.
+    struct Swap {
+        // owner is the address of Alice, who initiates the swap by calling
+        // `newSwap`. Only the owner is allowed to call `setReady` or `refund`.
+        address payable owner;
+        // claimer is the address of Bob. Only the claimer can call `claim` or
+        // sign a RelaySwap object that `claimRelayer` will accept the signature
+        // for.
+        address payable claimer;
+        // claimCommitment is the Keccak-256 hash of the expected secp256k1
+        // public key derived from the secret (private key) that Bob sends when
+        // claiming. Alice receives this commitment off-chain.
+        bytes32 claimCommitment;
+        // refundCommitment is the Keccak-256 hash of the expected secp256k1
+        // public key derived from the secret (private key) that Alice sends if
+        // refunding.
+        bytes32 refundCommitment;
+        // timeout1 is the block timestamp before which Alice can call
+        // either `setReady` or `refund`.
+        uint256 timeout1;
+        // timeout2 is the block timestamp after which Bob cannot claim, only
+        // Alice can refund.
+        uint256 timeout2;
+        // asset is address(0) for EVM native currency swaps, or it is the
+        // address of the token that Alice is providing.
+        address asset;
+        // value is the wei or token unit amount that Alice locked in the contract
+        uint256 value;
+        // nonce is a random value chosen by Alice
+        uint256 nonce;
+    }
+
+    // RelaySwap contains additional information required for relayed claim
+    // transactions. This entire structure is encoded and signed by the swap
+    // claimer, and the signature is passed to `claimRelayer`.
+    struct RelaySwap {
+        // swap specifies which swap is being claimed
+        Swap swap;
+        // fee is the wei amount paid to the relayer
+        uint256 fee;
+        // relayerHash Keccak-256 hash of (relayer's payout address || 4-byte salt)
+        bytes32 relayerHash;
+        // swapCreator is the address of the swap's contract
+        address swapCreator;
+    }
+
+    event New(
+        bytes32 swapID,
+        bytes32 claimKey,
+        bytes32 refundKey,
+        uint256 timeout1,
+        uint256 timeout2,
+        address asset,
+        uint256 value
+    );
+    event Ready(bytes32 indexed swapID);
+    event Claimed(bytes32 indexed swapID, bytes32 indexed s);
+    event Refunded(bytes32 indexed swapID, bytes32 indexed s);
+
+    // thrown when the value parameter to `newSwap` is zero
+    error ZeroValue();
+
+    // thrown when either of the claimCommitment or refundCommitment parameters
+    // passed to `newSwap` are zero
+    error InvalidSwapKey();
+
+    // thrown when the claimer parameter for `newSwap` is the zero address
+    error InvalidClaimer();
+
+    // thrown when the timeout1 or timeout2 parameters for `newSwap` are zero
+    error InvalidTimeout();
+
+    // thrown when msg.value of a `newSwap` transaction has the wrong value
+    error InvalidValue();
+
+    // thrown when trying to initiate a swap with an ID that already exists
+    error SwapAlreadyExists();
+
+    // thrown when trying to call `setReady` on a swap that is not in the
+    // PENDING stage
+    error SwapNotPending();
+
+    // thrown when the caller of `setReady` or `refund` is not the swap owner
+    error OnlySwapOwner();
+
+    // thrown when the signer of the relayed transaction is not the swap's
+    // claimer
+    error OnlySwapClaimer();
+
+    // thrown when trying to call `claim` or `refund` on an invalid swap
+    error InvalidSwap();
+
+    // thrown when trying to call `claim` or `refund` on a swap that's already
+    // completed
+    error SwapCompleted();
+
+    // thrown when trying to call `claim` on a swap that's not set to ready or
+    // the first timeout has not been reached
+    error TooEarlyToClaim();
+
+    // thrown when trying to call `claim` on a swap where the second timeout has
+    // been reached
+    error TooLateToClaim();
+
+    // thrown when it's the counterparty's turn to claim and refunding is not
+    // allowed
+    error NotTimeToRefund();
+
+    // thrown when the provided secret does not match its expected public key
+    // hash
+    error InvalidSecret();
+
+    // thrown when the signature of a `RelaySwap` is invalid
+    error InvalidSignature();
+
+    // thrown when the SwapCreator address is a `RelaySwap` is not the address
+    // of this contract
+    error InvalidContractAddress();
+
+    // thrown when the hash of the relayer address and salt passed to
+    // `claimRelayer` does not match the relayer hash in `RelaySwap`
+    error InvalidRelayerAddress();
+
+    // `newSwap` creates a new Swap instance using the passed parameters and
+    // locks Alice's native EVM currency or token asset in the contract. On
+    // success, the swap ID is returned.
+    //
+    // Note that the duration values are distinct from the timeout values:
+    //
+    //   _timeoutDuration1:
+    //      duration, in seconds, between the current block timestamp and
+    //      timeout1
+    //
+    //   _timeoutDuration2:
+    //      duration, in seconds, between timeout1 and timeout2
+    //
+    function newSwap(
+        bytes32 _claimCommitment,
+        bytes32 _refundCommitment,
+        address payable _claimer,
+        uint256 _timeoutDuration1,
+        uint256 _timeoutDuration2,
+        address _asset,
+        uint256 _value,
+        uint256 _nonce
+    ) public payable returns (bytes32) {
+        if (_value == 0) revert ZeroValue();
+        if (_asset == address(0)) {
+            if (_value != msg.value) revert InvalidValue();
+        } else {
+            // transfer the token amount to this contract
+            // WARN: fee-on-transfer tokens are not supported
+            IERC20(_asset).safeTransferFrom(msg.sender, address(this), _value);
+        }
+
+        if (_claimCommitment == 0 || _refundCommitment == 0) revert InvalidSwapKey();
+        if (_claimer == address(0)) revert InvalidClaimer();
+        if (_timeoutDuration1 == 0 || _timeoutDuration2 == 0) revert InvalidTimeout();
+
+        Swap memory swap = Swap({
+            owner: payable(msg.sender),
+            claimCommitment: _claimCommitment,
+            refundCommitment: _refundCommitment,
+            claimer: _claimer,
+            timeout1: block.timestamp + _timeoutDuration1,
+            timeout2: block.timestamp + _timeoutDuration1 + _timeoutDuration2,
+            asset: _asset,
+            value: _value,
+            nonce: _nonce
+        });
+
+        bytes32 swapID = keccak256(abi.encode(swap));
+
+        // ensure that we are not overriding an existing swap
+        if (swaps[swapID] != Stage.INVALID) revert SwapAlreadyExists();
+
+        emit New(
+            swapID,
+            _claimCommitment,
+            _refundCommitment,
+            swap.timeout1,
+            swap.timeout2,
+            swap.asset,
+            swap.value
+        );
+        swaps[swapID] = Stage.PENDING;
+        return swapID;
+    }
+
+    // Alice should call `setReady` before timeout1 and after verifying that Bob
+    // locked his swap funds.
+    function setReady(Swap memory _swap) public {
+        bytes32 swapID = keccak256(abi.encode(_swap));
+        if (swaps[swapID] != Stage.PENDING) revert SwapNotPending();
+        if (_swap.owner != msg.sender) revert OnlySwapOwner();
+        swaps[swapID] = Stage.READY;
+        emit Ready(swapID);
+    }
+
+    // Bob can call `claim` if either of these hold true:
+    // (1) Alice has set the swap to `ready` and it's before timeout1
+    // (2) It is between timeout1 and timeout2
+    function claim(Swap memory _swap, bytes32 _secret) public {
+        if (msg.sender != _swap.claimer) revert OnlySwapClaimer();
+        _claim(_swap, _secret);
+
+        if (_swap.asset == address(0)) {
+            // Transfer the swap value as the EVM's native currency
+            _swap.claimer.transfer(_swap.value);
+        } else {
+            // Transfer the swap value as a token amount.
+            // WARNING: this will FAIL for fee-on-transfer or rebasing tokens if
+            // the token transfer reverts (i.e. if this contract does not
+            // contain _swap.value tokens), exposing Bob's secret while giving
+            // him nothing.
+            IERC20(_swap.asset).safeTransfer(_swap.claimer, _swap.value);
+        }
+    }
+
+    // Anyone can call `claimRelayer` if they receive a signed _relaySwap object
+    // from Bob. The same rules for when Bob can call `claim` apply here when a
+    // 3rd party relays a claim for Bob. This version of claiming transfers a
+    // _relaySwap.fee to _relayer. To prevent front-running, while not requiring
+    // Bob to know the relayer's payout address, Bob only signs a salted hash of
+    // the relayer's payout address in _relaySwap.relayerHash.
+    // Note: claimRelayer will revert if the swap value is less than the relayer
+    // fee; in that case, Bob must call claim directly.
+    function claimRelayer(
+        RelaySwap memory _relaySwap,
+        bytes32 _secret,
+        address payable _relayer,
+        uint32 _salt,
+        uint8 v,
+        bytes32 r,
+        bytes32 s
+    ) public {
+        address signer = ecrecover(keccak256(abi.encode(_relaySwap)), v, r, s);
+        if (signer != _relaySwap.swap.claimer) revert InvalidSignature();
+        if (address(this) != _relaySwap.swapCreator) revert InvalidContractAddress();
+        if (keccak256(abi.encodePacked(_relayer, _salt)) != _relaySwap.relayerHash)
+            revert InvalidRelayerAddress();
+
+        _claim(_relaySwap.swap, _secret);
+
+        // send ether to swap claimer, subtracting the relayer fee
+        if (_relaySwap.swap.asset == address(0)) {
+            _relaySwap.swap.claimer.transfer(_relaySwap.swap.value - _relaySwap.fee);
+            payable(_relayer).transfer(_relaySwap.fee);
+        } else {
+            // WARN: this will FAIL for fee-on-transfer or rebasing tokens if the token
+            // transfer reverts (i.e. if this contract does not contain _swap.value tokens),
+            // exposing Bob's secret while giving him nothing.
+            IERC20(_relaySwap.swap.asset).safeTransfer(
+                _relaySwap.swap.claimer,
+                _relaySwap.swap.value - _relaySwap.fee
+            );
+            IERC20(_relaySwap.swap.asset).safeTransfer(_relayer, _relaySwap.fee);
+        }
+    }
+
+    function _claim(Swap memory _swap, bytes32 _secret) internal {
+        bytes32 swapID = keccak256(abi.encode(_swap));
+        Stage swapStage = swaps[swapID];
+        if (swapStage == Stage.INVALID) revert InvalidSwap();
+        if (swapStage == Stage.COMPLETED) revert SwapCompleted();
+        if (block.timestamp < _swap.timeout1 && swapStage != Stage.READY) revert TooEarlyToClaim();
+        if (block.timestamp >= _swap.timeout2) revert TooLateToClaim();
+
+        verifySecret(_secret, _swap.claimCommitment);
+        emit Claimed(swapID, _secret);
+        swaps[swapID] = Stage.COMPLETED;
+    }
+
+    // Alice can `refund` her swap funds:
+    // - Until timeout1, unless she called `setReady`
+    // - After timeout2, independent of whether she called `setReady`
+    function refund(Swap memory _swap, bytes32 _secret) public {
+        bytes32 swapID = keccak256(abi.encode(_swap));
+        Stage swapStage = swaps[swapID];
+        if (swapStage == Stage.INVALID) revert InvalidSwap();
+        if (swapStage == Stage.COMPLETED) revert SwapCompleted();
+        if (_swap.owner != msg.sender) revert OnlySwapOwner();
+        if (
+            block.timestamp < _swap.timeout2 &&
+            (block.timestamp > _swap.timeout1 || swapStage == Stage.READY)
+        ) revert NotTimeToRefund();
+
+        verifySecret(_secret, _swap.refundCommitment);
+        emit Refunded(swapID, _secret);
+
+        // send asset back to swap owner
+        swaps[swapID] = Stage.COMPLETED;
+        if (_swap.asset == address(0)) {
+            _swap.owner.transfer(_swap.value);
+        } else {
+            IERC20(_swap.asset).safeTransfer(_swap.owner, _swap.value);
+        }
+    }
+
+    function verifySecret(bytes32 _secret, bytes32 _hashedPubkey) internal pure {
+        if (!mulVerify(uint256(_secret), uint256(_hashedPubkey))) revert InvalidSecret();
+    }
+}

src/error.rs

@@ -0,0 +1,21 @@
+use crate::{ethereum, protocol};
+
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error("protocol error: {0}")]
+    ProtocolError(#[source] protocol::Error),
+    #[error("ethereum error: {0}")]
+    EthereumError(#[source] ethereum::Error),
+}
+
+impl From<protocol::Error> for Error {
+    fn from(e: protocol::Error) -> Self {
+        Error::ProtocolError(e)
+    }
+}
+
+impl From<ethereum::Error> for Error {
+    fn from(e: ethereum::Error) -> Self {
+        Error::EthereumError(e)
+    }
+}

src/ethereum/error.rs

@@ -0,0 +1,37 @@
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error("counterparty keys channel closed")]
+    CounterpartyKeysChannelClosed,
+    #[error("listening to Claimed event stream failed")]
+    ClaimedEventStreamFailed,
+    #[error("timeout_1 is in the past")]
+    Timeout1Passed,
+    #[error("timeout_1 is too close to now")]
+    Timeout1TooClose,
+    #[error("timeout_2 is in the past")]
+    Timeout2Passed,
+    #[error("ERC20 not supported yet")]
+    ERC20NotSupported,
+    #[error("failed to submit `{0}` transaction: {1}")]
+    FailedToSubmitTransaction(String, String),
+    #[error("failed to await pending `{0}` transaction")]
+    FailedToAwaitPendingTransaction(String, #[source] ethers::providers::ProviderError),
+    #[error("no receipt received for transaction")]
+    NoReceipt,
+    #[error("`{0}` transaction failed: {1:?}")]
+    TransactionFailed(String, ethers::types::TransactionReceipt),
+    #[error("failed to decode log")]
+    NewSwapLogDecodingFailed(#[source] ethers::abi::Error),
+    #[error("expected exactly one log, got {0}")]
+    NewSwapUnexpectedLogCount(usize),
+    #[error("expected exactly one topic, got {0}")]
+    NewSwapUnexpectedTopicCount(usize),
+    #[error("expected five tokens, got {0}")]
+    NewSwapUnexpectedLogTokenCount(usize),
+    #[error("expected exactly 32 bytes, got {0}")]
+    FixedBytesDecodingError(usize),
+    #[error("expected FixedBytes, got another token type: {0}")]
+    ExpectedFixedBytes(ethers::abi::Token),
+    #[error("expected two U256s, got something else")]
+    ExpectedTwoU256s,
+}

src/ethereum/initiator.rs

@@ -0,0 +1,223 @@
+use crate::{
+    ethereum::{
+        swap_creator::{Swap, SwapCreator},
+        Error,
+    },
+    protocol::traits::{HandleCounterpartyKeysReceivedResult, InitiateSwapArgs, Initiator},
+};
+
+use darkfi_serial::async_trait;
+use ethers::prelude::*;
+
+use log::info;
+use std::sync::Arc;
+
+/// Implemented on top of the non-initiating chain
+///
+/// Can probably become an extension trait of an RPC client eventually
+pub(crate) trait OtherChainClient {
+    fn claim_funds(
+        &self,
+        our_secret: [u8; 32],
+        counterparty_secret: [u8; 32],
+    ) -> Result<(), crate::Error>;
+}
+
+pub(crate) struct EthInitiator<M: Middleware, C: OtherChainClient> {
+    contract: SwapCreator<M>,
+    middleware: Arc<M>,
+    other_chain_client: C,
+    secret: [u8; 32],
+}
+
+#[allow(dead_code)]
+impl<M: Middleware, C: OtherChainClient> EthInitiator<M, C> {
+    pub(crate) fn new(
+        contract: SwapCreator<M>,
+        middleware: Arc<M>,
+        other_chain_client: C,
+        secret: [u8; 32],
+    ) -> Self {
+        Self { contract, middleware, other_chain_client, secret }
+    }
+}
+
+#[async_trait]
+impl<M: Middleware + 'static, C: OtherChainClient + Send + Sync> Initiator for EthInitiator<M, C> {
+    async fn handle_counterparty_keys_received(
+        &self,
+        args: InitiateSwapArgs,
+    ) -> Result<HandleCounterpartyKeysReceivedResult, crate::Error> {
+        use ethers::abi::ParamType;
+
+        let InitiateSwapArgs {
+            claim_commitment,
+            refund_commitment,
+            claimer,
+            timeout_duration_1,
+            timeout_duration_2,
+            asset,
+            value,
+            nonce,
+        } = args;
+
+        // TODO: ERC20 is *not* handled right now
+        if asset != Address::zero() {
+            return Err(Error::ERC20NotSupported.into());
+        }
+
+        let tx = self
+            .contract
+            .new_swap(
+                claim_commitment,
+                refund_commitment,
+                claimer,
+                timeout_duration_1,
+                timeout_duration_2,
+                asset,
+                value,
+                nonce,
+            )
+            .value(value);
+        let receipt = tx
+            .send()
+            .await
+            .map_err(|e| Error::FailedToSubmitTransaction("new_swap".to_string(), e.to_string()))?
+            .await
+            .map_err(|e| Error::FailedToAwaitPendingTransaction("new_swap".to_string(), e))?
+            .ok_or_else(|| Error::NoReceipt)?;
+
+        if receipt.status != Some(U64::from(1)) {
+            return Err(Error::TransactionFailed("new_swap".to_string(), receipt).into());
+        }
+
+        if receipt.logs.len() != 1 {
+            return Err(Error::NewSwapUnexpectedLogCount(receipt.logs.len()).into());
+        }
+
+        if receipt.logs[0].topics.len() != 1 {
+            return Err(Error::NewSwapUnexpectedTopicCount(receipt.logs[0].topics.len()).into());
+        }
+
+        let log_data = &receipt.logs[0].data;
+
+        // ABI-unpack log data
+        // note: there are other parameters emitted in the log, but we don't care about them
+        let mut tokens = ethers::abi::decode(
+            &vec![
+                ParamType::FixedBytes(32),
+                ParamType::FixedBytes(32),
+                ParamType::FixedBytes(32),
+                ParamType::Uint(256),
+                ParamType::Uint(256),
+            ],
+            &log_data.0,
+        )
+        .map_err(|e| Error::NewSwapLogDecodingFailed(e))?;
+
+        if tokens.len() != 5 {
+            return Err(Error::NewSwapUnexpectedLogTokenCount(tokens.len()).into());
+        }
+
+        let swap_id = match tokens.remove(0) {
+            ethers::abi::Token::FixedBytes(bytes) => {
+                // this shouldn't happen, would be an error in ethers-rs
+                if bytes.len() != 32 {
+                    return Err(Error::FixedBytesDecodingError(bytes.len()).into());
+                }
+
+                let mut swap_id = [0u8; 32];
+                swap_id.copy_from_slice(&bytes);
+                swap_id
+            }
+            token => {
+                return Err(Error::ExpectedFixedBytes(token).into());
+            }
+        };
+        let (timeout_1, timeout_2) = match (tokens.remove(2), tokens.remove(2)) {
+            // tokens index 3 and 4
+            (ethers::abi::Token::Uint(timeout_1), ethers::abi::Token::Uint(timeout_2)) => {
+                (timeout_1, timeout_2)
+            }
+            _ => {
+                return Err(Error::ExpectedTwoU256s.into());
+            }
+        };
+
+        let contract_swap = Swap {
+            owner: self.middleware.default_sender().expect("must have a default sender"),
+            claim_commitment,
+            refund_commitment,
+            claimer,
+            timeout_1,
+            timeout_2,
+            asset,
+            value,
+            nonce,
+        };
+
+        info!(
+            "initiated swap on-chain: contract_swap_id = {}",
+            ethers::utils::hex::encode(&swap_id)
+        );
+
+        Ok(HandleCounterpartyKeysReceivedResult {
+            contract_swap_id: swap_id,
+            contract_swap,
+            block_number: receipt
+                .block_number
+                .expect("block number must be set in receipt")
+                .as_u64(),
+        })
+    }
+
+    async fn handle_counterparty_funds_locked(
+        &self,
+        swap: super::swap_creator::Swap,
+        swap_id: [u8; 32],
+    ) -> Result<(), crate::Error> {
+        let tx = self.contract.set_ready(swap);
+        let receipt = tx
+            .send()
+            .await
+            .map_err(|e| Error::FailedToSubmitTransaction("set_ready".to_string(), e.to_string()))?
+            .await
+            .map_err(|e| Error::FailedToAwaitPendingTransaction("set_ready".to_string(), e))?
+            .ok_or_else(|| Error::NoReceipt)?;
+
+        if receipt.status != Some(U64::from(1)) {
+            return Err(Error::TransactionFailed("set_ready".to_string(), receipt).into());
+        }
+
+        info!("contract set to ready, contract_swap_id = {}", ethers::utils::hex::encode(swap_id));
+        Ok(())
+    }
+
+    async fn handle_counterparty_funds_claimed(
+        &self,
+        counterparty_secret: [u8; 32],
+    ) -> Result<(), crate::Error> {
+        self.other_chain_client.claim_funds(self.secret, counterparty_secret)
+    }
+
+    async fn handle_should_refund(
+        &self,
+        swap: super::swap_creator::Swap,
+    ) -> Result<(), crate::Error> {
+        let tx = self.contract.refund(swap, self.secret);
+
+        let receipt = tx
+            .send()
+            .await
+            .map_err(|e| Error::FailedToSubmitTransaction("refund".to_string(), e.to_string()))?
+            .await
+            .map_err(|e| Error::FailedToAwaitPendingTransaction("refund".to_string(), e))?
+            .ok_or_else(|| Error::NoReceipt)?;
+
+        if receipt.status != Some(U64::from(1)) {
+            return Err(Error::TransactionFailed("refund".to_string(), receipt).into());
+        }
+
+        Ok(())
+    }
+}

src/ethereum/initiator_event_watcher.rs

@@ -0,0 +1,92 @@
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use crate::{
+    ethereum::{swap_creator::SwapCreator, Error},
+    protocol::{
+        initiator::Event,
+        traits::{CounterpartyKeys, InitiatorEventWatcher},
+    },
+};
+use ethers::prelude::Middleware;
+use smol::{channel, stream::StreamExt as _};
+
+pub(crate) struct Watcher;
+
+#[darkfi_serial::async_trait]
+impl InitiatorEventWatcher for Watcher {
+    async fn run_received_counterparty_keys_watcher(
+        event_tx: channel::Sender<Event>,
+        counterparty_keys_rx: channel::Receiver<CounterpartyKeys>,
+    ) -> Result<(), crate::Error> {
+        let counterparty_keys = counterparty_keys_rx
+            .recv()
+            .await
+            .map_err(|_| crate::Error::from(Error::CounterpartyKeysChannelClosed))?;
+        event_tx.send(Event::ReceivedCounterpartyKeys(counterparty_keys)).await.unwrap();
+        Ok(())
+    }
+
+    async fn run_counterparty_funds_locked_watcher(
+        event_tx: channel::Sender<Event>,
+    ) -> Result<(), crate::Error> {
+        // TODO: from watching counterchain swap wallet
+        event_tx.send(Event::CounterpartyFundsLocked).await.unwrap();
+        Ok(())
+    }
+
+    async fn run_counterparty_funds_claimed_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
+            .claimed_filter() // claimed event sig is topic0
+            .from_block(from_block)
+            .address(contract.address().into())
+            .topic1(topic1);
+
+        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::ClaimedEventStreamFailed.into());
+        };
+
+        event_tx.send(Event::CounterpartyFundsClaimed(event.s)).await.unwrap();
+        Ok(())
+    }
+
+    async fn run_timeout_1_watcher(
+        event_tx: channel::Sender<Event>,
+        timeout_1: u64,
+        buffer_seconds: u64,
+    ) -> Result<(), crate::Error> {
+        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
+        let diff = timeout_1
+            .checked_sub(now)
+            .ok_or(Error::Timeout1Passed)?
+            .checked_sub(buffer_seconds)
+            .ok_or(Error::Timeout1TooClose)?;
+        let sleep_duration = std::time::Duration::from_secs(diff);
+
+        smol::Timer::after(sleep_duration).await;
+        event_tx.send(Event::AlmostTimeout1).await.unwrap();
+        Ok(())
+    }
+
+    async fn run_timeout_2_watcher(
+        event_tx: channel::Sender<Event>,
+        timeout_2: u64,
+    ) -> Result<(), crate::Error> {
+        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
+        let diff = timeout_2.checked_sub(now).ok_or(Error::Timeout2Passed)?;
+        let sleep_duration = std::time::Duration::from_secs(diff);
+
+        smol::Timer::after(sleep_duration).await;
+        event_tx.send(Event::PastTimeout2).await.unwrap();
+        Ok(())
+    }
+}

src/ethereum/mod.rs

@@ -0,0 +1,13 @@
+mod error;
+pub(crate) mod initiator;
+mod initiator_event_watcher;
+pub(crate) mod swap_creator;
+
+pub use error::Error;
+#[allow(unused_imports)]
+pub(crate) use initiator::EthInitiator;
+#[allow(unused_imports)]
+pub(crate) use initiator_event_watcher::Watcher;
+
+#[cfg(feature = "test-utils")]
+pub mod test_utils;

src/ethereum/swap_creator.rs

@@ -0,0 +1,3 @@
+use ethers::prelude::*;
+
+abigen!(SwapCreator, "./ethereum/out/SwapCreator.sol/SwapCreator.json",);

src/ethereum/test_utils.rs

@@ -0,0 +1,45 @@
+use std::{path::Path, sync::Arc, time::Duration};
+
+use ethers::{core::utils::Anvil, prelude::*, utils::AnvilInstance};
+
+/// Starts a local anvil instance and deploys the `SwapCreator` contract to it.
+///
+/// Returns the contract address, provider, wallet, and anvil instance.
+///
+/// # Panics
+///
+/// - if the contract cannot be found in the expected path
+/// - if the contract cannot be compiled
+/// - if the provider fails to connect to the anvil instance
+/// - if the contract fails to deploy
+#[allow(dead_code)]
+pub(crate) async fn deploy_swap_creator() -> (Address, Arc<Provider<Ws>>, LocalWallet, AnvilInstance)
+{
+    // compile contract for testing
+    let source = Path::new(&env!("CARGO_MANIFEST_DIR")).join("ethereum/src/SwapCreator.sol");
+    let input = CompilerInput::new(source.clone()).unwrap().first().unwrap().clone();
+    let compiled = Solc::default().compile(&input).expect("could not compile contract");
+    assert!(compiled.errors.is_empty(), "errors: {:?}", compiled.errors);
+
+    let (abi, bytecode, _) =
+        compiled.find("SwapCreator").expect("could not find contract").into_parts_or_default();
+
+    // setup anvil and signing wallet
+    let anvil = Anvil::new().spawn();
+    let wallet: LocalWallet = anvil.keys()[0].clone().into();
+    let provider = Arc::new(
+        Provider::<Ws>::connect(anvil.ws_endpoint())
+            .await
+            .unwrap()
+            .interval(Duration::from_millis(10u64)),
+    );
+    let signer =
+        SignerMiddleware::new(provider.clone(), wallet.clone().with_chain_id(anvil.chain_id()));
+
+    // deploy contract
+    let factory = ContractFactory::new(abi, bytecode, signer.into());
+    let contract = factory.deploy(()).unwrap().send().await.unwrap();
+    let contract_address = contract.address();
+
+    (contract_address, provider, wallet.with_chain_id(anvil.chain_id()), anvil)
+}

src/lib.rs

@@ -0,0 +1,8 @@
+pub(crate) mod error;
+mod ethereum;
+pub(crate) mod protocol;
+mod rpc;
+pub(crate) mod swapd;
+
+pub(crate) use error::Error;
+pub use swapd::{Swapd, SwapdArgs};

src/main.rs

@@ -16,28 +16,26 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{collections::HashSet, sync::Arc};
+use std::sync::Arc;
 
 use darkfi::{
     async_daemonize, cli_desc,
     rpc::server::{listen_and_serve, RequestHandler},
-    system::{StoppableTask, StoppableTaskPtr},
+    system::StoppableTask,
     util::path::expand_path,
     Error, Result,
 };
 use log::{error, info};
 use serde::Deserialize;
-use smol::{fs, lock::Mutex, stream::StreamExt, Executor};
+use smol::{fs, stream::StreamExt, Executor};
 use structopt::StructOpt;
 use structopt_toml::StructOptToml;
-use url::Url;
+
+use swapd::{Swapd, SwapdArgs};
 
 const CONFIG_FILE: &str = "swapd.toml";
 const CONFIG_FILE_CONTENTS: &str = include_str!("../swapd.toml");
 
-/// JSON-RPC server methods
-mod rpc;
-
 #[derive(Clone, Debug, Deserialize, StructOpt, StructOptToml)]
 #[serde(default)]
 #[structopt(name = "darkfi-mmproxy", about = cli_desc!())]
@@ -58,33 +56,6 @@ struct Args {
     swapd: SwapdArgs,
 }
 
-#[derive(Clone, Debug, Deserialize, StructOpt, StructOptToml)]
-#[structopt()]
-struct SwapdArgs {
-    #[structopt(long, default_value = "tcp://127.0.0.1:52821")]
-    /// darkfi-swapd JSON-RPC listen URL
-    swapd_rpc: Url,
-
-    #[structopt(long, default_value = "~/.local/darkfi/swapd")]
-    /// Path to swapd's filesystem database
-    swapd_db: String,
-}
-
-/// Swapd daemon state
-struct Swapd {
-    /// Main reference to the swapd filesystem databaase
-    _sled_db: sled::Db,
-    /// JSON-RPC connection tracker
-    rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
-}
-
-impl Swapd {
-    /// Instantiate `Swapd` state
-    async fn new(_swapd_args: &SwapdArgs, sled_db: sled::Db) -> Result<Self> {
-        Ok(Self { _sled_db: sled_db, rpc_connections: Mutex::new(HashSet::new()) })
-    }
-}
-
 async_daemonize!(realmain);
 async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     info!("Starting DarkFi Atomic Swap Daemon...");

src/protocol/error.rs

@@ -0,0 +1,15 @@
+use crate::protocol::traits::CounterpartyKeys;
+
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error("unexpected received counterparty keys event: {0}")]
+    UnexpectedReceivedCounterpartyKeysEvent(CounterpartyKeys),
+    #[error("unexpected counterparty funds locked event")]
+    UnexpectedCounterpartyFundsLockedEvent,
+    #[error("unexpected counterparty funds claimed event")]
+    UnexpectedCounterpartyFundsClaimedEvent([u8; 32]),
+    #[error("unexpected almost timeout 1 event")]
+    UnexpectedAlmostTimeout1Event,
+    #[error("unexpected past timeout 2 event")]
+    UnexpectedPastTimeout2Event,
+}

src/protocol/follower.rs

@@ -0,0 +1,41 @@
+use crate::protocol::traits::Follower;
+use smol::channel;
+
+#[allow(dead_code)]
+enum Event {
+    CounterpartyFundsLocked,
+    ReadyToClaim,
+    CounterpartyFundsRefunded,
+}
+
+#[allow(dead_code)]
+struct Swap {
+    handler: Box<dyn Follower>,
+    event_rx: channel::Receiver<Event>,
+}
+
+#[allow(dead_code)]
+impl Swap {
+    fn new(handler: Box<dyn Follower>, event_rx: channel::Receiver<Event>) -> Self {
+        Self { handler, event_rx }
+    }
+
+    async fn run(&mut self) {
+        loop {
+            match self.event_rx.recv().await {
+                Ok(Event::CounterpartyFundsLocked) => {
+                    self.handler.handle_counterparty_funds_locked();
+                }
+                Ok(Event::ReadyToClaim) => {
+                    self.handler.handle_ready_to_claim();
+                }
+                Ok(Event::CounterpartyFundsRefunded) => {
+                    self.handler.handle_counterparty_funds_refunded();
+                }
+                Err(_) => {
+                    break;
+                }
+            }
+        }
+    }
+}

src/protocol/initiator.rs

@@ -0,0 +1,342 @@
+use crate::protocol::{
+    traits::{
+        CounterpartyKeys, HandleCounterpartyKeysReceivedResult, InitiateSwapArgs, InitiationArgs,
+        Initiator,
+    },
+    Error,
+};
+use log::{info, warn};
+use smol::channel;
+
+#[derive(Debug)]
+pub(crate) enum Event {
+    ReceivedCounterpartyKeys(CounterpartyKeys),
+    CounterpartyFundsLocked,
+    CounterpartyFundsClaimed([u8; 32]),
+    AlmostTimeout1,
+    PastTimeout2,
+}
+
+#[allow(dead_code)]
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum State {
+    WaitingForCounterpartyKeys,
+    WaitingForCounterpartyFundsLocked,
+    WaitingForCounterpartyFundsClaimed,
+    Completed,
+}
+
+#[allow(dead_code)]
+struct Swap {
+    // the initial parameters required for the swap
+    args: InitiationArgs,
+
+    // the chain-specific event handler
+    // TODO: just make this a generic
+    handler: Box<dyn Initiator + Send + Sync>,
+
+    // the event receiver channel for the swap
+    // the [`Watcher`] sends events to this channel
+    event_rx: channel::Receiver<Event>,
+
+    // the current state of the swap
+    state_tx: async_watch::Sender<State>,
+    state_rx: async_watch::Receiver<State>,
+
+    // the info of the swap within the on-chain contract
+    contract_swap_info_tx: async_watch::Sender<Option<HandleCounterpartyKeysReceivedResult>>,
+    contract_swap_info_rx: async_watch::Receiver<Option<HandleCounterpartyKeysReceivedResult>>,
+}
+
+#[allow(dead_code)]
+impl Swap {
+    fn new(
+        args: InitiationArgs,
+        handler: Box<dyn Initiator + Send + Sync>,
+        event_rx: channel::Receiver<Event>,
+    ) -> (
+        Self,
+        async_watch::Receiver<State>,
+        async_watch::Receiver<Option<HandleCounterpartyKeysReceivedResult>>,
+    ) {
+        let state = async_watch::channel(State::WaitingForCounterpartyKeys);
+        let contract_swap_info = async_watch::channel(None);
+        (
+            Self {
+                args,
+                handler,
+                event_rx,
+                state_tx: state.0,
+                state_rx: state.1.clone(),
+                contract_swap_info_tx: contract_swap_info.0,
+                contract_swap_info_rx: contract_swap_info.1.clone(),
+            },
+            state.1,
+            contract_swap_info.1,
+        )
+    }
+
+    async fn run(self) -> Result<(), crate::Error> {
+        loop {
+            match self.event_rx.recv().await {
+                Ok(Event::ReceivedCounterpartyKeys(counterparty_keys)) => {
+                    info!("received counterparty keys");
+
+                    if !matches!(*self.state_rx.borrow(), State::WaitingForCounterpartyKeys) {
+                        warn!(
+                            "unexpected event ReceivedCounterpartyKeys, state is {:?}",
+                            *self.state_rx.borrow()
+                        );
+                        return Err(Error::UnexpectedReceivedCounterpartyKeysEvent(
+                            counterparty_keys,
+                        )
+                        .into());
+                    }
+
+                    let refund_commitment =
+                        ethers::utils::keccak256(&counterparty_keys.secp256k1_public_key);
+
+                    let args = InitiateSwapArgs {
+                        claim_commitment: self.args.claim_commitment,
+                        refund_commitment,
+                        claimer: self.args.claimer,
+                        timeout_duration_1: self.args.timeout_duration_1,
+                        timeout_duration_2: self.args.timeout_duration_2,
+                        asset: self.args.asset,
+                        value: self.args.value,
+                        nonce: self.args.nonce,
+                    };
+
+                    let contract_swap_info =
+                        Some(self.handler.handle_counterparty_keys_received(args).await?);
+
+                    let _ = self.contract_swap_info_tx.send(contract_swap_info.clone());
+                    self.state_tx
+                        .send(State::WaitingForCounterpartyFundsLocked)
+                        .expect("state channel should not be dropped");
+                }
+                Ok(Event::CounterpartyFundsLocked) => {
+                    info!("counterparty funds locked");
+                    if !matches!(*self.state_rx.borrow(), State::WaitingForCounterpartyFundsLocked)
+                    {
+                        return Err(Error::UnexpectedCounterpartyFundsLockedEvent.into());
+                    }
+
+                    let contract_swap_info = self
+                        .contract_swap_info_rx
+                        .borrow()
+                        .clone()
+                        .expect("contract swap info must be set");
+
+                    self.handler
+                        .handle_counterparty_funds_locked(
+                            contract_swap_info.contract_swap,
+                            contract_swap_info.contract_swap_id,
+                        )
+                        .await?;
+
+                    self.state_tx
+                        .send(State::WaitingForCounterpartyFundsClaimed)
+                        .expect("state channel should not be dropped");
+                }
+                Ok(Event::CounterpartyFundsClaimed(counterparty_secret)) => {
+                    if !matches!(*self.state_rx.borrow(), State::WaitingForCounterpartyFundsClaimed)
+                    {
+                        return Err(Error::UnexpectedCounterpartyFundsClaimedEvent(
+                            counterparty_secret,
+                        )
+                        .into());
+                    }
+
+                    self.handler.handle_counterparty_funds_claimed(counterparty_secret).await?;
+                    self.state_tx
+                        .send(State::Completed)
+                        .expect("state channel should not be dropped");
+                }
+                Ok(Event::AlmostTimeout1) => {
+                    match *self.state_rx.borrow() {
+                        State::WaitingForCounterpartyFundsLocked |
+                        State::WaitingForCounterpartyFundsClaimed => {}
+                        _ => {
+                            return Err(Error::UnexpectedAlmostTimeout1Event.into());
+                        }
+                    }
+
+                    // we're almost at timeout 1, and the counterparty hasn't locked,
+                    // so we need to refund
+                    if matches!(*self.state_rx.borrow(), State::WaitingForCounterpartyFundsLocked) {
+                        let contract_swap_info = self
+                            .contract_swap_info_rx
+                            .borrow()
+                            .clone()
+                            .expect("contract swap info must be set");
+
+                        self.handler.handle_should_refund(contract_swap_info.contract_swap).await?;
+
+                        self.state_tx
+                            .send(State::Completed)
+                            .expect("state channel should not be dropped");
+                    }
+                }
+                Ok(Event::PastTimeout2) => {
+                    if !matches!(*self.state_rx.borrow(), State::WaitingForCounterpartyFundsClaimed)
+                    {
+                        return Err(Error::UnexpectedPastTimeout2Event.into());
+                    }
+
+                    let contract_swap_info = self
+                        .contract_swap_info_rx
+                        .borrow()
+                        .clone()
+                        .expect("contract swap info must be set");
+
+                    // we're past timeout 2, and the counterparty hasn't claimed,
+                    // so we need to refund
+                    self.handler
+                        .handle_should_refund(contract_swap_info.contract_swap.clone())
+                        .await?;
+
+                    self.state_tx
+                        .send(State::Completed)
+                        .expect("state channel should not be dropped");
+                }
+                Err(_) => {
+                    info!("event channel closed, exiting");
+                    break;
+                }
+            }
+
+            if matches!(*self.state_rx.borrow(), State::Completed) {
+                info!("swap completed, exiting");
+                break;
+            }
+        }
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    use std::sync::Arc;
+
+    use smol::channel::bounded;
+
+    use crate::ethereum::{
+        initiator::OtherChainClient, swap_creator::SwapCreator, EthInitiator, Watcher,
+    };
+
+    use ethers::{
+        core::k256::elliptic_curve::sec1::ToEncodedPoint,
+        prelude::{Address, SignerMiddleware, U256},
+    };
+
+    use crate::protocol::traits::InitiatorEventWatcher as _;
+
+    struct MockOtherChainClient;
+
+    impl OtherChainClient for MockOtherChainClient {
+        fn claim_funds(
+            &self,
+            _our_secret: [u8; 32],
+            _counterparty_secret: [u8; 32],
+        ) -> Result<(), crate::Error> {
+            Ok(())
+        }
+    }
+
+    #[async_std::test]
+    async fn test_initiator_swap_success() {
+        let (event_tx, event_rx) = channel::bounded(1);
+
+        let (contract_address, provider, wallet, anvil) =
+            crate::ethereum::test_utils::deploy_swap_creator().await;
+        let signer = Arc::new(SignerMiddleware::new(provider, wallet));
+        let contract = SwapCreator::new(contract_address, signer.clone());
+
+        let other_chain_client = MockOtherChainClient;
+        let refund_secret = [0; 32]; // TODO generate an actual secp256k1 private key for refund testing
+        let initiator =
+            EthInitiator::new(contract.clone(), signer.clone(), other_chain_client, refund_secret);
+
+        // TODO: this is the same key as the initiator right now.
+        let counterparty_secret: [u8; 32] = anvil.keys()[0].to_bytes().try_into().unwrap();
+        let counterparty_public_key = anvil.keys()[0].public_key();
+        let pubkey_bytes: [u8; 64] =
+            counterparty_public_key.to_encoded_point(false).as_bytes()[1..].try_into().unwrap();
+        let claim_commitment = ethers::utils::keccak256(pubkey_bytes);
+
+        let args = InitiationArgs {
+            claim_commitment,
+            claimer: signer.address(),
+            timeout_duration_1: U256::from(120),
+            timeout_duration_2: U256::from(120),
+            asset: Address::zero(),                      // ETH
+            value: 1_000_000_000_000_000_000u128.into(), // 1 ETH
+            nonce: U256::zero(),                         // arbitrary
+        };
+
+        let (swap, mut state, contract_swap_id) =
+            Swap::new(args.clone(), Box::new(initiator), event_rx);
+        assert!(*state.borrow() == State::WaitingForCounterpartyKeys);
+
+        let swap_task = smol::spawn(async move { swap.run().await });
+
+        let (counterparty_keys_tx, counterparty_keys_rx) = bounded(1);
+        let join_handle = smol::spawn(Watcher::run_received_counterparty_keys_watcher(
+            event_tx.clone(),
+            counterparty_keys_rx,
+        ));
+
+        counterparty_keys_tx
+            .send(CounterpartyKeys { secp256k1_public_key: [0; 33] })
+            .await
+            .expect("should send counterparty keys");
+        state.changed().await.expect("state should change");
+        assert!(*state.borrow() == State::WaitingForCounterpartyFundsLocked);
+        join_handle.cancel().await.unwrap().unwrap();
+
+        Watcher::run_counterparty_funds_locked_watcher(event_tx.clone())
+            .await
+            .expect("watcher should run");
+        state.changed().await.expect("state should change");
+        assert!(*state.borrow() == State::WaitingForCounterpartyFundsClaimed);
+
+        let contract_swap = contract_swap_id.borrow().as_ref().unwrap().contract_swap.clone();
+
+        let contract_clone = contract.clone();
+        let claim_task = smol::spawn(async move {
+            let tx = contract_clone.claim(contract_swap, counterparty_secret);
+
+            let receipt = tx
+                .send()
+                .await
+                .expect("failed to submit transaction")
+                .await
+                .expect("failed to await pending transaction")
+                .expect("no receipt found");
+
+            assert!(
+                receipt.status == Some(ethers::types::U64::from(1)),
+                "`claim` transaction failed: {:?}",
+                receipt
+            );
+        });
+
+        Watcher::run_counterparty_funds_claimed_watcher(
+            event_tx,
+            contract,
+            &contract_swap_id.borrow().as_ref().unwrap().contract_swap_id,
+            contract_swap_id.borrow().as_ref().unwrap().block_number,
+        )
+        .await
+        .expect("watcher should run");
+        state.changed().await.expect("state should change");
+        assert!(*state.borrow() == State::Completed);
+
+        swap_task.await.expect("swap task should not fail");
+        claim_task.await;
+    }
+}

src/protocol/mod.rs

@@ -0,0 +1,7 @@
+//! This module contains the protocol traits and logic for DRK-ETH atomic swaps.
+mod error;
+mod follower;
+pub(crate) mod initiator;
+pub(crate) mod traits;
+
+pub use error::Error;

src/protocol/traits.rs

@@ -0,0 +1,140 @@
+use crate::ethereum::swap_creator::Swap; // TODO: shouldn't depend on this
+use crate::{error::Error, ethereum::swap_creator::SwapCreator, protocol::initiator::Event};
+use darkfi_serial::async_trait;
+use ethers::{prelude::*, utils::hex};
+use smol::channel;
+use std::{
+    fmt,
+    fmt::{Display, Formatter},
+};
+
+// Initial parameters required by the swap initiator.
+// TODO: make Address/U256 generic; these are ethers-specific right now
+#[allow(dead_code)]
+#[derive(Debug, Clone)]
+pub(crate) struct InitiationArgs {
+    pub(crate) claim_commitment: [u8; 32],
+    pub(crate) claimer: Address,
+    pub(crate) timeout_duration_1: U256,
+    pub(crate) timeout_duration_2: U256,
+    pub(crate) asset: Address,
+    pub(crate) value: U256,
+    pub(crate) nonce: U256,
+}
+
+// TODO: make Address/U256 generic; these are ethers-specific right now
+#[derive(Debug)]
+pub(crate) struct InitiateSwapArgs {
+    pub(crate) claim_commitment: [u8; 32],
+    pub(crate) refund_commitment: [u8; 32],
+    pub(crate) claimer: Address,
+    pub(crate) timeout_duration_1: U256,
+    pub(crate) timeout_duration_2: U256,
+    pub(crate) asset: Address,
+    pub(crate) value: U256,
+    pub(crate) nonce: U256,
+}
+
+// TODO: make this generic for both chains
+#[allow(dead_code)]
+#[derive(Debug)]
+pub(crate) struct CounterpartyKeys {
+    pub(crate) secp256k1_public_key: [u8; 33],
+}
+
+impl Display for CounterpartyKeys {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "CounterpartyKeys {{ secp256k1_public_key: {:?} }}",
+            hex::encode(self.secp256k1_public_key)
+        )
+    }
+}
+
+#[allow(dead_code)]
+#[derive(Debug, Clone)]
+pub(crate) struct HandleCounterpartyKeysReceivedResult {
+    // the ID of the swap within the on-chain contract
+    pub(crate) contract_swap_id: [u8; 32],
+
+    // the details of the swap within the on-chain contract
+    pub(crate) contract_swap: Swap,
+
+    // the block number at which the swap was initiated
+    pub(crate) block_number: u64,
+}
+
+/// the chain that initiates the swap; ie. the first-mover
+///
+/// the implementation of this trait must hold a signing key for
+/// chain A and chain B.
+///
+/// TODO: [`Swap`] should be a non-chain-specific type
+#[async_trait]
+pub(crate) trait Initiator {
+    // initiates the swap by locking funds on chain A
+    async fn handle_counterparty_keys_received(
+        &self,
+        args: InitiateSwapArgs,
+    ) -> Result<HandleCounterpartyKeysReceivedResult, Error>;
+
+    // handles the counterparty locking funds
+    async fn handle_counterparty_funds_locked(
+        &self,
+        swap: Swap,
+        swap_id: [u8; 32],
+    ) -> Result<(), Error>;
+
+    // handles the counterparty claiming funds
+    async fn handle_counterparty_funds_claimed(
+        &self,
+        counterparty_secret: [u8; 32],
+    ) -> Result<(), Error>;
+
+    // handles the timeout cases where we need to refund funds
+    async fn handle_should_refund(&self, swap: Swap) -> Result<(), Error>;
+}
+
+#[async_trait]
+pub(crate) trait InitiatorEventWatcher {
+    async fn run_received_counterparty_keys_watcher(
+        event_tx: channel::Sender<Event>,
+        counterparty_keys_rx: channel::Receiver<CounterpartyKeys>,
+    ) -> Result<(), Error>;
+
+    async fn run_counterparty_funds_locked_watcher(
+        event_tx: channel::Sender<Event>,
+    ) -> Result<(), Error>;
+
+    // TODO: make this generic for both chains
+    async fn run_counterparty_funds_claimed_watcher<M: Middleware>(
+        event_tx: channel::Sender<Event>,
+        contract: SwapCreator<M>,
+        contract_swap_id: &[u8; 32],
+        from_block: u64,
+    ) -> Result<(), Error>;
+
+    async fn run_timeout_1_watcher(
+        event_tx: channel::Sender<Event>,
+        timeout_1: u64,
+        buffer_seconds: u64,
+    ) -> Result<(), Error>;
+
+    async fn run_timeout_2_watcher(
+        event_tx: channel::Sender<Event>,
+        timeout_2: u64,
+    ) -> Result<(), Error>;
+}
+
+/// the chain that is the counterparty to the swap; ie. the second-mover
+pub(crate) trait Follower {
+    // handle the swap initiation by locking funds on chain B
+    fn handle_counterparty_funds_locked(&self);
+
+    // handle the funds being ready to be claimed by us
+    fn handle_ready_to_claim(&self);
+
+    // handle the counterparty refunding their funds, in case of a timeout
+    fn handle_counterparty_funds_refunded(&self);
+}

src/rpc.rs

@@ -29,7 +29,7 @@ use darkfi::{
 use darkfi_serial::async_trait;
 use smol::lock::MutexGuard;
 
-use super::Swapd;
+use crate::swapd::Swapd;
 
 #[async_trait]
 impl RequestHandler for Swapd {

src/swapd.rs

@@ -0,0 +1,35 @@
+use std::collections::HashSet;
+
+use darkfi::{system::StoppableTaskPtr, Result};
+use serde::Deserialize;
+use smol::lock::Mutex;
+use structopt::StructOpt;
+use structopt_toml::StructOptToml;
+use url::Url;
+
+#[derive(Clone, Debug, Deserialize, StructOpt, StructOptToml)]
+#[structopt()]
+pub struct SwapdArgs {
+    #[structopt(long, default_value = "tcp://127.0.0.1:52821")]
+    /// darkfi-swapd JSON-RPC listen URL
+    pub swapd_rpc: Url,
+
+    #[structopt(long, default_value = "~/.local/darkfi/swapd")]
+    /// Path to swapd's filesystem database
+    pub swapd_db: String,
+}
+
+/// Swapd daemon state
+pub struct Swapd {
+    /// Main reference to the swapd filesystem databaase
+    _sled_db: sled::Db,
+    /// JSON-RPC connection tracker
+    pub(crate) rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
+}
+
+impl Swapd {
+    /// Instantiate `Swapd` state
+    pub async fn new(_swapd_args: &SwapdArgs, sled_db: sled::Db) -> Result<Self> {
+        Ok(Self { _sled_db: sled_db, rpc_connections: Mutex::new(HashSet::new()) })
+    }
+}