feat(pubkey-manager): add pubkey watch function similar to balance watch

packages/komodo_defi_sdk/lib/src/balances/balance_manager.dart

@@ -1,4 +1,5 @@
 import 'dart:async';
+
 import 'package:flutter/foundation.dart'; // Add this import for debugPrint
 import 'package:komodo_defi_local_auth/komodo_defi_local_auth.dart';
 import 'package:komodo_defi_sdk/src/activation/activation_manager.dart';

packages/komodo_defi_sdk/lib/src/pubkeys/pubkey_manager.dart

@@ -1,24 +1,72 @@
+import 'dart:async';
+
 import 'package:komodo_defi_local_auth/komodo_defi_local_auth.dart';
 import 'package:komodo_defi_sdk/src/_internal_exports.dart';
 import 'package:komodo_defi_types/komodo_defi_type_utils.dart';
 import 'package:komodo_defi_types/komodo_defi_types.dart';
 
+/// Interface defining the contract for pubkey management operations
+abstract class IPubkeyManager {
+  /// Get pubkeys for a given asset, handling HD/non-HD differences internally
+  Future<AssetPubkeys> getPubkeys(Asset asset);
+
+  /// Watch pubkeys for a given asset, emitting the initial state if available
+  /// and polling for updates at a fixed interval. Optionally activates asset.
+  Stream<AssetPubkeys> watchPubkeys(
+    Asset asset, {
+    bool activateIfNeeded = true,
+  });
+
+  /// Get the last known pubkeys for an asset without triggering a refresh.
+  /// Returns null if no pubkeys have been fetched yet.
+  AssetPubkeys? lastKnown(AssetId assetId);
+
+  /// Create a new pubkey for an asset if supported
+  Future<PubkeyInfo> createNewPubkey(Asset asset);
+
+  /// Streamed version of [createNewPubkey]
+  Stream<NewAddressState> createNewPubkeyStream(Asset asset);
+
+  /// Unban pubkeys according to [unbanBy] criteria
+  Future<UnbanPubkeysResult> unbanPubkeys(UnbanBy unbanBy);
+
+  /// Pre-caches pubkeys for an asset to warm the cache and notify listeners
+  Future<void> preCachePubkeys(Asset asset);
+
+  /// Dispose of any resources
+  Future<void> dispose();
+}
+
 /// Manager responsible for handling pubkey operations across different assets
-class PubkeyManager {
-  PubkeyManager(this._client, this._auth, this._activationCoordinator);
+class PubkeyManager implements IPubkeyManager {
+  PubkeyManager(this._client, this._auth, this._activationCoordinator) {
+    _authSubscription = _auth.authStateChanges.listen(_handleAuthStateChanged);
+  }
 
   final ApiClient _client;
   final KomodoDefiLocalAuth _auth;
   final SharedActivationCoordinator _activationCoordinator;
 
+  // Internal state for watching pubkeys per asset
+  final Map<AssetId, AssetPubkeys> _pubkeysCache = {};
+  final Map<AssetId, StreamSubscription<dynamic>> _activeWatchers = {};
+  final Map<AssetId, StreamController<AssetPubkeys>> _pubkeysControllers = {};
+
+  StreamSubscription<KdfUser?>? _authSubscription;
+  WalletId? _currentWalletId;
+  bool _isDisposed = false;
+  final Duration _defaultPollingInterval = const Duration(seconds: 30);
+
   /// Get pubkeys for a given asset, handling HD/non-HD differences internally
+  @override
   Future<AssetPubkeys> getPubkeys(Asset asset) async {
     await retry(() => _activationCoordinator.activateAsset(asset));
     final strategy = await _resolvePubkeyStrategy(asset);
     return strategy.getPubkeys(asset.id, _client);
   }
 
   /// Create a new pubkey for an asset if supported
+  @override
   Future<PubkeyInfo> createNewPubkey(Asset asset) async {
     await retry(() => _activationCoordinator.activateAsset(asset));
     final strategy = await _resolvePubkeyStrategy(asset);
@@ -31,6 +79,7 @@ class PubkeyManager {
   }
 
   /// Streamed version of [createNewPubkey]
+  @override
   Stream<NewAddressState> createNewPubkeyStream(Asset asset) async* {
     await retry(() => _activationCoordinator.activateAsset(asset));
     final strategy = await _resolvePubkeyStrategy(asset);
@@ -44,6 +93,7 @@ class PubkeyManager {
   }
 
   /// Unban pubkeys according to [unbanBy] criteria
+  @override
   Future<UnbanPubkeysResult> unbanPubkeys(UnbanBy unbanBy) async {
     final response = await _client.rpc.wallet.unbanPubkeys(unbanBy: unbanBy);
     return response.result;
@@ -57,8 +107,220 @@ class PubkeyManager {
     return asset.pubkeyStrategy(kdfUser: currentUser);
   }
 
+  /// Stream of pubkeys per asset. Polls pubkeys (not balances) and emits updates.
+  /// Emits the initial known state if available.
+  @override
+  Stream<AssetPubkeys> watchPubkeys(
+    Asset asset, {
+    bool activateIfNeeded = true,
+  }) async* {
+    if (_isDisposed) {
+      throw StateError('PubkeyManager has been disposed');
+    }
+
+    // Emit last known pubkeys immediately if available
+    final lastKnown = _pubkeysCache[asset.id];
+    if (lastKnown != null) {
+      yield lastKnown;
+    }
+
+    final controller = _pubkeysControllers.putIfAbsent(
+      asset.id,
+      () => StreamController<AssetPubkeys>.broadcast(
+        onListen: () => _startWatchingPubkeys(asset, activateIfNeeded),
+        onCancel: () => _stopWatchingPubkeys(asset.id),
+      ),
+    );
+
+    yield* controller.stream;
+  }
+
+  @override
+  AssetPubkeys? lastKnown(AssetId assetId) {
+    if (_isDisposed) {
+      throw StateError('PubkeyManager has been disposed');
+    }
+    return _pubkeysCache[assetId];
+  }
+
+  Future<void> _startWatchingPubkeys(Asset asset, bool activateIfNeeded) async {
+    final controller = _pubkeysControllers[asset.id];
+    if (controller == null || _isDisposed) return;
+
+    // Cancel any existing watcher for this asset
+    await _activeWatchers[asset.id]?.cancel();
+    _activeWatchers.remove(asset.id);
+
+    // Ensure user is authenticated
+    final user = await _auth.currentUser;
+    if (user == null) {
+      // Do not emit an error; wait for authentication changes
+      return;
+    }
+    _currentWalletId = user.walletId;
+
+    // Emit last known immediately if available
+    final maybeKnown = _pubkeysCache[asset.id];
+    if (maybeKnown != null && !controller.isClosed) {
+      controller.add(maybeKnown);
+    }
+
+    try {
+      // Ensure activation if requested, otherwise only proceed if already active
+      bool isActive = await _activationCoordinator.isAssetActive(asset.id);
+      if (!isActive && activateIfNeeded) {
+        final activationResult = await _activationCoordinator.activateAsset(
+          asset,
+        );
+        isActive = activationResult.isSuccess;
+      }
+
+      if (isActive) {
+        final first = await getPubkeys(asset);
+        _pubkeysCache[asset.id] = first;
+        if (!controller.isClosed) controller.add(first);
+      }
+
+      // Periodic polling for pubkeys updates
+      final periodicStream = Stream<void>.periodic(_defaultPollingInterval);
+      _activeWatchers[asset.id] = periodicStream
+          .asyncMap<AssetPubkeys?>((_) async {
+            if (_isDisposed) return null;
+
+            // Check that user is still authenticated and wallet hasn't changed
+            final currentUser = await _auth.currentUser;
+            if (currentUser == null ||
+                currentUser.walletId != _currentWalletId) {
+              return null;
+            }
+
+            try {
+              bool active = await _activationCoordinator.isAssetActive(
+                asset.id,
+              );
+              if (!active && activateIfNeeded) {
+                final activationResult = await _activationCoordinator
+                    .activateAsset(asset);
+                active = activationResult.isSuccess;
+              }
+              if (active) {
+                final pubkeys = await getPubkeys(asset);
+                _pubkeysCache[asset.id] = pubkeys;
+                return pubkeys;
+              }
+            } catch (_) {
+              // Swallow transient errors; continue with last known state
+            }
+            return _pubkeysCache[asset.id];
+          })
+          .listen(
+            (AssetPubkeys? pubkeys) {
+              if (pubkeys != null && !controller.isClosed) {
+                controller.add(pubkeys);
+              }
+            },
+            onError: (Object error) {
+              if (!controller.isClosed) controller.addError(error);
+            },
+            onDone: () => _stopWatchingPubkeys(asset.id),
+            cancelOnError: false,
+          );
+    } catch (e) {
+      if (!controller.isClosed) controller.addError(e);
+    }
+  }
+
+  void _stopWatchingPubkeys(AssetId assetId) {
+    final watcher = _activeWatchers[assetId];
+    if (watcher != null) {
+      watcher.cancel();
+      _activeWatchers.remove(assetId);
+    }
+  }
+
+  @override
+  Future<void> preCachePubkeys(Asset asset) async {
+    if (_isDisposed) return;
+
+    final user = await _auth.currentUser;
+    if (user == null) return;
+
+    try {
+      final pubkeys = await getPubkeys(asset);
+      _pubkeysCache[asset.id] = pubkeys;
+
+      final controller = _pubkeysControllers[asset.id];
+      if (controller != null && !controller.isClosed) {
+        controller.add(pubkeys);
+      }
+    } catch (_) {
+      // Fail silently; this is a best-effort cache warm-up
+    }
+  }
+
+  Future<void> _handleAuthStateChanged(KdfUser? user) async {
+    if (_isDisposed) return;
+    final newWalletId = user?.walletId;
+    if (_currentWalletId != newWalletId) {
+      await _resetState();
+      _currentWalletId = newWalletId;
+    }
+  }
+
+  Future<void> _resetState() async {
+    // Cancel all active watchers
+    for (final subscription in _activeWatchers.values) {
+      await subscription.cancel();
+    }
+    _activeWatchers.clear();
+
+    // Wallet changes are normal user operations; do not signal as errors.
+    // If needed, consider using a dedicated event or state indicator.
+    for (final _ in _pubkeysControllers.values) {
+      // No-op: we intentionally keep controllers open so that consumers can
+      // re-listen, which will trigger a fresh watcher start when appropriate.
+      // Optionally, emit a state event here if the stream type supports it.
+    }
+
+    // Clear caches
+    _pubkeysCache.clear();
+
+    // Restart watchers for existing controllers
+    final existingControllers =
+        Map<AssetId, StreamController<AssetPubkeys>>.from(_pubkeysControllers);
+    for (final entry in existingControllers.entries) {
+      if (!entry.value.isClosed) {
+        // Asset recreation from AssetId is intentionally not performed here during state reset.
+        // This is because reconstructing an Asset instance from just its AssetId is non-trivial:
+        // it may require additional metadata or context that is not available at this point,
+        // and could depend on external state or configuration. Instead, we rely on callers to
+        // re-listen, which will trigger a restart with a fully constructed Asset. This approach
+        // avoids potential errors and ensures that watchers are started with complete information.
+        // We simply keep controllers; new listeners will call _startWatchingPubkeys as needed.
+      }
+    }
+  }
+
   /// Dispose of any resources
+  @override
   Future<void> dispose() async {
-    // No cleanup needed currently
+    if (_isDisposed) return;
+    _isDisposed = true;
+
+    await _authSubscription?.cancel();
+    _authSubscription = null;
+
+    for (final subscription in _activeWatchers.values) {
+      await subscription.cancel();
+    }
+    _activeWatchers.clear();
+
+    for (final controller in _pubkeysControllers.values) {
+      await controller.close();
+    }
+    _pubkeysControllers.clear();
+
+    _pubkeysCache.clear();
+    _currentWalletId = null;
   }
 }

packages/komodo_defi_sdk/pubspec.yaml

@@ -45,6 +45,7 @@ dependencies:
   logging: any
 dev_dependencies:
   index_generator: ^4.0.1
+  fake_async: ^1.3.3
   mocktail: ^1.0.4
   # test: ^1.25.7
   test: ^1.25.7