Refactor CCH order with FSM and actions

crates/fiber-bin/src/main.rs

@@ -21,7 +21,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};
@@ -330,7 +330,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(),
@@ -353,7 +355,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/Cargo.toml

@@ -68,6 +68,7 @@ home = "0.5.9"
 hyper = { version = "1.5" }
 jsonrpsee = { version = "0.25.1", features = ["client", "server", "macros"] }
 lnd-grpc-tonic-client = "0.3.0"
+tonic = "0.11.0"
 rocksdb = { package = "ckb-rocksdb", version = "=0.21.1", features = [
   "lz4",
 ], default-features = false }

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 @@
+pub(crate) mod backend_dispatchers;
+pub(crate) mod send_outgoing_payment;
+pub(crate) mod settle_incoming_invoice;
+pub(crate) 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,189 @@
+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::{
+        payment::{PaymentStatus, SendPaymentCommand},
+        types::Hash256,
+        NetworkActorCommand, NetworkActorMessage, ASSUME_NETWORK_ACTOR_ALIVE,
+    },
+};
+
+const BTC_PAYMENT_TIMEOUT_SECONDS: i32 = 60;
+
+pub struct SendOutgoingPaymentDispatcher;
+
+pub struct SendFiberOutgoingPaymentExecutor {
+    payment_hash: Hash256,
+    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 event = match call!(self.network_actor_ref, message).expect(ASSUME_NETWORK_ACTOR_ALIVE)
+        {
+            Ok(payment) => CchTrackingEvent::PaymentChanged {
+                payment_hash: payment.payment_hash,
+                payment_preimage: None,
+                status: payment.status,
+                failure_reason: None,
+            },
+            Err(err) if err.contains("Payment session already exists") => {
+                CchTrackingEvent::PaymentChanged {
+                    payment_hash: self.payment_hash,
+                    payment_preimage: None,
+                    status: PaymentStatus::Inflight,
+                    failure_reason: None,
+                }
+            }
+            Err(err) => {
+                let failure_reason = format!("SendFiberOutgoingPaymentExecutor failure: {:?}", err);
+                if Self::is_permanent_error(&err) {
+                    CchTrackingEvent::PaymentChanged {
+                        payment_hash: self.payment_hash,
+                        payment_preimage: None,
+                        status: PaymentStatus::Failed,
+                        failure_reason: Some(failure_reason),
+                    }
+                } else {
+                    return Err(anyhow!(failure_reason));
+                }
+            }
+        };
+
+        self.cch_actor_ref
+            .send_message(CchMessage::TrackingEvent(event))?;
+
+        Ok(())
+    }
+}
+
+impl SendFiberOutgoingPaymentExecutor {
+    fn is_permanent_error(err: &str) -> bool {
+        err.contains("InvalidParameter")
+    }
+}
+
+pub struct SendLightningOutgoingPaymentExecutor {
+    payment_hash: Hash256,
+    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!("SendLightningOutgoingPaymentExecutor req: {:?}", 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!(
+            "SendLightningOutgoingPaymentExecutor resp: {:?}",
+            payment_result_opt
+        );
+        let event = match payment_result_opt {
+            Some(Ok(payment)) => map_lnd_payment_changed_event(payment)?,
+            Some(Err(err)) if err.code() == tonic::Code::AlreadyExists => {
+                CchTrackingEvent::PaymentChanged {
+                    payment_hash: self.payment_hash,
+                    payment_preimage: None,
+                    status: PaymentStatus::Inflight,
+                    failure_reason: None,
+                }
+            }
+            Some(Err(err)) => {
+                let failure_reason =
+                    format!("SendLightningOutgoingPaymentExecutor failure: {:?}", err);
+                if Self::is_permanent_error(err) {
+                    CchTrackingEvent::PaymentChanged {
+                        payment_hash: self.payment_hash,
+                        payment_preimage: None,
+                        status: PaymentStatus::Failed,
+                        failure_reason: Some(failure_reason),
+                    }
+                } else {
+                    return Err(anyhow!(failure_reason));
+                }
+            }
+            None => {
+                return Err(anyhow!(
+                    "SendLightningOutgoingPaymentExecutor failed to get payment result because stream is closed"
+                ));
+            }
+        };
+        self.cch_actor_ref
+            .send_message(CchMessage::TrackingEvent(event))?;
+        Ok(())
+    }
+}
+
+impl SendLightningOutgoingPaymentExecutor {
+    fn is_permanent_error(status: tonic::Status) -> bool {
+        matches!(status.code(), tonic::Code::InvalidArgument)
+    }
+}
+
+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 {
+                payment_hash: order.payment_hash,
+                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 {
+                payment_hash: order.payment_hash,
+                cch_actor_ref: cch_actor_ref.clone(),
+                outgoing_pay_req: order.outgoing_pay_req.clone(),
+                lnd_connection: state.lnd_connection.clone(),
+            })),
+        }
+    }
+}