hondavert-dev/lib/core/bluetooth/bt_poller.dart

import 'dart:async';
import 'dart:typed_data';

import '../logging/app_log.dart';
import '../protocol/neg8.dart';
import 'bt_service.dart';

enum EcuType { s300, kpro }

typedef _Matcher = Uint8List? Function(List<int> rxBuf);

class _Pending {
  _Pending(this.match);
  final _Matcher match;
  final Completer<Uint8List?> completer = Completer<Uint8List?>();
}

/// Strict request/response poller for the HV BT adapter.
///
/// Rules (agreed with the device maker's feedback — the adapter MCU must not
/// be flooded):
///  * Exactly ONE request is outstanding at any time. A new request is only
///    sent after the previous one was answered with a complete, checksum-valid
///    128-byte frame, or after [responseTimeout] expired.
///  * After a timeout the link is left quiet for [quietAfterTimeout] and the
///    receive buffer is flushed, so a late reply can't be mistaken for the
///    answer to the next request.
///  * Requests are paced: at most one every [pollingInterval] (measured from
///    request to request) and never less than [minGap] after a reply.
///  * After [stallThreshold] consecutive timeouts, [onLinkStalled] fires so the
///    owner can reconnect.
///
/// Other commands (DTC reset, protocol probe) go through the same lock, so they
/// never interleave with a datalog request.
class BtPoller {
  BtPoller(
    this._service, {
    this.ecuType = EcuType.s300,
    Duration pollingInterval = defaultInterval,
    this.responseTimeout = const Duration(milliseconds: 500),
    this.onLinkStalled,
  }) : _pollingInterval = _sanitize(pollingInterval);

  static const Duration defaultInterval = Duration(milliseconds: 100);
  static const Duration minInterval = Duration(milliseconds: 80);
  static const Duration minGap = Duration(milliseconds: 20);
  static const Duration quietAfterTimeout = Duration(milliseconds: 150);
  static const int stallThreshold = 6;
  static const int frameLength = 128;
  static const int _maxBuffer = 4096;

  // S300 request: [0x1B, 0x00, NEG8] / KPro request: [0x1B, 0x01, NEG8]
  static final Uint8List _s300Request = Uint8List.fromList([0x1B, 0x00, 0xE5]);
  static final Uint8List _kproRequest = Uint8List.fromList([0x1B, 0x01, 0xE4]);

  final BtService _service;
  EcuType ecuType;
  final Duration responseTimeout;
  final void Function()? onLinkStalled;

  Duration _pollingInterval;
  Duration get pollingInterval => _pollingInterval;
  set pollingInterval(Duration d) => _pollingInterval = _sanitize(d);

  static Duration _sanitize(Duration d) => d < minInterval ? minInterval : d;

  final StreamController<Uint8List> _frameController =
      StreamController<Uint8List>.broadcast();

  /// Checksum-valid 128-byte frames, one per answered request.
  Stream<Uint8List> get frameStream => _frameController.stream;

  final List<int> _rxBuf = [];
  StreamSubscription<Uint8List>? _rxSub;
  _Pending? _pending;
  bool _running = false;
  int _generation = 0;

  // Serialises every exchange on the link (polls and one-off commands).
  Future<void> _lockTail = Future<void>.value();

  // ── Stats (read by the debug screen) ──────────────────────────────────────
  int requestsSent = 0;
  int validFrames = 0;
  int droppedFrames = 0; // checksum failures
  int timeouts = 0;
  int consecutiveTimeouts = 0;
  Duration? lastResponseTime;

  bool get isRunning => _running;

  void start() {
    if (_running) return;
    _running = true;
    final gen = ++_generation;
    _rxBuf.clear();
    consecutiveTimeouts = 0;
    _rxSub = _service.rawStream.listen(
      _onRx,
      onError: (Object e) {
        AppLog.w('BT stream error: $e');
        if (!_frameController.isClosed) _frameController.addError(e);
      },
    );
    AppLog.i('Poller started (${ecuType.name}, every '
        '${_pollingInterval.inMilliseconds} ms, timeout '
        '${responseTimeout.inMilliseconds} ms)');
    unawaited(_loop(gen));
  }

  void stop() {
    if (!_running) return;
    _running = false;
    _generation++;
    _rxSub?.cancel();
    _rxSub = null;
    final p = _pending;
    if (p != null && !p.completer.isCompleted) p.completer.complete(null);
    _pending = null;
    _rxBuf.clear();
    AppLog.i('Poller stopped (sent=$requestsSent ok=$validFrames '
        'timeouts=$timeouts badChecksum=$droppedFrames)');
  }

  void dispose() {
    stop();
    _frameController.close();
  }

  // ── Main loop ─────────────────────────────────────────────────────────────

  Future<void> _loop(int gen) async {
    bool alive() => _running && gen == _generation;

    while (alive()) {
      final started = DateTime.now();

      await _withLink(() async {
        if (!alive() || !_service.isConnected) return;
        final req = ecuType == EcuType.s300 ? _s300Request : _kproRequest;
        final frame = await _transact(req, _matchDatalogFrame, responseTimeout);
        if (!alive()) return;

        if (frame != null) {
          validFrames++;
          consecutiveTimeouts = 0;
          lastResponseTime = DateTime.now().difference(started);
          if (!_frameController.isClosed) _frameController.add(frame);
        } else {
          timeouts++;
          consecutiveTimeouts++;
          if (consecutiveTimeouts == 1 || consecutiveTimeouts % 10 == 0) {
            AppLog.w('No reply within ${responseTimeout.inMilliseconds} ms '
                '(consecutive=$consecutiveTimeouts, total=$timeouts)');
          }
          // Let any late reply arrive and be discarded before the next request.
          await Future<void>.delayed(quietAfterTimeout);
          _rxBuf.clear();
        }
      });

      if (!alive()) break;

      if (consecutiveTimeouts >= stallThreshold) {
        consecutiveTimeouts = 0;
        AppLog.w('Link stalled: $stallThreshold requests in a row unanswered');
        onLinkStalled?.call();
      }

      final elapsed = DateTime.now().difference(started);
      var wait = _pollingInterval - elapsed;
      if (wait < minGap) wait = minGap;
      await Future<void>.delayed(wait);
    }
  }

  // ── One-off commands ──────────────────────────────────────────────────────

  /// Sends [command] (between two polls, never overlapping one) and waits up
  /// to [timeout] for [expected] to appear in the reply bytes.
  Future<({bool acknowledged, Uint8List raw})> sendCommandAwaitReply(
    Uint8List command,
    Uint8List expected, {
    Duration timeout = const Duration(seconds: 2),
  }) {
    return _withLink(() async {
      await Future<void>.delayed(quietAfterTimeout);
      _rxBuf.clear();
      final match = await _transact(
        command,
        (buf) => _indexOf(buf, expected) >= 0 ? expected : null,
        timeout,
      );
      final raw = Uint8List.fromList(_rxBuf);
      _rxBuf.clear();
      return (acknowledged: match != null, raw: raw);
    });
  }

  /// Sends [command] and returns every byte received during [window].
  Future<Uint8List> sendCommandCollect(Uint8List command, Duration window) {
    return _withLink(() async {
      await Future<void>.delayed(quietAfterTimeout);
      _rxBuf.clear();
      await _transact(command, (_) => null, window);
      final raw = Uint8List.fromList(_rxBuf);
      _rxBuf.clear();
      return raw;
    });
  }

  // ── Internals ─────────────────────────────────────────────────────────────

  Future<T> _withLink<T>(Future<T> Function() action) {
    final previous = _lockTail;
    final release = Completer<void>();
    _lockTail = release.future;
    return previous.then((_) => action()).whenComplete(release.complete);
  }

  /// Writes [request] and waits for [match] to succeed or [timeout]. Must be
  /// called while holding the link lock.
  Future<Uint8List?> _transact(
    Uint8List request,
    _Matcher match,
    Duration timeout,
  ) async {
    _rxBuf.clear();
    final pending = _Pending(match);
    _pending = pending;
    try {
      await _service.write(request).timeout(const Duration(seconds: 1));
      requestsSent++;
    } catch (e) {
      AppLog.w('BT write failed: $e');
      if (identical(_pending, pending)) _pending = null;
      // Don't spin on a broken link.
      await Future<void>.delayed(quietAfterTimeout);
      return null;
    }
    _tryComplete();
    try {
      return await pending.completer.future.timeout(timeout);
    } on TimeoutException {
      return null;
    } finally {
      if (identical(_pending, pending)) _pending = null;
    }
  }

  void _onRx(Uint8List chunk) {
    if (_pending == null) {
      // Nobody is waiting: stray/late bytes. Drop them.
      _rxBuf.clear();
      return;
    }
    _rxBuf.addAll(chunk);
    if (_rxBuf.length > _maxBuffer) {
      _rxBuf.removeRange(0, _rxBuf.length - _maxBuffer);
    }
    _tryComplete();
  }

  void _tryComplete() {
    final p = _pending;
    if (p == null || p.completer.isCompleted) return;
    final result = p.match(_rxBuf);
    if (result != null) p.completer.complete(result);
  }

  /// Finds a complete, checksum-valid 128-byte datalog frame starting with
  /// the expected 2-byte header. Consumes bytes it has ruled out.
  Uint8List? _matchDatalogFrame(List<int> buf) {
    final id2 = ecuType == EcuType.s300 ? 0x00 : 0x01;
    while (true) {
      var i = 0;
      while (i + 1 < buf.length && !(buf[i] == 0x1B && buf[i + 1] == id2)) {
        i++;
      }
      if (buf.length - i < frameLength) {
        if (i > 0) buf.removeRange(0, i);
        return null;
      }
      final frame = Uint8List.fromList(buf.sublist(i, i + frameLength));
      if (validateFrame(frame)) {
        buf.removeRange(0, i + frameLength);
        return frame;
      }
      // Bad checksum: either a false header inside data or a corrupted frame.
      droppedFrames++;
      buf.removeRange(0, i + 1);
    }
  }

  static int _indexOf(List<int> buf, List<int> pattern) {
    for (var i = 0; i + pattern.length <= buf.length; i++) {
      var ok = true;
      for (var j = 0; j < pattern.length; j++) {
        if (buf[i + j] != pattern[j]) {
          ok = false;
          break;
        }
      }
      if (ok) return i;
    }
    return -1;
  }
}