nearle_pos/lib/data/sync/sync_engine.dart

import 'dart:async';
import 'dart:math';

import 'package:flutter/foundation.dart';

import '../../domain/repositories/sync_repository.dart';
import '../remote/order_transport.dart';

/// Why a drain was attempted. Shown in the events log, because "why did it try
/// then" is the first question when a sync misbehaves.
enum SyncTrigger {
  startup,
  saleCommitted,
  connectivityRegained,
  periodic,
  retry,
  headOfficeRequest,
  manual,
}

/// What the engine is doing, for the header and the events screen.
@immutable
class SyncEngineState {
  const SyncEngineState({
    this.isSyncing = false,
    this.pending = 0,
    this.online = true,
    this.consecutiveFailures = 0,
    this.lastSuccessAt,
    this.lastAttemptAt,
    this.nextAttemptAt,
    this.lastError,
    this.isHalted = false,
    this.lastTrigger,
  });

  final bool isSyncing;
  final int pending;
  final bool online;
  final int consecutiveFailures;
  final DateTime? lastSuccessAt;
  final DateTime? lastAttemptAt;

  /// When the backoff timer will fire. Null when nothing is scheduled.
  final DateTime? nextAttemptAt;

  final String? lastError;

  /// Set after a failure that retrying cannot fix — a bad credential, an
  /// unconfigured endpoint. The engine stops its own scheduling so it does not
  /// hammer a door that is locked; a manual sync or a config change resumes it.
  final bool isHalted;

  final SyncTrigger? lastTrigger;

  bool get isHealthy => !isHalted && consecutiveFailures == 0;

  SyncEngineState copyWith({
    bool? isSyncing,
    int? pending,
    bool? online,
    int? consecutiveFailures,
    DateTime? lastSuccessAt,
    DateTime? lastAttemptAt,
    DateTime? nextAttemptAt,
    String? lastError,
    bool? isHalted,
    SyncTrigger? lastTrigger,
    bool clearNextAttempt = false,
    bool clearError = false,
  }) =>
      SyncEngineState(
        isSyncing: isSyncing ?? this.isSyncing,
        pending: pending ?? this.pending,
        online: online ?? this.online,
        consecutiveFailures: consecutiveFailures ?? this.consecutiveFailures,
        lastSuccessAt: lastSuccessAt ?? this.lastSuccessAt,
        lastAttemptAt: lastAttemptAt ?? this.lastAttemptAt,
        nextAttemptAt:
            clearNextAttempt ? null : nextAttemptAt ?? this.nextAttemptAt,
        lastError: clearError ? null : lastError ?? this.lastError,
        isHalted: isHalted ?? this.isHalted,
        lastTrigger: lastTrigger ?? this.lastTrigger,
      );
}

/// Decides *when* bills are uploaded.
///
/// The repository knows how to send one batch; this knows when to ask, and what
/// to do when the answer is no. Together they turn the orders table into a
/// queue that empties itself:
///
/// * a sale is committed — try immediately, so a bill is usually up within
///   seconds of the drawer closing;
/// * the network returns — try at once rather than waiting out a poll;
/// * nothing happened for a while — poll, because interface state lies;
/// * head office asked — the MQTT downlink can pull a shift up on demand;
/// * the cashier pressed sync — always allowed, even while halted.
///
/// ### Two guarantees worth stating
///
/// **Single flight.** Only one drain runs at a time. Without this, a busy till
/// firing a trigger per sale would have several passes reading the same pending
/// rows and publishing them concurrently — every bill sent two or three times.
/// A trigger arriving mid-drain sets a flag and is honoured once the current
/// pass finishes, so nothing is dropped either.
///
/// **Backoff with jitter.** A failed attempt waits, and waits longer each time,
/// to a ceiling. The jitter matters more than it looks: when a shop's line
/// drops, every terminal in the store fails at the same instant, and without it
/// they would all retry in lockstep and keep colliding on the way back up.
class SyncEngine {
  SyncEngine({
    required SyncRepository repository,
    Stream<bool>? connectivity,
    Stream<DownlinkMessage>? downlink,
    Future<void> Function()? onCatalogueChanged,
    Duration idlePoll = const Duration(minutes: 5),
    Duration baseBackoff = const Duration(seconds: 2),
    Duration maxBackoff = const Duration(minutes: 5),
    Random? random,
    DateTime Function()? clock,
    Timer Function(Duration, void Function())? scheduleTimer,
  })  : _repository = repository,
        _connectivity = connectivity,
        _downlink = downlink,
        _onCatalogueChanged = onCatalogueChanged,
        _idlePoll = idlePoll,
        _baseBackoff = baseBackoff,
        _maxBackoff = maxBackoff,
        _random = random ?? Random(),
        _now = clock ?? DateTime.now,
        _schedule = scheduleTimer ?? Timer.new;

  final SyncRepository _repository;
  final Stream<bool>? _connectivity;
  final Stream<DownlinkMessage>? _downlink;
  final Future<void> Function()? _onCatalogueChanged;

  final Duration _idlePoll;
  final Duration _baseBackoff;
  final Duration _maxBackoff;
  final Random _random;
  final DateTime Function() _now;
  final Timer Function(Duration, void Function()) _schedule;

  final _states = StreamController<SyncEngineState>.broadcast();

  SyncEngineState _state = const SyncEngineState();
  SyncEngineState get state => _state;
  Stream<SyncEngineState> get states => _states.stream;

  /// Held for the whole of a drain. The single-flight guarantee rests on this
  /// being checked and set without an `await` in between.
  bool _draining = false;

  /// A trigger that arrived while a drain was already running.
  SyncTrigger? _queuedTrigger;

  Timer? _backoffTimer;
  Timer? _pollTimer;
  StreamSubscription<bool>? _connectivitySub;
  StreamSubscription<DownlinkMessage>? _downlinkSub;
  bool _stopped = false;

  // ------------------------------------------------------------------ Life
  Future<void> start() async {
    if (_stopped) throw StateError('This SyncEngine has been disposed.');

    _connectivitySub = _connectivity?.listen((online) {
      _emit(_state.copyWith(online: online));
      // Coming back is the single best moment to try; going offline is not
      // worth an attempt that is certain to fail.
      if (online) nudge(SyncTrigger.connectivityRegained);
    });

    _downlinkSub = _downlink?.listen(_onDownlink);

    _pollTimer = _startPoll();

    await _refreshPending();
    nudge(SyncTrigger.startup);
  }

  Timer _startPoll() => _schedule(_idlePoll, () {
        if (_stopped) return;
        _pollTimer = _startPoll();
        nudge(SyncTrigger.periodic);
      });

  void _onDownlink(DownlinkMessage message) {
    switch (message.kind) {
      case DownlinkKind.syncRequested:
        nudge(SyncTrigger.headOfficeRequest);
      case DownlinkKind.catalogueChanged:
        // Sending what we owe before pulling new prices keeps the bills we
        // already rang priced as they were rung.
        nudge(SyncTrigger.headOfficeRequest);
        unawaited(_onCatalogueChanged?.call() ?? Future<void>.value());
      case DownlinkKind.unknown:
        break;
    }
  }

  // -------------------------------------------------------------- Triggers
  /// Asks for a drain. Cheap, non-blocking, and safe to call on every sale.
  void nudge(SyncTrigger trigger) {
    if (_stopped) return;

    if (_draining) {
      // Remember it rather than dropping it: bills committed during this pass
      // were not in the set it read, and would otherwise wait for the poll.
      _queuedTrigger = trigger;
      return;
    }

    if (_state.isHalted && trigger != SyncTrigger.manual) return;
    if (!_state.online && trigger != SyncTrigger.manual) return;

    unawaited(_drain(trigger));
  }

  /// The cashier pressed sync. Runs even when halted or believed offline —
  /// they may know something the engine does not, and being told why it failed
  /// beats a button that does nothing.
  Future<SyncOutcome> syncNow({
    void Function(double progress, String stage)? onProgress,
  }) async {
    _backoffTimer?.cancel();
    _emit(_state.copyWith(isHalted: false, clearNextAttempt: true));
    return _drain(SyncTrigger.manual, onProgress: onProgress);
  }

  // ----------------------------------------------------------------- Drain
  Future<SyncOutcome> _drain(
    SyncTrigger trigger, {
    void Function(double progress, String stage)? onProgress,
  }) async {
    if (_draining) return const SyncOutcome(attempted: 0, uploaded: 0);
    _draining = true;

    _backoffTimer?.cancel();
    _emit(_state.copyWith(
      isSyncing: true,
      lastTrigger: trigger,
      lastAttemptAt: _now(),
      clearNextAttempt: true,
    ),);

    SyncOutcome outcome;
    try {
      try {
        outcome = await _repository.syncOrders(onProgress: onProgress);
      } on Object catch (e) {
        // The repository is meant to fold failures into the outcome; anything
        // escaping it is a defect, not a network fault. Treated as a retryable
        // failure so a bad build still empties its queue once fixed.
        outcome = SyncOutcome(attempted: 0, uploaded: 0, error: e.toString());
      }

      await _refreshPending();

      if (outcome.isSuccess) {
        _emit(_state.copyWith(
          isSyncing: false,
          consecutiveFailures: 0,
          lastSuccessAt: _now(),
          clearError: true,
          clearNextAttempt: true,
        ),);
      } else {
        _onFailure(outcome);
      }
    } finally {
      // Held until the bookkeeping is done, not just until the send is. Freed
      // any earlier, a queued trigger could start a second drain whose
      // `isSyncing: true` this one would then overwrite with `false`, leaving
      // the header claiming idle while an upload is in flight.
      _draining = false;
    }

    // Honour anything that arrived while we were busy. Bills committed during
    // the pass were not in the set it read.
    final queued = _queuedTrigger;
    _queuedTrigger = null;
    if (queued != null && outcome.isSuccess && _state.pending > 0) {
      nudge(queued);
    }

    return outcome;
  }

  void _onFailure(SyncOutcome outcome) {
    final failures = _state.consecutiveFailures + 1;

    if (!outcome.isRetryable) {
      _emit(_state.copyWith(
        isSyncing: false,
        consecutiveFailures: failures,
        lastError: outcome.error,
        isHalted: true,
        clearNextAttempt: true,
      ),);
      return;
    }

    final delay = backoffFor(failures);
    _emit(_state.copyWith(
      isSyncing: false,
      consecutiveFailures: failures,
      lastError: outcome.error,
      nextAttemptAt: _now().add(delay),
    ),);

    _backoffTimer = _schedule(delay, () {
      if (_stopped) return;
      nudge(SyncTrigger.retry);
    });
  }

  /// Doubles per failure to a ceiling, then ±20% so a store's terminals do not
  /// come back in lockstep.
  @visibleForTesting
  Duration backoffFor(int failures) {
    final exponent = (failures - 1).clamp(0, 30);
    final raw = _baseBackoff * pow(2, exponent).toDouble();
    final capped = raw > _maxBackoff ? _maxBackoff : raw;
    final jitter = 0.8 + _random.nextDouble() * 0.4;
    return Duration(
      milliseconds: (capped.inMilliseconds * jitter).round(),
    );
  }

  Future<void> _refreshPending() async {
    try {
      // Read first, then emit. Written as `copyWith(pending: await …)` the
      // receiver `_state` is evaluated before the await completes, so anything
      // that changed during the wait — connectivity dropping, most of all —
      // would be overwritten by the stale snapshot.
      final count = await _repository.unsyncedCount();
      _emit(_state.copyWith(pending: count));
    } on Object {
      // A count is decoration; failing to read it must not fail the drain.
    }
  }

  void _emit(SyncEngineState next) {
    _state = next;
    if (!_states.isClosed) _states.add(next);
  }

  Future<void> dispose() async {
    _stopped = true;
    _backoffTimer?.cancel();
    _pollTimer?.cancel();
    await _connectivitySub?.cancel();
    await _downlinkSub?.cancel();
    await _states.close();
  }
}