// ignore_for_file: prefer_final_fields

import 'dart:async';
import 'package:meta/meta.dart';
import 'package:protobuf/protobuf.dart';
import 'inbound_gateway.dart';
import 'packets/message_common.pb.dart';
import 'transport.dart';

/// A single frame queued for inbound dispatch.
class _InboundItem {
  final String typeName;
  final List<int> data;
  final int incomingNonce;
  final int responseNonce;

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

abstract class Communicator {
  static const int maxNonce = 2147483647;

  Map<String, IDataHandler> _handlerDic = {};
  Map<String, Future<void> Function(String, List<int>, int)>
      _requestHandlerDic = {};
  Map<int, _PendingRequest> _pendingRequests = {};
  late Map<String, GeneratedMessage Function(List<int>)> _instanceGenerator;
  Map<String, String> _canonicalNameMap = {};
  Future<void> _outboundWrite = Future.value();
  Transport? _transport;

  /// Inbound queue: serial receive worker chained via Future.
  Future<void> _inboundDispatch = Future.value();
  static const int _inboundQueueCapacity = 64;
  int _inboundQueueLength = 0;
  bool _isClosing = false;
  Completer<void>? _queueWaiter;

  /// Optional worker gateway. When attached, raw frames are decoded off the
  /// coordinator and reordered by [seq] before reaching [onReceivedData].
  /// Defaults to null: frames decode inline on the main isolate.
  InboundGateway? _inboundGateway;
  StreamSubscription<DecodedFrame>? _gatewaySubscription;
  int _frameSeq = 0;

  /// Frames handed to the gateway but not yet accepted into the dispatch queue.
  ///
  /// Decode is never the bottleneck, so the gateway's own in-flight capacity
  /// cannot throttle a stalled dispatch handler. Gating [onReceivedFrame] on
  /// this backlog makes a full inbound queue propagate to the transport read
  /// loop and keeps decoded frames from piling up unbounded behind a stuck
  /// handler — the same bound the inline path gets for free.
  int _gatewayBacklog = 0;
  Completer<void>? _gatewayBacklogWaiter;

  /// Monotonically increasing nonce. Shared by send / sendRequest / response.
  int _nonce = 0;
  int get nonce => _nonce;

  @protected
  set nonce(int value) => _nonce = value;

  @protected
  int nextNonce() {
    for (var i = 0; i < maxNonce; i++) {
      if (_nonce >= maxNonce) {
        _nonce = 0;
      }
      _nonce += 1;
      if (!_pendingRequests.containsKey(_nonce)) {
        return _nonce;
      }
    }
    throw StateError('no available nonce: all positive int32 nonces are pending');
  }

  /// Whether the connection is alive. Set by each transport implementation.
  bool _isAlive = false;
  bool get isAlive => _isAlive;

  @protected
  set isAlive(bool value) => _isAlive = value;

  Communicator();

  void initialize(
      Map<String, GeneratedMessage Function(List<int>)> instanceGenerator,
      {required Transport transport}) {
    final shortToFull = <String, String>{};
    for (final fullName in instanceGenerator.keys) {
      final shortName = _shortTypeName(fullName);
      if (shortToFull.containsKey(shortName) && shortToFull[shortName] != fullName) {
        throw ArgumentError('Duplicate alias mapping: $shortName maps to both ${shortToFull[shortName]} and $fullName');
      }
      shortToFull[shortName] = fullName;
    }

    _canonicalNameMap = {};
    for (final fullName in instanceGenerator.keys) {
      _canonicalNameMap[fullName] = fullName;
      final shortName = _shortTypeName(fullName);
      _canonicalNameMap[shortName] = fullName;
    }

    _instanceGenerator = instanceGenerator;
    _transport = transport;
  }

  String _canonicalize(String typeName) {
    return _canonicalNameMap[typeName] ??
        _canonicalNameMap[_shortTypeName(typeName)] ??
        typeName;
  }

  T Function(List<int>) getGenerator<T extends GeneratedMessage>(String type) {
    final canonical = _canonicalize(type);
    final generator = _instanceGenerator[canonical];
    if (generator == null) {
      throw Exception(
          'Must set protobuf packet creator before use it. Type: ${_qualifiedMessageNameOf<T>()}');
    }
    return generator as T Function(List<int>);
  }

  String _qualifiedMessageNameOf<T extends GeneratedMessage>() {
    final shortName = T.toString();
    for (final key in _instanceGenerator.keys) {
      if (key == shortName || _shortTypeName(key) == shortName) {
        if (key.contains('.')) {
          return key;
        }
      }
    }
    for (final key in _instanceGenerator.keys) {
      if (key == shortName || _shortTypeName(key) == shortName) {
        return key;
      }
    }
    return shortName;
  }

  String _shortTypeName(String typeName) {
    final lastDot = typeName.lastIndexOf('.');
    if (lastDot < 0 || lastDot == typeName.length - 1) {
      return typeName;
    }
    return typeName.substring(lastDot + 1);
  }

  V? _lookupByWireType<V>(Map<String, V> map, String typeName) {
    return map[_canonicalize(typeName)];
  }

  GeneratedMessage _decodeMessage(
    String typeName,
    List<int> data, {
    String? fallbackTypeName,
  }) {
    final canonical = _canonicalize(typeName);
    var generator = _instanceGenerator[canonical];
    if (generator == null && fallbackTypeName != null) {
      generator = _instanceGenerator[_canonicalize(fallbackTypeName)];
    }
    if (generator == null) {
      throw Exception(
          'Must set protobuf packet creator before use it. Type: $typeName');
    }
    return generator(data);
  }

  /// Serializes writes so stream transports do not interleave packets.
  Future<void> queuePacket(PacketBase base) {
    final write = _outboundWrite.then((_) => _transport!.writePacket(base));
    _outboundWrite = write.catchError((_) {});
    return write;
  }

  Future<void> send<T extends GeneratedMessage>(T data) async {
    if (isAlive) {
      await queuePacket(PacketBase()
        ..typeName = data.info_.qualifiedMessageName
        ..nonce = nextNonce()
        ..data = data.writeToBuffer());
    }
  }

  @protected
  void cancelPendingRequests() {
    final snapshot = Map<int, _PendingRequest>.from(_pendingRequests);
    _pendingRequests.clear();
    for (final pending in snapshot.values) {
      pending.completeError(
          StateError('connection closed'), StackTrace.current);
    }
  }

  /// Sends [data] as a request and waits for a typed response.
  ///
  /// The remote side must have registered an [addRequestListener] for [Req].
  /// [Res] must be registered in the parser map.
  ///
  /// ```dart
  /// final res = await client.sendRequest<GetUser, UserData>(GetUser()..id = 1);
  /// ```
  Future<Res>
      sendRequest<Req extends GeneratedMessage, Res extends GeneratedMessage>(
          Req data,
          {Duration timeout = const Duration(seconds: 30)}) async {
    if (!isAlive) return Future.error(StateError('not connected'));
    final requestNonce = nextNonce();
    final completer = Completer<Res>();
    final expectedResponseType = _qualifiedMessageNameOf<Res>();
    _pendingRequests[requestNonce] = _PendingRequest(
      expectedTypeName: expectedResponseType,
      complete: (typeName, bytes) {
        final canonicalExpected = _canonicalize(expectedResponseType);
        final canonicalReceived = _canonicalize(typeName);
        if (canonicalReceived != canonicalExpected) {
          completer.completeError(
            StateError(
                'Response type mismatch for nonce $requestNonce: expected $canonicalExpected, got $typeName'),
            StackTrace.current,
          );
          return;
        }
        final message = _decodeMessage(typeName, bytes,
            fallbackTypeName: expectedResponseType);
        if (message is! Res) {
          completer.completeError(
            StateError(
                'Response type mismatch for nonce $requestNonce: expected $canonicalExpected, got $typeName'),
            StackTrace.current,
          );
          return;
        }
        completer.complete(message);
      },
      completeError: (error, stackTrace) {
        completer.completeError(error, stackTrace);
      },
    );
    await queuePacket(PacketBase()
          ..typeName = data.info_.qualifiedMessageName
          ..nonce = requestNonce
          ..data = data.writeToBuffer())
        .catchError((error, stackTrace) {
      _pendingRequests.remove(requestNonce);
      completer.completeError(error, stackTrace);
    });
    return completer.future.timeout(timeout, onTimeout: () {
      _pendingRequests.remove(requestNonce);
      throw TimeoutException(
          'sendRequest timeout for nonce $requestNonce', timeout);
    });
  }

  /// Registers a request handler for [Req] that returns [Res].
  ///
  /// When a [Req] packet arrives the handler is called and the returned [Res]
  /// is automatically sent back to the caller.
  ///
  /// ```dart
  /// server.addRequestListener<GetUser, UserData>((req) async {
  ///   return UserData()..name = db.getUser(req.id).name;
  /// });
  /// ```
  void addRequestListener<Req extends GeneratedMessage,
      Res extends GeneratedMessage>(Future<Res> Function(Req) handler) {
    final reqType = _qualifiedMessageNameOf<Req>();
    final canonicalReq = _canonicalize(reqType);
    if (_handlerDic.containsKey(canonicalReq)) {
      throw StateError(
          'Type $reqType is already registered with addListener and cannot also use addRequestListener.');
    }
    _requestHandlerDic[canonicalReq] =
        (String typeName, List<int> bytes, int requestNonce) async {
      final message =
          _decodeMessage(typeName, bytes, fallbackTypeName: reqType);
      if (message is! Req) {
        throw StateError(
            'Request type mismatch for nonce $requestNonce: expected ${_canonicalize(reqType)}, got $typeName');
      }
      final req = message;
      final res = await handler(req);
      if (isAlive) {
        await queuePacket(PacketBase()
          ..typeName = res.info_.qualifiedMessageName
          ..nonce = nextNonce()
          ..responseNonce = requestNonce
          ..data = res.writeToBuffer());
      }
    };
  }

  /// Attaches a worker [gateway] in front of the receive coordinator.
  ///
  /// Decoded results are dispatched in input [seq] order, so the coordinator
  /// keeps its FIFO dispatch and sole ownership of stateful handling. Each
  /// result releases the backlog slot its frame reserved in [onReceivedFrame],
  /// only once the frame is actually handled. Attaching is opt-in; without it
  /// [onReceivedFrame] decodes inline.
  @protected
  void attachInboundGateway(InboundGateway gateway) {
    _inboundGateway = gateway;
    _gatewaySubscription = gateway.results.listen(_onGatewayResult);
  }

  /// Consumes one ordered gateway result and frees its reserved backlog slot
  /// once handled — not merely decoded — so a stalled handler keeps the
  /// transport read loop paused via [onReceivedFrame].
  void _onGatewayResult(DecodedFrame frame) {
    if (frame.isError) {
      onGatewayDecodeError(frame.seq, frame.error!);
      _releaseGatewayBacklog();
      return;
    }
    if (frame.responseNonce > 0) {
      // Responses complete out of band, exactly as the inline coordinator does.
      final pending = _pendingRequests.remove(frame.responseNonce);
      pending?.complete(frame.typeName, frame.data);
      _releaseGatewayBacklog();
      return;
    }
    // Data/request frames serialize on the dispatch chain. The reserved backlog
    // slot is held until dispatch completes, so the chain depth — and thus the
    // outstanding frame bound — matches the inline coordinator's queue.
    final item = _InboundItem(
      typeName: frame.typeName,
      data: frame.data,
      incomingNonce: frame.incomingNonce,
      responseNonce: frame.responseNonce,
    );
    _inboundDispatch = _inboundDispatch
        .then((_) => _dispatchInbound(item))
        .whenComplete(_releaseGatewayBacklog);
  }

  /// Frees one backlog slot and wakes a waiting [onReceivedFrame] if the gateway
  /// has fallen back under capacity.
  void _releaseGatewayBacklog() {
    if (_gatewayBacklog > 0) _gatewayBacklog--;
    final waiter = _gatewayBacklogWaiter;
    if (waiter != null && _gatewayBacklog < _inboundQueueCapacity) {
      _gatewayBacklogWaiter = null;
      waiter.complete();
    }
  }

  /// Handles an ordered decode failure surfaced by the gateway.
  ///
  /// A malformed envelope means the inbound framing can no longer be trusted, so
  /// the connection is torn down and pending requests are cleaned up. Overrides
  /// (e.g. [BaseClient]) close the transport as part of [close].
  @protected
  void onGatewayDecodeError(int seq, Object error) {
    if (!isAlive || _isClosing) return;
    unawaited(close());
  }

  int _nextFrameSeq() {
    _frameSeq += 1;
    return _frameSeq;
  }

  /// Receives a raw [PacketBase] frame from a transport.
  ///
  /// When a gateway is attached the frame is submitted with an internal [seq]
  /// for off-coordinator decode + reorder; otherwise it is decoded inline and
  /// forwarded to [onReceivedData].
  @protected
  Future<void> onReceivedFrame(List<int> frame) async {
    if (!isAlive || _isClosing) return;
    final gateway = _inboundGateway;
    if (gateway != null) {
      // Throttle on the dispatch backlog before handing the frame off. This
      // propagates a full inbound queue to the transport read loop even though
      // the gateway's in-flight capacity (decode) is never the bottleneck.
      while (isAlive && !_isClosing && _gatewayBacklog >= _inboundQueueCapacity) {
        _gatewayBacklogWaiter ??= Completer<void>();
        await _gatewayBacklogWaiter!.future;
      }
      if (!isAlive || _isClosing) return;
      _gatewayBacklog++;
      // Await so gateway backpressure propagates to the transport read loop.
      await gateway.submit(InboundFrame(seq: _nextFrameSeq(), bytes: frame));
      return;
    }
    final common = PacketBase.fromBuffer(frame);
    await onReceivedData(common.typeName, common.data,
        incomingNonce: common.nonce, responseNonce: common.responseNonce);
  }

  /// Blocks until the inbound dispatch queue has room. Shared by the inline and
  /// gateway receive paths so a stalled handler throttles the transport equally.
  Future<void> _awaitInboundCapacity() async {
    while (_inboundQueueLength >= _inboundQueueCapacity) {
      _queueWaiter ??= Completer<void>();
      await _queueWaiter!.future;
    }
  }

  /// Enqueues an inbound packet for serial dispatch.
  ///
  /// The receive worker processes items one at a time, preserving FIFO order
  /// even when a request handler is async.
  Future<void> onReceivedData(String typeName, List<int> data,
      {int incomingNonce = 0, int responseNonce = 0}) async {
    if (!isAlive || _isClosing) return;
    if (responseNonce > 0) {
      final pending = _pendingRequests.remove(responseNonce);
      if (pending != null) {
        pending.complete(typeName, data);
      }
      return;
    }
    await _awaitInboundCapacity();
    if (!isAlive || _isClosing) return;
    _inboundQueueLength++;
    final item = _InboundItem(
      typeName: typeName,
      data: data,
      incomingNonce: incomingNonce,
      responseNonce: responseNonce,
    );
    // Chain dispatch onto the previous future to preserve order.
    _inboundDispatch = _inboundDispatch
        .then((_) => _dispatchInbound(item))
        .whenComplete(() {
      _inboundQueueLength--;
      if (_inboundQueueLength < _inboundQueueCapacity && _queueWaiter != null) {
        final w = _queueWaiter;
        _queueWaiter = null;
        w!.complete();
      }
    });
  }

  Future<void> close() async {
    if (!isAlive || _isClosing) return;
    _isClosing = true;
    await _inboundDispatch;
    isAlive = false;
    _isClosing = false;
    if (_queueWaiter != null) {
      _queueWaiter!.complete();
      _queueWaiter = null;
    }
    _drainGatewayBacklog();
    await _closeInboundGateway();
    cancelPendingRequests();
  }

  void shutdown() {
    if (!isAlive) return;
    isAlive = false;
    _isClosing = false;
    _inboundQueueLength = 0;
    _inboundDispatch = Future.value();
    if (_queueWaiter != null) {
      _queueWaiter!.complete();
      _queueWaiter = null;
    }
    _drainGatewayBacklog();
    unawaited(_closeInboundGateway());
    cancelPendingRequests();
  }

  /// Releases any [onReceivedFrame] blocked on backlog and clears the counter so
  /// teardown never strands the transport read loop.
  void _drainGatewayBacklog() {
    _gatewayBacklog = 0;
    final waiter = _gatewayBacklogWaiter;
    if (waiter != null) {
      _gatewayBacklogWaiter = null;
      waiter.complete();
    }
  }

  Future<void> _closeInboundGateway() async {
    final subscription = _gatewaySubscription;
    final gateway = _inboundGateway;
    _gatewaySubscription = null;
    _inboundGateway = null;
    await subscription?.cancel();
    await gateway?.close();
  }

  /// Dispatches a single inbound item sequentially.
  Future<void> _dispatchInbound(_InboundItem item) async {
    if (item.responseNonce > 0) {
      final pending = _pendingRequests.remove(item.responseNonce);
      if (pending == null) return;
      pending.complete(item.typeName, item.data);
      return;
    }
    final canonical = _canonicalize(item.typeName);
    final requestHandler = _requestHandlerDic[canonical];
    if (requestHandler != null) {
      await requestHandler(item.typeName, item.data, item.incomingNonce);
      return;
    }
    final handler = _handlerDic[canonical];
    if (handler != null) {
      handler.onMessage(item.typeName, item.data);
    }
  }

  void addListener<T extends GeneratedMessage>(void Function(T) listener) {
    var type = _qualifiedMessageNameOf<T>();
    var canonical = _canonicalize(type);
    if (_requestHandlerDic.containsKey(canonical)) {
      throw StateError(
          'Type $type is already registered with addRequestListener and cannot also use addListener.');
    }
    if (!_handlerDic.containsKey(canonical)) {
      _handlerDic[canonical] = DataHandler<T>((typeName, data) {
        final message = _decodeMessage(typeName, data, fallbackTypeName: type);
        if (message is! T) {
          throw StateError(
              'Message type mismatch: expected $canonical, got $typeName');
        }
        return message;
      });
    }
    var handler = _handlerDic[canonical] as DataHandler<T>;
    handler.addListener(listener);
  }

  void removeListener<T extends GeneratedMessage>(void Function(T) listener) {
    var type = _qualifiedMessageNameOf<T>();
    var canonical = _canonicalize(type);
    if (_handlerDic.containsKey(canonical)) {
      var handler = _handlerDic[canonical] as DataHandler<T>;
      handler.removeListener(listener);
    }
  }
}

abstract class IDataHandler {
  void onMessage(String typeName, List<int> data);
}

class DataHandler<T extends GeneratedMessage> implements IDataHandler {
  T Function(String, List<int>) _generator;
  List<void Function(T)> _listeners = [];

  DataHandler(this._generator);

  @override
  void onMessage(String typeName, List<int> data) {
    for (var listener in _listeners) {
      listener.call(_generator(typeName, data));
    }
  }

  void addListener(void Function(T) handler) {
    removeListener(handler); // 중복 리스너 불허
    _listeners.add(handler);
  }

  void removeListener(void Function(T) handler) {
    _listeners.remove(handler);
  }
}

class _PendingRequest {
  final String expectedTypeName;
  final void Function(String, List<int>) complete;
  final void Function(Object, StackTrace) completeError;

  _PendingRequest({
    required this.expectedTypeName,
    required this.complete,
    required this.completeError,
  });
}