Add PaymentMetadataStore wrapper and persistence

Add the `PaymentMetadataStore` struct that wraps a `DataStore` with a
reverse index from `PaymentId` to `MetadataId`. This enables efficient
lookup of metadata entries by payment ID.

API methods:
- `insert` / `get` / `remove` — basic CRUD with reverse index upkeep
- `add_payment_id` — associate additional payment IDs with an entry
- `get_for_payment_id` — reverse index lookup
- `get_lsp_fee_limits_for_payment_id` — convenience for LSP fee checks
- `remove_payment_id` — clean up reverse index when payments are removed

Also add the `payment_metadata` namespace constants in `io/mod.rs`.

Generated with the help of AI (Claude Code).

Co-Authored-By: HAL 9000

Author: tnull

src/io/mod.rs

@@ -82,3 +82,7 @@ pub(crate) const STATIC_INVOICE_STORE_PRIMARY_NAMESPACE: &str = "static_invoices
 /// The pending payment information will be persisted under this prefix.
 pub(crate) const PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE: &str = "pending_payments";
 pub(crate) const PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
+
+/// The payment metadata will be persisted under this prefix.
+pub(crate) const PAYMENT_METADATA_PERSISTENCE_PRIMARY_NAMESPACE: &str = "payment_metadata";
+pub(crate) const PAYMENT_METADATA_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";

src/payment/metadata_store.rs

@@ -5,16 +5,26 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
+use std::collections::{HashMap, HashSet};
+use std::sync::{Arc, Mutex};
+
 use lightning::impl_writeable_tlv_based;
 use lightning::impl_writeable_tlv_based_enum;
 use lightning::ln::channelmanager::PaymentId;
 use lightning::offers::offer::Offer as LdkOffer;
 use lightning::offers::refund::Refund as LdkRefund;
 use lightning_invoice::Bolt11Invoice as LdkBolt11Invoice;
 
-use crate::data_store::{StorableObject, StorableObjectId, StorableObjectUpdate};
+use crate::data_store::{DataStore, StorableObject, StorableObjectId, StorableObjectUpdate};
 use crate::hex_utils;
+use crate::io::{
+	PAYMENT_METADATA_PERSISTENCE_PRIMARY_NAMESPACE,
+	PAYMENT_METADATA_PERSISTENCE_SECONDARY_NAMESPACE,
+};
+use crate::logger::{log_error, LdkLogger, Logger};
 use crate::payment::store::LSPFeeLimits;
+use crate::types::DynStore;
+use crate::Error;
 
 /// An opaque identifier for a payment metadata entry.
 #[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)]
@@ -113,3 +123,146 @@ impl StorableObjectUpdate<PaymentMetadataEntry> for PaymentMetadataEntryUpdate {
 		self.id
 	}
 }
+
+/// A store for payment metadata, backed by a [`DataStore`].
+///
+/// Maintains a reverse index from [`PaymentId`] to [`MetadataId`] for efficient lookups.
+pub(crate) struct PaymentMetadataStore {
+	inner: DataStore<PaymentMetadataEntry, Arc<Logger>>,
+	reverse_index: Mutex<HashMap<PaymentId, HashSet<MetadataId>>>,
+}
+
+impl PaymentMetadataStore {
+	pub(crate) fn new(
+		entries: Vec<PaymentMetadataEntry>, kv_store: Arc<DynStore>, logger: Arc<Logger>,
+	) -> Self {
+		let mut reverse_index: HashMap<PaymentId, HashSet<MetadataId>> = HashMap::new();
+		for entry in &entries {
+			for payment_id in &entry.payment_ids {
+				reverse_index.entry(*payment_id).or_default().insert(entry.id);
+			}
+		}
+
+		let inner = DataStore::new(
+			entries,
+			PAYMENT_METADATA_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
+			PAYMENT_METADATA_PERSISTENCE_SECONDARY_NAMESPACE.to_string(),
+			kv_store,
+			logger,
+		);
+
+		Self { inner, reverse_index: Mutex::new(reverse_index) }
+	}
+
+	/// Insert a new metadata entry and update the reverse index.
+	pub(crate) fn insert(&self, entry: PaymentMetadataEntry) -> Result<MetadataId, Error> {
+		let id = entry.id;
+		let payment_ids = entry.payment_ids.clone();
+
+		self.inner.insert(entry)?;
+
+		let mut locked_index = self.reverse_index.lock().unwrap();
+		for payment_id in payment_ids {
+			locked_index.entry(payment_id).or_default().insert(id);
+		}
+
+		Ok(id)
+	}
+
+	/// Associate an additional [`PaymentId`] with an existing metadata entry.
+	pub(crate) fn add_payment_id(
+		&self, metadata_id: MetadataId, payment_id: PaymentId,
+	) -> Result<(), Error> {
+		if let Some(mut entry) = self.inner.get(&metadata_id) {
+			if !entry.payment_ids.contains(&payment_id) {
+				entry.payment_ids.push(payment_id);
+				let update = PaymentMetadataEntryUpdate {
+					id: metadata_id,
+					payment_ids: Some(entry.payment_ids),
+				};
+				self.inner.update(update)?;
+				self.reverse_index
+					.lock()
+					.unwrap()
+					.entry(payment_id)
+					.or_default()
+					.insert(metadata_id);
+			}
+			Ok(())
+		} else {
+			Err(Error::PersistenceFailed)
+		}
+	}
+
+	/// Get a metadata entry by its ID.
+	pub(crate) fn get(&self, metadata_id: &MetadataId) -> Option<PaymentMetadataEntry> {
+		self.inner.get(metadata_id)
+	}
+
+	/// Get all metadata entries associated with a given payment ID.
+	pub(crate) fn get_for_payment_id(&self, payment_id: &PaymentId) -> Vec<PaymentMetadataEntry> {
+		let locked_index = self.reverse_index.lock().unwrap();
+		if let Some(metadata_ids) = locked_index.get(payment_id) {
+			metadata_ids.iter().filter_map(|mid| self.inner.get(mid)).collect()
+		} else {
+			Vec::new()
+		}
+	}
+
+	/// Convenience method to get the [`LSPFeeLimits`] for a given payment ID, if any.
+	pub(crate) fn get_lsp_fee_limits_for_payment_id(
+		&self, payment_id: &PaymentId,
+	) -> Option<LSPFeeLimits> {
+		let entries = self.get_for_payment_id(payment_id);
+		for entry in entries {
+			if let PaymentMetadataKind::LSPFeeLimits { limits } = entry.kind {
+				return Some(limits);
+			}
+		}
+		None
+	}
+
+	/// Remove a metadata entry and clean up the reverse index.
+	pub(crate) fn remove(&self, metadata_id: &MetadataId) -> Result<(), Error> {
+		if let Some(entry) = self.inner.get(metadata_id) {
+			let mut locked_index = self.reverse_index.lock().unwrap();
+			for payment_id in &entry.payment_ids {
+				if let Some(set) = locked_index.get_mut(payment_id) {
+					set.remove(metadata_id);
+					if set.is_empty() {
+						locked_index.remove(payment_id);
+					}
+				}
+			}
+		}
+		self.inner.remove(metadata_id)
+	}
+
+	/// Remove a [`PaymentId`] from all associated metadata entries.
+	///
+	/// This should be called when a payment store entry is removed to keep the reverse index
+	/// consistent. If a metadata entry's `payment_ids` becomes empty after removal, it is
+	/// **not** automatically deleted (the metadata may still be useful).
+	pub(crate) fn remove_payment_id(&self, payment_id: &PaymentId) -> Result<(), Error> {
+		let metadata_ids = {
+			let mut locked_index = self.reverse_index.lock().unwrap();
+			match locked_index.remove(payment_id) {
+				Some(ids) => ids,
+				None => return Ok(()),
+			}
+		};
+
+		for metadata_id in metadata_ids {
+			if let Some(mut entry) = self.inner.get(&metadata_id) {
+				entry.payment_ids.retain(|id| id != payment_id);
+				let update = PaymentMetadataEntryUpdate {
+					id: metadata_id,
+					payment_ids: Some(entry.payment_ids),
+				};
+				self.inner.update(update)?;
+			}
+		}
+
+		Ok(())
+	}
+}