import 'dart:async';
import 'dart:convert';
import 'package:dartssh2/dartssh2.dart';
import '../core/failure.dart';

abstract interface class MonitorAdapter {
  Future<Map<String, String>> sample(SSHClient client);
  void cancel();
}

class LinuxMonitor implements MonitorAdapter {
  final Set<SSHSession> _active = {};
  int _epoch = 0;
  @override
  void cancel() {
    _epoch++;
    for (final session in _active) {
      session.close();
    }
    _active.clear();
  }

  Future<SSHSession> _execute(
    SSHClient client,
    String command,
    int epoch,
  ) async {
    return client
        .execute(command)
        .then((session) {
          if (epoch != _epoch) {
            session.close();
            throw const Failure('cancelled');
          }
          _active.add(session);
          return session;
        })
        .timeout(
          const Duration(seconds: 10),
          onTimeout: () {
            cancel();
            throw TimeoutException('SSH channel');
          },
        );
  }

  static const command = r'''LC_ALL=C; export LC_ALL
test "$(uname -s)" = Linux || exit 70
printf 'system='; uname -sr
printf 'release='; sed -n 's/^PRETTY_NAME=//p' /etc/os-release
printf 'uptime='; cut -d ' ' -f 1 /proc/uptime
printf 'load='; cut -d ' ' -f 1-3 /proc/loadavg
awk '/^MemTotal:/{t=$2}/^MemAvailable:/{a=$2}END{if(t>0)printf "ram=%.1f%% (%d / %d MiB)\n",100*(t-a)/t,(t-a)/1024,t/1024}' /proc/meminfo
df -Pk / | awk 'NR==2{printf "disk=%s (%s / %s KiB)\n",$5,$3,$2}'
a=$(awk '/^cpu /{t=0;for(i=2;i<=9;i++)t+=$i;printf "%.0f %.0f\n",t,$5+$6;exit}' /proc/stat)
sleep 1
b=$(awk '/^cpu /{t=0;for(i=2;i<=9;i++)t+=$i;printf "%.0f %.0f\n",t,$5+$6;exit}' /proc/stat)
awk -v a="$a" -v b="$b" 'BEGIN{split(a,x," ");split(b,y," ");t=y[1]-x[1];if(t>0)printf "cpu=%.1f%%\n",100*(t-y[2]+x[2])/t}'
''';
  @override
  Future<Map<String, String>> sample(SSHClient client) async {
    final epoch = _epoch;
    final watch = Stopwatch()..start();
    final ping = await _execute(client, 'true', epoch);
    try {
      await ping.done.timeout(const Duration(seconds: 10));
    } finally {
      ping.close();
      _active.remove(ping);
    }
    final rtt = watch.elapsedMilliseconds;
    final session = await _execute(client, command, epoch);
    final output = StringBuffer();
    var bytes = 0;
    final sub = session.stdout
        .cast<List<int>>()
        .transform(const Utf8Decoder(allowMalformed: true))
        .listen((s) {
          bytes += s.length;
          if (bytes <= 65536) {
            output.write(s);
          } else {
            session.close();
          }
        });
    final err = session.stderr.listen((_) {});
    try {
      await session.done.timeout(const Duration(seconds: 15));
      if (session.exitCode != 0 || bytes > 65536) {
        throw const Failure('unsupportedSystem');
      }
      return {
        'rtt': '$rtt ms',
        for (final line in output.toString().split('\n'))
          if (line.contains('='))
            line.substring(0, line.indexOf('=')): line.substring(
              line.indexOf('=') + 1,
            ),
      };
    } finally {
      await sub.cancel();
      await err.cancel();
      session.close();
      _active.remove(session);
    }
  }
}