All checks were successful
ci/woodpecker/push/billing_fees Pipeline was successful
ci/woodpecker/push/db Pipeline was successful
ci/woodpecker/push/bff Pipeline was successful
ci/woodpecker/push/discovery Pipeline was successful
ci/woodpecker/push/fx_ingestor Pipeline was successful
ci/woodpecker/push/fx_oracle Pipeline was successful
ci/woodpecker/push/frontend Pipeline was successful
ci/woodpecker/push/gateway_mntx Pipeline was successful
ci/woodpecker/push/gateway_chain Pipeline was successful
ci/woodpecker/push/gateway_tgsettle Pipeline was successful
ci/woodpecker/push/nats Pipeline was successful
ci/woodpecker/push/ledger Pipeline was successful
ci/woodpecker/push/notification Pipeline was successful
ci/woodpecker/push/payments_orchestrator Pipeline was successful
255 lines
7.0 KiB
Dart
255 lines
7.0 KiB
Dart
import 'dart:async';
|
|
import 'dart:math';
|
|
|
|
import 'package:flutter/foundation.dart';
|
|
|
|
import 'package:collection/collection.dart';
|
|
|
|
import 'package:pshared/models/describable.dart';
|
|
import 'package:pshared/models/payment/wallet.dart';
|
|
import 'package:pshared/models/wallet/chain_asset.dart';
|
|
import 'package:pshared/provider/organizations.dart';
|
|
import 'package:pshared/provider/resource.dart';
|
|
import 'package:pshared/service/payment/wallets.dart';
|
|
import 'package:pshared/utils/exception.dart';
|
|
|
|
|
|
class WalletsProvider with ChangeNotifier {
|
|
final WalletsService _service;
|
|
OrganizationsProvider? _organizations;
|
|
|
|
WalletsProvider(this._service);
|
|
|
|
Resource<List<Wallet>> _resource = Resource(data: []);
|
|
Resource<List<Wallet>> get resource => _resource;
|
|
|
|
List<Wallet> get wallets => _resource.data ?? [];
|
|
bool get isLoading => _resource.isLoading;
|
|
Exception? get error => _resource.error;
|
|
|
|
bool _isRefreshingBalances = false;
|
|
bool get isRefreshingBalances => _isRefreshingBalances;
|
|
|
|
final Set<String> _refreshingWallets = <String>{};
|
|
bool isWalletRefreshing(String walletRef) => _refreshingWallets.contains(walletRef);
|
|
|
|
// Expose current org id so UI controller can reset per-org state if needed.
|
|
String? get organizationRef =>
|
|
(_organizations?.isOrganizationSet ?? false) ? _organizations!.current.id : null;
|
|
|
|
// Used to ignore stale async results (org changes / overlapping requests).
|
|
int _opSeq = 0;
|
|
|
|
// Per-wallet refresh sequence guard.
|
|
final Map<String, int> _walletSeq = <String, int>{};
|
|
|
|
// Keep modest concurrency to avoid hammering the backend.
|
|
static const int _balanceConcurrency = 6;
|
|
|
|
void update(OrganizationsProvider organizations) {
|
|
_organizations = organizations;
|
|
if (organizations.isOrganizationSet) {
|
|
unawaited(loadWalletsWithBalances());
|
|
}
|
|
}
|
|
|
|
Future<Wallet> updateWallet(Wallet newWallet) {
|
|
throw Exception('update wallet is not implemented');
|
|
}
|
|
|
|
Future<void> loadWalletsWithBalances() async {
|
|
final org = _organizations;
|
|
if (org == null || !org.isOrganizationSet) return;
|
|
|
|
final orgId = org.current.id;
|
|
final seq = ++_opSeq;
|
|
|
|
_isRefreshingBalances = false;
|
|
_refreshingWallets.clear();
|
|
|
|
_applyResource(_resource.copyWith(isLoading: true, error: null), notify: true);
|
|
|
|
try {
|
|
final base = await _service.getWallets(orgId);
|
|
|
|
final result = await _withBalances(orgId, base);
|
|
if (seq != _opSeq) return;
|
|
|
|
_applyResource(
|
|
Resource<List<Wallet>>(
|
|
data: result.wallets,
|
|
isLoading: false,
|
|
error: result.error,
|
|
),
|
|
notify: true,
|
|
);
|
|
} catch (e) {
|
|
if (seq != _opSeq) return;
|
|
|
|
_applyResource(
|
|
_resource.copyWith(isLoading: false, error: toException(e)),
|
|
notify: true,
|
|
);
|
|
}
|
|
}
|
|
|
|
Future<void> refreshBalances() async {
|
|
final org = _organizations;
|
|
if (org == null || !org.isOrganizationSet) return;
|
|
if (wallets.isEmpty) return;
|
|
|
|
final orgRef = org.current.id;
|
|
final seq = ++_opSeq;
|
|
|
|
_isRefreshingBalances = true;
|
|
_applyResource(_resource.copyWith(error: null), notify: false);
|
|
notifyListeners();
|
|
|
|
try {
|
|
final result = await _withBalances(orgRef, wallets);
|
|
if (seq != _opSeq) return;
|
|
|
|
_applyResource(
|
|
_resource.copyWith(data: result.wallets, error: result.error),
|
|
notify: false,
|
|
);
|
|
} catch (e) {
|
|
if (seq != _opSeq) return;
|
|
|
|
_applyResource(_resource.copyWith(error: toException(e)), notify: false);
|
|
} finally {
|
|
if (seq == _opSeq) {
|
|
_isRefreshingBalances = false;
|
|
notifyListeners();
|
|
}
|
|
}
|
|
}
|
|
|
|
Future<void> refreshBalance(String walletRef) async {
|
|
final org = _organizations;
|
|
if (org == null || !org.isOrganizationSet) return;
|
|
|
|
if (_refreshingWallets.contains(walletRef)) return;
|
|
|
|
final existing = wallets.firstWhereOrNull((w) => w.id == walletRef);
|
|
if (existing == null) return;
|
|
|
|
final orgId = org.current.id;
|
|
final seq = (_walletSeq[walletRef] ?? 0) + 1;
|
|
_walletSeq[walletRef] = seq;
|
|
|
|
_refreshingWallets.add(walletRef);
|
|
notifyListeners();
|
|
|
|
try {
|
|
final balance = await _service.getBalance(orgId, walletRef);
|
|
if ((_walletSeq[walletRef] ?? 0) != seq) return;
|
|
|
|
final next = _replaceWallet(walletRef, (w) => w.copyWith(balance: balance));
|
|
if (next == null) return;
|
|
|
|
_applyResource(_resource.copyWith(data: next), notify: false);
|
|
} catch (e) {
|
|
if ((_walletSeq[walletRef] ?? 0) != seq) return;
|
|
|
|
_applyResource(_resource.copyWith(error: toException(e)), notify: false);
|
|
} finally {
|
|
if ((_walletSeq[walletRef] ?? 0) == seq) {
|
|
_refreshingWallets.remove(walletRef);
|
|
notifyListeners();
|
|
}
|
|
}
|
|
}
|
|
|
|
Future<void> create({
|
|
required Describable describable,
|
|
required ChainAsset asset,
|
|
required String? ownerRef,
|
|
}) async {
|
|
final org = _organizations;
|
|
if (org == null || !org.isOrganizationSet) return;
|
|
|
|
_applyResource(_resource.copyWith(isLoading: true, error: null), notify: true);
|
|
|
|
try {
|
|
await _service.create(
|
|
organizationRef: org.current.id,
|
|
describable: describable,
|
|
asset: asset,
|
|
ownerRef: ownerRef,
|
|
);
|
|
await loadWalletsWithBalances();
|
|
} catch (e) {
|
|
_applyResource(_resource.copyWith(isLoading: false, error: toException(e)), notify: true);
|
|
rethrow;
|
|
}
|
|
}
|
|
|
|
// ---------- internals ----------
|
|
|
|
void _applyResource(Resource<List<Wallet>> newResource, {required bool notify}) {
|
|
_resource = newResource;
|
|
if (notify) notifyListeners();
|
|
}
|
|
|
|
List<Wallet>? _replaceWallet(String walletRef, Wallet Function(Wallet) updater) {
|
|
final idx = wallets.indexWhere((w) => w.id == walletRef);
|
|
if (idx < 0) return null;
|
|
|
|
final next = List<Wallet>.from(wallets);
|
|
next[idx] = updater(next[idx]);
|
|
return next;
|
|
}
|
|
|
|
Future<_WalletLoadResult> _withBalances(String orgRef, List<Wallet> base) async {
|
|
Exception? firstError;
|
|
|
|
final withBalances = await _mapConcurrent<Wallet, Wallet>(
|
|
base,
|
|
_balanceConcurrency,
|
|
(wallet) async {
|
|
try {
|
|
final balance = await _service.getBalance(orgRef, wallet.id);
|
|
return wallet.copyWith(balance: balance);
|
|
} catch (e) {
|
|
firstError ??= toException(e);
|
|
return wallet;
|
|
}
|
|
},
|
|
);
|
|
|
|
return _WalletLoadResult(withBalances, firstError);
|
|
}
|
|
|
|
static Future<List<R>> _mapConcurrent<T, R>(
|
|
List<T> items,
|
|
int concurrency,
|
|
Future<R> Function(T) fn,
|
|
) async {
|
|
if (items.isEmpty) return <R>[];
|
|
|
|
final results = List<R?>.filled(items.length, null);
|
|
var nextIndex = 0;
|
|
|
|
Future<void> worker() async {
|
|
while (true) {
|
|
final i = nextIndex++;
|
|
if (i >= items.length) return;
|
|
results[i] = await fn(items[i]);
|
|
}
|
|
}
|
|
|
|
final workers = List.generate(min(concurrency, items.length), (_) => worker());
|
|
await Future.wait(workers);
|
|
|
|
return results.cast<R>();
|
|
}
|
|
}
|
|
|
|
class _WalletLoadResult {
|
|
final List<Wallet> wallets;
|
|
final Exception? error;
|
|
|
|
const _WalletLoadResult(this.wallets, this.error);
|
|
}
|