PilotVault/Fly App/lib/uploader.dart

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

enum UploadStatus { disabled, connecting, connected, error }

/// Streams DJI events to the API Server over a WebSocket and dispatches
/// commands the server pushes back. Auto-reconnects while enabled.
///
/// Wire format matches the server's `/ws/device` endpoint:
///  * outbound: the raw DJI event maps (registration/connection/battery/telemetry)
///  * inbound : `{"type":"command","command":"...","payload":{...}}`
class ServerUploader {
  ServerUploader({required this.onCommand, this.snapshotProvider});

  /// Invoked when the server pushes a command frame.
  final void Function(String command, Map<String, dynamic> payload) onCommand;

  /// Returns the app's last-known events (registration/connection/battery/
  /// telemetry) to replay right after (re)connecting, so the server reflects
  /// current state even though those events fired in the past.
  final List<Map<String, dynamic>> Function()? snapshotProvider;

  /// High-frequency telemetry is throttled to this interval; other event
  /// types (battery, connection, registration) are sent immediately.
  Duration telemetryInterval = const Duration(milliseconds: 250);

  WebSocket? _socket;
  bool _enabled = false;
  String _host = '';
  String _deviceId = 'phone';
  Timer? _reconnectTimer;
  DateTime _lastTelemetrySent = DateTime.fromMillisecondsSinceEpoch(0);

  final StreamController<UploadStatus> _statusController =
      StreamController<UploadStatus>.broadcast();
  Stream<UploadStatus> get status => _statusController.stream;

  UploadStatus _status = UploadStatus.disabled;
  UploadStatus get currentStatus => _status;

  void _setStatus(UploadStatus s) {
    _status = s;
    if (!_statusController.isClosed) _statusController.add(s);
  }

  /// host accepts "10.0.0.5", "10.0.0.5:8080", or "ws://10.0.0.5:8080".
  Future<void> enable(String host, {String deviceId = 'phone'}) async {
    _host = host.trim();
    _deviceId = deviceId.trim().isEmpty ? 'phone' : deviceId.trim();
    _enabled = true;
    await _connect();
  }

  Future<void> disable() async {
    _enabled = false;
    _reconnectTimer?.cancel();
    await _socket?.close();
    _socket = null;
    _setStatus(UploadStatus.disabled);
  }

  Future<void> _connect() async {
    if (!_enabled) return;
    _setStatus(UploadStatus.connecting);
    try {
      final Uri uri = _buildUri(_host, _deviceId);
      final WebSocket sock =
          await WebSocket.connect(uri.toString()).timeout(const Duration(seconds: 6));
      _socket = sock;
      _setStatus(UploadStatus.connected);
      // Replay last-known state so the server is immediately in sync.
      final List<Map<String, dynamic>> snapshot = snapshotProvider?.call() ?? const <Map<String, dynamic>>[];
      for (final Map<String, dynamic> e in snapshot) {
        try {
          sock.add(jsonEncode(e));
        } catch (_) {}
      }
      sock.listen(
        _onData,
        onDone: _onDisconnect,
        onError: (Object _) => _onDisconnect(),
        cancelOnError: true,
      );
    } catch (_) {
      _socket = null;
      _setStatus(UploadStatus.error);
      _scheduleReconnect();
    }
  }

  Uri _buildUri(String host, String id) {
    String h = host;
    if (h.startsWith('ws://')) h = h.substring(5);
    if (h.startsWith('wss://')) h = h.substring(6);
    final int slash = h.indexOf('/');
    if (slash >= 0) h = h.substring(0, slash);
    if (!h.contains(':')) h = '$h:8080';
    return Uri.parse('ws://$h/ws/device?id=${Uri.encodeQueryComponent(id)}');
  }

  void _onData(dynamic data) {
    if (data is! String) return;
    try {
      final Map<String, dynamic> msg =
          (jsonDecode(data) as Map).cast<String, dynamic>();
      if (msg['type'] == 'command') {
        final String? cmd = msg['command'] as String?;
        final Map<String, dynamic> payload =
            (msg['payload'] as Map?)?.cast<String, dynamic>() ?? <String, dynamic>{};
        if (cmd != null) onCommand(cmd, payload);
      }
    } catch (_) {
      // ignore malformed frames
    }
  }

  void _onDisconnect() {
    _socket = null;
    if (_enabled) {
      _setStatus(UploadStatus.error);
      _scheduleReconnect();
    }
  }

  void _scheduleReconnect() {
    _reconnectTimer?.cancel();
    if (!_enabled) return;
    _reconnectTimer = Timer(const Duration(seconds: 3), _connect);
  }

  /// Feed an event coming from `DjiService.events()`.
  void onEvent(Map<String, dynamic> event) {
    final WebSocket? sock = _socket;
    if (sock == null || _status != UploadStatus.connected) return;

    if (event['type'] == 'telemetry') {
      final DateTime now = DateTime.now();
      if (now.difference(_lastTelemetrySent) < telemetryInterval) return;
      _lastTelemetrySent = now;
    }
    try {
      sock.add(jsonEncode(event));
    } catch (_) {
      // drop on transient write failure; reconnect logic handles the rest
    }
  }

  void dispose() {
    _enabled = false;
    _reconnectTimer?.cancel();
    _socket?.close();
    _statusController.close();
  }
}