feat(platform-wallet): snapshot readers + unified V002 schema migration

packages/rs-platform-wallet-storage/Cargo.toml

@@ -247,3 +247,9 @@ kv = ["sqlite"]
 __test-helpers = ["sqlite"]
 # e2e tests that drive the #3692 manager-apply path; enabled in the integrated stack (dash-evo-tool).
 rehydration-apply = []
+# Pass-through to `platform-wallet/shielded` so the feature-gated
+# `PlatformWalletChangeSet::shielded` field is visible to the exhaustive
+# `versions::touched_domains` destructure (the R8 forgotten-domain guard).
+# Storage persists no shielded state itself; this only aligns visibility so
+# an added always-on field stays a compile error.
+shielded = ["platform-wallet/shielded"]

packages/rs-platform-wallet-storage/migrations/V002__unified.rs

@@ -0,0 +1,75 @@
+//! Unified additive migration for `platform-wallet-storage` (#3968).
+//!
+//! Additive-only: V001 stays byte-identical so refinery's applied-migration
+//! checksum for version 1 never diverges on an existing store. V002 lifts
+//! `max_supported_version()` from 1 to 2 automatically (the value is derived
+//! from the embedded list) and lands three concerns in one migration event:
+//!
+//! - `core_address_pool` — per-index address-pool rows with a `used` flag,
+//!   the first-class row store that replaces `core_utxos` script-derivation
+//!   for the address-reuse guard. `pool_type` is in the primary key so an
+//!   External (receive) and Internal (change) pool never collide at the same
+//!   `address_index`. `script` (the address' `script_pubkey`) is stored so
+//!   the reader returns used addresses verbatim and the UTXO writer can
+//!   attribute an outpoint to its owning account, both without re-deriving.
+//! - `meta_data_versions` — per-`(wallet_id, domain)` monotonic `seq`
+//!   bumped inside the flush transaction, the cache-invalidation keystone.
+//!   No FK (a domain row may be written before its typed parent syncs,
+//!   mirroring the `meta_*` tables); a soft-cascade trigger reaps rows on
+//!   wallet delete.
+//! - `meta_store_generation` — a single-row store-generation token,
+//!   initialized with `randomblob(16)` so the rendered SQL stays deterministic (the
+//!   content fingerprint pins the text, the runtime value is unique per
+//!   store). Regenerated on restore.
+//!
+//! No MAC column ships here — manifest authentication is deferred out of
+//! this workstream (dev-plan §7).
+
+pub fn migration() -> String {
+    "\
+CREATE TABLE core_address_pool (
+    wallet_id BLOB NOT NULL,
+    account_index INTEGER NOT NULL,
+    key_class INTEGER NOT NULL DEFAULT 0,
+    pool_type INTEGER NOT NULL CHECK (pool_type IN (0, 1, 2, 3)),
+    address_index INTEGER NOT NULL,
+    script BLOB NOT NULL,
+    used INTEGER NOT NULL DEFAULT 0 CHECK (used IN (0, 1)),
+    PRIMARY KEY (wallet_id, account_index, key_class, pool_type, address_index),
+    FOREIGN KEY (wallet_id) REFERENCES wallets(wallet_id) ON DELETE CASCADE
+);
+
+CREATE INDEX idx_core_address_pool_used
+    ON core_address_pool(wallet_id, used);
+
+-- The UTXO writer attributes an outpoint to its owning account by matching
+-- the outpoint's script against a pool row.
+CREATE INDEX idx_core_address_pool_script
+    ON core_address_pool(wallet_id, script);
+
+CREATE TABLE meta_data_versions (
+    wallet_id BLOB NOT NULL,
+    domain TEXT NOT NULL,
+    seq INTEGER NOT NULL DEFAULT 0,
+    PRIMARY KEY (wallet_id, domain)
+);
+
+-- Soft-cascade reap, matching the meta_* tables: no FK (a domain may be
+-- bumped before its typed parent exists), so a trigger clears rows when
+-- the owning wallet is deleted.
+CREATE TRIGGER cascade_meta_data_versions_on_wallet_delete
+AFTER DELETE ON wallets
+FOR EACH ROW
+BEGIN
+    DELETE FROM meta_data_versions WHERE wallet_id = OLD.wallet_id;
+END;
+
+CREATE TABLE meta_store_generation (
+    id INTEGER NOT NULL PRIMARY KEY CHECK (id = 0),
+    generation BLOB NOT NULL
+);
+
+INSERT INTO meta_store_generation (id, generation) VALUES (0, randomblob(16));
+"
+    .to_string()
+}

packages/rs-platform-wallet-storage/src/sqlite/backup.rs

@@ -274,6 +274,17 @@ pub fn restore_from(dest_db_path: &Path, src_backup: &Path) -> Result<(), Wallet
 
     // 10. Re-tighten perms (idempotent; SQLite may re-materialise -wal/-shm).
     apply_secure_permissions(dest_db_path)?;
+
+    // 11. Regenerate the store-generation token so the restored copy is
+    //     distinguishable from its source — a client cache keyed on the old
+    //     generation misses rather than serving stale entries. A pre-V002
+    //     backup has no generation table; it is (re)seeded on its later
+    //     migration to V002.
+    {
+        let conn =
+            crate::sqlite::conn::open_conn(dest_db_path, crate::sqlite::conn::Access::ReadWrite)?;
+        crate::sqlite::schema::versions::regenerate_generation(&conn)?;
+    }
     Ok(())
 }
 

packages/rs-platform-wallet-storage/src/sqlite/migrations.rs

@@ -175,6 +175,52 @@ pub fn embedded_migrations_fingerprint() -> [u8; 32] {
     hasher.finalize().into()
 }
 
+/// SHA-256 over `(version, name, rendered SQL)` of every embedded migration
+/// in version order. Unlike [`embedded_migrations_fingerprint`] this is
+/// content-level: it pins each migration's SQL body, so an in-place DDL edit
+/// (e.g. renaming a table inside a same-named file) breaks the golden test.
+/// This is the guard the D0 schema freeze relies on; the identity-only
+/// fingerprint cannot catch a same-name body edit.
+///
+/// The SQL *text* is deterministic even where a value is generated at run
+/// time (`randomblob(16)`): the literal string is hashed, not the runtime
+/// bytes.
+#[cfg(any(test, feature = "__test-helpers"))]
+pub fn embedded_migrations_sql_fingerprint() -> [u8; 32] {
+    use sha2::{Digest, Sha256};
+    let mut migrations = migrations::runner().get_migrations().clone();
+    migrations.sort_by_key(|m| m.version());
+    let mut hasher = Sha256::new();
+    for m in &migrations {
+        hasher.update((m.version() as u32).to_be_bytes());
+        hasher.update([0u8]);
+        hasher.update(m.name().as_bytes());
+        hasher.update([0u8]);
+        let sql = m
+            .sql()
+            .expect("embedded migrations always carry rendered SQL");
+        hasher.update(sql.as_bytes());
+        hasher.update([0u8]);
+    }
+    hasher.finalize().into()
+}
+
+/// Rendered SQL of every embedded migration, in version order. Used by the
+/// schema-freeze grep guard to scan for retired table names.
+#[cfg(any(test, feature = "__test-helpers"))]
+pub fn embedded_migrations_sql() -> Vec<String> {
+    let mut migrations = migrations::runner().get_migrations().clone();
+    migrations.sort_by_key(|m| m.version());
+    migrations
+        .iter()
+        .map(|m| {
+            m.sql()
+                .expect("embedded migrations always carry rendered SQL")
+                .to_string()
+        })
+        .collect()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

packages/rs-platform-wallet-storage/src/sqlite/persister.rs

@@ -7,7 +7,7 @@ use std::sync::{Arc, Mutex, MutexGuard, OnceLock};
 use rusqlite::{Connection, OptionalExtension};
 
 use platform_wallet::changeset::{
-    ClientStartState, Merge, PersistenceError, PlatformWalletChangeSet, PlatformWalletPersistence,
+    ClientStartState, PersistenceError, PlatformWalletChangeSet, PlatformWalletPersistence,
 };
 use platform_wallet::wallet::platform_wallet::WalletId;
 
@@ -940,13 +940,27 @@ impl PlatformWalletPersistence for SqlitePersister {
                 .map_err(PersistenceError::from)?;
             let identity_keys = schema::identity_keys::load_state(&conn, &wallet_id)
                 .map_err(PersistenceError::from)?;
-            // Every address that ever held a UTXO (spent + unspent) is "used":
-            // the address-reuse guard so a used-then-emptied address is never
-            // handed back as a fresh receive address. The in-band pool snapshot
-            // was retired, so we derive this from the full core_utxos set.
-            let used_core_addresses =
-                schema::core_state::load_used_addresses(&conn, &wallet_id, network)
+            // Used addresses drive the reuse guard: a used-then-emptied
+            // address must never be handed back as a fresh receive address.
+            // Union the verbatim `core_address_pool` used-set with the
+            // `core_utxos`-derived set (spent + unspent). The guard is
+            // monotonic, so a mixed store — historical UTXOs plus a later
+            // partial pool snapshot that never enumerates them — must surface
+            // both; neither source may shadow the other. Deduped by script.
+            let used_core_addresses = {
+                let mut seen = std::collections::HashSet::new();
+                let mut union = Vec::new();
+                let pool = schema::core_pool::load_used_addresses(&conn, &wallet_id, network)
                     .map_err(PersistenceError::from)?;
+                let utxo = schema::core_state::load_used_addresses(&conn, &wallet_id, network)
+                    .map_err(PersistenceError::from)?;
+                for addr in pool.into_iter().chain(utxo) {
+                    if seen.insert(addr.script_pubkey().to_bytes()) {
+                        union.push(addr);
+                    }
+                }
+                union
+            };
 
             state.wallets.insert(
                 wallet_id,
@@ -994,25 +1008,9 @@ impl PlatformWalletPersistence for SqlitePersister {
 /// from the public fields so no storage-only helper leaks into the
 /// `rs-platform-wallet` API.
 fn populated_field_count(cs: &PlatformWalletChangeSet) -> usize {
-    [
-        cs.core.is_empty(),
-        cs.identities.is_empty(),
-        cs.identity_keys.is_empty(),
-        cs.contacts.is_empty(),
-        cs.platform_addresses.is_empty(),
-        cs.asset_locks.is_empty(),
-        cs.token_balances.is_empty(),
-        cs.dashpay_profiles.as_ref().is_none_or(|m| m.is_empty()),
-        cs.dashpay_payments_overlay
-            .as_ref()
-            .is_none_or(|m| m.is_empty()),
-        cs.wallet_metadata.is_none(),
-        cs.account_registrations.is_empty(),
-        cs.account_address_pools.is_empty(),
-    ]
-    .iter()
-    .filter(|empty| !**empty)
-    .count()
+    // Single source of truth with the version-domain mapping: each populated
+    // field is exactly one touched domain.
+    schema::versions::touched_domains(cs).len()
 }
 
 fn validate_config(config: &SqlitePersisterConfig) -> Result<(), WalletStorageError> {
@@ -1088,10 +1086,12 @@ fn apply_changeset_to_tx(
     if !cs.account_registrations.is_empty() {
         schema::accounts::apply_registrations(tx, wallet_id, &cs.account_registrations)?;
     }
-    // `account_address_pools` is intentionally NOT applied: UTXO attribution
-    // is hardcoded to the default account (index 0) in `core_state`, so the
-    // pool snapshot is no longer a storage input. The changeset field is kept
-    // for API stability and still feeds non-storage consumers.
+    // Pools land before core so the UTXO writer can attribute each outpoint
+    // to its owning account by matching the outpoint's script against a
+    // freshly-written `core_address_pool` row.
+    if !cs.account_address_pools.is_empty() {
+        schema::core_pool::apply_pools(tx, wallet_id, &cs.account_address_pools)?;
+    }
     if let Some(core) = cs.core.as_ref() {
         schema::core_state::apply(tx, wallet_id, core)?;
     }
@@ -1121,6 +1121,9 @@ fn apply_changeset_to_tx(
             cs.dashpay_payments_overlay.as_ref(),
         )?;
     }
+    // Bump each touched domain's version inside this same tx so a domain's
+    // cache-invalidation marker commits atomically with its data.
+    schema::versions::bump_touched_domains(tx, wallet_id, cs)?;
     Ok(())
 }
 

packages/rs-platform-wallet-storage/src/sqlite/schema/core_pool.rs

@@ -0,0 +1,140 @@
+//! Writer + account-attribution helper for the `core_address_pool` table.
+//!
+//! Per-index address-pool rows carrying a `used` flag, scoped by
+//! `(wallet_id, account_index, key_class, pool_type, address_index)`. The
+//! first-class row store the reader consumes verbatim — no `core_utxos`
+//! script-derivation, no horizon-walk re-derivation. Populated from the
+//! `account_address_pools` changeset snapshots; the UTXO writer reads it
+//! back to attribute an outpoint to its owning account.
+
+use rusqlite::{params, OptionalExtension, Transaction};
+
+use platform_wallet::changeset::AccountAddressPoolEntry;
+use platform_wallet::wallet::platform_wallet::WalletId;
+
+use key_wallet::managed_account::address_pool::AddressPoolType;
+
+use crate::sqlite::error::WalletStorageError;
+use crate::sqlite::schema::accounts;
+
+/// Stored `pool_type` discriminant. Kept in the primary key so an External
+/// and an Internal pool never collide at the same `address_index`.
+pub(crate) fn pool_type_to_i64(pool_type: AddressPoolType) -> i64 {
+    match pool_type {
+        AddressPoolType::External => 0,
+        AddressPoolType::Internal => 1,
+        AddressPoolType::Absent => 2,
+        AddressPoolType::AbsentHardened => 3,
+    }
+}
+
+const UPSERT_POOL_SQL: &str = "INSERT INTO core_address_pool \
+        (wallet_id, account_index, key_class, pool_type, address_index, script, used) \
+     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) \
+     ON CONFLICT(wallet_id, account_index, key_class, pool_type, address_index) DO UPDATE SET \
+        script = excluded.script, \
+        used = MAX(used, excluded.used)";
+
+/// Expand `account_address_pools` snapshots into per-index
+/// `core_address_pool` rows. Idempotent: `script` is derivation-stable and
+/// `used` is monotonic (`MAX`), so re-applying the same snapshot is a no-op
+/// and a used address can never revert to unused (the reuse-guard invariant).
+pub fn apply_pools(
+    tx: &Transaction<'_>,
+    wallet_id: &WalletId,
+    pools: &[AccountAddressPoolEntry],
+) -> Result<(), WalletStorageError> {
+    if pools.is_empty() {
+        return Ok(());
+    }
+    let mut stmt = tx.prepare_cached(UPSERT_POOL_SQL)?;
+    for entry in pools {
+        let account_index = i64::from(accounts::account_index(&entry.account_type));
+        // TODO(key_class): PlatformPayment carries a real key_class; every
+        // other account maps to the 0 sentinel until the pool snapshot
+        // threads a per-pool key class.
+        let key_class = i64::from(accounts::account_key_class(&entry.account_type));
+        let pool_type = pool_type_to_i64(entry.pool_type);
+        for info in &entry.addresses {
+            stmt.execute(params![
+                wallet_id.as_slice(),
+                account_index,
+                key_class,
+                pool_type,
+                i64::from(info.index),
+                info.script_pubkey.as_bytes(),
+                info.used,
+            ])?;
+        }
+    }
+    Ok(())
+}
+
+/// Owning account index for a UTXO, matched by its `script_pubkey` against a
+/// pool row. `None` when no pool row covers the script — the UTXO writer
+/// then falls back to account 0 (the one-way historical-attribution default,
+/// R7): funds are never dropped, only conservatively bucketed.
+pub fn account_index_for_script(
+    tx: &Transaction<'_>,
+    wallet_id: &WalletId,
+    script: &[u8],
+) -> Result<Option<u32>, WalletStorageError> {
+    let idx: Option<i64> = tx
+        .prepare_cached(
+            "SELECT account_index FROM core_address_pool \
+             WHERE wallet_id = ?1 AND script = ?2 LIMIT 1",
+        )?
+        .query_row(params![wallet_id.as_slice(), script], |row| row.get(0))
+        .optional()?;
+    idx.map(|v| crate::sqlite::util::safe_cast::i64_to_u32("core_address_pool.account_index", v))
+        .transpose()
+}
+
+/// Used addresses for a wallet, read verbatim from `core_address_pool`
+/// (`used = 1`) with no re-derivation. Possibly empty. The caller **unions**
+/// this with the `core_utxos`-derived set — the reuse guard is monotonic, so
+/// a mixed store (historical UTXOs a later partial pool snapshot never
+/// enumerates) must surface both sources, never drop the historical ones.
+///
+/// `network` turns each stored `script` back into an [`Address`]; a script
+/// that isn't a valid address is a hard error — corruption is never silently
+/// dropped, matching [`crate::sqlite::schema::core_state::load_used_addresses`].
+pub fn load_used_addresses(
+    conn: &rusqlite::Connection,
+    wallet_id: &WalletId,
+    network: dashcore::Network,
+) -> Result<Vec<dashcore::Address>, WalletStorageError> {
+    let mut stmt = conn.prepare(
+        "SELECT DISTINCT script FROM core_address_pool \
+         WHERE wallet_id = ?1 AND used = 1 ORDER BY script",
+    )?;
+    let rows = stmt.query_map(params![wallet_id.as_slice()], |row| {
+        row.get::<_, Vec<u8>>(0)
+    })?;
+    let mut out = Vec::new();
+    for r in rows {
+        let script = dashcore::ScriptBuf::from_bytes(r?);
+        let address = dashcore::Address::from_script(&script, network).map_err(|_| {
+            WalletStorageError::blob_decode("core_address_pool.script not an address")
+        })?;
+        out.push(address);
+    }
+    Ok(out)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn pool_type_discriminants_are_stable_and_distinct() {
+        let all = [
+            AddressPoolType::External,
+            AddressPoolType::Internal,
+            AddressPoolType::Absent,
+            AddressPoolType::AbsentHardened,
+        ];
+        let mapped: Vec<i64> = all.iter().copied().map(pool_type_to_i64).collect();
+        assert_eq!(mapped, vec![0, 1, 2, 3]);
+    }
+}