import 'dart:async'; import 'package:drift/drift.dart'; import 'package:flutter/foundation.dart'; import 'package:flutter_secure_storage/flutter_secure_storage.dart'; import 'package:sentry_flutter/sentry_flutter.dart'; import '../database/app_database.dart'; import '../models/bar_tab.dart'; import '../models/payment_method.dart'; import '../models/remote_command.dart'; import '../utils/app_config.dart'; import 'bar_tab_service.dart'; import 'kooltap_api_client.dart'; typedef _Credentials = ({String deviceId, String token}); typedef _CommandOutcome = ({CommandResult result, String? message}); class RemoteSyncService { final AppDatabase database; final BarTabService barTabService; final KoolTapApiClient apiClient; final FlutterSecureStorage secureStorage; /// How long tab changes must be quiet before they are synced. final Duration changeDebounce; /// The longest a tab change waits for a sync while changes keep coming. final Duration changeMaxDelay; RemoteSyncService({ required this.database, required this.barTabService, required this.apiClient, FlutterSecureStorage? secureStorage, this.changeDebounce = const Duration(milliseconds: kSyncChangeDebounceMs), this.changeMaxDelay = const Duration(milliseconds: kSyncChangeMaxDelayMs), }) : secureStorage = secureStorage ?? const FlutterSecureStorage(); static const _deviceIdKey = 'remote_device_id'; static const _authTokenKey = 'remote_auth_token'; static const _venueIdKey = 'remote_venue_id'; static const _syncStateId = 'default'; static const _pollingInterval = Duration( seconds: kSyncPollingIntervalSeconds, ); static const _maxBackoff = Duration(seconds: kSyncMaxBackoffSeconds); final _eventController = StreamController.broadcast(); Stream get events => _eventController.stream; bool _isSyncing = false; bool get isSyncing => _isSyncing; Timer? _timer; bool _isRunning = false; bool _paused = false; int _consecutiveFailures = 0; StreamSubscription>? _tabChanges; Timer? _changeTimer; DateTime? _firstPendingChange; bool _syncQueued = false; Future get venueId => secureStorage.read(key: _venueIdKey); Future get isConfigured async { final token = await secureStorage.read(key: _authTokenKey); final venueId = await this.venueId; return (token?.isNotEmpty ?? false) && (venueId?.isNotEmpty ?? false); } Future<_Credentials?> _credentials() async { final deviceId = await secureStorage.read(key: _deviceIdKey); final token = await secureStorage.read(key: _authTokenKey); if (deviceId == null || token == null) return null; return (deviceId: deviceId, token: token); } void startPeriodicSync() { _isRunning = true; // The first emission is the current state, not a change _tabChanges ??= barTabService .watchOpenTabs() .skip(1) .listen((_) => _onTabsChanged()); _scheduleNext(_pollingInterval); } void stopPeriodicSync() { _isRunning = false; _timer?.cancel(); _timer = null; _changeTimer?.cancel(); _changeTimer = null; _firstPendingChange = null; _syncQueued = false; unawaited(_tabChanges?.cancel()); _tabChanges = null; } void pause() { _paused = true; } void resume() { _paused = false; // Catch up right away instead of waiting for the next poll _onTabsChanged(); } /// Pushes tab changes to the server soon after they happen instead of /// waiting for the next poll. A burst of changes (several drinks rung up in /// a row) becomes one sync, sent once things are quiet for [changeDebounce] /// but never later than [changeMaxDelay] after the first change. void _onTabsChanged() { // While failing, the backoff timer decides when to retry if (!_isRunning || _paused || _consecutiveFailures > 0) return; final now = DateTime.now(); _firstPendingChange ??= now; final untilMaxDelay = changeMaxDelay - now.difference(_firstPendingChange!); final delay = untilMaxDelay < changeDebounce ? (untilMaxDelay.isNegative ? Duration.zero : untilMaxDelay) : changeDebounce; _changeTimer?.cancel(); _changeTimer = Timer(delay, () { _firstPendingChange = null; unawaited(sync()); }); } /// Every sync schedules the next one, so a failure can stretch the delay /// without stopping the loop. void _scheduleNext(Duration delay) { _timer?.cancel(); if (!_isRunning) return; _timer = Timer(delay, () { if (_paused) { _scheduleNext(_pollingInterval); return; } unawaited(sync()); }); } Duration _backoff() { final exponent = (_consecutiveFailures - 1).clamp(0, 3); final delay = _pollingInterval * (1 << exponent); return delay > _maxBackoff ? _maxBackoff : delay; } Future sync() async { if (_isSyncing) { // Changes made during a sync must not wait for the next poll _syncQueued = true; return; } final credentials = await _credentials(); if (credentials == null || !await isConfigured) return; _isSyncing = true; Duration nextDelay; try { await _doSync(credentials); _consecutiveFailures = 0; nextDelay = _pollingInterval; _emit(RemoteSyncEventType.syncCompleted); } on UnauthorizedException { debugPrint('RemoteSyncService: token revoked'); await secureStorage.delete(key: _authTokenKey); stopPeriodicSync(); _emit(RemoteSyncEventType.unauthorized); return; } on RateLimitedException catch (e) { nextDelay = await _recordFailure(e, retryAfter: e.retryAfter); } catch (e, stack) { unawaited(Sentry.captureException(e, stackTrace: stack)); nextDelay = await _recordFailure(e); } finally { _isSyncing = false; } _scheduleNext(nextDelay); if (_syncQueued) { _syncQueued = false; if (_consecutiveFailures == 0) unawaited(sync()); } } Future _recordFailure(Object error, {Duration? retryAfter}) async { debugPrint('RemoteSyncService: sync failed: $error'); _consecutiveFailures++; await _updateSyncState(success: false); _emit(RemoteSyncEventType.syncFailed, message: error.toString()); return retryAfter ?? _backoff(); } Future _doSync(_Credentials credentials) async { final state = await _getSyncState(); final openTabs = await barTabService.getOpenTabs(); final response = await apiClient.sync( token: credentials.token, request: SyncRequest( deviceId: credentials.deviceId, cursor: state?.serverCursor, openTabs: openTabs.map(_toSnapshot).toList(), ), ); for (final command in response.commands) { await _processCommand(command, credentials); } final openNames = openTabs.map((tab) => tab.customerName).toSet(); for (final name in response.plannedTabNames) { if (name.trim().isEmpty || !openNames.add(name)) continue; await barTabService.createTab(customerName: name); } await _updateSyncState(success: true, cursor: response.cursor); } SnapshotTab _toSnapshot(BarTab tab) { return SnapshotTab( id: tab.id, customerName: tab.customerName, version: tab.version, totalInCents: tab.totalInCents, itemCount: tab.itemCount, ); } Future _processCommand( RemoteCommand command, _Credentials credentials, ) async { final processed = await _getProcessedCommand(command.id); if (processed != null) { // The server only returns pending commands, so the earlier ack never arrived: resend it await _ackCommand(credentials, command.id, processed.result); return; } final outcome = await _executeCommand(command); // Recorded before acking, so a failed ack never re-executes the command. await _recordProcessed(command, outcome.result); await _ackCommand(credentials, command.id, outcome.result, outcome.message); } Future<_CommandOutcome> _executeCommand(RemoteCommand command) async { if (command.isExpired) { return (result: CommandResult.expired, message: 'Command expired'); } return switch (command.type) { CommandType.closeTab => _closeTab(command), null => ( result: CommandResult.failed, message: 'Unknown command type: ${command.rawType}', ), }; } Future<_CommandOutcome> _closeTab(RemoteCommand command) async { try { final tab = await barTabService.getTabById(command.tabId); await barTabService.closeTab( command.tabId, paymentMethod: PaymentMethod.fromValue(command.paymentMethod), expectedVersion: command.expectedVersion, ); _eventController.add( RemoteSyncEvent( type: RemoteSyncEventType.tabClosedRemotely, tabCustomerName: tab?.customerName, requestedBy: command.requestedBy, ), ); return (result: CommandResult.completed, message: null); } on TabNotFoundException { return _conflict('Tab not found'); } on TabVersionConflictException { return _conflict('Tab was modified since the request'); } catch (e, stack) { debugPrint('RemoteSyncService: command processing failed: $e'); unawaited(Sentry.captureException(e, stackTrace: stack)); return (result: CommandResult.failed, message: e.toString()); } } _CommandOutcome _conflict(String message) { _emit(RemoteSyncEventType.conflict, message: message); return (result: CommandResult.conflict, message: message); } void _emit(RemoteSyncEventType type, {String? message}) { _eventController.add(RemoteSyncEvent(type: type, message: message)); } Future _ackCommand( _Credentials credentials, String commandId, CommandResult status, [ String? result, ]) { return apiClient.ackCommand( token: credentials.token, deviceId: credentials.deviceId, ack: CommandAck(commandId: commandId, status: status, result: result), ); } Future _getSyncState() { return (database.select( database.remoteSyncState, )..where((row) => row.deviceId.equals(_syncStateId))).getSingleOrNull(); } Future _updateSyncState({required bool success, String? cursor}) async { final now = DateTime.now(); final existing = await _getSyncState(); await database .into(database.remoteSyncState) .insertOnConflictUpdate( RemoteSyncStateCompanion.insert( deviceId: _syncStateId, serverCursor: Value(cursor ?? existing?.serverCursor), lastSuccessfulSyncAt: Value( success ? now : existing?.lastSuccessfulSyncAt, ), lastAttemptedSyncAt: Value(now), consecutiveFailureCount: Value( success ? 0 : (existing?.consecutiveFailureCount ?? 0) + 1, ), ), ); } Future _getProcessedCommand(String commandId) { return (database.select( database.processedRemoteCommands, )..where((row) => row.commandId.equals(commandId))).getSingleOrNull(); } Future _recordProcessed( RemoteCommand command, CommandResult result, ) async { await database .into(database.processedRemoteCommands) .insertOnConflictUpdate( ProcessedRemoteCommandsCompanion.insert( commandId: command.id, commandType: command.rawType, processedAt: DateTime.now(), result: result, ), ); } Future setCredentials({ required String deviceId, required String token, required String venueId, }) async { await secureStorage.write(key: _deviceIdKey, value: deviceId); await secureStorage.write(key: _authTokenKey, value: token); await secureStorage.write(key: _venueIdKey, value: venueId); } Future clearCredentials() async { await secureStorage.delete(key: _deviceIdKey); await secureStorage.delete(key: _authTokenKey); await secureStorage.delete(key: _venueIdKey); } Future get lastSuccessfulSyncAt async { final state = await _getSyncState(); return state?.lastSuccessfulSyncAt; } Future> getRecentActivity() { return (database.select(database.processedRemoteCommands) ..orderBy([(row) => OrderingTerm.desc(row.processedAt)]) ..limit(50)) .get(); } void dispose() { stopPeriodicSync(); _eventController.close(); } }