improve handling of stale or duplicate events

src/main.rs

@@ -157,6 +157,16 @@ async fn main() -> Result<(), anyhow::Error> {
         }
     }
 
+    let plugin_clone_cleanup = plugin.clone();
+    tokio::spawn(async move {
+        loop {
+            if let Err(e) = tasks::cleanup_event_ids(plugin_clone_cleanup.clone()).await {
+                log::warn!("Error in cleanup_event_ids thread: {e}");
+            }
+            time::sleep(Duration::from_secs(10)).await;
+        }
+    });
+
     plugin.join().await
 }
 

src/nwc.rs

@@ -31,9 +31,9 @@ use crate::{
     nwc_keysend::pay_keysend_response,
     nwc_lookups::{list_transactions_response, lookup_invoice_response},
     nwc_pay::pay_invoice_response,
-    structs::{NwcStore, PluginState, WalletService},
+    structs::{ID_MAX_AGE, NwcStore, PluginState, WalletService},
     tasks::budget_task,
-    util::{build_capabilities, build_notifications_vec, is_read_only_nwc},
+    util::{build_capabilities, build_notifications_vec, is_read_only_nwc, save_event_id},
 };
 
 #[allow(clippy::too_many_lines)]
@@ -225,6 +225,7 @@ pub fn stop_nwc_budget_job(plugin: &Plugin<PluginState>, label: &String) {
     }
 }
 
+#[allow(clippy::too_many_lines)]
 async fn nwc_request_handler(
     notification: ClientNotification,
     nostr_client: &client::Client,
@@ -246,44 +247,73 @@ async fn nwc_request_handler(
         | ClientNotification::Shutdown => return Ok(()),
     };
 
-    if let Some(expi) = event.tags.expiration() {
-        if expi < Timestamp::now() {
-            return Ok(());
-        }
-    }
     log::debug!("relay_url:{relay_url} subscription_id:{subscription_id} {event:?}");
     let mut use_nip44 = check_nip44_support(&event);
 
     let request = decrypt_request(&event.content, wallet_keys, &client_pubkey, &mut use_nip44)?;
 
-    let responses = match request.params {
-        nip47::RequestParams::PayInvoice(pay_invoice_request) => {
-            pay_invoice_response(plugin.clone(), pay_invoice_request, label).await
+    let responses = if event.tags.expiration().is_some()
+        && event.tags.expiration().unwrap() < Timestamp::now()
+    {
+        vec![(
+            nip47::Response {
+                result_type: request.method,
+                error: Some(nip47::NIP47Error {
+                    code: nip47::ErrorCode::Other,
+                    message: "Event expired".to_owned(),
+                }),
+                result: None,
+            },
+            None,
+        )]
+    } else if event.created_at.as_secs() < (Timestamp::now().as_secs() - ID_MAX_AGE) {
+        vec![(
+            nip47::Response {
+                result_type: request.method,
+                error: Some(nip47::NIP47Error {
+                    code: nip47::ErrorCode::Other,
+                    message: "Event created too far in the past".to_owned(),
+                }),
+                result: None,
+            },
+            None,
+        )]
+    } else {
+        {
+            let mut rpc = plugin.state().rpc_lock.lock().await;
+            save_event_id(&mut rpc, event.id.to_hex(), event.created_at).await?;
         }
-        nip47::RequestParams::PayKeysend(pay_keysend_request) => {
-            pay_keysend_response(plugin.clone(), pay_keysend_request, label).await
-        }
-        nip47::RequestParams::MakeInvoice(make_invoice_request) => {
-            make_invoice_response(plugin.clone(), make_invoice_request).await
-        }
-        nip47::RequestParams::LookupInvoice(lookup_invoice_request) => {
-            lookup_invoice_response(plugin.clone(), lookup_invoice_request).await
-        }
-        nip47::RequestParams::ListTransactions(list_transactions_request) => {
-            list_transactions_response(plugin.clone(), list_transactions_request).await
-        }
-        nip47::RequestParams::GetBalance => get_balance_response(plugin.clone(), label).await,
-        nip47::RequestParams::GetInfo => get_info_response(plugin.clone(), label).await,
-        nip47::RequestParams::MakeHoldInvoice(make_hold_invoice_request) => {
-            make_hold_invoice_response(plugin.clone(), make_hold_invoice_request).await
-        }
-        nip47::RequestParams::CancelHoldInvoice(cancel_hold_invoice_request) => {
-            cancel_hold_invoice_response(plugin.clone(), cancel_hold_invoice_request).await
-        }
-        nip47::RequestParams::SettleHoldInvoice(settle_hold_invoice_request) => {
-            settle_hold_invoice_response(plugin.clone(), settle_hold_invoice_request).await
+
+        match request.params {
+            nip47::RequestParams::PayInvoice(pay_invoice_request) => {
+                pay_invoice_response(plugin.clone(), pay_invoice_request, label).await
+            }
+            nip47::RequestParams::PayKeysend(pay_keysend_request) => {
+                pay_keysend_response(plugin.clone(), pay_keysend_request, label).await
+            }
+            nip47::RequestParams::MakeInvoice(make_invoice_request) => {
+                make_invoice_response(plugin.clone(), make_invoice_request).await
+            }
+            nip47::RequestParams::LookupInvoice(lookup_invoice_request) => {
+                lookup_invoice_response(plugin.clone(), lookup_invoice_request).await
+            }
+            nip47::RequestParams::ListTransactions(list_transactions_request) => {
+                list_transactions_response(plugin.clone(), list_transactions_request).await
+            }
+            nip47::RequestParams::GetBalance => get_balance_response(plugin.clone(), label).await,
+            nip47::RequestParams::GetInfo => get_info_response(plugin.clone(), label).await,
+            nip47::RequestParams::MakeHoldInvoice(make_hold_invoice_request) => {
+                make_hold_invoice_response(plugin.clone(), make_hold_invoice_request).await
+            }
+            nip47::RequestParams::CancelHoldInvoice(cancel_hold_invoice_request) => {
+                cancel_hold_invoice_response(plugin.clone(), cancel_hold_invoice_request).await
+            }
+            nip47::RequestParams::SettleHoldInvoice(settle_hold_invoice_request) => {
+                settle_hold_invoice_response(plugin.clone(), settle_hold_invoice_request).await
+            }
         }
     };
+
     for (response, id) in responses {
         let content =
             match encrypt_response_content(&response, wallet_keys, &client_pubkey, use_nip44) {

src/structs.rs

@@ -15,6 +15,8 @@ use tonic::transport::Channel;
 use crate::hold::hold_client::HoldClient;
 
 pub const NOT_INV_ERR: &str = "Not an invoice or invalid invoice";
+pub const ID_STORE: &str = "eventids";
+pub const ID_MAX_AGE: u64 = 7_200;
 
 #[derive(Clone)]
 pub struct PluginState {

src/tasks.rs

@@ -1,13 +1,17 @@
-use std::{path::Path, time::Duration};
+use std::{path::Path, str::FromStr, time::Duration};
 
 use anyhow::anyhow;
 use cln_plugin::Plugin;
-use cln_rpc::ClnRpc;
+use cln_rpc::{
+    ClnRpc,
+    model::requests::{DeldatastoreRequest, ListdatastoreRequest},
+};
 use nostr::types::Timestamp;
 use tokio::{sync::oneshot, time};
 
 use crate::{
-    structs::PluginState,
+    PLUGIN_NAME,
+    structs::{ID_MAX_AGE, ID_STORE, PluginState},
     util::{load_nwc_store, update_nwc_store},
 };
 
@@ -57,3 +61,39 @@ pub async fn budget_task(
     }
     Ok(())
 }
+
+pub async fn cleanup_event_ids(plugin: Plugin<PluginState>) -> Result<(), anyhow::Error> {
+    let mut rpc = ClnRpc::new(
+        Path::new(&plugin.configuration().lightning_dir).join(&plugin.configuration().rpc_file),
+    )
+    .await?;
+
+    loop {
+        {
+            let ids = rpc
+                .call_typed(&ListdatastoreRequest {
+                    key: Some(vec![format!("{}-{}", PLUGIN_NAME, ID_STORE)]),
+                })
+                .await?
+                .datastore;
+
+            let now = Timestamp::now();
+            for id in ids {
+                if id.string.is_none() {
+                    continue;
+                }
+                let timestamp = Timestamp::from_str(&id.string.unwrap())?;
+                if now.as_secs() - timestamp.as_secs() > ID_MAX_AGE {
+                    rpc.call_typed(&DeldatastoreRequest {
+                        generation: None,
+                        key: id.key.clone(),
+                    })
+                    .await?;
+                    log::debug!("Cleaned up event id: {}", id.key.last().unwrap());
+                }
+            }
+        }
+
+        tokio::time::sleep(Duration::from_secs(120)).await;
+    }
+}

src/util.rs

@@ -1,13 +1,12 @@
 use anyhow::anyhow;
 use cln_plugin::Plugin;
 use cln_rpc::{
-    model::requests::{DatastoreMode, DatastoreRequest, ListdatastoreRequest},
     ClnRpc,
+    model::requests::{DatastoreMode, DatastoreRequest, ListdatastoreRequest},
 };
-use nostr::nips::nip47;
+use nostr::{nips::nip47, types::Timestamp};
 
 use crate::{
-    structs::{NwcStore, PluginState},
     OPT_NOTIFICATIONS,
     PLUGIN_NAME,
     WALLET_HOLD_METHODS,
@@ -15,6 +14,7 @@ use crate::{
     WALLET_NOTIFICATIONS,
     WALLET_PAY_METHODS,
     WALLET_READ_METHODS,
+    structs::{ID_STORE, NwcStore, PluginState},
 };
 
 pub fn budget_amount_check(
@@ -97,6 +97,23 @@ pub fn is_read_only_nwc(nwc_store: &NwcStore) -> bool {
     false
 }
 
+pub async fn save_event_id(
+    rpc: &mut ClnRpc,
+    id: String,
+    timestamp: Timestamp,
+) -> Result<(), anyhow::Error> {
+    rpc.call_typed(&DatastoreRequest {
+        key: vec![format!("{}-{}", PLUGIN_NAME, ID_STORE), id.clone()],
+        generation: None,
+        hex: None,
+        mode: Some(DatastoreMode::MUST_CREATE),
+        string: Some(timestamp.to_string()),
+    })
+    .await?;
+    log::debug!("stored event id:{id}");
+    Ok(())
+}
+
 pub fn at_or_above_version(my_version: &str, min_version: &str) -> Result<bool, anyhow::Error> {
     let clean_start_my_version = my_version
         .split_once('v')

tests/test_cln-nip47.py

@@ -26,7 +26,6 @@ from nostr_sdk import (
     MakeInvoiceRequest,
     Method,
     NostrSdkError,
-    NostrSigner,
     NostrWalletConnect,
     NostrWalletConnectUri,
     PayInvoiceRequest,