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 '../database/stock_queries.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;

  /// Most stock adjustment outcomes sent per sync.
  static const maxStockResultsPerSync = 500;

  /// How long a handled stock adjustment is remembered after the server
  /// acknowledged its outcome.
  static const stockResultRetention = Duration(days: 30);

  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<Set<String>>? _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;
    unawaited(pruneStockAdjustmentResults());
    _tabChanges ??= _watchNewOutboxEvents();
    _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;
  }

  /// Every local tab change queues an event, so a new event in the outbox
  /// means a change to push. Watching the tabs instead would also fire on
  /// the server state a sync writes, and sync again forever.
  StreamSubscription<Set<String>> _watchNewOutboxEvents() {
    Set<String>? known;

    final ids = database.selectOnly(database.tabEvents)
      ..addColumns([database.tabEvents.id]);

    return ids
        .map((row) => row.read(database.tabEvents.id)!)
        .watch()
        .map((rows) => rows.toSet())
        .listen((current) {
          // The first emission is the current state, not a change
          final isNew = known != null && !current.every(known!.contains);
          known = current;
          if (isNew) _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<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 {
      final appliedStock = await _doSync(credentials);
      _consecutiveFailures = 0;
      // Show the new stock on the dashboard within seconds. The follow-up
      // skips the adjustments it already applied, so this doesn't loop.
      nextDelay = appliedStock ? Duration.zero : _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();
  }

  /// Returns whether any stock adjustment was applied, which only happens
  /// with a server that doesn't own the stock.
  Future<bool> _doSync(_Credentials credentials) async {
    final state = await _getSyncState();
    // One read, so the snapshot and the stock match the events that are sent
    final (outbox, openTabs, products, stockResults) = await database
        .transaction(
          () async => (
            await _pendingEvents(),
            await barTabService.getOpenTabs(),
            await database.select(database.products).get(),
            await _pendingStockResults(),
          ),
        );

    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>,
        ],
        products: products.map(_toCatalogEntry).toList(),
        stockAdjustmentResults: [
          for (final row in stockResults)
            jsonDecode(row.payload) as Map<String, dynamic>,
        ],
        settlementsCursor: state?.settlementsCursor,
      ),
    );

    await _clearSentEvents(response);
    await _markStockResultsReceived(response);
    await barTabService.recordSettlements(response.settlements);

    final serverTabs = response.tabs;
    if (serverTabs != null) {
      // After the acked events left the outbox: what is left still has to
      // reach the server and is replayed on top
      await barTabService.applyServerState(serverTabs);
    }

    // Same rebase for the stock: the server counts every tablet's sales
    final serverStock = response.stock;
    if (serverStock != null) await database.rebaseStock(serverStock);

    for (final command in response.commands) {
      await _processCommand(command, credentials);
    }

    // A server that holds the tabs creates planned tabs itself; doing it
    // here too would open them once per tablet
    if (serverTabs == null) {
      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);
      }
    }

    // A server that owns the stock applies the dashboard's adjustments
    // itself; doing it here too would apply a delivery once per tablet
    var appliedStock = false;
    if (serverStock == null) {
      for (final adjustment in response.stockAdjustments) {
        if (await _applyStockAdjustment(adjustment)) appliedStock = true;
      }
    }

    await _updateSyncState(
      success: true,
      cursor: response.cursor,
      settlementsCursor: response.settlementsCursor,
    );

    return appliedStock;
  }

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

  Future<List<StockAdjustmentResultRow>> _pendingStockResults() {
    return (database.select(database.stockAdjustmentResults)
          ..where((row) => row.acknowledged.not())
          ..orderBy([
            (row) => OrderingTerm.asc(row.createdAt),
            (row) => OrderingTerm.asc(const CustomExpression<int>('rowid')),
          ])
          ..limit(maxStockResultsPerSync))
        .get();
  }

  /// Received outcomes stay as the record that their adjustment was handled;
  /// [pruneStockAdjustmentResults] removes them later.
  Future<void> _markStockResultsReceived(SyncResponse response) async {
    for (final rejected in response.rejectedStockAdjustmentResults) {
      debugPrint(
        'RemoteSyncService: stock result ${rejected.id} rejected: '
        '${rejected.reason}',
      );
    }

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

    await (database.update(
      database.stockAdjustmentResults,
    )..where((row) => row.id.isIn(ids))).write(
      const StockAdjustmentResultsCompanion(acknowledged: Value(true)),
    );
  }

  /// Applies [adjustment] unless it was handled before, and stores the
  /// outcome in the same transaction, so a crash can neither apply it twice
  /// nor lose it. Returns whether the stock changed.
  Future<bool> _applyStockAdjustment(StockAdjustment adjustment) async {
    final type = adjustment.type;
    if (type == null) return false;

    return database.transaction(() async {
      final handled = await (database.select(
        database.stockAdjustmentResults,
      )..where((row) => row.id.equals(adjustment.id))).getSingleOrNull();
      if (handled != null) return false;

      final product = await (database.select(
        database.products,
      )..where((row) => row.id.equals(adjustment.productId))).getSingleOrNull();

      final Map<String, dynamic> result;
      if (product == null) {
        result = _rejectedStockResult(adjustment, 'unknown_product');
      } else if (type == StockAdjustmentType.add && product.infiniteStock) {
        result = _rejectedStockResult(adjustment, 'stock_not_tracked');
      } else {
        final stockAfter = switch (type) {
          StockAdjustmentType.add =>
            product.stockQuantity + adjustment.quantity,
          StockAdjustmentType.set => adjustment.quantity,
        };
        // A stock count on an untracked product starts tracking it
        await (database.update(
          database.products,
        )..where((row) => row.id.equals(product.id))).write(
          ProductsCompanion(
            stockQuantity: Value(stockAfter),
            infiniteStock: const Value(false),
          ),
        );
        result = {
          'id': adjustment.id,
          'status': 'applied',
          'stockAfter': stockAfter,
        };
      }

      await database
          .into(database.stockAdjustmentResults)
          .insert(
            StockAdjustmentResultsCompanion.insert(
              id: adjustment.id,
              payload: jsonEncode(result),
              createdAt: DateTime.now(),
            ),
          );

      return result['status'] == 'applied';
    });
  }

  Map<String, dynamic> _rejectedStockResult(
    StockAdjustment adjustment,
    String reason,
  ) {
    debugPrint(
      'RemoteSyncService: stock adjustment ${adjustment.id} rejected: $reason',
    );
    return {'id': adjustment.id, 'status': 'rejected', 'reason': reason};
  }

  /// Forgets handled stock adjustments whose outcome the server received
  /// more than [stockResultRetention] ago.
  Future<void> pruneStockAdjustmentResults({DateTime? now}) async {
    final cutoff = (now ?? DateTime.now()).subtract(stockResultRetention);

    await (database.delete(database.stockAdjustmentResults)..where(
          (row) => row.acknowledged & row.createdAt.isSmallerThanValue(cutoff),
        ))
        .go();
  }

  /// Hidden products are included: they still exist. A server that owns the
  /// stock ignores `stock`; it gets the counts as `stock_set` events.
  Map<String, dynamic> _toCatalogEntry(ProductRow product) {
    final category = product.category.trim();

    return {
      'id': product.id,
      'name': product.name,
      'category': category.isEmpty ? null : category,
      'priceInCents': product.priceInCents,
      'stock': product.infiniteStock ? null : product.stockQuantity,
    };
  }

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

  /// Settlements are deduplicated by id, so a cursor lost in a crash only
  /// makes the server send some of them again.
  Future<void> _updateSyncState({
    required bool success,
    String? cursor,
    String? settlementsCursor,
  }) async {
    final now = DateTime.now();
    final existing = await _getSyncState();

    await database
        .into(database.remoteSyncState)
        .insertOnConflictUpdate(
          RemoteSyncStateCompanion.insert(
            deviceId: _syncStateId,
            serverCursor: Value(cursor ?? existing?.serverCursor),
            settlementsCursor: Value(
              settlementsCursor ?? existing?.settlementsCursor,
            ),
            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();
  }
}