PilotVault/Fly App/backup_pocketbase_app/pb_service.dart

import 'dart:async';

import 'package:pocketbase/pocketbase.dart';
import 'package:shared_preferences/shared_preferences.dart';

enum BackendStatus { signedOut, signingIn, online, error }

/// PocketBase-backed service: authentication, telemetry persistence, device
/// presence, and inbound command subscription. A single global [pb] instance is
/// shared across the app.
class PbService {
  PocketBase? _pb;
  SharedPreferences? _prefs;
  String _deviceId = 'phone';

  Timer? _heartbeat;
  String? _devicesRecordId;
  UnsubscribeFunc? _cmdUnsub;
  DateTime _lastTelemetry = DateTime.fromMillisecondsSinceEpoch(0);

  /// Called when a command record targeting this device is created.
  void Function(String command, Map<String, dynamic> payload)? onCommand;

  /// Returns the app's current state for the device presence record.
  /// Expected keys: connected (bool), model (String), registration (String).
  Map<String, dynamic> Function()? stateProvider;

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

  BackendStatus _status = BackendStatus.signedOut;
  BackendStatus get currentStatus => _status;

  bool get isAuthed => _pb?.authStore.isValid ?? false;
  String get userEmail => _pb?.authStore.record?.getStringValue('email') ?? '';
  String get deviceId => _deviceId;
  String get baseUrl => _pb?.baseURL ?? '';

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

  /// Restores a persisted session (if any) on app start.
  Future<void> init() async {
    _prefs = await SharedPreferences.getInstance();
    _deviceId = _prefs!.getString('pb_device') ?? 'phone';
    final String? url = _prefs!.getString('pb_url');
    final String? auth = _prefs!.getString('pb_auth');
    if (url != null && auth != null && auth.isNotEmpty) {
      _pb = _build(url, auth);
      if (_pb!.authStore.isValid) {
        _setStatus(BackendStatus.online);
        await _afterAuth();
      }
    }
  }

  PocketBase _build(String url, String? initialAuth) {
    final AsyncAuthStore store = AsyncAuthStore(
      save: (String data) async => _prefs?.setString('pb_auth', data),
      clear: () async => _prefs?.remove('pb_auth'),
      initial: initialAuth,
    );
    return PocketBase(url, authStore: store);
  }

  String get defaultBaseUrl => _prefs?.getString('pb_url') ?? 'http://10.2.1.101:8090';

  Future<void> signIn({
    required String baseUrl,
    required String email,
    required String password,
    String deviceId = 'phone',
  }) async {
    _deviceId = deviceId.trim().isEmpty ? 'phone' : deviceId.trim();
    final String url = _normalize(baseUrl);
    _setStatus(BackendStatus.signingIn);
    try {
      final PocketBase pb = _build(url, null);
      await pb.collection('users').authWithPassword(email.trim(), password);
      _pb = pb;
      await _prefs?.setString('pb_url', url);
      await _prefs?.setString('pb_device', _deviceId);
      _setStatus(BackendStatus.online);
      await _afterAuth();
    } catch (e) {
      _setStatus(BackendStatus.error);
      rethrow;
    }
  }

  Future<void> signOut() async {
    _heartbeat?.cancel();
    _heartbeat = null;
    try {
      await _upsertDevice(online: false);
    } catch (_) {}
    try {
      await _cmdUnsub?.call();
    } catch (_) {}
    _cmdUnsub = null;
    _devicesRecordId = null;
    _pb?.authStore.clear();
    _setStatus(BackendStatus.signedOut);
  }

  Future<void> _afterAuth() async {
    await _upsertDevice(online: true);
    _startHeartbeat();
    await _subscribeCommands();
  }

  String _normalize(String url) {
    String u = url.trim();
    if (!u.startsWith('http://') && !u.startsWith('https://')) u = 'http://$u';
    return u;
  }

  // ── Telemetry ingestion ────────────────────────────────────────────────────

  /// Feed an event from `DjiService.events()`.
  void onEvent(Map<String, dynamic> event) {
    final PocketBase? pb = _pb;
    if (pb == null || !pb.authStore.isValid) return;
    final String? type = event['type'] as String?;

    if (type == 'telemetry') {
      final DateTime now = DateTime.now();
      if (now.difference(_lastTelemetry) < const Duration(milliseconds: 500)) return;
      _lastTelemetry = now;
    }

    _appendEvent(type ?? 'event', event);

    // Reflect registration/connection promptly in the presence record.
    if (type == 'registration' || type == 'connection') {
      _upsertDevice(online: true);
    }
  }

  Future<void> _appendEvent(String kind, Map<String, dynamic> event) async {
    final PocketBase? pb = _pb;
    if (pb == null) return;
    try {
      await pb.collection('telemetry').create(body: <String, dynamic>{
        'device': _deviceId,
        'kind': kind,
        'payload': event,
        'owner': pb.authStore.record?.id,
      });
    } catch (_) {
      // best-effort; drop on transient failure
    }
  }

  // ── Device presence ─────────────────────────────────────────────────────────

  void _startHeartbeat() {
    _heartbeat?.cancel();
    _heartbeat = Timer.periodic(const Duration(seconds: 6), (_) => _upsertDevice(online: true));
  }

  Future<void> _upsertDevice({required bool online}) async {
    final PocketBase? pb = _pb;
    if (pb == null) return;
    final Map<String, dynamic> st = stateProvider?.call() ?? <String, dynamic>{};
    final Map<String, dynamic> body = <String, dynamic>{
      'device': _deviceId,
      'online': online,
      'connected': st['connected'] ?? false,
      'model': st['model'] ?? '',
      'registration': st['registration'] ?? '',
      'lastSeen': DateTime.now().millisecondsSinceEpoch ~/ 1000,
      'owner': pb.authStore.record?.id,
    };
    try {
      if (_devicesRecordId == null) {
        try {
          final RecordModel existing =
              await pb.collection('devices').getFirstListItem('device="$_deviceId"');
          _devicesRecordId = existing.id;
        } catch (_) {
          // none yet
        }
      }
      if (_devicesRecordId == null) {
        final RecordModel rec = await pb.collection('devices').create(body: body);
        _devicesRecordId = rec.id;
      } else {
        await pb.collection('devices').update(_devicesRecordId!, body: body);
      }
    } catch (_) {
      _devicesRecordId = null; // record may have been removed; recreate next tick
    }
  }

  // ── Commands ────────────────────────────────────────────────────────────────

  Future<void> _subscribeCommands() async {
    final PocketBase? pb = _pb;
    if (pb == null) return;
    try {
      await _cmdUnsub?.call();
    } catch (_) {}
    _cmdUnsub = await pb.collection('commands').subscribe(
      '*',
      (RecordSubscriptionEvent e) {
        if (e.action != 'create') return;
        final RecordModel? rec = e.record;
        if (rec == null) return;
        if (rec.getStringValue('device') != _deviceId) return;
        final String cmd = rec.getStringValue('command');
        final Map<String, dynamic> payload =
            (rec.get<Map<String, dynamic>?>('payload') ?? <String, dynamic>{});
        if (cmd.isNotEmpty) onCommand?.call(cmd, payload);
      },
      filter: 'device="$_deviceId"',
    );
  }

  void dispose() {
    _heartbeat?.cancel();
    _statusController.close();
  }
}

/// App-wide PocketBase service instance.
final PbService pb = PbService();