app-flutter/lib/boundary.dart

// Client side of the display boundary, protocol v0 (contracts docs/BOUNDARY.md).
//
// This side always connects and owns reconnection. The link drops routinely —
// ~3 s at every power change, and an asleep peer looks connected — so silence is
// treated as a dead link, and commands are retried under the same id until acked.

import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'dart:math';

import 'package:flutter/foundation.dart';

enum LinkState { connecting, up, down }

class PendingCommand {
  PendingCommand(this.id, this.cmd, this.args) : queuedAt = DateTime.now();
  final String id;
  final String cmd;
  final Map<String, dynamic> args;
  final DateTime queuedAt;
}

class Ack {
  Ack(this.id, this.ok, this.detail, this.at);
  final String id;
  final bool ok;
  final String detail;
  final DateTime at;
}

class Boundary extends ChangeNotifier {
  Boundary(this.url, {this.connectTimeout = const Duration(seconds: 3)});

  String url;
  final Duration connectTimeout;

  LinkState state = LinkState.down;
  String detail = 'not started';
  Map<String, dynamic>? hello;
  Map<String, dynamic> signals = {};

  /// Last envelope of every signal this gateway has sent. A name here but not in
  /// [signals] is missing right now — distinct from null and stale.
  final Map<String, Map<String, dynamic>> seen = {};
  DateTime? lastState;
  DateTime? lastMessage;

  /// Orin clock minus tablet clock, from heartbeats. Large values mean one side is unsynced.
  Duration? clockOffset;
  final List<String> recent = [];
  final List<PendingCommand> pending = [];
  final List<Ack> acks = [];

  WebSocket? _ws;
  Timer? _hb;
  Timer? _watchdog;
  bool _running = false;
  int _attempt = 0;
  final _rng = Random();

  /// True when any message seen on this connection said mock.
  bool get mock => hello?['mock'] == true || signals.values.any((e) => e is Map && e['mock'] == true);

  Duration get _hbInterval => Duration(milliseconds: (hello?['hb_ms'] as int?) ?? 1000);
  Duration get _timeout => Duration(milliseconds: (hello?['timeout_ms'] as int?) ?? 5000);

  void start() {
    if (_running) return;
    _running = true;
    _loop();
  }

  Future<void> stop() async {
    _running = false;
    await _ws?.close();
  }

  /// Change the target; the current connection is dropped and the loop reconnects.
  void retarget(String newUrl) {
    url = newUrl;
    seen.clear();
    _ws?.close();
  }

  /// Queue a command; it is sent now if connected and resent after every reconnect until acked.
  String command(String cmd, [Map<String, dynamic> args = const {}]) {
    final id = '${DateTime.now().microsecondsSinceEpoch.toRadixString(36)}-${_rng.nextInt(1 << 32).toRadixString(36)}';
    final c = PendingCommand(id, cmd, args);
    pending.add(c);
    _send(c);
    notifyListeners();
    return id;
  }

  /// Send once, never resent: for commands that must not run twice, like `poweroff`, whose
  /// effect wipes svc-gw's ack cache. Null when not connected; nothing is queued then.
  String? commandOnce(String cmd, [Map<String, dynamic> args = const {}]) {
    if (_ws == null || hello == null) return null;
    final id = '${DateTime.now().microsecondsSinceEpoch.toRadixString(36)}-${_rng.nextInt(1 << 32).toRadixString(36)}';
    _add(jsonEncode({'t': 'cmd', 'id': id, 'cmd': cmd, 'args': args}));
    return id;
  }

  /// The ack for [id], if it arrives within [within].
  Future<Ack?> ackFor(String id, {Duration within = const Duration(seconds: 3)}) async {
    final end = DateTime.now().add(within);
    while (DateTime.now().isBefore(end)) {
      for (final a in acks) {
        if (a.id == id) return a;
      }
      await Future.delayed(const Duration(milliseconds: 100));
    }
    return null;
  }

  Future<void> _loop() async {
    while (_running) {
      _set(LinkState.connecting, 'connecting to $url');
      try {
        final ws = await WebSocket.connect(url).timeout(connectTimeout);
        _ws = ws;
        _attempt = 0;
        lastMessage = DateTime.now();
        _armWatchdog();
        await for (final raw in ws) {
          _onMessage(raw as String);
        }
        _set(LinkState.down, 'closed${ws.closeReason != null && ws.closeReason!.isNotEmpty ? ': ${ws.closeReason}' : ''}');
      } catch (e) {
        _set(LinkState.down, '$e'.split('\n').first);
      } finally {
        _hb?.cancel();
        _watchdog?.cancel();
        _ws = null;
        hello = null;
      }
      if (!_running) break;
      // 0.5 s, 1 s, 2 s, then every 3 s: a replugged hub is back in about 3 s.
      final wait = Duration(milliseconds: min(3000, 500 * (1 << min(_attempt, 3))));
      _attempt++;
      await Future.delayed(wait);
    }
  }

  void _onMessage(String raw) {
    lastMessage = DateTime.now();
    recent.insert(0, raw.length > 400 ? '${raw.substring(0, 400)}…' : raw);
    if (recent.length > 30) recent.removeLast();
    final Map<String, dynamic> m;
    try {
      m = jsonDecode(raw) as Map<String, dynamic>;
    } catch (_) {
      return;
    }
    switch (m['t']) {
      case 'hello':
        hello = m;
        _set(LinkState.up, '${m['server']} on ${m['host']}');
        _hb?.cancel();
        _hb = Timer.periodic(_hbInterval, (_) => _add('{"t":"hb"}'));
        _armWatchdog();
        for (final c in pending) {
          _send(c);
        }
      case 'hb':
        final ts = m['ts'];
        if (ts is int) {
          clockOffset = DateTime.fromMillisecondsSinceEpoch(ts).difference(DateTime.now());
        }
      case 'state':
        signals = Map<String, dynamic>.from(m['signals'] as Map);
        signals.forEach((k, v) {
          if (v is Map) seen[k] = Map<String, dynamic>.from(v);
        });
        lastState = DateTime.now();
      case 'ack':
        final id = m['id'] as String?;
        pending.removeWhere((c) => c.id == id);
        acks.insert(0, Ack(id ?? '?', m['ok'] == true, '${m['detail'] ?? ''}', DateTime.now()));
        if (acks.length > 10) acks.removeLast();
    }
    notifyListeners();
  }

  /// Silence for timeout_ms means the link is dead, whatever the socket says.
  void _armWatchdog() {
    _watchdog?.cancel();
    _watchdog = Timer.periodic(const Duration(milliseconds: 500), (_) {
      final last = lastMessage;
      if (last != null && DateTime.now().difference(last) > _timeout) {
        detail = 'no message for ${_timeout.inMilliseconds} ms';
        _hb?.cancel();
        _ws?.close();
      }
      notifyListeners();
    });
  }

  void _send(PendingCommand c) {
    if (_ws == null || hello == null) return;
    final args = Map<String, dynamic>.from(c.args);
    if (c.cmd == 'mark') {
      args['delay_ms'] = DateTime.now().difference(c.queuedAt).inMilliseconds;
    }
    _add(jsonEncode({'t': 'cmd', 'id': c.id, 'cmd': c.cmd, 'args': args}));
  }

  /// A send on a link that just died is a no-op: the reconnect resends what matters.
  void _add(String msg) {
    final ws = _ws;
    if (ws == null || ws.closeCode != null) return;
    try {
      ws.add(msg);
    } on StateError {
      // closed between the check and the add
    }
  }

  void _set(LinkState s, String d) {
    state = s;
    detail = d;
    if (s != LinkState.up) signals = {};
    notifyListeners();
  }

  @override
  void dispose() {
    _running = false;
    _hb?.cancel();
    _watchdog?.cancel();
    _ws?.close();
    super.dispose();
  }
}