kooltab/lib/services/remote_sync_service.dart

import 'dart:async';
import 'dart:convert';
import 'dart:io';

import 'package:drift/drift.dart';
import 'package:flutter/foundation.dart';
import 'package:flutter_secure_storage/flutter_secure_storage.dart';
import 'package:http/http.dart' as http;
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';

  /// Most outbox events sent per sync; a larger backlog drains over syncs.
  static const maxEventsPerSync = 500;

  static const _pollingInterval = Duration(
    seconds: kSyncPollingIntervalSeconds,
  );
  static const _maxBackoff = Duration(seconds: kSyncMaxBackoffSeconds);

  final _eventController = StreamController<RemoteSyncEvent>.broadcast();

  Stream<RemoteSyncEvent> get events => _eventController.stream;

  bool _isSyncing = false;
  bool get isSyncing => _isSyncing;

  Timer? _timer;
  bool _isRunning = false;
  bool _paused = false;
  int _consecutiveFailures = 0;

  StreamSubscription<List<BarTab>>? _tabChanges;
  Timer? _changeTimer;
  DateTime? _firstPendingChange;
  bool _syncQueued = false;

  _Credentials? _cachedCredentials;

  Future<String?> get venueId => secureStorage.read(key: _venueIdKey);

  Future<bool> get isConfigured async => await _credentials() != null;

  Future<String?> get authToken async => (await _credentials())?.token;

  Future<_Credentials?> _credentials() async {
    if (_cachedCredentials != null) return _cachedCredentials;

    final deviceId = await secureStorage.read(key: _deviceIdKey);
    final token = await secureStorage.read(key: _authTokenKey);
    final venueId = await this.venueId;
    if (deviceId == null ||
        token == null ||
        token.isEmpty ||
        venueId == null ||
        venueId.isEmpty) {
      return null;
    }

    return _cachedCredentials = (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;
  }

  /// 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<void> 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) return;

    _isSyncing = true;
    Duration nextDelay;

    try {
      await _doSync(credentials);
      _consecutiveFailures = 0;
      nextDelay = _pollingInterval;
      _emit(RemoteSyncEventType.syncCompleted);
    } on UnauthorizedException {
      debugPrint('RemoteSyncService: token revoked');
      _cachedCredentials = null;
      await secureStorage.delete(key: _authTokenKey);
      stopPeriodicSync();
      _emit(RemoteSyncEventType.unauthorized);
      return;
    } on RateLimitedException catch (e) {
      nextDelay = await _recordFailure(e, retryAfter: e.retryAfter);
    } catch (e, stack) {
      if (!_isConnectivityError(e)) {
        unawaited(Sentry.captureException(e, stackTrace: stack));
      }
      nextDelay = await _recordFailure(e);
    } finally {
      _isSyncing = false;
    }

    _scheduleNext(nextDelay);

    if (_syncQueued) {
      _syncQueued = false;
      if (_consecutiveFailures == 0) unawaited(sync());
    }
  }

  /// Offline tablets are expected, not worth an error report.
  bool _isConnectivityError(Object error) {
    return error is IOException ||
        error is TimeoutException ||
        error is http.ClientException;
  }

  Future<Duration> _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<void> _doSync(_Credentials credentials) async {
    final state = await _getSyncState();
    // Read before the snapshot, so the snapshot is never older than the events
    final outbox = await _pendingEvents();
    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(),
        events: [
          for (final row in outbox)
            jsonDecode(row.payload) as Map<String, dynamic>,
        ],
      ),
    );

    await _clearSentEvents(response);

    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);
  }

  Future<List<TabEventRow>> _pendingEvents() {
    return (database.select(database.tabEvents)
          ..orderBy([
            (row) => OrderingTerm.asc(row.createdAt),
            // Same-second events keep their insertion order
            (row) => OrderingTerm.asc(const CustomExpression<int>('rowid')),
          ])
          ..limit(maxEventsPerSync))
        .get();
  }

  /// Events the server stored or refused leave the outbox. Events created
  /// while the sync was in flight are in neither list and go next time.
  Future<void> _clearSentEvents(SyncResponse response) async {
    for (final rejected in response.rejectedEvents) {
      debugPrint(
        'RemoteSyncService: event ${rejected.id} rejected: ${rejected.reason}',
      );
    }

    final ids = {
      ...response.acknowledgedEventIds,
      for (final rejected in response.rejectedEvents)
        if (rejected.id != null) rejected.id!,
    };
    if (ids.isEmpty) return;

    await (database.delete(
      database.tabEvents,
    )..where((row) => row.id.isIn(ids))).go();
  }

  SnapshotTab _toSnapshot(BarTab tab) {
    return SnapshotTab(
      id: tab.id,
      customerName: tab.customerName,
      version: tab.version,
      totalInCents: tab.totalInCents,
      itemCount: tab.itemCount,
      items: [
        for (final item in tab.items)
          SnapshotTabItem(
            productName: item.productName,
            quantity: item.quantity,
            unitPriceInCents: item.unitPriceInCents,
          ),
      ],
    );
  }

  Future<void> _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,
        commandId: command.id,
      );

      _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<void> _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<RemoteSyncStateRow?> _getSyncState() {
    return (database.select(
      database.remoteSyncState,
    )..where((row) => row.deviceId.equals(_syncStateId))).getSingleOrNull();
  }

  Future<void> _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<ProcessedRemoteCommandRow?> _getProcessedCommand(String commandId) {
    return (database.select(
      database.processedRemoteCommands,
    )..where((row) => row.commandId.equals(commandId))).getSingleOrNull();
  }

  Future<void> _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<void> 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);
    _cachedCredentials = (deviceId: deviceId, token: token);
  }

  Future<void> clearCredentials() async {
    _cachedCredentials = null;
    await secureStorage.delete(key: _deviceIdKey);
    await secureStorage.delete(key: _authTokenKey);
    await secureStorage.delete(key: _venueIdKey);
  }

  Future<DateTime?> get lastSuccessfulSyncAt async {
    final state = await _getSyncState();
    return state?.lastSuccessfulSyncAt;
  }

  Future<List<ProcessedRemoteCommandRow>> getRecentActivity() {
    return (database.select(database.processedRemoteCommands)
          ..orderBy([(row) => OrderingTerm.desc(row.processedAt)])
          ..limit(50))
        .get();
  }

  void dispose() {
    stopPeriodicSync();
    _eventController.close();
  }
}