lib/core/sync/master_data_sync.dartDart
import 'dart:convert';
import 'package:flutter/foundation.dart';
import 'package:http/http.dart' as http;
import '../auth/auth_store.dart';
import 'master_data_sync_repository.dart';
class SyncClientException implements Exception {
const SyncClientException(this.message, {this.unauthorized = false});
final String message;
final bool unauthorized;
@override
String toString() => message;
}
class SyncPushResponse {
const SyncPushResponse(this.results);
final List<Map<String, dynamic>> results;
}
class SyncPullResponse {
const SyncPullResponse({
required this.changes,
required this.cursor,
required this.hasMore,
});
final List<Map<String, dynamic>> changes;
final int cursor;
final bool hasMore;
}
abstract interface class MasterDataSyncApi {
Future<SyncPushResponse> push(
String accessToken,
List<MasterDataOutboxOperation> operations,
);
Future<SyncPullResponse> pull(String accessToken, {required int cursor});
}
class HttpMasterDataSyncApi implements MasterDataSyncApi {
HttpMasterDataSyncApi({http.Client? client, String? baseUrl})
: _client = client ?? http.Client(),
baseUrl =
(baseUrl ??
const String.fromEnvironment(
'MACHWEIS_API_BASE_URL',
defaultValue: 'https://machweisapi.haselhuhn-media.com',
))
.replaceFirst(RegExp(r'/$'), '');
final http.Client _client;
final String baseUrl;
@override
Future<SyncPushResponse> push(
String accessToken,
List<MasterDataOutboxOperation> operations,
) async {
if (kIsWeb) {
throw const SyncClientException('Web-Synchronisation ist deaktiviert.');
}
try {
final response = await _client.post(
Uri.parse('$baseUrl/api/v1/sync/push'),
headers: {
'Authorization': 'Bearer $accessToken',
'Content-Type': 'application/json',
'Idempotency-Key': operations
.map((item) => item.idempotencyKey)
.join('.'),
},
body: jsonEncode({
'operations': operations.map((item) => item.toApiJson()).toList(),
}),
);
_debugResponse('POST /api/v1/sync/push', response);
final data = _responseData(response);
final results = data['results'] as List<dynamic>? ?? const [];
if (kDebugMode) {
for (final item in results.whereType<Map>()) {
if (item['status'] == 'rejected') {
debugPrint(
'[MACHWEIS Sync] Push abgelehnt: ${item['entity']}/${item['action'] ?? 'operation'} (${item['code'] ?? 'unknown'})',
);
}
}
}
return SyncPushResponse(
results.map((item) => Map<String, dynamic>.from(item as Map)).toList(),
);
} on http.ClientException {
throw const SyncClientException(
'Offline – Synchronisation wird später erneut versucht.',
);
}
}
@override
Future<SyncPullResponse> pull(
String accessToken, {
required int cursor,
}) async {
if (kIsWeb) {
throw const SyncClientException('Web-Synchronisation ist deaktiviert.');
}
try {
final response = await _client.get(
Uri.parse('$baseUrl/api/v1/sync/pull?cursor=$cursor&limit=100'),
headers: {'Authorization': 'Bearer $accessToken'},
);
_debugResponse('GET /api/v1/sync/pull', response);
final data = _responseData(response);
return SyncPullResponse(
changes: (data['changes'] as List<dynamic>? ?? const [])
.map((item) => Map<String, dynamic>.from(item as Map))
.toList(),
cursor: data['cursor'] as int? ?? cursor,
hasMore: data['has_more'] as bool? ?? false,
);
} on http.ClientException {
throw const SyncClientException(
'Offline – Synchronisation wird später erneut versucht.',
);
}
}
Map<String, dynamic> _responseData(http.Response response) {
final body = response.body.isEmpty
? const <String, dynamic>{}
: jsonDecode(response.body) as Map<String, dynamic>;
if (response.statusCode == 401) {
throw const SyncClientException(
'Anmeldung muss erneuert werden.',
unauthorized: true,
);
}
if (response.statusCode >= 500) {
throw const SyncClientException(
'Der Sync-Server ist vorübergehend nicht erreichbar.',
);
}
if (response.statusCode != 200 &&
response.statusCode != 403 &&
response.statusCode != 409 &&
response.statusCode != 422) {
throw const SyncClientException('Unerwartete Antwort des Sync-Servers.');
}
return Map<String, dynamic>.from(
body['data'] as Map<String, dynamic>? ?? const {},
);
}
void _debugResponse(String request, http.Response response) {
if (kDebugMode) {
debugPrint('[MACHWEIS Sync] $request: HTTP ${response.statusCode}');
}
}
}
enum MasterDataSyncStatus {
unavailable,
needsDecision,
idle,
syncing,
success,
offline,
conflict,
error,
}
typedef SyncAccessTokenProvider = Future<String> Function({bool forceRefresh});
typedef SyncTenantProvider = String? Function();
class MasterDataSyncController extends ChangeNotifier {
MasterDataSyncController({
MasterDataSyncRepository? repository,
MasterDataSyncApi? api,
SyncAccessTokenProvider? tokenProvider,
SyncTenantProvider? tenantProvider,
bool? supported,
}) : _repository = repository ?? SqliteMasterDataSyncRepository(),
_api = api ?? HttpMasterDataSyncApi(),
_tokenProvider = tokenProvider ?? AuthStore.instance.accessToken,
_tenantProvider =
tenantProvider ?? (() => AuthStore.instance.session?.tenantId),
_supported =
supported ??
(!kIsWeb &&
(defaultTargetPlatform == TargetPlatform.android ||
defaultTargetPlatform == TargetPlatform.iOS));
static final instance = MasterDataSyncController();
final MasterDataSyncRepository _repository;
final MasterDataSyncApi _api;
final SyncAccessTokenProvider _tokenProvider;
final SyncTenantProvider _tenantProvider;
final bool _supported;
MasterDataSyncStatus status = MasterDataSyncStatus.unavailable;
int pendingCount = 0;
int conflictCount = 0;
DateTime? lastSyncAt;
String? message;
String? conflictMessage;
String? _tenantId;
Future<void> Function()? onDataChanged;
bool get requiresInitialDecision =>
status == MasterDataSyncStatus.needsDecision;
bool get isSyncing => status == MasterDataSyncStatus.syncing;
Future<void> initialize() async {
final tenant = _tenantProvider();
if (!_supported || tenant == null) {
status = MasterDataSyncStatus.unavailable;
notifyListeners();
return;
}
_tenantId = tenant;
await _repository.clearLastError(tenant);
final state = await _repository.state(tenant);
lastSyncAt = state.lastSyncAt;
message = state.lastError;
if (state.decision == null) {
if (await _repository.hasLocalOnlyData(tenant)) {
status = MasterDataSyncStatus.needsDecision;
} else {
await _repository.setInitialDecision(
tenant,
InitialSyncDecision.serverOnly,
);
status = MasterDataSyncStatus.idle;
}
} else {
status = MasterDataSyncStatus.idle;
}
await _refreshCounts();
notifyListeners();
}
Future<void> chooseInitialSync(InitialSyncDecision decision) async {
final tenant = _tenantId;
if (tenant == null) return;
await _repository.setInitialDecision(tenant, decision);
status = MasterDataSyncStatus.idle;
message = decision == InitialSyncDecision.importLocal
? 'Lokale Stammdaten wurden zum Import vorgemerkt.'
: 'Serverdaten werden ergänzt; lokale Altdaten bleiben unangetastet.';
await _refreshCounts();
notifyListeners();
}
Future<void> synchronize({DateTime? now}) async {
final tenant = _tenantId;
if (!_supported || tenant == null || requiresInitialDecision || isSyncing) {
return;
}
final startedAt = now ?? DateTime.now();
status = MasterDataSyncStatus.syncing;
message = 'Stammdaten werden synchronisiert …';
notifyListeners();
var operations = <MasterDataOutboxOperation>[];
try {
await _repository.clearLastError(tenant);
await _repository.retryNow(tenant);
operations = await _repository.dueOperations(tenant, startedAt);
var token = await _tokenProvider();
if (operations.isNotEmpty) {
final pushed = await _withRefresh(
token,
(value) => _api.push(value, operations),
);
token = pushed.$1;
await _repository.applyPushResults(pushed.$2.results);
}
var state = await _repository.state(tenant);
var hasMore = true;
while (hasMore) {
final pulled = await _withRefresh(
token,
(value) => _api.pull(value, cursor: state.cursor),
);
token = pulled.$1;
await _repository.applyPull(
tenant,
pulled.$2.changes,
pulled.$2.cursor,
);
hasMore = pulled.$2.hasMore;
state = await _repository.state(tenant);
}
await _repository.completeSync(tenant, startedAt);
await _refreshCounts();
conflictMessage =
conflictCount > 0 && _repository is SqliteMasterDataSyncRepository
? await (_repository).latestConflictMessage(tenant)
: null;
lastSyncAt = startedAt;
status = conflictCount > 0
? MasterDataSyncStatus.conflict
: MasterDataSyncStatus.success;
message = conflictCount > 0
? '$conflictCount Konflikt(e) benötigen eine Prüfung${conflictMessage == null ? '.' : ': $conflictMessage'}'
: 'Stammdaten sind synchronisiert.';
await onDataChanged?.call();
} on SyncClientException catch (error) {
if (operations.isNotEmpty) {
await _repository.failOperations(operations, error.message, startedAt);
}
await _repository.failSync(tenant, error.message);
await _refreshCounts();
status = error.unauthorized
? MasterDataSyncStatus.error
: MasterDataSyncStatus.offline;
message = error.message;
} on Object {
const error = 'Synchronisation konnte nicht abgeschlossen werden.';
await _repository.failSync(tenant, error);
await _refreshCounts();
status = MasterDataSyncStatus.error;
message = error;
} finally {
notifyListeners();
}
}
Future<void> discardConflicts() async {
final tenant = _tenantId;
if (tenant == null || _repository is! SqliteMasterDataSyncRepository) {
return;
}
await (_repository).discardConflicts(tenant);
await _refreshCounts();
conflictMessage = null;
message =
'Lokale Konfliktänderungen wurden verworfen; die Serverdaten bleiben unverändert.';
status = MasterDataSyncStatus.idle;
notifyListeners();
}
Future<(String, T)> _withRefresh<T>(
String token,
Future<T> Function(String token) action,
) async {
try {
return (token, await action(token));
} on SyncClientException catch (error) {
if (!error.unauthorized) rethrow;
if (kDebugMode) {
debugPrint(
'[MACHWEIS Sync] HTTP 401: Access-Token wird erneuert und der Request einmal wiederholt.',
);
}
final refreshed = await _tokenProvider(forceRefresh: true);
return (refreshed, await action(refreshed));
}
}
Future<void> _refreshCounts() async {
final tenant = _tenantId;
if (tenant == null) return;
pendingCount = await _repository.pendingCount(tenant);
conflictCount = await _repository.conflictCount(tenant);
}
}