Add sync queue orchestration and local mutation envelopes

2026-02-15 16:56:55 +01:00
parent aaabc51d2d
commit d711f270dd
8 changed files with 1411 additions and 101 deletions
@@ -0,0 +1,179 @@
+import 'package:flutter/foundation.dart';
+import 'package:flutter_riverpod/flutter_riverpod.dart';
+import 'package:relationship_saver/features/local/local_repository.dart';
+import 'package:relationship_saver/features/sync/sync_queue_repository.dart';
+import 'package:relationship_saver/features/sync/sync_state.dart';
+import 'package:relationship_saver/integrations/backend/backend_gateway_provider.dart';
+
+/// Result details for one sync action.
+@immutable
+class SyncRunResult {
+  const SyncRunResult({
+    required this.success,
+    required this.operation,
+    required this.message,
+    required this.pendingAfter,
+    this.cursor,
+    this.accepted = 0,
+    this.rejected = 0,
+    this.pulled = 0,
+    this.error,
+  });
+
+  final bool success;
+  final String operation;
+  final String message;
+  final int pendingAfter;
+  final String? cursor;
+  final int accepted;
+  final int rejected;
+  final int pulled;
+  final Object? error;
+}
+
+/// Coordinates push/pull sync operations across queue, gateway, and local store.
+class SyncCoordinator {
+  const SyncCoordinator(this._ref);
+
+  final Ref _ref;
+
+  Future<SyncRunResult> pushPending() async {
+    final SyncState queueState = await _ref.read(
+      syncQueueRepositoryProvider.future,
+    );
+    if (queueState.pendingChanges.isEmpty) {
+      return SyncRunResult(
+        success: true,
+        operation: 'push',
+        message: 'No pending local changes to push.',
+        pendingAfter: 0,
+        cursor: queueState.cursor,
+      );
+    }
+
+    final DateTime now = DateTime.now();
+    final syncQueue = _ref.read(syncQueueRepositoryProvider.notifier);
+    try {
+      final result = await _ref
+          .read(backendGatewayProvider)
+          .pushChanges(
+            changes: queueState.pendingChanges,
+            cursor: queueState.cursor,
+          );
+      await syncQueue.applyPushResult(result, at: now);
+      final SyncState after = await _ref.read(
+        syncQueueRepositoryProvider.future,
+      );
+      return SyncRunResult(
+        success: true,
+        operation: 'push',
+        message:
+            'Push complete. accepted=${result.accepted.length}, rejected=${result.rejected.length}.',
+        accepted: result.accepted.length,
+        rejected: result.rejected.length,
+        pendingAfter: after.pendingChanges.length,
+        cursor: after.cursor,
+      );
+    } catch (error) {
+      await syncQueue.markFailure(error, at: now);
+      final SyncState after = await _ref.read(
+        syncQueueRepositoryProvider.future,
+      );
+      return SyncRunResult(
+        success: false,
+        operation: 'push',
+        message: 'Push failed. Local changes stay queued.',
+        pendingAfter: after.pendingChanges.length,
+        cursor: after.cursor,
+        error: error,
+      );
+    }
+  }
+
+  Future<SyncRunResult> pullRemote() async {
+    final SyncState queueState = await _ref.read(
+      syncQueueRepositoryProvider.future,
+    );
+    final DateTime now = DateTime.now();
+    final syncQueue = _ref.read(syncQueueRepositoryProvider.notifier);
+
+    try {
+      final result = await _ref
+          .read(backendGatewayProvider)
+          .pullChanges(cursor: queueState.cursor);
+      await _ref
+          .read(localRepositoryProvider.notifier)
+          .applyRemoteChanges(result.changes);
+      await syncQueue.applyPullResult(result, at: now);
+      final SyncState after = await _ref.read(
+        syncQueueRepositoryProvider.future,
+      );
+      return SyncRunResult(
+        success: true,
+        operation: 'pull',
+        message: 'Pulled ${result.changes.length} change(s) from backend.',
+        pulled: result.changes.length,
+        pendingAfter: after.pendingChanges.length,
+        cursor: after.cursor,
+      );
+    } catch (error) {
+      await syncQueue.markFailure(error, at: now);
+      final SyncState after = await _ref.read(
+        syncQueueRepositoryProvider.future,
+      );
+      return SyncRunResult(
+        success: false,
+        operation: 'pull',
+        message: 'Pull failed. Staying in local-first mode.',
+        pendingAfter: after.pendingChanges.length,
+        cursor: after.cursor,
+        error: error,
+      );
+    }
+  }
+
+  Future<SyncRunResult> syncNow() async {
+    final SyncRunResult pushResult = await pushPending();
+    if (!pushResult.success) {
+      return SyncRunResult(
+        success: false,
+        operation: 'sync',
+        message: 'Sync paused at push step. ${pushResult.message}',
+        pendingAfter: pushResult.pendingAfter,
+        cursor: pushResult.cursor,
+        accepted: pushResult.accepted,
+        rejected: pushResult.rejected,
+        error: pushResult.error,
+      );
+    }
+
+    final SyncRunResult pullResult = await pullRemote();
+    if (!pullResult.success) {
+      return SyncRunResult(
+        success: false,
+        operation: 'sync',
+        message: 'Sync paused at pull step. ${pullResult.message}',
+        pendingAfter: pullResult.pendingAfter,
+        cursor: pullResult.cursor,
+        accepted: pushResult.accepted,
+        rejected: pushResult.rejected,
+        error: pullResult.error,
+      );
+    }
+
+    return SyncRunResult(
+      success: true,
+      operation: 'sync',
+      message:
+          'Sync complete. accepted=${pushResult.accepted}, rejected=${pushResult.rejected}, pulled=${pullResult.pulled}.',
+      pendingAfter: pullResult.pendingAfter,
+      cursor: pullResult.cursor,
+      accepted: pushResult.accepted,
+      rejected: pushResult.rejected,
+      pulled: pullResult.pulled,
+    );
+  }
+}
+
+final Provider<SyncCoordinator> syncCoordinatorProvider =
+    Provider<SyncCoordinator>(SyncCoordinator.new);