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.
///
/// A decode failure is also a result: [DecodedFrame.error] produces a frame
/// with the same [seq] but [error] set. Emitting it through the ordered
/// [InboundGateway.results] stream keeps the reorder buffer from stalling on a
/// missing seq and lets the coordinator tear the connection down deterministically.
class DecodedFrame {
  final int seq;
  final String typeName;
  final List<int> data;
  final int incomingNonce;
  final int responseNonce;

  /// Non-null when decoding this [seq] failed. Success frames leave it null.
  final Object? error;

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

  /// An ordered decode failure for [seq]. Occupies its seq slot so successors
  /// still release, but carries no envelope payload.
  const DecodedFrame.error({
    required this.seq,
    required this.error,
  })  : typeName = '',
        data = const [],
        incomingNonce = 0,
        responseNonce = 0;

  /// Whether this frame represents a decode failure rather than a payload.
  bool get isError => error != null;
}

/// 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;

  /// Frames submitted to the worker whose result has not yet come back.
  ///
  /// Bounded by the gateway's capacity; used as backlog/cleanup evidence.
  int get inFlightCount;

  /// Frames already decoded but buffered in the reorder buffer waiting for an
  /// earlier seq. Drains to zero once the contiguous run is released.
  int get queuedCount;

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

  /// Submits a raw frame for decoding.
  ///
  /// Awaits worker readiness before handing off, so a frame is never silently
  /// dropped while the isolate is still spawning, and applies backpressure when
  /// [inFlightCount] reaches capacity: the future does not complete until an
  /// earlier result releases a slot. The decoded result (or an ordered
  /// [DecodedFrame.error]) is later emitted on [results].
  Future<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 = {};

  /// Frames buffered out of order, still waiting for an earlier seq.
  int get pendingCount => _pending.length;

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

  /// Drops all buffered frames and resets the expected sequence to its initial
  /// value, so a reused gateway leaves no residual backlog after close.
  void clear() {
    _pending.clear();
    _nextSeq = 1;
  }
}

/// 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;

  // Synchronous decode never leaves work outstanding across an await.
  @override
  int get inFlightCount => 0;

  @override
  int get queuedCount => _reorder.pendingCount;

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

  @override
  Future<void> submit(InboundFrame frame) async {
    if (_closed) return;
    DecodedFrame decoded;
    try {
      final common = PacketBase.fromBuffer(frame.bytes);
      decoded = DecodedFrame(
        seq: frame.seq,
        typeName: common.typeName,
        data: common.data,
        incomingNonce: common.nonce,
        responseNonce: common.responseNonce,
      );
    } catch (error) {
      // A malformed envelope still occupies its seq so successors release and
      // the coordinator can tear the connection down in order.
      decoded = DecodedFrame.error(seq: frame.seq, error: error);
    }
    for (final ready in _reorder.release(decoded)) {
      if (!_controller.isClosed) _controller.add(ready);
    }
  }

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