lib/core/sync/sync_queue_store.dartDart
import 'dart:convert';
import 'dart:math' as math;
import 'package:flutter/foundation.dart';
import '../storage/local_storage.dart';
enum SyncAction { create, update, finalize, delete }
class SyncOperation {
const SyncOperation({
required this.id,
required this.entityType,
required this.entityId,
required this.action,
required this.payload,
required this.createdAt,
this.tenantId,
this.idempotencyKey,
this.baseRevision = 0,
this.conflict = false,
this.attempts = 0,
this.lastError,
this.nextAttemptAt,
});
final String id;
final String entityType;
final String entityId;
final SyncAction action;
final Map<String, dynamic> payload;
final DateTime createdAt;
final String? tenantId;
final String? idempotencyKey;
final int baseRevision;
final bool conflict;
final int attempts;
final String? lastError;
final DateTime? nextAttemptAt;
bool isDue(DateTime now) =>
nextAttemptAt == null || !nextAttemptAt!.isAfter(now);
SyncOperation failed(String message, DateTime now) {
final nextAttempts = attempts + 1;
final delayMinutes = math.min(60, math.pow(2, nextAttempts).toInt());
return SyncOperation(
id: id,
entityType: entityType,
entityId: entityId,
action: action,
payload: payload,
createdAt: createdAt,
tenantId: tenantId,
idempotencyKey: idempotencyKey,
baseRevision: baseRevision,
conflict: conflict,
attempts: nextAttempts,
lastError: message,
nextAttemptAt: now.add(Duration(minutes: delayMinutes)),
);
}
Map<String, Object?> toJson() => {
'id': id,
'entityType': entityType,
'entityId': entityId,
'action': action.name,
'payload': payload,
'createdAt': createdAt.toIso8601String(),
'tenantId': tenantId,
'idempotencyKey': idempotencyKey,
'baseRevision': baseRevision,
'conflict': conflict,
'attempts': attempts,
'lastError': lastError,
'nextAttemptAt': nextAttemptAt?.toIso8601String(),
};
factory SyncOperation.fromJson(Map<String, dynamic> json) => SyncOperation(
id: json['id'] as String,
entityType: json['entityType'] as String,
entityId: json['entityId'] as String,
action: SyncAction.values.firstWhere(
(value) => value.name == json['action'],
orElse: () => SyncAction.update,
),
payload: Map<String, dynamic>.from(
json['payload'] as Map<String, dynamic>? ?? const {},
),
createdAt: DateTime.parse(json['createdAt'] as String),
tenantId: json['tenantId'] as String?,
idempotencyKey: json['idempotencyKey'] as String?,
baseRevision: json['baseRevision'] as int? ?? 0,
conflict: json['conflict'] as bool? ?? false,
attempts: json['attempts'] as int? ?? 0,
lastError: json['lastError'] as String?,
nextAttemptAt: json['nextAttemptAt'] == null
? null
: DateTime.parse(json['nextAttemptAt'] as String),
);
}
abstract interface class SyncTransport {
Future<void> send(SyncOperation operation);
}
class SyncQueueStore extends ChangeNotifier {
SyncQueueStore._();
static final instance = SyncQueueStore._();
static const _storageKey = 'machweis_sync_queue_v1';
static const _reportRevisionKey = 'machweis_report_revisions_v1';
final List<SyncOperation> _operations = [];
final Map<String, int> _reportRevisions = {};
final Set<String> _sendingReportOperations = {};
bool _isProcessing = false;
List<SyncOperation> get operations => List.unmodifiable(_operations);
int get pendingCount => _operations.length;
bool get isProcessing => _isProcessing;
Future<void> initialize() async {
final stored = await LocalStorage.instance.getString(_storageKey);
if (stored == null) return;
try {
final values = jsonDecode(stored) as List<dynamic>;
_operations
..clear()
..addAll(
values.map(
(value) => SyncOperation.fromJson(value as Map<String, dynamic>),
),
);
} on Object {
await LocalStorage.instance.remove(_storageKey);
}
final revisions = await LocalStorage.instance.getString(_reportRevisionKey);
if (revisions != null) {
try {
_reportRevisions
..clear()
..addAll(
(jsonDecode(revisions) as Map<String, dynamic>).map(
(key, value) => MapEntry(key, value as int),
),
);
} on Object {
await LocalStorage.instance.remove(_reportRevisionKey);
}
}
}
int reportRevision(String tenantId, String reportId) =>
_reportRevisions['$tenantId:$reportId'] ?? 0;
Future<void> enqueue({
required String entityType,
required String entityId,
required SyncAction action,
required Map<String, dynamic> payload,
String? tenantId,
int baseRevision = 0,
}) async {
final pendingCreateIndex = _operations.indexWhere(
(item) =>
item.entityType == entityType &&
item.entityId == entityId &&
item.tenantId == tenantId &&
!_sendingReportOperations.contains(item.id) &&
item.action == SyncAction.create,
);
if (action == SyncAction.delete && pendingCreateIndex != -1) {
_operations.removeWhere(
(item) =>
item.entityType == entityType &&
item.entityId == entityId &&
item.tenantId == tenantId,
);
} else if ((action == SyncAction.update || action == SyncAction.finalize) &&
pendingCreateIndex != -1) {
final pendingCreate = _operations[pendingCreateIndex];
_operations[pendingCreateIndex] = SyncOperation(
id: pendingCreate.id,
entityType: entityType,
entityId: entityId,
action: SyncAction.create,
payload: payload,
createdAt: pendingCreate.createdAt,
tenantId: tenantId ?? pendingCreate.tenantId,
idempotencyKey: pendingCreate.idempotencyKey ?? pendingCreate.id,
baseRevision: pendingCreate.baseRevision,
);
_operations.removeWhere(
(item) =>
item.entityType == entityType &&
item.entityId == entityId &&
item.tenantId == tenantId &&
item.id != pendingCreate.id,
);
} else {
_operations.removeWhere(
(item) =>
item.entityType == entityType &&
item.entityId == entityId &&
item.tenantId == tenantId &&
item.action == action,
);
final now = DateTime.now();
_operations.add(
SyncOperation(
id: '$entityType-$entityId-${action.name}-${now.microsecondsSinceEpoch}',
entityType: entityType,
entityId: entityId,
action: action,
payload: payload,
createdAt: now,
tenantId: tenantId,
idempotencyKey:
'$entityType-$entityId-${action.name}-${now.microsecondsSinceEpoch}',
baseRevision: baseRevision,
),
);
}
await _persistAndNotify();
}
Future<void> adoptLegacyReports(String tenantId) async {
var changed = false;
for (var index = 0; index < _operations.length; index++) {
final operation = _operations[index];
if (operation.entityType != 'report' || operation.tenantId != null) {
continue;
}
_operations[index] = SyncOperation(
id: operation.id,
entityType: operation.entityType,
entityId: operation.entityId,
action: operation.action,
payload: operation.payload,
createdAt: operation.createdAt,
tenantId: tenantId,
idempotencyKey: operation.id,
baseRevision: operation.action == SyncAction.create
? 0
: ((operation.payload['version'] as int? ?? 1) - 1),
attempts: operation.attempts,
lastError: operation.lastError,
nextAttemptAt: operation.nextAttemptAt,
);
changed = true;
}
if (changed) await _persistAndNotify();
}
List<SyncOperation> dueReports(
String tenantId,
DateTime now, {
bool force = false,
}) {
final due = <SyncOperation>[];
final selectedReports = <String>{};
final conflictedReports = _operations
.where(
(item) =>
item.entityType == 'report' &&
item.tenantId == tenantId &&
item.conflict,
)
.map((item) => item.entityId)
.toSet();
for (final item in _operations) {
if (item.entityType != 'report' ||
item.tenantId != tenantId ||
item.conflict ||
conflictedReports.contains(item.entityId) ||
selectedReports.contains(item.entityId) ||
(!force && !item.isDue(now))) {
continue;
}
selectedReports.add(item.entityId);
due.add(item);
}
return due;
}
void beginReportSend(Iterable<SyncOperation> operations) {
_sendingReportOperations.addAll(operations.map((item) => item.id));
}
void endReportSend(Iterable<SyncOperation> operations) {
_sendingReportOperations.removeAll(operations.map((item) => item.id));
}
int reportPendingCount(String tenantId) => _operations
.where(
(item) =>
item.entityType == 'report' &&
item.tenantId == tenantId &&
!item.conflict,
)
.length;
int reportConflictCount(String tenantId) => _operations
.where(
(item) =>
item.entityType == 'report' &&
item.tenantId == tenantId &&
item.conflict,
)
.length;
Future<void> applyReportResult(
SyncOperation operation,
Map<String, dynamic> result,
) async {
final index = _operations.indexWhere((item) => item.id == operation.id);
if (index == -1) return;
if (result['status'] == 'applied') {
final tenantId = operation.tenantId;
final revision = result['row_version'] as int?;
if (tenantId != null && revision != null) {
_reportRevisions['$tenantId:${operation.entityId}'] = revision;
for (
var successorIndex = 0;
successorIndex < _operations.length;
successorIndex++
) {
final successor = _operations[successorIndex];
if (successor.id == operation.id ||
successor.entityType != 'report' ||
successor.tenantId != tenantId ||
successor.entityId != operation.entityId) {
continue;
}
_operations[successorIndex] = SyncOperation(
id: successor.id,
entityType: successor.entityType,
entityId: successor.entityId,
action: successor.action == SyncAction.create
? SyncAction.update
: successor.action,
payload: successor.payload,
createdAt: successor.createdAt,
tenantId: successor.tenantId,
idempotencyKey: successor.idempotencyKey,
baseRevision: revision,
attempts: successor.attempts,
lastError: successor.lastError,
nextAttemptAt: successor.nextAttemptAt,
);
}
await LocalStorage.instance.setString(
_reportRevisionKey,
jsonEncode(_reportRevisions),
);
}
_operations.removeAt(index);
} else {
final message =
result['message'] as String? ??
'Server und Gerät enthalten unterschiedliche Berichtsversionen.';
_operations[index] = SyncOperation(
id: operation.id,
entityType: operation.entityType,
entityId: operation.entityId,
action: operation.action,
payload: operation.payload,
createdAt: operation.createdAt,
tenantId: operation.tenantId,
idempotencyKey: operation.idempotencyKey,
baseRevision: operation.baseRevision,
conflict: true,
attempts: operation.attempts,
lastError: message,
);
}
await _persistAndNotify();
}
Future<void> failReport(
SyncOperation operation,
String message,
DateTime now,
) async {
final index = _operations.indexWhere((item) => item.id == operation.id);
if (index == -1) return;
_operations[index] = operation.failed(message, now);
await _persistAndNotify();
}
Future<void> process(SyncTransport transport, {DateTime? now}) async {
if (_isProcessing) return;
_isProcessing = true;
notifyListeners();
final currentTime = now ?? DateTime.now();
try {
for (final operation in List<SyncOperation>.from(_operations)) {
if (!operation.isDue(currentTime)) continue;
try {
await transport.send(operation);
_operations.removeWhere((item) => item.id == operation.id);
} on Object catch (error) {
final index = _operations.indexWhere(
(item) => item.id == operation.id,
);
if (index != -1) {
_operations[index] = operation.failed(
error.toString(),
currentTime,
);
}
}
await _persistAndNotify();
}
} finally {
_isProcessing = false;
notifyListeners();
}
}
Future<void> retryNow(String operationId) async {
final index = _operations.indexWhere((item) => item.id == operationId);
if (index == -1) return;
final operation = _operations[index];
_operations[index] = SyncOperation(
id: operation.id,
entityType: operation.entityType,
entityId: operation.entityId,
action: operation.action,
payload: operation.payload,
createdAt: operation.createdAt,
attempts: operation.attempts,
lastError: operation.lastError,
);
await _persistAndNotify();
}
Future<void> _persistAndNotify() async {
await LocalStorage.instance.setString(
_storageKey,
jsonEncode(_operations.map((item) => item.toJson()).toList()),
);
notifyListeners();
}
}