feat(watch): Android Wear Data Layer adapter + foreground service (#91-D)

lib/infrastructure/watch_bridge/native_watch_bridge_channel.dart

@@ -0,0 +1,137 @@
+import 'dart:async';
+
+import 'package:flutter/services.dart';
+import 'package:watch_bridge_contract/watch_bridge_contract.dart';
+
+final class WatchBridgeConnectionEvent {
+  const WatchBridgeConnectionEvent({
+    required this.isReachable,
+    this.requestsResync = false,
+  });
+
+  final bool isReachable;
+  final bool requestsResync;
+}
+
+abstract interface class WatchBridgeNativeChannel {
+  Stream<WatchCommandEnvelope> get commands;
+
+  Stream<WatchBridgeConnectionEvent> get connectionEvents;
+
+  Future<void> publishProjection(WatchSessionProjection projection);
+
+  Future<void> sendCommandAck(
+    WatchCommandEnvelope command,
+    WatchCommandAck ack, {
+    int? revisionAtAck,
+  });
+
+  Future<void> requestCapabilityRefresh();
+
+  Future<void> startForegroundService();
+
+  Future<void> stopForegroundService();
+}
+
+final class MethodChannelWatchBridgeNativeChannel
+    implements WatchBridgeNativeChannel {
+  const MethodChannelWatchBridgeNativeChannel({
+    MethodChannel methodChannel = const MethodChannel(_methodChannelName),
+    EventChannel commandChannel = const EventChannel(_commandChannelName),
+    EventChannel connectionChannel = const EventChannel(_connectionChannelName),
+  }) : _methodChannel = methodChannel,
+       _commandChannel = commandChannel,
+       _connectionChannel = connectionChannel;
+
+  static const _methodChannelName = 'gametime.watch_bridge/methods';
+  static const _commandChannelName = 'gametime.watch_bridge/commands';
+  static const _connectionChannelName = 'gametime.watch_bridge/connection';
+
+  final MethodChannel _methodChannel;
+  final EventChannel _commandChannel;
+  final EventChannel _connectionChannel;
+
+  @override
+  Stream<WatchCommandEnvelope> get commands {
+    return _commandChannel
+        .receiveBroadcastStream()
+        .where((event) {
+          return event is Map;
+        })
+        .map((event) {
+          return WatchCommandEnvelope.fromJson(_stringObjectMap(event));
+        });
+  }
+
+  @override
+  Stream<WatchBridgeConnectionEvent> get connectionEvents {
+    return _connectionChannel
+        .receiveBroadcastStream()
+        .where((event) {
+          return event is Map;
+        })
+        .map((event) {
+          final json = _stringObjectMap(event);
+          return WatchBridgeConnectionEvent(
+            isReachable: json['isReachable'] == true,
+            requestsResync: json['requestsResync'] == true,
+          );
+        });
+  }
+
+  @override
+  Future<void> publishProjection(WatchSessionProjection projection) {
+    return _invokeIgnoringMissingPlugin(
+      'publishProjection',
+      projection.toJson(),
+    );
+  }
+
+  @override
+  Future<void> requestCapabilityRefresh() {
+    return _invokeIgnoringMissingPlugin('requestCapabilityRefresh');
+  }
+
+  @override
+  Future<void> sendCommandAck(
+    WatchCommandEnvelope command,
+    WatchCommandAck ack, {
+    int? revisionAtAck,
+  }) {
+    return _invokeIgnoringMissingPlugin('sendCommandAck', {
+      'schemaVersion': watchBridgeSchemaVersion,
+      'commandId': command.commandId,
+      'sessionId': command.sessionId,
+      'expectedRevision': command.expectedRevision,
+      'status': ack.name,
+      'revisionAtAck': revisionAtAck,
+      'ackedAtEpochMs': DateTime.now().toUtc().millisecondsSinceEpoch,
+    });
+  }
+
+  @override
+  Future<void> startForegroundService() {
+    return _invokeIgnoringMissingPlugin('startForegroundService');
+  }
+
+  @override
+  Future<void> stopForegroundService() {
+    return _invokeIgnoringMissingPlugin('stopForegroundService');
+  }
+
+  Future<void> _invokeIgnoringMissingPlugin(
+    String method, [
+    Object? arguments,
+  ]) {
+    return _methodChannel
+        .invokeMethod<void>(method, arguments)
+        .onError<MissingPluginException>((_, _) {});
+  }
+}
+
+Map<String, Object?> _stringObjectMap(Object? value) {
+  if (value is Map) {
+    return value.map((key, value) => MapEntry(key.toString(), value));
+  }
+  return const {};
+}

lib/infrastructure/watch_bridge/watch_bridge.dart

@@ -0,0 +1,2 @@
+export 'native_watch_bridge_channel.dart';
+export 'wear_data_layer_adapter.dart';

lib/infrastructure/watch_bridge/wear_data_layer_adapter.dart

@@ -0,0 +1,183 @@
+import 'dart:async';
+
+import 'package:watch_bridge_contract/watch_bridge_contract.dart';
+
+import '../../application/watch_companion_use_cases.dart';
+import 'native_watch_bridge_channel.dart';
+
+final class WatchWearDataLayerAdapter implements WatchProjectionPublisher {
+  WatchWearDataLayerAdapter({
+    required WatchBridgeNativeChannel nativeChannel,
+    required WatchCommandIngress commandIngress,
+    required WatchProjectionSource projectionSource,
+    Duration heartbeatInterval = const Duration(seconds: 5),
+  }) : _nativeChannel = nativeChannel,
+       _commandIngress = commandIngress,
+       _projectionSource = projectionSource,
+       _heartbeatInterval = heartbeatInterval;
+
+  final WatchBridgeNativeChannel _nativeChannel;
+  final WatchCommandIngress _commandIngress;
+  final WatchProjectionSource _projectionSource;
+  final Duration _heartbeatInterval;
+  final _commandAcks = <_WatchAdapterCommandKey, WatchCommandAck>{};
+  final _subscriptions = <StreamSubscription<dynamic>>[];
+  Future<void> _commandTail = Future<void>.value();
+  Timer? _heartbeatTimer;
+  WatchSessionProjection? _latestProjection;
+  bool _started = false;
+  bool _foregroundActive = false;
+
+  Future<void> start() async {
+    if (_started) {
+      return;
+    }
+    _started = true;
+    _subscriptions.add(
+      _projectionSource.projections.listen((projection) {
+        unawaited(publish(projection));
+      }),
+    );
+    _subscriptions.add(
+      _nativeChannel.commands.listen((command) {
+        unawaited(_enqueueCommand(command));
+      }),
+    );
+    _subscriptions.add(
+      _nativeChannel.connectionEvents.listen((event) {
+        if (event.isReachable || event.requestsResync) {
+          unawaited(_projectionSource.emitCurrentProjection());
+        }
+      }),
+    );
+    await _projectionSource.emitCurrentProjection();
+    await _nativeChannel.requestCapabilityRefresh();
+  }
+
+  Future<void> stop() async {
+    _heartbeatTimer?.cancel();
+    _heartbeatTimer = null;
+    for (final subscription in _subscriptions) {
+      await subscription.cancel();
+    }
+    _subscriptions.clear();
+    _started = false;
+  }
+
+  @override
+  Future<void> publish(WatchSessionProjection projection) async {
+    _latestProjection = projection;
+    await _nativeChannel.publishProjection(projection);
+    await _syncForegroundService(projection);
+    _syncHeartbeat(projection);
+  }
+
+  Future<void> _enqueueCommand(WatchCommandEnvelope command) {
+    final run = _commandTail.then(
+      (_) => _handleCommand(command),
+      onError: (_) => _handleCommand(command),
+    );
+    _commandTail = run.then((_) {}, onError: (_) {});
+    return run;
+  }
+
+  Future<void> _handleCommand(WatchCommandEnvelope command) async {
+    final key = _WatchAdapterCommandKey(command);
+    final cachedAck = _commandAcks[key];
+    if (cachedAck != null) {
+      await _sendAck(command, WatchCommandAck.acceptedNoOp);
+      return;
+    }
+    final ack = await _commandIngress.dispatch(command);
+    if (ack == WatchCommandAck.accepted ||
+        ack == WatchCommandAck.acceptedNoOp) {
+      _rememberAck(key, ack);
+    }
+    await _sendAck(command, ack);
+  }
+
+  Future<void> _sendAck(
+    WatchCommandEnvelope command,
+    WatchCommandAck ack,
+  ) async {
+    int? revisionAtAck;
+    try {
+      revisionAtAck = (await _projectionSource.currentProjection()).revision;
+    } on Exception {
+      revisionAtAck = _latestProjection?.revision;
+    }
+    await _nativeChannel.sendCommandAck(
+      command,
+      ack,
+      revisionAtAck: revisionAtAck,
+    );
+  }
+
+  void _rememberAck(_WatchAdapterCommandKey key, WatchCommandAck ack) {
+    _commandAcks[key] = ack;
+    if (_commandAcks.length <= 128) {
+      return;
+    }
+    _commandAcks.remove(_commandAcks.keys.first);
+  }
+
+  Future<void> _syncForegroundService(WatchSessionProjection projection) async {
+    final shouldRun =
+        projection.phase != WatchSessionPhase.noActiveSession &&
+        projection.deviceSessionId.isNotEmpty;
+    if (shouldRun == _foregroundActive) {
+      return;
+    }
+    _foregroundActive = shouldRun;
+    if (shouldRun) {
+      await _nativeChannel.startForegroundService();
+    } else {
+      await _nativeChannel.stopForegroundService();
+    }
+  }
+
+  void _syncHeartbeat(WatchSessionProjection projection) {
+    if (!_hasRunningTimer(projection)) {
+      _heartbeatTimer?.cancel();
+      _heartbeatTimer = null;
+      return;
+    }
+    _heartbeatTimer ??= Timer.periodic(_heartbeatInterval, (_) {
+      unawaited(_projectionSource.emitCurrentProjection());
+    });
+  }
+}
+
+bool _hasRunningTimer(WatchSessionProjection projection) {
+  final timers = [
+    if (projection.dominantTimer != null) projection.dominantTimer!,
+    ...projection.secondaryTimers,
+  ];
+  return timers.any((timer) => timer.runState == WatchTimerRunState.running);
+}
+
+final class _WatchAdapterCommandKey {
+  _WatchAdapterCommandKey(WatchCommandEnvelope command)
+    : sessionId = command.sessionId,
+      expectedRevision = command.expectedRevision,
+      commandId = command.commandId,
+      type = command.type;
+
+  final String sessionId;
+  final int expectedRevision;
+  final String commandId;
+  final WatchCommandType type;
+
+  @override
+  bool operator ==(Object other) {
+    return identical(this, other) ||
+        other is _WatchAdapterCommandKey &&
+            sessionId == other.sessionId &&
+            expectedRevision == other.expectedRevision &&
+            commandId == other.commandId &&
+            type == other.type;
+  }
+
+  @override
+  int get hashCode => Object.hash(sessionId, expectedRevision, commandId, type);
+}