import 'dart:async';
import 'dart:ffi';
import 'dart:io';
import 'dart:isolate';
import 'dart:math' as math;
import 'dart:typed_data';
import 'package:ffi/ffi.dart';
import 'icmp_packet.dart';
import 'posix.dart';

/// Settings of an ICMP echo measurement. [count] 0 means continuous until
/// stopped. [ttl] null keeps the system default TTL / hop limit.
class PingSettings {
  final int count, size;
  final int? ttl;
  final bool dontFragment;
  final Duration interval, timeout;
  const PingSettings({
    this.count = 4,
    this.interval = const Duration(seconds: 1),
    this.timeout = const Duration(seconds: 2),
    this.size = 56,
    this.ttl,
    this.dontFragment = false,
  });
  Map<String, Object?> toJson() => {
    'count': count == 0 ? 'continuous' : count,
    'intervalMs': interval.inMilliseconds,
    'timeoutMs': timeout.inMilliseconds,
    'payloadBytes': size,
    'ttl': ttl ?? 'system default',
    'dontFragment': dontFragment,
  };
}

/// How traceroute probes are sent.
enum TraceMethod {
  /// ICMP echo requests with increasing TTL.
  icmp,

  /// UDP datagrams to high ports; the destination answers with ICMP port
  /// unreachable.
  udp,

  /// TCP SYN to a port; the destination answers with SYN-ACK or reset.
  tcp,
}

/// Settings of a traceroute.
class TraceSettings {
  final int maxHops, probes, size, firstTtl;
  final Duration timeout, interval;
  final TraceMethod method;

  /// UDP: first destination port (incremented per probe); TCP: the port.
  final int? port;
  const TraceSettings({
    this.maxHops = 30,
    this.probes = 3,
    this.timeout = const Duration(seconds: 2),
    this.size = 32,
    this.firstTtl = 1,
    this.interval = Duration.zero,
    this.method = TraceMethod.icmp,
    this.port,
  });
  int get effectivePort =>
      port ?? (method == TraceMethod.tcp ? 80 : defaultUdpPort);
  static const defaultUdpPort = 33434;
  Map<String, Object?> toJson() => {
    'method': method.name,
    'maxHops': maxHops,
    'probesPerHop': probes,
    'firstTtl': firstTtl,
    'timeoutMs': timeout.inMilliseconds,
    'intervalMs': interval.inMilliseconds,
    if (method != TraceMethod.icmp) 'port': effectivePort,
  };
}

/// One echo request outcome.
enum ProbeStatus { reply, timeout, timeExceeded, unreachable, sendFailed }

class ProbeEvent {
  /// Sequence number for ping; probe number within the hop for traceroute.
  final int seq;

  /// Hop number (TTL / hop limit) for traceroute; 0 for ping.
  final int ttl;
  final ProbeStatus status;
  final InternetAddress? from;
  final double? rttMs;
  final int? icmpCode;

  /// When the probe was sent.
  final DateTime? time;

  /// Additional detail: 'portUnreachable' (UDP destination reached),
  /// 'tcpOpen' / 'tcpRefused' (TCP destination reached), 'messageTooLong'
  /// or 'errno N'.
  final String? note;
  const ProbeEvent(
    this.seq,
    this.status, {
    this.ttl = 0,
    this.from,
    this.rttMs,
    this.icmpCode,
    this.time,
    this.note,
  });
}

/// Ping statistics in the usual min/avg/max/loss form.
class PingStats {
  final int sent, received;
  final double? min, avg, max;
  const PingStats(this.sent, this.received, this.min, this.avg, this.max);
  double get lossPercent => sent == 0 ? 0 : 100 * (sent - received) / sent;

  factory PingStats.of(Iterable<ProbeEvent> events) {
    final rtts = [
      for (final e in events)
        if (e.status == ProbeStatus.reply && e.rttMs != null) e.rttMs!,
    ];
    return PingStats(
      events.length,
      rtts.length,
      rtts.isEmpty ? null : rtts.reduce(math.min),
      rtts.isEmpty ? null : rtts.reduce((a, b) => a + b) / rtts.length,
      rtts.isEmpty ? null : rtts.reduce(math.max),
    );
  }
}

/// Raised when the platform does not allow an unprivileged ICMP socket or
/// a socket call fails before any packet could be sent.
class IcmpUnavailable implements Exception {
  /// 'icmpUnsupported', 'icmpNotPermitted', 'networkError',
  /// 'tcpTraceUnsupported', 'ttlUnsupported', 'dontFragmentUnsupported'.
  final String code;
  final int? errno;
  const IcmpUnavailable(this.code, [this.errno]);
  @override
  String toString() => 'IcmpUnavailable($code, errno $errno)';
}

/// A running measurement: [events] streams results; [cancel] stops it at
/// once and closes the socket.
class IcmpRun {
  final Stream<ProbeEvent> events;
  final Future<void> Function() cancel;
  IcmpRun(this.events, this.cancel);
}

class IcmpProbe {
  /// Null when [method] can run on this platform without privileges,
  /// otherwise the localization key explaining why not.
  ///
  /// iOS/macOS deliver ICMP errors for UDP and TCP probes to an unprivileged
  /// ICMP datagram socket, so all methods work. Android/Linux report ICMP
  /// errors for UDP through the socket error queue; for TCP connections the
  /// kernel reports only an error code without the router's address, so TCP
  /// traceroute is not available there.
  static String? unsupported(TraceMethod method) {
    final p = Posix.current;
    if (p == null) return 'icmpUnsupported';
    if (method == TraceMethod.tcp && !p.darwin) return 'tcpTraceUnsupported';
    return null;
  }

  /// Sends [settings].count echo requests to [target].
  static IcmpRun ping(InternetAddress target, PingSettings settings) => _start({
    'mode': 'ping',
    'address': target.rawAddress,
    'count': settings.count,
    'interval': settings.interval.inMicroseconds,
    'timeout': settings.timeout.inMicroseconds,
    'size': settings.size,
    'ttl': ?settings.ttl,
    'df': settings.dontFragment,
  });

  /// Probes each hop to [target] with increasing TTL / hop limit.
  static IcmpRun traceroute(InternetAddress target, TraceSettings settings) =>
      _start({
        'mode': 'trace',
        'method': settings.method.name,
        'address': target.rawAddress,
        'maxHops': settings.maxHops,
        'probes': settings.probes,
        'timeout': settings.timeout.inMicroseconds,
        'interval': settings.interval.inMicroseconds,
        'size': settings.size,
        'firstTtl': settings.firstTtl,
        'port': settings.effectivePort,
      });

  static IcmpRun _start(Map<String, Object> config) {
    final controller = StreamController<ProbeEvent>();
    final fromWorker = ReceivePort();
    SendPort? toWorker;
    Isolate? isolate;
    var done = false;
    void finish() {
      if (done) return;
      done = true;
      fromWorker.close();
      controller.close();
    }

    fromWorker.listen((message) {
      if (message is SendPort) {
        toWorker = message;
      } else if (message is Map) {
        if (message['error'] != null) {
          controller.addError(
            IcmpUnavailable(
              message['error'] as String,
              message['errno'] as int?,
            ),
          );
        } else {
          controller.add(_decode(message));
        }
      } else if (message == null) {
        finish();
      }
    });
    Isolate.spawn(
      _worker,
      [fromWorker.sendPort, config],
      onExit: fromWorker.sendPort,
      onError: fromWorker.sendPort,
    ).then(
      (i) {
        isolate = i;
        if (done) i.kill(priority: Isolate.immediate);
      },
      onError: (Object e) {
        controller.addError(const IcmpUnavailable('icmpUnsupported'));
        finish();
      },
    );
    Future<void> cancel() async {
      // The worker closes its socket when told; kill it if it does not exit.
      toWorker?.send('cancel');
      final pending = isolate;
      if (pending != null) {
        Timer(const Duration(milliseconds: 500), () {
          pending.kill(priority: Isolate.immediate);
        });
      }
      finish();
    }

    controller.onCancel = cancel;
    return IcmpRun(controller.stream, cancel);
  }

  static ProbeEvent _decode(Map m) => ProbeEvent(
    m['seq'] as int,
    ProbeStatus.values[m['status'] as int],
    ttl: m['ttl'] as int,
    from: m['from'] == null
        ? null
        : InternetAddress.fromRawAddress(m['from'] as Uint8List),
    rttMs: m['rtt'] as double?,
    icmpCode: m['code'] as int?,
    time: m['time'] == null
        ? null
        : DateTime.fromMicrosecondsSinceEpoch(m['time'] as int),
    note: m['note'] as String?,
  );
}

// ---------------------------------------------------------------------------
// Worker isolate: owns the sockets and all blocking calls.

Future<void> _worker(List<Object> args) async {
  final out = args[0] as SendPort, config = args[1] as Map<String, Object>;
  final inbox = ReceivePort();
  out.send(inbox.sendPort);
  var cancelled = false;
  inbox.listen((_) => cancelled = true);
  final posix = Posix.current;
  if (posix == null) {
    out.send({'error': 'icmpUnsupported'});
    inbox.close();
    return;
  }
  final target = InternetAddress.fromRawAddress(config['address'] as Uint8List);
  final session = _Session(posix, target);
  try {
    if (config['mode'] == 'ping') {
      session.open();
      await _ping(session, config, out, () => cancelled);
    } else {
      final method = TraceMethod.values.byName(config['method'] as String);
      if (method == TraceMethod.tcp && !posix.darwin) {
        throw const IcmpUnavailable('tcpTraceUnsupported');
      }
      // ICMP errors for UDP/TCP probes: Darwin delivers them to the ICMP
      // socket; Linux to the UDP socket's own error queue.
      if (method == TraceMethod.icmp || posix.darwin) session.open();
      await _trace(session, method, config, out, () => cancelled);
    }
  } on IcmpUnavailable catch (e) {
    out.send({'error': e.code, 'errno': e.errno});
  } finally {
    session.close();
    inbox.close();
  }
}

void _send(SendPort out, ProbeEvent e) => out.send({
  'seq': e.seq,
  'ttl': e.ttl,
  'status': e.status.index,
  'from': e.from?.rawAddress,
  'rtt': e.rttMs,
  'code': e.icmpCode,
  'time': e.time?.microsecondsSinceEpoch,
  'note': e.note,
});

Future<void> _ping(
  _Session s,
  Map<String, Object> c,
  SendPort out,
  bool Function() cancelled,
) async {
  final count = c['count'] as int, size = c['size'] as int;
  final interval = c['interval'] as int, timeout = c['timeout'] as int;
  final ttl = c['ttl'] as int?;
  if (ttl != null && !s.hopLimit(ttl)) {
    throw IcmpUnavailable('ttlUnsupported', s.p.errno);
  }
  if (c['df'] == true && !s.dontFragment()) {
    throw IcmpUnavailable('dontFragmentUnsupported', s.p.errno);
  }
  final clock = s.clock;
  final pending = <int, int>{}; // seq -> send time (µs)
  final sentAt = <int, DateTime>{};
  var nextSend = 0, sent = 0, seq = 0;
  bool more() => count == 0 || sent < count;
  while (!cancelled()) {
    final now = clock.elapsedMicroseconds;
    if (more() && now >= nextSend) {
      sent++;
      // Sequence numbers are 16 bits; a continuous ping wraps around.
      seq = seq % 0xffff + 1;
      final time = DateTime.now();
      if (s.send(seq, size)) {
        pending[seq] = clock.elapsedMicroseconds;
        sentAt[seq] = time;
      } else {
        _send(
          out,
          ProbeEvent(
            seq,
            ProbeStatus.sendFailed,
            time: time,
            note: _errnoNote(s.p, s.p.errno),
          ),
        );
      }
      nextSend += interval;
    }
    for (final entry in pending.entries.toList()) {
      if (clock.elapsedMicroseconds - entry.value > timeout) {
        pending.remove(entry.key);
        _send(
          out,
          ProbeEvent(
            entry.key,
            ProbeStatus.timeout,
            time: sentAt.remove(entry.key),
          ),
        );
      }
    }
    if (!more() && pending.isEmpty) break;
    final wait = more()
        ? nextSend - clock.elapsedMicroseconds
        : timeout -
              (clock.elapsedMicroseconds - pending.values.reduce(math.min));
    for (final r in s.receiveEcho(math.max(0, math.min(wait, 50000)))) {
      final start = pending.remove(r.seq);
      if (start == null) continue;
      _send(
        out,
        ProbeEvent(
          r.seq,
          r.status,
          from: r.from,
          rttMs: (r.receivedAt - start) / 1000,
          icmpCode: r.code,
          time: sentAt.remove(r.seq),
        ),
      );
    }
    // Lets the isolate process a cancel message between short polls.
    await Future<void>.delayed(Duration.zero);
  }
}

String _errnoNote(Posix p, int errno) =>
    errno == p.eMsgSize ? 'messageTooLong' : 'errno $errno';

/// A probe sender for one traceroute method.
abstract class _Prober {
  /// Sends probe [n] with [ttl]; returns the send time on the session clock,
  /// or null (with errno set) when sending failed.
  int? send(int n, int ttl);

  /// Waits up to [micros] for the answer to the probe sent last.
  _Received? poll(int micros);
  void close();
}

Future<void> _trace(
  _Session s,
  TraceMethod method,
  Map<String, Object> c,
  SendPort out,
  bool Function() cancelled,
) async {
  final maxHops = c['maxHops'] as int, probes = c['probes'] as int;
  final timeout = c['timeout'] as int, size = c['size'] as int;
  final interval = c['interval'] as int, firstTtl = c['firstTtl'] as int;
  final port = c['port'] as int;
  final _Prober prober = switch (method) {
    TraceMethod.icmp => _IcmpProber(s, size),
    TraceMethod.udp =>
      s.p.darwin
          ? _DarwinUdpProber(s, port, size)
          : _LinuxUdpProber(s, port, size),
    TraceMethod.tcp => _DarwinTcpProber(s, port),
  };
  final clock = s.clock;
  var n = 0;
  try {
    for (var ttl = firstTtl; ttl <= maxHops && !cancelled(); ttl++) {
      var reached = false, unreachable = false;
      for (var probe = 1; probe <= probes && !cancelled(); probe++) {
        if (n > 0 && interval > 0) {
          final until = clock.elapsedMicroseconds + interval;
          while (!cancelled() && clock.elapsedMicroseconds < until) {
            await Future<void>.delayed(const Duration(milliseconds: 10));
          }
          if (cancelled()) return;
        }
        n++;
        final time = DateTime.now();
        final sent = prober.send(n, ttl);
        if (sent == null) {
          _send(
            out,
            ProbeEvent(
              probe,
              ProbeStatus.sendFailed,
              ttl: ttl,
              time: time,
              note: _errnoNote(s.p, s.p.errno),
            ),
          );
          continue;
        }
        _Received? answer;
        while (!cancelled()) {
          final left = timeout - (clock.elapsedMicroseconds - sent);
          if (left <= 0) break;
          answer = prober.poll(math.min(left, 50000));
          if (answer != null) break;
          await Future<void>.delayed(Duration.zero);
        }
        if (cancelled()) return;
        if (answer == null) {
          _send(
            out,
            ProbeEvent(probe, ProbeStatus.timeout, ttl: ttl, time: time),
          );
          continue;
        }
        reached |= answer.status == ProbeStatus.reply;
        unreachable |= answer.status == ProbeStatus.unreachable;
        _send(
          out,
          ProbeEvent(
            probe,
            answer.status,
            ttl: ttl,
            from: answer.from,
            rttMs: (answer.receivedAt - sent) / 1000,
            icmpCode: answer.code,
            time: time,
            note: answer.note,
          ),
        );
      }
      if (reached || unreachable) break;
    }
  } finally {
    prober.close();
  }
}

class _Received {
  final int seq;
  final ProbeStatus status;
  final InternetAddress? from;
  final int? code;
  final int receivedAt; // µs on the session clock
  final String? note;
  _Received(
    this.seq,
    this.status,
    this.from,
    this.code,
    this.receivedAt, [
    this.note,
  ]);
}

/// A received ICMP message with where and when it arrived.
class _Inbound {
  final IcmpMessage message;
  final InternetAddress? from;
  final int at;
  _Inbound(this.message, this.from, this.at);
}

ProbeStatus? _errorStatus(IcmpKind kind) => switch (kind) {
  IcmpKind.timeExceeded => ProbeStatus.timeExceeded,
  IcmpKind.unreachable => ProbeStatus.unreachable,
  _ => null,
};

class _IcmpProber implements _Prober {
  final _Session s;
  final int size;
  int _seq = 0;
  _IcmpProber(this.s, this.size);
  @override
  int? send(int n, int ttl) {
    s.hopLimit(ttl);
    _seq = n & 0xffff;
    final at = s.clock.elapsedMicroseconds;
    return s.send(_seq, size) ? at : null;
  }

  @override
  _Received? poll(int micros) {
    for (final r in s.receiveEcho(micros)) {
      if (r.seq == _seq) return r;
    }
    return null;
  }

  @override
  void close() {}
}

/// Classic traceroute ports: one destination port per probe.
int _probePort(int base, int n) => base + (n - 1) % math.max(1, 65535 - base);

/// Darwin: what an ICMP message about one of our UDP/TCP probes means.
/// Port unreachable ends the trace like in the classic tool; when it comes
/// from another address than the target (anycast, a host firewall), the
/// note says so.
_Received? _transportAnswer(_Session s, _Inbound m) {
  final msg = m.message;
  if (msg.portUnreachable(s.v6)) {
    return _Received(
      0,
      ProbeStatus.reply,
      m.from,
      msg.code,
      m.at,
      m.from == s.target ? 'portUnreachable' : 'portUnreachableOther',
    );
  }
  final status = _errorStatus(msg.kind);
  if (status == null) return null;
  return _Received(0, status, m.from, msg.code, m.at);
}

/// Darwin: UDP probes from an ordinary UDP socket; the ICMP errors they
/// cause arrive on the session's ICMP socket, matched by the quoted ports.
class _DarwinUdpProber implements _Prober {
  final _Session s;
  final int basePort, size;
  late final int fd;
  int _port = 0, _source = 0;
  _DarwinUdpProber(this.s, this.basePort, this.size) {
    fd = s.p.socket(
      s.v6 ? s.p.afInet6 : s.p.afInet,
      s.p.sockDgram,
      s.p.ipprotoUdp,
    );
    if (fd < 0) throw IcmpUnavailable('networkError', s.p.errno);
  }

  @override
  int? send(int n, int ttl) {
    s.setTtl(fd, ttl);
    _port = _probePort(basePort, n);
    final at = s.clock.elapsedMicroseconds;
    if (!s.sendDatagram(fd, _port, size)) return null;
    if (_source == 0) _source = s.localPort(fd);
    return at;
  }

  @override
  _Received? poll(int micros) {
    for (final m in s.receiveRaw(micros)) {
      final msg = m.message;
      if (msg.protocol != 17 ||
          msg.destinationPort != _port ||
          msg.sourcePort != _source) {
        continue;
      }
      final answer = _transportAnswer(s, m);
      if (answer != null) return answer;
    }
    return null;
  }

  @override
  void close() => s.p.close(fd);
}

/// Darwin: a non-blocking TCP connect with the TTL set; routers answer the
/// SYN with time exceeded (seen on the ICMP socket), the destination with
/// SYN-ACK (connected) or a reset (refused).
class _DarwinTcpProber implements _Prober {
  final _Session s;
  final int port;
  int fd = -1, _source = 0;
  _DarwinTcpProber(this.s, this.port);

  @override
  int? send(int n, int ttl) {
    _closeSocket();
    final p = s.p;
    fd = p.socket(s.v6 ? p.afInet6 : p.afInet, p.sockStream, p.ipprotoTcp);
    if (fd < 0) return null;
    s.setTtl(fd, ttl);
    final flags = p.fcntl(fd, p.fGetFl, 0);
    p.fcntl(fd, p.fSetFl, flags | p.oNonBlock);
    final address = p.sockaddr(s.target, port);
    final buffer = calloc<Uint8>(address.length);
    try {
      buffer.asTypedList(address.length).setAll(0, address);
      final at = s.clock.elapsedMicroseconds;
      final result = p.connect(fd, buffer, address.length);
      // EINPROGRESS is the normal case; a refusal may be immediate on
      // loopback and is then reported by the next poll.
      if (result != 0) {
        final errno = p.errno;
        if (errno != p.eInProgress && errno != p.eConnRefused) return null;
      }
      _source = s.localPort(fd);
      return at;
    } finally {
      calloc.free(buffer);
    }
  }

  _Received? _matching(int micros) {
    for (final m in s.receiveRaw(micros)) {
      final msg = m.message;
      if (msg.protocol == 6 &&
          msg.destinationPort == port &&
          msg.sourcePort == _source) {
        final answer = _transportAnswer(s, m);
        if (answer != null) return answer;
      }
    }
    return null;
  }

  @override
  _Received? poll(int micros) {
    final p = s.p;
    final early = _matching(0);
    if (early != null || fd < 0) return early;
    final fds = calloc<PollFd>(2);
    try {
      fds[0]
        ..fd = s.fd
        ..events = Posix.pollIn
        ..revents = 0;
      fds[1]
        ..fd = fd
        ..events = Posix.pollOut
        ..revents = 0;
      final ready = p.poll(fds, 2, (micros / 1000).ceil());
      if (ready <= 0) return null;
      if (fds[0].revents & Posix.pollIn != 0) {
        final answer = _matching(0);
        if (answer != null) return answer;
      }
      if (fds[1].revents != 0) {
        final at = s.clock.elapsedMicroseconds;
        final error = s.socketError(fd);
        if (error == 0 || error == p.eConnRefused) {
          _closeSocket();
          return _Received(
            0,
            ProbeStatus.reply,
            s.target,
            null,
            at,
            error == 0 ? 'tcpOpen' : 'tcpRefused',
          );
        }
        // Another error (no route after a time exceeded): the ICMP message
        // names the router and is normally already readable.
        final answer = _matching(20000);
        _closeSocket();
        return answer ??
            _Received(
              0,
              ProbeStatus.unreachable,
              null,
              null,
              at,
              'errno $error',
            );
      }
    } finally {
      calloc.free(fds);
    }
    return null;
  }

  void _closeSocket() {
    if (fd >= 0) s.p.close(fd);
    fd = -1;
  }

  @override
  void close() => _closeSocket();
}

/// Linux/Android: UDP probes on a socket with IP_RECVERR; ICMP errors come
/// back on its error queue together with the original destination port.
class _LinuxUdpProber implements _Prober {
  final _Session s;
  final int basePort, size;
  late final int fd;
  int _port = 0;
  _LinuxUdpProber(this.s, this.basePort, this.size) {
    final p = s.p;
    fd = p.socket(s.v6 ? p.afInet6 : p.afInet, p.sockDgram, p.ipprotoUdp);
    if (fd < 0) throw IcmpUnavailable('networkError', p.errno);
    s.setInt(
      fd,
      s.v6 ? p.ipprotoIpv6 : p.ipprotoIp,
      s.v6 ? p.ipv6RecvErr : p.ipRecvErr,
      1,
    );
  }

  @override
  int? send(int n, int ttl) {
    s.setTtl(fd, ttl);
    _port = _probePort(basePort, n);
    final at = s.clock.elapsedMicroseconds;
    return s.sendDatagram(fd, _port, size) ? at : null;
  }

  @override
  _Received? poll(int micros) {
    for (final e in s.readErrorQueue(fd, micros)) {
      if (e.port != _port) continue;
      final (type, code, offender) = e.error;
      final unreachable = s.v6 ? IcmpType.unreachable6 : IcmpType.unreachable4;
      final exceeded = s.v6 ? IcmpType.timeExceeded6 : IcmpType.timeExceeded4;
      if (type == unreachable && code == (s.v6 ? 4 : 3)) {
        return _Received(
          0,
          ProbeStatus.reply,
          offender,
          code,
          e.at,
          offender == s.target ? 'portUnreachable' : 'portUnreachableOther',
        );
      }
      if (type == exceeded) {
        return _Received(0, ProbeStatus.timeExceeded, offender, code, e.at);
      }
      if (type == unreachable) {
        return _Received(0, ProbeStatus.unreachable, offender, code, e.at);
      }
    }
    return null;
  }

  @override
  void close() => s.p.close(fd);
}

typedef _QueuedError = ({
  Uint8List data,
  int port,
  (int, int, InternetAddress?) error,
  int at,
});

/// One ICMP socket and the buffers used with it.
class _Session {
  final Posix p;
  final InternetAddress target;
  final bool v6;
  // Darwin keeps the id we choose; Linux replaces it with the socket's own
  // identifier, so on Linux replies are matched by sequence only (the kernel
  // already delivers only this socket's replies).
  final int id = math.Random.secure().nextInt(0xffff);
  int fd = -1;

  /// Send and receive times are both taken from this clock.
  final Stopwatch clock = Stopwatch()..start();
  final Pointer<Uint8> _buffer = calloc<Uint8>(_bufferSize);
  final Pointer<Uint8> _address = calloc<Uint8>(128);
  final Pointer<Uint32> _addressLength = calloc<Uint32>();
  final Pointer<PollFd> _poll = calloc<PollFd>();
  static const _bufferSize = 65536;

  _Session(this.p, this.target) : v6 = target.type == InternetAddressType.IPv6;

  void open() {
    fd = p.socket(
      v6 ? p.afInet6 : p.afInet,
      p.sockDgram,
      v6 ? p.ipprotoIcmpv6 : p.ipprotoIcmp,
    );
    if (fd < 0) {
      final errno = p.errno;
      throw IcmpUnavailable(
        errno == p.eAccess || errno == p.ePerm
            ? 'icmpNotPermitted'
            : 'icmpUnsupported',
        errno,
      );
    }
    if (!p.darwin) {
      // Deliver ICMP errors (time exceeded, unreachable) to the error queue.
      setInt(
        fd,
        v6 ? p.ipprotoIpv6 : p.ipprotoIp,
        v6 ? p.ipv6RecvErr : p.ipRecvErr,
        1,
      );
    }
  }

  bool setInt(int socket, int level, int name, int value) {
    final v = calloc<Int32>()..value = value;
    try {
      return p.setsockopt(socket, level, name, v.cast(), 4) == 0;
    } finally {
      calloc.free(v);
    }
  }

  bool setTtl(int socket, int ttl) => setInt(
    socket,
    v6 ? p.ipprotoIpv6 : p.ipprotoIp,
    v6 ? p.ipv6UnicastHops : p.ipTtl,
    ttl,
  );

  bool hopLimit(int ttl) => setTtl(fd, ttl);

  bool dontFragment() => setInt(
    fd,
    v6 ? p.ipprotoIpv6 : p.ipprotoIp,
    v6 ? p.ipv6DontFrag : p.ipDontFrag,
    p.dontFragValue,
  );

  int socketError(int socket) {
    final value = calloc<Int32>(), length = calloc<Uint32>()..value = 4;
    try {
      if (p.getsockopt(socket, p.solSocket, p.soError, value.cast(), length) !=
          0) {
        return p.errno;
      }
      return value.value;
    } finally {
      calloc.free(value);
      calloc.free(length);
    }
  }

  int localPort(int socket) {
    _addressLength.value = 128;
    if (p.getsockname(socket, _address, _addressLength) != 0) return 0;
    return p.port(_address.asTypedList(4));
  }

  bool _sendTo(int socket, int length, int port) {
    final address = p.sockaddr(target, port);
    final to = calloc<Uint8>(address.length);
    try {
      to.asTypedList(address.length).setAll(0, address);
      return p.sendto(socket, _buffer, length, 0, to, address.length) == length;
    } finally {
      calloc.free(to);
    }
  }

  bool send(int seq, int size) {
    final packet = buildEchoRequest(v6: v6, id: id, seq: seq, size: size);
    _buffer.asTypedList(packet.length).setAll(0, packet);
    return _sendTo(fd, packet.length, 0);
  }

  bool sendDatagram(int socket, int port, int size) {
    final payload = _buffer.asTypedList(size);
    for (var i = 0; i < size; i++) {
      payload[i] = 0x40 + i % 32;
    }
    return _sendTo(socket, size, port);
  }

  /// Echo replies and errors for this session's echo requests.
  List<_Received> receiveEcho(int micros) {
    final out = <_Received>[];
    for (final m in receiveRaw(micros)) {
      final message = m.message;
      if (message.seq == null) continue;
      if (p.darwin && message.id != id) continue; // another process's echo
      final status = message.kind == IcmpKind.echoReply
          ? ProbeStatus.reply
          : _errorStatus(message.kind);
      if (status == null) continue;
      if (status == ProbeStatus.reply && m.from != null && m.from != target) {
        continue;
      }
      out.add(_Received(message.seq!, status, m.from, message.code, m.at));
    }
    return out;
  }

  /// All ICMP messages that arrive within [micros] on the ICMP socket. On
  /// Linux, errors for echo requests come from the error queue.
  List<_Inbound> receiveRaw(int micros) {
    final out = <_Inbound>[];
    if (fd < 0) return out;
    _poll.ref
      ..fd = fd
      ..events = Posix.pollIn
      ..revents = 0;
    final ready = p.poll(_poll, 1, (micros / 1000).ceil());
    if (ready <= 0) return out;
    final revents = _poll.ref.revents;
    if (revents & Posix.pollIn != 0) _readMessages(out);
    if (!p.darwin && revents & Posix.pollErr != 0) {
      for (final e in readErrorQueue(fd, 0)) {
        final sent = sentEcho(e.data, v6: v6);
        if (sent == null) continue;
        final (type, code, offender) = e.error;
        final kind =
            type == (v6 ? IcmpType.timeExceeded6 : IcmpType.timeExceeded4)
            ? IcmpKind.timeExceeded
            : type == (v6 ? IcmpType.unreachable6 : IcmpType.unreachable4)
            ? IcmpKind.unreachable
            : IcmpKind.other;
        out.add(
          _Inbound(
            IcmpMessage(kind, type, code, id: sent.id, seq: sent.seq),
            offender,
            e.at,
          ),
        );
      }
    }
    return out;
  }

  void _readMessages(List<_Inbound> out) {
    while (true) {
      _addressLength.value = 128;
      final n = p.recvfrom(
        fd,
        _buffer,
        _bufferSize,
        p.msgDontWait,
        _address,
        _addressLength,
      );
      if (n <= 0) return;
      final at = clock.elapsedMicroseconds;
      final data = Uint8List.fromList(_buffer.asTypedList(n));
      final message = parseIcmp(data, v6: v6, ipHeader: p.darwin && !v6);
      if (message == null) continue;
      final from = p.address(_address.asTypedList(128), _addressLength.value);
      out.add(_Inbound(message, from, at));
    }
  }

  /// Linux: reads the error queue of [socket], waiting up to [micros] for
  /// an entry. Each entry carries the offending packet's payload, its
  /// original destination port and the ICMP type, code and sender.
  List<_QueuedError> readErrorQueue(int socket, int micros) {
    final out = <_QueuedError>[];
    if (micros > 0) {
      _poll.ref
        ..fd = socket
        ..events = Posix.pollIn
        ..revents = 0;
      final ready = p.poll(_poll, 1, (micros / 1000).ceil());
      if (ready <= 0 || _poll.ref.revents & Posix.pollErr == 0) return out;
    }
    final msg = calloc<MsgHdr>();
    final iov = calloc<IoVec>();
    const controlSize = 512;
    final control = calloc<Uint8>(controlSize);
    final name = calloc<Uint8>(128);
    try {
      while (true) {
        iov.ref
          ..base = _buffer
          ..length = _bufferSize;
        msg.ref
          ..name = name
          ..nameLength = 128
          ..iov = iov
          ..iovLength = 1
          ..control = control
          ..controlLength = controlSize
          ..flags = 0;
        final n = p.recvmsg(socket, msg, p.msgErrQueue | p.msgDontWait);
        if (n < 0) return out;
        final at = clock.elapsedMicroseconds;
        final error = _extendedError(msg.ref);
        if (error == null) continue;
        out.add((
          data: Uint8List.fromList(_buffer.asTypedList(n)),
          port: p.port(name.asTypedList(4)),
          error: error,
          at: at,
        ));
      }
    } finally {
      calloc.free(name);
      calloc.free(control);
      calloc.free(iov);
      calloc.free(msg);
    }
  }

  (int, int, InternetAddress?)? _extendedError(MsgHdr msg) {
    final word = sizeOf<IntPtr>();
    final header = sizeOf<CmsgHdr>();
    var offset = 0;
    final base = msg.control;
    while (offset + header <= msg.controlLength) {
      final cmsg = Pointer<CmsgHdr>.fromAddress(base.address + offset).ref;
      if (cmsg.length < header) return null;
      final level = v6 ? p.ipprotoIpv6 : p.ipprotoIp;
      final type = v6 ? p.ipv6RecvErr : p.ipRecvErr;
      final dataOffset = offset + ((header + word - 1) & ~(word - 1));
      if (cmsg.level == level && cmsg.type == type) {
        // struct sock_extended_err: u32 errno; u8 origin, type, code, pad;
        // u32 info, data; followed by the offender's sockaddr.
        final ee = Pointer<Uint8>.fromAddress(
          base.address + dataOffset,
        ).asTypedList(cmsg.length - (dataOffset - offset));
        final origin = ee[4];
        if (origin != Posix.originIcmp && origin != Posix.originIcmp6) {
          return null;
        }
        final offender = ee.length >= 16 + 8
            ? p.address(Uint8List.fromList(ee.sublist(16)), ee.length - 16)
            : null;
        return (ee[5], ee[6], offender);
      }
      offset += (cmsg.length + word - 1) & ~(word - 1);
    }
    return null;
  }

  void close() {
    if (fd >= 0) p.close(fd);
    fd = -1;
    calloc.free(_buffer);
    calloc.free(_address);
    calloc.free(_addressLength);
    calloc.free(_poll);
  }
}