meshcore-client/lib/services/repeater_command_service.dart

import 'dart:async';
import 'package:flutter/foundation.dart';
import '../models/contact.dart';
import '../models/path_selection.dart';
import '../connector/meshcore_connector.dart';
import '../connector/meshcore_protocol.dart';
import '../utils/app_logger.dart';

/// A repeater reply that arrived with no command waiting to receive it.
///
/// [command] carries the original request when the reply could be traced back
/// to one that already timed out, and is null when the reply cannot be
/// attributed to anything this service sent.
class UnmatchedRepeaterResponse {
  final String repeaterKeyHex;
  final String response;
  final String? command;
  final Duration? sinceTimeout;

  const UnmatchedRepeaterResponse({
    required this.repeaterKeyHex,
    required this.response,
    this.command,
    this.sinceTimeout,
  });

  /// True when this answers a request that timed out rather than arriving
  /// unsolicited.
  bool get isLateReply => command != null;
}

/// Thrown when a CLI command's window closes with no reply.
///
/// Carries the window that was actually armed, so callers can report the real
/// figure. The previous message printed `(timeoutMs / 1000).ceil()`, so a
/// 4074 ms window announced "timeout after 5 seconds": every window in
/// (4000, 5000] reported the same 5, and the number shown was never the number
/// used (#531).
class RepeaterCommandTimeout implements Exception {
  final String command;
  final int timeoutMs;

  const RepeaterCommandTimeout({
    required this.command,
    required this.timeoutMs,
  });

  /// The armed window in seconds to one decimal, e.g. `28.7`.
  String get secondsText => (timeoutMs / 1000).toStringAsFixed(1);

  /// Non-localized fallback for logs and for callers without a
  /// [BuildContext]. UI paths should prefer [secondsText] with their own
  /// localized string.
  @override
  String toString() => 'Command timed out after $secondsText seconds';
}

class _ExpiredCommand {
  final String command;
  final DateTime expiredAt;

  const _ExpiredCommand({required this.command, required this.expiredAt});
}

class RepeaterCommandService {
  final MeshCoreConnector _connector;
  final Map<String, Completer<String>> _pendingCommands = {};
  final Map<String, Timer> _commandTimeouts = {};
  final Map<String, String> _commandPrefixes = {};
  final Map<String, String> _pendingByPrefix = {};
  final Map<String, _ExpiredCommand> _expiredCommands = {};
  int _prefixCounter = 0;

  static const int maxRetries = 5;

  /// How long a timed-out command stays remembered so a reply arriving after
  /// its window can still be presented with the request it answers.
  static const Duration lateReplyRetention = Duration(minutes: 2);

  /// Invoked when a reply cannot be handed to a waiting command. Consumers
  /// must surface this to the user: the reply is a real answer from the
  /// repeater and dropping it silently loses it for good (#528).
  void Function(UnmatchedRepeaterResponse)? onUnmatchedResponse;

  RepeaterCommandService(this._connector);

  /// Send a CLI command to a repeater with automatic retries
  /// Returns a future that completes when a response is received or after max retries
  Future<String> sendCommand(
    Contact repeater,
    String command, {
    Function(String)? onResponse,
    Function(int)? onAttempt,
    int retries = maxRetries,
  }) async {
    final attemptCount = retries < 1 ? 1 : retries;
    final selection = await _connector.preparePathForContactSend(repeater);
    final attemptPrefixes = <String>[];

    for (int attempt = 0; attempt < attemptCount; attempt++) {
      onAttempt?.call(attempt + 1);
      try {
        final response = await _sendCommandAttempt(
          repeater,
          command,
          selection,
          attempt,
          attemptPrefixes,
        );
        // The caller has its answer, so a straggler from an earlier attempt of
        // this same command is noise rather than a lost response.
        for (final prefix in attemptPrefixes) {
          _expiredCommands.remove(prefix);
        }
        onResponse?.call(response);
        return response;
      } catch (e) {
        if (attempt == attemptCount - 1) rethrow;
      }
    }

    throw Exception('Command failed after $attemptCount attempts');
  }

  Future<String> _sendCommandAttempt(
    Contact repeater,
    String command,
    PathSelection selection,
    int attempt,
    List<String> attemptPrefixes,
  ) async {
    final repeaterKey = repeater.publicKeyHex;
    final prefix = _nextPrefixToken();
    final commandId = '${repeaterKey}_$prefix';
    final completer = _registerPending(commandId, prefix);
    attemptPrefixes.add(prefix);

    try {
      final framedCommand = '$prefix$command';
      final pathLengthValue = selection.useFlood ? -1 : selection.hopCount;
      final timestampSeconds = DateTime.now().millisecondsSinceEpoch ~/ 1000;
      _connector.trackRepeaterAck(
        contact: repeater,
        selection: selection,
        text: framedCommand,
        timestampSeconds: timestampSeconds,
        attempt: attempt,
      );
      final frame = buildSendCliCommandFrame(
        repeater.publicKey,
        framedCommand,
        attempt: attempt,
        timestampSeconds: timestampSeconds,
      );
      final responseBytes = frame.length > maxFrameSize
          ? frame.length
          : maxFrameSize;
      final timeoutMs = _connector.calculateCliTimeout(
        pathLength: pathLengthValue,
        messageBytes: responseBytes,
      );
      await _connector.sendFrame(frame);
      _commandTimeouts[commandId]?.cancel();
      _commandTimeouts[commandId] = Timer(
        Duration(milliseconds: timeoutMs),
        () {
          final completer = _pendingCommands[commandId];
          if (completer != null && !completer.isCompleted) {
            // Prune here as well as on receive. A run of commands that all
            // time out with no traffic coming back would otherwise accumulate
            // records until disposal, since handleResponse is the only other
            // place that prunes and it never runs (Gemini review).
            _pruneExpiredCommands();
            // Remember what this prefix asked so a reply arriving after the
            // window can still reach the user with its question attached.
            _expiredCommands[prefix] = _ExpiredCommand(
              command: command,
              expiredAt: DateTime.now(),
            );
            completer.completeError(
              RepeaterCommandTimeout(command: command, timeoutMs: timeoutMs),
            );
            _cleanup(commandId);
          }
        },
      );
    } catch (e) {
      _cleanup(commandId);
      throw Exception('Failed to send command: $e');
    }

    try {
      return await completer.future;
    } finally {
      _cleanup(commandId);
    }
  }

  /// Call this when a text message response is received from a repeater
  void handleResponse(Contact repeater, String responseText) {
    final repeaterKey = repeater.publicKeyHex;
    _pruneExpiredCommands();

    String? prefix;
    String responsePayload = responseText;
    if (responseText.length >= 3 && responseText[2] == '|') {
      prefix = responseText.substring(0, 3);
      responsePayload = responseText.substring(3).trimLeft();
    }

    // A reply that carries a prefix is self-identifying. If that prefix matches
    // nothing pending, the answer belongs to a command that already completed
    // or expired, and forcing it onto whatever else is in flight would report
    // one command's output as another command's result (#532). Only a reply
    // with no prefix at all, which carries no correlation token, may fall back
    // to matching by repeater.
    final matchedId = prefix != null ? _pendingByPrefix[prefix] : null;
    final commandId =
        matchedId ?? (prefix != null ? '' : _solePendingFor(repeaterKey));

    if (commandId.isNotEmpty) {
      final completer = _pendingCommands[commandId];
      if (completer != null && !completer.isCompleted) {
        completer.complete(responsePayload);
        _cleanup(commandId);
        return;
      }
    }

    // Nothing is waiting for this reply. It is still a real answer from the
    // repeater, so it gets surfaced and logged rather than dropped (#528).
    _surfaceUnmatchedResponse(repeaterKey, prefix, responsePayload);
  }

  void _surfaceUnmatchedResponse(
    String repeaterKey,
    String? prefix,
    String responsePayload,
  ) {
    final expired = prefix != null ? _expiredCommands.remove(prefix) : null;
    final sinceTimeout = expired == null
        ? null
        : DateTime.now().difference(expired.expiredAt);

    if (expired != null) {
      appLogger.warn(
        'Late reply to "${expired.command}" from $repeaterKey arrived '
        '${sinceTimeout!.inMilliseconds}ms after its window closed',
        tag: 'RepeaterCommand',
      );
    } else {
      appLogger.warn(
        'Reply from $repeaterKey matched no pending or recently expired '
        'command (prefix: ${prefix ?? 'none'})',
        tag: 'RepeaterCommand',
      );
    }

    onUnmatchedResponse?.call(
      UnmatchedRepeaterResponse(
        repeaterKeyHex: repeaterKey,
        response: responsePayload,
        command: expired?.command,
        sinceTimeout: sinceTimeout,
      ),
    );
  }

  /// The one command awaiting a reply from [repeaterKey], or `''` when that is
  /// ambiguous.
  ///
  /// An unprefixed reply carries no correlation token, so position is all that
  /// is left to match on. Firmware only echoes the prefix when the command is
  /// longer than four characters including it (`simple_repeater/MyMesh.cpp`
  /// `strlen(command) > 4 && command[2] == '|'`), so a very short command can
  /// legitimately answer without one and the fallback has to stay.
  ///
  /// It is only sound with exactly one candidate though. With two or more,
  /// "first" is map-iteration order, and choosing one would hand a caller
  /// another command's output. Ambiguity is surfaced instead of guessed.
  String _solePendingFor(String repeaterKey) {
    String? only;
    for (final id in _pendingCommands.keys) {
      if (!id.startsWith(repeaterKey)) continue;
      if (only != null) return '';
      only = id;
    }
    return only ?? '';
  }

  Completer<String> _registerPending(String commandId, String prefix) {
    final completer = Completer<String>();
    _pendingCommands[commandId] = completer;
    _commandPrefixes[commandId] = prefix;
    _pendingByPrefix[prefix] = commandId;
    return completer;
  }

  /// Puts a command in flight without performing the send, so a test can set up
  /// two concurrent commands to one repeater. Deliberately routes through the
  /// same [_registerPending] the real send path uses, so the two cannot drift.
  @visibleForTesting
  Future<String> registerPendingForTest(String repeaterKeyHex, String prefix) =>
      _registerPending('${repeaterKeyHex}_$prefix', prefix).future;

  /// Records a timed-out command so a reply arriving later can still be
  /// attributed to it. Exposed because a test cannot drive a real
  /// send-then-time-out cycle without a connected transport.
  @visibleForTesting
  void recordExpiredCommandForTest(String prefix, String command) {
    _expiredCommands[prefix] = _ExpiredCommand(
      command: command,
      expiredAt: DateTime.now(),
    );
  }

  void _pruneExpiredCommands() {
    if (_expiredCommands.isEmpty) return;
    final cutoff = DateTime.now().subtract(lateReplyRetention);
    _expiredCommands.removeWhere(
      (_, entry) => entry.expiredAt.isBefore(cutoff),
    );
  }

  void _cleanup(String commandId) {
    _commandTimeouts[commandId]?.cancel();
    _commandTimeouts.remove(commandId);
    _pendingCommands.remove(commandId);
    final prefix = _commandPrefixes.remove(commandId);
    if (prefix != null) {
      _pendingByPrefix.remove(prefix);
    }
  }

  void dispose() {
    for (final timer in _commandTimeouts.values) {
      timer.cancel();
    }
    _commandTimeouts.clear();
    _pendingCommands.clear();
    _commandPrefixes.clear();
    _pendingByPrefix.clear();
    _expiredCommands.clear();
    onUnmatchedResponse = null;
  }

  String _nextPrefixToken() {
    for (var i = 0; i < 256; i++) {
      final value = _prefixCounter++ & 0xFF;
      final token = '${value.toRadixString(16).padLeft(2, '0').toUpperCase()}|';
      if (!_pendingByPrefix.containsKey(token)) {
        return token;
      }
    }
    return '00|';
  }
}