import 'dart:async';
import 'dart:isolate';

import 'inbound_gateway.dart';
import 'packets/message_common.pb.dart';

/// Long-running isolate worker gateway for Dart IO targets.
///
/// A single isolate is spawned once via [start] and kept alive to avoid
/// repeating spawn cost per frame. Raw frames are sent over a [SendPort] with
/// their [InboundFrame.seq]; the isolate decodes the [PacketBase] envelope
/// (pure work only) and sends the fields back. Results are reordered by [seq]
/// and emitted on [results] for the main-isolate coordinator to dispatch.
///
/// The gateway deliberately owns no stateful dispatch: listeners, request
/// handlers, the pending-request map, and the outbound write queue all remain
/// on the main isolate's receive coordinator.
class IsolateInboundGateway implements InboundGateway {
  final StreamController<DecodedFrame> _controller =
      StreamController<DecodedFrame>.broadcast();
  final FrameReorderBuffer _reorder = FrameReorderBuffer();

  Isolate? _isolate;
  SendPort? _toWorker;
  ReceivePort? _fromWorker;
  Completer<void>? _ready;
  bool _closed = false;

  @override
  Stream<DecodedFrame> get results => _controller.stream;

  @override
  Future<void> start() async {
    if (_isolate != null || _closed) return;
    final ready = Completer<void>();
    _ready = ready;
    final fromWorker = ReceivePort();
    _fromWorker = fromWorker;
    fromWorker.listen(_onWorkerMessage);
    _isolate = await Isolate.spawn(_gatewayEntry, fromWorker.sendPort);
    await ready.future;
  }

  void _onWorkerMessage(dynamic message) {
    if (message is SendPort) {
      _toWorker = message;
      _ready?.complete();
      _ready = null;
      return;
    }
    if (message is List) {
      final decoded = DecodedFrame(
        seq: message[0] as int,
        typeName: message[1] as String,
        data: (message[2] as List).cast<int>(),
        incomingNonce: message[3] as int,
        responseNonce: message[4] as int,
      );
      for (final ready in _reorder.release(decoded)) {
        if (!_controller.isClosed) _controller.add(ready);
      }
    }
  }

  @override
  void submit(InboundFrame frame) {
    if (_closed) return;
    _toWorker?.send([frame.seq, frame.bytes]);
  }

  @override
  Future<void> close() async {
    if (_closed) return;
    _closed = true;
    _toWorker = null;
    _isolate?.kill(priority: Isolate.immediate);
    _isolate = null;
    _fromWorker?.close();
    _fromWorker = null;
    await _controller.close();
  }
}

/// Isolate entry point: decodes [PacketBase] envelopes and returns fields.
///
/// Runs only pure decode work. The first message back is the worker's
/// [SendPort]; subsequent messages are `[seq, bytes]` decode requests.
void _gatewayEntry(SendPort toMain) {
  final fromMain = ReceivePort();
  toMain.send(fromMain.sendPort);
  fromMain.listen((dynamic message) {
    if (message == null) {
      fromMain.close();
      return;
    }
    final request = message as List;
    final seq = request[0] as int;
    final bytes = (request[1] as List).cast<int>();
    final common = PacketBase.fromBuffer(bytes);
    toMain.send([
      seq,
      common.typeName,
      common.data,
      common.nonce,
      common.responseNonce,
    ]);
  });
}
