import 'dart:async'; import 'package:wakelock_plus/wakelock_plus.dart'; import 'package:flutter_riverpod/flutter_riverpod.dart'; import 'package:flutter/foundation.dart' show debugPrint; import '../../database/database.dart' hide TelemetrySample; import 'package:drift/drift.dart' as drift; import 'package:geolocator/geolocator.dart'; import '../../models/telemetry_sample.dart'; import '../../models/field_test_event.dart'; import '../../models/recording_health_status.dart'; import '../../services/telemetry_service.dart'; import '../../services/audio_alert_service.dart'; import '../records/records_service.dart'; import '../garage/garage_providers.dart'; import '../../services/field_test/field_test_logger.dart'; enum RideState { idle, ready, riding, paused, finished } class RideSession { final int? dbId; final RideState state; final RecordingHealthStatus healthStatus; /// RID-3C-1: the ONE authoritative in-memory recording-health evidence model. /// /// Ephemeral runtime state only - never persisted. `RideSession` has no JSON or /// Drift serialization, so exposing the contract here needs no schema change, /// migration, persistence change or cloud schema change, and creates no second /// health provider, StateNotifier, singleton or duplicate model. final RecordingHealthContract health; final double tripDistanceMiles; final double topSpeedMph; final double maxLeanLeft; final double maxLeanRight; final double maxAccelG; final double maxBrakingG; final DateTime? startTime; // Phase 4 Stats final int movingTimeSeconds; final int stoppedTimeSeconds; final double elevationGain; final double elevationLoss; final double averageSpeedMph; final int totalSamplesReceived; final int totalSamplesPersisted; RideSession({ this.dbId, this.state = RideState.idle, this.healthStatus = RecordingHealthStatus.recordingArmed, this.health = const RecordingHealthContract(), this.tripDistanceMiles = 0.0, this.topSpeedMph = 0.0, this.maxLeanLeft = 0.0, this.maxLeanRight = 0.0, this.maxAccelG = 0.0, this.maxBrakingG = 0.0, this.startTime, this.movingTimeSeconds = 0, this.stoppedTimeSeconds = 0, this.elevationGain = 0.0, this.elevationLoss = 0.0, this.averageSpeedMph = 0.0, this.totalSamplesReceived = 0, this.totalSamplesPersisted = 0, }); RideSession copyWith({ int? dbId, RideState? state, RecordingHealthStatus? healthStatus, RecordingHealthContract? health, double? tripDistanceMiles, double? topSpeedMph, double? maxLeanLeft, double? maxLeanRight, double? maxAccelG, double? maxBrakingG, DateTime? startTime, int? movingTimeSeconds, int? stoppedTimeSeconds, double? elevationGain, double? elevationLoss, double? averageSpeedMph, int? totalSamplesReceived, int? totalSamplesPersisted, }) { return RideSession( dbId: dbId ?? this.dbId, state: state ?? this.state, healthStatus: healthStatus ?? this.healthStatus, health: health ?? this.health, tripDistanceMiles: tripDistanceMiles ?? this.tripDistanceMiles, topSpeedMph: topSpeedMph ?? this.topSpeedMph, maxLeanLeft: maxLeanLeft ?? this.maxLeanLeft, maxLeanRight: maxLeanRight ?? this.maxLeanRight, maxAccelG: maxAccelG ?? this.maxAccelG, maxBrakingG: maxBrakingG ?? this.maxBrakingG, startTime: startTime ?? this.startTime, movingTimeSeconds: movingTimeSeconds ?? this.movingTimeSeconds, stoppedTimeSeconds: stoppedTimeSeconds ?? this.stoppedTimeSeconds, elevationGain: elevationGain ?? this.elevationGain, elevationLoss: elevationLoss ?? this.elevationLoss, averageSpeedMph: averageSpeedMph ?? this.averageSpeedMph, totalSamplesReceived: totalSamplesReceived ?? this.totalSamplesReceived, totalSamplesPersisted: totalSamplesPersisted ?? this.totalSamplesPersisted, ); } } class RideStateMachine extends StateNotifier { final AppDatabase db; final TelemetryService telemetryService; final RecordsService recordsService; final AudioAlertService audioAlertService; final Ref ref; StreamSubscription? _telemetrySub; TelemetrySample? _lastSample; double _accumulatedDistance = 0; // meters internally double _movingTimeAccumulator = 0.0; // seconds internally double _stoppedTimeAccumulator = 0.0; // seconds internally // Batching final List _telemetryBatch = []; Timer? _batchWriteTimer; Timer? _watchdogTimer; DateTime? _lastTelemetryReceivedTime; // Event Debouncing (Store last trigger time per event type) final Map _lastEventTimes = {}; static const Duration _eventCooldown = Duration(seconds: 5); // RID-1: re-entrancy guard for ride start. Raised synchronously before the // first await inside startRide() so concurrent callers cannot each create // their own ride row and cloudRideId. Its lifetime is owned exclusively by // the in-flight start operation (try/finally); it is intentionally NOT // cleared anywhere else, because clearing it while a start is still awaiting // would re-open the duplicate-start race. bool _isStarting = false; // RID-2: re-entrancy guard for ride finish. finishRide() previously left // state.state == riding across every await (only healthStatus changed), so // two rapid Stop taps both entered finalization and duplicated db.endRide, // personal-record evaluation, logger.stopSession, the field-test summary row // and the "Session ended" announcement. Raised synchronously before the // first await; lifetime is owned exclusively by the in-flight finish // operation (try/finally) and it is intentionally NOT cleared anywhere else. bool _isFinishing = false; // RID-3A: re-entrancy guard for telemetry batch persistence. Raised // synchronously before the first await inside _flushBatch(); its lifetime is // owned solely by the in-flight persistence operation (try/finally) and it is // intentionally NOT cleared anywhere else, because clearing it while a // persistence await is outstanding would re-open the shared-buffer race. bool _isFlushingTelemetryBatch = false; // RID-3B-1: coordination primitive exposing the in-flight telemetry // persistence operation, so an explicit Stop can await the current owner // instead of observing a no-op flush. This is NOT a second buffer and NOT a // serialization framework: it publishes exactly one Future for exactly one // owned snapshot and is cleared when that snapshot's attempt settles. Future? _activeTelemetryFlush; // RID-3B-1: late private telemetry-acceptance cutoff (invariant I3). Owned // solely by the in-flight finish operation and reopened in its finally path, // so a failed-but-retryable ride keeps recording and the next ride is // unaffected. Strictly private: no public RideState or RideSession change. bool _telemetryAcceptanceClosed = false; // RID-3B-1: bound on the convergent final drain. Telemetry arrives at 20 Hz // (one sample per 50 ms) while each pass awaits a single db.batch round-trip, // so the samples that can accumulate during a pass are normally 0 or 1 and // convergence is reached in 1-2 passes. Four passes tolerate a persistence // latency of roughly 150 ms per pass while still bounding finalization to at // most four batch writes. The bound never fabricates durability: if telemetry // is still outstanding when it is exhausted, finalization FAILS and remains // retryable rather than transitioning to finished. static const int _maxFinalDrainPasses = 4; // Explicit per-ride counter of telemetry samples discarded due to queue capacity bounds. int _telemetrySamplesLost = 0; int get telemetrySamplesLost => _telemetrySamplesLost; RideStateMachine( this.db, this.telemetryService, this.recordsService, this.audioAlertService, this.ref, ) : super(RideSession()) { _telemetrySub = telemetryService.telemetryStream.listen(_onTelemetryUpdate); // Write to DB every 3 seconds for tight crash-durability _batchWriteTimer = Timer.periodic( const Duration(seconds: 3), (_) => _runScheduledFlush(), ); // Watchdog timer (every 2 seconds) checks if telemetry pipeline is alive _watchdogTimer = Timer.periodic( const Duration(seconds: 2), (_) => _checkWatchdog(), ); // Recover any unfinished rides from previous sessions _recoverInterruptedRides(); } // --------------------------------------------------------------------------- // RID-3C-1: truthful, evidence-based recording health. // // Evidence markers are MONOTONIC values taken from TelemetryService's own // monotonic domain (TelemetrySample.monotonicTimestampMs, sourced from a // Stopwatch), so a user or system wall-clock jump can never make stale evidence // look fresh. Wall-clock DateTimes are retained only as contract metadata. No // new timing subsystem is introduced. // // Presence - never magnitude - proves a sensor sample arrived, so a genuine // zero-valued lean, pitch or g-force still counts as real evidence. // --------------------------------------------------------------------------- int? _latestObservedMonotonicMs; int? _lastGpsEvidenceMonotonicMs; DateTime? _lastSeenGpsTimestamp; int? _lastImuEvidenceMonotonicMs; int? _lastDbCommitMonotonicMs; bool _dbPersistenceFailed = false; /// Telemetry evidence time, advanced ONLY for samples accepted for the active /// ride. Deliberately separate from `_lastTelemetryReceivedTime`, which the /// existing watchdog updates before RID-3B-1's acceptance cutoff and which must /// not be moved: a post-cutoff sample may not refresh health evidence. DateTime? _lastAcceptedTelemetryTime; void _resetHealthEvidence() { _latestObservedMonotonicMs = null; _lastGpsEvidenceMonotonicMs = null; _lastSeenGpsTimestamp = null; _lastImuEvidenceMonotonicMs = null; _lastDbCommitMonotonicMs = null; _dbPersistenceFailed = false; _lastAcceptedTelemetryTime = null; } /// Monotonic age in ms of a marker, or null when it cannot be established. /// Null ALWAYS means "not fresh" - this model fails closed. int? _monotonicAgeMs(int? markerMonotonicMs) { final reference = _latestObservedMonotonicMs; if (reference == null || markerMonotonicMs == null) return null; final age = reference - markerMonotonicMs; // A negative age is not physically meaningful here; fail closed rather than // reporting it as fresh. return age < 0 ? null : age; } bool _isFresh(int? ageMs, int boundMs) => ageMs != null && ageMs <= boundMs; /// Records what an ACCEPTED telemetry sample actually proves. Called only after /// RID-3B-1's acceptance cutoff gate, so post-cutoff samples are excluded. void _recordTelemetryEvidence(TelemetrySample sample) { final monotonicMs = sample.monotonicTimestampMs; if (monotonicMs == null) return; final latest = _latestObservedMonotonicMs; if (latest == null || monotonicMs > latest) { _latestObservedMonotonicMs = monotonicMs; } // Real GPS fix evidence: genuinely newer fix timestamp is present. if (sample.gpsTimestamp != null) { final gpsTime = sample.gpsTimestamp!; if (_lastSeenGpsTimestamp == null || gpsTime.isAfter(_lastSeenGpsTimestamp!)) { _lastSeenGpsTimestamp = gpsTime; _lastGpsEvidenceMonotonicMs = monotonicMs; } } // Real IMU sample evidence: the IMU timestamp is present. if (sample.imuTimestamp != null) _lastImuEvidenceMonotonicMs = monotonicMs; _lastAcceptedTelemetryTime = DateTime.now(); } /// Recomputes the ONE authoritative RecordingHealthContract from real evidence /// and maps it to RecordingHealthStatus. Invoked only from the canonical /// telemetry-acceptance and watchdog paths - never from a bare timer tick, a UI /// frame, or a service-presence assumption. /// /// RID-3B-1 priority is preserved: `resolveStatus` treats recordingFinalizing, /// recordingComplete and recordingFailed as authoritative lifecycle overrides, /// so ordinary health evaluation can never return a finishing or finished ride /// to recordingActive. void _evaluateRecordingHealth() { final now = DateTime.now(); const sensorBound = RecordingHealthContract.evidenceStaleAfterMs; // A. Active, valid ride/session identity. final hasValidSession = state.dbId != null && state.state == RideState.riding; // C. Telemetry evidence, using the SAME 4 s bound as the existing canonical // watchdog so contract and status agree by construction. This wall-clock term // is also what prevents monotonic freeze-blindness: when samples stop, the // monotonic reference stops advancing, but this term still goes stale. final accepted = _lastAcceptedTelemetryTime; final telemetryAgeMs = accepted == null ? null : now.difference(accepted).inMilliseconds; // Fail closed on a backwards clock jump: negative age is not freshness. final telemetryActive = telemetryAgeMs != null && telemetryAgeMs >= 0 && telemetryAgeMs <= sensorBound; // B. GPS and D. IMU evidence: monotonic, presence-proven. final gpsOperational = _isFresh( _monotonicAgeMs(_lastGpsEvidenceMonotonicMs), sensorBound, ); final sensorActive = _isFresh( _monotonicAgeMs(_lastImuEvidenceMonotonicMs), sensorBound, ); // E. Database evidence: advances ONLY after a real successful telemetry // commit (the RID-3B-2 commit point). A scheduled flush is not evidence, and // a failed pre-commit batch does not advance it. final dbOperational = !_dbPersistenceFailed && _isFresh( _monotonicAgeMs(_lastDbCommitMonotonicMs), RecordingHealthContract.databaseWriteStaleAfterMs, ); final previous = state.health; final health = previous.copyWith( isForegroundServiceAlive: gpsOperational, isTelemetryPipelineActive: telemetryActive, isAtLeastOneSensorActive: sensorActive, isDatabasePersistenceOperational: dbOperational, isValidSessionId: hasValidSession, lastGPSFixTime: previous.lastGPSFixTime, lastTelemetryTime: accepted ?? previous.lastTelemetryTime, lastIMUSampleTime: previous.lastIMUSampleTime, lastDatabaseWriteTime: previous.lastDatabaseWriteTime, ); state = state.copyWith( health: health, healthStatus: health.resolveStatus(lifecycle: state.healthStatus), ); } void _checkWatchdog() { if (state.state != RideState.riding) return; final now = DateTime.now(); // Existing canonical stall detection: the 4-second bound and the telemetryGap // diagnostic are unchanged. RID-3C-1 adds NO second watchdog; it makes this // one the source of heartbeat and consecutive-miss evidence for the contract. final elapsedSec = _lastTelemetryReceivedTime == null ? null : now.difference(_lastTelemetryReceivedTime!).inSeconds; final stalled = elapsedSec != null && elapsedSec > 4; final previousMisses = state.health.consecutiveWatchdogMisses; final misses = stalled ? previousMisses + 1 : 0; if (stalled && previousMisses == 0) { // First detection of this stall episode: logged once, as before. try { ref .read(fieldTestLoggerProvider) .log( eventType: FieldTestEventType.telemetryGap, severity: EventSeverity.warning, message: 'Watchdog: Telemetry pipeline stalled ($elapsedSec seconds without samples)', ); } catch (_) {} } state = state.copyWith( health: state.health.copyWith( lastHeartbeatTime: now, consecutiveWatchdogMisses: misses, ), ); // The watchdog evaluates the contract instead of maintaining a contradictory // parallel truth, so RecordingHealthStatus and RecordingHealthContract agree. _evaluateRecordingHealth(); } Future _recoverInterruptedRides() async { try { final unclosedRides = await (db.select( db.rides, )..where((r) => r.endTime.isNull())).get(); for (final r in unclosedRides) { // Find last recorded telemetry sample for this ride to determine actual end time final lastSample = await (db.select(db.telemetrySamples) ..where((t) => t.rideId.equals(r.id)) ..orderBy([ (t) => drift.OrderingTerm( expression: t.timestamp, mode: drift.OrderingMode.desc, ), ]) ..limit(1)) .getSingleOrNull(); final calculatedEndTime = lastSample?.timestamp ?? r.startTime.add(Duration(seconds: r.movingTime + r.stoppedTime)); await (db.update(db.rides)..where((t) => t.id.equals(r.id))).write( RidesCompanion(endTime: drift.Value(calculatedEndTime)), ); try { final logger = ref.read(fieldTestLoggerProvider); logger.log( eventType: FieldTestEventType.databaseRecovery, severity: EventSeverity.critical, message: 'Interrupted ride #${r.id} recovered after unexpected exit', rideId: r.id, ); await logger.generateAndSaveSummary(r.id, wasRecovered: true); } catch (_) {} } } catch (_) {} } void setReady() { state = state.copyWith( state: RideState.ready, healthStatus: RecordingHealthStatus.recordingArmed, ); } /// RID-1: idempotent ride start. /// /// The guard is raised synchronously, before the first `await`, so that /// concurrent callers — the 20 Hz motion auto-start in [_onTelemetryUpdate] /// and rapid taps on the dashboard Start buttons — collapse into a single /// start operation instead of each minting its own ride row, `cloudRideId` /// and orphaned open ride. /// /// The guard is released in `finally` so a genuinely failed start never /// leaves the machine permanently unable to record, and a later legitimate /// retry still succeeds. Frozen semantics (auto-start threshold, PR gating, /// kickstand isolation, recovery, `cloudRideId` generation) are untouched. Future startRide() async { if (_isStarting) return; if (state.state == RideState.riding) return; _isStarting = true; try { await _performStartRide(); } finally { _isStarting = false; } } /// Performs the actual ride start. Guarded exclusively by [startRide] and /// must never be invoked directly. Future _performStartRide() async { try { ref .read(fieldTestLoggerProvider) .log( eventType: FieldTestEventType.rideStartRequested, severity: EventSeverity.info, message: 'Ride start requested by user or motion trigger', ); } catch (_) {} final defaultBike = ref.read(defaultMotorcycleProvider); final startTime = DateTime.now(); final id = await db.startRide( startTime: startTime, motorcycleId: defaultBike?.id, motorcycleYear: defaultBike?.year, motorcycleMake: defaultBike?.makeName, motorcycleModel: defaultBike?.modelName, motorcycleVariant: defaultBike?.variantName, ); // RID-3C-1 (warm-up): a freshly created ride has NO evidence yet, so it must // not claim recordingActive. Ride creation is never blocked waiting for GPS, // IMU or the 3-second batch; health warms up automatically as real evidence // arrives and _evaluateRecordingHealth promotes the status. _resetHealthEvidence(); _telemetrySamplesLost = 0; state = state.copyWith( dbId: id, state: RideState.riding, healthStatus: RecordingHealthStatus.recordingArmed, health: const RecordingHealthContract(), startTime: startTime, tripDistanceMiles: 0, topSpeedMph: 0, maxLeanLeft: 0, maxLeanRight: 0, maxAccelG: 0, maxBrakingG: 0, averageSpeedMph: 0, movingTimeSeconds: 0, stoppedTimeSeconds: 0, elevationGain: 0, elevationLoss: 0, totalSamplesReceived: 0, totalSamplesPersisted: 0, ); _lastSample = null; _accumulatedDistance = 0; _movingTimeAccumulator = 0.0; _stoppedTimeAccumulator = 0.0; _lastTelemetryReceivedTime = DateTime.now(); // RID-3C-1: publish truthful warm-up health immediately, so the contract // reflects the real session identity even before the first sample arrives. // All evidence is still absent, so this resolves to recordingArmed and can // never produce a false recordingActive. _evaluateRecordingHealth(); try { WakelockPlus.enable().catchError((_) {}); } catch (_) {} audioAlertService.speak('Session started'); // Auto-start Field Test Session with full forensic logging try { final logger = ref.read(fieldTestLoggerProvider); logger.startSession(id); logger.log( eventType: FieldTestEventType.foregroundServiceStarted, severity: EventSeverity.info, message: 'Foreground service confirmed active for ride #$id', rideId: id, ); logger.log( eventType: FieldTestEventType.wakelockAcquired, severity: EventSeverity.info, message: 'Screen wakelock acquired for ride #$id', rideId: id, ); logger.log( eventType: FieldTestEventType.telemetryPipelineStarted, severity: EventSeverity.info, message: '20Hz telemetry aggregation pipeline started for ride #$id', rideId: id, ); } catch (_) {} } void pauseRide() { state = state.copyWith(state: RideState.paused); try { WakelockPlus.disable().catchError((_) {}); } catch (_) {} try { ref .read(fieldTestLoggerProvider) .log( eventType: FieldTestEventType.appPaused, severity: EventSeverity.info, message: 'Ride session paused', ); } catch (_) {} } void resumeRide() { // RID-3C-1: resuming does not by itself prove recording health. Evidence ages // out monotonically across the pause, so the next accepted sample // re-establishes truthfully whether the ride is active or degraded. state = state.copyWith( state: RideState.riding, healthStatus: RecordingHealthStatus.recordingArmed, ); _lastSample = null; // reset to avoid jumping _lastTelemetryReceivedTime = DateTime.now(); // RID-3C-1: re-publish truthful health for the resumed session. _evaluateRecordingHealth(); try { WakelockPlus.enable().catchError((_) {}); } catch (_) {} try { ref .read(fieldTestLoggerProvider) .log( eventType: FieldTestEventType.appResumed, severity: EventSeverity.info, message: 'Ride session resumed', ); } catch (_) {} } /// RID-2: idempotent ride finish. /// /// The guard is raised synchronously, before the first `await`, so concurrent /// Stop taps collapse into a single finalization instead of each persisting /// their own end-of-ride write and forensic summary. A finish after a /// successful completion is a no-op. /// /// The guard is released in `finally`, so a persistence failure never leaves /// the machine permanently unable to finalize and a later explicit retry /// still succeeds. Failure semantics are deliberately unchanged: a failed /// `db.endRide` still propagates to callers that await this method, still /// leaves the session in its existing recoverable riding/finalizing state, /// and still does not fabricate an `endTime`, emit a finished transition, or /// enqueue an unpersisted ride for cloud sync. Future finishRide() async { if (_isFinishing) return; if (state.state == RideState.finished) return; _isFinishing = true; try { await _performFinishRide(); } finally { _isFinishing = false; // RID-3B-1: the telemetry-acceptance cutoff is owned solely by this // in-flight finish operation, so it is reopened here. A failed but // retryable ride therefore keeps recording, and the next ride is // unaffected. Deliberately NOT cleared from reset(): source analysis shows // reset() has no callers, and clearing finish-owned state elsewhere would // weaken the barrier. _telemetryAcceptanceClosed = false; } } /// Performs the actual ride finalization. Guarded exclusively by /// [finishRide] and must never be invoked directly. Future _performFinishRide() async { if (state.dbId != null) { state = state.copyWith( healthStatus: RecordingHealthStatus.recordingFinalizing, ); try { ref .read(fieldTestLoggerProvider) .log( eventType: FieldTestEventType.rideFinishRequested, severity: EventSeverity.info, message: 'Ride finish requested for ride #${state.dbId}', rideId: state.dbId, ); } catch (_) {} // RID-3B-1: the explicit-Stop durability barrier replaces the single // tolerant flush. It awaits any in-flight periodic flush owner, drains to // convergence, closes telemetry acceptance, and performs a final strict // drain - throwing if durability cannot be established (invariant I2). await _establishTelemetryDurabilityForFinish(); await db.endRide( state.dbId!, RidesCompanion( distanceMiles: drift.Value(state.tripDistanceMiles), topSpeedMph: drift.Value(state.topSpeedMph), maxLeanLeft: drift.Value(state.maxLeanLeft), maxLeanRight: drift.Value(state.maxLeanRight), maxAccelG: drift.Value(state.maxAccelG), maxBrakingG: drift.Value(state.maxBrakingG), movingTime: drift.Value(state.movingTimeSeconds), stoppedTime: drift.Value(state.stoppedTimeSeconds), averageSpeed: drift.Value(state.averageSpeedMph), elevationGain: drift.Value(state.elevationGain), elevationLoss: drift.Value(state.elevationLoss), ), ); try { ref .read(fieldTestLoggerProvider) .log( eventType: FieldTestEventType.rideFinalized, severity: EventSeverity.info, message: 'Ride record #${state.dbId} finalized in database', rideId: state.dbId, ); } catch (_) {} // Check for personal records final runStats = PersonalRecords( topSpeedMph: state.topSpeedMph, maxLeanLeft: state.maxLeanLeft, maxLeanRight: state.maxLeanRight, maxAccelG: state.maxAccelG, maxBrakingG: state.maxBrakingG, longestRideMiles: state.tripDistanceMiles, greatestElevationGain: state.elevationGain, ); await recordsService.checkAndSaveRecords(runStats, (recordName, value) { audioAlertService.speak('New record: $recordName'); }); } // Auto-stop Field Test Session if active try { final logger = ref.read(fieldTestLoggerProvider); if (logger.isSessionActive) { logger.telemetrySamplesLost = _telemetrySamplesLost; await logger.stopSession(); logger.log( eventType: FieldTestEventType.rideSummaryCreated, severity: EventSeverity.info, message: 'Field test summary generated and saved for ride #${state.dbId}', rideId: state.dbId, ); } } catch (_) {} try { WakelockPlus.disable().catchError((_) {}); } catch (_) {} audioAlertService.speak('Session ended'); // RID-3B-1: the sync-visible finished edge is the LAST side effect of // finalization. It is emitted only after telemetry durability, endTime, // personal records and the forensic summary are complete, so SyncObserver // can keep observing only this edge (invariant I1). state = state.copyWith( state: RideState.finished, healthStatus: RecordingHealthStatus.recordingComplete, ); } void reset() { state = RideSession(); _resetHealthEvidence(); } void _onTelemetryUpdate(TelemetrySample sample) { _lastTelemetryReceivedTime = DateTime.now(); // Motion-triggered auto-start safeguard: // If in `ready` or `idle` and vehicle speed is detected >= 2.0 MPH, // automatically promote to active recording so physical rides are NEVER lost! if (state.state == RideState.ready) { if (sample.speed != null && sample.speed! >= 1.5) { startRide(); return; } } if (state.state != RideState.riding) return; // RID-3B-1 (invariant I3): telemetry accepted after the finalization // boundary is never attributed to the finishing ride. The gate is private // and is reopened by finishRide()'s finally path, so a retryable ride and // every subsequent ride continue to record normally. if (_telemetryAcceptanceClosed) return; // RID-3C-1: an accepted sample is EVIDENCE, not a health verdict. Record only // what it actually proves - a real GPS fix timestamp, a real IMU sample // timestamp, monotonic timing - and let the contract decide the status. // // RID-3B-1 is preserved twice over: this runs AFTER the acceptance cutoff gate // above, so a post-cutoff sample can never refresh health evidence, and the // existing RID-2 `_isFinishing` guard together with `resolveStatus`'s // lifecycle overrides still prevent recordingFinalizing from being clobbered // back to recordingActive. _recordTelemetryEvidence(sample); if (!_isFinishing) { _evaluateRecordingHealth(); } _queueForBatch(sample); _processStats(sample); try { if (sample.quality != null) { ref .read(fieldTestLoggerProvider) .processTelemetrySample(sample, sample.quality!); } } catch (_) {} } static const int _maxBatchCapacity = 1000; void _queueForBatch(TelemetrySample sample) { if (state.dbId == null) return; // Bound queue to prevent memory leaks during extended DB locks if (_telemetryBatch.length >= _maxBatchCapacity) { _telemetryBatch.removeAt(0); _telemetrySamplesLost++; try { final logger = ref.read(fieldTestLoggerProvider); logger.recordTelemetryLoss(1); logger.log( eventType: FieldTestEventType.databaseQueueOverflow, message: 'Telemetry batch buffer reached capacity (1000); dropping oldest sample', severity: EventSeverity.warning, ); } catch (_) {} } _telemetryBatch.add( TelemetrySamplesCompanion( rideId: drift.Value(state.dbId!), timestamp: drift.Value(sample.timestamp), latitude: drift.Value(sample.latitude), longitude: drift.Value(sample.longitude), altitude: drift.Value(sample.altitude), speed: drift.Value(sample.speed), heading: drift.Value(sample.heading), gpsAccuracy: drift.Value(sample.gpsAccuracy), leanAngle: drift.Value(sample.leanAngle), pitchAngle: drift.Value(sample.pitchAngle), accelX: drift.Value(sample.accelX), accelY: drift.Value(sample.accelY), accelZ: drift.Value(sample.accelZ), longitudinalG: drift.Value(sample.longitudinalG), lateralG: drift.Value(sample.lateralG), ), ); state = state.copyWith( totalSamplesReceived: state.totalSamplesReceived + 1, ); } /// RID-3A ownership semantics, plus RID-3B-1 active-flush publication. /// /// Takes exclusive ownership of the current buffer contents, removes them from /// the live buffer BEFORE awaiting persistence, and restores them to the FRONT /// on failure so chronological identity order survives. Returns true only when /// the owned snapshot is durable (or there was nothing to own); it never /// reports success for a failed persistence. /// /// Tolerant by design for the periodic path: a persistence failure is /// retained, restored and logged with the existing bounded warning, never /// thrown into the timer callback. /// RID-3B-1: finish-only strict drain. /// /// Awaits any active telemetry flush owner first - looping, because a periodic /// tick may take ownership again while waiting - then takes ownership under /// RID-3A's guard and persists. Returns false whenever durability could not be /// established, so finishRide can distinguish success from failure. It never /// reports success for a failed persistence and never silently swallows a /// durability failure. No automatic retry worker is added. Future _drainTelemetryBatchStrictly() async { while (_isFlushingTelemetryBatch) { final active = _activeTelemetryFlush; if (active == null) break; try { await active; } catch (_) { // The owner retains, restores and logs its own failure; ownership is // re-checked by the loop condition. } } if (_telemetryBatch.isEmpty) return true; // Another owner is still active and could not be awaited, so durability is // not established and must not be assumed. if (_isFlushingTelemetryBatch) return false; return _persistOwnedTelemetrySnapshot(); } /// RID-3B-1: the explicit-Stop durability barrier (invariants I1-I3). /// /// Awaits any in-flight periodic flush, drains the buffer to convergence within /// a bounded policy while telemetry capture continues, closes acceptance, then /// performs one final strict drain for anything queued between the last /// convergence check and the cutoff assignment. /// /// Throws if durability cannot be established, so the caller aborts /// finalization without writing endTime, evaluating personal records, /// generating a forensic summary, announcing completion, or emitting the /// sync-visible finished edge. Future _establishTelemetryDurabilityForFinish() async { var durable = true; var passes = 0; // Bounded convergent drain. Telemetry keeps being accepted during this phase // so existing capture semantics are preserved; the final few samples are // never intentionally dropped to simplify finalization. while (passes < _maxFinalDrainPasses && (_telemetryBatch.isNotEmpty || _activeTelemetryFlush != null)) { passes++; if (!await _drainTelemetryBatchStrictly()) { durable = false; break; } } if (durable) { // I3: close acceptance, then one final strict drain for anything queued // between the last convergence check and the cutoff assignment. _telemetryAcceptanceClosed = true; if (!await _drainTelemetryBatchStrictly()) { durable = false; } } // Convergence must actually have been reached. The cap never fabricates // durability: telemetry still outstanding means finalization fails. if (durable && (_telemetryBatch.isNotEmpty || _activeTelemetryFlush != null)) { durable = false; } if (!durable) { // I2: restore a truthful recoverable recording status and propagate, so // RID-2's existing UI boundary containment handles the rejection and the // ride remains retryable. RID-3A has already restored any failed owned // snapshot to the front of the buffer in correct order. state = state.copyWith( healthStatus: RecordingHealthStatus.recordingActive, ); throw StateError( 'Ride finalization aborted: telemetry durability could not be established', ); } } Future _persistOwnedTelemetrySnapshot() async { if (_isFlushingTelemetryBatch || _telemetryBatch.isEmpty) return true; _isFlushingTelemetryBatch = true; final batchToInsert = List.from(_telemetryBatch); _telemetryBatch.clear(); // Publish this exact in-flight operation so an explicit Stop can await the // current owner instead of observing a no-op flush. One Future for one owned // snapshot - a coordination primitive, not a second buffering system. final completion = Completer(); _activeTelemetryFlush = completion.future; try { // ---- PRE-COMMIT PHASE --------------------------------------------------- // This is the ONLY block that may fail persistence. RID-3A ownership, // ordering and front-restoration semantics are unchanged. try { await db.batch((batch) { batch.insertAll(db.telemetrySamples, batchToInsert); }); } catch (e) { // Persistence failed BEFORE the commit point, so the owned snapshot is // not durable. Restore it to the FRONT of the live buffer, preserving // both the snapshot's internal order and the order of samples that // arrived during the await, so a later legitimate flush can retry them. // No automatic retry worker is added. _telemetryBatch.insertAll(0, batchToInsert); // Enforce live queue capacity after failed-snapshot restoration by // dropping the excess oldest samples (from the front), preserving the // existing oldest-first overflow policy and queue ordering. final excess = _telemetryBatch.length - _maxBatchCapacity; if (excess > 0) { _telemetryBatch.removeRange(0, excess); _telemetrySamplesLost += excess; if (mounted) { try { final logger = ref.read(fieldTestLoggerProvider); logger.recordTelemetryLoss(excess); for (var i = 0; i < excess; i++) { logger.log( eventType: FieldTestEventType.databaseQueueOverflow, message: 'Telemetry batch buffer reached capacity (1000); dropping oldest sample', severity: EventSeverity.warning, ); } } catch (_) {} } } // RID-3C-1: a pre-commit failure means this telemetry is NOT durable, so // database health evidence must not advance and is marked non-operational // until a real commit succeeds. _dbPersistenceFailed = true; // Keep in queue for retry on next flush interval. A notifier that has // begun disposal must not touch ref, so the diagnostic is skipped in that // case; it remains bounded either way. if (mounted) { try { ref .read(fieldTestLoggerProvider) .log( eventType: FieldTestEventType.databaseQueueWarning, message: 'Database batch write retry: $e', severity: EventSeverity.warning, ); } catch (_) {} } return false; } // ---- COMMIT POINT REACHED ----------------------------------------------- // The successful return of db.batch(...) above is the telemetry durability // commit point. The owned snapshot is now durable: nothing below may restore // it to the live buffer or change the durability verdict. // Post-commit bookkeeping is notifier-owned, so it is conditional on the // notifier still being alive and isolated in its own guard. `mounted` is // StateNotifier's real property, so this boundary is identical in debug and // release - correctness never depends on a debug assertion firing. After // disposal there is no state read, no state assignment and no ref access. // // Accepted RID-3B-2 counter contract: when disposal races a committed flush, // the durable row count may exceed the dead notifier's final in-memory // totalSamplesPersisted value. That is a terminal counter snapshot on a // disposed notifier, NOT a persistence failure, and no live consumer may // rely on it. Committed samples are never restored to make the counter look // consistent, and no database counter or schema field is added. if (mounted) { try { // RID-3C-1: database health evidence advances ONLY here, strictly after // the RID-3B-2 commit point. A scheduled flush is not evidence, and a // failed pre-commit batch never reaches this block. _lastDbCommitMonotonicMs = _latestObservedMonotonicMs; _dbPersistenceFailed = false; state = state.copyWith( totalSamplesPersisted: state.totalSamplesPersisted + batchToInsert.length, health: state.health.copyWith( lastDatabaseWriteTime: DateTime.now(), ), ); _evaluateRecordingHealth(); } catch (_) { // Post-commit bookkeeping failure must never be classified as a // persistence failure and must never restore the committed snapshot. } } return true; } finally { _isFlushingTelemetryBatch = false; if (!completion.isCompleted) completion.complete(); _activeTelemetryFlush = null; } } /// RID-3B-2 timer Future ownership. /// /// A Timer.periodic callback is synchronous, so the Future returned by /// _flushBatch() would otherwise be discarded and any rejection would surface /// as an uncaught zone error attributed to the timer callback. The Future is /// explicitly owned here instead. /// /// This is a defensive ownership boundary only: no second scheduler, no new /// timer, and no change to the 3-second crash-durability cadence. The timer /// callback itself stays synchronous, so cancellation semantics are unchanged. /// /// Diagnostics are bounded to runtimeType only. The exception object is never /// logged because its message may carry implementation or database detail, and /// no stack trace or telemetry payload is recorded. void _runScheduledFlush() { unawaited( _flushBatch().catchError((Object error, StackTrace stackTrace) { debugPrint('Scheduled telemetry flush failed: ${error.runtimeType}'); }), ); } Future _flushBatch() async { // RID-3A telemetry batch ownership (exclusive snapshot, clear-before-await, // front-restoration on failure, success-only counter) now lives in // _persistOwnedTelemetrySnapshot(), unchanged in behaviour. RID-3B-1 also // publishes the in-flight operation there so an explicit Stop can await the // current owner instead of observing a no-op flush. The periodic path stays // tolerant and never throws into the timer callback; timer Future ownership // and teardown safety are RID-3B-2 scope and are deliberately unchanged. // // The guard is raised synchronously, before the first `await`, so a // competing legitimate flush (the 3 s periodic timer against the flush // awaited by finishRide) returns without taking ownership instead of // snapshotting the very same samples. // // The snapshot is removed from the live buffer BEFORE the persistence await, // so the in-flight operation owns those samples exclusively and telemetry // arriving during the await accumulates in a fresh live buffer. Nothing is // removed from the live buffer after the await - that post-await // `removeRange` is what previously allowed an overlapping flush to discard // newer samples it had never written, while re-inserting samples another // flush had already persisted. // // The incremental ride-stats persistence below is deliberately left outside // this guard so its existing every-tick cadence is unchanged. await _persistOwnedTelemetrySnapshot(); // RID-3B-2: the notifier may have been disposed while this telemetry flush // was in flight, because Timer.cancel() cannot abort an already-started async // callback. The database commit above is already authoritative and is not // revisited; everything below is notifier-owned, so return without reading // state or ref. Incremental stats are never fabricated after disposal, and // stats persistence is never moved ahead of the telemetry commit. if (!mounted) return; // Incrementally persist ongoing ride summary stats to SQLite so crashes/terminations never lose stats if (state.dbId != null && state.state == RideState.riding) { try { await (db.update( db.rides, )..where((t) => t.id.equals(state.dbId!))).write( RidesCompanion( distanceMiles: drift.Value(state.tripDistanceMiles), topSpeedMph: drift.Value(state.topSpeedMph), maxLeanLeft: drift.Value(state.maxLeanLeft), maxLeanRight: drift.Value(state.maxLeanRight), maxAccelG: drift.Value(state.maxAccelG), maxBrakingG: drift.Value(state.maxBrakingG), movingTime: drift.Value(state.movingTimeSeconds), stoppedTime: drift.Value(state.stoppedTimeSeconds), averageSpeed: drift.Value(state.averageSpeedMph), elevationGain: drift.Value(state.elevationGain), elevationLoss: drift.Value(state.elevationLoss), ), ); } catch (_) {} } } void _processStats(TelemetrySample sample) { // Completely ignore rejected samples (NaN, teleportation jumps, extreme spikes) if (sample.quality?.isRejected == true) return; double newTopSpeed = state.topSpeedMph; double newMaxLeanLeft = state.maxLeanLeft; double newMaxLeanRight = state.maxLeanRight; double newMaxAccelG = state.maxAccelG; double newMaxBrakingG = state.maxBrakingG; int newMovingTime = state.movingTimeSeconds; int newStoppedTime = state.stoppedTimeSeconds; double newElevationGain = state.elevationGain; double newElevationLoss = state.elevationLoss; if (sample.isSpeedFresh && sample.speed != null && sample.speed! <= 220.0) { if (sample.speed! > state.topSpeedMph) { newTopSpeed = sample.speed!; } } // Continuous sub-second time accumulation across 20Hz frames if (_lastSample != null && sample.speed != null && sample.isSpeedFresh && _lastSample!.isSpeedFresh) { final dt = sample.timestamp.difference(_lastSample!.timestamp).inMilliseconds / 1000.0; if (dt > 0 && dt < 2.0) { if (sample.speed! > 1.8) { _movingTimeAccumulator += dt; } else { _stoppedTimeAccumulator += dt; } newMovingTime = _movingTimeAccumulator.floor(); newStoppedTime = _stoppedTimeAccumulator.floor(); } } // Speed-gate dynamic motorcycle metrics (require speed >= 3.0 MPH to prevent stationary kickstand / dismount artifacts from registering as personal riding records) final isDynamicallyMoving = sample.isSpeedFresh && sample.speed != null && sample.speed! >= 3.0; if (isDynamicallyMoving && sample.leanAngle != null && sample.leanAngle!.abs() <= 65.0) { if (sample.leanAngle! < 0 && sample.leanAngle!.abs() > state.maxLeanLeft) { newMaxLeanLeft = sample.leanAngle!.abs(); } else if (sample.leanAngle! > 0 && sample.leanAngle! > state.maxLeanRight) { newMaxLeanRight = sample.leanAngle!; } } if (isDynamicallyMoving && sample.longitudinalG != null && sample.longitudinalG!.abs() <= 2.5) { if (sample.longitudinalG! > 0 && sample.longitudinalG! > state.maxAccelG) { newMaxAccelG = sample.longitudinalG!; } else if (sample.longitudinalG! < 0 && sample.longitudinalG!.abs() > state.maxBrakingG) { newMaxBrakingG = sample.longitudinalG!.abs(); } } if (sample.latitude != null && sample.longitude != null && _lastSample?.latitude != null && _lastSample?.longitude != null) { final distMeters = Geolocator.distanceBetween( _lastSample!.latitude!, _lastSample!.longitude!, sample.latitude!, sample.longitude!, ); // Accumulate distance ONLY if moving and GPS accuracy is reasonable (suppresses stationary jitter at stop lights) final hasFreshGps = sample.isSpeedFresh && (_lastSample?.isSpeedFresh ?? false); final isMoving = (sample.speed != null && sample.speed! >= 1.0) || (_lastSample!.speed != null && _lastSample!.speed! >= 1.0); final hasGoodGps = sample.gpsAccuracy != null && sample.gpsAccuracy! <= 35.0; if (hasFreshGps && isMoving && hasGoodGps && distMeters > 0.3 && distMeters < 150.0) { _accumulatedDistance += distMeters; } // Elevation tracking if (hasFreshGps && sample.altitude != null && _lastSample!.altitude != null) { final elevationDelta = sample.altitude! - _lastSample!.altitude!; // Threshold filter to prevent phantom gain (e.g. > 1 meter change) if (elevationDelta.abs() > 1.0 && elevationDelta.abs() < 50.0) { if (elevationDelta > 0) { newElevationGain += elevationDelta; } else { newElevationLoss += elevationDelta.abs(); } } } } double tripDistanceMiles = _accumulatedDistance / 1609.34; // Calculate avg speed (only moving time) double avgSpeed = 0; if (newMovingTime > 0) { // miles per hour avgSpeed = (tripDistanceMiles / newMovingTime) * 3600; } _detectEvents(sample); _lastSample = sample; state = state.copyWith( topSpeedMph: newTopSpeed, maxLeanLeft: newMaxLeanLeft, maxLeanRight: newMaxLeanRight, maxAccelG: newMaxAccelG, maxBrakingG: newMaxBrakingG, tripDistanceMiles: tripDistanceMiles, movingTimeSeconds: newMovingTime, stoppedTimeSeconds: newStoppedTime, elevationGain: newElevationGain, elevationLoss: newElevationLoss, averageSpeedMph: avgSpeed, ); } void _detectEvents(TelemetrySample sample) { if (state.dbId == null) return; final now = DateTime.now(); void checkAndLogEvent(String type, double value, bool condition) { if (condition) { final lastTime = _lastEventTimes[type]; if (lastTime == null || now.difference(lastTime) > _eventCooldown) { _lastEventTimes[type] = now; db .into(db.rideEvents) .insert( RideEventsCompanion.insert( rideId: state.dbId!, timestamp: now, eventType: type, value: value, latitude: drift.Value(sample.latitude), longitude: drift.Value(sample.longitude), ), ); } } } if (sample.longitudinalG != null) { checkAndLogEvent( 'Hard Braking', sample.longitudinalG!, sample.longitudinalG! < -0.8, ); checkAndLogEvent( 'Hard Acceleration', sample.longitudinalG!, sample.longitudinalG! > 0.6, ); } if (sample.leanAngle != null) { checkAndLogEvent( 'Extreme Lean', sample.leanAngle!, sample.leanAngle!.abs() > 45.0, ); } if (sample.isSpeedFresh && sample.speed != null) { checkAndLogEvent('High Speed', sample.speed!, sample.speed! > 100.0); } } @override void dispose() { _telemetrySub?.cancel(); _batchWriteTimer?.cancel(); _watchdogTimer?.cancel(); super.dispose(); } } final rideStateMachineProvider = StateNotifierProvider((ref) { final db = ref.watch(databaseProvider); final telemetryService = ref.watch(telemetryServiceProvider); final recordsService = ref.watch(recordsServiceProvider.notifier); final audioAlertService = ref.watch(audioAlertServiceProvider); return RideStateMachine( db, telemetryService, recordsService, audioAlertService, ref, ); });