implement nip60 wallet registry

Cargo.lock

@@ -6131,12 +6131,23 @@ dependencies = [
 name = "nip60"
 version = "2.0.0-alpha3"
 dependencies = [
+ "anyhow",
  "cdk",
  "cdk-nostr",
  "cdk-redb",
+ "common",
+ "flume 0.11.1",
  "gpui-pre",
+ "gpui_tokio",
+ "hkdf",
+ "log",
  "nostr",
  "nostr-sdk",
+ "serde_json",
+ "sha2 0.10.9",
+ "smallvec",
+ "state",
+ "tokio",
 ]
 
 [[package]]

crates/nip60/Cargo.toml

@@ -5,9 +5,21 @@ edition.workspace = true
 publish.workspace = true
 
 [dependencies]
+state = { path = "../state" }
+common = { path = "../common" }
+
 gpui.workspace = true
+gpui_tokio.workspace = true
 nostr.workspace = true
 nostr-sdk.workspace = true
+anyhow.workspace = true
+smallvec.workspace = true
+log.workspace = true
+flume.workspace = true
+serde_json.workspace = true
+hkdf.workspace = true
+sha2.workspace = true
+tokio = { version = "1", default-features = false, features = [ "rt" ] }
 
 cdk.workspace = true
 cdk-redb.workspace = true

crates/nip60/src/keys.rs

@@ -0,0 +1,178 @@
+use anyhow::{Error, anyhow};
+use hkdf::Hkdf;
+use nostr::nips::nip60::TokenEvent;
+use nostr_sdk::prelude::*;
+use sha2::Sha256;
+
+const SEED_SALT: &[u8] = b"coop nip60 seed salt";
+const SEED_INFO: &[u8] = b"coop cashu seed";
+
+pub(crate) struct WalletData {
+    pub(crate) event_id: EventId,
+    pub(crate) created_at: Timestamp,
+    pub(crate) privkey: SecretKey,
+    pub(crate) mints: Vec<Url>,
+}
+
+impl Clone for WalletData {
+    fn clone(&self) -> Self {
+        Self {
+            event_id: self.event_id,
+            created_at: self.created_at,
+            privkey: self.privkey.clone(),
+            mints: self.mints.clone(),
+        }
+    }
+}
+
+pub(crate) fn generate_privkey() -> SecretKey {
+    SecretKey::generate()
+}
+
+/// Derive the cdk seed from the wallet key.
+pub(crate) fn derive_seed(privkey: &SecretKey) -> Result<[u8; 64], Error> {
+    let ikm = privkey.to_secret_bytes();
+    let hk = Hkdf::<Sha256>::new(Some(SEED_SALT), &ikm);
+    let mut seed = [0u8; 64];
+    hk.expand(SEED_INFO, &mut seed)
+        .map_err(|error| anyhow!("seed derivation failed: {error}"))?;
+    Ok(seed)
+}
+
+pub(crate) fn parse_wallet_content(content: &str) -> Result<(SecretKey, Vec<Url>), Error> {
+    let rows: Vec<Vec<String>> =
+        serde_json::from_str(content).map_err(|error| anyhow!("invalid wallet event: {error}"))?;
+
+    let mut privkey = None;
+    let mut mints = Vec::new();
+    for row in rows {
+        let (Some(key), Some(value)) = (row.first(), row.get(1)) else {
+            continue;
+        };
+        match key.as_str() {
+            "privkey" if privkey.is_none() => privkey = Some(SecretKey::from_hex(value)?),
+            "mint" => {
+                if let Ok(mint) = Url::parse(value) {
+                    mints.push(mint);
+                }
+            }
+            _ => {}
+        }
+    }
+
+    let Some(privkey) = privkey else {
+        return Err(anyhow!("wallet event is missing the private key"));
+    };
+    if mints.is_empty() {
+        return Err(anyhow!("wallet event is missing mints"));
+    }
+
+    Ok((privkey, mints))
+}
+
+pub(crate) async fn decrypt_wallet_event<S>(
+    signer: &S,
+    me: &PublicKey,
+    event: &Event,
+) -> Result<WalletData, Error>
+where
+    S: AsyncNip44 + ?Sized,
+{
+    if event.pubkey != *me {
+        return Err(anyhow!("wallet event was not published by this account"));
+    }
+
+    let content = signer.nip44_decrypt_async(me, &event.content).await?;
+    let (privkey, mints) = parse_wallet_content(&content)?;
+
+    Ok(WalletData {
+        event_id: event.id,
+        created_at: event.created_at,
+        privkey,
+        mints,
+    })
+}
+
+pub(crate) async fn decrypt_token_event<S>(
+    signer: &S,
+    me: &PublicKey,
+    event: &Event,
+) -> Result<TokenEvent, Error>
+where
+    S: AsyncNip44 + ?Sized,
+{
+    if event.pubkey != *me {
+        return Err(anyhow!("token event was not published by this account"));
+    }
+
+    let content = signer.nip44_decrypt_async(me, &event.content).await?;
+
+    let token =
+        serde_json::from_str(&content).map_err(|error| anyhow!("invalid token event: {error}"))?;
+
+    Ok(token)
+}
+
+pub(crate) async fn build_wallet_event<S>(
+    signer: &S,
+    me: &PublicKey,
+    privkey: &SecretKey,
+    mints: &[Url],
+) -> Result<Event, Error>
+where
+    S: AsyncGetPublicKey + AsyncSignEvent + AsyncNip44 + ?Sized,
+{
+    let mut rows = vec![vec!["privkey".to_string(), privkey.to_secret_hex()]];
+    for mint in mints {
+        rows.push(vec!["mint".to_string(), mint.to_string()]);
+    }
+
+    let content = signer
+        .nip44_encrypt_async(me, &serde_json::to_string(&rows)?)
+        .await?;
+
+    Ok(EventBuilder::new(Kind::CashuWallet, content)
+        .finalize_async(signer)
+        .await?)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_wallet_content() {
+        let privkey = SecretKey::generate();
+        let mint = Url::parse("https://mint.example.com").expect("valid mint URL");
+
+        let content = serde_json::to_string(&vec![
+            vec!["privkey".to_string(), privkey.to_secret_hex()],
+            vec!["mint".to_string(), mint.to_string()],
+        ])
+        .expect("serialization cannot fail");
+
+        let (parsed_key, mints) = parse_wallet_content(&content).expect("valid content");
+        assert_eq!(parsed_key.to_secret_hex(), privkey.to_secret_hex());
+        assert_eq!(mints, vec![mint]);
+    }
+
+    #[test]
+    fn wallet_content_rejects_missing_privkey() {
+        let mint = Url::parse("https://mint.example.com").expect("valid mint URL");
+        let content = serde_json::to_string(&vec![vec!["mint".to_string(), mint.to_string()]])
+            .expect("serialization cannot fail");
+
+        assert!(parse_wallet_content(&content).is_err());
+    }
+
+    #[test]
+    fn seed_is_deterministic_per_key() {
+        let key = SecretKey::generate();
+        let first = derive_seed(&key).expect("seed derivation cannot fail");
+        let second = derive_seed(&key).expect("seed derivation cannot fail");
+        assert_eq!(first, second);
+
+        let other = derive_seed(&SecretKey::generate()).expect("seed derivation cannot fail");
+        assert_ne!(first, other);
+    }
+}

crates/nip60/src/lib.rs

@@ -1,49 +1,783 @@
 #![cfg(not(target_arch = "wasm32"))]
 
-use gpui::App;
-use nostr::prelude::*;
+mod keys;
 
-/// Initialize the wallet registry.
-pub fn init(_cx: &mut App) {}
+use std::collections::{HashMap, HashSet};
+use std::str::FromStr;
+use std::sync::Arc;
 
-/// Compile-only phase 0 spike: proves the `nostr::nips::nip60` types, kinds,
-/// and the cdk crates resolve against the workspace's git nostr. Replaced by
-/// the real implementation.
-#[allow(dead_code)]
-fn spike() {
-    let _kinds = (
-        Kind::CashuWallet,
-        Kind::CashuWalletUnspentProof,
-        Kind::CashuWalletSpendingHistory,
-        Kind::CashuWalletQuote,
-    );
+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::secret::Secret;
+use cdk::wallet::types::ProofInfo;
+use cdk::wallet::{Wallet, WalletBuilder};
+use cdk_redb::WalletRedbDatabase;
+use common::config_dir;
+use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task};
+use gpui_tokio::Tokio;
+use nostr::nips::nip60::TokenEvent;
+use nostr_sdk::prelude::*;
+use smallvec::{SmallVec, smallvec};
+use state::NostrRegistry;
 
-    let _history =
-        nostr::nips::nip60::SpendingHistory::new(nostr::nips::nip60::TransactionDirection::In, 1);
-    let _builder = cdk::wallet::WalletBuilder::new();
-    let _store: Option<cdk_redb::WalletRedbDatabase> = None;
-    let _inbox: Option<cdk_nostr::inbox::NostrInbox> = None;
+pub fn init(cx: &mut App) {
+    WalletRegistry::set_global(cx.new(WalletRegistry::new), cx);
+}
 
-    if let Ok(mint) = nostr::types::Url::parse("https://mint.example.com") {
-        let proof = nostr::nips::nip60::CashuProof {
-            id: String::new(),
-            amount: 1,
-            secret: String::new(),
-            c: String::new(),
+struct GlobalWalletRegistry(Entity<WalletRegistry>);
+
+impl Global for GlobalWalletRegistry {}
+
+#[derive(Debug, Clone)]
+pub enum WalletEvent {
+    Updated,
+    Error(String),
+}
+
+impl EventEmitter<WalletEvent> for WalletRegistry {}
+
+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>,
+    store: Option<Arc<WalletRedbDatabase>>,
+    wallets: HashMap<MintUrl, Wallet>,
+    loading: bool,
+    store_opening: bool,
+    importing: bool,
+    import_queued: bool,
+    notification_listener: Option<Task<()>>,
+    tasks: SmallVec<[Task<()>; 2]>,
+    _subscriptions: SmallVec<[Subscription; 2]>,
+}
+
+struct LoadedState {
+    wallet: Option<keys::WalletData>,
+    tokens: HashMap<EventId, TokenEvent>,
+    stale: HashSet<EventId>,
+}
+
+impl WalletRegistry {
+    pub fn global(cx: &App) -> Entity<Self> {
+        cx.global::<GlobalWalletRegistry>().0.clone()
+    }
+
+    fn set_global(state: Entity<Self>, cx: &mut App) {
+        cx.set_global(GlobalWalletRegistry(state));
+    }
+
+    fn new(cx: &mut Context<Self>) -> Self {
+        let nostr = NostrRegistry::global(cx);
+        let mut subscriptions = smallvec![];
+
+        subscriptions.push(cx.subscribe(&nostr, |this, _nostr, event, cx| {
+            if event.signer_changed() {
+                this.reset(cx);
+                this.handle_notifications(cx);
+                this.load(cx);
+            }
+        }));
+
+        Self {
+            wallet: None,
+            tokens: HashMap::new(),
+            stale: HashSet::new(),
+            store: None,
+            wallets: HashMap::new(),
+            loading: false,
+            store_opening: false,
+            importing: false,
+            import_queued: false,
+            notification_listener: None,
+            tasks: smallvec![],
+            _subscriptions: subscriptions,
+        }
+    }
+
+    pub fn has_wallet(&self) -> bool {
+        self.wallet.is_some()
+    }
+
+    /// Spendable balance in satoshis across the active token events.
+    pub fn balance(&self) -> u64 {
+        self.tokens
+            .values()
+            .map(|token| token.proofs.iter().map(|proof| proof.amount).sum::<u64>())
+            .sum()
+    }
+
+    /// 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 {
+            self.emit_error("Wallet is still syncing, try again".to_string(), cx);
+            return;
+        }
+
+        let mint = match Url::parse(&mint) {
+            Ok(mint) => mint,
+            Err(_) => {
+                self.emit_error("Invalid mint URL".to_string(), cx);
+                return;
+            }
         };
-        let _wallet_event = nostr::nips::nip60::WalletEvent::new("privkey", vec![mint.clone()]);
-        let _token_event =
-            nostr::nips::nip60::TokenEvent::new(mint, vec![proof]).destroyed("id".to_string());
+
+        let (privkey, mints) = match self
+            .wallet
+            .as_ref()
+            .map(|wallet| (wallet.privkey.clone(), wallet.mints.clone()))
+        {
+            Some((privkey, mut mints)) => {
+                if mints.contains(&mint) {
+                    return;
+                }
+                mints.push(mint.clone());
+                (privkey, mints)
+            }
+            None => (keys::generate_privkey(), vec![mint.clone()]),
+        };
+
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client().clone();
+        let signer = nostr.read(cx).signer().clone();
+
+        let publish = cx.background_spawn(async move {
+            let me = signer.get_public_key_async().await?;
+            let event = keys::build_wallet_event(&signer, &me, &privkey, &mints).await?;
+            client.send_event(&event).await?;
+            client.database().save_event(&event).await?;
+
+            Ok::<_, Error>(keys::WalletData {
+                event_id: event.id,
+                created_at: event.created_at,
+                privkey,
+                mints,
+            })
+        });
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            let data = match publish.await {
+                Ok(data) => data,
+                Err(error) => {
+                    this.update(cx, |this, cx| {
+                        this.emit_error(format!("{error:#}"), cx);
+                    })
+                    .ok();
+                    return;
+                }
+            };
+
+            this.update(cx, |this, cx| {
+                // Our event wins ties so a relay echo of the same event is a no-op
+                let take_over = match &this.wallet {
+                    Some(current) => data.created_at >= current.created_at,
+                    None => true,
+                };
+
+                if take_over {
+                    this.wallet = Some(data);
+                    this.ensure_wallets(cx);
+                    this.emit_updated(cx);
+                }
+            })
+            .ok();
+        }));
+    }
+
+    fn reset(&mut self, cx: &mut Context<Self>) {
+        self.wallet = None;
+        self.tokens.clear();
+        self.stale.clear();
+        self.store = None;
+        self.wallets.clear();
+        self.loading = false;
+        self.store_opening = false;
+        self.importing = false;
+        self.import_queued = false;
+        self.notification_listener = None;
+        self.tasks.clear();
+        cx.notify();
+    }
+
+    fn handle_notifications(&mut self, cx: &mut Context<Self>) {
+        self.notification_listener = None;
+
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client().clone();
+        let signer = nostr.read(cx).signer().clone();
+
+        let (tx, rx) = flume::bounded::<Event>(100);
+
+        self.notification_listener = Some(cx.background_spawn(async move {
+            let me = signer.get_public_key_async().await.ok();
+            let mut notifications = client.notifications();
+
+            const MAX_PROCESSED: usize = 1000;
+            let mut processed_events = HashSet::new();
+
+            while let Some(notification) = notifications.next().await {
+                if let ClientNotification::Message { message, .. } = notification
+                    && let RelayMessage::Event { event, .. } = *message
+                    && me == Some(event.pubkey)
+                    && (matches!(
+                        event.kind,
+                        Kind::CashuWallet
+                            | Kind::CashuWalletUnspentProof
+                            | Kind::CashuWalletSpendingHistory
+                    ) || is_token_deletion(&event))
+                {
+                    // Prune the dedup set before it grows unbounded
+                    if processed_events.len() >= MAX_PROCESSED {
+                        processed_events.clear();
+                    }
+
+                    if !processed_events.insert(event.id) {
+                        continue;
+                    }
+
+                    if tx.send_async(event.into_owned()).await.is_err() {
+                        break;
+                    }
+                }
+            }
+        }));
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            while let Ok(event) = rx.recv_async().await {
+                this.update(cx, |this, cx| {
+                    this.absorb(event, cx);
+                })
+                .ok();
+            }
+        }));
+    }
+
+    fn absorb(&mut self, event: Event, cx: &mut Context<Self>) {
+        match event.kind {
+            Kind::CashuWallet => self.absorb_wallet(event, cx),
+            Kind::CashuWalletUnspentProof => self.absorb_token(event, cx),
+            Kind::EventDeletion => self.absorb_deletion(event, cx),
+            _ => {}
+        }
+    }
+
+    fn absorb_wallet(&mut self, event: Event, cx: &mut Context<Self>) {
+        let newer = match &self.wallet {
+            Some(current) => event.created_at > current.created_at,
+            None => true,
+        };
+
+        if !newer {
+            return;
+        }
+
+        let nostr = NostrRegistry::global(cx);
+        let signer = nostr.read(cx).signer().clone();
+
+        let decrypted = cx.background_spawn(async move {
+            let me = signer.get_public_key_async().await?;
+            keys::decrypt_wallet_event(&signer, &me, &event).await
+        });
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            let data = match decrypted.await {
+                Ok(data) => data,
+                Err(error) => {
+                    this.update(cx, |this, cx| {
+                        this.emit_error(format!("{error:#}"), cx);
+                    })
+                    .ok();
+                    return;
+                }
+            };
+
+            this.update(cx, |this, cx| {
+                this.wallet = Some(data);
+                this.ensure_wallets(cx);
+                this.emit_updated(cx);
+            })
+            .ok();
+        }));
+    }
+
+    fn absorb_token(&mut self, event: Event, cx: &mut Context<Self>) {
+        if self.stale.contains(&event.id) {
+            return;
+        }
+
+        let nostr = NostrRegistry::global(cx);
+        let signer = nostr.read(cx).signer().clone();
+        let event_id = event.id;
+
+        let decrypted = cx.background_spawn(async move {
+            let me = signer.get_public_key_async().await?;
+            keys::decrypt_token_event(&signer, &me, &event).await
+        });
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            let token = match decrypted.await {
+                Ok(token) => token,
+                Err(error) => {
+                    this.update(cx, |this, cx| {
+                        this.emit_error(format!("{error:#}"), cx);
+                    })
+                    .ok();
+                    return;
+                }
+            };
+
+            this.update(cx, |this, cx| {
+                // The replacement may arrive before the tokens it replaces
+                for old in &token.del {
+                    if let Ok(old_id) = EventId::from_hex(old) {
+                        this.tokens.remove(&old_id);
+                        this.stale.insert(old_id);
+                    }
+                }
+                this.tokens.insert(event_id, token);
+                this.emit_updated(cx);
+                this.schedule_import(cx);
+            })
+            .ok();
+        }));
+    }
+
+    fn absorb_deletion(&mut self, event: Event, cx: &mut Context<Self>) {
+        let mut removed_any = false;
+        for tag in event.tags.iter() {
+            if tag.kind() != "e" {
+                continue;
+            }
+            let Some(hex) = tag.content() else {
+                continue;
+            };
+            if let Ok(id) = EventId::from_hex(hex) {
+                if self.tokens.remove(&id).is_some() {
+                    removed_any = true;
+                }
+                self.stale.insert(id);
+            }
+        }
+        if removed_any {
+            self.emit_updated(cx);
+            self.schedule_import(cx);
+        }
+    }
+
+    /// Open the cdk store and build a wallet per mint that is not open yet.
+    fn ensure_wallets(&mut self, cx: &mut Context<Self>) {
+        let Some(data) = self.wallet.clone() else {
+            return;
+        };
+
+        let pending: Vec<Url> = data
+            .mints
+            .iter()
+            .filter(|mint| {
+                MintUrl::from_str(mint.as_str())
+                    .map(|mint_url| !self.wallets.contains_key(&mint_url))
+                    .unwrap_or(true)
+            })
+            .cloned()
+            .collect();
+
+        if pending.is_empty() {
+            return;
+        }
+
+        if let Some(store) = self.store.clone() {
+            let seed = match keys::derive_seed(&data.privkey) {
+                Ok(seed) => seed,
+                Err(error) => {
+                    self.emit_error(format!("{error:#}"), cx);
+                    return;
+                }
+            };
+            for mint in pending {
+                let Ok(mint_url) = MintUrl::from_str(mint.as_str()) else {
+                    continue;
+                };
+
+                match WalletBuilder::new()
+                    .mint_url(mint_url.clone())
+                    .unit(CurrencyUnit::Sat)
+                    .localstore(store.clone())
+                    .seed(seed)
+                    .build()
+                {
+                    Ok(wallet) => {
+                        self.wallets.insert(mint_url, wallet);
+                    }
+                    Err(error) => self.emit_error(format!("{error:#}"), cx),
+                }
+            }
+
+            cx.notify();
+            self.schedule_import(cx);
+
+            return;
+        }
+
+        if self.store_opening {
+            return;
+        }
+        self.store_opening = true;
+
+        let privkey = data.privkey.clone();
+        let mints = data.mints.clone();
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            match Tokio::spawn_result(cx, open_store(privkey, mints)).await {
+                Ok((store, wallets)) => {
+                    this.update(cx, |this, cx| {
+                        if this.store.is_none() {
+                            this.store = Some(store);
+                        }
+
+                        for (mint_url, wallet) in wallets {
+                            this.wallets.insert(mint_url, wallet);
+                        }
+
+                        this.store_opening = false;
+                        cx.notify();
+                        this.schedule_import(cx);
+                    })
+                    .ok();
+                }
+                Err(error) => {
+                    this.update(cx, |this, cx| {
+                        this.store_opening = false;
+                        this.emit_error(format!("{error:#}"), cx);
+                    })
+                    .ok();
+                }
+            }
+        }));
+    }
+
+    /// Mirror the active token events into the cdk store.
+    fn schedule_import(&mut self, cx: &mut Context<Self>) {
+        if self.importing {
+            self.import_queued = true;
+            return;
+        }
+
+        let Some(store) = self.store.clone() else {
+            return;
+        };
+
+        if self.wallets.is_empty() {
+            return;
+        }
+
+        let mut active: Vec<(MintUrl, Vec<Proof>)> = Vec::new();
+        for token in self.tokens.values() {
+            let Ok(mint_url) = MintUrl::from_str(token.mint.as_str()) else {
+                continue;
+            };
+
+            let mut proofs = Vec::new();
+            for proof in &token.proofs {
+                let Ok(keyset_id) = Id::from_str(&proof.id) else {
+                    continue;
+                };
+
+                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;
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            let result = Tokio::spawn_result(cx, async move {
+                let mut added = Vec::new();
+                let mut ys = HashSet::new();
+
+                for (mint_url, proofs) in active {
+                    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;
+
+                if this.import_queued {
+                    this.import_queued = false;
+                    this.schedule_import(cx);
+                }
+
+                if let Err(error) = result {
+                    this.emit_error(format!("{error:#}"), cx);
+                }
+            })
+            .ok();
+        }));
+    }
+
+    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);
+    }
+
+    fn load(&mut self, cx: &mut Context<Self>) {
+        if self.loading {
+            return;
+        }
+        self.loading = true;
+
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client().clone();
+        let signer = nostr.read(cx).signer().clone();
+
+        let subscribe_client = client.clone();
+        let cached = cx.background_spawn(async move {
+            let me = signer.get_public_key_async().await?;
+            let state = load_inner(&me, &client, &signer).await?;
+            Ok::<_, Error>((me, state))
+        });
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            let (me, state) = match cached.await {
+                Ok(loaded) => loaded,
+                Err(error) => {
+                    this.update(cx, |this, cx| {
+                        this.emit_error(format!("{error:#}"), cx);
+                    })
+                    .ok();
+                    this.update(cx, |this, _cx| {
+                        this.loading = false;
+                    })
+                    .ok();
+                    return;
+                }
+            };
+
+            this.update(cx, |this, cx| {
+                this.apply_loaded(state, cx);
+            })
+            .ok();
+
+            this.update(cx, |this, _cx| {
+                this.loading = false;
+            })
+            .ok();
+
+            // Subscribe only after the cache snapshot; from then on the state
+            // advances exclusively through the notification listener.
+            let subscribed =
+                cx.background_spawn(
+                    async move { subscribe_client.subscribe(wallet_filter(me)).await },
+                );
+            if let Err(error) = subscribed.await {
+                log::warn!("wallet subscription failed: {error:#}");
+            }
+        }));
+    }
+
+    fn emit_updated(&self, cx: &mut Context<Self>) {
+        cx.emit(WalletEvent::Updated);
+        cx.notify();
+    }
+
+    fn emit_error(&self, message: String, cx: &mut Context<Self>) {
+        log::error!("wallet: {message}");
+        cx.emit(WalletEvent::Error(message));
     }
 }
 
-/// Only compiles when `nostr` and `cdk_nostr` resolve to the same crate.
-#[allow(dead_code)]
-fn spike_unified(key: nostr::key::SecretKey) -> cdk_nostr::nostr::key::SecretKey {
-    key
+async fn load_inner<S>(me: &PublicKey, client: &Client, signer: &S) -> Result<LoadedState, Error>
+where
+    S: AsyncNip44 + ?Sized,
+{
+    let events = client.database().query(wallet_filter(*me)).await?;
+
+    let mut wallet_event: Option<&Event> = None;
+    let mut token_events = Vec::new();
+    let mut deletions = Vec::new();
+
+    for event in &events {
+        match event.kind {
+            Kind::CashuWallet => {
+                let replace = match &wallet_event {
+                    Some(current) => event.created_at > current.created_at,
+                    None => true,
+                };
+
+                if replace {
+                    wallet_event = Some(event);
+                }
+            }
+            Kind::CashuWalletUnspentProof => token_events.push(event),
+            Kind::EventDeletion if is_token_deletion(event) => deletions.push(event),
+            _ => {}
+        }
+    }
+
+    let mut stale = HashSet::new();
+    for event in deletions {
+        for tag in event.tags.iter() {
+            if tag.kind() != "e" {
+                continue;
+            }
+
+            let Some(hex) = tag.content() else {
+                continue;
+            };
+
+            if let Ok(id) = EventId::from_hex(hex) {
+                stale.insert(id);
+            }
+        }
+    }
+
+    let mut tokens = HashMap::new();
+    for event in token_events {
+        if stale.contains(&event.id) {
+            continue;
+        }
+        let token = match keys::decrypt_token_event(signer, me, event).await {
+            Ok(token) => token,
+            Err(error) => {
+                log::warn!("skipping unreadable token event {}: {error:#}", event.id);
+                continue;
+            }
+        };
+        tokens.insert(event.id, token);
+    }
+
+    let superseded: HashSet<EventId> = tokens
+        .values()
+        .flat_map(|token| token.del.iter())
+        .filter_map(|hex| EventId::from_hex(hex).ok())
+        .collect();
+
+    tokens.retain(|id, _| !stale.contains(id) && !superseded.contains(id));
+    stale.extend(superseded);
+
+    let wallet = match wallet_event {
+        Some(event) => match keys::decrypt_wallet_event(signer, me, event).await {
+            Ok(wallet) => Some(wallet),
+            Err(error) => {
+                log::warn!("skipping wallet event {}: {error:#}", event.id);
+                None
+            }
+        },
+        None => None,
+    };
+
+    Ok(LoadedState {
+        wallet,
+        tokens,
+        stale,
+    })
 }
 
-#[allow(dead_code)]
-fn spike_sdk_unified(client: nostr_sdk::client::Client) -> cdk_nostr::nostr_sdk::client::Client {
-    client
+async fn open_store(
+    privkey: SecretKey,
+    mints: Vec<Url>,
+) -> Result<(Arc<WalletRedbDatabase>, HashMap<MintUrl, Wallet>), Error> {
+    let dir = config_dir()
+        .join("cashu")
+        .join(Keys::new(privkey.clone()).public_key().to_hex());
+    std::fs::create_dir_all(&dir)?;
+
+    let store = Arc::new(WalletRedbDatabase::new(&dir.join("wallet.redb"))?);
+    let seed = keys::derive_seed(&privkey)?;
+
+    let mut wallets = HashMap::new();
+    for mint in &mints {
+        let mint_url = MintUrl::from_str(mint.as_str())?;
+        let wallet = WalletBuilder::new()
+            .mint_url(mint_url.clone())
+            .unit(CurrencyUnit::Sat)
+            .localstore(store.clone())
+            .seed(seed)
+            .build()?;
+        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))
+}
+
+fn wallet_filter(author: PublicKey) -> Filter {
+    Filter::new().author(author).kinds([
+        Kind::CashuWallet,
+        Kind::CashuWalletUnspentProof,
+        Kind::CashuWalletSpendingHistory,
+        Kind::EventDeletion,
+    ])
+}
+
+fn is_token_deletion(event: &Event) -> bool {
+    event.kind == Kind::EventDeletion
+        && event
+            .tags
+            .iter()
+            .any(|tag| tag.kind() == "k" && tag.content() == Some("7375"))
 }

docs/nip60.md

@@ -193,13 +193,27 @@ wired into desktop. `cargo check -p nip60` and the full desktop check pass.
 The remaining spike code in `crates/nip60/src/lib.rs` is deleted in
 phase 1.
 
-### Phase 1 — bootstrap + load + balance
+### Phase 1 — bootstrap + load + balance (done)
 
-1. `crates/nip60` skeleton: registry, `init`, signer-changed reset.
-2. `keys.rs`: wallet key generation, seed derivation, `17375` parse/build.
-3. Load path: query cached + subscribed events, build the unspent set,
-   open cdk wallets, import proofs, expose `balance()` and emit `Updated`.
-4. `setup(mint_url)` + mint list update (republish `17375`).
+Implemented in `crates/nip60` (`lib.rs`, `keys.rs`):
+
+- `WalletRegistry` follows the nip29/nip4e pattern: global entity, signer
+  reset, flume notification listener, task + subscription SmallVecs.
+- `keys.rs`: wallet key generation, HKDF seed derivation, `17375`
+  encrypt/parse through the signer (works with bunkers), unit tests.
+- Load: subscribe + query cache for 17375/7375/7376 and kind 5 `k:7375`,
+  decrypt via signer, prune deleted (`k:7375` deletes) and superseded
+  (`del` chains) token events, expose `has_wallet()`/`balance()`.
+- cdk: redb store per wallet pubkey under `config_dir()/cashu`, one `Wallet`
+  per mint (`sat`), unspent proofs imported/diffed against the store on
+  every sync, `recover_incomplete_sagas` + `mint_unissued_quotes` fire on
+  first store open.
+- `setup(mint)` creates the wallet (first mint) or adds a mint, publishing
+  `17375` and saving it to the local cache.
+
+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
 
@@ -240,6 +254,8 @@ phase 1.
 - 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
+  reserved by local in-flight operations lands with phase 2 operations.
 - 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