400 lines
12 KiB
Dart
400 lines
12 KiB
Dart
import 'package:dio/dio.dart';
|
|
import 'package:flutter/foundation.dart';
|
|
import 'package:miaoji_zhang/shared/api/api_client.dart';
|
|
import 'package:miaoji_zhang/shared/services/current_ledger_store.dart';
|
|
import 'package:miaoji_zhang/shared/services/local_database.dart';
|
|
import 'package:miaoji_zhang/shared/services/session_store.dart';
|
|
|
|
enum CloudSyncPhase {
|
|
idle,
|
|
syncing,
|
|
synced,
|
|
pending,
|
|
offline,
|
|
disabled,
|
|
reauthenticate,
|
|
failed,
|
|
}
|
|
|
|
class SyncService extends ChangeNotifier {
|
|
SyncService._();
|
|
|
|
static final instance = SyncService._();
|
|
static final conflictRevision = ValueNotifier<int>(0);
|
|
|
|
final _dio = ApiClient.instance.dio;
|
|
Future<void>? _running;
|
|
CloudSyncPhase _phase = CloudSyncPhase.idle;
|
|
DateTime? _lastSyncedAt;
|
|
int _pendingCount = 0;
|
|
bool _connectivityInterrupted = false;
|
|
bool _hadFailures = false;
|
|
|
|
int get conflictCount => LocalDatabase.instance.conflicts().length;
|
|
CloudSyncPhase get phase => _phase;
|
|
DateTime? get lastSyncedAt => _lastSyncedAt;
|
|
int get pendingCount => _pendingCount;
|
|
|
|
String get statusLabel => switch (_phase) {
|
|
CloudSyncPhase.syncing => '正在同步',
|
|
CloudSyncPhase.synced => '已同步',
|
|
CloudSyncPhase.pending => '$_pendingCount 项待同步',
|
|
CloudSyncPhase.offline => '网络不可用，$_pendingCount 项待同步',
|
|
CloudSyncPhase.disabled => '云同步已关闭',
|
|
CloudSyncPhase.reauthenticate => '登录已过期，等待重新登录',
|
|
CloudSyncPhase.failed => '同步失败，点击重试',
|
|
CloudSyncPhase.idle => _pendingCount > 0 ? '$_pendingCount 项待同步' : '等待同步',
|
|
};
|
|
|
|
void refreshLocalStatus() {
|
|
final session = SessionStore.instance;
|
|
if (!session.isAccount || !session.cloudSyncEnabled) {
|
|
_setPhase(CloudSyncPhase.disabled, 0);
|
|
return;
|
|
}
|
|
if (session.needsReauth) {
|
|
_setPhase(CloudSyncPhase.reauthenticate, _safePendingCount());
|
|
return;
|
|
}
|
|
final pending = _safePendingCount();
|
|
_setPhase(
|
|
pending > 0 ? CloudSyncPhase.pending : CloudSyncPhase.idle,
|
|
pending,
|
|
);
|
|
}
|
|
|
|
Future<void> run() {
|
|
final session = SessionStore.instance;
|
|
if (!session.isAccount || !session.cloudSyncEnabled) {
|
|
_setPhase(CloudSyncPhase.disabled, 0);
|
|
return Future.value();
|
|
}
|
|
if (session.needsReauth) {
|
|
_setPhase(CloudSyncPhase.reauthenticate, _safePendingCount());
|
|
return Future.value();
|
|
}
|
|
final active = _running;
|
|
if (active != null) return active;
|
|
_connectivityInterrupted = false;
|
|
_hadFailures = false;
|
|
_setPhase(CloudSyncPhase.syncing, _safePendingCount());
|
|
return _running = _execute().whenComplete(() => _running = null);
|
|
}
|
|
|
|
Future<void> _execute() async {
|
|
try {
|
|
await _runCore();
|
|
if (SessionStore.instance.needsReauth) {
|
|
_setPhase(CloudSyncPhase.reauthenticate, _safePendingCount());
|
|
return;
|
|
}
|
|
final pending = _safePendingCount();
|
|
if (_connectivityInterrupted) {
|
|
_setPhase(CloudSyncPhase.offline, pending);
|
|
} else if (_hadFailures) {
|
|
_setPhase(CloudSyncPhase.failed, pending);
|
|
} else if (pending > 0) {
|
|
_setPhase(CloudSyncPhase.pending, pending);
|
|
} else {
|
|
_lastSyncedAt = DateTime.now();
|
|
_setPhase(CloudSyncPhase.synced, 0);
|
|
}
|
|
} catch (error) {
|
|
final pending = _safePendingCount();
|
|
_setPhase(
|
|
isConnectivityError(error)
|
|
? CloudSyncPhase.offline
|
|
: CloudSyncPhase.failed,
|
|
pending,
|
|
);
|
|
}
|
|
}
|
|
|
|
int _safePendingCount() {
|
|
try {
|
|
return LocalDatabase.instance.pendingSync().length;
|
|
} catch (_) {
|
|
return 0;
|
|
}
|
|
}
|
|
|
|
void _setPhase(CloudSyncPhase phase, int pending) {
|
|
if (_phase == phase && _pendingCount == pending) return;
|
|
_phase = phase;
|
|
_pendingCount = pending;
|
|
notifyListeners();
|
|
}
|
|
|
|
Future<void> _runCore() async {
|
|
final database = LocalDatabase.instance;
|
|
for (final item in database.pendingSync()) {
|
|
final queueId = item['id'] as int;
|
|
try {
|
|
await _process(item);
|
|
database.completeSync(queueId);
|
|
} on _DependencyPending {
|
|
continue;
|
|
} on DioException catch (error) {
|
|
if (error.response?.statusCode == 409) {
|
|
final data = error.response?.data;
|
|
final remote = data is Map && data['server'] is Map
|
|
? Map<String, dynamic>.from(data['server'] as Map)
|
|
: <String, dynamic>{};
|
|
database.addConflict(
|
|
item['entityType'] as String,
|
|
item['entityId'] as int,
|
|
item['operation'] as String,
|
|
item['payload'] as Map<String, dynamic>,
|
|
remote,
|
|
);
|
|
database.completeSync(queueId);
|
|
conflictRevision.value++;
|
|
continue;
|
|
}
|
|
if (error.response?.statusCode == 401) {
|
|
await SessionStore.instance.markNeedsReauth();
|
|
return;
|
|
}
|
|
if (isConnectivityError(error)) {
|
|
_connectivityInterrupted = true;
|
|
return;
|
|
}
|
|
database.failSync(queueId);
|
|
_hadFailures = true;
|
|
} catch (_) {
|
|
database.failSync(queueId);
|
|
_hadFailures = true;
|
|
}
|
|
}
|
|
await CurrentLedgerStore.instance.ensureLoaded(force: true);
|
|
}
|
|
|
|
Future<void> _process(Map<String, dynamic> item) async {
|
|
final entityType = item['entityType'] as String;
|
|
final entityId = item['entityId'] as int;
|
|
final operation = item['operation'] as String;
|
|
final payload = Map<String, dynamic>.from(
|
|
item['payload'] as Map<String, dynamic>,
|
|
);
|
|
switch (entityType) {
|
|
case 'ledger':
|
|
await _syncLedger(entityId, operation, payload);
|
|
return;
|
|
case 'category':
|
|
await _syncCategory(entityId, operation, payload);
|
|
return;
|
|
case 'transaction':
|
|
await _syncTransaction(entityId, operation, payload);
|
|
return;
|
|
case 'budget':
|
|
await _syncBudget(operation, payload);
|
|
return;
|
|
}
|
|
}
|
|
|
|
Future<void> _syncLedger(
|
|
int entityId,
|
|
String operation,
|
|
Map<String, dynamic> payload,
|
|
) async {
|
|
if (operation == 'create') {
|
|
final response = await _dio.post('/api/ledgers', data: payload);
|
|
LocalDatabase.instance.remapLedger(
|
|
entityId,
|
|
Map<String, dynamic>.from(response.data as Map),
|
|
);
|
|
return;
|
|
}
|
|
final remoteId = _resolvedId('ledger', entityId);
|
|
if (operation == 'update') {
|
|
await _dio.put('/api/ledgers/$remoteId', data: payload);
|
|
} else if (operation == 'delete') {
|
|
await _dio.delete('/api/ledgers/$remoteId');
|
|
} else if (operation == 'select') {
|
|
await _dio.put('/api/ledgers/$remoteId/default');
|
|
}
|
|
}
|
|
|
|
Future<void> _syncCategory(
|
|
int entityId,
|
|
String operation,
|
|
Map<String, dynamic> payload,
|
|
) async {
|
|
if (operation == 'create') {
|
|
final response = await _dio.post('/api/categories', data: payload);
|
|
LocalDatabase.instance.remapCategory(
|
|
entityId,
|
|
Map<String, dynamic>.from(response.data as Map),
|
|
);
|
|
return;
|
|
}
|
|
if (operation == 'reorder') {
|
|
final ids = (payload['categoryIds'] as List)
|
|
.map((value) => _resolvedId('category', (value as num).toInt()))
|
|
.toList();
|
|
await _dio.put(
|
|
'/api/categories/reorder',
|
|
data: {'type': payload['type'], 'categoryIds': ids},
|
|
);
|
|
return;
|
|
}
|
|
final remoteId = _resolvedId('category', entityId);
|
|
if (operation == 'update') {
|
|
await _dio.put('/api/categories/$remoteId', data: payload);
|
|
} else if (operation == 'delete') {
|
|
await _dio.delete('/api/categories/$remoteId');
|
|
}
|
|
}
|
|
|
|
Future<void> _syncTransaction(
|
|
int entityId,
|
|
String operation,
|
|
Map<String, dynamic> payload,
|
|
) async {
|
|
final data = Map<String, dynamic>.from(payload);
|
|
if (data['ledgerId'] is num) {
|
|
data['ledgerId'] = _resolvedId(
|
|
'ledger',
|
|
(data['ledgerId'] as num).toInt(),
|
|
);
|
|
}
|
|
if (data['categoryId'] is num) {
|
|
data['categoryId'] = _resolvedId(
|
|
'category',
|
|
(data['categoryId'] as num).toInt(),
|
|
);
|
|
}
|
|
if (operation == 'create') {
|
|
final response = await _dio.post('/api/transactions', data: data);
|
|
LocalDatabase.instance.replaceLocalTransaction(
|
|
entityId,
|
|
Map<String, dynamic>.from(response.data as Map),
|
|
);
|
|
return;
|
|
}
|
|
final remoteId = _resolvedId('transaction', entityId);
|
|
if (operation == 'update') {
|
|
final response = await _dio.put(
|
|
'/api/transactions/$remoteId',
|
|
data: data,
|
|
);
|
|
LocalDatabase.instance.cacheTransaction(
|
|
Map<String, dynamic>.from(response.data as Map),
|
|
);
|
|
} else if (operation == 'delete') {
|
|
await _dio.delete(
|
|
'/api/transactions/$remoteId',
|
|
queryParameters: {
|
|
if (data['baseUpdatedAt'] != null)
|
|
'baseUpdatedAt': data['baseUpdatedAt'],
|
|
},
|
|
);
|
|
} else if (operation == 'restore') {
|
|
final response = await _dio.post(
|
|
'/api/transactions/$remoteId/restore',
|
|
queryParameters: {
|
|
if (data['baseUpdatedAt'] != null)
|
|
'baseUpdatedAt': data['baseUpdatedAt'],
|
|
},
|
|
);
|
|
LocalDatabase.instance.cacheTransaction(
|
|
Map<String, dynamic>.from(response.data as Map),
|
|
);
|
|
}
|
|
}
|
|
|
|
Future<void> _syncBudget(
|
|
String operation,
|
|
Map<String, dynamic> payload,
|
|
) async {
|
|
final ledgerId = _resolvedId(
|
|
'ledger',
|
|
(payload['ledgerId'] as num).toInt(),
|
|
);
|
|
if (operation == 'batch') {
|
|
final items = (payload['items'] as List).map((value) {
|
|
final item = Map<String, dynamic>.from(value as Map);
|
|
if (item['categoryId'] is num) {
|
|
item['categoryId'] = _resolvedId(
|
|
'category',
|
|
(item['categoryId'] as num).toInt(),
|
|
);
|
|
}
|
|
return item;
|
|
}).toList();
|
|
await _dio.put(
|
|
'/api/budgets/batch',
|
|
queryParameters: {
|
|
'year': payload['year'],
|
|
'month': payload['month'],
|
|
'ledgerId': ledgerId,
|
|
},
|
|
data: {'recurring': payload['recurring'], 'items': items},
|
|
);
|
|
return;
|
|
}
|
|
final categoryId = payload['categoryId'] is num
|
|
? _resolvedId('category', (payload['categoryId'] as num).toInt())
|
|
: null;
|
|
await _dio.put(
|
|
'/api/budgets',
|
|
queryParameters: {
|
|
'year': payload['year'],
|
|
'month': payload['month'],
|
|
'ledgerId': ledgerId,
|
|
},
|
|
data: {
|
|
'categoryId': categoryId,
|
|
'amount': payload['amount'],
|
|
'recurring': payload['recurring'],
|
|
},
|
|
);
|
|
}
|
|
|
|
int _resolvedId(String entityType, int value) {
|
|
final resolved = LocalDatabase.instance.remoteId(entityType, value);
|
|
if (resolved == null) throw const _DependencyPending();
|
|
return resolved;
|
|
}
|
|
|
|
Future<void> resolveConflict(
|
|
Map<String, dynamic> conflict, {
|
|
required bool keepLocal,
|
|
}) async {
|
|
final database = LocalDatabase.instance;
|
|
final remote = conflict['remote'] as Map<String, dynamic>;
|
|
if (keepLocal && conflict['entityType'] == 'transaction') {
|
|
final entityId = conflict['entityId'] as int;
|
|
final remoteId = _resolvedId('transaction', entityId);
|
|
final operation = conflict['operation'] as String;
|
|
final payload = Map<String, dynamic>.from(
|
|
conflict['local'] as Map<String, dynamic>,
|
|
)..remove('baseUpdatedAt');
|
|
if (operation == 'update') {
|
|
final response = await _dio.put(
|
|
'/api/transactions/$remoteId',
|
|
data: payload,
|
|
);
|
|
database.cacheTransaction(
|
|
Map<String, dynamic>.from(response.data as Map),
|
|
);
|
|
} else if (operation == 'delete') {
|
|
await _dio.delete('/api/transactions/$remoteId');
|
|
} else if (operation == 'restore') {
|
|
final response = await _dio.post('/api/transactions/$remoteId/restore');
|
|
database.cacheTransaction(
|
|
Map<String, dynamic>.from(response.data as Map),
|
|
);
|
|
}
|
|
} else if (!keepLocal && remote.isNotEmpty) {
|
|
database.cacheTransaction(remote);
|
|
}
|
|
database.resolveConflict(conflict['id'] as int);
|
|
conflictRevision.value++;
|
|
}
|
|
}
|
|
|
|
class _DependencyPending implements Exception {
|
|
const _DependencyPending();
|
|
}
|