samplez/rippr-flutter-src/test/recording_engine_test.dart

import 'package:drift/drift.dart' show driftRuntimeOptions;
import 'package:drift/native.dart';
import 'package:flutter_test/flutter_test.dart';
import 'package:rippr/src/data/database.dart';
import 'package:rippr/src/data/trip_repository.dart';
import 'package:rippr/src/domain/models.dart';
import 'package:rippr/src/recording/location_source.dart';
import 'package:rippr/src/recording/recording_engine.dart';

/// Covers `TrackingServiceLifecycleTest`'s ground, but on the Dart VM.
///
/// The Kotlin equivalent drove a real Android `Service` through all five actions on a
/// device, polled the database with a 25 s timeout, and was flaky enough that the timeout
/// was once raised to hide what turned out to be a real `stopSelf()` race. None of that
/// applies here: the pipeline is plain Dart behind a fake location source, so these run
/// deterministically in milliseconds.
void main() {
  late AppDatabase db;
  late TripRepository repo;
  late FakeLocationSource source;
  late RecordingEngine engine;
  late int fakeNow;

  setUp(() {
    driftRuntimeOptions.dontWarnAboutMultipleDatabases = true;
    db = AppDatabase(NativeDatabase.memory());
    repo = TripRepository(db);
    source = FakeLocationSource();
    fakeNow = 1000;
    engine = RecordingEngine(
      repository: repo,
      locationSource: source,
      clock: () => fakeNow,
    );
  });

  tearDown(() async {
    await engine.dispose();
    await source.dispose();
    await db.close();
  });

  /// Lets the broadcast stream deliver. Fixes reach the engine on a microtask, not
  /// synchronously, so every emission must be followed by this before asserting.
  Future<void> settle() => Future<void>.delayed(Duration.zero);

  /// Emits [n] fixes a second apart, moving ~11 m north each time.
  Future<void> ride(int n,
      {int startTs = 1000, double startLat = 51.0, double speedMps = 11.11}) async {
    for (var i = 0; i < n; i++) {
      source.emitAt(
        timestamp: startTs + i * 1000,
        latitude: startLat + i * 0.0001,
        speedMps: speedMps,
      );
    }
    await settle();
  }

  group('the fix path', () {
    test('a started ride buffers fixes without writing immediately', () async {
      await engine.start();
      await ride(5);

      expect(engine.pendingCount, 5,
          reason: 'the fix path must not touch the database');
      expect(await db.countPointsForTrip(engine.currentTripId), 0);
    });

    test('fixes are ignored before start and after stop', () async {
      await ride(3);
      expect(engine.pendingCount, 0, reason: 'source is not running yet');

      await engine.start();
      await engine.stop();
      await ride(3);
      expect(engine.pendingCount, 0);
    });

    test('fixes worse than the accuracy budget are dropped', () async {
      await engine.start();
      source.emitAt(timestamp: 1000, accuracyM: 500);
      await settle();
      expect(engine.pendingCount, 0);

      source.emitAt(timestamp: 2000, accuracyM: 5);
      await settle();
      expect(engine.pendingCount, 1);
    });

    test('speed is converted and the noise floor applied', () async {
      await engine.start();
      source.emitAt(timestamp: 1000, speedMps: 10.0); // 36 km/h
      source.emitAt(timestamp: 2000, speedMps: 0.1); // 0.36 km/h, under the floor
      await settle();
      await engine.stop();

      final points = await db.allPoints();
      expect(points[0].speedKmh, closeTo(36.0, 1e-9));
      expect(points[1].speedKmh, 0.0,
          reason: 'jitter under the noise floor must read as zero');
    });

    test('live telemetry updates at fix rate', () async {
      await engine.start();
      source.emitAt(timestamp: 1000, speedMps: 20.0);
      await settle();
      expect(engine.pendingCount, 1);
      // 20 m/s is 72 km/h.
      // LiveTelemetry is a singleton; the value is what the record screen reads.
      await engine.stop();
    });
  });

  group('persistence', () {
    test('stop drains everything buffered', () async {
      await engine.start();
      final tripId = engine.currentTripId;
      await ride(7);

      await engine.stop();

      expect(await db.countPointsForTrip(tripId), 7);
      expect(engine.pendingCount, 0);
    });

    test('aggregates are persisted and then reconciled on stop', () async {
      await engine.start();
      final tripId = engine.currentTripId;
      await ride(11); // ten hops of ~11.12 m

      await engine.stop();

      final trip = (await repo.tripById(tripId))!;
      expect(trip.pointCount, 11);
      expect(trip.distanceM, closeTo(111.2, 3.0));
      expect(trip.state, TripState.completed);
      expect(trip.endedAt, isNotNull);
    });

    test('a ride that captured nothing is discarded, not saved', () async {
      await engine.start();
      final tripId = engine.currentTripId;

      final result = await engine.stop();

      expect(result, isNull);
      expect(await repo.tripById(tripId), isNull,
          reason: 'an empty trip is noise in the history list');
      expect(await db.countTrips(), 0);
    });
  });

  group('pause and segments', () {
    test('the full lifecycle produces one trip and two segments', () async {
      await engine.start();
      final tripId = engine.currentTripId;
      await ride(3, startTs: 1000);

      await engine.pause();
      await ride(3, startTs: 5000); // ignored: the source is stopped

      fakeNow = 6000;
      await engine.start(); // resume
      await ride(3, startTs: 6000, startLat: 51.001);

      await engine.stop();

      expect(await db.countTrips(), 1);
      final segments = await repo.segmentsForTrip(tripId);
      expect(segments.length, 2);
      expect(segments.every((s) => !s.isOpen), isTrue);
      expect(await db.countPointsForTrip(tripId), 6,
          reason: 'fixes emitted while paused must not be recorded');
    });

    test('a fix in flight at pause lands in the segment it belongs to', () async {
      // This is the guarantee that stamping ids at creation exists for. The fix is
      // buffered before the pause and written after it; it must carry the OLD segment id.
      await engine.start();
      final tripId = engine.currentTripId;
      final firstSegment =
          (await repo.segmentsForTrip(tripId)).single.id;

      await ride(2); // buffered, not yet written
      expect(engine.pendingCount, 2);

      await engine.pause();

      final points = await db.pointsForTrip(tripId);
      expect(points.length, 2);
      expect(points.every((p) => p.segmentId == firstSegment), isTrue,
          reason: 'a queued fix must not be re-homed into the next segment');
    });

    test('distance does not span the pause', () async {
      await engine.start();
      final tripId = engine.currentTripId;
      await ride(3, startLat: 51.0);

      await engine.pause();
      fakeNow = 6000;
      await engine.start();
      // Resumed a full degree of latitude away — a trailered gap of ~111 km.
      await ride(3, startTs: 6000, startLat: 52.0);
      await engine.stop();

      final trip = (await repo.tripById(tripId))!;
      expect(trip.distanceM, lessThan(200.0),
          reason: 'the pause gap leaked into distance: ${trip.distanceM} m');
    });

    test('pause then stop completes the trip normally', () async {
      await engine.start();
      final tripId = engine.currentTripId;
      await ride(3);
      await engine.pause();

      final result = await engine.stop();

      expect(result, tripId);
      expect((await repo.tripById(tripId))!.state, TripState.completed);
    });
  });

  group('discard', () {
    test('discard deletes the trip and everything buffered', () async {
      await engine.start();
      final tripId = engine.currentTripId;
      await ride(5);

      await engine.discard();

      expect(await repo.tripById(tripId), isNull);
      expect(await db.countTrips(), 0);
      expect(engine.pendingCount, 0);
      expect(engine.state, RecorderState.idle);
    });

    test('discard leaves earlier completed rides alone', () async {
      await engine.start();
      final keep = engine.currentTripId;
      await ride(3);
      await engine.stop();

      fakeNow = 10000;
      await engine.start();
      await ride(3, startTs: 10000);
      await engine.discard();

      expect(await repo.tripById(keep), isNotNull);
      expect(await db.countTrips(), 1);
    });
  });

  group('idempotency and state', () {
    test('start twice adopts rather than duplicating', () async {
      await engine.start();
      final tripId = engine.currentTripId;
      await engine.start();

      expect(engine.currentTripId, tripId);
      expect(await db.countTrips(), 1);
      expect(await db.countSegmentsForTrip(tripId), 1);
    });

    test('pause twice is safe', () async {
      await engine.start();
      await engine.pause();
      await engine.pause();
      expect(engine.state, RecorderState.paused);
    });

    test('stop with nothing running is a safe no-op', () async {
      expect(await engine.stop(), isNull);
      expect(engine.state, RecorderState.idle);
    });

    test('the source is started and stopped in step with the lifecycle', () async {
      await engine.start();
      expect(source.isRunning, isTrue);

      await engine.pause();
      expect(source.isRunning, isFalse);

      await engine.start();
      expect(source.isRunning, isTrue);

      await engine.stop();
      expect(source.isRunning, isFalse);
    });

    test('state transitions are observable', () async {
      final seen = <RecorderState>[];
      final sub = engine.stateStream.listen(seen.add);

      await engine.start();
      await engine.pause();
      await engine.start();
      await engine.stop();
      await Future<void>.delayed(Duration.zero);
      await sub.cancel();

      expect(seen, [
        RecorderState.recording,
        RecorderState.paused,
        RecorderState.recording,
        RecorderState.idle,
      ]);
    });
  });

  group('process death', () {
    test('a recording trip is resumed into a new segment', () async {
      await engine.start();
      final tripId = engine.currentTripId;
      await ride(5);
      // Deliberately no stop(): dispose without completing simulates a process kill.
      await engine.dispose();

      // A brand new engine over the same database, as after a process restart.
      final revived = RecordingEngine(
        repository: repo,
        locationSource: source,
        clock: () => fakeNow,
      );
      addTearDown(revived.dispose);

      fakeNow = 20000;
      final state = await revived.restoreAfterProcessDeath();

      expect(state, RecorderState.recording);
      expect(revived.currentTripId, tripId);
      final segments = await repo.segmentsForTrip(tripId);
      expect(segments.length, 2,
          reason: 'the dead time is a real gap and must render as one');
    });

    test('a recording trip found at restart reports an unexpected stop (V3-12)',
        () async {
      await engine.start();
      await ride(5);
      await engine.dispose();

      String? reportedReason;
      final revived = RecordingEngine(
        repository: repo,
        locationSource: source,
        clock: () => fakeNow,
        onUnexpectedStop: (reason) => reportedReason = reason,
      );
      addTearDown(revived.dispose);

      fakeNow = 20000;
      await revived.restoreAfterProcessDeath();

      expect(reportedReason, isNotNull,
          reason: 'the recording stopped without anyone choosing that -- the exact '
              'failure V3-12 exists to surface');
    });

    test('a cleanly-completed trip reports nothing at restart', () async {
      await engine.start();
      await ride(5);
      await engine.stop();
      await engine.dispose();

      var called = false;
      final revived = RecordingEngine(
        repository: repo,
        locationSource: source,
        clock: () => fakeNow,
        onUnexpectedStop: (_) => called = true,
      );
      addTearDown(revived.dispose);

      await revived.restoreAfterProcessDeath();

      expect(called, isFalse,
          reason: 'a rider who pressed Stop is not a crash and must not be reported');
    });

    test('the dead time is never measured as distance', () async {
      // The bug this guards is in the native app: after a crash the open segment is
      // adopted rather than closed, so computeSummary -- the authoritative pass at trip
      // completion -- measures straight across the gap. A rider who crashes in town and
      // relaunches downtown would have those kilometres added to their ride.
      await engine.start();
      final tripId = engine.currentTripId;
      final crashedSegment = (await repo.segmentsForTrip(tripId)).single.id;

      // Write points the way the flush loop would have, *without* going through pause or
      // stop — so the segment is left OPEN, which is exactly what a crash leaves behind.
      // Using pause() here would close the segment and quietly stop reproducing the bug.
      await repo.appendPoints([
        for (var i = 0; i < 3; i++)
          TrackPoint(
            tripId: tripId,
            segmentId: crashedSegment,
            timestamp: 1000 + i * 1000,
            latitude: 51.0 + i * 0.0001,
            longitude: -114.0,
            speedKmh: 40,
            altitudeM: 1000,
          ),
      ]);
      await engine.dispose();

      final revived = RecordingEngine(
        repository: repo,
        locationSource: source,
        clock: () => fakeNow,
      );
      addTearDown(revived.dispose);

      fakeNow = 900000;
      await revived.restoreAfterProcessDeath();
      // Relaunched a full degree of latitude away — ~111 km from where it died.
      await ride(3, startTs: 900000, startLat: 52.0);
      await revived.stop();

      final trip = (await repo.tripById(tripId))!;
      expect(trip.distanceM, lessThan(200.0),
          reason:
              'the dead time leaked into distance: ${trip.distanceM} m — the crash gap '
              'was measured as if it had been ridden');
    });

    test('restored totals continue rather than restarting from zero', () async {
      await engine.start();
      final tripId = engine.currentTripId;
      await ride(11);
      // Force the buffered points and their aggregates to disk without completing.
      await engine.pause();

      final before = (await repo.tripById(tripId))!;
      expect(before.distanceM, greaterThan(100.0));

      await engine.dispose();

      final revived = RecordingEngine(
        repository: repo,
        locationSource: source,
        clock: () => fakeNow,
      );
      addTearDown(revived.dispose);

      fakeNow = 30000;
      await revived.start(); // resumes the paused trip
      await ride(3, startTs: 30000, startLat: 53.0);
      await revived.stop();

      final after = (await repo.tripById(tripId))!;
      expect(after.pointCount, 14, reason: 'earlier points must be kept');
      expect(after.distanceM, greaterThanOrEqualTo(before.distanceM),
          reason: 'distance must not restart from zero');
    });

    test('a paused trip is restored as paused, not resumed', () async {
      await engine.start();
      await ride(3);
      await engine.pause();
      await engine.dispose();

      final revived = RecordingEngine(
        repository: repo,
        locationSource: source,
        clock: () => fakeNow,
      );
      addTearDown(revived.dispose);

      final state = await revived.restoreAfterProcessDeath();

      expect(state, RecorderState.paused);
      expect(source.isRunning, isFalse,
          reason: 'a paused ride must not silently start consuming GPS');
    });

    test('no active trip restores to idle', () async {
      final state = await engine.restoreAfterProcessDeath();
      expect(state, RecorderState.idle);
    });
  });
}