395 lines
15 KiB
Dart
395 lines
15 KiB
Dart
import 'dart:async';
|
|
import 'package:flutter/widgets.dart';
|
|
import 'sync_models.dart';
|
|
import 'sync_outbox.dart';
|
|
import 'sync_provider.dart';
|
|
import '../../database/database.dart';
|
|
import '../auth/auth_service.dart';
|
|
|
|
/// Diagnostic summary returned by [SyncManager.reconcileCompletedRides].
|
|
class CompletedRideReconciliationReport {
|
|
final int discoveredCompletedRides;
|
|
final int alreadyTrackedCount;
|
|
final int recoveredQueuedCount;
|
|
final int recoveredAuthRequiredCount;
|
|
final int recoveredAlreadySyncedCount;
|
|
final int foreignOwnerSkippedCount;
|
|
final int errorCount;
|
|
final List<String> errors;
|
|
|
|
const CompletedRideReconciliationReport({
|
|
this.discoveredCompletedRides = 0,
|
|
this.alreadyTrackedCount = 0,
|
|
this.recoveredQueuedCount = 0,
|
|
this.recoveredAuthRequiredCount = 0,
|
|
this.recoveredAlreadySyncedCount = 0,
|
|
this.foreignOwnerSkippedCount = 0,
|
|
this.errorCount = 0,
|
|
this.errors = const [],
|
|
});
|
|
|
|
int get totalRecovered =>
|
|
recoveredQueuedCount +
|
|
recoveredAuthRequiredCount +
|
|
recoveredAlreadySyncedCount;
|
|
|
|
@override
|
|
String toString() =>
|
|
'CompletedRideReconciliationReport(discovered: $discoveredCompletedRides, '
|
|
'tracked: $alreadyTrackedCount, queued: $recoveredQueuedCount, '
|
|
'authRequired: $recoveredAuthRequiredCount, alreadySynced: $recoveredAlreadySyncedCount, '
|
|
'foreignSkipped: $foreignOwnerSkippedCount, errors: $errorCount)';
|
|
}
|
|
|
|
/// Central provider-agnostic synchronization coordinator.
|
|
///
|
|
/// Sync work is persisted in [SyncOutbox]. The manager reconciles persisted
|
|
/// work on startup and whenever the app returns to the foreground. Network
|
|
/// failures remain queued instead of being discarded.
|
|
class SyncManager extends ChangeNotifier with WidgetsBindingObserver {
|
|
final SyncOutbox _outbox;
|
|
final Map<String, SyncProvider> _providers = {};
|
|
String _activeProviderId = 'fake';
|
|
bool _isProcessing = false;
|
|
Future<CompletedRideReconciliationReport>? _activeReconciliation;
|
|
Future<void>? _currentProcessingFuture;
|
|
bool _needsAnotherPass = false;
|
|
DateTime? Function() _nowProvider;
|
|
final AppDatabase? _db;
|
|
final AuthService? _authService;
|
|
|
|
SyncManager(this._outbox, {List<SyncProvider>? initialProviders, String? defaultProviderId, DateTime? Function()? nowProvider, this._db, this._authService, bool autoStart = true}) : _nowProvider = nowProvider ?? (() => DateTime.now()) {
|
|
if (initialProviders != null) {
|
|
for (final p in initialProviders) { registerProvider(p); }
|
|
}
|
|
if (defaultProviderId != null) _activeProviderId = defaultProviderId;
|
|
WidgetsBinding.instance.addObserver(this);
|
|
if (autoStart) {
|
|
unawaited(_startupRecoveryAndProcess());
|
|
}
|
|
}
|
|
|
|
void setNowProvider(DateTime? Function() provider) => _nowProvider = provider;
|
|
DateTime _now() => _nowProvider() ?? DateTime.now();
|
|
String get activeProviderId => _activeProviderId;
|
|
set activeProviderId(String id) { if (_activeProviderId != id) { _activeProviderId = id; notifyListeners(); } }
|
|
bool get isProcessing => _isProcessing;
|
|
int get outboxCount => _outbox.count;
|
|
List<SyncItem> get allItems => _outbox.getAllItems();
|
|
List<SyncItem> get pendingItems => _outbox.getPendingItems();
|
|
|
|
SyncManagerState get currentState {
|
|
int pending = 0, synced = 0, failed = 0;
|
|
DateTime? lastSuccess;
|
|
String? lastErr;
|
|
for (final item in _outbox.getAllItems()) {
|
|
if (item.status == SyncStatus.synced) {
|
|
synced++;
|
|
if (item.uploadedAt != null && (lastSuccess == null || item.uploadedAt!.isAfter(lastSuccess))) lastSuccess = item.uploadedAt;
|
|
} else if (item.status == SyncStatus.failed || item.status == SyncStatus.cancelled) {
|
|
failed++;
|
|
if (item.lastError != null) lastErr = item.lastError;
|
|
} else {
|
|
pending++;
|
|
}
|
|
}
|
|
return SyncManagerState(isProcessing: _isProcessing, pendingCount: pending, syncedCount: synced, failedCount: failed, lastSuccessfulSync: lastSuccess, lastError: lastErr, activeProviderId: _activeProviderId);
|
|
}
|
|
|
|
void registerProvider(SyncProvider provider) {
|
|
_providers[provider.providerId] = provider;
|
|
if (_activeProviderId == 'none' || _activeProviderId == 'fake') _activeProviderId = provider.providerId;
|
|
notifyListeners();
|
|
}
|
|
SyncProvider? getProvider(String providerId) => _providers[providerId];
|
|
|
|
Future<SyncItem> enqueueRide(int localRideId, {String? targetProviderId, bool autoProcess = true}) async {
|
|
final providerId = targetProviderId ?? _activeProviderId;
|
|
if (providerId == 'firebase') {
|
|
final user = _authService?.currentUser;
|
|
if (user != null && !user.isAnonymous && _db != null) {
|
|
await _db.claimRideForCloudOwner(localRideId, user.uid);
|
|
}
|
|
}
|
|
final item = await _outbox.enqueue(localRideId, providerId, now: _now());
|
|
notifyListeners();
|
|
if (autoProcess) unawaited(processQueue());
|
|
return item;
|
|
}
|
|
|
|
Future<int> backupAllRideIds(Iterable<int> rideIds, {String? targetProviderId}) async {
|
|
var enqueued = 0;
|
|
for (final rideId in rideIds) {
|
|
await enqueueRide(rideId, targetProviderId: targetProviderId, autoProcess: false);
|
|
enqueued++;
|
|
}
|
|
if (enqueued > 0) unawaited(processQueue());
|
|
return enqueued;
|
|
}
|
|
|
|
Future<void> _startupRecoveryAndProcess() async {
|
|
await reconcileCompletedRides();
|
|
await processQueue();
|
|
}
|
|
|
|
Future<void> retryAuthRequired() async {
|
|
final user = _authService?.currentUser;
|
|
final isAuthenticated = user != null && !user.isAnonymous;
|
|
if (isAuthenticated && _db != null) {
|
|
await reconcileCompletedRides();
|
|
}
|
|
for (final item in _outbox.getAllItems().where((i) => i.status == SyncStatus.authRequired)) {
|
|
if (isAuthenticated && _db != null && item.providerId == 'firebase') {
|
|
await _db.claimRideForCloudOwner(item.localRideId, user.uid);
|
|
}
|
|
await _outbox.update(item.copyWith(status: SyncStatus.queued, clearLastError: true, clearNextRetry: true));
|
|
}
|
|
await processQueue();
|
|
}
|
|
|
|
/// Discovers completed rides in local SQLite that are missing from [SyncOutbox]
|
|
/// or need cloud confirmation recovery.
|
|
Future<CompletedRideReconciliationReport> reconcileCompletedRides({
|
|
String? targetProviderId,
|
|
}) async {
|
|
if (_activeReconciliation != null) {
|
|
return _activeReconciliation!;
|
|
}
|
|
final future = _performReconcileCompletedRides(targetProviderId: targetProviderId);
|
|
_activeReconciliation = future;
|
|
try {
|
|
return await future;
|
|
} finally {
|
|
_activeReconciliation = null;
|
|
}
|
|
}
|
|
|
|
Future<CompletedRideReconciliationReport> _performReconcileCompletedRides({
|
|
String? targetProviderId,
|
|
}) async {
|
|
if (_db == null) {
|
|
return const CompletedRideReconciliationReport();
|
|
}
|
|
|
|
final providerId = targetProviderId ?? _activeProviderId;
|
|
final provider = _providers[providerId];
|
|
final user = _authService?.currentUser;
|
|
final isAuthenticated = user != null && !user.isAnonymous;
|
|
|
|
int discovered = 0;
|
|
int alreadyTracked = 0;
|
|
int recoveredQueued = 0;
|
|
int recoveredAuthRequired = 0;
|
|
int recoveredAlreadySynced = 0;
|
|
int foreignOwnerSkipped = 0;
|
|
int errorCount = 0;
|
|
final List<String> errors = [];
|
|
|
|
try {
|
|
final completedRides = await (_db.select(_db.rides)
|
|
..where((r) => r.endTime.isNotNull()))
|
|
.get();
|
|
discovered = completedRides.length;
|
|
|
|
for (final ride in completedRides) {
|
|
try {
|
|
// Check if already tracked in outbox for this provider
|
|
final existingItem = _outbox.getItem(ride.id, providerId);
|
|
if (existingItem != null) {
|
|
alreadyTracked++;
|
|
continue;
|
|
}
|
|
|
|
// Foreign account check: never claim or upload another user's ride
|
|
if (ride.cloudOwnerUid != null &&
|
|
isAuthenticated &&
|
|
ride.cloudOwnerUid != user.uid) {
|
|
foreignOwnerSkipped++;
|
|
continue;
|
|
}
|
|
if (ride.cloudOwnerUid != null && !isAuthenticated) {
|
|
foreignOwnerSkipped++;
|
|
continue;
|
|
}
|
|
|
|
// Claim unclaimed ride if user is authenticated and target is firebase
|
|
if (ride.cloudOwnerUid == null && isAuthenticated && providerId == 'firebase') {
|
|
await _db.claimRideForCloudOwner(ride.id, user.uid);
|
|
}
|
|
|
|
// Stable cloud identity check
|
|
final remoteRideId = await _db.ensureCloudRideId(ride.id);
|
|
|
|
// Check if already confirmed uploaded in cloud backend
|
|
final isConfirmed = provider != null &&
|
|
await provider.isRideConfirmedInCloud(remoteRideId);
|
|
|
|
if (isConfirmed) {
|
|
final syncedItem = SyncItem(
|
|
localRideId: ride.id,
|
|
providerId: providerId,
|
|
status: SyncStatus.synced,
|
|
remoteId: remoteRideId,
|
|
createdAt: ride.endTime ?? ride.startTime,
|
|
uploadedAt: _now(),
|
|
);
|
|
await _outbox.update(syncedItem);
|
|
recoveredAlreadySynced++;
|
|
} else {
|
|
final providerRequiresAuth = provider != null &&
|
|
await provider.requiresAuthentication();
|
|
|
|
if (providerRequiresAuth || (!isAuthenticated && providerId == 'firebase')) {
|
|
final authReqItem = SyncItem(
|
|
localRideId: ride.id,
|
|
providerId: providerId,
|
|
status: SyncStatus.authRequired,
|
|
createdAt: ride.endTime ?? ride.startTime,
|
|
lastError: 'Authentication required for provider "$providerId"',
|
|
);
|
|
await _outbox.update(authReqItem);
|
|
recoveredAuthRequired++;
|
|
} else {
|
|
final queuedItem = SyncItem(
|
|
localRideId: ride.id,
|
|
providerId: providerId,
|
|
status: SyncStatus.queued,
|
|
createdAt: ride.endTime ?? ride.startTime,
|
|
);
|
|
await _outbox.update(queuedItem);
|
|
recoveredQueued++;
|
|
}
|
|
}
|
|
} catch (itemErr) {
|
|
errorCount++;
|
|
errors.add('Ride #${ride.id}: $itemErr');
|
|
}
|
|
}
|
|
} catch (e) {
|
|
errorCount++;
|
|
errors.add('Reconciliation failed: $e');
|
|
}
|
|
|
|
final report = CompletedRideReconciliationReport(
|
|
discoveredCompletedRides: discovered,
|
|
alreadyTrackedCount: alreadyTracked,
|
|
recoveredQueuedCount: recoveredQueued,
|
|
recoveredAuthRequiredCount: recoveredAuthRequired,
|
|
recoveredAlreadySyncedCount: recoveredAlreadySynced,
|
|
foreignOwnerSkippedCount: foreignOwnerSkipped,
|
|
errorCount: errorCount,
|
|
errors: errors,
|
|
);
|
|
|
|
if (report.totalRecovered > 0) {
|
|
notifyListeners();
|
|
}
|
|
return report;
|
|
}
|
|
|
|
static Duration calculateRetryDelay(int attemptCount) {
|
|
switch (attemptCount) {
|
|
case 1: return const Duration(seconds: 5);
|
|
case 2: return const Duration(seconds: 30);
|
|
case 3: return const Duration(minutes: 2);
|
|
case 4: return const Duration(minutes: 5);
|
|
case 5: return const Duration(minutes: 15);
|
|
default: return const Duration(minutes: 30);
|
|
}
|
|
}
|
|
|
|
@override
|
|
void didChangeAppLifecycleState(AppLifecycleState state) {
|
|
if (state == AppLifecycleState.resumed) {
|
|
unawaited(_startupRecoveryAndProcess());
|
|
}
|
|
}
|
|
|
|
Future<void> processQueue() async {
|
|
if (_currentProcessingFuture != null) {
|
|
_needsAnotherPass = true;
|
|
return _currentProcessingFuture!;
|
|
}
|
|
final future = _performProcessQueue();
|
|
_currentProcessingFuture = future;
|
|
try {
|
|
await future;
|
|
} finally {
|
|
_currentProcessingFuture = null;
|
|
}
|
|
}
|
|
|
|
Future<void> _performProcessQueue() async {
|
|
do {
|
|
_needsAnotherPass = false;
|
|
_isProcessing = true;
|
|
notifyListeners();
|
|
try {
|
|
final now = _now();
|
|
for (final item in _outbox.getEligibleItems(now)) {
|
|
final provider = _providers[item.providerId];
|
|
if (provider == null) {
|
|
await _outbox.update(item.copyWith(status: SyncStatus.failed, lastError: 'Provider "${item.providerId}" not registered', lastAttemptAt: now));
|
|
continue;
|
|
}
|
|
if (!await provider.isAvailable()) {
|
|
await _outbox.update(item.copyWith(status: SyncStatus.waitingForNetwork, lastAttemptAt: now));
|
|
continue;
|
|
}
|
|
if (await provider.requiresAuthentication()) {
|
|
await _outbox.update(item.copyWith(status: SyncStatus.authRequired, lastError: 'Authentication required for provider "${item.providerId}"', lastAttemptAt: now));
|
|
continue;
|
|
}
|
|
final uploadingItem = item.copyWith(status: SyncStatus.uploading, lastAttemptAt: now);
|
|
await _outbox.update(uploadingItem);
|
|
SyncResult result;
|
|
try {
|
|
result = await provider.uploadRide(uploadingItem);
|
|
} catch (e) {
|
|
result = SyncResult.temporaryFailure('Unhandled provider exception: $e', timestamp: now);
|
|
}
|
|
if (result.success) {
|
|
await _outbox.update(uploadingItem.copyWith(status: SyncStatus.synced, remoteId: result.remoteId, uploadedAt: result.timestamp, clearLastError: true, clearNextRetry: true));
|
|
} else if (result.requiresAuth) {
|
|
await _outbox.update(uploadingItem.copyWith(status: SyncStatus.authRequired, lastError: result.errorMessage));
|
|
} else if (result.isRetryable) {
|
|
final attempts = uploadingItem.attemptCount + 1;
|
|
await _outbox.update(uploadingItem.copyWith(status: SyncStatus.retryWait, attemptCount: attempts, nextRetryAt: now.add(calculateRetryDelay(attempts)), lastError: result.errorMessage));
|
|
} else {
|
|
await _outbox.update(uploadingItem.copyWith(status: SyncStatus.failed, attemptCount: uploadingItem.attemptCount + 1, lastError: result.errorMessage, clearNextRetry: true));
|
|
}
|
|
}
|
|
} finally {
|
|
_isProcessing = false;
|
|
notifyListeners();
|
|
}
|
|
} while (_needsAnotherPass);
|
|
}
|
|
|
|
Future<void> cancelSync(int localRideId, String providerId) async {
|
|
final item = _outbox.getItem(localRideId, providerId);
|
|
if (item != null && !item.status.isTerminal) {
|
|
await _outbox.update(item.copyWith(status: SyncStatus.cancelled));
|
|
notifyListeners();
|
|
}
|
|
}
|
|
|
|
Future<void> retryAllPending() async {
|
|
final retryable = _outbox.getAllItems().where((i) => i.status.isPending || i.status == SyncStatus.failed).toList(growable: false);
|
|
for (final item in retryable) {
|
|
await _outbox.update(item.copyWith(status: SyncStatus.queued, clearLastError: true, clearNextRetry: true));
|
|
}
|
|
await processQueue();
|
|
}
|
|
|
|
bool _disposed = false;
|
|
|
|
@override
|
|
void dispose() {
|
|
if (_disposed) return;
|
|
_disposed = true;
|
|
WidgetsBinding.instance.removeObserver(this);
|
|
super.dispose();
|
|
}
|
|
}
|