Aggregates refactor and lightning aggregate

Cargo.toml

@@ -6,7 +6,6 @@ edition = "2021"
 [dependencies]
 payday_core = { path = "./payday_core" }
 payday_node_lnd = { path = "./payday_node_lnd" }
-payday_btc = { path = "./payday_btc" }
 payday_surrealdb = { path = "./payday_surrealdb" }
 payday_postgres = { path = "./payday_postgres" }
 tokio.workspace = true
@@ -17,7 +16,6 @@ tokio-stream.workspace = true
 
 [workspace]
 members = [
-  "payday_btc",
   "payday_core",
   "payday_node_lnd",
   "payday_postgres",
@@ -31,7 +29,8 @@ serde = { version = "1.0.203", features = ["derive"] }
 serde_json = "1.0.118"
 tokio-stream = "0.1.15"
 chrono = "0.4"
-cqrs-es = "0.4.11"
+cqrs-es = "0.4.12"
+postgres-es = "0.4.12"
 tokio = { version = "1.38.0", features = ["full"] }
 sqlx = { version = "0.8", features = ["postgres", "json"] }
 futures = "0.3.30"

payday_btc/src/lib.rs

@@ -5,9 +5,3 @@ use std::str::FromStr;
 
 use bitcoin::{Address, Network};
 use payday_core::Result;
-
-/// Given a Bitcoin address string and a network, parses and validates the address.
-/// Returns a checked address result.
-pub fn to_address(addr: &str, network: Network) -> Result<Address> {
-    Ok(Address::from_str(addr)?.require_network(network)?)
-}

payday_core/src/aggregate/lightning_aggregate.rs

@@ -0,0 +1,162 @@
+use crate::{
+    api::lightining_api::LightningTransactionEvent,
+    payment::{
+        amount::Amount,
+        currency::Currency,
+        invoice::{Error, InvoiceId},
+    },
+};
+use async_trait::async_trait;
+use cqrs_es::{Aggregate, DomainEvent};
+use lightning_invoice::Bolt11Invoice;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct LightningInvoice {
+    pub invoice_id: InvoiceId,
+    pub node_id: String,
+    pub r_hash: String,
+    pub invoice: String,
+    pub amount: Amount,
+    pub received_amount: Amount,
+    pub overpaid: bool,
+    pub paid: bool,
+}
+
+#[async_trait]
+pub trait LightningInvoiceService: Send + Sync {}
+
+pub struct LightningInvoiceServices {}
+
+#[allow(clippy::large_enum_variant)]
+#[derive(Debug, Deserialize)]
+pub enum LightningInvoiceCommand {
+    CreateInvoice {
+        invoice_id: InvoiceId,
+        node_id: String,
+        amount: Amount,
+        invoice: Bolt11Invoice,
+    },
+    SettleInvoice {
+        received_amount: Amount,
+    },
+}
+
+impl From<LightningTransactionEvent> for LightningInvoiceCommand {
+    fn from(event: LightningTransactionEvent) -> Self {
+        match event {
+            LightningTransactionEvent::Settled(tx) => LightningInvoiceCommand::SettleInvoice {
+                received_amount: tx.amount_paid,
+            },
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub enum LightningInvoiceEvent {
+    InvoiceCreated {
+        invoice_id: InvoiceId,
+        node_id: String,
+        r_hash: String,
+        amount: Amount,
+        invoice: String,
+    },
+    InvoiceSettled {
+        received_amount: Amount,
+        overpaid: bool,
+        paid: bool,
+    },
+}
+
+impl DomainEvent for LightningInvoiceEvent {
+    fn event_type(&self) -> String {
+        let event_type = match self {
+            LightningInvoiceEvent::InvoiceCreated { .. } => "LightningInvoiceCreated",
+            LightningInvoiceEvent::InvoiceSettled { .. } => "LightningInvoiceSettled",
+        };
+        event_type.to_string()
+    }
+
+    fn event_version(&self) -> String {
+        "1.0.0".to_string()
+    }
+}
+
+#[async_trait]
+impl Aggregate for LightningInvoice {
+    type Command = LightningInvoiceCommand;
+    type Event = LightningInvoiceEvent;
+    type Error = Error;
+    type Services = ();
+
+    fn aggregate_type() -> String {
+        "LightningInvoice".to_string()
+    }
+
+    async fn handle(
+        &self,
+        command: Self::Command,
+        _service: &Self::Services,
+    ) -> Result<Vec<Self::Event>, Self::Error> {
+        match command {
+            LightningInvoiceCommand::CreateInvoice {
+                invoice_id,
+                node_id,
+                amount,
+                invoice,
+            } => {
+                if amount.currency != Currency::Btc {
+                    return Err(Error::InvalidCurrency(
+                        amount.currency.to_string(),
+                        Currency::Btc.to_string(),
+                    ));
+                }
+
+                let r_hash = invoice.payment_hash().to_string();
+                let invoice = invoice.to_string();
+
+                Ok(vec![LightningInvoiceEvent::InvoiceCreated {
+                    invoice_id,
+                    node_id,
+                    r_hash,
+                    amount,
+                    invoice,
+                }])
+            }
+            LightningInvoiceCommand::SettleInvoice { received_amount } => {
+                Ok(vec![LightningInvoiceEvent::InvoiceSettled {
+                    received_amount,
+                    overpaid: received_amount.amount > self.amount.amount,
+                    paid: received_amount.amount >= self.amount.amount,
+                }])
+            }
+        }
+    }
+
+    fn apply(&mut self, event: Self::Event) {
+        match event {
+            LightningInvoiceEvent::InvoiceCreated {
+                invoice_id,
+                node_id,
+                r_hash,
+                amount,
+                invoice,
+            } => {
+                self.invoice_id = invoice_id;
+                self.node_id = node_id;
+                self.r_hash = r_hash;
+                self.amount = amount;
+                self.invoice = invoice;
+            }
+            LightningInvoiceEvent::InvoiceSettled {
+                received_amount,
+                overpaid,
+                paid,
+            } => {
+                self.received_amount = received_amount;
+                self.overpaid = overpaid;
+                self.paid = paid;
+            }
+        }
+    }
+}

payday_core/src/aggregate/mod.rs

@@ -0,0 +1,2 @@
+pub mod lightning_aggregate;
+pub mod on_chain_aggregate;

payday_btc/src/on_chain_aggregate.rs → payday_core/src/aggregate/on_chain_aggregate.rs

@@ -1,13 +1,17 @@
+use crate::{
+    api::on_chain_api::OnChainTransactionEvent,
+    payment::{
+        amount::Amount,
+        currency::Currency,
+        invoice::{Error, InvoiceId},
+    },
+};
 use async_trait::async_trait;
 use cqrs_es::{Aggregate, DomainEvent};
-use payday_core::api::on_chain_api::OnChainTransactionEvent;
-use payday_core::payment::amount::Amount;
-use payday_core::payment::currency::Currency;
-use payday_core::payment::invoice::{Error, InvoiceId};
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct BtcOnChainInvoice {
+pub struct OnChainInvoice {
     pub invoice_id: InvoiceId,
     pub node_id: String,
     pub address: String,
@@ -20,7 +24,7 @@ pub struct BtcOnChainInvoice {
     pub paid: bool,
 }
 
-impl Default for BtcOnChainInvoice {
+impl Default for OnChainInvoice {
     fn default() -> Self {
         Self {
             invoice_id: "".to_string(),
@@ -143,14 +147,14 @@ impl DomainEvent for OnChainInvoiceEvent {
 }
 
 #[async_trait]
-impl Aggregate for BtcOnChainInvoice {
+impl Aggregate for OnChainInvoice {
     type Command = OnChainInvoiceCommand;
     type Event = OnChainInvoiceEvent;
     type Error = Error;
     type Services = ();
 
     fn aggregate_type() -> String {
-        "BtcOnChainInvoice".to_string()
+        "OnChainInvoice".to_string()
     }
 
     async fn handle(
@@ -242,12 +246,12 @@ impl Aggregate for BtcOnChainInvoice {
 
 #[cfg(test)]
 mod aggregate_tests {
+    use crate::payment::currency::Currency;
     use cqrs_es::test::TestFramework;
-    use payday_core::payment::currency::Currency;
 
     use super::*;
 
-    type OnChainInvoiceTestFramework = TestFramework<BtcOnChainInvoice>;
+    type OnChainInvoiceTestFramework = TestFramework<OnChainInvoice>;
 
     #[test]
     fn test_create_invoice() {

payday_core/src/api/mod.rs

@@ -1,2 +1,3 @@
 pub mod lightining_api;
 pub mod on_chain_api;
+pub mod payment_event;

payday_core/src/api/payment_event.rs

@@ -0,0 +1,49 @@
+use crate::payment::amount::Amount;
+use crate::payment::invoice::InvoiceId;
+use crate::Result;
+
+pub trait PaymentEventPublisherApi {
+    /// Publishes a transaction event.
+    fn publish_received_event(&self, event: PaymentReceivedEvent) -> Result<()>;
+}
+
+pub enum PaymentReceivedEvent {
+    /// A payment for an invoice has been received but not confirmed yet.
+    OnChainUnconfirmed(OnChainPaymentEvent),
+    /// A payment for an invoice has been received and has been confirmed.
+    OnChainConfirmed(OnChainPaymentEvent),
+    /// We received a payment where there is no invoice in the system.
+    OnChainUnexpected(OnChainUexpectedPaymentEvent),
+    /// We received a payment for an invoice that has already been paid.
+    OnChainUsedAddress(OnChainUsedAddressPaymentEvent),
+}
+
+pub struct OnChainData {
+    pub node_id: String,
+    pub address: String,
+    pub confirmations: u64,
+    pub transaction_id: Option<String>,
+}
+
+pub struct OnChainPaymentEvent {
+    pub invoice_id: InvoiceId,
+    pub invoice_amount: Amount,
+    pub received_amount: Amount,
+    pub underpayment: bool,
+    pub overpayment: bool,
+    pub on_chain_data: OnChainData,
+    pub paid: bool,
+}
+
+pub struct OnChainUexpectedPaymentEvent {
+    pub received_amount: Amount,
+    pub on_chain_data: OnChainData,
+    pub paid: bool,
+}
+
+pub struct OnChainUsedAddressPaymentEvent {
+    pub received_amount: Amount,
+    pub on_chain_data: OnChainData,
+    pub original_invoice_id: InvoiceId,
+    pub paid: bool,
+}

payday_core/src/lib.rs

@@ -1,15 +1,12 @@
-use std::pin::Pin;
-
-use tokio_stream::Stream;
-
 pub use error::Error;
 
+pub mod aggregate;
 pub mod api;
 pub mod date;
 pub mod error;
 pub mod events;
 pub mod payment;
 pub mod persistence;
+pub mod processor;
 
 pub type Result<T> = std::result::Result<T, Error>;
-pub type PaydayStream<T> = Pin<Box<dyn Stream<Item = T>>>;

payday_core/src/processor/mod.rs

@@ -0,0 +1 @@
+pub mod on_chain_processor;