import 'dart:async';

import 'packets/message_common.pb.dart';

/// A raw inbound frame queued for gateway preprocessing.
///
/// [seq] is an internal monotonically increasing sequence assigned by the
/// receive side before submitting to the gateway. It is used only to restore
/// input order after parallel/asynchronous decoding and never touches the wire
/// format or any public protocol field.
class InboundFrame {
  final int seq;
  final List<int> bytes;

  const InboundFrame({required this.seq, required this.bytes});
}

/// Result of decoding a [PacketBase] envelope inside a gateway.
///
/// Carries the same envelope fields that [Communicator.onReceivedData] expects,
/// plus the originating [seq] so results can be reordered before dispatch.
class DecodedFrame {
  final int seq;
  final String typeName;
  final List<int> data;
  final int incomingNonce;
  final int responseNonce;

  const DecodedFrame({
    required this.seq,
    required this.typeName,
    required this.data,
    required this.incomingNonce,
    required this.responseNonce,
  });
}

/// Worker gateway contract.
///
/// A gateway performs only pure raw-bytes work (envelope decode / lightweight
/// preprocessing) off the receive coordinator. It must NOT own stateful
/// dispatch, the pending-request map, listeners, or the outbound write queue —
/// those stay on the main isolate's receive coordinator. Results are emitted on
/// [results] in input [seq] order so the coordinator can dispatch
/// deterministically.
abstract interface class InboundGateway {
  /// Reordered decode results, emitted in input [InboundFrame.seq] order.
  Stream<DecodedFrame> get results;

  /// Prepares the gateway (e.g. spawns a long-running isolate). Idempotent.
  Future<void> start();

  /// Submits a raw frame for decoding. Fire-and-forget; the decoded result is
  /// later emitted on [results].
  void submit(InboundFrame frame);

  /// Releases gateway resources and closes [results].
  Future<void> close();
}

/// Restores input order from results that may arrive out of order.
///
/// Sequences are expected to be contiguous starting at 1. A submitted frame is
/// buffered until every earlier sequence has been released, then it and any
/// now-contiguous successors are returned in order.
class FrameReorderBuffer {
  int _nextSeq = 1;
  final Map<int, DecodedFrame> _pending = {};

  /// Buffers [frame] and returns the frames now releasable in seq order.
  List<DecodedFrame> release(DecodedFrame frame) {
    if (frame.seq < _nextSeq) {
      // Already released (or duplicate); ignore to avoid emitting twice.
      return const [];
    }
    _pending[frame.seq] = frame;
    final ready = <DecodedFrame>[];
    while (_pending.containsKey(_nextSeq)) {
      ready.add(_pending.remove(_nextSeq)!);
      _nextSeq++;
    }
    return ready;
  }
}

/// In-process gateway that decodes synchronously on the calling isolate.
///
/// Used as the web fallback (where worker isolates are unavailable) and as a
/// lightweight default where isolate hand-off cost is not worth it. It still
/// emits results in [seq] order via a [FrameReorderBuffer], so callers see the
/// same ordering contract as the isolate-backed gateway.
class SyncInboundGateway implements InboundGateway {
  final StreamController<DecodedFrame> _controller =
      StreamController<DecodedFrame>.broadcast();
  final FrameReorderBuffer _reorder = FrameReorderBuffer();
  bool _closed = false;

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

  @override
  Future<void> start() async {}

  @override
  void submit(InboundFrame frame) {
    if (_closed) return;
    final common = PacketBase.fromBuffer(frame.bytes);
    final decoded = DecodedFrame(
      seq: frame.seq,
      typeName: common.typeName,
      data: common.data,
      incomingNonce: common.nonce,
      responseNonce: common.responseNonce,
    );
    for (final ready in _reorder.release(decoded)) {
      if (!_controller.isClosed) _controller.add(ready);
    }
  }

  @override
  Future<void> close() async {
    if (_closed) return;
    _closed = true;
    await _controller.close();
  }
}
