refactor: 统一云服务会话管理与生命周期

lib/cloud/cloud_session_manager.dart

@@ -0,0 +1,143 @@
+import 'dart:async';
+import 'dart:convert';
+
+import 'package:flutter_cloud_sync/flutter_cloud_sync.dart';
+
+typedef CloudServices = ({CloudProvider? provider, CloudAuthService? auth});
+typedef CloudServicesFactory = Future<CloudServices> Function(
+    CloudServiceConfig);
+
+/// Owns one active configuration's services. The factory remains uncached.
+class CloudSession {
+  CloudSession(this.config, this.services);
+
+  final CloudServiceConfig config;
+  final CloudServices services;
+  final Set<void Function()> _onClose = {};
+  Future<void>? _closing;
+  bool get isClosed => _closing != null;
+
+  /// Consumers stop scheduling work before their network resources are released.
+  void Function() onClose(void Function() callback) {
+    if (isClosed) {
+      callback();
+    } else {
+      _onClose.add(callback);
+    }
+    return () => _onClose.remove(callback);
+  }
+
+  Future<void> close() {
+    if (_closing != null) return _closing!;
+    final done = Completer<void>();
+    _closing = done.future;
+    Future<void>(() async {
+      Object? firstError;
+      StackTrace? firstStack;
+      try {
+        for (final callback in _onClose.toList()) {
+          try {
+            callback();
+          } catch (error, stack) {
+            firstError ??= error;
+            firstStack ??= stack;
+          }
+        }
+      } finally {
+        _onClose.clear();
+        await services.provider?.dispose();
+      }
+      if (firstError != null) {
+        Error.throwWithStackTrace(firstError!, firstStack!);
+      }
+    }).then(done.complete, onError: done.completeError);
+    return done.future;
+  }
+}
+
+class CloudSessionManager {
+  CloudSessionManager(
+      {CloudServicesFactory factory = createCloudServices,
+      CloudServicesFactory? temporaryFactory})
+      : _factory = factory,
+        _temporaryFactory = temporaryFactory ??
+            ((config) => createCloudServices(config, persistSession: false));
+
+  final CloudServicesFactory _factory;
+  final CloudServicesFactory _temporaryFactory;
+  CloudSession? _active;
+  Future<void> _tail = Future.value();
+  Future<CloudSession>? _pending;
+  String? _requestedKey;
+  int _generation = 0;
+  bool _disposed = false;
+
+  Future<CloudSession> activate(CloudServiceConfig config) {
+    if (_disposed) return Future.error(StateError('Session manager closed'));
+    // Includes credentials: editing credentials must replace the session too.
+    // Never log this key, since configuration can contain passwords.
+    final key = jsonEncode(config.toJson());
+    if (_requestedKey == key && _pending != null) return _pending!;
+    _requestedKey = key;
+    final generation = ++_generation;
+    final next = _tail.then((_) async {
+      _checkGeneration(generation);
+      final previous = _active;
+      _active = null;
+      await previous?.close();
+      _checkGeneration(generation);
+      final services = await _factory(config);
+      final session = CloudSession(config, services);
+      if (_disposed || generation != _generation) {
+        await session.close();
+        throw StateError('Cloud configuration changed during initialization');
+      }
+      // Backend-specific recovery belongs in session assembly, not UI providers.
+      final auth = services.auth;
+      if (auth is BeeCountCloudAuthService) {
+        auth.setRecoveryCredentials(
+          email: config.beecountCloudEmail,
+          password: config.beecountCloudPassword,
+        );
+      }
+      _active = session;
+      return session;
+    });
+    _pending = next;
+    _tail = next.then<void>((_) {}, onError: (Object error, StackTrace stack) {
+      if (generation == _generation) {
+        _requestedKey = null;
+        _pending = null;
+      }
+    });
+    return next;
+  }
+
+  void _checkGeneration(int generation) {
+    if (_disposed || generation != _generation) {
+      throw StateError('Cloud configuration changed');
+    }
+  }
+
+  /// Draft configurations never replace the active session. Always released.
+  Future<T> withTemporarySession<T>(CloudServiceConfig config,
+      Future<T> Function(CloudSession) action) async {
+    if (_disposed) throw StateError('Session manager closed');
+    final session = CloudSession(config, await _temporaryFactory(config));
+    try {
+      if (_disposed) throw StateError('Session manager closed');
+      return await action(session);
+    } finally {
+      await session.close();
+    }
+  }
+
+  Future<void> dispose() async {
+    _disposed = true;
+    ++_generation;
+    await _tail;
+    final previous = _active;
+    _active = null;
+    await previous?.close();
+  }
+}

lib/cloud/sync/sync_engine.dart

@@ -78,6 +78,7 @@ class SyncEngine implements app.SyncService {
   /// 状态缓存
   final Map<int, app.SyncStatus> _statusCache = {};
   bool _localChanged = false;
+  bool _disposed = false;
 
   /// WebSocket 实时监听
   StreamSubscription<BeeCountCloudRealtimeEvent>? _realtimeSubscription;
@@ -361,6 +362,9 @@ class SyncEngine implements app.SyncService {
 
   /// 释放资源
   void dispose() {
+    if (_disposed) return;
+    _disposed = true;
+    ledgerIdResolver = null;
     stopListeningRealtime();
     _eventsController.close();
   }
@@ -369,6 +373,7 @@ class SyncEngine implements app.SyncService {
 
   /// 执行完整同步（先 push 后 pull）
   Future<SyncResult> sync({required String ledgerId}) async {
+    if (_disposed) return const SyncResult(error: 'Cloud session closed');
     logger.info('SyncEngine', '开始同步 ledger=$ledgerId');
     try {
       final ledgerIdInt = int.tryParse(ledgerId) ?? -1;
@@ -532,14 +537,16 @@ class SyncEngine implements app.SyncService {
   /// 此时设备全局 cursor 可能已经前移、普通 `_pull` 再也拉不回历史。
   ///
   /// 返回新增（非已存在）的账本数，调用方可据此决定要不要 bump 刷新信号。
-  /// 并发互斥锁 — **static** 跨 SyncEngine 实例共享。
-  /// 关键 bug:join page 拿 syncEngineProvider(family) 的 engine,WS listener
-  /// 拿 cloudSyncServiceProvider 创建的 engine,两个不同 instance!instance-level
-  /// 字段互不知道,各跑各的。改 static 后整个进程同一时间只有一个 fetch-then-write
-  /// 在跑。
-  static Completer<int>? _syncLedgersInFlight;
+  /// 同一会话共用一个 engine；切换会话不能复用旧服务器的 in-flight 结果。
+  Completer<int>? _syncLedgersInFlight;
+
+  // Different engines using the same database must not interleave ledger
+  // lookup/insert/GC. Queue work, not results: a new session must fetch its own
+  // server's ledger list after the previous operation has finished.
+  static final _ledgerSyncTails = Expando<Future<void>>();
 
   Future<int> syncLedgersFromServer() async {
+    if (_disposed) throw StateError('Cloud session closed');
     final existing = _syncLedgersInFlight;
     if (existing != null) {
       logger.info('SyncEngine', 'syncLedgersFromServer 已在执行中,等待 in-flight 结果');
@@ -548,7 +555,14 @@ class SyncEngine implements app.SyncService {
     final completer = Completer<int>();
     _syncLedgersInFlight = completer;
     try {
-      final n = await _syncLedgersFromServerLocked();
+      final previous = _ledgerSyncTails[db] ?? Future<void>.value();
+      final work = previous.then((_) async {
+        if (_disposed) return 0;
+        return _syncLedgersFromServerLocked();
+      });
+      _ledgerSyncTails[db] = work.then<void>((_) {},
+          onError: (Object error, StackTrace stack) {});
+      final n = await work;
       completer.complete(n);
       return n;
     } catch (e, st) {
@@ -563,6 +577,7 @@ class SyncEngine implements app.SyncService {
     logger.info('SyncEngine', 'syncLedgersFromServer start');
     try {
       final remote = await provider.readLedgers();
+      if (_disposed) return 0;
       int upserted = 0;
       int inserted = 0;
       // 新设备登录场景:Editor 已是 server LedgerMember 但本地 ledgers 表为空。
@@ -572,6 +587,7 @@ class SyncEngine implements app.SyncService {
       // 也会让单个失败影响其它账本。
       final newSharedLedgerSyncIds = <String>[];
       for (final r in remote) {
+        if (_disposed) return inserted;
         final syncId = r.ledgerId;
         if (syncId.isEmpty) continue;
         // 用 get() 不用 getSingleOrNull() — 历史可能已经产生过同 syncId 多行
@@ -711,6 +727,7 @@ class SyncEngine implements app.SyncService {
   /// 否则单飞失效。这俩内部应该只处理 ledger-scope change(transaction / budget /
   /// ledger / ledger_snapshot)。
   Future<int> pushUserGlobalEntities() async {
+    if (_disposed) throw StateError('Cloud session closed');
     final inFlight = _userGlobalPushInFlight;
     if (inFlight != null) {
       logger.info('SyncEngine', 'pushUserGlobalEntities 已在执行,复用 in-flight');
@@ -904,6 +921,7 @@ class SyncEngine implements app.SyncService {
   /// user-global change(account / category / tag)由 [pushUserGlobalEntities] 统一推
   /// (在 [_doPush] 开头调用),避免多账本场景下并行 push 重复推送 user-global。
   Future<int> push(String ledgerId) async {
+    if (_disposed) throw StateError('Cloud session closed');
     final inFlight = _pushInFlight[ledgerId];
     if (inFlight != null) {
       logger.info('SyncEngine', 'push(ledger=$ledgerId) 已在执行,复用 in-flight');
@@ -1062,6 +1080,7 @@ class SyncEngine implements app.SyncService {
   ///   轮)
   /// - replay(sinceOverride 非 null)语义独立,等 in-flight 完成后再自己跑
   Future<int> pull(String ledgerId, {int? sinceOverride}) async {
+    if (_disposed) throw StateError('Cloud session closed');
     // 1. in-flight 单飞
     final inFlight = _pullInFlight;
     if (inFlight != null) {
@@ -1214,13 +1233,15 @@ class SyncEngine implements app.SyncService {
   /// - SQLite busy/locked → 单条 retry 2 次
   Future<_PullPageOutcome> _applyPullPage(
       List<BeeCountCloudSyncChange> changes) async {
+    if (_disposed) throw StateError('Cloud session closed');
     int applied = 0;
     int skipped = 0;
     BeeCountCloudSyncChange? failingChange;
 
     try {
       await db.transaction(() async {
         for (final ch in changes) {
+          if (_disposed) throw StateError('Cloud session closed');
           failingChange = ch;
           final ok = await _applyOneWithBusyRetry(ch);
           if (ok) {

lib/cloud/sync/sync_engine_profile.dart

@@ -15,12 +15,14 @@ extension SyncEngineProfile on SyncEngine {
   /// PR 3:不再接 callback,所有字段更新走 [events] stream emit
   /// `ProfileFieldApplied` 事件,UI 通过 syncEventStreamProvider 订阅处理。
   Future<bool> syncMyProfile() async {
+    if (_disposed) return false;
     final localVersion = await AvatarService.getStoredRemoteVersion();
     logger.info('avatar_sync',
         'syncMyProfile start, localVersion=$localVersion');
     bool anyChanged = false;
     try {
       final profile = await provider.getMyProfile();
+      if (_disposed) return false;
 
       // === theme_primary_color ===
       final theme = profile.themePrimaryColor;
@@ -87,6 +89,7 @@ extension SyncEngineProfile on SyncEngine {
         userId: profile.userId,
         version: remoteVersion > 0 ? remoteVersion : null,
       );
+      if (_disposed) return false;
       logger.info('avatar_sync', 'downloaded size=${bytes.length}B');
       await AvatarService.saveAvatarFromBytes(bytes);
       await AvatarService.setStoredRemoteVersion(remoteVersion);

lib/cloud/sync/sync_engine_realtime.dart

@@ -8,6 +8,7 @@ part of 'sync_engine.dart';
 extension SyncEngineRealtime on SyncEngine {
   /// 开始监听 WebSocket 实时事件，收到变更通知时自动触发 pull
   void startListeningRealtime() {
+    if (_disposed) return;
     _realtimeSubscription?.cancel();
     // 启动 WebSocket 连接，否则 realtimeEvents 流永远为空
     provider.startRealtime().catchError((e) {
@@ -66,6 +67,7 @@ extension SyncEngineRealtime on SyncEngine {
   /// 2 秒防抖：WiFi ↔ 移动网络切换、或 WS reconnect 接着 connectivity 事件
   /// 这种"连续上线信号"只触发 1 次 sync。
   void _scheduleAutoSync({required String reason}) {
+    if (_disposed) return;
     _autoSyncDebounce?.cancel();
     _autoSyncDebounce = Timer(const Duration(seconds: 2), () async {
       if (_autoSyncing) {
@@ -617,6 +619,7 @@ extension SyncEngineRealtime on SyncEngine {
 
   /// 防抖调度 pull（1 秒内多次触发只执行一次）
   void _schedulePull(String? ledgerId) {
+    if (_disposed) return;
     _pullDebounce?.cancel();
     _pullDebounce = Timer(const Duration(seconds: 1), () async {
       if (_autoPulling) return;

lib/cloud/sync/sync_providers.dart

@@ -3,6 +3,7 @@ import 'package:flutter_cloud_sync/flutter_cloud_sync.dart'
     hide SyncStatus;
 
 import '../../providers/database_providers.dart';
+import '../../providers/sync_providers.dart' show activeCloudServicesProvider;
 import 'change_tracker.dart';
 import 'sync_engine.dart';
 
@@ -23,6 +24,13 @@ final changeTrackerProvider = Provider<ChangeTracker>((ref) {
 /// 装配 callback 后才启动,但 dispose 由 Riverpod GC family entry 时统一触发。
 final syncEngineProvider = Provider.family<SyncEngine, BeeCountCloudProvider>(
   (ref, provider) {
+    // Loading with the same previous value is not a new session.
+    final session = ref.watch(activeCloudServicesProvider.select(
+        (value) => value.valueOrNull));
+    if (session == null || session.isClosed ||
+        !identical(session.services.provider, provider)) {
+      throw StateError('SyncEngine requires the active cloud session');
+    }
     final db = ref.watch(databaseProvider);
     final tracker = ref.watch(changeTrackerProvider);
     final repo = ref.watch(repositoryProvider);
@@ -32,7 +40,11 @@ final syncEngineProvider = Provider.family<SyncEngine, BeeCountCloudProvider>(
       changeTracker: tracker,
       repo: repo,
     );
-    ref.onDispose(() => engine.dispose());
+    final unregister = session.onClose(engine.dispose);
+    ref.onDispose(() {
+      unregister();
+      engine.dispose();
+    });
     return engine;
   },
 );

lib/cloud/transactions_sync_manager.dart

@@ -11,6 +11,7 @@ import '../services/data_import_service.dart';
 import '../services/system/logger_service.dart';
 import 'sync_diff_service.dart';
 import 'sync_service.dart';
+import 'cloud_session_manager.dart';
 import 'transactions_json.dart';
 
 /// 账本交易的云同步管理器
@@ -20,6 +21,7 @@ class TransactionsSyncManager implements SyncService {
   final fcs.CloudServiceConfig config;
   final BeeDatabase db;
   final BaseRepository repo;
+  final CloudSession session;
 
   fcs.CloudSyncManager<int>? _syncManager;
   fcs.CloudProvider? _provider;
@@ -34,6 +36,7 @@ class TransactionsSyncManager implements SyncService {
     required this.config,
     required this.db,
     required this.repo,
+    required this.session,
   });
 
   @override
@@ -47,6 +50,7 @@ class TransactionsSyncManager implements SyncService {
 
   /// 确保服务已初始化（延迟初始化）
   Future<void> _ensureInitialized() async {
+    if (session.isClosed) throw StateError('Cloud session closed');
     if (_isInitialized) return;
     if (_isInitializing) {
       // 等待初始化完成
@@ -67,7 +71,7 @@ class TransactionsSyncManager implements SyncService {
 
   /// 初始化 CloudProvider 和 SyncManager
   Future<void> _initialize() async {
-    final services = await fcs.createCloudServices(config);
+    final services = session.services;
     _provider = services.provider;
 
     if (_provider == null) {