implement nip60 wallet operations

crates/nip60/src/keys.rs

@@ -136,6 +136,79 @@ where
         .await?)
 }
 
+pub(crate) async fn build_token_event<S>(
+    signer: &S,
+    me: &PublicKey,
+    token: &TokenEvent,
+) -> Result<Event, Error>
+where
+    S: AsyncGetPublicKey + AsyncSignEvent + AsyncNip44 + ?Sized,
+{
+    let content = signer
+        .nip44_encrypt_async(me, &serde_json::to_string(token)?)
+        .await?;
+
+    Ok(EventBuilder::new(Kind::CashuWalletUnspentProof, content)
+        .finalize_async(signer)
+        .await?)
+}
+
+/// NIP-09 deletion carrying the `k: 7375` tag required by NIP-60.
+pub(crate) async fn build_token_deletion<S>(signer: &S, ids: &[EventId]) -> Result<Event, Error>
+where
+    S: AsyncGetPublicKey + AsyncSignEvent + AsyncNip44 + ?Sized,
+{
+    let mut tags: Vec<Tag> = ids.iter().map(|id| Tag::event(*id)).collect();
+    tags.push(Tag::custom("k", ["7375".to_string()]));
+
+    Ok(EventBuilder::new(Kind::EventDeletion, "")
+        .tags(tags)
+        .finalize_async(signer)
+        .await?)
+}
+
+#[allow(clippy::too_many_arguments)]
+pub(crate) async fn build_history_event<S>(
+    signer: &S,
+    me: &PublicKey,
+    direction: TransactionDirection,
+    amount: u64,
+    created: EventId,
+    destroyed: &[EventId],
+) -> Result<Event, Error>
+where
+    S: AsyncGetPublicKey + AsyncSignEvent + AsyncNip44 + ?Sized,
+{
+    let mut rows = vec![
+        vec!["direction".to_string(), direction.to_string()],
+        vec!["amount".to_string(), amount.to_string()],
+    ];
+
+    for id in destroyed {
+        rows.push(vec![
+            "e".to_string(),
+            id.to_hex(),
+            String::new(),
+            "destroyed".to_string(),
+        ]);
+    }
+
+    rows.push(vec![
+        "e".to_string(),
+        created.to_hex(),
+        String::new(),
+        "created".to_string(),
+    ]);
+
+    let content = signer
+        .nip44_encrypt_async(me, &serde_json::to_string(&rows)?)
+        .await?;
+
+    Ok(EventBuilder::new(Kind::CashuWalletSpendingHistory, content)
+        .finalize_async(signer)
+        .await?)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/nip60/src/lib.rs

@@ -1,18 +1,15 @@
 #![cfg(not(target_arch = "wasm32"))]
 
 mod keys;
+mod ops;
 
 use std::collections::{HashMap, HashSet};
 use std::str::FromStr;
 use std::sync::Arc;
 
 use anyhow::Error;
-use cdk::Amount;
-use cdk::cdk_database::WalletDatabase;
 use cdk::mint_url::MintUrl;
-use cdk::nuts::{CurrencyUnit, Id, Proof, State};
+use cdk::nuts::CurrencyUnit;
-use cdk::secret::Secret;
-use cdk::wallet::types::ProofInfo;
 use cdk::wallet::{Wallet, WalletBuilder};
 use cdk_redb::WalletRedbDatabase;
 use common::config_dir;
@@ -34,22 +31,33 @@ impl Global for GlobalWalletRegistry {}
 #[derive(Debug, Clone)]
 pub enum WalletEvent {
     Updated,
+    TokenCreated(String),
     Error(String),
 }
 
 impl EventEmitter<WalletEvent> for WalletRegistry {}
 
+/// A pending lightning invoice waiting to be paid and claimed.
+#[derive(Debug, Clone)]
+pub struct MintQuoteEntry {
+    pub quote_id: String,
+    pub request: String,
+    pub amount: u64,
+    pub mint: Url,
+}
+
 pub struct WalletRegistry {
     wallet: Option<keys::WalletData>,
     tokens: HashMap<EventId, TokenEvent>,
     /// Token events killed by a kind 5 delete or superseded by a `del` chain
     stale: HashSet<EventId>,
+    quotes: Vec<MintQuoteEntry>,
     store: Option<Arc<WalletRedbDatabase>>,
     wallets: HashMap<MintUrl, Wallet>,
     loading: bool,
     store_opening: bool,
-    importing: bool,
+    job_running: bool,
-    import_queued: bool,
+    job_queued: Vec<ops::Job>,
     notification_listener: Option<Task<()>>,
     tasks: SmallVec<[Task<()>; 2]>,
     _subscriptions: SmallVec<[Subscription; 2]>,
@@ -86,12 +94,13 @@ impl WalletRegistry {
             wallet: None,
             tokens: HashMap::new(),
             stale: HashSet::new(),
+            quotes: Vec::new(),
             store: None,
             wallets: HashMap::new(),
             loading: false,
             store_opening: false,
-            importing: false,
+            job_running: false,
-            import_queued: false,
+            job_queued: Vec::new(),
             notification_listener: None,
             tasks: smallvec![],
             _subscriptions: subscriptions,
@@ -110,6 +119,167 @@ impl WalletRegistry {
             .sum()
     }
 
+    /// Lightning invoices waiting to be paid and claimed.
+    pub fn mint_quotes(&self) -> &[MintQuoteEntry] {
+        &self.quotes
+    }
+
+    /// Create a lightning invoice that funds the wallet once paid.
+    pub fn create_mint_quote(&mut self, amount: u64, cx: &mut Context<Self>) {
+        let Some(mint) = self
+            .wallet
+            .as_ref()
+            .and_then(|wallet| wallet.mints.first().cloned())
+        else {
+            self.emit_error("No wallet is set up".to_string(), cx);
+            return;
+        };
+
+        let Ok(mint_url) = MintUrl::from_str(mint.as_str()) else {
+            self.emit_error("Invalid mint URL".to_string(), cx);
+            return;
+        };
+
+        let Some(wallet) = self.wallets.get(&mint_url).cloned() else {
+            self.emit_error("Wallet is not ready".to_string(), cx);
+            return;
+        };
+
+        let created = cx.background_spawn(async move {
+            let quote = wallet
+                .mint_quote(
+                    cdk::nuts::PaymentMethod::Known(cdk::nuts::nut00::KnownMethod::Bolt11),
+                    Some(cdk::Amount::from(amount)),
+                    None,
+                    None,
+                )
+                .await?;
+
+            Ok::<_, Error>((quote.id, quote.request))
+        });
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            let (quote_id, request) = match created.await {
+                Ok(quote) => quote,
+                Err(error) => {
+                    this.update(cx, |this, cx| {
+                        this.emit_error(format!("{error:#}"), cx);
+                    })
+                    .ok();
+                    return;
+                }
+            };
+
+            this.update(cx, |this, cx| {
+                this.quotes.push(MintQuoteEntry {
+                    quote_id,
+                    request,
+                    amount,
+                    mint: mint.clone(),
+                });
+                this.emit_updated(cx);
+            })
+            .ok();
+        }));
+    }
+
+    /// Check a mint quote with the mint and claim it once it is paid.
+    pub fn claim_mint_quote(&mut self, quote_id: String, cx: &mut Context<Self>) {
+        let Some(entry) = self.quotes.iter().find(|quote| quote.quote_id == quote_id) else {
+            self.emit_error("Unknown mint quote".to_string(), cx);
+            return;
+        };
+
+        let Ok(mint_url) = MintUrl::from_str(entry.mint.as_str()) else {
+            self.emit_error("Invalid mint URL".to_string(), cx);
+            return;
+        };
+
+        self.schedule_job(
+            ops::Job::ClaimQuote {
+                mint: mint_url,
+                quote_id,
+            },
+            cx,
+        );
+    }
+
+    /// Pay a lightning invoice from the wallet.
+    pub fn melt(&mut self, invoice: String, cx: &mut Context<Self>) {
+        let Some(mint) = self
+            .wallet
+            .as_ref()
+            .and_then(|wallet| wallet.mints.first().cloned())
+        else {
+            self.emit_error("No wallet is set up".to_string(), cx);
+            return;
+        };
+
+        let Ok(mint_url) = MintUrl::from_str(mint.as_str()) else {
+            self.emit_error("Invalid mint URL".to_string(), cx);
+            return;
+        };
+
+        self.schedule_job(
+            ops::Job::Melt {
+                mint: mint_url,
+                invoice,
+            },
+            cx,
+        );
+    }
+
+    /// Send ecash as a token string the recipient can paste anywhere.
+    pub fn send_token(&mut self, amount: u64, cx: &mut Context<Self>) {
+        let Some(mint) = self
+            .wallet
+            .as_ref()
+            .and_then(|wallet| wallet.mints.first().cloned())
+        else {
+            self.emit_error("No wallet is set up".to_string(), cx);
+            return;
+        };
+
+        let Ok(mint_url) = MintUrl::from_str(mint.as_str()) else {
+            self.emit_error("Invalid mint URL".to_string(), cx);
+            return;
+        };
+
+        self.schedule_job(
+            ops::Job::SendToken {
+                mint: mint_url,
+                amount,
+            },
+            cx,
+        );
+    }
+
+    /// Receive a cashu token created elsewhere.
+    pub fn receive_token(&mut self, token: String, cx: &mut Context<Self>) {
+        let mint = match cdk::nuts::Token::from_str(&token) {
+            Ok(token) => token.mint_url(),
+            Err(_) => {
+                self.emit_error("Invalid token".to_string(), cx);
+                return;
+            }
+        };
+
+        let mint = match mint {
+            Ok(mint) => mint,
+            Err(_) => {
+                self.emit_error("Token has no mint".to_string(), cx);
+                return;
+            }
+        };
+
+        if !self.wallets.contains_key(&mint) {
+            self.emit_error("Token mint is not in the wallet".to_string(), cx);
+            return;
+        }
+
+        self.schedule_job(ops::Job::Receive { mint, token }, cx);
+    }
+
     /// Create the wallet (first mint) or add a mint to an existing one.
     pub fn setup(&mut self, mint: String, cx: &mut Context<Self>) {
         if self.loading {
@@ -191,12 +361,13 @@ impl WalletRegistry {
         self.wallet = None;
         self.tokens.clear();
         self.stale.clear();
+        self.quotes.clear();
         self.store = None;
         self.wallets.clear();
         self.loading = false;
         self.store_opening = false;
-        self.importing = false;
+        self.job_running = false;
-        self.import_queued = false;
+        self.job_queued.clear();
         self.notification_listener = None;
         self.tasks.clear();
         cx.notify();
@@ -339,7 +510,7 @@ impl WalletRegistry {
                 }
                 this.tokens.insert(event_id, token);
                 this.emit_updated(cx);
-                this.schedule_import(cx);
+                this.schedule_job(ops::Job::Reconcile, cx);
             })
             .ok();
         }));
@@ -363,7 +534,7 @@ impl WalletRegistry {
         }
         if removed_any {
             self.emit_updated(cx);
-            self.schedule_import(cx);
+            self.schedule_job(ops::Job::Reconcile, cx);
         }
     }
 
@@ -416,7 +587,7 @@ impl WalletRegistry {
             }
 
             cx.notify();
-            self.schedule_import(cx);
+            self.schedule_job(ops::Job::Reconcile, cx);
 
             return;
         }
@@ -443,7 +614,7 @@ impl WalletRegistry {
 
                         this.store_opening = false;
                         cx.notify();
-                        this.schedule_import(cx);
+                        this.schedule_job(ops::Job::Reconcile, cx);
                     })
                     .ok();
                 }
@@ -458,119 +629,126 @@ impl WalletRegistry {
         }));
     }
 
-    /// Mirror the active token events into the cdk store.
+    /// Queue a store-touching job.
-    fn schedule_import(&mut self, cx: &mut Context<Self>) {
+    fn schedule_job(&mut self, job: ops::Job, cx: &mut Context<Self>) {
-        if self.importing {
+        if let Some(queued) = self.job_queued.last_mut()
-            self.import_queued = true;
+            && matches!(queued, ops::Job::Reconcile)
+            && matches!(job, ops::Job::Reconcile)
+        {
             return;
         }
 
-        let Some(store) = self.store.clone() else {
+        if self.job_running {
-            return;
+            self.job_queued.push(job);
-        };
-
-        if self.wallets.is_empty() {
             return;
         }
 
-        let mut active: Vec<(MintUrl, Vec<Proof>)> = Vec::new();
+        self.start_job(job, cx);
-        for token in self.tokens.values() {
+    }
-            let Ok(mint_url) = MintUrl::from_str(token.mint.as_str()) else {
-                continue;
-            };
 
-            let mut proofs = Vec::new();
+    fn start_job(&mut self, job: ops::Job, cx: &mut Context<Self>) {
-            for proof in &token.proofs {
+        if self.store.is_none() || self.wallets.is_empty() {
-                let Ok(keyset_id) = Id::from_str(&proof.id) else {
+            self.emit_error("Wallet is not ready".to_string(), cx);
-                    continue;
+            return;
-                };
-
-                let Ok(c) = cdk::nuts::PublicKey::from_str(&proof.c) else {
-                    continue;
-                };
-
-                proofs.push(Proof::new(
-                    Amount::from(proof.amount),
-                    keyset_id,
-                    Secret::new(proof.secret.clone()),
-                    c,
-                ));
-            }
-
-            active.push((mint_url, proofs));
         }
 
-        self.importing = true;
+        let wallets = self.wallets.clone();
+        let tokens = self.tokens.clone();
+
+        let signer = NostrRegistry::global(cx).read(cx).signer().clone();
+        let client = NostrRegistry::global(cx).read(cx).client().clone();
+        self.job_running = true;
 
         self.tasks.push(cx.spawn(async move |this, cx| {
             let result = Tokio::spawn_result(cx, async move {
-                let mut added = Vec::new();
+                let ctx = ops::JobCtx {
-                let mut ys = HashSet::new();
+                    wallets,
+                    signer,
+                    client,
+                    tokens,
+                };
 
-                for (mint_url, proofs) in active {
+                ops::run_job(job, ctx).await
-                    for proof in proofs {
-                        let info = match ProofInfo::new(
-                            proof,
-                            mint_url.clone(),
-                            State::Unspent,
-                            CurrencyUnit::Sat,
-                        ) {
-                            Ok(info) => info,
-                            Err(error) => {
-                                log::warn!("skipping proof: {error:#}");
-                                continue;
-                            }
-                        };
-                        if ys.insert(info.y) {
-                            added.push(info);
-                        }
-                    }
-                }
-
-                let existing = store
-                    .get_proofs(
-                        None,
-                        Some(CurrencyUnit::Sat),
-                        Some(vec![State::Unspent]),
-                        None,
-                    )
-                    .await?;
-
-                let removed = existing
-                    .iter()
-                    .filter(|info| !ys.contains(&info.y))
-                    .map(|info| info.y)
-                    .collect();
-
-                store.update_proofs(added, removed).await?;
-
-                Ok(())
             })
             .await;
 
             this.update(cx, |this, cx| {
-                this.importing = false;
+                this.job_running = false;
+                this.apply_outcome(result, cx);
 
-                if this.import_queued {
+                for job in std::mem::take(&mut this.job_queued) {
-                    this.import_queued = false;
+                    this.schedule_job(job, cx);
-                    this.schedule_import(cx);
-                }
-
-                if let Err(error) = result {
-                    this.emit_error(format!("{error:#}"), cx);
                 }
             })
             .ok();
         }));
     }
 
+    fn apply_outcome(&mut self, result: Result<ops::Outcome, Error>, cx: &mut Context<Self>) {
+        let outcome = match result {
+            Ok(outcome) => outcome,
+            Err(error) => {
+                self.emit_error(format!("{error:#}"), cx);
+                return;
+            }
+        };
+
+        match outcome {
+            ops::Outcome::Idle => {}
+            ops::Outcome::Reconciled { updates } => {
+                let changed = !updates.is_empty();
+
+                for update in updates {
+                    self.apply_token_update(update);
+                }
+
+                if changed {
+                    self.emit_updated(cx);
+                }
+            }
+            ops::Outcome::QuoteMinted {
+                quote_id,
+                update,
+                amount,
+            } => {
+                self.quotes.retain(|quote| quote.quote_id != quote_id);
+                self.apply_token_update(update);
+                self.emit_updated(cx);
+                log::info!("minted {amount} sats from quote {quote_id}");
+            }
+            ops::Outcome::Melted { update, amount } => {
+                self.apply_token_update(update);
+                self.emit_updated(cx);
+                log::info!("melted {amount} sats");
+            }
+            ops::Outcome::Sent { update, token } => {
+                self.apply_token_update(update);
+                cx.emit(WalletEvent::TokenCreated(token));
+                self.emit_updated(cx);
+            }
+            ops::Outcome::Received { update, amount } => {
+                self.apply_token_update(update);
+                self.emit_updated(cx);
+                log::info!("received {amount} sats");
+            }
+        }
+    }
+
+    fn apply_token_update(&mut self, update: ops::TokenUpdate) {
+        for id in &update.destroyed {
+            self.tokens.remove(id);
+            self.stale.insert(*id);
+        }
+        self.tokens.insert(update.event_id, update.token);
+    }
+
     fn apply_loaded(&mut self, state: LoadedState, cx: &mut Context<Self>) {
         self.wallet = state.wallet;
         self.tokens = state.tokens;
         self.stale = state.stale;
         cx.notify();
         self.ensure_wallets(cx);
-        self.schedule_import(cx);
+        self.schedule_job(ops::Job::Reconcile, cx);
     }
 
     fn load(&mut self, cx: &mut Context<Self>) {
@@ -750,18 +928,6 @@ async fn open_store(
         wallets.insert(mint_url, wallet);
     }
 
-    let recovery = wallets.clone();
-    tokio::spawn(async move {
-        for (_, wallet) in recovery {
-            if let Err(error) = wallet.recover_incomplete_sagas().await {
-                log::warn!("saga recovery failed: {error:#}");
-            }
-            if let Err(error) = wallet.mint_unissued_quotes().await {
-                log::warn!("minting unissued quotes failed: {error:#}");
-            }
-        }
-    });
-
     Ok((store, wallets))
 }
 

crates/nip60/src/ops.rs

@@ -0,0 +1,334 @@
+use std::collections::{HashMap, HashSet};
+use std::str::FromStr;
+
+use anyhow::{Error, anyhow};
+use cdk::Amount;
+use cdk::amount::SplitTarget;
+use cdk::mint_url::MintUrl;
+use cdk::nuts::nut00::KnownMethod;
+use cdk::nuts::{CurrencyUnit, Id, MeltQuoteState, MintQuoteState, PaymentMethod, Proof, State};
+use cdk::secret::Secret;
+use cdk::wallet::types::ProofInfo;
+use cdk::wallet::{ReceiveOptions, SendOptions, Wallet};
+use nostr::nips::nip60::{CashuProof, TokenEvent, TransactionDirection};
+use nostr_sdk::prelude::*;
+
+use crate::keys;
+
+/// A job mutating the wallet store or reconciling it with relays.
+#[derive(Debug, Clone)]
+pub(crate) enum Job {
+    Reconcile,
+    ClaimQuote { mint: MintUrl, quote_id: String },
+    Melt { mint: MintUrl, invoice: String },
+    SendToken { mint: MintUrl, amount: u64 },
+    Receive { mint: MintUrl, token: String },
+}
+
+pub(crate) struct TokenUpdate {
+    pub(crate) event_id: EventId,
+    pub(crate) token: TokenEvent,
+    pub(crate) destroyed: Vec<EventId>,
+}
+
+pub(crate) enum Outcome {
+    /// Nothing changed (e.g. an unpaid quote)
+    Idle,
+    /// Token events published by a reconcile pass
+    Reconciled {
+        updates: Vec<TokenUpdate>,
+    },
+    QuoteMinted {
+        quote_id: String,
+        update: TokenUpdate,
+        amount: u64,
+    },
+    Melted {
+        update: TokenUpdate,
+        amount: u64,
+    },
+    Sent {
+        update: TokenUpdate,
+        token: String,
+    },
+    Received {
+        update: TokenUpdate,
+        amount: u64,
+    },
+}
+
+pub(crate) struct JobCtx<S> {
+    pub(crate) wallets: HashMap<MintUrl, Wallet>,
+    pub(crate) signer: S,
+    pub(crate) client: Client,
+    /// Relay token events at schedule time.
+    pub(crate) tokens: HashMap<EventId, TokenEvent>,
+}
+
+pub(crate) async fn run_job<S>(job: Job, ctx: JobCtx<S>) -> Result<Outcome, Error>
+where
+    S: AsyncGetPublicKey + AsyncSignEvent + AsyncNip44 + Send + Sync + 'static,
+{
+    match job {
+        Job::Reconcile => reconcile(ctx).await,
+        Job::ClaimQuote { mint, quote_id } => {
+            let wallet = wallet_for(&ctx, &mint)?;
+            let quote = wallet.check_mint_quote(&quote_id).await?;
+
+            if quote.state != MintQuoteState::Paid {
+                return Ok(Outcome::Idle);
+            }
+
+            let proofs = wallet.mint(&quote_id, SplitTarget::default(), None).await?;
+            let amount: u64 = proofs.iter().map(|proof| u64::from(proof.amount)).sum();
+
+            let update = sync_token_events(
+                &ctx.signer,
+                &ctx.client,
+                wallet,
+                nostr_mint(&mint)?,
+                &previous_ids(&ctx.tokens, &mint),
+                Some((TransactionDirection::In, amount)),
+            )
+            .await?;
+
+            Ok(Outcome::QuoteMinted {
+                quote_id,
+                update,
+                amount,
+            })
+        }
+        Job::Melt { mint, invoice } => {
+            let wallet = wallet_for(&ctx, &mint)?;
+            let quote = wallet
+                .melt_quote(
+                    PaymentMethod::Known(KnownMethod::Bolt11),
+                    invoice,
+                    None,
+                    None,
+                )
+                .await?;
+
+            let finalized = wallet
+                .prepare_melt(&quote.id, HashMap::new())
+                .await?
+                .confirm()
+                .await?;
+
+            if finalized.state() != MeltQuoteState::Paid {
+                return Err(anyhow!("melt did not settle: {:?}", finalized.state()));
+            }
+
+            let amount = u64::from(finalized.total_amount());
+            let update = sync_token_events(
+                &ctx.signer,
+                &ctx.client,
+                wallet,
+                nostr_mint(&mint)?,
+                &previous_ids(&ctx.tokens, &mint),
+                Some((TransactionDirection::Out, amount)),
+            )
+            .await?;
+
+            Ok(Outcome::Melted { update, amount })
+        }
+        Job::SendToken { mint, amount } => {
+            let wallet = wallet_for(&ctx, &mint)?;
+            let prepared = wallet
+                .prepare_send(Amount::from(amount), SendOptions::default())
+                .await?;
+            let token = prepared.confirm(None).await?.to_string();
+
+            let update = sync_token_events(
+                &ctx.signer,
+                &ctx.client,
+                wallet,
+                nostr_mint(&mint)?,
+                &previous_ids(&ctx.tokens, &mint),
+                Some((TransactionDirection::Out, amount)),
+            )
+            .await?;
+
+            Ok(Outcome::Sent { update, token })
+        }
+        Job::Receive { mint, token } => {
+            let wallet = wallet_for(&ctx, &mint)?;
+            let received = wallet.receive(&token, ReceiveOptions::default()).await?;
+
+            let update = sync_token_events(
+                &ctx.signer,
+                &ctx.client,
+                wallet,
+                nostr_mint(&mint)?,
+                &previous_ids(&ctx.tokens, &mint),
+                Some((TransactionDirection::In, u64::from(received))),
+            )
+            .await?;
+
+            Ok(Outcome::Received {
+                update,
+                amount: u64::from(received),
+            })
+        }
+    }
+}
+
+/// Claim paid quotes, then converge each mint.
+async fn reconcile<S>(ctx: JobCtx<S>) -> Result<Outcome, Error>
+where
+    S: AsyncGetPublicKey + AsyncSignEvent + AsyncNip44 + Sync,
+{
+    let mut tokens = ctx.tokens.clone();
+    let mut updates = Vec::new();
+
+    for (mint_url, wallet) in &ctx.wallets {
+        let minted = wallet.mint_unissued_quotes().await?;
+        if minted == Amount::ZERO {
+            continue;
+        }
+
+        let update = sync_token_events(
+            &ctx.signer,
+            &ctx.client,
+            wallet,
+            nostr_mint(mint_url)?,
+            &previous_ids(&tokens, mint_url),
+            Some((TransactionDirection::In, u64::from(minted))),
+        )
+        .await?;
+
+        apply_update(&mut tokens, &update);
+        updates.push(update);
+    }
+
+    for (mint_url, wallet) in &ctx.wallets {
+        let mut relay_ys = HashSet::new();
+        let mut added = Vec::new();
+
+        for token in tokens
+            .values()
+            .filter(|token| token.mint.as_str() == mint_url.to_string())
+        {
+            for proof in &token.proofs {
+                let info = match proof_info(proof, mint_url) {
+                    Ok(info) => info,
+                    Err(error) => {
+                        log::warn!("skipping proof: {error:#}");
+                        continue;
+                    }
+                };
+                if relay_ys.insert(info.y) {
+                    added.push(info);
+                }
+            }
+        }
+
+        let unspent = wallet.get_unspent_proofs().await?;
+        let removed = unspent
+            .iter()
+            .filter(|proof| match proof.y() {
+                Ok(y) => !relay_ys.contains(&y),
+                Err(_) => false,
+            })
+            .map(|proof| proof.y())
+            .collect::<Result<Vec<_>, _>>()?;
+
+        wallet.localstore.update_proofs(added, removed).await?;
+    }
+
+    Ok(Outcome::Reconciled { updates })
+}
+
+/// Publish the store's current unspent set as a fresh token event.
+async fn sync_token_events<S>(
+    signer: &S,
+    client: &Client,
+    wallet: &Wallet,
+    mint: Url,
+    previous: &[EventId],
+    history: Option<(TransactionDirection, u64)>,
+) -> Result<TokenUpdate, Error>
+where
+    S: AsyncGetPublicKey + AsyncSignEvent + AsyncNip44 + Sync + ?Sized,
+{
+    let me = signer.get_public_key_async().await?;
+    let unspent = wallet.get_unspent_proofs().await?;
+
+    let mut proofs = Vec::new();
+    for proof in &unspent {
+        proofs.push(cashu_proof(proof)?);
+    }
+
+    let token = TokenEvent::new(mint, proofs);
+    let event = keys::build_token_event(signer, &me, &token).await?;
+    client.send_event(&event).await?;
+
+    if !previous.is_empty() {
+        let deletion = keys::build_token_deletion(signer, previous).await?;
+        client.send_event(&deletion).await?;
+    }
+
+    if let Some((direction, amount)) = history {
+        let history_event =
+            keys::build_history_event(signer, &me, direction, amount, event.id, previous).await?;
+        client.send_event(&history_event).await?;
+    }
+
+    Ok(TokenUpdate {
+        event_id: event.id,
+        token,
+        destroyed: previous.to_vec(),
+    })
+}
+
+fn apply_update(tokens: &mut HashMap<EventId, TokenEvent>, update: &TokenUpdate) {
+    for id in &update.destroyed {
+        tokens.remove(id);
+    }
+    tokens.insert(update.event_id, update.token.clone());
+}
+
+fn previous_ids(tokens: &HashMap<EventId, TokenEvent>, mint: &MintUrl) -> Vec<EventId> {
+    tokens
+        .iter()
+        .filter(|(_, token)| token.mint.as_str() == mint.to_string())
+        .map(|(id, _)| *id)
+        .collect()
+}
+
+fn wallet_for<'a, S>(ctx: &'a JobCtx<S>, mint: &MintUrl) -> Result<&'a Wallet, Error> {
+    ctx.wallets
+        .get(mint)
+        .ok_or_else(|| anyhow!("no open wallet for mint {mint}"))
+}
+
+fn nostr_mint(mint: &MintUrl) -> Result<Url, Error> {
+    Url::parse(&mint.to_string()).map_err(|error| anyhow!("invalid mint url: {error}"))
+}
+
+fn cashu_proof(proof: &Proof) -> Result<CashuProof, Error> {
+    Ok(CashuProof {
+        id: proof.keyset_id.to_string(),
+        amount: u64::from(proof.amount),
+        secret: proof.secret.to_string(),
+        c: proof.c.to_string(),
+    })
+}
+
+fn proof_info(proof: &CashuProof, mint: &MintUrl) -> Result<ProofInfo, Error> {
+    let keyset_id = Id::from_str(&proof.id)?;
+    let c = cdk::nuts::PublicKey::from_str(&proof.c)?;
+    let info = ProofInfo::new(
+        Proof::new(
+            Amount::from(proof.amount),
+            keyset_id,
+            Secret::new(proof.secret.clone()),
+            c,
+        ),
+        mint.clone(),
+        State::Unspent,
+        CurrencyUnit::Sat,
+    )?;
+
+    Ok(info)
+}

docs/nip60.md

@@ -215,12 +215,36 @@ Deferred to later phases: token rollover publishing (phase 2), remote
 proof validation at the mint (phase 2 reconciliation), replacing proofs
 reserved by local in-flight operations (phase 2 guard).
 
-### Phase 2 — moving money
+### Phase 2 — moving money (done)
 
-1. `ops.rs` sync tail: rollover `7375`, NIP-09 deletes (`k` tag), `7376`
+Implemented in `crates/nip60/src/ops.rs` plus registry wiring:
-   history.
+
-2. `create_mint_quote` (+ claim-on-load), `melt`, `send_token`,
+- All store mutations go through a serialized job queue
-   `receive_token`, each with error propagation to the registry event.
+  (`schedule_job`): one job at a time on the background executor,
+  snapshots taken at schedule time, one queued job slot. This keeps ops
+  and relay-driven imports from interleaving on the store, which a loose
+  sync tail would not guarantee.
+- Jobs: `Reconcile` (claim paid quotes via `mint_unissued_quotes`, then
+  converge each mint — publish the store's unspent set when it diverges
+  from the relay view, and import relay-only proofs into the store),
+  `ClaimQuote`, `Melt`, `SendToken`, `Receive`.
+- The sync tail (`sync_token_events`) publishes a fresh `7375` with the
+  store's unspent proofs, NIP-09-deletes the previous token events with
+  the required `k: 7375` tag, and publishes a `7376` history event with
+  `created`/`destroyed` markers.
+- Registry API: `create_mint_quote(amount)` (invoice stored in
+  `mint_quotes()`, UI polls `claim_mint_quote(id)`), `melt(invoice)`,
+  `send_token(amount)` (token delivered via `WalletEvent::TokenCreated`),
+  `receive_token(token)` (mint resolved from the token).
+- Claim-on-load moved from a fire-and-forget task into `Reconcile`:
+  the fire-and-forget variant raced the import pass, which could wipe
+  freshly claimed proofs from the store before they were published.
+- History amounts: mint/receive record what was received; melt records
+  invoice amount + fee (`FinalizedMelt::total_amount`); send records the
+  sent amount.
+
+Deferred: NUT-20/quote events (`7374`, per spec local state is fine),
+sending tokens in-chat (v1 copies the token string).
 
 ### Phase 3 — NIP-61 redemption (cdk-nostr)
 
@@ -254,8 +278,9 @@ reserved by local in-flight operations (phase 2 guard).
 - Concurrent spends from two devices double-spend; recovery is the spec's
   Appendix-1 validation (check proofs at the mint, roll over survivors) —
   implemented as reconciliation on sync, not real-time locking.
-- Import replaces the unspent proof set wholesale; a guard for proofs
+- Import replaces the unspent proof set wholesale; the serialized job
-  reserved by local in-flight operations lands with phase 2 operations.
+  queue closes the in-flight-operation window, and mint-side double-spend
+  rejection remains the backstop.
 - Tracking cdk via git main is a moving target; the lockfile pins a rev
   and must be bumped deliberately (same policy as the rust-nostr git dep).
   Watch for the next cdk release line built against nostr 0.45.x — moving