// RID-3A regression coverage: telemetry batch ownership / exactly-once buffer.
//
// Converted from the RID-3 forensic reproduction. Before this fix, two
// overlapping legitimate flushes (the 3 s periodic timer against the flush
// awaited by finishRide) shared one mutable buffer that was cleared AFTER the
// persistence await by count rather than identity. For 30 accepted samples that
// produced 30 rows representing only 20 distinct samples: indices 0-9 twice,
// indices 10-19 destroyed, indices 20-29 once - while the counters reported a
// perfect 30/30.
//
// Race windows are controlled with Completer gates on db.batch, not sleeps. The
// only real-time waits are for the production Timer.periodic(3 s) to fire,
// because that interval is a hardcoded constant with no injectable seam and
// creating one would be a production change.
import 'dart:async';

import 'package:flutter/foundation.dart';
import 'package:flutter_test/flutter_test.dart';
import 'package:flutter_riverpod/flutter_riverpod.dart';
import 'package:geolocator/geolocator.dart';
import 'package:shared_preferences/shared_preferences.dart';
import 'package:drift/native.dart';
import 'package:drift/drift.dart' as drift;
import 'package:motorcycle_dash/database/database.dart'
    hide TelemetrySample, FieldTestSummary, FieldTestEvent;
import 'package:motorcycle_dash/database/database.dart'
    as dbgen
    show TelemetrySample;
import 'package:motorcycle_dash/models/telemetry_sample.dart';
import 'package:motorcycle_dash/models/field_test_summary.dart';
import 'package:motorcycle_dash/models/field_test_event.dart';
import 'package:motorcycle_dash/services/telemetry_service.dart';
import 'package:motorcycle_dash/services/location_service.dart';
import 'package:motorcycle_dash/services/sensor_fusion_service.dart';
import 'package:motorcycle_dash/services/compass_service.dart';
import 'package:motorcycle_dash/services/audio_alert_service.dart';
import 'package:motorcycle_dash/services/field_test/field_test_logger.dart';
import 'package:motorcycle_dash/services/validation/telemetry_validator.dart';
import 'package:motorcycle_dash/features/rides/ride_state_machine.dart';
import 'package:motorcycle_dash/models/recording_health_status.dart';
import 'package:motorcycle_dash/features/records/records_service.dart';

/// Drift's generated row class for telemetry_samples is also named
/// TelemetrySample, colliding with the telemetry model.
typedef TelemetryRow = dbgen.TelemetrySample;

/// Gates, counts and can fail db.batch - the persistence API used by
/// RideStateMachine._flushBatch() (ride_state_machine.dart:560).
class _Rid3aAppDatabase extends AppDatabase {
  _Rid3aAppDatabase() : super.forTesting(NativeDatabase.memory());

  int batchInvocations = 0;
  bool gateBatch = false;
  final List<Completer<void>> batchGates = [];
  int batchFailuresRemaining = 0;

  @override
  Future<void> batch(
    FutureOr<void> Function(drift.Batch batch) runInBatch,
  ) async {
    batchInvocations++;
    // Gate FIRST, then optionally fail, so a flush can be parked in-flight and
    // made to fail on release - required by the ordered-restoration test.
    if (gateBatch) {
      final gate = Completer<void>();
      batchGates.add(gate);
      await gate.future;
    }
    if (batchFailuresRemaining > 0) {
      batchFailuresRemaining--;
      throw StateError('RID-3A simulated telemetry persistence failure');
    }
    return super.batch(runInBatch);
  }
}

/// FieldTestLogger runs its OWN 3 s periodic flush() -> _db.batch
/// (field_test_logger.dart:138, :592). Those batches are indistinguishable from
/// telemetry _flushBatch() batches at the executor level, so the logger flush is
/// neutralised here to keep db.batch accounting unambiguous. Field-test event
/// persistence is not under test in RID-3A.
class _Rid3aFieldTestLogger extends FieldTestLogger {
  _Rid3aFieldTestLogger(super.db);

  int stopSessionInvocations = 0;
  int overflowEventCount = 0;
  final List<FieldTestEventType> loggedEventTypes = [];

  @override
  void log({
    required FieldTestEventType eventType,
    required EventSeverity severity,
    required String message,
    DateTime? timestamp,
    int? rideId,
    DateTime? telemetryTimestamp,
    int? gpsAgeMs,
    double? speed,
    double? latitude,
    double? longitude,
    double? gpsAccuracy,
    double? leanAngle,
    double? longitudinalG,
    double? lateralG,
    double? overallConfidence,
    Map<String, dynamic>? metadata,
  }) {
    loggedEventTypes.add(eventType);
    if (eventType == FieldTestEventType.databaseQueueOverflow) {
      overflowEventCount++;
    }
    super.log(
      eventType: eventType,
      severity: severity,
      message: message,
      timestamp: timestamp,
      rideId: rideId,
      telemetryTimestamp: telemetryTimestamp,
      gpsAgeMs: gpsAgeMs,
      speed: speed,
      latitude: latitude,
      longitude: longitude,
      gpsAccuracy: gpsAccuracy,
      leanAngle: leanAngle,
      longitudinalG: longitudinalG,
      lateralG: lateralG,
      overallConfidence: overallConfidence,
      metadata: metadata,
    );
  }

  @override
  Future<void> flush() async {}

  @override
  Future<FieldTestSummary> stopSession() async {
    stopSessionInvocations++;
    return super.stopSession();
  }
}

class MockAudioAlertService extends Fake implements AudioAlertService {
  final List<String> spokenMessages = [];
  @override
  Future<void> speak(String text, {bool force = false}) async {
    spokenMessages.add(text);
  }
}

class FakeTelemetryService extends TelemetryService {
  final StreamController<TelemetrySample> _controller =
      StreamController<TelemetrySample>.broadcast();

  FakeTelemetryService(
    super.loc,
    super.fusion,
    super.compass,
    super.logger,
    super.validator,
  );

  @override
  Stream<TelemetrySample> get telemetryStream => _controller.stream;

  void emitSample(TelemetrySample sample) => _controller.add(sample);

  @override
  void dispose() => _controller.close();
}

class _NullLocationService extends Fake implements LocationService {
  @override
  void Function(Position position, bool accepted, String? rejectionReason)?
  onGpsDiagnostic;
  @override
  Stream<Position> getFilteredPositionStream() => const Stream.empty();
}

class _NullSensorFusionService extends Fake implements SensorFusionService {
  @override
  Stream<double> get leanAngleStream => const Stream.empty();
  @override
  Stream<double> get pitchAngleStream => const Stream.empty();
  @override
  Stream<GForceData> get gForceStream => const Stream.empty();
  @override
  double get latestRollRateDegPerSec => 0.0;
  @override
  double get latestRawAccelX => 0.0;
  @override
  double get latestRawAccelY => 0.0;
  @override
  double get latestRawAccelZ => 0.0;
  @override
  double get latestRawGyroX => 0.0;
  @override
  double get latestRawGyroY => 0.0;
  @override
  double get latestRawGyroZ => 0.0;
}

class _NullCompassService extends Fake implements CompassService {
  @override
  Stream<double> get headingStream => const Stream.empty();
}

void main() {
  TestWidgetsFlutterBinding.ensureInitialized();

  /// The production flush interval is Timer.periodic(3 s)
  /// (ride_state_machine.dart:145). Waiting slightly longer guarantees exactly
  /// one tick deterministically.
  const tickWait = Duration(milliseconds: 3300);

  late _Rid3aAppDatabase db;
  late _Rid3aFieldTestLogger logger;
  late FakeTelemetryService fakeTelemetry;
  late MockAudioAlertService mockAudio;
  late TelemetryValidator validator;
  late SharedPreferences prefs;

  final sampleBase = DateTime(2026, 9, 10, 12);

  setUp(() async {
    SharedPreferences.setMockInitialValues({});
    prefs = await SharedPreferences.getInstance();
    db = _Rid3aAppDatabase();
    logger = _Rid3aFieldTestLogger(db);
    validator = TelemetryValidator();
    fakeTelemetry = FakeTelemetryService(
      _NullLocationService(),
      _NullSensorFusionService(),
      _NullCompassService(),
      logger,
      validator,
    );
    mockAudio = MockAudioAlertService();
  });

  tearDown(() async {
    for (final g in db.batchGates) {
      if (!g.isCompleted) g.complete();
    }
    db.batchGates.clear();
    db.gateBatch = false;
    fakeTelemetry.dispose();
    logger.dispose();
    await db.close();
  });

  ProviderContainer buildContainer() {
    final container = ProviderContainer(
      overrides: [
        sharedPreferencesProvider.overrideWithValue(prefs),
        databaseProvider.overrideWithValue(db),
        telemetryServiceProvider.overrideWithValue(fakeTelemetry),
        audioAlertServiceProvider.overrideWithValue(mockAudio),
        fieldTestLoggerProvider.overrideWithValue(logger),
      ],
    );
    addTearDown(container.dispose);
    return container;
  }

  Future<int> startRide(ProviderContainer container) async {
    final notifier = container.read(rideStateMachineProvider.notifier);
    await notifier.startRide();
    return container.read(rideStateMachineProvider).dbId!;
  }

  /// Emits [count] uniquely identifiable samples from index [startIndex].
  /// Identity is carried in `speed` (1000 + index) so every persisted row traces
  /// back to exactly one injected sample, independent of DateTime precision.
  void emitSamples(int startIndex, int count) {
    for (var i = 0; i < count; i++) {
      final index = startIndex + i;
      fakeTelemetry.emitSample(
        TelemetrySample(
          timestamp: sampleBase.add(Duration(milliseconds: index * 50)),
          latitude: 34.0 + (index * 0.0001),
          longitude: -85.0,
          speed: 1000.0 + index,
          heading: 90.0,
          gpsAccuracy: 3.0,
        ),
      );
    }
  }

  Future<List<TelemetryRow>> persistedRows() =>
      (db.select(db.telemetrySamples)..orderBy([
            (t) => drift.OrderingTerm(
              expression: t.timestamp,
              mode: drift.OrderingMode.asc,
            ),
            (t) => drift.OrderingTerm(
              expression: t.id,
              mode: drift.OrderingMode.asc,
            ),
          ]))
          .get();

  Map<int, int> countsByIndex(List<TelemetryRow> rows) {
    final counts = <int, int>{};
    for (final row in rows) {
      final speed = row.speed;
      if (speed == null || speed < 1000) continue;
      counts[(speed - 1000.0).round()] =
          (counts[(speed - 1000.0).round()] ?? 0) + 1;
    }
    return counts;
  }

  List<int> duplicatedIn(Map<int, int> counts) =>
      counts.entries.where((e) => e.value > 1).map((e) => e.key).toList()
        ..sort();

  List<int> missingOf(int total, Map<int, int> counts) => [
    for (var i = 0; i < total; i++)
      if ((counts[i] ?? 0) == 0) i,
  ];

  group('RID-3A: telemetry batch ownership', () {
    test('single flush persists every sample exactly once, in order', () async {
      final container = buildContainer();
      final notifier = container.read(rideStateMachineProvider.notifier);
      final rideId = await startRide(container);

      emitSamples(0, 10);
      await pumpEventQueue(times: 30);
      expect(container.read(rideStateMachineProvider).totalSamplesReceived, 10);

      // One legitimate flush source, no competition.
      await notifier.finishRide();
      await pumpEventQueue(times: 30);

      final rows = await persistedRows();
      final counts = countsByIndex(rows);
      final session = container.read(rideStateMachineProvider);

      expect(db.batchInvocations, 1);
      expect(rows, hasLength(10), reason: 'N accepted -> N durable rows');
      expect(counts.length, 10, reason: 'N distinct identities');
      expect(duplicatedIn(counts), isEmpty, reason: '0 duplicates');
      expect(missingOf(10, counts), isEmpty, reason: '0 missing');
      expect(rows.map((r) => r.speed).toList(), [
        for (var i = 0; i < 10; i++) 1000.0 + i,
      ], reason: 'correct chronological order');
      expect(rows.map((r) => r.rideId).toSet(), {
        rideId,
      }, reason: 'correct ride association');
      expect(session.totalSamplesReceived, 10);
      expect(
        session.totalSamplesPersisted,
        10,
        reason: 'buffer empty after a successful final flush',
      );
    });

    test('overlapping periodic and finishRide flushes persist all 30 accepted '
        'samples exactly once', () async {
      final container = buildContainer();
      final notifier = container.read(rideStateMachineProvider.notifier);
      final rideId = await startRide(container);

      // Batch A: indices 0..9.
      emitSamples(0, 10);
      await pumpEventQueue(times: 30);

      db.gateBatch = true;
      // Tick 1: the periodic flush takes exclusive ownership of 0..9, removes
      // them from the live buffer, and parks inside db.batch.
      await Future<void>.delayed(tickWait);
      expect(
        db.batchInvocations,
        1,
        reason: 'periodic flush parked in db.batch',
      );
      final flushA = db.batchGates.single;

      // Newer telemetry (indices 10..29) arrives while A is still in flight.
      // Injected BEFORE the competing flush so the live buffer is provably
      // non-empty - otherwise the competing flush's exclusion could be
      // attributed to an empty buffer rather than to the ownership guard.
      emitSamples(10, 20);
      await pumpEventQueue(times: 40);
      expect(container.read(rideStateMachineProvider).totalSamplesReceived, 30);

      // Competing legitimate flush: finishRide() now awaits the active owner
      // (RID-3B-1) instead of observing a no-op. While flush A is still parked it
      // must take no ownership of A's snapshot and issue no batch of its own; it
      // drains the newer samples only after A has committed.
      final finishing = notifier.finishRide();
      await pumpEventQueue(times: 30);
      expect(
        db.batchInvocations,
        1,
        reason: 'while the owner is parked, no second flush may take ownership',
      );

      // Release ONLY flush A, leaving the gate armed so the RID-3B-1 durability
      // barrier's own drain parks as the next gate. That yields a deterministic
      // observation point at which flush A's committed write is isolated: A has
      // persisted and nothing else has been written yet.
      flushA.complete();
      await pumpEventQueue(times: 40);

      final rowsOwnedByA = await persistedRows();
      final countsOwnedByA = countsByIndex(rowsOwnedByA);
      expect(
        duplicatedIn(countsOwnedByA),
        isEmpty,
        reason: 'no duplication from overlapping ownership',
      );
      expect(
        missingOf(10, countsOwnedByA),
        isEmpty,
        reason: 'A persisted its whole owned snapshot',
      );
      // The real RID-3A ownership property: flush A's write contains exactly its
      // own identity set, and none of the newer samples - those remain buffered
      // for later ownership. This is observed BEFORE finishRide can complete the
      // durability barrier, not after `await finishing`, because RID-3B-1 now
      // intentionally drains all accepted pre-boundary telemetry before
      // finalizing. The previous "only 10 rows after await finishing" expectation
      // depended on the finish flush being a no-op, which RID-3B-1 removed.
      expect(
        countsOwnedByA.keys.toList()..sort(),
        [for (var i = 0; i < 10; i++) i],
        reason: 'flush A owns exactly indices 0-9',
      );
      expect(
        countsOwnedByA.keys.where((i) => i >= 10).toList(),
        isEmpty,
        reason: 'no newer sample was included in flush A write',
      );
      expect(
        db.batchGates,
        hasLength(2),
        reason: 'the barrier now owns the newer batch and is parked',
      );

      // Release the barrier's drain and let finishRide complete. It persists
      // 10..29 itself, so no later periodic flush is required.
      db.gateBatch = false;
      db.batchGates.last.complete();
      await finishing;
      await pumpEventQueue(times: 40);

      final rows = await persistedRows();
      final counts = countsByIndex(rows);
      final session = container.read(rideStateMachineProvider);

      debugPrint('==== RID-3A OVERLAP (after fix) ====');
      debugPrint('batch invocations       : ${db.batchInvocations}');
      debugPrint('accepted samples        : ${session.totalSamplesReceived}');
      debugPrint('durable rows            : ${rows.length}');
      debugPrint('distinct identities     : ${counts.length}');
      debugPrint('duplicated indices      : ${duplicatedIn(counts)}');
      debugPrint('missing indices         : ${missingOf(30, counts)}');
      debugPrint(
        'received / persisted    : '
        '${session.totalSamplesReceived} / ${session.totalSamplesPersisted}',
      );
      debugPrint(
        'rideIds                 : ${rows.map((r) => r.rideId).toSet()}',
      );

      // Previously: 30 rows / 20 distinct, indices 0-9 twice, 10-19 destroyed.
      expect(rows, hasLength(30), reason: '30 accepted -> 30 durable rows');
      expect(counts.length, 30, reason: '30 distinct identities');
      expect(duplicatedIn(counts), isEmpty, reason: '0 duplicates');
      expect(missingOf(30, counts), isEmpty, reason: '0 missing');
      expect(
        rows.map((r) => r.speed).toList(),
        [for (var i = 0; i < 30; i++) 1000.0 + i],
        reason: 'chronological identity order preserved',
      );
      expect(
        rows.map((r) => r.rideId).toSet(),
        {rideId},
        reason: 'every sample against the correct ride',
      );
      expect(
        session.totalSamplesPersisted,
        30,
        reason:
            'counters agree with distinct durable samples; nothing left '
            'intentionally buffered after the final flush',
      );
    });

    test('telemetry arriving during an in-flight flush is persisted exactly '
        'once, after the owned batch', () async {
      final container = buildContainer();
      final rideId = await startRide(container);

      // Batch A: indices 0..4.
      emitSamples(0, 5);
      await pumpEventQueue(times: 30);

      db.gateBatch = true;
      await Future<void>.delayed(tickWait);
      expect(db.batchInvocations, 1);
      final flushA = db.batchGates.single;

      // Batch B: indices 5..9, arriving while A's persistence is parked.
      emitSamples(5, 5);
      await pumpEventQueue(times: 30);
      expect(container.read(rideStateMachineProvider).totalSamplesReceived, 10);

      db.gateBatch = false;
      flushA.complete();
      await pumpEventQueue(times: 40);

      // Legitimate later flush persists B.
      await Future<void>.delayed(tickWait);
      await pumpEventQueue(times: 40);

      final rows = await persistedRows();
      final counts = countsByIndex(rows);
      final session = container.read(rideStateMachineProvider);

      debugPrint('==== RID-3A DURING-FLUSH ARRIVAL (after fix) ====');
      debugPrint('durable rows         : ${rows.length}');
      debugPrint('distinct identities  : ${counts.length}');
      debugPrint('duplicated           : ${duplicatedIn(counts)}');
      debugPrint('missing              : ${missingOf(10, counts)}');
      debugPrint('order                : ${rows.map((r) => r.speed).toList()}');

      expect(rows, hasLength(10));
      expect(
        duplicatedIn(counts),
        isEmpty,
        reason: 'A and B each exactly once',
      );
      expect(missingOf(10, counts), isEmpty, reason: 'nothing missing');
      expect(rows.map((r) => r.speed).toList(), [
        for (var i = 0; i < 10; i++) 1000.0 + i,
      ], reason: 'A persisted before B');
      expect(rows.map((r) => r.rideId).toSet(), {rideId});
      expect(session.totalSamplesPersisted, 10);
    });

    test('a failed flush restores its snapshot AHEAD of newer samples and a '
        'later flush persists A B C D E F exactly once each', () async {
      final container = buildContainer();
      final rideId = await startRide(container);

      // A B C = indices 0, 1, 2.
      emitSamples(0, 3);
      await pumpEventQueue(times: 30);

      db.gateBatch = true;
      db.batchFailuresRemaining = 1;
      await Future<void>.delayed(tickWait);
      expect(db.batchInvocations, 1);
      final failingFlush = db.batchGates.single;

      // D E F = indices 3, 4, 5 arrive while the doomed persistence is parked.
      emitSamples(3, 3);
      await pumpEventQueue(times: 30);
      expect(container.read(rideStateMachineProvider).totalSamplesReceived, 6);

      db.gateBatch = false;
      failingFlush.complete(); // db.batch now throws for this snapshot
      await pumpEventQueue(times: 40);

      var session = container.read(rideStateMachineProvider);
      expect(
        session.totalSamplesPersisted,
        0,
        reason: 'a failed persistence must not increment the success counter',
      );
      expect(await persistedRows(), isEmpty, reason: 'nothing durable yet');
      expect(
        session.state,
        RideState.riding,
        reason: 'telemetry recording continues so a later flush can recover',
      );

      // Existing periodic timer provides the legitimate retry. No automatic
      // retry worker is introduced.
      await Future<void>.delayed(tickWait);
      await pumpEventQueue(times: 40);

      final rows = await persistedRows();
      final counts = countsByIndex(rows);
      session = container.read(rideStateMachineProvider);

      debugPrint('==== RID-3A FAILURE + ORDERED RESTORATION (after fix) ====');
      debugPrint('batch invocations    : ${db.batchInvocations}');
      debugPrint('durable rows         : ${rows.length}');
      debugPrint('distinct identities  : ${counts.length}');
      debugPrint('duplicated           : ${duplicatedIn(counts)}');
      debugPrint('missing              : ${missingOf(6, counts)}');
      debugPrint('order                : ${rows.map((r) => r.speed).toList()}');
      debugPrint(
        'received / persisted : '
        '${session.totalSamplesReceived} / ${session.totalSamplesPersisted}',
      );

      expect(rows, hasLength(6));
      expect(
        duplicatedIn(counts),
        isEmpty,
        reason: 'retry causes no duplicates',
      );
      expect(missingOf(6, counts), isEmpty, reason: 'no rows missing');
      expect(
        rows.map((r) => r.speed).toList(),
        [1000.0, 1001.0, 1002.0, 1003.0, 1004.0, 1005.0],
        reason:
            'failed snapshot A B C restored AHEAD of newer D E F, both '
            'internal orders preserved',
      );
      expect(rows.map((r) => r.rideId).toSet(), {rideId});
      expect(
        session.totalSamplesPersisted,
        6,
        reason: 'counter reflects only the successfully persisted snapshot',
      );
    });

    test('failed flush restoration above 1000 capacity bounds queue to 1000 '
        'by dropping oldest samples', () async {
      final container = buildContainer();
      final notifier = container.read(rideStateMachineProvider.notifier);
      final rideId = await startRide(container);

      // 1. Emit 600 samples (indices 0..599).
      emitSamples(0, 600);
      await pumpEventQueue(times: 30);
      expect(
        container.read(rideStateMachineProvider).totalSamplesReceived,
        600,
      );

      // 2. Start flush of those 600 samples, but gate db.batch.
      db.gateBatch = true;
      db.batchFailuresRemaining = 1;
      final finishAttempt = notifier.finishRide();
      await pumpEventQueue(times: 10);
      expect(db.batchInvocations, 1);
      final failingFlush = db.batchGates.single;

      // 3. While write is pending, emit 500 new samples (indices 600..1099).
      // Total received = 1100.
      emitSamples(600, 500);
      await pumpEventQueue(times: 30);
      expect(
        container.read(rideStateMachineProvider).totalSamplesReceived,
        1100,
      );

      // 4. Fail the write. Snapshot of 600 is restored to front of the 500 new samples.
      // Combined = 1100 > 1000.
      db.gateBatch = false;
      failingFlush.complete();
      await expectLater(finishAttempt, throwsStateError);
      await pumpEventQueue(times: 30);

      // Assert ride remains open and no final summary produced after failed Stop
      final intermediateSession = container.read(rideStateMachineProvider);
      expect(
        intermediateSession.state,
        RideState.riding,
        reason: 'ride remains riding on stop failure',
      );
      expect(
        intermediateSession.healthStatus,
        RecordingHealthStatus.recordingActive,
        reason: 'recordingActive restored on stop failure',
      );
      final intermediateRides = await db.select(db.rides).get();
      expect(
        intermediateRides.singleWhere((r) => r.id == rideId).endTime,
        isNull,
        reason: 'no endTime written to DB on stop failure',
      );
      expect(
        logger.stopSessionInvocations,
        0,
        reason: 'no logger stopSession called on stop failure',
      );
      expect(
        await db.select(db.fieldTestSummaries).get(),
        isEmpty,
        reason: 'no final summary produced on stop failure',
      );
      expect(
        notifier.telemetrySamplesLost,
        100,
        reason: '100 samples lost on restoration excess',
      );
      expect(logger.telemetrySamplesLost, 100);

      // 5. Emit 1 more sample (index 1100).
      emitSamples(1100, 1);
      await pumpEventQueue(times: 30);
      expect(
        container.read(rideStateMachineProvider).totalSamplesReceived,
        1101,
      );
      expect(
        notifier.telemetrySamplesLost,
        101,
        reason: '101 samples lost after 1 drop on arrival',
      );
      expect(logger.telemetrySamplesLost, 101);

      // 6. Now do a successful finish flush to persist whatever remains in queue.
      await notifier.finishRide();
      await pumpEventQueue(times: 30);

      final rows = await persistedRows();
      final counts = countsByIndex(rows);
      final session = container.read(rideStateMachineProvider);

      debugPrint('==== RID-3A OVERFLOW REGRESSION ====');
      debugPrint('durable rows         : ${rows.length}');
      debugPrint('distinct identities  : ${counts.length}');
      debugPrint('overflow events      : ${logger.overflowEventCount}');
      debugPrint(
        'first row speed      : ${rows.isNotEmpty ? rows.first.speed : "empty"}',
      );
      debugPrint(
        'last row speed       : ${rows.isNotEmpty ? rows.last.speed : "empty"}',
      );

      // Must be capped at 1000.
      expect(rows, hasLength(1000));
      // First 101 oldest samples (indices 0..100) dropped (100 on restoration + 1 on sample 1100).
      // Verify complete retained sample identity sequence in exact order.
      expect(
        rows.map((r) => r.speed).toList(),
        [for (var i = 101; i <= 1100; i++) 1000.0 + i],
        reason: 'complete sequence of 1000 retained samples in exact order',
      );
      expect(duplicatedIn(counts), isEmpty);
      expect(logger.overflowEventCount, 101);
      expect(session.totalSamplesPersisted, 1000);
      expect(rows.map((r) => r.rideId).toSet(), {rideId});

      final summaries = await db.select(db.fieldTestSummaries).get();
      expect(summaries, hasLength(1));
      final summary = summaries.single;
      expect(
        summary.databaseSamplesLost,
        101,
        reason: 'databaseSamplesLost reports exact dropped telemetry',
      );
      expect(summary.databaseSamplesWritten, 1000);
      expect(notifier.telemetrySamplesLost, 101);
      expect(logger.telemetrySamplesLost, 101);
    });

    test('repeated flush failures keep queue bounded to exactly 1000 capacity', () async {
      final container = buildContainer();
      final notifier = container.read(rideStateMachineProvider.notifier);
      final rideId = await startRide(container);

      // Cycle 1: Emit 800 samples (0..799).
      emitSamples(0, 800);
      await pumpEventQueue(times: 30);

      db.gateBatch = true;
      db.batchFailuresRemaining = 1;
      final finish1 = notifier.finishRide();
      await pumpEventQueue(times: 10);
      expect(db.batchInvocations, 1);
      final flush1 = db.batchGates.single;

      // 300 arrive during await (800..1099). Total received: 1100.
      emitSamples(800, 300);
      await pumpEventQueue(times: 30);

      db.gateBatch = false;
      flush1.complete();
      await expectLater(finish1, throwsStateError);
      await pumpEventQueue(times: 30);

      // Assert ride remains open and no final summary produced after fail 1
      final sessionAfterFail1 = container.read(rideStateMachineProvider);
      expect(sessionAfterFail1.state, RideState.riding);
      expect(
        sessionAfterFail1.healthStatus,
        RecordingHealthStatus.recordingActive,
      );
      final ridesAfterFail1 = await db.select(db.rides).get();
      expect(
        ridesAfterFail1.singleWhere((r) => r.id == rideId).endTime,
        isNull,
      );
      expect(logger.stopSessionInvocations, 0);
      expect(await db.select(db.fieldTestSummaries).get(), isEmpty);

      // First failure: 800 + 300 = 1100 -> 100 oldest dropped (0..99). Retained: 100..1099.
      expect(logger.overflowEventCount, 100);
      expect(notifier.telemetrySamplesLost, 100);
      expect(logger.telemetrySamplesLost, 100);

      // Cycle 2: Flush 2 starts, owns all 1000 currently in queue.
      db.batchGates.clear();
      db.gateBatch = true;
      db.batchFailuresRemaining = 1;
      final finish2 = notifier.finishRide();
      await pumpEventQueue(times: 10);
      expect(db.batchInvocations, 2);
      final flush2 = db.batchGates.single;

      // 200 arrive during await (1100..1299). Total received: 1300.
      emitSamples(1100, 200);
      await pumpEventQueue(times: 30);

      db.gateBatch = false;
      flush2.complete();
      await expectLater(finish2, throwsStateError);
      await pumpEventQueue(times: 30);

      // Assert ride remains open and no final summary produced after fail 2
      final sessionAfterFail2 = container.read(rideStateMachineProvider);
      expect(sessionAfterFail2.state, RideState.riding);
      expect(
        sessionAfterFail2.healthStatus,
        RecordingHealthStatus.recordingActive,
      );
      final ridesAfterFail2 = await db.select(db.rides).get();
      expect(
        ridesAfterFail2.singleWhere((r) => r.id == rideId).endTime,
        isNull,
      );
      expect(logger.stopSessionInvocations, 0);
      expect(await db.select(db.fieldTestSummaries).get(), isEmpty);

      // Second failure: 1000 + 200 = 1200 -> 200 oldest dropped (100..299). Retained: 300..1299.
      expect(logger.overflowEventCount, 300);
      expect(notifier.telemetrySamplesLost, 300);
      expect(logger.telemetrySamplesLost, 300);

      // Subsequent 50 arrivals (1300..1349).
      emitSamples(1300, 50);
      await pumpEventQueue(times: 30);
      // Each drops 1 oldest (300..349). Retained: 350..1349 (1000 samples).
      expect(logger.overflowEventCount, 350);
      expect(notifier.telemetrySamplesLost, 350);
      expect(logger.telemetrySamplesLost, 350);

      // Successful finalization persists retained 1000.
      await notifier.finishRide();
      await pumpEventQueue(times: 30);

      final rows = await persistedRows();
      final counts = countsByIndex(rows);
      final session = container.read(rideStateMachineProvider);

      debugPrint('==== RID-3A REPEATED FAILURES ====');
      debugPrint('durable rows         : ${rows.length}');
      debugPrint('distinct identities  : ${counts.length}');
      debugPrint('overflow events      : ${logger.overflowEventCount}');
      debugPrint(
        'first row speed      : ${rows.isNotEmpty ? rows.first.speed : "empty"}',
      );
      debugPrint(
        'last row speed       : ${rows.isNotEmpty ? rows.last.speed : "empty"}',
      );

      expect(rows, hasLength(1000));
      // Verify complete retained sample identity sequence in exact order.
      expect(
        rows.map((r) => r.speed).toList(),
        [for (var i = 350; i <= 1349; i++) 1000.0 + i],
        reason:
            'complete sequence of 1000 retained samples in exact order after repeated drops',
      );
      expect(duplicatedIn(counts), isEmpty);
      expect(logger.overflowEventCount, 350);
      expect(session.totalSamplesPersisted, 1000);
      expect(rows.map((r) => r.rideId).toSet(), {rideId});

      final summaries = await db.select(db.fieldTestSummaries).get();
      expect(summaries, hasLength(1));
      final summary = summaries.single;
      expect(
        summary.databaseSamplesLost,
        350,
        reason: 'databaseSamplesLost reflects 350 total dropped telemetry',
      );
      expect(summary.databaseSamplesWritten, 1000);
      expect(notifier.telemetrySamplesLost, 350);
      expect(logger.telemetrySamplesLost, 350);
    });

    test(
      'clean ride with zero telemetry loss produces databaseSamplesLost == 0 and new ride resets counter',
      () async {
        final container = buildContainer();
        final notifier = container.read(rideStateMachineProvider.notifier);
        final logger = container.read(fieldTestLoggerProvider);
        final rideId = await startRide(container);

        // Emit 50 samples (< 1000 buffer capacity)
        emitSamples(0, 50);
        await pumpEventQueue(times: 30);

        expect(notifier.telemetrySamplesLost, 0);
        expect(logger.telemetrySamplesLost, 0);

        // Finish cleanly
        await notifier.finishRide();
        await pumpEventQueue(times: 30);

        final summaries = await db.select(db.fieldTestSummaries).get();
        expect(summaries, hasLength(1));
        expect(summaries.single.databaseSamplesLost, 0);
        expect(summaries.single.databaseSamplesWritten, 50);

        // Start new ride: counters must reset to 0
        final rideId2 = await startRide(container);
        expect(rideId2, isNot(rideId));
        expect(notifier.telemetrySamplesLost, 0);
        expect(logger.telemetrySamplesLost, 0);
      },
    );
  });
}