feat: cleaning of cancelled invoices

protos/hold.proto

@@ -12,6 +12,9 @@ service Hold {
   rpc Settle (SettleRequest) returns (SettleResponse) {}
   rpc Cancel (CancelRequest) returns (CancelResponse) {}
 
+  // Cleans cancelled invoices
+  rpc Clean (CleanRequest) returns (CleanResponse) {}
+
   rpc Track (TrackRequest) returns (stream TrackResponse) {}
   rpc TrackAll (TrackAllRequest) returns (stream TrackAllResponse) {}
 }
@@ -103,6 +106,14 @@ message CancelRequest {
 }
 message CancelResponse {}
 
+message CleanRequest {
+  // Clean everything older than age seconds
+  optional uint64 age = 1;
+}
+message CleanResponse {
+  uint64 cleaned = 1;
+}
+
 message TrackRequest {
   bytes payment_hash = 1;
 }

src/commands/clean.rs

@@ -0,0 +1,44 @@
+use crate::commands::structs::{parse_args, FromArr, ParamsError};
+use crate::database::helpers::invoice_helper::InvoiceHelper;
+use crate::encoder::InvoiceEncoder;
+use crate::State;
+use cln_plugin::Plugin;
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+#[derive(Debug, Deserialize)]
+struct CleanRequest {
+    age: Option<u64>,
+}
+
+impl FromArr for CleanRequest {
+    fn from_arr(arr: Vec<Value>) -> anyhow::Result<Self>
+    where
+        Self: Sized,
+    {
+        if arr.is_empty() {
+            return Ok(Self { age: None });
+        }
+
+        Ok(Self {
+            age: Some(arr[0].as_u64().ok_or(ParamsError::ParseError)?),
+        })
+    }
+}
+
+#[derive(Debug, Serialize)]
+struct CleanResponse {
+    pub cleaned: usize,
+}
+
+pub async fn clean<T, E>(plugin: Plugin<State<T, E>>, args: Value) -> anyhow::Result<Value>
+where
+    T: InvoiceHelper + Sync + Send + Clone,
+    E: InvoiceEncoder + Sync + Send + Clone,
+{
+    let params = parse_args::<CleanRequest>(args)?;
+
+    let cleaned = plugin.state().invoice_helper.clean_cancelled(params.age)?;
+
+    Ok(serde_json::to_value(&CleanResponse { cleaned })?)
+}

src/commands/mod.rs

@@ -1,10 +1,12 @@
 mod cancel;
+mod clean;
 mod invoice;
 mod list;
 mod settle;
 mod structs;
 
 pub use cancel::cancel;
+pub use clean::clean;
 pub use invoice::invoice;
 pub use list::list_invoices;
 pub use settle::settle;

src/database/helpers/invoice_helper.rs

@@ -3,13 +3,20 @@ use crate::database::model::{
 };
 use crate::database::schema::{htlcs, invoices};
 use crate::database::Pool;
-use anyhow::Result;
-use diesel::{insert_into, update, BelongingToDsl, ExpressionMethods, GroupedBy};
+use anyhow::{anyhow, Result};
+use chrono::{TimeDelta, Utc};
+use diesel::dsl::delete;
+use diesel::{
+    insert_into, update, BelongingToDsl, BoolExpressionMethods, Connection, ExpressionMethods,
+    GroupedBy,
+};
 use diesel::{QueryDsl, RunQueryDsl, SelectableHelper};
+use std::ops::Sub;
 
 pub trait InvoiceHelper {
     fn insert(&self, invoice: &InvoiceInsertable) -> Result<usize>;
     fn insert_htlc(&self, htlc: &HtlcInsertable) -> Result<usize>;
+
     fn set_invoice_state(
         &self,
         id: i64,
@@ -29,6 +36,9 @@ pub trait InvoiceHelper {
         state: InvoiceState,
         new_state: InvoiceState,
     ) -> Result<usize>;
+
+    fn clean_cancelled(&self, age: Option<u64>) -> Result<usize>;
+
     fn get_all(&self) -> Result<Vec<HoldInvoice>>;
     fn get_paginated(&self, index_start: i64, limit: u64) -> Result<Vec<HoldInvoice>>;
     fn get_by_payment_hash(&self, payment_hash: &[u8]) -> Result<Option<HoldInvoice>>;
@@ -107,6 +117,35 @@ impl InvoiceHelper for InvoiceHelperDatabase {
             .execute(&mut self.pool.get()?)?)
     }
 
+    fn clean_cancelled(&self, age: Option<u64>) -> Result<usize> {
+        let age = match TimeDelta::new(age.unwrap_or(0) as i64, 0) {
+            Some(age) => age,
+            None => return Err(anyhow!("invalid age")),
+        };
+
+        let now = Utc::now().naive_utc().sub(age);
+
+        let mut con = self.pool.get()?;
+        con.transaction(|tx| {
+            let invoice_clause = invoices::dsl::state
+                .eq(InvoiceState::Cancelled.to_string())
+                .and(invoices::dsl::created_at.le(now));
+
+            let invoices = invoices::dsl::invoices
+                .select(Invoice::as_select())
+                .filter(invoice_clause.clone())
+                .load(tx)?;
+
+            delete(
+                htlcs::dsl::htlcs
+                    .filter(htlcs::dsl::invoice_id.eq_any(invoices.iter().map(|i| i.id))),
+            )
+            .execute(tx)?;
+
+            Ok(delete(invoices::dsl::invoices.filter(invoice_clause)).execute(tx)?)
+        })
+    }
+
     fn get_all(&self) -> Result<Vec<HoldInvoice>> {
         let mut con = self.pool.get()?;
 

src/grpc/service.rs

@@ -5,9 +5,9 @@ use crate::grpc::service::hold::hold_server::Hold;
 use crate::grpc::service::hold::invoice_request::Description;
 use crate::grpc::service::hold::list_request::Constraint;
 use crate::grpc::service::hold::{
-    CancelRequest, CancelResponse, GetInfoRequest, GetInfoResponse, InvoiceRequest,
-    InvoiceResponse, ListRequest, ListResponse, SettleRequest, SettleResponse, TrackAllRequest,
-    TrackAllResponse, TrackRequest, TrackResponse,
+    CancelRequest, CancelResponse, CleanRequest, CleanResponse, GetInfoRequest, GetInfoResponse,
+    InvoiceRequest, InvoiceResponse, ListRequest, ListResponse, SettleRequest, SettleResponse,
+    TrackAllRequest, TrackAllResponse, TrackRequest, TrackResponse,
 };
 use crate::grpc::transformers::{transform_invoice_state, transform_route_hints};
 use crate::settler::Settler;
@@ -192,6 +192,22 @@ where
         Ok(Response::new(CancelResponse {}))
     }
 
+    async fn clean(
+        &self,
+        request: Request<CleanRequest>,
+    ) -> Result<Response<CleanResponse>, Status> {
+        let params = request.into_inner();
+        match self.invoice_helper.clean_cancelled(params.age) {
+            Ok(deleted) => Ok(Response::new(CleanResponse {
+                cleaned: deleted as u64,
+            })),
+            Err(err) => Err(Status::new(
+                Code::Internal,
+                format!("could not clean invoices: {}", err),
+            )),
+        }
+    }
+
     type TrackStream = Pin<Box<dyn Stream<Item = Result<TrackResponse, Status>> + Send>>;
 
     async fn track(

src/main.rs

@@ -74,6 +74,11 @@ async fn main() -> Result<()> {
                 .description("Cancels a hold invoice")
                 .usage("payment_hash"),
         )
+        .rpcmethod_from_builder(
+            RpcMethodBuilder::new("cleanholdinvoices", commands::clean)
+                .description("Cleans canceled hold invoices")
+                .usage("[age]"),
+        )
         .configure()
         .await?
     {

tests-regtest/hold/regtest_grpc.py

@@ -7,6 +7,7 @@
 
 from hold.protos.hold_pb2 import (
     CancelRequest,
+    CleanRequest,
     GetInfoRequest,
     GetInfoResponse,
     Hop,
@@ -241,6 +242,32 @@ def test_list_pagination(self, cl: HoldStub) -> None:
         assert len(page.invoices) == 5
         assert page.invoices[0].id == 3
 
+    def test_clean_cancelled(self, cl: HoldStub) -> None:
+        # One that we are not going to cancel which should not be cleaned
+        (_, payment_hash) = new_preimage_bytes()
+        cl.Invoice(InvoiceRequest(payment_hash=payment_hash, amount_msat=1))
+
+        (_, payment_hash) = new_preimage_bytes()
+        invoice: InvoiceResponse = cl.Invoice(
+            InvoiceRequest(payment_hash=payment_hash, amount_msat=1_000)
+        )
+
+        pay = LndPay(1, invoice.bolt11)
+        pay.start()
+        time.sleep(1)
+
+        cl.Cancel(CancelRequest(payment_hash=payment_hash))
+        pay.join()
+
+        res = cl.Clean(CleanRequest(age=0))
+        assert res.cleaned > 0
+
+        res = cl.List(ListRequest(payment_hash=payment_hash))
+        assert len(res.invoices) == 0
+
+        res = cl.List(ListRequest())
+        assert len(res.invoices) > 0
+
     def test_track_settle(self, cl: HoldStub) -> None:
         (preimage, payment_hash) = new_preimage_bytes()
         invoice: InvoiceResponse = cl.Invoice(

tests-regtest/hold/regtest_rpc.py

@@ -121,3 +121,26 @@ def test_cancel(self) -> None:
 
         # Cancelling again should not error
         assert lightning("cancelholdinvoice", payment_hash) == {}
+
+    def test_clean(self) -> None:
+        # One that we are not going to cancel which should not be cleaned
+        (_, payment_hash) = new_preimage()
+        _ = lightning("holdinvoice", payment_hash, "1000")["bolt11"]
+
+        (_, payment_hash) = new_preimage()
+        invoice = lightning("holdinvoice", payment_hash, "1000")["bolt11"]
+
+        payer = LndPay(1, invoice)
+        payer.start()
+        time.sleep(1)
+
+        lightning("cancelholdinvoice", payment_hash)
+
+        cleaned = lightning("cleanholdinvoices")["cleaned"]
+        assert cleaned > 0
+
+        res = lightning("listholdinvoices", payment_hash)["holdinvoices"]
+        assert len(res) == 0
+
+        res = lightning("listholdinvoices")["holdinvoices"]
+        assert len(res) > 0