GameTime/lib/infrastructure/watch_bridge/wear_data_layer_adapter.dart

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);
}