refactor(cch): refactor lnd trackers

- Define payment and invoice events to be sent to CCH for both fiber and
  lnd.
- These events can be emitted from fiber store changes in a later PR.
- Extract lnd trackers into its own module.
- Limit concurrent lnd invoice trackers to 5, scheduling them at
  5-minute intervals using a round-robin strategy.

Author: doitian

crates/fiber-lib/src/cch/actor.rs

@@ -40,8 +40,8 @@ pub enum CchError {
     ReceiveBTCOrderAlreadyPaid,
     #[error("ReceiveBTC received payment amount is too small")]
     ReceiveBTCReceivedAmountTooSmall,
-    #[error("ReceiveBTC expected preimage but missing")]
-    ReceiveBTCMissingPreimage,
+    #[error("Expect preimage in settled payment but missing")]
+    SettledPaymentMissingPreimage,
     #[error("System time error: {0}")]
     SystemTimeError(#[from] SystemTimeError),
     #[error("JSON serialization error: {0}")]

crates/fiber-lib/src/cch/error.rs

@@ -0,0 +1,97 @@
+use lnd_grpc_tonic_client::lnrpc;
+
+use crate::{cch::CchOrderStatus, fiber::types::Hash256};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum CchIncomingPaymentStatus {
+    // The income payment is in-flight
+    InFlight = 0,
+    // Incoming payment TLCs have been accepted
+    Accepted = 1,
+    Settled = 2,
+    Failed = 3,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum CchOutgoingPaymentStatus {
+    // The outgoing payment is in-flight
+    InFlight = 0,
+    Settled = 2,
+    Failed = 3,
+}
+
+impl From<CchIncomingPaymentStatus> for CchOrderStatus {
+    fn from(status: CchIncomingPaymentStatus) -> Self {
+        match status {
+            CchIncomingPaymentStatus::InFlight => CchOrderStatus::Pending,
+            CchIncomingPaymentStatus::Accepted => CchOrderStatus::IncomingAccepted,
+            CchIncomingPaymentStatus::Settled => CchOrderStatus::Succeeded,
+            CchIncomingPaymentStatus::Failed => CchOrderStatus::Failed,
+        }
+    }
+}
+
+impl From<CchOutgoingPaymentStatus> for CchOrderStatus {
+    fn from(status: CchOutgoingPaymentStatus) -> Self {
+        match status {
+            CchOutgoingPaymentStatus::InFlight => CchOrderStatus::OutgoingInFlight,
+            CchOutgoingPaymentStatus::Settled => CchOrderStatus::OutgoingSettled,
+            CchOutgoingPaymentStatus::Failed => CchOrderStatus::Failed,
+        }
+    }
+}
+
+/// Lnd invoice is the incoming part of a CCHOrder to receive BTC from Lightning to Fiber
+impl From<lnrpc::invoice::InvoiceState> for CchIncomingPaymentStatus {
+    fn from(state: lnrpc::invoice::InvoiceState) -> Self {
+        use lnrpc::invoice::InvoiceState;
+        match state {
+            InvoiceState::Open => CchIncomingPaymentStatus::InFlight,
+            InvoiceState::Settled => CchIncomingPaymentStatus::Settled,
+            InvoiceState::Canceled => CchIncomingPaymentStatus::Failed,
+            InvoiceState::Accepted => CchIncomingPaymentStatus::Accepted,
+        }
+    }
+}
+
+/// Lnd payment is the outgoing part of a CCHOrder to send BTC from Fiber to Lightning
+impl From<lnrpc::payment::PaymentStatus> for CchOutgoingPaymentStatus {
+    fn from(status: lnrpc::payment::PaymentStatus) -> Self {
+        use lnrpc::payment::PaymentStatus;
+        match status {
+            PaymentStatus::Unknown => CchOutgoingPaymentStatus::InFlight,
+            PaymentStatus::InFlight => CchOutgoingPaymentStatus::InFlight,
+            PaymentStatus::Succeeded => CchOutgoingPaymentStatus::Settled,
+            PaymentStatus::Failed => CchOutgoingPaymentStatus::Failed,
+            PaymentStatus::Initiated => CchOutgoingPaymentStatus::InFlight,
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub enum CchIncomingEvent {
+    InvoiceChanged {
+        /// The payment hash of the invoice.
+        payment_hash: Hash256,
+        /// The preimage of the invoice.
+        payment_preimage: Option<Hash256>,
+        status: CchIncomingPaymentStatus,
+    },
+
+    PaymentChanged {
+        /// The payment hash of the invoice.
+        payment_hash: Hash256,
+        /// The preimage of the invoice.
+        payment_preimage: Option<Hash256>,
+        status: CchOutgoingPaymentStatus,
+    },
+}
+
+impl CchIncomingEvent {
+    pub fn payment_hash(&self) -> &Hash256 {
+        match self {
+            CchIncomingEvent::InvoiceChanged { payment_hash, .. } => payment_hash,
+            CchIncomingEvent::PaymentChanged { payment_hash, .. } => payment_hash,
+        }
+    }
+}

crates/fiber-lib/src/cch/events.rs

@@ -4,6 +4,11 @@ pub use actor::{start_cch, CchActor, CchArgs, CchMessage, ReceiveBTC, SendBTC};
 mod error;
 pub use error::{CchError, CchResult};
 
+mod events;
+pub use events::{CchIncomingEvent, CchIncomingPaymentStatus, CchOutgoingPaymentStatus};
+mod trackers;
+pub use trackers::{LndConnectionInfo, LndTrackerActor, LndTrackerArgs, LndTrackerMessage};
+
 mod config;
 pub use config::{
     CchConfig, DEFAULT_BTC_FINAL_TLC_EXPIRY_TIME, DEFAULT_CKB_FINAL_TLC_EXPIRY_DELTA,
@@ -15,3 +20,6 @@ pub use order::{CchInvoice, CchOrder, CchOrderStatus};
 
 mod orders_db;
 pub use orders_db::CchOrdersDb;
+
+#[cfg(test)]
+pub mod tests;

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

@@ -1,5 +1,4 @@
 use lightning_invoice::Bolt11Invoice;
-use lnd_grpc_tonic_client::lnrpc;
 use serde::{Deserialize, Serialize};
 use serde_with::{serde_as, DisplayFromStr};
 
@@ -15,7 +14,7 @@ use crate::{
 #[derive(Debug, Copy, Clone, Serialize, Deserialize, Eq, PartialEq)]
 #[serde(rename_all = "snake_case")]
 pub enum CchOrderStatus {
-    /// Order is created and has not send out payments yet.
+    /// Order is created and has not received the incoming payment
     Pending = 0,
     /// HTLC in the incoming payment is accepted.
     IncomingAccepted = 1,
@@ -29,31 +28,6 @@ pub enum CchOrderStatus {
     Failed = 5,
 }
 
-/// Lnd payment is the outgoing part of a CCHOrder to send BTC from Fiber to Lightning
-impl From<lnrpc::payment::PaymentStatus> for CchOrderStatus {
-    fn from(status: lnrpc::payment::PaymentStatus) -> Self {
-        use lnrpc::payment::PaymentStatus;
-        match status {
-            PaymentStatus::Succeeded => CchOrderStatus::OutgoingSettled,
-            PaymentStatus::Failed => CchOrderStatus::Failed,
-            _ => CchOrderStatus::OutgoingInFlight,
-        }
-    }
-}
-
-/// Lnd invoice is the incoming part of a CCHOrder to receive BTC from Lightning to Fiber
-impl From<lnrpc::invoice::InvoiceState> for CchOrderStatus {
-    fn from(state: lnrpc::invoice::InvoiceState) -> Self {
-        use lnrpc::invoice::InvoiceState;
-        match state {
-            InvoiceState::Accepted => CchOrderStatus::IncomingAccepted,
-            InvoiceState::Canceled => CchOrderStatus::Failed,
-            InvoiceState::Settled => CchOrderStatus::Succeeded,
-            _ => CchOrderStatus::Pending,
-        }
-    }
-}
-
 /// The generated proxy invoice for the incoming payment.
 ///
 /// The JSON representation:

crates/fiber-lib/src/cch/order.rs

@@ -0,0 +1,186 @@
+use std::sync::Arc;
+
+use crate::{
+    cch::{LndConnectionInfo, LndTrackerActor, LndTrackerArgs, LndTrackerMessage},
+    fiber::types::Hash256,
+};
+
+use ractor::{concurrency::Duration as RactorDuration, Actor, ActorRef, OutputPort};
+use tokio_util::{sync::CancellationToken, task::TaskTracker};
+
+// Helper function to create test arguments
+fn create_test_args() -> LndTrackerArgs {
+    let port = Arc::new(OutputPort::default());
+    let tracker = TaskTracker::new();
+    let token = CancellationToken::new();
+    let lnd_connection = LndConnectionInfo {
+        // Tracker will keep running because this URI is unreachable
+        uri: "https://localhost:10009".parse().unwrap(),
+        cert: None,
+        macaroon: None,
+    };
+
+    LndTrackerArgs {
+        port,
+        lnd_connection,
+        token,
+        tracker,
+    }
+}
+
+// Helper function to create a test payment hash
+fn test_payment_hash(value: u8) -> Hash256 {
+    let mut bytes = [0u8; 32];
+    bytes[0] = value;
+    Hash256::from(bytes)
+}
+
+// Helper function to create a test `LndTrackerActor` (without spawning trackers)
+async fn create_test_actor() -> (ActorRef<LndTrackerMessage>, tokio::task::JoinHandle<()>) {
+    // Use spawn instead of spawn_linked to avoid needing a root actor
+    let args = create_test_args();
+    let (actor_ref, actor_handle) =
+        Actor::spawn(Some("test_lnd_tracker".to_string()), LndTrackerActor, args)
+            .await
+            .expect("Failed to spawn test actor");
+
+    (actor_ref, actor_handle)
+}
+
+// Test completion decrements active_invoice_trackers counter
+#[tokio::test]
+async fn test_completion_decrements_counter() {
+    let (actor_ref, _handle) = create_test_actor().await;
+    let payment_hash = test_payment_hash(1);
+
+    // Add invoice to queue (without processing to avoid LND calls)
+    actor_ref
+        .cast(LndTrackerMessage::TrackInvoice(payment_hash))
+        .expect("Failed to send TrackInvoice");
+    tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
+
+    // Send completion message (simulating a tracker that finished)
+    actor_ref
+        .cast(LndTrackerMessage::InvoiceTrackerCompleted {
+            payment_hash,
+            completed_successfully: true,
+        })
+        .expect("Failed to send completion");
+
+    tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
+
+    // Verify counter behavior (should handle completion gracefully)
+    let final_state = actor_ref
+        .call(
+            |reply_port| LndTrackerMessage::GetState(reply_port),
+            Some(RactorDuration::from_millis(1000)),
+        )
+        .await
+        .expect("Actor should be responsive after completion");
+
+    assert!(final_state.is_success());
+    let final_state = final_state.unwrap();
+    assert_eq!(final_state.invoice_queue_len, 0);
+    assert_eq!(final_state.active_invoice_trackers, 0);
+}
+
+// Test completion triggers queue processing for waiting invoices
+#[tokio::test]
+async fn test_completion_triggers_queue_processing() {
+    let (actor_ref, _handle) = create_test_actor().await;
+
+    // Add 6 invoices to queue
+    for i in 0..6 {
+        let payment_hash = test_payment_hash(i);
+        actor_ref
+            .cast(LndTrackerMessage::TrackInvoice(payment_hash))
+            .expect("Failed to send TrackInvoice");
+    }
+
+    tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
+
+    // Verify invoices are queued
+    let state_before = actor_ref
+        .call(
+            |reply_port| LndTrackerMessage::GetState(reply_port),
+            Some(RactorDuration::from_millis(1000)),
+        )
+        .await
+        .expect("Failed to get state")
+        .expect("Failed to get state");
+
+    assert_eq!(
+        state_before.invoice_queue_len, 1,
+        "Should have 1 invoice in queue"
+    );
+    assert_eq!(
+        state_before.active_invoice_trackers, 5,
+        "Should have 5 active invoice trackers"
+    );
+
+    let completed_hash = test_payment_hash(1);
+    actor_ref
+        .cast(LndTrackerMessage::InvoiceTrackerCompleted {
+            payment_hash: completed_hash,
+            completed_successfully: true,
+        })
+        .expect("Failed to send completion");
+
+    tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
+
+    // Verify actor is still responsive
+    let state_after = actor_ref
+        .call(
+            |reply_port| LndTrackerMessage::GetState(reply_port),
+            Some(RactorDuration::from_millis(1000)),
+        )
+        .await
+        .expect("Failed to get state")
+        .expect("Failed to get state");
+
+    assert_eq!(
+        state_after.invoice_queue_len, 0,
+        "Should have 0 invoices in queue"
+    );
+    assert_eq!(
+        state_after.active_invoice_trackers, 5,
+        "Should have 5 active invoice trackers"
+    );
+}
+
+// Test timeout re-queues active invoices to end of queue
+#[tokio::test]
+async fn test_timeout_requeues_active_invoices() {
+    let (actor_ref, _handle) = create_test_actor().await;
+    let payment_hash = test_payment_hash(1);
+
+    // Add invoice to queue (without processing to avoid LND calls)
+    actor_ref
+        .cast(LndTrackerMessage::TrackInvoice(payment_hash))
+        .expect("Failed to send TrackInvoice");
+    tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
+
+    // Send completion message (simulating a tracker that finished)
+    actor_ref
+        .cast(LndTrackerMessage::InvoiceTrackerCompleted {
+            payment_hash,
+            completed_successfully: false,
+        })
+        .expect("Failed to send completion");
+
+    tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
+
+    // Verify counter behavior (should handle completion gracefully)
+    let final_state = actor_ref
+        .call(
+            |reply_port| LndTrackerMessage::GetState(reply_port),
+            Some(RactorDuration::from_millis(1000)),
+        )
+        .await
+        .expect("Actor should be responsive after completion");
+
+    assert!(final_state.is_success());
+    let final_state = final_state.unwrap();
+    assert_eq!(final_state.invoice_queue_len, 0);
+    assert_eq!(final_state.active_invoice_trackers, 1);
+}

crates/fiber-lib/src/cch/tests/lnd_trackers_tests.rs

@@ -0,0 +1 @@
+mod lnd_trackers_tests;