import 'dart:async'; import 'package:flutter_riverpod/flutter_riverpod.dart'; import '../../../core/observations/observed_event.dart'; import '../../../core/observations/vehicle_data_port.dart'; import '../../../core/observations/vehicle_data_update.dart'; import '../../../core/transport/replay_vehicle_data_port.dart'; final vehicleDataPortProvider = Provider((ref) { final port = const ReplayVehicleDataPort(); ref.onDispose(() => unawaited(port.close())); return port; }); final observabilityControllerProvider = NotifierProvider( ObservabilityController.new, ); class ObservabilityController extends Notifier { static const _criticalAnnouncementWindow = Duration(seconds: 2); static const _timelineCapacity = 500; StreamSubscription? _subscription; DateTime? _lastCriticalAnnouncementAt; @override ObservabilityState build() { final source = ref.read(vehicleDataPortProvider); ref.onDispose(_cancelSubscription); Future.microtask(() => _observe(source)); return ObservabilityState.initial(source); } void _observe(VehicleDataPort source) { _subscription = source.observe().listen( _handleUpdate, onError: (Object _, StackTrace _) { state = state.copyWith( status: ObservationStatus.disconnected, accessibilityAnnouncement: 'Observation connection closed.', replaceAccessibilityAnnouncement: true, ); }, onDone: () { state = state.copyWith( status: source.isReplay ? ObservationStatus.replayComplete : ObservationStatus.disconnected, accessibilityAnnouncement: source.isReplay ? 'Replay complete.' : 'Observation connection closed.', replaceAccessibilityAnnouncement: true, ); }, ); } void _handleUpdate(VehicleDataUpdate update) { switch (update) { case ObservationReceived(:final event): _record(event); case ObservationDropped(:final kind, :final count): state = state.copyWith( status: ObservationStatus.receiving, parserDropCount: state.parserDropCount + (kind == ObservationDropKind.parser ? count : 0), semanticDropCount: state.semanticDropCount + (kind == ObservationDropKind.semantic ? count : 0), ); } } void _record(ObservedEvent event) { final announcement = _announcementFor(event); final retention = _isTimelineEvent(event) ? _retainTimelineEvent(state.events, event) : _TimelineRetention(state.events); final heartbeat = event is HeartbeatEvent ? event : state.lastHeartbeat; final observedStreams = switch (event) { TelemetryEvent(:final kind) => {...state.observedTelemetryKinds, kind}, _ => state.observedTelemetryKinds, }; state = state.copyWith( status: ObservationStatus.receiving, lastFrameAt: event.receivedAt, events: retention.events, lastHeartbeat: heartbeat, observedTelemetryKinds: Set.unmodifiable(observedStreams), localTimelineEvictionCount: state.localTimelineEvictionCount + retention.evictionCount, accessibilityAnnouncement: announcement.message, replaceAccessibilityAnnouncement: true, coalescedAlertCount: state.coalescedAlertCount + (announcement.coalesced ? 1 : 0), ); } _AnnouncementDecision _announcementFor(ObservedEvent event) { final candidate = switch (event) { ErrorEvent(:final path) => 'Error received: $path', ShutdownEvent(:final reason) => 'Orderly shutdown observed: ${reason.label}', _ => null, }; if (candidate == null) { return const _AnnouncementDecision(); } final now = DateTime.now(); final previous = _lastCriticalAnnouncementAt; if (previous != null && now.difference(previous) < _criticalAnnouncementWindow) { return const _AnnouncementDecision(coalesced: true); } _lastCriticalAnnouncementAt = now; return _AnnouncementDecision(message: candidate); } void _cancelSubscription() { _subscription?.cancel(); } _TimelineRetention _retainTimelineEvent( List current, ObservedEvent incoming, ) { if (current.length < _timelineCapacity) { return _TimelineRetention([...current, incoming]); } final retained = List.of(current); final optionalIndex = retained.indexWhere((event) => !_isCritical(event)); if (optionalIndex >= 0) { retained.removeAt(optionalIndex); retained.add(incoming); return _TimelineRetention(retained, evictionCount: 1); } if (_isCritical(incoming)) { retained.removeAt(0); retained.add(incoming); return _TimelineRetention(retained, evictionCount: 1); } return _TimelineRetention(retained, evictionCount: 1); } } bool _isCritical(ObservedEvent event) => event is ErrorEvent || event is ShutdownEvent; bool _isTimelineEvent(ObservedEvent event) => event is LogEvent || _isCritical(event); class _TimelineRetention { _TimelineRetention(Iterable events, {this.evictionCount = 0}) : events = List.unmodifiable(events); final List events; final int evictionCount; } enum ObservationStatus { starting('Starting'), receiving('Observing'), replayComplete('Replay complete'), disconnected('Disconnected'); const ObservationStatus(this.label); final String label; } class ObservabilityState { const ObservabilityState({ required this.sourceLabel, required this.isReplay, required this.status, required this.events, required this.lastHeartbeat, required this.observedTelemetryKinds, required this.parserDropCount, required this.semanticDropCount, required this.coalescedAlertCount, required this.localTimelineEvictionCount, this.lastFrameAt, this.accessibilityAnnouncement, }); factory ObservabilityState.initial(VehicleDataPort source) { return ObservabilityState( sourceLabel: source.label, isReplay: source.isReplay, status: ObservationStatus.starting, events: const [], lastHeartbeat: null, observedTelemetryKinds: const {}, parserDropCount: 0, semanticDropCount: 0, coalescedAlertCount: 0, localTimelineEvictionCount: 0, ); } final String sourceLabel; final bool isReplay; final ObservationStatus status; final List events; final HeartbeatEvent? lastHeartbeat; final Set observedTelemetryKinds; final int parserDropCount; final int semanticDropCount; final int coalescedAlertCount; final int localTimelineEvictionCount; final DateTime? lastFrameAt; final String? accessibilityAnnouncement; ObservabilityState copyWith({ ObservationStatus? status, List? events, HeartbeatEvent? lastHeartbeat, Set? observedTelemetryKinds, int? parserDropCount, int? semanticDropCount, int? coalescedAlertCount, int? localTimelineEvictionCount, DateTime? lastFrameAt, String? accessibilityAnnouncement, bool replaceAccessibilityAnnouncement = false, }) { return ObservabilityState( sourceLabel: sourceLabel, isReplay: isReplay, status: status ?? this.status, events: events ?? this.events, lastHeartbeat: lastHeartbeat ?? this.lastHeartbeat, observedTelemetryKinds: observedTelemetryKinds ?? this.observedTelemetryKinds, parserDropCount: parserDropCount ?? this.parserDropCount, semanticDropCount: semanticDropCount ?? this.semanticDropCount, coalescedAlertCount: coalescedAlertCount ?? this.coalescedAlertCount, localTimelineEvictionCount: localTimelineEvictionCount ?? this.localTimelineEvictionCount, lastFrameAt: lastFrameAt ?? this.lastFrameAt, accessibilityAnnouncement: replaceAccessibilityAnnouncement ? accessibilityAnnouncement : this.accessibilityAnnouncement, ); } } class _AnnouncementDecision { const _AnnouncementDecision({this.message, this.coalesced = false}); final String? message; final bool coalesced; }