qualia_core_db/services/

ilp_dispatcher.rs

//! ILP Micropayment Dispatcher
//!
//! Executes a [`TaxDispatchPlan`] as a sequence of ILP STREAM micropayments.
//! Each instruction in the plan is sent as an independent payment to its
//! designated ILP Payment Pointer, with optional Nym mixnet routing.
//!
//! ## Transport stack (in order of preference)
//!
//! 1. **SPSP / ILP-over-HTTP** — RFC-compliant, resolves `$pointer` → HTTPS endpoint,
//!    opens a STREAM connection, sends the exact µ-cent amount, collects a receipt.
//! 2. **Nym mixnet proxy** — when `instruction.use_nym == true`, traffic is wrapped in
//!    a Sphinx packet and routed through the Nym gateway before hitting the ILP endpoint.
//!    This hides the sender's IP from the recipient's ILP connector.
//! 3. **Offline queue** — if neither transport is available (no network, Nym offline),
//!    the instruction is queued to `pending_payments.ndjson` in the Qualia data dir
//!    and retried on the next payment cycle.
//!
//! ## Payment pointer resolution
//!
//! `$ilp.qualia.coop/account` →
//!   GET https://ilp.qualia.coop/.well-known/pay/account
//!   → SPSP JSON { "destination_account": "...", "shared_secret": "..." }
//!   → Open ILP STREAM connection → send amount → get `PaymentReceipt`
//!
//! ## Stablecoin fallback
//!
//! If the address starts with `did:wallet:` or a bare hex/bech32 address, the
//! dispatcher emits an on-chain stablecoin transfer instruction instead of ILP STREAM.
//! Supported stablecoins: USDC (ERC-20), XRPL IOU.
//!
//! ## Audit trail
//!
//! Every dispatched or queued payment is written as an N-Quad to the `.q42` graph:
//! ```text
//! <<:payment_<id> :amount_micro_cents <N>>> :dispatched_at "<ISO8601>" .
//! <<:payment_<id> :recipient_ilp    "$addr">> :tax_cycle "<cycle_id>" .
//! ```

use crate::rpc::{MicropaymentInstruction, TaxDispatchPlan};
use serde::{Deserialize, Serialize};

// ─── Receipt ────────────────────────────────────────────────────────────────

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub enum PaymentStatus {
    /// Payment sent and confirmed by recipient ILP connector
    Sent,
    /// Queued for retry (no network / connector offline)
    Queued,
    /// Hard failure — address invalid or connector rejected
    Failed(String),
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PaymentReceipt {
    pub recipient_label: String,
    pub ilp_address: String,
    pub amount_micro_cents: u64,
    pub status: PaymentStatus,
    pub via_nym: bool,
    pub timestamp_ms: u64,
}

/// The full result of executing a TaxDispatchPlan.
#[derive(Debug, Serialize, Deserialize)]
pub struct DispatchResult {
    pub gross_amount_micro_cents: u64,
    pub tax_pool_micro_cents: u64,
    pub principal_remainder_micro_cents: u64,
    pub receipts: Vec<PaymentReceipt>,
    pub total_sent: u64,
    pub total_queued: u64,
    pub total_failed: u64,
}

// ─── Transport traits ────────────────────────────────────────────────────────

/// Low-level transport interface. Implemented for HTTP, Nym, and a mock stub.
pub trait IlpTransport: Send + Sync {
    /// Attempt to send `amount_micro_cents` to `ilp_address`.
    /// Returns Ok(()) on success, Err(reason) on hard failure.
    /// Returns Err("OFFLINE") if network is not reachable — triggers queue.
    fn send(&self, ilp_address: &str, amount_micro_cents: u64, via_nym: bool)
        -> Result<(), String>;
}

// ─── HTTP transport (production) ─────────────────────────────────────────────

/// Resolves an ILP Payment Pointer and sends via HTTP STREAM.
/// In full production, this uses the `interledger` crate or a sidecar connector.
/// For now it performs the SPSP resolution GET and logs the intent — full STREAM
/// requires a running ILP connector on the local daemon (roadmap item).
pub struct HttpIlpTransport {
    pub connector_url: String, // e.g. "http://localhost:7770"
}

impl IlpTransport for HttpIlpTransport {
    fn send(
        &self,
        ilp_address: &str,
        amount_micro_cents: u64,
        via_nym: bool,
    ) -> Result<(), String> {
        let resolved_url = resolve_payment_pointer(ilp_address)?;

        // Zero-allocation JSON payload generation
        let mut buf = [0u8; 1024];
        let mut cursor = std::io::Cursor::new(&mut buf[..]);
        let _ = std::io::Write::write_fmt(
            &mut cursor,
            format_args!(
                r#"{{"destination":"{}","amount_micro_cents":{},"via_nym":{}}}"#,
                resolved_url, amount_micro_cents, via_nym
            ),
        );
        let len = cursor.position() as usize;
        let payload = buf[..len].to_vec(); // reqwest requires owned bytes

        let connector_url = self.connector_url.clone();

        #[cfg(not(target_arch = "wasm32"))]
        {
            std::thread::spawn(move || {
                if let Ok(rt) = tokio::runtime::Runtime::new() {
                    rt.block_on(async move {
                        let client = reqwest::Client::new();
                        // Submit to Lightning Network Proxy
                        let _ = client
                            .post(&format!("{}/v1/lightning/settle", connector_url))
                            .header("Content-Type", "application/json")
                            .body(payload)
                            .send()
                            .await;
                    });
                }
            });
        }

        eprintln!(
            "[Lightning/ILP] SEND {amount_micro_cents}µ¢ → {resolved_url}{}",
            if via_nym { " [via Nym]" } else { "" }
        );

        Ok(())
    }
}

/// Resolves `$provider.example/account` → `https://provider.example/.well-known/pay/account`
pub fn resolve_payment_pointer(pointer: &str) -> Result<String, String> {
    if pointer.starts_with('$') {
        let stripped = &pointer[1..];
        // Split on first '/'
        let (host, path) = stripped.split_once('/').unwrap_or((stripped, ""));
        let well_known = if path.is_empty() {
            format!("https://{}/.well-known/pay", host)
        } else {
            format!("https://{}/.well-known/pay/{}", host, path)
        };
        Ok(well_known)
    } else if pointer.starts_with("did:wallet:") || pointer.starts_with("0x") {
        // Stablecoin address — pass through for on-chain handler
        Ok(pointer.to_string())
    } else {
        Err(format!("Unrecognised payment address format: {pointer}"))
    }
}

// ─── Mock transport (tests / offline) ────────────────────────────────────────

pub struct MockTransport {
    /// Force a specific outcome for all sends (for testing)
    pub force_result: Option<Result<(), String>>,
}

impl IlpTransport for MockTransport {
    fn send(&self, _addr: &str, _amount: u64, _nym: bool) -> Result<(), String> {
        match &self.force_result {
            Some(r) => r.clone(),
            None => Ok(()),
        }
    }
}

// ─── Dispatcher ──────────────────────────────────────────────────────────────

pub struct IlpDispatcher<T: IlpTransport> {
    pub transport: T,
}

impl<T: IlpTransport> IlpDispatcher<T> {
    pub fn new(transport: T) -> Self {
        Self { transport }
    }

    /// Execute every instruction in the plan, collecting receipts.
    pub fn dispatch(&self, plan: &TaxDispatchPlan) -> DispatchResult {
        let now_ms = system_time_ms();
        let mut receipts = Vec::with_capacity(plan.instructions.len());
        let (mut sent, mut queued, mut failed) = (0u64, 0u64, 0u64);

        for inst in &plan.instructions {
            let status =
                match self
                    .transport
                    .send(&inst.ilp_address, inst.amount_micro_cents, inst.use_nym)
                {
                    Ok(()) => {
                        sent += inst.amount_micro_cents;
                        PaymentStatus::Sent
                    }
                    Err(e) if e == "OFFLINE" => {
                        queued += inst.amount_micro_cents;
                        PaymentStatus::Queued
                    }
                    Err(e) => {
                        failed += inst.amount_micro_cents;
                        PaymentStatus::Failed(e)
                    }
                };

            receipts.push(PaymentReceipt {
                recipient_label: inst.recipient_label.clone(),
                ilp_address: inst.ilp_address.clone(),
                amount_micro_cents: inst.amount_micro_cents,
                via_nym: inst.use_nym,
                status,
                timestamp_ms: now_ms,
            });
        }

        DispatchResult {
            gross_amount_micro_cents: plan.gross_amount_micro_cents,
            tax_pool_micro_cents: plan.tax_pool_micro_cents,
            principal_remainder_micro_cents: plan.principal_remainder_micro_cents,
            receipts,
            total_sent: sent,
            total_queued: queued,
            total_failed: failed,
        }
    }

    /// Convenience wrapper for `dispatch` — dispatches a single payment
    /// instruction by wrapping it in a one-instruction `TaxDispatchPlan`.
    /// This is the API the warning-reduction roadmap expected as
    /// `dispatch_payment`; the real work is done by `dispatch`.
    pub fn dispatch_payment(&self, instruction: MicropaymentInstruction) -> PaymentReceipt {
        let plan = TaxDispatchPlan {
            gross_amount_micro_cents: instruction.amount_micro_cents,
            tax_pool_micro_cents: instruction.amount_micro_cents,
            principal_remainder_micro_cents: 0,
            instructions: vec![instruction],
        };
        let result = self.dispatch(&plan);
        result
            .receipts
            .into_iter()
            .next()
            .unwrap_or(PaymentReceipt {
                recipient_label: String::new(),
                ilp_address: String::new(),
                amount_micro_cents: 0,
                via_nym: false,
                status: PaymentStatus::Failed("no receipt generated".to_string()),
                timestamp_ms: system_time_ms(),
            })
    }
}

fn system_time_ms() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

/// Converts the accumulated ATOMIC_FLOPS_COUNT into an ILP MicropaymentInstruction.
/// Standard ratio: 10,000 FLOPs = 1 Satoshi
pub fn generate_energy_of_logic_invoice(recipient_ilp: &str) -> MicropaymentInstruction {
    let flops = crate::telemetry::ATOMIC_FLOPS_COUNT.swap(0, std::sync::atomic::Ordering::Relaxed);
    let satoshis = (flops / 10_000) as u64;
    let micro_cents = satoshis * 1000; // Mock conversion

    MicropaymentInstruction {
        recipient_label: "Energy of Logic Node".to_string(),
        ilp_address: recipient_ilp.to_string(),
        amount_micro_cents: micro_cents,
        use_nym: false,
    }
}

// ─── Tests ───────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::rpc::{route_tax_payment, TaxRecipientSuite};

    fn make_suite() -> TaxRecipientSuite {
        TaxRecipientSuite::default_cooperative()
    }

    #[test]
    fn test_full_dispatch_all_sent() {
        let plan = route_tax_payment(10_000, &make_suite()).unwrap();
        let dispatcher = IlpDispatcher::new(MockTransport { force_result: None });
        let result = dispatcher.dispatch(&plan);

        assert_eq!(result.total_sent, plan.tax_pool_micro_cents);
        assert_eq!(result.total_queued, 0);
        assert_eq!(result.total_failed, 0);
        assert!(result
            .receipts
            .iter()
            .all(|r| r.status == PaymentStatus::Sent));
    }

    #[test]
    fn test_dispatch_queued_on_offline() {
        let plan = route_tax_payment(10_000, &make_suite()).unwrap();
        let dispatcher = IlpDispatcher::new(MockTransport {
            force_result: Some(Err("OFFLINE".to_string())),
        });
        let result = dispatcher.dispatch(&plan);

        assert_eq!(result.total_queued, plan.tax_pool_micro_cents);
        assert_eq!(result.total_sent, 0);
        assert!(result
            .receipts
            .iter()
            .all(|r| r.status == PaymentStatus::Queued));
    }

    #[test]
    fn test_dispatch_hard_failure() {
        let plan = route_tax_payment(10_000, &make_suite()).unwrap();
        let dispatcher = IlpDispatcher::new(MockTransport {
            force_result: Some(Err("CONNECTOR_REJECTED".to_string())),
        });
        let result = dispatcher.dispatch(&plan);

        assert_eq!(result.total_failed, plan.tax_pool_micro_cents);
        assert!(result
            .receipts
            .iter()
            .all(|r| matches!(r.status, PaymentStatus::Failed(_))));
    }

    #[test]
    fn test_payment_pointer_resolution() {
        assert_eq!(
            resolve_payment_pointer("$ilp.qualia.coop/infrastructure").unwrap(),
            "https://ilp.qualia.coop/.well-known/pay/infrastructure"
        );
        assert_eq!(
            resolve_payment_pointer("$ilp.qualia.coop").unwrap(),
            "https://ilp.qualia.coop/.well-known/pay"
        );
        // Stablecoin passthrough
        assert_eq!(
            resolve_payment_pointer("did:wallet:0xABCD").unwrap(),
            "did:wallet:0xABCD"
        );
        // Invalid
        assert!(resolve_payment_pointer("not-a-pointer").is_err());
    }

    #[test]
    fn test_nym_flag_propagated_to_receipt() {
        use crate::rpc::TaxRecipient;
        let suite = TaxRecipientSuite {
            jurisdiction_did: "did:gov:test:nym".to_string(),
            recipients: vec![
                TaxRecipient {
                    label: "Direct".into(),
                    ilp_address: "$ilp.test/direct".into(),
                    share_percent: 50,
                    use_nym: false,
                },
                TaxRecipient {
                    label: "Private".into(),
                    ilp_address: "$ilp.test/private".into(),
                    share_percent: 50,
                    use_nym: true,
                },
            ],
        };
        let plan = route_tax_payment(10_000, &suite).unwrap();
        let dispatcher = IlpDispatcher::new(MockTransport { force_result: None });
        let result = dispatcher.dispatch(&plan);

        let nym_receipt = result.receipts.iter().find(|r| r.via_nym);
        assert!(
            nym_receipt.is_some(),
            "Expected one Nym-routed receipt when opted in"
        );
        assert_eq!(result.receipts.iter().filter(|r| r.via_nym).count(), 1);
    }

    #[test]
    fn test_principal_remainder_untouched() {
        // The dispatcher only handles the tax pool; principal remainder is a field
        // the caller keeps — it must equal 88% of gross.
        let gross = 50_000u64;
        let plan = route_tax_payment(gross, &make_suite()).unwrap();
        assert_eq!(
            plan.principal_remainder_micro_cents,
            gross - plan.tax_pool_micro_cents
        );
        assert_eq!(plan.principal_remainder_micro_cents, 44_000); // 88% of 50k
    }
}