bark/

mailbox.rs

pub extern crate ark;

pub extern crate bip39;
pub extern crate lightning_invoice;
pub extern crate lnurl as lnurllib;

use std::collections::HashMap;
use std::ops::ControlFlow;
use std::sync::{Arc, Weak};

use anyhow::Context;
use ark::tree::signed::UnlockHash;
use bitcoin::hashes::Hash;
use bitcoin::Amount;
use bitcoin::hex::DisplayHex;
use bitcoin::secp256k1::Keypair;
use futures::{FutureExt, Stream, StreamExt};
use log::{debug, error, info, trace, warn};
use tokio_util::sync::CancellationToken;


use ark::{ProtocolEncoding, Vtxo, VtxoId};
use ark::lightning::{PaymentHash, Preimage};
use ark::mailbox::{MailboxAuthorization, MailboxIdentifier};
use ark::vtxo::Full;
use ark::vtxo::policy::signing::VtxoSigner;
use server_rpc::{protos, MAX_NB_MAILBOX_RECOVERY_IDS};
use server_rpc::protos::mailbox_server::MailboxMessage;

use crate::{Wallet, WalletInner, SUBSCRIBE_REQUEST_TIMEOUT};
use crate::actions::DriveMode;
use crate::actions::lightning::pay::Progress;
use crate::movement::{MovementDestination, MovementStatus};
use crate::movement::update::MovementUpdate;
use crate::subsystem::{ArkoorMovement, Subsystem};
use crate::utils::ReconnectBackoff;


/// The maximum number of times we will call the fetch mailbox endpoint in one go
///
/// We can't trust the server to honestly tell us to keep trying more forever.
/// A malicious server could send us empty messages or invalid messages and
/// lock up our resources forever. So we limit the number of times we will fetch.
/// If a user actually has more messages left, he will have to call sync again.
///
/// (Note that currently the server sends 100 messages per fetch, so this would
/// only happen for users with more than 1000 pending items.)
const MAX_MAILBOX_REQUEST_BURST: usize = 10;

/// Key for the lock that serializes the arkoor receive dedup within a wallet.
///
/// Several consumers of one wallet process the same mailbox messages: the
/// daemon's always-on stream ([`Wallet::subscribe_process_mailbox_messages`])
/// runs alongside the startup / periodic `sync()` (which calls
/// [`Wallet::sync_mailbox`]), and a REST deployment can fire concurrent
/// `/sync` and `/sync/mailbox` requests. The server hands each consumer the
/// same messages from its checkpoint, so without serialization the
/// peek-then-store dedup in arkoor processing races: every consumer that wins
/// the [`crate::persist::BarkPersister::get_wallet_vtxo`] check before the
/// others store records its own receive movement, double-counting the receive.
///
/// Holding this lock across that check-then-store (in
/// [`Wallet::process_received_arkoor_package`]) makes it atomic across
/// consumers. The VTXO row is `INSERT OR IGNORE`, so once one consumer has
/// stored the package the rest see the VTXO already present and skip the
/// movement.
const MAILBOX_PROCESSING_LOCK_KEY: &str = "mailbox.processing";

/// How long a consumer waits for the arkoor dedup lock before giving up.
///
/// The critical section is a couple of server round trips plus local DB
/// writes, normally well under a second. On timeout we error rather than
/// block, handled like any other arkoor processing error: this message's
/// checkpoint isn't advanced and processing halts (see
/// [`Wallet::process_mailbox_message`]) so a later message can't bury it. A
/// timeout only happens if a holder is stuck for the full duration; normally
/// waiters acquire the lock and dedup as usual.
const MAILBOX_PROCESSING_LOCK_TIMEOUT: std::time::Duration =
	std::time::Duration::from_secs(30);

/// Whether a structurally-valid arkoor vtxo is safe to store as a spendable
/// received payment.
///
/// [`Wallet::validate_vtxo`] only proves the vtxo's genesis chain was cosigned
/// by the server pubkey embedded in the vtxo. A received payment additionally
/// has to be a final, self-custodial coin, so we require that:
///
/// - the policy is [`VtxoPolicyKind::Pubkey`]: an in-flight HTLC (or any other
///   server contract) is not a payment, and the server refuses to let us spend
///   it as an arkoor/round/offboard input;
/// - the vtxo commits to *our* server's pubkey, not a foreign server's.
///
/// Expiry is not a reason to refuse. The sender paid with a vtxo that was
/// valid when the server cosigned it; if we only sync after it expired, the
/// payment is still ours and the coin needs a refresh, not a rejection.
/// Refusing would drop the payment, since the mailbox checkpoint advances.
///
/// Ownership (that the wallet holds a signable key) is checked by the caller,
/// which has async database access.
fn check_arkoor_receive_policy(
	vtxo: &Vtxo<Full>,
	server_pubkey: bitcoin::secp256k1::PublicKey,
) -> anyhow::Result<()> {
	let kind = vtxo.policy().policy_type();
	if kind != ark::vtxo::policy::VtxoPolicyKind::Pubkey {
		bail!("not a final payment VTXO (policy: {})", kind);
	}
	if vtxo.server_pubkey() != server_pubkey {
		bail!("VTXO commits to a foreign server pubkey {}", vtxo.server_pubkey());
	}
	Ok(())
}

impl Wallet {
	/// Get the keypair used for the server mailbox
	pub fn mailbox_keypair(&self) -> Keypair {
		self.inner.seed.to_mailbox_keypair()
	}

	/// Get the keypair used for the server recovery mailbox
	pub fn recovery_mailbox_keypair(&self) -> Keypair {
		self.inner.seed.to_recovery_mailbox_keypair()
	}

	/// Get this wallet's server mailbox ID
	pub fn mailbox_identifier(&self) -> MailboxIdentifier {
		let mailbox_kp = self.mailbox_keypair();
		MailboxIdentifier::from_pubkey(mailbox_kp.public_key())
	}

	/// Get this wallet's server recovery mailbox ID
	pub fn recovery_mailbox_identifier(&self) -> MailboxIdentifier {
		let mailbox_kp = self.recovery_mailbox_keypair();
		MailboxIdentifier::from_pubkey(mailbox_kp.public_key())
	}

	/// Create a mailbox authorization that is valid until the given expiry time
	///
	/// This authorization can be used by third parties to lookup your mailbox
	/// with the Ark server.
	pub fn mailbox_authorization(
		&self,
		authorization_expiry: chrono::DateTime<chrono::Local>,
	) -> MailboxAuthorization {
		MailboxAuthorization::new(&self.mailbox_keypair(), authorization_expiry)
	}

	/// Subscribe to mailbox message stream.
	///
	/// If `since` is `None`, the stream will start from the last checkpoint stored in the database.
	///
	/// Returns a stream of mailbox messages.
	pub async fn subscribe_mailbox_messages(
		&self,
		since_checkpoint: Option<u64>,
	) -> anyhow::Result<impl Stream<Item = anyhow::Result<MailboxMessage>> + Unpin + use<>> {
		let (mut srv, _) = self.require_server().await?;

		let checkpoint = if let Some(since) = since_checkpoint {
			since
		} else {
			self.get_mailbox_checkpoint().await?
		};

		// we just need a short authorization for the stream initialization
		let expiry = chrono::Local::now() + std::time::Duration::from_secs(10);
		let auth = self.mailbox_authorization(expiry);
		let mailbox_id = auth.mailbox();

		let mut req = tonic::IntoRequest::into_request(protos::mailbox_server::MailboxRequest {
			mailbox_id: mailbox_id.serialize(),
			authorization: Some(auth.serialize()),
			checkpoint: checkpoint,
		});
		req.set_timeout(SUBSCRIBE_REQUEST_TIMEOUT);
		trace!("Requesting mailbox stream from checkpoint {}", checkpoint);

		let stream = srv.mailbox_client.subscribe_mailbox(req).await?.into_inner().map(|m| {
			let m = m.context("received error on mailbox message stream")?;
			Ok::<_, anyhow::Error>(m)
		});

		Ok(stream)
	}

	/// Similar to [Wallet::subscribe_mailbox_messages] but it will also process each mailbox
	/// message indefinitely. This method won't stop until the given `shutdown` `CancellationToken`
	/// is triggered.
	///
	/// If `since_checkpoint` is `None`, the stream will start from the last checkpoint stored in
	/// the database.
	///
	/// Returns only once the stream is closed.
	pub async fn subscribe_process_mailbox_messages(
		&self,
		since_checkpoint: Option<u64>,
		shutdown: CancellationToken,
	) -> anyhow::Result<()> {
		Wallet::subscribe_process_mailbox_messages_weak(
			Arc::downgrade(&self.inner), since_checkpoint, shutdown,
		).await
	}

	/// The body of [Wallet::subscribe_process_mailbox_messages], holding the
	/// wallet only weakly.
	///
	/// The subscription is always-on, so this future must not own a strong
	/// wallet reference across its lifetime: the daemon relies on the wallet's
	/// strong count reaching zero — running [crate::WalletInner]'s drop glue —
	/// to learn that it should shut down. The weak reference is therefore only
	/// upgraded for one unit of work at a time (a resubscribe or a single
	/// message); between messages this future owns just the gRPC stream.
	///
	/// Returns cleanly when the wallet is dropped, like on shutdown.
	pub(crate) async fn subscribe_process_mailbox_messages_weak(
		wallet: Weak<WalletInner>,
		since_checkpoint: Option<u64>,
		shutdown: CancellationToken,
	) -> anyhow::Result<()> {
		let upgrade = || wallet.upgrade().map(|inner| Wallet { inner });

		// Count consecutive reconnects that made no progress. A delivered
		// message resets it (the connection is healthy); enough failures in a
		// row means the server is unreachable, so we bail and let the daemon
		// mark itself disconnected and back off a full sync interval.
		let mut reconnect_count = 0;
		const MAX_RECONNECT_ATTEMPTS: usize = 5;
		let mut backoff = ReconnectBackoff::new();

		loop {
			let mut stream = {
				let Some(wallet) = upgrade() else { return Ok(()) };
				wallet.subscribe_mailbox_messages(since_checkpoint).await?
			};
			trace!("Connected to mailbox stream with server");

			'stream: loop {
				futures::select! {
					message = stream.next().fuse() => {
						match message {
							Some(Ok(message)) => {
								// A delivered message proves the connection is
								// healthy, so stop counting reconnects as failures.
								reconnect_count = 0;
								let Some(wallet) = upgrade() else { return Ok(()) };
								if wallet.process_mailbox_message(message).await.is_break() {
									// A message failed without advancing its
									// checkpoint. Stop consuming this stream and
									// resubscribe from the unadvanced checkpoint
									// so it's redelivered before a later message
									// can bury it. We deliberately do NOT reset
									// the backoff here: if the same message keeps
									// failing, the growing delay keeps this from
									// becoming a tight redeliver loop.
									trace!("Halting mailbox stream after unadvanced message; resubscribing");
									break 'stream;
								}
								// Progress made: reconnect promptly next time.
								backoff.reset();
							},
							// A tonic h2 stream reset is almost always a
							// proxy- or server-side idle timeout rather than
							// a real failure; resubscribe quietly. Unlike before
							// we count it toward the reconnect cap, so a server
							// that keeps resetting us is treated as unreachable
							// rather than triggering an unbounded resubscribe.
							Some(Err(e)) if crate::utils::is_h2_stream_error(&e) => {
								reconnect_count += 1;
								trace!("Mailbox stream reset by server, reconnecting: {e:#}");
								break 'stream;
							},
							Some(Err(e)) => {
								return Err(e).context("error on mailbox message stream");
							},
							None => {
								reconnect_count += 1;
								warn!("Mailbox stream dropped by server, reconnecting");
								break 'stream;
							},
						}
					},
					_ = shutdown.cancelled().fuse() => {
						info!("Shutdown signal received! Shutting mailbox messages process...");
						return Ok(());
					},
				}
			}

			// Reached only by breaking out of 'stream to resubscribe.
			if reconnect_count >= MAX_RECONNECT_ATTEMPTS {
				bail!("Mailbox stream dropped by server, giving up to retry later");
			}

			// Back off before resubscribing so a fast-closing or rate-limited
			// stream can't become a tight reconnect loop that floods the server
			// with opened-then-reset streams.
			futures::select! {
				_ = backoff.wait().fuse() => {},
				_ = shutdown.cancelled().fuse() => {
					info!("Shutdown signal received! Shutting mailbox messages process...");
					return Ok(());
				},
			}
		}
	}

	/// Sync with the mailbox on the Ark server and look for out-of-round received VTXOs.
	pub async fn sync_mailbox(&self) -> anyhow::Result<()> {
		let (mut srv, _) = self.require_server().await?;

		// we should be able to do all our syncing in 10 minutes
		let expiry = chrono::Local::now() + std::time::Duration::from_secs(10 * 60);
		let auth = self.mailbox_authorization(expiry);
		let mailbox_id = auth.mailbox();

		for _ in 0..MAX_MAILBOX_REQUEST_BURST {
			let checkpoint = self.get_mailbox_checkpoint().await?;
			let mailbox_req = protos::mailbox_server::MailboxRequest {
				mailbox_id: mailbox_id.serialize(),
				authorization: Some(auth.serialize()),
				checkpoint,
			};

			let mailbox_resp = srv.mailbox_client.read_mailbox(mailbox_req).await
				.context("error fetching mailbox")?.into_inner();
			debug!("Ark server has {} mailbox messages for us", mailbox_resp.messages.len());

			for mailbox_msg in mailbox_resp.messages {
				if self.process_mailbox_message(mailbox_msg).await.is_break() {
					// A message failed without advancing its checkpoint. Stop
					// so we don't advance past it; the next sync retries from
					// the same checkpoint.
					return Ok(());
				}
			}

			if !mailbox_resp.have_more {
				break;
			}
		}

		Ok(())
	}

	/// Turn raw byte arrays into VTXOs, then validate them.
	///
	/// This function only returns an error when it cannot process any VTXOs,
	/// for example when the server is down or the chain tip is unknown.
	/// No errors are returned for issues on individual VTXOs.
	async fn process_raw_vtxos(
		&self,
		raw_vtxos: Vec<Vec<u8>>,
	) -> anyhow::Result<Vec<Vtxo<Full>>> {
		let ark_info  = self.require_server().await
			.context("refuse_htlc_send_vtxo_posted_to_arkoor_mailbox")?.1;
		let tip = self.inner.chain.tip().await
			.context("cannot vet received arkoor VTXOs, no chain tip")?;

		let mut invalid_vtxos = Vec::with_capacity(raw_vtxos.len());
		let mut valid_vtxos = Vec::with_capacity(raw_vtxos.len());

		for bytes in &raw_vtxos {
			let vtxo = match Vtxo::<Full>::deserialize(&bytes) {
				Ok(vtxo) => vtxo,
				Err(e) => {
					error!("Failed to deserialize arkoor VTXO: {}: {}", bytes.as_hex(), e);
					invalid_vtxos.push(bytes);
					continue;
				}
			};

			if let Err(e) = self.validate_vtxo(&vtxo).await {
				error!("Received invalid arkoor VTXO {} from server: {}", vtxo.id(), e);
				invalid_vtxos.push(bytes);
				continue;
			}

			if let Err(e) = check_arkoor_receive_policy(&vtxo, ark_info.server_pubkey) {
				error!("Refusing received arkoor VTXO {}: {}", vtxo.id(), e);
				invalid_vtxos.push(bytes);
				continue;
			}
			if self.find_signable_clause(&vtxo).await.is_none() {
				error!("Refusing received arkoor VTXO {}: not owned by this wallet", vtxo.id());
				invalid_vtxos.push(bytes);
				continue;
			}

			if vtxo.expiry_height() <= tip {
				warn!("Received arkoor VTXO {} that expired at height {} (tip {}); \
					keeping it, it needs a refresh", vtxo.id(), vtxo.expiry_height(), tip,
				);
			}

			valid_vtxos.push(vtxo);
		}

		// We log all invalid VTXOs to keep track
		if !invalid_vtxos.is_empty() {
			error!("Received {} invalid arkoor VTXOs out of {} from server",
				invalid_vtxos.len(), raw_vtxos.len(),
			);
		}

		Ok(valid_vtxos)
	}

	/// Process a single mailbox message and report whether the caller should
	/// keep consuming the mailbox.
	///
	/// Returns [`ControlFlow::Break`] when an arkoor package failed to process
	/// and its checkpoint was therefore not advanced. Because checkpoints are
	/// monotonic, the caller must stop before a later message stores a higher
	/// checkpoint and buries the unprocessed one; the next sync/resubscribe
	/// re-fetches from the unadvanced checkpoint and retries.
	pub(crate) async fn process_mailbox_message(
		&self,
		mailbox_msg: MailboxMessage,
	) -> ControlFlow<()> {
		use protos::mailbox_server::mailbox_message::Message;

		// Each arm returns whether the checkpoint should advance. Only
		// arkoor returns false on processing error so the server
		// redelivers and we retry. Every other arm advances regardless,
		// either because the work is idempotent and re-done on every
		// wallet sync, or because the message is informational/ignored.
		let advance = match mailbox_msg.message {
			Some(Message::Arkoor(msg)) => {
				match self.process_received_arkoor_package(msg.vtxos).await {
					Ok(()) => true,
					Err(e) => {
						error!("Error processing received arkoor package: {:#}", e);
						false
					}
				}
			}
			Some(Message::RoundParticipationCompleted(m)) => {
				info!("Server informed that round participation is ready, unlock_hash:{:?}",
					UnlockHash::from_slice(&m.unlock_hash).ok(),
				);
				if let Err(e) = self.sync_pending_rounds().await {
					error!("Error syncing pending rounds: {:#}", e);
				}
				true
			},
			Some(Message::IncomingLightningPayment(msg)) => {
				if let Err(e) = self.handle_lightning_receive_notification(msg).await {
					error!("Error handling lightning receive notification: {:#}", e);
				}
				true
			},
			Some(Message::RecoveryVtxoIds(_)) => {
				trace!("Received recovery VTXO IDs, ignoring");
				true
			}
			Some(Message::LightningSendFinished(msg)) => {
				if let Err(e) = self.handle_lightning_send_finished(msg, mailbox_msg.checkpoint).await {
					error!("Error handling lightning send finished notification: {:#}", e);
				}
				true
			}
			None => {
				warn!("Received unknown mailbox message kind at checkpoint {}; bark may need to be upgraded",
					mailbox_msg.checkpoint);
				true
			}
		};

		if advance {
			if let Err(e) = self.store_mailbox_checkpoint(mailbox_msg.checkpoint).await {
				error!("Error storing mailbox checkpoint: {:#}", e);
			}
			ControlFlow::Continue(())
		} else {
			// An arkoor package didn't process and its checkpoint wasn't
			// advanced. Stop here so a later message can't store a higher
			// checkpoint and bury it.
			ControlFlow::Break(())
		}
	}

	async fn process_received_arkoor_package(
		&self,
		raw_vtxos: Vec<Vec<u8>>,
	) -> anyhow::Result<()> {
		let vtxos = self.process_raw_vtxos(raw_vtxos).await?;

		// Serialize the receive dedup across all consumers of this wallet's
		// mailbox so two of them can't both record a movement for the same
		// package. See MAILBOX_PROCESSING_LOCK_KEY. On lock failure we
		// return like any other arkoor processing error, leaving this
		// message's checkpoint unadvanced.
		let _guard = self.inner.lock_manager.lock(
			MAILBOX_PROCESSING_LOCK_KEY, MAILBOX_PROCESSING_LOCK_TIMEOUT,
		).await.context("failed to acquire mailbox processing lock")?;

		// Everything works off of a deduplicated set. This is important to avoid accounting
		// mistakes in the movement and marking the entire package as spendable, even if it contains
		// VTXOs we already marked as spent.
		let mut new_vtxos = Vec::<Vtxo<Full>>::with_capacity(vtxos.len());
		for vtxo in vtxos {
			// Skip if the VTXO is already in the wallet or is in the VTXO list multiple times.
			let vtxo_id = vtxo.id();
			if new_vtxos.iter().any(|v| v.id() == vtxo_id) {
				debug!("Ignoring duplicate VTXO {} in arkoor package", vtxo_id);
			} else if self.inner.db.get_wallet_vtxo(vtxo_id).await?.is_some() {
				debug!("Ignoring already-known VTXO {} in arkoor package", vtxo_id);
			} else {
				trace!("Received VTXO {} for {} in arkoor package", vtxo_id, vtxo.amount());
				new_vtxos.push(vtxo);
			}
		}
		if new_vtxos.is_empty() {
			return Ok(());
		}

		// Redundantly re-register the received vtxos with the server. An
		// up-to-date sender already does this after cosign, but older
		// senders may not, so we do it on receive too to make sure the
		// server has signed_tx rows for our spendable vtxos. Any failure
		// is logged and swallowed: the receive must still proceed so we
		// don't lose track of the vtxos locally, and later spends will
		// retry registration if still needed.
		if let Err(e) = self.register_vtxo_transactions_with_server(&new_vtxos).await {
			warn!("Failed to register received arkoor vtxo transactions with server: {:#}", e);
		}

		// TODO: Make sure we don't effectively skip the entire package if the movement fails to
		//       be recorded due to bark getting killed after this point.
		let balance = new_vtxos
			.iter()
			.map(|vtxo| vtxo.amount()).sum::<Amount>()
			.to_signed()?;
		self.store_spendable_vtxos(&new_vtxos).await?;

		// Build received_on destinations from received VTXOs, aggregated by address
		let mut received_by_address = HashMap::<ark::Address, Amount>::new();
		for vtxo in &new_vtxos {
			if let Ok(Some((index, _))) = self.pubkey_keypair(&vtxo.user_pubkey()).await {
				if let Ok(address) = self.peek_address(index).await {
					*received_by_address.entry(address).or_default() += vtxo.amount();
				}
			}
		}
		let received_on: Vec<_> = received_by_address
			.iter()
			.map(|(addr, amount)| MovementDestination::ark(addr.clone(), *amount))
			.collect();

		let movement_id = self.inner.movements.new_finished_movement(
			Subsystem::ARKOOR,
			ArkoorMovement::Receive.to_string(),
			MovementStatus::Successful,
			MovementUpdate::new()
				.produced_vtxos(&new_vtxos)
				.intended_and_effective_balance(balance)
				.received_on(received_on),
		).await?;

		info!("Received arkoor (movement {}) for {}", movement_id, balance);

		Ok(())
	}

	/// Handle a lightning receive notification from the mailbox.
	///
	/// This is a signal that the server has received a lightning payment for us
	/// and we should come online to claim it.
	async fn handle_lightning_receive_notification(
		&self,
		notif: protos::mailbox_server::IncomingLightningPaymentMessage,
	) -> anyhow::Result<()> {
		let payment_hash = PaymentHash::try_from(notif.payment_hash)
			.context("invalid payment hash in lightning receive notification")?;

		debug!("Lightning receive notification: payment_hash={}", payment_hash);

		match self.try_claim_lightning_receive(payment_hash, false).await {
			Ok(_) => info!("Lightning receive claimed via mailbox notification for {}", payment_hash),
			Err(e) => error!("Failed to claim lightning receive for {}: {:#}", payment_hash, e),
		}

		Ok(())
	}

	/// Handle a lightning send finished notification from the mailbox.
	///
	/// This notification indicates that the server has completed processing
	/// a lightning payment we initiated, either successfully or with failure.
	async fn handle_lightning_send_finished(
		&self,
		notif: protos::mailbox_server::LightningSendFinishedMessage,
		checkpoint: u64,
	) -> anyhow::Result<()> {
		let payment_hash = PaymentHash::try_from(notif.payment_hash)
			.context("invalid payment hash in lightning send finished notification")?;

		let known_preimage = notif.preimage
			.and_then(|bytes| Preimage::try_from(bytes).ok());

		if known_preimage.is_some() {
			debug!("Lightning send finished notification (success): payment_hash={}", payment_hash);
		} else {
			debug!("Lightning send finished notification (failed): payment_hash={}", payment_hash);
		}

		// Errors are logged but not propagated: we always advance the
		// mailbox checkpoint to avoid re-processing the same
		// notification on the next poll.
		match self.is_invoice_paid(payment_hash).await {
			Ok(true) => {
				debug!("Lightning send {} already settled; ignoring notification", payment_hash);
			},
			Ok(false) => {
				let lookup = self.lightning_send_checkpoint(payment_hash).await;
				match lookup {
					Ok(Some(send)) => {
						let result = match (&send.progress, known_preimage) {
							(Progress::PaymentInitiated(htlcs), Some(preimage)) => {
								let htlcs = htlcs.clone();
								self.settle_lightning_send_with_preimage(send, htlcs, preimage).await
							},
							(Progress::PaymentInitiated(_), None) => {
								self.drive_action(send, DriveMode::UntilParkOrDone).await
							},
							_ => {
								debug!("Lightning send finished notification for {} but checkpoint is not PaymentInitiated; ignoring", payment_hash);
								Ok(())
							},
						};
						match result {
							Ok(()) => info!("Processed lightning send finished for {}", payment_hash),
							Err(e) => error!("Failed to process lightning send finished for {}: {:#}", payment_hash, e),
						}
					},
					Ok(None) => {
						warn!("Lightning send finished notification for unknown payment hash {}", payment_hash);
					},
					Err(e) => {
						error!("Failed to look up lightning send checkpoint for {}: {:#}", payment_hash, e);
					},
				}
			},
			Err(e) => {
				error!("Failed to look up paid_invoice for {}: {:#}", payment_hash, e);
			},
		}

		self.store_mailbox_checkpoint(checkpoint).await?;
		Ok(())
	}

	/// Post vtxo IDs to the server's recovery mailbox
	pub async fn post_recovery_vtxo_ids(
		&self,
		vtxo_ids: impl IntoIterator<Item = VtxoId>,
	) -> anyhow::Result<()> {
		let vtxo_ids = vtxo_ids.into_iter().map(|id| id.to_bytes().to_vec()).collect::<Vec<_>>();
		if vtxo_ids.is_empty() {
			return Ok(());
		}
		let nb_vtxos = vtxo_ids.len();

		// Prove ownership of the recovery mailbox; short validity is enough as
		// it's consumed by this single request.
		let expiry = chrono::Local::now() + std::time::Duration::from_secs(60);
		let auth = MailboxAuthorization::new(&self.recovery_mailbox_keypair(), expiry);
		let mailbox_id = self.recovery_mailbox_identifier().serialize();

		let (mut srv, _) = self.require_server().await?;
		for chunk in vtxo_ids.chunks(MAX_NB_MAILBOX_RECOVERY_IDS) {
			let req = protos::mailbox_server::PostRecoveryVtxoIdsRequest {
				mailbox_id: mailbox_id.clone(),
				vtxo_ids: chunk.to_vec(),
				authorization: Some(auth.serialize()),
			};

			srv.mailbox_client.post_recovery_vtxo_ids(req).await
				.context("error posting recovery vtxo IDs to server")?;
		}

		debug!("Posted {} recovery vtxo IDs to server", nb_vtxos);
		Ok(())
	}

	/// Return the stored mailbox checkpoint — the tip position the wallet
	/// has consumed up to. After a successful [`Self::sync_mailbox`], this value
	/// reflects the server's latest advertised tip.
	pub async fn get_mailbox_checkpoint(&self) -> anyhow::Result<u64> {
		Ok(self.inner.db.get_mailbox_checkpoint().await?)
	}

	async fn store_mailbox_checkpoint(&self, checkpoint: u64) -> anyhow::Result<()> {
		Ok(self.inner.db.store_mailbox_checkpoint(checkpoint).await?)
	}
}

#[cfg(test)]
mod test {
	use super::*;

	use ark::SECP;
	use ark::test_util::VTXO_VECTORS;
	use bitcoin::secp256k1::Keypair;

	/// The arkoor receive path must only store final, self-custodial coins for
	/// our own server. This pins the two ways a received vtxo can be unsafe.
	#[test]
	fn arkoor_receive_policy_gate() {
		let vectors = &*VTXO_VECTORS;
		let server_pubkey = vectors.server_key.public_key();

		// A final Pubkey payment vtxo committing to our server is acceptable.
		let good = &vectors.board_vtxo;
		check_arkoor_receive_policy(good, server_pubkey)
			.expect("a Pubkey vtxo must be accepted");

		// An HTLC-send vtxo is an in-flight contract, not a payment.
		let htlc = &vectors.arkoor_htlc_out_vtxo;
		check_arkoor_receive_policy(htlc, server_pubkey)
			.expect_err("an HTLC-send vtxo must be refused as a payment");

		// A vtxo committing to some other server's key is not ours to trust.
		let foreign_server = Keypair::from_seckey_slice(&SECP, &[0x11; 32])
			.unwrap().public_key();
		check_arkoor_receive_policy(good, foreign_server)
			.expect_err("a vtxo for a foreign server must be refused");
	}
}