// 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 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? hello; Map 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> seen = {}; DateTime? lastState; DateTime? lastMessage; /// Orin clock minus tablet clock, from heartbeats. Large values mean one side is unsynced. Duration? clockOffset; final List recent = []; final List pending = []; final List 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 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 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 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 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 _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 m; try { m = jsonDecode(raw) as Map; } 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.from(m['signals'] as Map); signals.forEach((k, v) { if (v is Map) seen[k] = Map.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.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(); } }