nearle_pos/test/unit/sync_engine_test.dart

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

import 'package:flutter_test/flutter_test.dart';
import 'package:nearle_pos/data/remote/order_transport.dart';
import 'package:nearle_pos/data/sync/sync_engine.dart';
import 'package:nearle_pos/domain/entities/shift_report.dart';
import 'package:nearle_pos/domain/entities/sync_event.dart';
import 'package:nearle_pos/domain/entities/transaction.dart';
import 'package:nearle_pos/domain/repositories/sync_repository.dart';

/// A repository whose every answer the test dictates.
///
/// The engine is a scheduler; what it schedules is irrelevant here. Driving it
/// with a stub is what makes "did it retry, and when" answerable without a
/// database or a network.
class _StubRepository implements SyncRepository {
  _StubRepository();

  /// Answers handed out in order; the last one repeats once exhausted.
  final List<SyncOutcome> scripted = [];
  int calls = 0;
  int pending = 0;

  /// Completed by the test to hold a drain open, so overlapping triggers can
  /// be observed.
  Completer<void>? gate;

  @override
  Future<SyncOutcome> syncOrders({
    void Function(double progress, String stage)? onProgress,
  }) async {
    calls++;
    if (gate != null) await gate!.future;
    if (scripted.isEmpty) return const SyncOutcome(attempted: 0, uploaded: 0);
    return scripted[(calls - 1).clamp(0, scripted.length - 1)];
  }

  @override
  Future<int> unsyncedCount() async => pending;

  /// Counted so a test can prove registrations go up before bills do.
  int customerSyncs = 0;

  /// Set to make the registration pass throw, proving it cannot hold up the
  /// bills behind it.
  bool customerSyncThrows = false;

  @override
  Future<int> unsyncedCustomerCount() async => 0;

  @override
  Future<SyncOutcome> syncCustomers() async {
    customerSyncs++;
    if (customerSyncThrows) throw Exception('registration upload failed');
    return const SyncOutcome(attempted: 0, uploaded: 0);
  }

  @override
  Future<int> purgeExpired() async => 0;

  @override
  bool get hasCatalogue => true;

  @override
  DateTime? get lastImportAt => null;

  @override
  String? get catalogueRevision => null;

  @override
  List<SyncEvent> get events => const [];

  @override
  Future<SyncEvent> importCatalogue({
    void Function(double progress, String stage)? onProgress,
  }) =>
      throw UnimplementedError();

  @override
  Future<List<SaleTransaction>> unsyncedOrders() async => const [];

  @override
  Future<List<OrderSyncRow>> orderSyncRows({int limit = 200}) async => const [];

  @override
  Future<ShiftReport> todayReport({
    required String terminalId,
    required String cashierName,
    bool scopeToCashier = false,
  }) =>
      throw UnimplementedError();
}

/// Captures what the engine asked to be scheduled instead of really waiting.
class _FakeScheduler {
  final List<Duration> delays = [];
  final List<void Function()> callbacks = [];

  Timer schedule(Duration d, void Function() cb) {
    delays.add(d);
    callbacks.add(cb);
    return Timer(Duration.zero, () {})..cancel();
  }

  /// Runs the most recently scheduled callback — the backoff retry.
  void fireLast() => callbacks.last();
}

void main() {
  late _StubRepository repo;
  late _FakeScheduler scheduler;

  SyncEngine build({
    Stream<bool>? connectivity,
    Stream<DownlinkMessage>? downlink,
    Random? random,
  }) =>
      SyncEngine(
        repository: repo,
        connectivity: connectivity,
        downlink: downlink,
        // Fixed seed so the jitter band is assertable rather than flaky.
        random: random ?? Random(7),
        scheduleTimer: scheduler.schedule,
      );

  setUp(() {
    repo = _StubRepository();
    scheduler = _FakeScheduler();
  });

  group('single flight', () {
    test('overlapping triggers do not start a second pass over the same bills',
        () async {
      // Without this guarantee a busy till firing a trigger per sale would have
      // several drains reading the same pending rows at once, and every bill
      // would go up two or three times.
      repo.gate = Completer<void>();
      final engine = build();

      engine
        ..nudge(SyncTrigger.saleCommitted)
        ..nudge(SyncTrigger.saleCommitted)
        ..nudge(SyncTrigger.saleCommitted);

      await Future<void>.delayed(Duration.zero);
      expect(repo.calls, 1, reason: 'three triggers must yield one drain');

      repo.gate!.complete();
      repo.gate = null;
      await Future<void>.delayed(Duration.zero);

      await engine.dispose();
    });

    test('a trigger arriving mid-drain is honoured once that drain finishes',
        () async {
      // Bills committed during a pass were not in the set it read. Dropping
      // the trigger would leave them waiting for the next poll.
      repo
        ..gate = Completer<void>()
        ..pending = 3;
      final engine = build();

      engine.nudge(SyncTrigger.startup);
      await Future<void>.delayed(Duration.zero);
      expect(repo.calls, 1);

      engine.nudge(SyncTrigger.saleCommitted);
      repo.gate!.complete();
      repo.gate = null;

      await Future<void>.delayed(Duration.zero);
      await Future<void>.delayed(Duration.zero);

      expect(repo.calls, 2, reason: 'the queued trigger must be replayed');
      await engine.dispose();
    });

    test('a queued trigger with nothing left owing does not run a second pass',
        () async {
      // The counterpart to the test above. The drain that just finished emptied
      // the queue, so replaying the trigger would send nothing and only churn
      // the connection.
      repo
        ..gate = Completer<void>()
        ..pending = 0;
      final engine = build();

      engine.nudge(SyncTrigger.startup);
      await Future<void>.delayed(Duration.zero);

      engine.nudge(SyncTrigger.saleCommitted);
      repo.gate!.complete();
      repo.gate = null;
      await _settle();

      expect(repo.calls, 1);
      await engine.dispose();
    });
  });

  group('backoff', () {
    test('doubles per failure and stops at the ceiling', () {
      final engine = SyncEngine(
        repository: repo,
        baseBackoff: const Duration(seconds: 2),
        maxBackoff: const Duration(minutes: 5),
        // No jitter, so the shape of the curve is what is being asserted.
        random: _ZeroJitter(),
        scheduleTimer: scheduler.schedule,
      );

      // 0.8 is the bottom of the jitter band, which _ZeroJitter pins.
      expect(engine.backoffFor(1).inMilliseconds, 1600); // 2s
      expect(engine.backoffFor(2).inMilliseconds, 3200); // 4s
      expect(engine.backoffFor(3).inMilliseconds, 6400); // 8s
      expect(engine.backoffFor(9).inSeconds, 240); // 512s → capped
      expect(engine.backoffFor(30).inSeconds, 240); // still capped
    });

    test('jitter keeps every delay inside ±20% of the nominal wait', () {
      // A shop's terminals all fail at the same instant when the line drops.
      // Without jitter they would retry in lockstep and keep colliding.
      final engine = build(random: Random(1));

      final seen = <int>{};
      for (var i = 0; i < 200; i++) {
        final ms = engine.backoffFor(4).inMilliseconds;
        expect(ms, greaterThanOrEqualTo((16000 * 0.8).round()));
        expect(ms, lessThanOrEqualTo((16000 * 1.2).round()));
        seen.add(ms);
      }
      expect(seen.length, greaterThan(50), reason: 'delays must actually vary');
    });

    test('a failed drain schedules a retry and a success clears it', () async {
      repo.scripted.addAll([
        const SyncOutcome(attempted: 2, uploaded: 0, error: 'line dropped'),
        const SyncOutcome(attempted: 2, uploaded: 2),
      ]);

      final engine = build();
      engine.nudge(SyncTrigger.saleCommitted);
      await _settle();

      expect(engine.state.consecutiveFailures, 1);
      expect(engine.state.nextAttemptAt, isNotNull);
      expect(scheduler.delays, hasLength(1));

      scheduler.fireLast();
      await _settle();

      expect(engine.state.consecutiveFailures, 0);
      expect(engine.state.lastError, isNull);
      expect(engine.state.nextAttemptAt, isNull);
      expect(engine.state.lastSuccessAt, isNotNull);

      await engine.dispose();
    });
  });

  group('halting', () {
    test('a refused batch halts instead of retrying the same bytes forever',
        () async {
      // A rejection is a decision, not a fault. Re-sending gets the same
      // answer, and a loop would bury the one message a person needs to see.
      repo.scripted.add(const SyncOutcome(
        attempted: 1,
        uploaded: 0,
        rejected: 1,
        error: 'duplicate invoice number',
        isRetryable: false,
      ),);

      final engine = build();
      engine.nudge(SyncTrigger.saleCommitted);
      await _settle();

      expect(engine.state.isHalted, isTrue);
      expect(scheduler.delays, isEmpty, reason: 'no retry may be scheduled');

      // Further background triggers are ignored while halted.
      engine
        ..nudge(SyncTrigger.periodic)
        ..nudge(SyncTrigger.saleCommitted);
      await _settle();
      expect(repo.calls, 1);

      await engine.dispose();
    });

    test('pressing sync clears a halt and tries again', () async {
      // The button is how a cashier retries once the back office is fixed.
      repo.scripted.add(const SyncOutcome(
        attempted: 1,
        uploaded: 0,
        error: 'bad credential',
        isRetryable: false,
      ),);

      final engine = build();
      engine.nudge(SyncTrigger.saleCommitted);
      await _settle();
      expect(engine.state.isHalted, isTrue);

      repo.scripted
        ..clear()
        ..add(const SyncOutcome(attempted: 1, uploaded: 1));
      repo.calls = 0;

      await engine.syncNow();
      expect(repo.calls, 1);
      expect(engine.state.isHalted, isFalse);

      await engine.dispose();
    });
  });

  group('triggers', () {
    test('the network coming back starts a drain; going away does not',
        () async {
      final connectivity = StreamController<bool>();
      final engine = build(connectivity: connectivity.stream);
      await engine.start();

      // start() fires a drain of its own. Let it finish before taking the
      // baseline, so this measures what connectivity did rather than how many
      // awaits happen to sit in front of the repository call.
      await _settle();
      final atStart = repo.calls;

      connectivity.add(false);
      await _settle();
      expect(repo.calls, atStart,
          reason: 'an attempt certain to fail is not worth making',);
      expect(engine.state.online, isFalse);

      connectivity.add(true);
      await _settle();
      expect(repo.calls, atStart + 1);

      await connectivity.close();
      await engine.dispose();
    });

    test('background triggers are ignored while offline, manual is not',
        () async {
      final connectivity = StreamController<bool>();
      final engine = build(connectivity: connectivity.stream);
      await engine.start();

      connectivity.add(false);
      await _settle();
      final atStart = repo.calls;

      engine.nudge(SyncTrigger.saleCommitted);
      await _settle();
      expect(repo.calls, atStart);

      // The cashier may know something the engine does not, and being told why
      // it failed beats a button that does nothing.
      await engine.syncNow();
      expect(repo.calls, atStart + 1);

      await connectivity.close();
      await engine.dispose();
    });

    test('head office can pull a shift up over the downlink', () async {
      final downlink = StreamController<DownlinkMessage>();
      // Bills owed, otherwise a request to sync correctly does nothing.
      repo.pending = 2;
      final engine = build(downlink: downlink.stream);
      await engine.start();
      // Let the startup drain finish, so this measures the downlink and not a
      // race with it.
      await _settle();

      final atStart = repo.calls;
      downlink.add(const DownlinkMessage(kind: DownlinkKind.syncRequested));
      await _settle();

      expect(repo.calls, atStart + 1);

      await downlink.close();
      await engine.dispose();
    });

    test('an unrecognised downlink message is ignored, not acted on', () async {
      final downlink = StreamController<DownlinkMessage>();
      repo.pending = 2;
      final engine = build(downlink: downlink.stream);
      await engine.start();
      await _settle();

      final atStart = repo.calls;
      downlink.add(const DownlinkMessage(kind: DownlinkKind.unknown));
      await _settle();

      expect(repo.calls, atStart);

      await downlink.close();
      await engine.dispose();
    });
  });

  group('registrations', () {
    test('go up before the bills that might refer to them', () async {
      final engine = build();

      engine.nudge(SyncTrigger.saleCommitted);
      await _settle();

      expect(repo.customerSyncs, 1);
      expect(repo.calls, 1);
    });

    test('failing to upload one cannot strand a day of takings', () async {
      // A shopper waiting to go up must never be the reason money stays on the
      // terminal. The registration pass is allowed to fail on its own.
      repo.customerSyncThrows = true;
      final engine = build();

      engine.nudge(SyncTrigger.saleCommitted);
      await _settle();

      expect(repo.calls, 1, reason: 'the bills still went');
      expect(engine.state.consecutiveFailures, 0);
      expect(engine.state.isHalted, isFalse);

      await engine.dispose();
    });
  });

  test('a repository that throws is treated as a retryable failure, not a crash',
      () async {
    // Anything escaping the repository is a defect. The engine must still empty
    // its queue once the defect is fixed, rather than dying on the first sale.
    final engine = SyncEngine(
      repository: _ThrowingRepository(),
      random: Random(3),
      scheduleTimer: scheduler.schedule,
    );

    engine.nudge(SyncTrigger.saleCommitted);
    await _settle();

    expect(engine.state.consecutiveFailures, 1);
    expect(engine.state.isHalted, isFalse);
    expect(scheduler.delays, hasLength(1));

    await engine.dispose();
  });
}

/// Pins the jitter multiplier at its lower bound so the backoff curve itself
/// can be asserted.
class _ZeroJitter implements Random {
  @override
  double nextDouble() => 0;

  @override
  bool nextBool() => false;

  @override
  int nextInt(int max) => 0;
}

class _ThrowingRepository extends _StubRepository {
  @override
  Future<SyncOutcome> syncOrders({
    void Function(double progress, String stage)? onProgress,
  }) async =>
      throw StateError('boom');
}

/// Lets queued microtasks run. The engine never really waits, so a handful of
/// turns is enough for everything it schedules to settle.
Future<void> _settle() async {
  for (var i = 0; i < 6; i++) {
    await Future<void>.delayed(Duration.zero);
  }
}