refactor: use fsm and actions pattern in CCH

- Implement a state machine for CCH orders to manage transitions based on incoming invoice and outgoing payment events.
- Introduce action dispatchers for handling various order actions, such as sending payments and settling invoices.
- Refactor existing code to improve modularity and maintainability, including the separation of event handling and order state management.

Author: doitian

.cursor/rules/base-branch.mdc

@@ -0,0 +1,4 @@
+---
+alwaysApply: true
+---
+The base branch is develop

.cursor/rules/nextest.mdc

@@ -0,0 +1,13 @@
+---
+alwaysApply: true
+---
+# Run tests using `cargo nextest`
+
+Always execute the Rust test suite with `cargo nextest` instead of the default `cargo test`.
+
+## How to run
+
+- `cargo nextest run --all-targets --all-features` for a full local verification run.
+- Use the `Makefile` target `make test` when you want the exact flags used in automation (`RUST_LOG=off cargo nextest run --no-fail-fast -p fnn -p fiber-bin`).
+
+Running tests this way avoids discrepancies with CI and ensures flaky tests are surfaced quickly.

.cursor/worktrees.json

@@ -0,0 +1,6 @@
+{
+  "setup-worktree": [
+    "cp $ROOT_WORKTREE_PATH/mise.toml mise.toml",
+    "mkdir -p .vscode && cp -f $ROOT_WORKTREE_PATH/.vscode/settings.json .vscode/settings.json"
+  ]
+}

crates/fiber-bin/src/main.rs

@@ -20,7 +20,7 @@ use fnn::tasks::{
 use fnn::watchtower::{
     WatchtowerActor, WatchtowerMessage, DEFAULT_WATCHTOWER_CHECK_INTERVAL_SECONDS,
 };
-use fnn::{start_cch, start_network, Config, NetworkServiceEvent};
+use fnn::{start_network, CchActor, Config, NetworkServiceEvent};
 use jsonrpsee::http_client::HttpClientBuilder;
 use jsonrpsee::ws_client::{HeaderMap, HeaderValue};
 use ractor::{port::OutputPortSubscriberTrait as _, Actor, ActorRef, OutputPort};
@@ -327,7 +327,9 @@ pub async fn main() -> Result<(), ExitMessage> {
                 })?
                 .read_or_generate_secret_key()
                 .map_err(|err| ExitMessage(format!("failed to read secret key: {}", err)))?;
-            match start_cch(
+            match Actor::spawn_linked(
+                Some("cch actor".to_string()),
+                CchActor,
                 CchArgs {
                     config: cch_config,
                     tracker: new_tokio_task_tracker(),
@@ -350,7 +352,7 @@ pub async fn main() -> Result<(), ExitMessage> {
                         ));
                     }
                 }
-                Ok(actor) => {
+                Ok((actor, _handle)) => {
                     if let Some(port) = cch_fiber_store_event_port {
                         actor.subscribe_to_port(&port);
                     }

crates/fiber-lib/src/cch/actions/backend_dispatchers.rs

@@ -0,0 +1,27 @@
+use crate::cch::{CchInvoice, CchOrder};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PaymentHandlerType {
+    Fiber,
+    Lightning,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum InvoiceHandlerType {
+    Fiber,
+    Lightning,
+}
+
+pub fn dispatch_invoice_handler(order: &CchOrder) -> InvoiceHandlerType {
+    match order.incoming_invoice {
+        CchInvoice::Fiber(_) => InvoiceHandlerType::Fiber,
+        CchInvoice::Lightning(_) => InvoiceHandlerType::Lightning,
+    }
+}
+pub fn dispatch_payment_handler(order: &CchOrder) -> PaymentHandlerType {
+    // Payment use the inverse handler of the invoice.
+    match order.incoming_invoice {
+        CchInvoice::Fiber(_) => PaymentHandlerType::Lightning,
+        CchInvoice::Lightning(_) => PaymentHandlerType::Fiber,
+    }
+}

crates/fiber-lib/src/cch/actions/mod.rs

@@ -0,0 +1,60 @@
+mod backend_dispatchers;
+mod send_outgoing_payment;
+mod settle_incoming_invoice;
+mod track_incoming_invoice;
+mod track_outgoing_payment;
+use send_outgoing_payment::SendOutgoingPaymentDispatcher;
+use settle_incoming_invoice::SettleIncomingInvoiceDispatcher;
+use track_incoming_invoice::TrackIncomingInvoiceDispatcher;
+use track_outgoing_payment::TrackOutgoingPaymentDispatcher;
+
+use anyhow::Result;
+use ractor::ActorRef;
+
+use crate::cch::{actor::CchState, order::CchOrderAction, CchMessage, CchOrder};
+
+#[async_trait::async_trait]
+pub trait ActionExecutor: Send + Sync {
+    async fn execute(self: Box<Self>) -> Result<()>;
+}
+
+pub struct ActionDispatcher;
+
+impl ActionDispatcher {
+    fn dispatch(
+        state: &mut CchState,
+        cch_actor_ref: &ActorRef<CchMessage>,
+        order: &CchOrder,
+        action: CchOrderAction,
+    ) -> Option<Box<dyn ActionExecutor>> {
+        match action {
+            CchOrderAction::TrackIncomingInvoice => {
+                TrackIncomingInvoiceDispatcher::dispatch(state, cch_actor_ref, order)
+            }
+            CchOrderAction::SendOutgoingPayment => {
+                SendOutgoingPaymentDispatcher::dispatch(state, cch_actor_ref, order)
+            }
+            CchOrderAction::TrackOutgoingPayment => {
+                TrackOutgoingPaymentDispatcher::dispatch(state, cch_actor_ref, order)
+            }
+            CchOrderAction::SettleIncomingInvoice => {
+                SettleIncomingInvoiceDispatcher::dispatch(state, cch_actor_ref, order)
+            }
+        }
+    }
+
+    /// Execute an action.
+    ///
+    /// Executor cannot modify the order directly, but can send events to the actor.
+    pub async fn execute(
+        state: &mut CchState,
+        cch_actor_ref: &ActorRef<CchMessage>,
+        order: &CchOrder,
+        action: CchOrderAction,
+    ) -> Result<()> {
+        if let Some(executor) = Self::dispatch(state, cch_actor_ref, order, action) {
+            return executor.execute().await;
+        }
+        Ok(())
+    }
+}

crates/fiber-lib/src/cch/actions/send_outgoing_payment.rs

@@ -0,0 +1,120 @@
+use anyhow::{anyhow, Result};
+use futures::StreamExt as _;
+use lnd_grpc_tonic_client::routerrpc;
+use ractor::{call, ActorRef};
+
+use crate::{
+    cch::{
+        actions::{
+            backend_dispatchers::{dispatch_payment_handler, PaymentHandlerType},
+            ActionExecutor,
+        },
+        actor::CchState,
+        trackers::{map_lnd_payment_changed_event, CchTrackingEvent, LndConnectionInfo},
+        CchMessage, CchOrder, CchOrderStatus,
+    },
+    fiber::{
+        network::SendPaymentCommand, NetworkActorCommand, NetworkActorMessage,
+        ASSUME_NETWORK_ACTOR_ALIVE,
+    },
+};
+
+const BTC_PAYMENT_TIMEOUT_SECONDS: i32 = 60;
+
+pub struct SendOutgoingPaymentDispatcher;
+
+pub struct SendFiberOutgoingPaymentExecutor {
+    cch_actor_ref: ActorRef<CchMessage>,
+    network_actor_ref: ActorRef<NetworkActorMessage>,
+    outgoing_pay_req: String,
+}
+
+#[async_trait::async_trait]
+impl ActionExecutor for SendFiberOutgoingPaymentExecutor {
+    async fn execute(self: Box<Self>) -> Result<()> {
+        let outgoing_pay_req = self.outgoing_pay_req;
+        let message = |rpc_reply| -> NetworkActorMessage {
+            NetworkActorMessage::Command(NetworkActorCommand::SendPayment(
+                SendPaymentCommand {
+                    invoice: Some(outgoing_pay_req),
+                    ..Default::default()
+                },
+                rpc_reply,
+            ))
+        };
+
+        let payment = call!(self.network_actor_ref, message)
+            .expect(ASSUME_NETWORK_ACTOR_ALIVE)
+            .map_err(|err| anyhow!("{}", err))?;
+
+        self.cch_actor_ref.send_message(CchMessage::TrackingEvent(
+            CchTrackingEvent::PaymentChanged {
+                payment_hash: payment.payment_hash,
+                payment_preimage: None,
+                status: payment.status,
+            },
+        ))?;
+
+        Ok(())
+    }
+}
+
+pub struct SendLightningOutgoingPaymentExecutor {
+    cch_actor_ref: ActorRef<CchMessage>,
+    outgoing_pay_req: String,
+    lnd_connection: LndConnectionInfo,
+}
+
+#[async_trait::async_trait]
+impl ActionExecutor for SendLightningOutgoingPaymentExecutor {
+    async fn execute(self: Box<Self>) -> Result<()> {
+        let req = routerrpc::SendPaymentRequest {
+            payment_request: self.outgoing_pay_req,
+            timeout_seconds: BTC_PAYMENT_TIMEOUT_SECONDS,
+            ..Default::default()
+        };
+        tracing::debug!("[inbounding tlc] SendPaymentRequest: {:?}", req);
+
+        let mut client = self.lnd_connection.create_router_client().await?;
+        // TODO: set a fee
+        let mut stream = client.send_payment_v2(req).await?.into_inner();
+        // Wait for the first message then quit
+        let payment_result_opt = stream.next().await;
+        tracing::debug!("[inbounding tlc] payment result: {:?}", payment_result_opt);
+        if let Some(Ok(payment)) = payment_result_opt {
+            self.cch_actor_ref.send_message(CchMessage::TrackingEvent(
+                map_lnd_payment_changed_event(payment)?,
+            ))?;
+        }
+        Ok(())
+    }
+}
+
+impl SendOutgoingPaymentDispatcher {
+    pub fn should_dispatch(order: &CchOrder) -> bool {
+        order.status == CchOrderStatus::IncomingAccepted
+    }
+
+    pub fn dispatch(
+        state: &mut CchState,
+        cch_actor_ref: &ActorRef<CchMessage>,
+        order: &CchOrder,
+    ) -> Option<Box<dyn ActionExecutor>> {
+        if !Self::should_dispatch(order) {
+            return None;
+        }
+
+        match dispatch_payment_handler(order) {
+            PaymentHandlerType::Fiber => Some(Box::new(SendFiberOutgoingPaymentExecutor {
+                cch_actor_ref: cch_actor_ref.clone(),
+                network_actor_ref: state.network_actor.clone(),
+                outgoing_pay_req: order.outgoing_pay_req.clone(),
+            })),
+            PaymentHandlerType::Lightning => Some(Box::new(SendLightningOutgoingPaymentExecutor {
+                cch_actor_ref: cch_actor_ref.clone(),
+                outgoing_pay_req: order.outgoing_pay_req.clone(),
+                lnd_connection: state.lnd_connection.clone(),
+            })),
+        }
+    }
+}

crates/fiber-lib/src/cch/actions/settle_incoming_invoice.rs

@@ -0,0 +1,95 @@
+use anyhow::Result;
+use lnd_grpc_tonic_client::invoicesrpc;
+use ractor::{call, ActorRef};
+
+use crate::{
+    cch::{
+        actions::{
+            backend_dispatchers::{dispatch_invoice_handler, InvoiceHandlerType},
+            ActionExecutor,
+        },
+        actor::CchState,
+        trackers::LndConnectionInfo,
+        CchMessage, CchOrder, CchOrderStatus,
+    },
+    fiber::{types::Hash256, NetworkActorCommand, NetworkActorMessage, ASSUME_NETWORK_ACTOR_ALIVE},
+};
+
+pub struct SettleIncomingInvoiceDispatcher;
+
+pub struct SettleFiberIncomingInvoiceExecutor {
+    payment_hash: Hash256,
+    payment_preimage: Hash256,
+    network_actor_ref: ActorRef<NetworkActorMessage>,
+}
+
+#[async_trait::async_trait]
+impl ActionExecutor for SettleFiberIncomingInvoiceExecutor {
+    async fn execute(self: Box<Self>) -> Result<()> {
+        let payment_hash = self.payment_hash;
+        let payment_preimage = self.payment_preimage;
+        let command = move |rpc_reply| -> NetworkActorMessage {
+            NetworkActorMessage::Command(NetworkActorCommand::SettleInvoice(
+                payment_hash,
+                payment_preimage,
+                rpc_reply,
+            ))
+        };
+        call!(self.network_actor_ref, command).expect(ASSUME_NETWORK_ACTOR_ALIVE)?;
+        Ok(())
+    }
+}
+
+pub struct SettleLightningIncomingInvoiceExecutor {
+    payment_preimage: Hash256,
+    lnd_connection: LndConnectionInfo,
+}
+
+#[async_trait::async_trait]
+impl ActionExecutor for SettleLightningIncomingInvoiceExecutor {
+    async fn execute(self: Box<Self>) -> Result<()> {
+        // settle the lnd invoice
+        let req = invoicesrpc::SettleInvoiceMsg {
+            preimage: self.payment_preimage.into(),
+        };
+        tracing::debug!("[settled tlc] SettleInvoiceMsg: {:?}", req);
+
+        let mut client = self.lnd_connection.create_invoices_client().await?;
+        // TODO: set a fee
+        let resp = client.settle_invoice(req).await?.into_inner();
+        tracing::debug!("[settled tlc] SettleInvoiceResp: {:?}", resp);
+        Ok(())
+    }
+}
+
+impl SettleIncomingInvoiceDispatcher {
+    pub fn should_dispatch(order: &CchOrder) -> bool {
+        order.status == CchOrderStatus::OutgoingSucceeded
+    }
+
+    pub fn dispatch(
+        state: &mut CchState,
+        _cch_actor_ref: &ActorRef<CchMessage>,
+        order: &CchOrder,
+    ) -> Option<Box<dyn ActionExecutor>> {
+        if !Self::should_dispatch(order) {
+            return None;
+        }
+        let payment_preimage = order.payment_preimage?;
+
+        match dispatch_invoice_handler(order) {
+            // `CchActor` will track all fiber invoices, so there's nothing to do here to track a single invoice.
+            InvoiceHandlerType::Fiber => Some(Box::new(SettleFiberIncomingInvoiceExecutor {
+                payment_hash: order.payment_hash,
+                payment_preimage,
+                network_actor_ref: state.network_actor.clone(),
+            })),
+            InvoiceHandlerType::Lightning => {
+                Some(Box::new(SettleLightningIncomingInvoiceExecutor {
+                    payment_preimage,
+                    lnd_connection: state.lnd_connection.clone(),
+                }))
+            }
+        }
+    }
+}