import 'dart:async';
import 'dart:convert';
import 'dart:typed_data';

import 'package:dartssh2/dartssh2.dart';
import '../about/external/data/Docker.dart';
import '../about/external/data/Health.dart';
import '../about/external/data/Kafka.dart';
import '../about/external/data/Redis.dart';
import '../about/internal/application/ServerProfile.dart';
import '../about/internal/application/ServiceConfig.dart';
import '../about/internal/application/StepTracker.dart';
import 'ConnectComms.dart';

enum LogLevel { info, step, success, error, remote, warn }

class LogLine {
  final LogLevel level;
  final String text;
  final DateTime time;
  LogLine(this.level, this.text) : time = DateTime.now();
}

/// Result of a deploy attempt, surfaced to the UI.
class DeployResult {
  final bool ok;
  final bool canRevert;
  final String message;
  DeployResult(this.ok, this.canRevert, this.message);
}

/// A remote filesystem entry (for the file browser).
class RemoteEntry {
  final String name;
  final bool isDir;
  final int size;
  final int? modifyTime;
  RemoteEntry(this.name, this.isDir, this.size, this.modifyTime);
}

/// Live transfer stats for an upload or download.
class TransferProgress {
  final int sent;
  final int total;
  final double bytesPerSec;
  final Duration? eta;
  final bool uploading;
  const TransferProgress(
    this.sent,
    this.total,
    this.bytesPerSec,
    this.eta,
    this.uploading,
  );

  double? get fraction => total <= 0 ? null : (sent / total).clamp(0.0, 1.0);
  bool get complete => total > 0 && sent >= total;
}

/// Smooths transfer speed with an exponential moving average and throttles
/// emissions to ~10/s.
class _SpeedMeter {
  final _start = DateTime.now();
  DateTime _lastEmit;
  DateTime _lastT;
  int _lastSent = 0;
  double _bps = 0;
  _SpeedMeter()
    : _lastEmit = DateTime.now().subtract(const Duration(seconds: 1)),
      _lastT = DateTime.now();

  /// Updates speed; returns true if the caller should emit now.
  bool tick(int sent, {bool force = false}) {
    final now = DateTime.now();
    final dt = now.difference(_lastT).inMicroseconds / 1e6;
    if (dt >= 0.15) {
      final inst = (sent - _lastSent) / dt;
      _bps = _bps == 0 ? inst : _bps * 0.6 + inst * 0.4;
      _lastT = now;
      _lastSent = sent;
    } else if (_bps == 0 && sent > 0) {
      final el = now.difference(_start).inMicroseconds / 1e6;
      if (el > 0) _bps = sent / el;
    }
    if (force || now.difference(_lastEmit).inMilliseconds >= 100) {
      _lastEmit = now;
      return true;
    }
    return false;
  }

  double get bps => _bps;
  Duration? eta(int sent, int total) {
    if (_bps <= 0 || total <= 0) return null;
    final remaining = (total - sent) / _bps;
    return Duration(milliseconds: (remaining * 1000).round());
  }
}

/// Drives one SSH session against a server: safe jar deployment, log
/// streaming, and generic SFTP file transfer.
class SshService implements ConnectComms {
  SSHClient? _client;
  String? _connectedServerId;
  bool _rootUser = false; // when logged in as root, skip the sudo wrapper
  SSHSession? _logSession;
  @override
  bool get isConnected => _client != null;

  // Revert state after a deploy that left a backup in place.
  bool _canRevert = false;
  bool _revertIsWeb = false;
  bool _cancelled = false;
  String? _revertTarget;
  String? _revertBackup;
  ServerProfile? _revertServer;
  ServiceConfig? _revertService;
  @override
  bool get canRevert => _canRevert;

  final _logController = StreamController<LogLine>.broadcast();
  @override
  Stream<LogLine> get logs => _logController.stream;

  final _progressController = StreamController<TransferProgress?>.broadcast();
  @override
  Stream<TransferProgress?> get transferProgress => _progressController.stream;

  final steps = StepTracker();
  @override
  Stream<List<DeployStep>> get stepStream => steps.stream;
  String? _uploadStepKey; // step whose progress the SFTP upload feeds

  void _log(LogLevel level, String text) {
    for (final line in const LineSplitter().convert(text)) {
      if (line.trim().isEmpty && level == LogLevel.remote) continue;
      _logController.add(LogLine(level, line));
    }
  }

  Future<void> _connect(ServerProfile s) async {
    // Reuse only a live connection to the same server. A long build between
    // batch deploys can let the server drop an idle connection — detect that
    // (isClosed) and reconnect instead of using a dead client.
    if (_client != null && _connectedServerId == s.id && !_client!.isClosed) {
      return;
    }
    if (_client != null) disconnect();
    _log(LogLevel.step, 'Connecting to ${s.username}@${s.host}:${s.port} …');
    final socket = await SSHSocket.connect(
      s.host,
      s.port,
      timeout: const Duration(seconds: 15),
    );
    final client = SSHClient(
      socket,
      username: s.username,
      onPasswordRequest: () => s.password,
    );
    await client.authenticated;
    _client = client;
    _connectedServerId = s.id;
    _rootUser = s.username.trim().toLowerCase() == 'root';
    _log(
      LogLevel.success,
      'Connected ✓${_rootUser ? ' (root — running without sudo)' : ''}',
    );
  }

  /// Public connect used by the file browser.
  @override
  Future<void> connect(ServerProfile s) => _connect(s);

  /// Runs a command, streaming stdout/stderr to the log. Feeds the sudo
  /// password over stdin when [sudo] is set. Returns the exit code.
  Future<int> _run(
    String command, {
    bool sudo = false,
    String? password,
    LogLevel echo = LogLevel.remote,
  }) async {
    final client = _client!;
    final useSudo = sudo && !_rootUser;
    final cmd = useSudo
        ? "sudo -S -p '' bash -c ${_shQuote(command)}"
        : command;
    final session = await client.execute(cmd);
    if (useSudo && password != null) {
      session.stdin.add(Uint8List.fromList(utf8.encode('$password\n')));
    }
    await session.stdin.close();
    await Future.wait([
      session.stdout
          .cast<List<int>>()
          .transform(utf8.decoder)
          .forEach((d) => _log(echo, d.trimRight())),
      session.stderr
          .cast<List<int>>()
          .transform(utf8.decoder)
          .forEach((d) => _log(echo, d.trimRight())),
    ]);
    await session.done;
    return session.exitCode ?? -1;
  }

  Future<String> _capture(
    String command, {
    bool sudo = false,
    String? password,
  }) async {
    final client = _client!;
    final useSudo = sudo && !_rootUser;
    final cmd = useSudo
        ? "sudo -S -p '' bash -c ${_shQuote(command)}"
        : command;
    final session = await client.execute(cmd);
    if (useSudo && password != null) {
      session.stdin.add(Uint8List.fromList(utf8.encode('$password\n')));
    }
    await session.stdin.close();
    final out = StringBuffer();
    await Future.wait([
      session.stdout
          .cast<List<int>>()
          .transform(utf8.decoder)
          .forEach(out.write),
      session.stderr.cast<List<int>>().transform(utf8.decoder).drain<void>(),
    ]);
    await session.done;
    return out.toString().trim();
  }

  static String _shQuote(String s) => "'${s.replaceAll("'", "'\\''")}'";

  // ---- jar deployment -------------------------------------------------------

  @override
  Future<DeployResult> deploy({
    required ServerProfile server,
    required ServiceConfig service,
    required Uint8List jarBytes,
    required String jarLabel,
  }) async {
    _canRevert = false;
    _revertIsWeb = false;
    _cancelled = false;
    final sudo = server.useSudo;
    final pw = server.password;
    final hasService = service.serviceName.trim().isNotEmpty;
    steps.reset([
      ('connect', 'Connect'),
      ('upload', 'Upload jar'),
      ('stop', 'Stop service'),
      ('backup', 'Back up current'),
      ('install', 'Install jar'),
      ('start', 'Start & verify'),
      ('cleanup', 'Finalize'),
      ('logs', 'Fetch logs'),
    ]);
    try {
      steps.active('connect');
      await _connect(server);
      steps.done('connect');

      final target = service.remotePath(jarLabel);
      final backup = '$target.bak';
      final dir = target.substring(0, target.lastIndexOf('/'));
      final staging =
          '/tmp/peachdeploy-${DateTime.now().millisecondsSinceEpoch}.jar';

      await _run('mkdir -p ${_shQuote(dir)}', sudo: sudo, password: pw);

      // 1. Upload to /tmp first — the live jar is untouched until success.
      steps.active('upload');
      _uploadStepKey = 'upload';
      _log(
        LogLevel.step,
        'Uploading $jarLabel (${_fmtBytes(jarBytes.length)}) → $staging',
      );
      await _sftpUpload(jarBytes, staging);
      _uploadStepKey = null;
      steps.done('upload');
      _log(LogLevel.success, 'Upload complete ✓');

      // 2. Stop the service before swapping.
      steps.active('stop');
      if (hasService) {
        _log(LogLevel.step, 'Stopping service "${service.serviceName}" …');
        await _run(
          'systemctl stop ${_shQuote(service.serviceName)}',
          sudo: sudo,
          password: pw,
        );
        _log(LogLevel.success, 'Service stopped ✓');
        steps.done('stop');
      } else {
        steps.skip('stop');
      }

      // 3. Back up the current jar to <name>.bak.
      steps.active('backup');
      _log(LogLevel.step, 'Backing up current jar → $backup');
      await _run(
        'rm -f ${_shQuote(backup)}; '
        'if [ -f ${_shQuote(target)} ]; then mv ${_shQuote(target)} ${_shQuote(backup)}; '
        'echo "backed up"; else echo "no previous jar"; fi',
        sudo: sudo,
        password: pw,
      );
      steps.done('backup');

      // 4. Install the new jar under the required name.
      steps.active('install');
      _log(LogLevel.step, 'Installing → $target');
      await _run(
        'mv -f ${_shQuote(staging)} ${_shQuote(target)}',
        sudo: sudo,
        password: pw,
      );
      if (service.isGeneric) {
        await _run('chmod +x ${_shQuote(target)}', sudo: sudo, password: pw);
      }
      // Verify the artifact actually landed (SFTP/mv can fail on a dropped
      // connection during a long batch).
      final placed = await _capture(
        'if [ -f ${_shQuote(target)} ]; then echo "OK \$(wc -c < ${_shQuote(target)})"; else echo MISSING; fi',
        sudo: sudo,
        password: pw,
      );
      _log(LogLevel.remote, 'artifact check: $placed');
      if (placed.contains('MISSING')) {
        steps.failRemaining();
        // Restore the previous jar so the service isn't left broken.
        await _run(
          'if [ -f ${_shQuote(backup)} ]; then mv -f ${_shQuote(backup)} ${_shQuote(target)}; fi',
          sudo: sudo,
          password: pw,
        );
        _log(LogLevel.error, 'Artifact not present at $target after install.');
        return DeployResult(
          false,
          false,
          'Install failed — jar not present at $target.',
        );
      }
      steps.done('install');
      _log(LogLevel.success, 'New artifact installed ✓');

      // 5. Start and verify.
      steps.active('start');
      var ok = true;
      if (hasService) {
        _log(LogLevel.step, 'Starting service "${service.serviceName}" …');
        await _run(
          'systemctl start ${_shQuote(service.serviceName)}',
          sudo: sudo,
          password: pw,
        );
        await Future.delayed(const Duration(seconds: 2));
        final state = await _capture(
          'systemctl is-active ${_shQuote(service.serviceName)} || true',
          sudo: sudo,
          password: pw,
        );
        ok = state.contains('active') && !state.contains('inactive');
        if (ok) {
          _log(LogLevel.success, 'Service is active ✓');
        } else {
          _log(LogLevel.error, 'Service is "$state" — start FAILED.');
          await _run(
            'systemctl --no-pager status ${_shQuote(service.serviceName)} | head -n 12',
            sudo: sudo,
            password: pw,
          );
        }
      }
      ok ? steps.done('start') : steps.fail('start');

      steps.active('cleanup');
      if (ok) {
        await _run('rm -f ${_shQuote(backup)}', sudo: sudo, password: pw);
        _log(LogLevel.success, 'Deployment successful 🎉');
        _canRevert = false;
        steps.done('cleanup');
      } else {
        _canRevert = true;
        _revertTarget = target;
        _revertBackup = backup;
        _revertServer = server;
        _revertService = service;
        _log(
          LogLevel.warn,
          'Backup preserved at $backup — you can revert to the previous jar.',
        );
        steps.skip('cleanup');
      }

      steps.active('logs');
      _log(LogLevel.step, 'Fetching recent logs …');
      await _runLogCommand(server, service);
      steps.done('logs');

      return DeployResult(
        ok,
        _canRevert,
        ok
            ? 'Deployed $jarLabel and service is active.'
            : 'Service failed to start.',
      );
    } catch (e) {
      steps.failRemaining();
      if (_cancelled) {
        _log(LogLevel.warn, 'Deployment cancelled by user.');
        return DeployResult(false, _canRevert, 'Deployment cancelled.');
      }
      _log(LogLevel.error, 'Deploy failed: $e');
      return DeployResult(false, _canRevert, 'Error: $e');
    }
  }

  // ---- web (nginx) deployment ----------------------------------------------

  /// Deploys a static site (Angular/React/Flutter web/static) to the nginx web
  /// root. [archiveBytes] is a .tar.gz of the built site's contents.
  @override
  Future<DeployResult> deployWeb({
    required ServerProfile server,
    required ServiceConfig service,
    required Uint8List archiveBytes,
    required int fileCount,
  }) async {
    _canRevert = false;
    _revertIsWeb = true;
    _cancelled = false;
    final sudo = server.useSudo;
    final pw = server.password;
    final root = service.remoteRoot;
    final backup = '$root.bak';
    final nginx = service.nginxService.trim().isEmpty
        ? 'nginx'
        : service.nginxService.trim();
    final ts = DateTime.now().millisecondsSinceEpoch;
    final archive = '/tmp/peachdeploy-web-$ts.tgz';
    final staging = '/tmp/peachweb-$ts'; // extract & verify here first
    steps.reset([
      ('connect', 'Connect'),
      ('upload', 'Upload bundle'),
      ('extract', 'Extract & verify'),
      ('backup', 'Back up web root'),
      ('publish', 'Publish $fileCount files'),
      ('reload', 'Reload nginx'),
      ('cleanup', 'Finalize'),
      ('logs', 'Fetch logs'),
    ]);
    try {
      steps.active('connect');
      await _connect(server);
      steps.done('connect');

      // Ensure the web root (e.g. /var/www/<project>) exists — create it and
      // any parent folders if missing.
      _log(LogLevel.step, 'Ensuring web root $root exists …');
      await _run('mkdir -p ${_shQuote(root)}', sudo: sudo, password: pw);

      // 1. Upload the bundle to /tmp.
      steps.active('upload');
      _uploadStepKey = 'upload';
      _log(
        LogLevel.step,
        'Uploading site bundle (${_fmtBytes(archiveBytes.length)}, $fileCount files) …',
      );
      await _sftpUpload(archiveBytes, archive);
      _uploadStepKey = null;
      steps.done('upload');
      _log(LogLevel.success, 'Upload complete ✓');

      // 2. Extract to a STAGING dir and verify BEFORE touching the live root,
      //    so a bad/empty bundle can never wipe the site.
      steps.active('extract');
      _log(LogLevel.step, 'Extracting bundle to $staging …');
      await _run(
        'rm -rf ${_shQuote(staging)} && mkdir -p ${_shQuote(staging)} && '
        'tar -xzf ${_shQuote(archive)} -C ${_shQuote(staging)}',
        sudo: sudo,
        password: pw,
      );
      final verify = await _capture(
        'if [ -f ${_shQuote('$staging/index.html')} ]; then '
        'echo "OK \$(find ${_shQuote(staging)} -type f | wc -l)"; '
        'else echo "MISSING \$(find ${_shQuote(staging)} -type f | wc -l)"; fi',
        sudo: sudo,
        password: pw,
      );
      _log(LogLevel.remote, 'staging check: $verify');
      if (!verify.contains('OK')) {
        steps.failRemaining();
        await _run(
          'rm -rf ${_shQuote(staging)} ${_shQuote(archive)}',
          sudo: sudo,
          password: pw,
        );
        _log(
          LogLevel.error,
          'No index.html in the bundle — the LIVE site at $root was NOT touched. '
          'Check the build output folder (must be the folder that contains index.html, '
          'e.g. dist/<project>/browser).',
        );
        return DeployResult(
          false,
          false,
          'Bundle has no index.html — nothing published (site untouched).',
        );
      }
      steps.done('extract');

      // 3. Back up the current web root (only now that we know staging is good).
      steps.active('backup');
      _log(LogLevel.step, 'Backing up web root → $backup');
      await _run(
        'mkdir -p ${_shQuote(root)}; rm -rf ${_shQuote(backup)}; '
        'if [ -d ${_shQuote(root)} ] && [ -n "\$(ls -A ${_shQuote(root)} 2>/dev/null)" ]; '
        'then cp -a ${_shQuote(root)} ${_shQuote(backup)}; fi',
        sudo: sudo,
        password: pw,
      );
      steps.done('backup');

      // 4. Atomic-ish swap: clear the root and copy staging contents in.
      steps.active('publish');
      _log(LogLevel.step, 'Publishing new site to $root …');
      final publishCode = await _run(
        'find ${_shQuote(root)} -mindepth 1 -delete && '
        'cp -a ${_shQuote('$staging/.')} ${_shQuote(root)}/ && '
        'rm -rf ${_shQuote(staging)} ${_shQuote(archive)}',
        sudo: sudo,
        password: pw,
      );
      final live = await _capture(
        'if [ -f ${_shQuote('$root/index.html')} ]; then echo "OK \$(find ${_shQuote(root)} -type f | wc -l)"; else echo MISSING; fi',
        sudo: sudo,
        password: pw,
      );
      _log(LogLevel.remote, 'live check: $live');
      if (publishCode != 0 || live.contains('MISSING')) {
        // Restore the backup so the site is never left empty.
        _log(LogLevel.warn, 'Publish issue — restoring previous site …');
        await _run(
          'find ${_shQuote(root)} -mindepth 1 -delete; '
          'if [ -d ${_shQuote(backup)} ]; then cp -a ${_shQuote('$backup/.')} ${_shQuote(root)}/; fi',
          sudo: sudo,
          password: pw,
        );
        steps.failRemaining();
        return DeployResult(
          false,
          false,
          'Publish failed — restored the previous site.',
        );
      }
      steps.done('publish');
      final listing = await _capture(
        'ls -la ${_shQuote(root)} | head -n 25',
        sudo: sudo,
        password: pw,
      );
      _log(LogLevel.success, 'Published to $root:');
      _log(LogLevel.remote, listing);

      // 4. Test config and reload nginx.
      steps.active('reload');
      var ok = true;
      _log(LogLevel.step, 'Testing nginx config & reloading …');
      final test = await _capture(
        'nginx -t 2>&1 || true',
        sudo: sudo,
        password: pw,
      );
      _log(LogLevel.remote, test);
      if (test.contains('test failed') || test.contains('emerg')) {
        ok = false;
        _log(LogLevel.error, 'nginx config test FAILED.');
      } else {
        await _run(
          'systemctl reload ${_shQuote(nginx)}',
          sudo: sudo,
          password: pw,
        );
        await Future.delayed(const Duration(seconds: 1));
        final state = await _capture(
          'systemctl is-active ${_shQuote(nginx)} || true',
          sudo: sudo,
          password: pw,
        );
        ok = state.contains('active') && !state.contains('inactive');
        _log(
          ok ? LogLevel.success : LogLevel.error,
          ok ? 'nginx reloaded ✓' : 'nginx is "$state".',
        );
      }
      ok ? steps.done('reload') : steps.fail('reload');

      steps.active('cleanup');
      if (ok) {
        await _run('rm -rf ${_shQuote(backup)}', sudo: sudo, password: pw);
        _log(LogLevel.success, 'Website deployed 🎉');
        _canRevert = false;
        steps.done('cleanup');
      } else {
        _canRevert = true;
        _revertTarget = root;
        _revertBackup = backup;
        _revertServer = server;
        _revertService = service;
        _log(LogLevel.warn, 'Previous site kept at $backup — you can revert.');
        steps.skip('cleanup');
      }

      steps.active('logs');
      await _runLogCommand(server, service);
      steps.done('logs');

      return DeployResult(
        ok,
        _canRevert,
        ok ? 'Published $fileCount files to $root.' : 'nginx failed to reload.',
      );
    } catch (e) {
      steps.failRemaining();
      if (_cancelled) {
        _log(LogLevel.warn, 'Deployment cancelled by user.');
        return DeployResult(false, _canRevert, 'Deployment cancelled.');
      }
      _log(LogLevel.error, 'Web deploy failed: $e');
      return DeployResult(false, _canRevert, 'Error: $e');
    }
  }

  /// Best-effort cancel of an in-flight deploy: interrupts the current command
  /// by dropping the connection. Steps already applied on the server are not
  /// rolled back automatically.
  @override
  void requestCancel() {
    _cancelled = true;
    _log(LogLevel.warn, 'Cancelling — closing connection …');
    try {
      _client?.close();
    } catch (_) {}
    _client = null;
    _connectedServerId = null;
  }

  /// Restarts a service (or reloads nginx for web apps) without deploying,
  /// then shows recent logs. Returns success.
  @override
  Future<DeployResult> restartOnly(
    ServerProfile server,
    ServiceConfig service,
  ) async {
    final sudo = server.useSudo;
    final pw = server.password;
    try {
      await _connect(server);
      if (service.isWeb) {
        final nginx = service.nginxService.trim().isEmpty
            ? 'nginx'
            : service.nginxService.trim();
        _log(LogLevel.step, 'Reloading $nginx …');
        await _run(
          'systemctl reload ${_shQuote(nginx)}',
          sudo: sudo,
          password: pw,
        );
        final state = await _capture(
          'systemctl is-active ${_shQuote(nginx)} || true',
          sudo: sudo,
          password: pw,
        );
        final ok = state.contains('active') && !state.contains('inactive');
        _log(
          ok ? LogLevel.success : LogLevel.error,
          ok ? 'nginx reloaded ✓' : 'nginx is "$state".',
        );
        await _runLogCommand(server, service);
        return DeployResult(
          ok,
          false,
          ok ? 'nginx reloaded.' : 'Reload failed.',
        );
      }
      final unit = service.serviceName.trim();
      if (unit.isEmpty) {
        return DeployResult(
          false,
          false,
          'No systemd unit set for this service.',
        );
      }
      _log(LogLevel.step, 'Restarting $unit …');
      await _run(
        'systemctl restart ${_shQuote(unit)}',
        sudo: sudo,
        password: pw,
      );
      await Future.delayed(const Duration(seconds: 2));
      final state = await _capture(
        'systemctl is-active ${_shQuote(unit)} || true',
        sudo: sudo,
        password: pw,
      );
      final ok = state.contains('active') && !state.contains('inactive');
      _log(
        ok ? LogLevel.success : LogLevel.error,
        ok ? 'Service is active ✓' : 'Service is "$state".',
      );
      await _runLogCommand(server, service);
      return DeployResult(
        ok,
        false,
        ok ? 'Service restarted.' : 'Restart failed.',
      );
    } catch (e) {
      _log(LogLevel.error, 'Restart failed: $e');
      return DeployResult(false, false, 'Error: $e');
    }
  }

  @override
  Future<DeployResult> revert() async {
    final server = _revertServer;
    final service = _revertService;
    final target = _revertTarget;
    final backup = _revertBackup;
    if (server == null ||
        service == null ||
        target == null ||
        backup == null ||
        !_canRevert) {
      return DeployResult(false, false, 'Nothing to revert.');
    }
    final sudo = server.useSudo;
    final pw = server.password;
    try {
      if (_revertIsWeb) {
        _log(LogLevel.step, 'Reverting web root to previous site …');
        await _run(
          'if [ -d ${_shQuote(backup)} ]; then '
          'find ${_shQuote(target)} -mindepth 1 -delete; '
          'cp -a ${_shQuote('$backup/.')} ${_shQuote(target)}/; '
          'rm -rf ${_shQuote(backup)}; '
          'else echo "no backup to restore"; fi',
          sudo: sudo,
          password: pw,
        );
        final nginx = service.nginxService.trim().isEmpty
            ? 'nginx'
            : service.nginxService.trim();
        await _run(
          'systemctl reload ${_shQuote(nginx)}',
          sudo: sudo,
          password: pw,
        );
        _log(LogLevel.success, 'Reverted to previous site ✓');
      } else {
        _log(LogLevel.step, 'Reverting to previous jar …');
        if (service.serviceName.trim().isNotEmpty) {
          await _run(
            'systemctl stop ${_shQuote(service.serviceName)}',
            sudo: sudo,
            password: pw,
          );
        }
        await _run('rm -f ${_shQuote(target)}', sudo: sudo, password: pw);
        await _run(
          'mv -f ${_shQuote(backup)} ${_shQuote(target)}',
          sudo: sudo,
          password: pw,
        );
        if (service.serviceName.trim().isNotEmpty) {
          await _run(
            'systemctl start ${_shQuote(service.serviceName)}',
            sudo: sudo,
            password: pw,
          );
        }
        _log(LogLevel.success, 'Reverted to previous jar ✓');
      }
      _canRevert = false;
      await _runLogCommand(server, service);
      return DeployResult(true, false, 'Reverted to the previous version.');
    } catch (e) {
      _log(LogLevel.error, 'Revert failed: $e');
      return DeployResult(false, true, 'Revert error: $e');
    }
  }

  @override
  Future<void> discardBackup() async {
    final server = _revertServer;
    final backup = _revertBackup;
    if (server == null || backup == null) return;
    _log(LogLevel.step, 'Discarding backup $backup …');
    final cmd = _revertIsWeb
        ? 'rm -rf ${_shQuote(backup)}'
        : 'rm -f ${_shQuote(backup)}';
    await _run(cmd, sudo: server.useSudo, password: server.password);
    _canRevert = false;
    _log(LogLevel.success, 'Backup discarded.');
  }

  // ---- logs -----------------------------------------------------------------

  Future<void> _runLogCommand(
    ServerProfile server,
    ServiceConfig service,
  ) async {
    await _run(
      service.effectiveLogCommand,
      sudo: server.useSudo,
      password: server.password,
      echo: LogLevel.remote,
    );
  }

  /// Writes and registers a systemd unit file: upload → move to
  /// /etc/systemd/system → daemon-reload → (enable) → (start) → status.
  @override
  Future<DeployResult> installSystemdUnit({
    required ServerProfile server,
    required String unitName,
    required String unitContent,
    bool enable = true,
    bool startNow = true,
  }) async {
    final sudo = server.useSudo;
    final pw = server.password;
    final tmp =
        '/tmp/peachdeploy-$unitName-${DateTime.now().millisecondsSinceEpoch}.service';
    final target = '/etc/systemd/system/$unitName.service';
    steps.reset([
      ('connect', 'Connect'),
      ('upload', 'Upload unit file'),
      ('install', 'Install unit'),
      ('reload', 'daemon-reload'),
      ('enable', 'Enable on boot'),
      ('start', 'Start service'),
      ('status', 'Check status'),
    ]);
    try {
      steps.active('connect');
      await _connect(server);
      steps.done('connect');

      steps.active('upload');
      _log(LogLevel.step, 'Uploading unit file → $tmp');
      await _sftpUpload(Uint8List.fromList(utf8.encode(unitContent)), tmp);
      steps.done('upload');

      steps.active('install');
      _log(LogLevel.step, 'Installing → $target');
      await _run(
        'mv -f ${_shQuote(tmp)} ${_shQuote(target)} && chmod 644 ${_shQuote(target)}',
        sudo: sudo,
        password: pw,
      );
      steps.done('install');

      steps.active('reload');
      await _run('systemctl daemon-reload', sudo: sudo, password: pw);
      steps.done('reload');

      steps.active('enable');
      if (enable) {
        await _run(
          'systemctl enable ${_shQuote(unitName)}',
          sudo: sudo,
          password: pw,
        );
        steps.done('enable');
      } else {
        steps.skip('enable');
      }

      steps.active('start');
      if (startNow) {
        await _run(
          'systemctl restart ${_shQuote(unitName)}',
          sudo: sudo,
          password: pw,
        );
        await Future.delayed(const Duration(seconds: 2));
        steps.done('start');
      } else {
        steps.skip('start');
      }

      steps.active('status');
      final st = await _capture(
        'systemctl is-active ${_shQuote(unitName)} || true',
        sudo: sudo,
        password: pw,
      );
      await _run(
        'systemctl --no-pager status ${_shQuote(unitName)} | head -n 12',
        sudo: sudo,
        password: pw,
      );
      steps.done('status');

      final ok =
          !startNow || (st.contains('active') && !st.contains('inactive'));
      _log(
        ok ? LogLevel.success : LogLevel.warn,
        'Unit $unitName installed${startNow ? ' — is "$st"' : ''}.',
      );
      return DeployResult(
        ok,
        false,
        ok
            ? '$unitName registered${startNow ? ' and active' : ''}.'
            : '$unitName installed but is "$st".',
      );
    } catch (e) {
      steps.failRemaining();
      _log(LogLevel.error, 'Unit install failed: $e');
      return DeployResult(false, false, 'Error: $e');
    }
  }

  // ---- environment (configuration tab) --------------------------------------

  /// Detects the server's package manager: apt | dnf | yum | apk | zypper |
  /// pacman | unknown.
  @override
  Future<String> detectPackageManager(ServerProfile server) async {
    await _connect(server);
    final out = await _capture(
      'command -v apt-get >/dev/null 2>&1 && echo apt; '
      'command -v dnf >/dev/null 2>&1 && echo dnf; '
      'command -v yum >/dev/null 2>&1 && echo yum; '
      'command -v apk >/dev/null 2>&1 && echo apk; '
      'command -v zypper >/dev/null 2>&1 && echo zypper; '
      'command -v pacman >/dev/null 2>&1 && echo pacman',
      sudo: false,
    );
    final lines = out
        .trim()
        .split('\n')
        .where((l) => l.trim().isNotEmpty)
        .toList();
    return lines.isEmpty ? 'unknown' : lines.first.trim();
  }

  /// Runs each detect command; returns key -> version line ('' when missing).
  @override
  Future<Map<String, String>> probeEnvironment(
    ServerProfile server,
    List<String> detectByKey,
  ) async {
    // detectByKey is a flat list of "keycommand" pairs to avoid importing
    // the model here.
    await _connect(server);
    final res = <String, String>{};
    for (final pair in detectByKey) {
      final i = pair.indexOf('');
      final key = pair.substring(0, i);
      final cmd = pair.substring(i + 1);
      try {
        res[key] = (await _capture(cmd, sudo: false)).trim();
      } catch (_) {
        res[key] = '';
      }
    }
    return res;
  }

  /// Installs a set of packages (one combined command) and enables each
  /// listed systemd unit. Used by the new-server setup wizard.
  @override
  Future<DeployResult> installEnvironment({
    required ServerProfile server,
    required String command,
    required List<String> units,
  }) async {
    final sudo = server.useSudo;
    final pw = server.password;
    steps.reset([
      ('connect', 'Connect'),
      ('install', 'Install packages'),
      ('enable', 'Enable services'),
      ('verify', 'Finalize'),
    ]);
    try {
      steps.active('connect');
      await _connect(server);
      steps.done('connect');

      steps.active('install');
      _log(LogLevel.step, '\$ $command');
      final code = await _run(command, sudo: sudo, password: pw);
      if (code != 0) {
        steps.failRemaining();
        return DeployResult(false, false, 'Install failed (exit $code).');
      }
      steps.done('install');

      steps.active('enable');
      for (final u in units.where((u) => u.trim().isNotEmpty)) {
        _log(LogLevel.step, 'Enabling $u …');
        await _run(
          'systemctl enable --now ${_shQuote(u)} || true',
          sudo: sudo,
          password: pw,
        );
      }
      steps.done('enable');

      steps.active('verify');
      steps.done('verify');
      _log(LogLevel.success, 'Environment ready ✓');
      return DeployResult(true, false, 'Environment set up.');
    } catch (e) {
      steps.failRemaining();
      _log(LogLevel.error, 'Setup failed: $e');
      return DeployResult(false, false, 'Error: $e');
    }
  }

  /// Runs a guided install command (already confirmed by the user) and
  /// optionally enables a systemd unit afterwards.
  @override
  Future<DeployResult> installPackages({
    required ServerProfile server,
    required String command,
    String? enableUnit,
  }) async {
    final sudo = server.useSudo;
    final pw = server.password;
    steps.reset([
      ('connect', 'Connect'),
      ('install', 'Install packages'),
      ('enable', 'Enable service'),
      ('verify', 'Finalize'),
    ]);
    try {
      steps.active('connect');
      await _connect(server);
      steps.done('connect');

      steps.active('install');
      _log(LogLevel.step, '\$ $command');
      final code = await _run(command, sudo: sudo, password: pw);
      if (code != 0) {
        steps.failRemaining();
        _log(LogLevel.error, 'Install failed (exit $code).');
        return DeployResult(false, false, 'Install failed (exit $code).');
      }
      steps.done('install');

      steps.active('enable');
      if (enableUnit != null && enableUnit.trim().isNotEmpty) {
        await _run(
          'systemctl enable --now ${_shQuote(enableUnit)} || true',
          sudo: sudo,
          password: pw,
        );
        steps.done('enable');
      } else {
        steps.skip('enable');
      }

      steps.active('verify');
      steps.done('verify');
      _log(LogLevel.success, 'Installation complete ✓');
      return DeployResult(true, false, 'Installed successfully.');
    } catch (e) {
      steps.failRemaining();
      _log(LogLevel.error, 'Install failed: $e');
      return DeployResult(false, false, 'Error: $e');
    }
  }

  /// Probes a service's health per its configured [HealthCheckType]. Quiet:
  /// does not stream to the log console (used by the health dashboard).
  @override
  Future<HealthResult> checkHealth(
    ServerProfile server,
    ServiceConfig service,
  ) async {
    final pw = server.password;
    final sudo = server.useSudo;
    try {
      await _connect(server);
      switch (service.healthType) {
        case HealthCheckType.none:
          return HealthResult(HealthStatus.disabled, 'disabled');
        case HealthCheckType.auto:
          final unit = _unitFor(service);
          if (unit.isEmpty) {
            return HealthResult(HealthStatus.unknown, 'no unit');
          }
          final out = await _capture(
            'systemctl is-active ${_shQuote(unit)} || true',
            sudo: false,
          );
          final s = out.trim().split('\n').last.trim();
          final ok = s.contains('active') && !s.contains('inactive');
          return HealthResult(
            ok ? HealthStatus.healthy : HealthStatus.unhealthy,
            s.isEmpty ? 'unknown' : s,
          );
        case HealthCheckType.http:
          final url = service.healthTarget.trim();
          if (url.isEmpty) {
            return HealthResult(HealthStatus.unknown, 'no URL');
          }
          final out = await _capture(
            'curl -s -o /dev/null -w "%{http_code}" --max-time 6 ${_shQuote(url)} 2>/dev/null || echo 000',
            sudo: false,
          );
          final code = int.tryParse(out.trim().split('\n').last.trim()) ?? 0;
          final ok = code >= 200 && code < 400;
          return HealthResult(
            ok ? HealthStatus.healthy : HealthStatus.unhealthy,
            'HTTP $code',
          );
        case HealthCheckType.tcp:
          final port = service.healthTarget.trim();
          if (port.isEmpty) {
            return HealthResult(HealthStatus.unknown, 'no port');
          }
          final out = await _capture(
            'timeout 4 bash -c "echo > /dev/tcp/127.0.0.1/$port" >/dev/null 2>&1 && echo open || echo closed',
            sudo: false,
          );
          final ok = out.contains('open');
          return HealthResult(
            ok ? HealthStatus.healthy : HealthStatus.unhealthy,
            ok ? 'port $port open' : 'port $port closed',
          );
        case HealthCheckType.command:
          final cmd = service.healthTarget.trim();
          if (cmd.isEmpty) {
            return HealthResult(HealthStatus.unknown, 'no command');
          }
          final out = await _capture(
            '($cmd) >/dev/null 2>&1 && echo OK || echo FAIL',
            sudo: sudo,
            password: pw,
          );
          final ok = out.contains('OK');
          return HealthResult(
            ok ? HealthStatus.healthy : HealthStatus.unhealthy,
            ok ? 'exit 0' : 'non-zero exit',
          );
      }
    } catch (e) {
      return HealthResult(HealthStatus.unknown, 'error: $e');
    }
  }

  // ---- redis viewer ---------------------------------------------------------

  String _redisCli({
    required String host,
    required int port,
    required String pass,
    required int db,
  }) {
    // `timeout 8` guards against a hung connect; --no-auth-warning only when a
    // password is passed (older redis-cli rejects the flag otherwise).
    final b = StringBuffer('timeout 8 redis-cli');
    b.write(' -h ${_shQuote(host.trim().isEmpty ? '127.0.0.1' : host.trim())}');
    b.write(' -p $port');
    if (pass.trim().isNotEmpty) {
      b.write(' -a ${_shQuote(pass.trim())} --no-auth-warning');
    }
    if (db != 0) b.write(' -n $db');
    return b.toString();
  }

  /// (ok, detail) — verifies redis-cli exists and PING returns PONG.
  @override
  Future<(bool, String)> redisPing(
    ServerProfile server, {
    String host = '127.0.0.1',
    int port = 6379,
    String pass = '',
    int db = 0,
  }) async {
    await _connect(server);
    final has = await _capture(
      'command -v redis-cli >/dev/null 2>&1 && echo yes || echo no',
      sudo: false,
    );
    if (has.contains('no')) {
      return (false, 'redis-cli is not installed on the server.');
    }
    final out = await _capture(
      '${_redisCli(host: host, port: port, pass: pass, db: db)} PING 2>&1',
      sudo: false,
    );
    final ok = out.trim().toUpperCase().contains('PONG');
    return (ok, out.trim());
  }

  /// Reads `requirepass` from the server's Redis config (needs root/sudo to
  /// read /etc/redis). Returns '' if none is set / not found.
  @override
  Future<String> redisConfigPassword(ServerProfile server) async {
    await _connect(server);
    final line = await _capture(
      'for f in /etc/redis/redis.conf /etc/redis.conf /etc/redis/redis-server.conf '
      '/etc/redis/6379.conf; do [ -f "\$f" ] && grep -iE "^[[:space:]]*requirepass[[:space:]]" "\$f"; done | tail -n1',
      sudo: server.useSudo,
      password: server.password,
    );
    final m = RegExp(
      r'requirepass\s+(.+)',
      caseSensitive: false,
    ).firstMatch(line.trim());
    if (m == null) return '';
    return m.group(1)!.trim().replaceAll('"', '').replaceAll("'", '');
  }

  /// Total number of keys in the selected DB (`DBSIZE`).
  @override
  Future<int> redisDbSize(
    ServerProfile server, {
    String host = '127.0.0.1',
    int port = 6379,
    String pass = '',
    int db = 0,
  }) async {
    await _connect(server);
    final out = await _capture(
      '${_redisCli(host: host, port: port, pass: pass, db: db)} DBSIZE 2>&1',
      sudo: false,
    );
    return int.tryParse(
          RegExp(r'\d+').firstMatch(out.trim())?.group(0) ?? '',
        ) ??
        0;
  }

  /// Returns (number of DBs, {dbIndex: keyCount}) from CONFIG + INFO keyspace.
  @override
  Future<(int, Map<int, int>)> redisDatabases(
    ServerProfile server, {
    String host = '127.0.0.1',
    int port = 6379,
    String pass = '',
  }) async {
    await _connect(server);
    final cli = _redisCli(host: host, port: port, pass: pass, db: 0);
    final ks = await _capture('$cli INFO keyspace 2>&1', sudo: false);
    final counts = <int, int>{};
    for (final line in ks.split('\n')) {
      final m = RegExp(r'^db(\d+):keys=(\d+)').firstMatch(line.trim());
      if (m != null) counts[int.parse(m.group(1)!)] = int.parse(m.group(2)!);
    }
    final cfg = await _capture('$cli CONFIG GET databases 2>&1', sudo: false);
    final nums = RegExp(
      r'\d+',
    ).allMatches(cfg).map((m) => m.group(0)!).toList();
    final dbCount = nums.isEmpty ? 16 : int.tryParse(nums.last) ?? 16;
    return (dbCount, counts);
  }

  /// Scans keys returning (key, type) pairs, optionally filtered to
  /// [filterType]. Uses a server-side Lua SCAN+TYPE (one round trip); scan is
  /// capped at [limit] keys. Returns [] on error (caller falls back).
  @override
  Future<List<(String, String)>> redisKeysTyped(
    ServerProfile server, {
    String pattern = '*',
    String filterType = '',
    int limit = 50000,
    String host = '127.0.0.1',
    int port = 6379,
    String pass = '',
    int db = 0,
  }) async {
    await _connect(server);
    final cli = _redisCli(host: host, port: port, pass: pass, db: db);
    const script =
        'local cur="0" local out={} local n=0 repeat local r=redis.call("SCAN",cur,"MATCH",ARGV[1],"COUNT",500) cur=r[1] for i=1,#r[2] do local k=r[2][i] local t=redis.call("TYPE",k).ok if ARGV[3]=="" or t==ARGV[3] then out[#out+1]=k out[#out+1]=t end n=n+1 if n>=tonumber(ARGV[2]) then cur="0" break end end until cur=="0" return out';
    final out = await _capture(
      "$cli EVAL '$script' 0 ${_shQuote(pattern)} ${_shQuote(limit.toString())} ${_shQuote(filterType)} 2>&1",
      sudo: false,
    );
    final lines = out
        .split('\n')
        .map((l) => l.trim())
        .where((l) => l.isNotEmpty)
        .toList();
    if (lines.isEmpty ||
        out.contains('(error)') ||
        out.toUpperCase().contains('ERR ') ||
        out.contains('NOAUTH')) {
      return [];
    }
    final result = <(String, String)>[];
    for (var i = 0; i + 1 < lines.length; i += 2) {
      result.add((lines[i], lines[i + 1]));
    }
    return result;
  }

  /// Lists keys matching [pattern] (SCAN-based; capped at [limit] for safety).
  @override
  Future<List<String>> redisKeys(
    ServerProfile server, {
    String pattern = '*',
    int limit = 100000,
    String host = '127.0.0.1',
    int port = 6379,
    String pass = '',
    int db = 0,
  }) async {
    await _connect(server);
    final cli = _redisCli(host: host, port: port, pass: pass, db: db);
    final out = await _capture(
      '$cli --scan --pattern ${_shQuote(pattern)} 2>/dev/null | head -n $limit',
      sudo: false,
    );
    return out
        .split('\n')
        .map((l) => l.trim())
        .where((l) => l.isNotEmpty)
        .toList();
  }

  /// Reads a key's type, TTL and value (formatted per type).
  @override
  Future<RedisEntry> redisGet(
    ServerProfile server,
    String key, {
    String host = '127.0.0.1',
    int port = 6379,
    String pass = '',
    int db = 0,
  }) async {
    await _connect(server);
    final cli = _redisCli(host: host, port: port, pass: pass, db: db);
    final k = _shQuote(key);
    final type = (await _capture('$cli TYPE $k 2>&1', sudo: false)).trim();
    final ttl = (await _capture('$cli TTL $k 2>&1', sudo: false)).trim();
    final readCmd = switch (type) {
      'list' => '$cli LRANGE $k 0 -1',
      'set' => '$cli SMEMBERS $k',
      'hash' => '$cli HGETALL $k',
      'zset' => '$cli ZRANGE $k 0 -1 WITHSCORES',
      'stream' => '$cli XRANGE $k - +',
      'string' => '$cli GET $k',
      _ => '$cli GET $k',
    };
    final value = (await _capture('$readCmd 2>&1', sudo: false)).trim();
    return RedisEntry(key: key, type: type, ttl: ttl, value: value);
  }

  // ---- docker ---------------------------------------------------------------

  @override
  Future<DockerStatus> dockerStatus(ServerProfile server) async {
    await _connect(server);
    final has = await _capture(
      'command -v docker >/dev/null 2>&1 && echo yes || echo no',
      sudo: false,
    );
    if (has.contains('no')) return DockerStatus(false, false, '');
    final ver = (await _capture(
      'docker --version 2>/dev/null',
      sudo: false,
    )).trim();
    final info = await _capture(
      'docker info >/dev/null 2>&1 && echo up || echo down',
      sudo: server.useSudo,
      password: server.password,
    );
    return DockerStatus(true, info.contains('up'), ver);
  }

  static const _dockerSep = '~|~';

  @override
  Future<List<DockerContainer>> dockerPs(
    ServerProfile server, {
    bool all = true,
  }) async {
    await _connect(server);
    final fmt =
        '{{.ID}}$_dockerSep{{.Names}}$_dockerSep{{.Image}}$_dockerSep{{.State}}$_dockerSep{{.Status}}$_dockerSep{{.Ports}}';
    final out = await _capture(
      "docker ps ${all ? '-a' : ''} --format ${_shQuote(fmt)} 2>&1",
      sudo: server.useSudo,
      password: server.password,
    );
    final list = <DockerContainer>[];
    for (final line in out.split('\n')) {
      final l = line.trim();
      if (l.isEmpty) continue;
      final parts = l.split(_dockerSep);
      if (parts.length < 6) continue; // skip errors/noise
      list.add(
        DockerContainer(
          id: parts[0],
          name: parts[1],
          image: parts[2],
          state: parts[3],
          status: parts[4],
          ports: parts[5],
        ),
      );
    }
    return list;
  }

  /// action: start | stop | restart | rm
  @override
  Future<String> dockerAction(
    ServerProfile server,
    String id,
    String action,
  ) async {
    await _connect(server);
    final cmd = action == 'rm'
        ? 'docker rm -f ${_shQuote(id)}'
        : 'docker $action ${_shQuote(id)}';
    return (await _capture(
      '$cmd 2>&1',
      sudo: server.useSudo,
      password: server.password,
    )).trim();
  }

  @override
  Future<String> dockerLogs(
    ServerProfile server,
    String id, {
    int tail = 300,
  }) async {
    await _connect(server);
    return (await _capture(
      'docker logs --tail $tail ${_shQuote(id)} 2>&1',
      sudo: server.useSudo,
      password: server.password,
    )).trim();
  }

  @override
  Future<String> dockerTest(ServerProfile server) async {
    await _connect(server);
    return (await _capture(
      'docker run --rm hello-world 2>&1',
      sudo: server.useSudo,
      password: server.password,
    )).trim();
  }

  /// Builds and runs `docker run`. Returns (ok, output).
  @override
  Future<(bool, String)> dockerRun(
    ServerProfile server,
    DockerRunSpec spec,
  ) async {
    await _connect(server);
    final b = StringBuffer('docker run');
    b.write(spec.detached ? ' -d' : ' --rm');
    if (spec.name.trim().isNotEmpty) {
      b.write(' --name ${_shQuote(spec.name.trim())}');
    }
    if (spec.restart.trim().isNotEmpty && spec.detached) {
      b.write(' --restart ${_shQuote(spec.restart.trim())}');
    }
    for (final p in spec.portList) {
      b.write(' -p ${_shQuote(p)}');
    }
    for (final e in spec.envList) {
      b.write(' -e ${_shQuote(e)}');
    }
    for (final v in spec.volumeList) {
      b.write(' -v ${_shQuote(v)}');
    }
    b.write(' ${_shQuote(spec.image.trim())}');
    if (spec.command.trim().isNotEmpty) b.write(' ${spec.command.trim()}');

    _log(LogLevel.step, '\$ ${b.toString()}');
    final out = await _capture(
      '${b.toString()} 2>&1',
      sudo: server.useSudo,
      password: server.password,
    );
    final t = out.trim();
    // Success: docker prints the (64-hex) container id; errors contain text.
    final ok =
        RegExp(
          r'^[0-9a-f]{12,64}$',
          multiLine: true,
        ).hasMatch(t.split('\n').last.trim()) &&
        !t.toLowerCase().contains('error');
    _log(ok ? LogLevel.success : LogLevel.error, t);
    return (ok, t);
  }

  // ---- kafka monitor --------------------------------------------------------

  /// Full path to a Kafka CLI script, single-quoted for the shell.
  String _kbin(String binDir, String suffix, String name) =>
      _shQuote('$binDir/$name$suffix');

  List<int> _ints(String s) => s
      .split(',')
      .map((e) => e.trim())
      .where((e) => e.isNotEmpty)
      .map((e) => int.tryParse(e) ?? -1)
      .where((e) => e >= 0)
      .toList();

  /// Locates the Kafka tooling (bin dir + script suffix) on the server.
  @override
  Future<KafkaStatus> kafkaDetect(ServerProfile server) async {
    await _connect(server);
    final java = await _capture(
      'command -v java >/dev/null 2>&1 && echo yes || echo no',
      sudo: false,
    );
    // Prefer PATH, then fall back to the usual install locations.
    const finder =
        'p=""; '
        'for c in kafka-topics.sh kafka-topics; do '
        'if command -v "\$c" >/dev/null 2>&1; then p=\$(command -v "\$c"); break; fi; done; '
        'if [ -z "\$p" ]; then '
        'for d in /opt/kafka/bin /usr/local/kafka/bin /opt/bitnami/kafka/bin '
        '/home/kafka/kafka/bin /usr/share/kafka/bin /opt/confluent/bin '
        '/usr/lib/kafka/bin \$(ls -d /opt/kafka*/bin 2>/dev/null) '
        '\$(ls -d /opt/kafka_*/bin 2>/dev/null) \$(ls -d /usr/local/kafka*/bin 2>/dev/null); do '
        'for s in kafka-topics.sh kafka-topics; do '
        'if [ -x "\$d/\$s" ]; then p="\$d/\$s"; break 2; fi; done; done; fi; '
        'echo "\$p"';
    final lines = (await _capture(
      finder,
      sudo: false,
    )).split('\n').map((l) => l.trim()).where((l) => l.isNotEmpty).toList();
    final path = lines.isEmpty ? '' : lines.last;
    if (path.isEmpty) {
      return KafkaStatus(
        detected: false,
        java: java.contains('yes'),
        detail:
            'kafka-topics not found on PATH or in the common install directories '
            '(/opt/kafka, /usr/local/kafka, /opt/bitnami/kafka …).',
      );
    }
    final slash = path.lastIndexOf('/');
    final binDir = slash >= 0 ? path.substring(0, slash) : '.';
    final base = slash >= 0 ? path.substring(slash + 1) : path;
    final suffix = base.endsWith('.sh') ? '.sh' : '';
    final ver = (await _capture(
      'timeout 15 ${_shQuote(path)} --version 2>/dev/null | head -n1',
      sudo: false,
    )).trim();
    return KafkaStatus(
      detected: true,
      binDir: binDir,
      suffix: suffix,
      java: java.contains('yes'),
      version: ver,
      detail: 'Found $path',
    );
  }

  /// Connectivity probe — returns the broker-api-versions output (or the error).
  @override
  Future<String> kafkaTest(
    ServerProfile server, {
    required String binDir,
    required String suffix,
    required String bootstrap,
  }) async {
    await _connect(server);
    final out = await _capture(
      'timeout 25 ${_kbin(binDir, suffix, 'kafka-broker-api-versions')} '
      '--bootstrap-server ${_shQuote(bootstrap)} 2>&1 | head -n 30',
      sudo: false,
    );
    return out.trim();
  }

  /// topic:partition -> offset, for `--time` -1 (latest) or -2 (earliest).
  Future<Map<String, int>> _kafkaOffsets(
    String binDir,
    String suffix,
    String bootstrap,
    int time,
  ) async {
    var out = await _capture(
      'timeout 25 ${_kbin(binDir, suffix, 'kafka-get-offsets')} '
      '--bootstrap-server ${_shQuote(bootstrap)} --time $time 2>/dev/null',
      sudo: false,
    );
    if (out.trim().isEmpty) {
      // Older Kafka: fall back to the GetOffsetShell tool.
      out = await _capture(
        'timeout 25 ${_kbin(binDir, suffix, 'kafka-run-class')} '
        'kafka.tools.GetOffsetShell --broker-list ${_shQuote(bootstrap)} '
        '--time $time 2>/dev/null',
        sudo: false,
      );
    }
    final map = <String, int>{};
    for (final raw in out.split('\n')) {
      final m = RegExp(r'^([^:]+):(\d+):(-?\d+)$').firstMatch(raw.trim());
      if (m != null) {
        map['${m.group(1)}:${m.group(2)}'] = int.tryParse(m.group(3)!) ?? 0;
      }
    }
    return map;
  }

  /// Lists every topic with its partitions, replication and message offsets.
  @override
  Future<List<KafkaTopic>> kafkaTopics(
    ServerProfile server, {
    required String binDir,
    required String suffix,
    required String bootstrap,
    bool includeInternal = false,
  }) async {
    await _connect(server);
    final describe = await _capture(
      'timeout 25 ${_kbin(binDir, suffix, 'kafka-topics')} '
      '--bootstrap-server ${_shQuote(bootstrap)} --describe 2>/dev/null',
      sudo: false,
    );

    final topics = <KafkaTopic>[];
    List<KafkaPartition> curParts = [];
    for (final raw in describe.split('\n')) {
      final line = raw.trim();
      if (line.isEmpty) continue;
      if (line.contains('PartitionCount:')) {
        final name = RegExp(r'Topic:\s*(\S+)').firstMatch(line)?.group(1) ?? '';
        final rf =
            int.tryParse(
              RegExp(
                    r'ReplicationFactor:\s*(\d+)',
                  ).firstMatch(line)?.group(1) ??
                  '',
            ) ??
            0;
        final cfg =
            RegExp(r'Configs:\s*(.*)').firstMatch(line)?.group(1)?.trim() ?? '';
        curParts = [];
        topics.add(
          KafkaTopic(
            name: name,
            replicationFactor: rf,
            partitions: curParts,
            configs: cfg,
          ),
        );
      } else if (line.contains('Partition:') && line.contains('Leader:')) {
        final pid =
            int.tryParse(
              RegExp(r'Partition:\s*(\d+)').firstMatch(line)?.group(1) ?? '',
            ) ??
            0;
        final leader =
            int.tryParse(
              RegExp(r'Leader:\s*(-?\d+)').firstMatch(line)?.group(1) ?? '',
            ) ??
            -1;
        final replicas = _ints(
          RegExp(r'Replicas:\s*([0-9,]*)').firstMatch(line)?.group(1) ?? '',
        );
        final isr = _ints(
          RegExp(r'Isr:\s*([0-9,]*)').firstMatch(line)?.group(1) ?? '',
        );
        curParts.add(
          KafkaPartition(id: pid, leader: leader, replicas: replicas, isr: isr),
        );
      }
    }

    // Overlay message offsets (best effort — topics still render without them).
    final latest = await _kafkaOffsets(binDir, suffix, bootstrap, -1);
    final earliest = await _kafkaOffsets(binDir, suffix, bootstrap, -2);
    for (final t in topics) {
      for (final p in t.partitions) {
        p.latest = latest['${t.name}:${p.id}'] ?? p.latest;
        p.earliest = earliest['${t.name}:${p.id}'] ?? p.earliest;
      }
    }

    final result = includeInternal
        ? topics
        : topics.where((t) => !t.internal).toList();
    result.sort((a, b) => a.name.compareTo(b.name));
    return result;
  }

  /// Splits a Kafka CLI table row on runs of 2+ spaces (columns are padded).
  List<String> _cols(String line) => line
      .split(RegExp(r'\s{2,}'))
      .map((e) => e.trim())
      .where((e) => e.isNotEmpty)
      .toList();

  /// Lists consumer groups with state, members and per-partition lag.
  @override
  Future<List<KafkaConsumerGroup>> kafkaGroups(
    ServerProfile server, {
    required String binDir,
    required String suffix,
    required String bootstrap,
  }) async {
    await _connect(server);
    final cg = _kbin(binDir, suffix, 'kafka-consumer-groups');
    final bs = _shQuote(bootstrap);

    final groups = <String, KafkaConsumerGroup>{};

    // 1) All group ids (so idle/empty groups still appear).
    final listOut = await _capture(
      'timeout 25 $cg --bootstrap-server $bs --list 2>/dev/null',
      sudo: false,
    );
    for (final raw in listOut.split('\n')) {
      final id = raw.trim();
      if (id.isEmpty || id.contains(' ')) continue;
      groups[id] = KafkaConsumerGroup(groupId: id);
    }

    // 2) State / members / coordinator per group.
    final stateOut = await _capture(
      'timeout 30 $cg --bootstrap-server $bs --describe --all-groups --state 2>/dev/null',
      sudo: false,
    );
    for (final raw in stateOut.split('\n')) {
      final line = raw.trim();
      if (line.isEmpty || line.startsWith('GROUP')) continue;
      final c = _cols(line);
      if (c.length < 4) continue;
      final members = int.tryParse(c.last);
      if (members == null) continue; // not a data row
      final state = c[c.length - 2];
      final id = c[0];
      final g = groups.putIfAbsent(id, () => KafkaConsumerGroup(groupId: id));
      g.state = state;
      g.members = members;
    }

    // 3) Offsets / lag per topic-partition.
    final offOut = await _capture(
      'timeout 40 $cg --bootstrap-server $bs --describe --all-groups --offsets 2>/dev/null',
      sudo: false,
    );
    for (final raw in offOut.split('\n')) {
      final line = raw.trim();
      if (line.isEmpty || line.startsWith('GROUP')) continue;
      final c = _cols(line);
      if (c.length < 6) continue;
      final partition = int.tryParse(c[2]);
      if (partition == null) continue; // header/noise
      int num(String s) => int.tryParse(s.trim()) ?? 0;
      final g = groups.putIfAbsent(
        c[0],
        () => KafkaConsumerGroup(groupId: c[0]),
      );
      g.assignments.add(
        KafkaGroupAssignment(
          topic: c[1],
          partition: partition,
          currentOffset: num(c[3]),
          logEndOffset: num(c[4]),
          lag: num(c[5]),
          consumerId: c.length > 6 ? c[6] : '',
          host: c.length > 7 ? c[7] : '',
          clientId: c.length > 8 ? c[8] : '',
        ),
      );
    }

    final list = groups.values.toList()
      ..sort((a, b) => a.groupId.compareTo(b.groupId));
    return list;
  }

  /// Cluster snapshot: brokers, cluster id, controller and health counts.
  @override
  Future<KafkaCluster> kafkaCluster(
    ServerProfile server, {
    required String binDir,
    required String suffix,
    required String bootstrap,
  }) async {
    await _connect(server);
    final bs = _shQuote(bootstrap);

    // Brokers via broker-api-versions (one broker per header line).
    final apiOut = await _capture(
      'timeout 25 ${_kbin(binDir, suffix, 'kafka-broker-api-versions')} '
      '--bootstrap-server $bs 2>/dev/null',
      sudo: false,
    );
    final brokers = <KafkaBroker>[];
    for (final raw in apiOut.split('\n')) {
      final m = RegExp(
        r'^(\S+):(\d+)\s+\(id:\s*(\d+)\s+rack:\s*(\S+)\)',
      ).firstMatch(raw.trim());
      if (m != null) {
        final rack = m.group(4)!;
        brokers.add(
          KafkaBroker(
            id: int.parse(m.group(3)!),
            host: m.group(1)!,
            port: int.parse(m.group(2)!),
            rack: rack == 'null' ? '' : rack,
          ),
        );
      }
    }
    brokers.sort((a, b) => a.id.compareTo(b.id));

    // Cluster id (best effort).
    String clusterId = '';
    final cid = await _capture(
      'timeout 20 ${_kbin(binDir, suffix, 'kafka-cluster')} cluster-id '
      '--bootstrap-server $bs 2>/dev/null',
      sudo: false,
    );
    clusterId =
        RegExp(r'Cluster ID:\s*(\S+)').firstMatch(cid)?.group(1)?.trim() ?? '';

    // Controller / quorum leader (KRaft) — best effort.
    int controllerId = -1;
    final quorum = await _capture(
      'timeout 20 ${_kbin(binDir, suffix, 'kafka-metadata-quorum')} '
      '--bootstrap-server $bs describe --status 2>/dev/null',
      sudo: false,
    );
    controllerId =
        int.tryParse(
          RegExp(r'LeaderId:\s*(\d+)').firstMatch(quorum)?.group(1) ?? '',
        ) ??
        -1;
    for (final b in brokers) {
      if (b.id == controllerId) b.isController = true;
    }

    // Health: under-replicated & unavailable partitions.
    List<String> parsePartitions(String out) {
      final res = <String>[];
      for (final raw in out.split('\n')) {
        final line = raw.trim();
        final t = RegExp(r'Topic:\s*(\S+)').firstMatch(line)?.group(1);
        final p = RegExp(r'Partition:\s*(\d+)').firstMatch(line)?.group(1);
        if (t != null && p != null) res.add('$t-$p');
      }
      return res;
    }

    final ur = await _capture(
      'timeout 25 ${_kbin(binDir, suffix, 'kafka-topics')} '
      '--bootstrap-server $bs --describe --under-replicated-partitions 2>/dev/null',
      sudo: false,
    );
    final un = await _capture(
      'timeout 25 ${_kbin(binDir, suffix, 'kafka-topics')} '
      '--bootstrap-server $bs --describe --unavailable-partitions 2>/dev/null',
      sudo: false,
    );

    return KafkaCluster(
      brokers: brokers,
      clusterId: clusterId,
      controllerId: controllerId,
      underReplicated: parsePartitions(ur),
      unavailable: parsePartitions(un),
    );
  }

  String _unitFor(ServiceConfig s) => s.isWeb
      ? (s.nginxService.trim().isEmpty ? 'nginx' : s.nginxService.trim())
      : s.serviceName.trim();

  static String _fmtDt(DateTime d) {
    String two(int n) => n.toString().padLeft(2, '0');
    return '${d.year}-${two(d.month)}-${two(d.day)} '
        '${two(d.hour)}:${two(d.minute)}:${two(d.second)}';
  }

  /// Public one-shot log fetch (connects if needed). Optionally restricts to a
  /// time window via journalctl `--since` / `--until` (when using the default
  /// journalctl command).
  @override
  Future<void> fetchLogs(
    ServerProfile server,
    ServiceConfig service, {
    DateTime? since,
    DateTime? until,
  }) async {
    await _connect(server);
    _log(LogLevel.step, 'Fetching logs for ${service.name} …');
    final hasRange = since != null || until != null;
    if (hasRange && service.logCommand.trim().isEmpty) {
      final unit = _unitFor(service).isEmpty ? 'app' : _unitFor(service);
      final b = StringBuffer(
        'journalctl -u ${_shQuote(unit)} -n 2000 --no-pager',
      );
      if (since != null) b.write(' --since ${_shQuote(_fmtDt(since))}');
      if (until != null) b.write(' --until ${_shQuote(_fmtDt(until))}');
      await _run(
        b.toString(),
        sudo: server.useSudo,
        password: server.password,
        echo: LogLevel.remote,
      );
    } else {
      await _runLogCommand(server, service);
    }
  }

  /// The distinct `root` directives from the live nginx config (`nginx -T`),
  /// so the deploy screen can warn if the web root doesn't match what nginx
  /// actually serves. Returns [] if it can't be read.
  @override
  Future<List<String>> nginxRoots(ServerProfile server) async {
    try {
      await _connect(server);
      final out = await _capture(
        "nginx -T 2>/dev/null | grep -E '^[[:space:]]*root[[:space:]]' | "
        "sed -E 's/^[[:space:]]*root[[:space:]]+//; s/;.*//; s/[[:space:]]+\$//' | sort -u",
        sudo: server.useSudo,
        password: server.password,
      );
      return out
          .split('\n')
          .map((l) => l.trim())
          .where((l) => l.isNotEmpty)
          .toList();
    } catch (_) {
      return [];
    }
  }

  /// Current systemd state of the service's unit ('active', 'inactive',
  /// 'failed', 'activating', 'unknown', or 'no-unit').
  @override
  Future<String> serviceStatus(
    ServerProfile server,
    ServiceConfig service,
  ) async {
    await _connect(server);
    final unit = _unitFor(service);
    if (unit.isEmpty) return 'no-unit';
    final out = await _capture(
      'systemctl is-active ${_shQuote(unit)} || true',
      sudo: server.useSudo,
      password: server.password,
    );
    final s = out.trim().split('\n').last.trim();
    return s.isEmpty ? 'unknown' : s;
  }

  /// start | stop | restart the service's unit (reload for nginx web apps).
  @override
  Future<DeployResult> serviceControl(
    ServerProfile server,
    ServiceConfig service,
    String action,
  ) async {
    final unit = _unitFor(service);
    if (unit.isEmpty) {
      return DeployResult(
        false,
        false,
        'No systemd unit set for this service.',
      );
    }
    final sudo = server.useSudo;
    final pw = server.password;
    // nginx should reload, not restart, on "restart".
    final effective = (service.isWeb && action == 'restart')
        ? 'reload'
        : action;
    try {
      await _connect(server);
      _log(LogLevel.step, 'systemctl $effective $unit …');
      await _run(
        'systemctl $effective ${_shQuote(unit)}',
        sudo: sudo,
        password: pw,
      );
      await Future.delayed(const Duration(seconds: 2));
      final st = await serviceStatus(server, service);
      final ok = action == 'stop'
          ? (st.contains('inactive') ||
                st.contains('dead') ||
                st.contains('failed'))
          : (st.contains('active') && !st.contains('inactive'));
      _log(ok ? LogLevel.success : LogLevel.warn, '$unit is now "$st".');
      await _runLogCommand(server, service);
      return DeployResult(ok, false, '$unit is $st.');
    } catch (e) {
      _log(LogLevel.error, 'Control failed: $e');
      return DeployResult(false, false, 'Error: $e');
    }
  }

  @override
  Future<void> followLogs(ServerProfile server, ServiceConfig service) async {
    await _connect(server);
    stopLogs();
    final unit = service.isWeb
        ? (service.nginxService.isEmpty ? 'nginx' : service.nginxService)
        : (service.serviceName.isEmpty ? 'app' : service.serviceName);
    final base = 'journalctl -u ${_shQuote(unit)} -n 50 -f';
    final useSudo = server.useSudo && !_rootUser;
    final cmd = useSudo ? "sudo -S -p '' $base" : base;
    _log(LogLevel.step, 'Live tailing logs (press stop to end) …');
    final session = await _client!.execute(cmd);
    _logSession = session;
    if (useSudo) {
      session.stdin.add(
        Uint8List.fromList(utf8.encode('${server.password}\n')),
      );
    }
    await session.stdin.close();
    unawaited(
      session.stdout
          .cast<List<int>>()
          .transform(utf8.decoder)
          .forEach((d) => _log(LogLevel.remote, d.trimRight())),
    );
    unawaited(
      session.stderr
          .cast<List<int>>()
          .transform(utf8.decoder)
          .forEach((d) => _log(LogLevel.remote, d.trimRight())),
    );
  }

  @override
  void stopLogs() {
    _logSession?.close();
    _logSession = null;
  }

  // ---- generic SFTP file transfer ------------------------------------------

  @override
  Future<List<RemoteEntry>> listDir(ServerProfile server, String path) async {
    await _connect(server);
    final sftp = await _client!.sftp();
    final names = await sftp.listdir(path.isEmpty ? '.' : path);
    final entries = <RemoteEntry>[];
    for (final n in names) {
      if (n.filename == '.' || n.filename == '..') continue;
      entries.add(
        RemoteEntry(
          n.filename,
          n.attr.isDirectory,
          n.attr.size ?? 0,
          n.attr.modifyTime,
        ),
      );
    }
    entries.sort((a, b) {
      if (a.isDir != b.isDir) return a.isDir ? -1 : 1;
      return a.name.toLowerCase().compareTo(b.name.toLowerCase());
    });
    return entries;
  }

  @override
  Future<String> resolveHome(ServerProfile server) async {
    await _connect(server);
    final sftp = await _client!.sftp();
    try {
      return await sftp.absolute('.');
    } catch (_) {
      return '.';
    }
  }

  @override
  Future<void> uploadFile(
    ServerProfile server,
    Uint8List bytes,
    String remotePath,
  ) async {
    await _connect(server);
    await _sftpUpload(bytes, remotePath);
  }

  @override
  Future<Uint8List> downloadFile(
    ServerProfile server,
    String remotePath,
  ) async {
    await _connect(server);
    final sftp = await _client!.sftp();
    final file = await sftp.open(remotePath);
    final attrs = await file.stat();
    final total = attrs.size ?? 0;
    final builder = BytesBuilder(copy: false);
    final meter = _SpeedMeter();
    await for (final chunk in file.read()) {
      builder.add(chunk);
      if (meter.tick(builder.length)) {
        _emitTransfer(builder.length, total, meter, uploading: false);
      }
    }
    await file.close();
    _emitTransfer(
      total == 0 ? builder.length : total,
      total == 0 ? builder.length : total,
      meter,
      uploading: false,
      force: true,
    );
    return builder.toBytes();
  }

  void _emitTransfer(
    int sent,
    int total,
    _SpeedMeter meter, {
    required bool uploading,
    bool force = false,
  }) {
    final tp = TransferProgress(
      sent,
      total,
      meter.bps,
      meter.eta(sent, total),
      uploading,
    );
    _progressController.add(tp);
    if (_uploadStepKey != null) {
      steps.progress(_uploadStepKey!, tp.fraction);
    }
  }

  @override
  Future<void> deleteRemote(
    ServerProfile server,
    String remotePath,
    bool isDir,
  ) async {
    await _connect(server);
    final sftp = await _client!.sftp();
    if (isDir) {
      await sftp.rmdir(remotePath);
    } else {
      await sftp.remove(remotePath);
    }
  }

  Future<void> _sftpUpload(Uint8List bytes, String remotePath) async {
    final sftp = await _client!.sftp();
    final file = await sftp.open(
      remotePath,
      mode:
          SftpFileOpenMode.create |
          SftpFileOpenMode.write |
          SftpFileOpenMode.truncate,
    );
    final total = bytes.length;
    var sent = 0;
    const chunk = 32 * 1024;
    final meter = _SpeedMeter();
    final stream =
        Stream<Uint8List>.fromIterable([
          for (var i = 0; i < total; i += chunk)
            Uint8List.sublistView(
              bytes,
              i,
              i + chunk > total ? total : i + chunk,
            ),
        ]).map((c) {
          sent += c.length;
          if (meter.tick(sent)) {
            _emitTransfer(sent, total, meter, uploading: true);
          }
          return c;
        });
    await file.write(stream);
    await file.close();
    _emitTransfer(total, total, meter, uploading: true, force: true);
  }

  // ---- lifecycle ------------------------------------------------------------

  @override
  void disconnect() {
    stopLogs();
    _client?.close();
    _client = null;
    _connectedServerId = null;
    _log(LogLevel.info, 'Disconnected.');
  }

  @override
  void dispose() {
    stopLogs();
    _client?.close();
    _client = null;
    _connectedServerId = null;
    _logController.close();
    _progressController.close();
    steps.dispose();
  }

  static String _fmtBytes(int b) {
    if (b < 1024) return '$b B';
    if (b < 1024 * 1024) return '${(b / 1024).toStringAsFixed(1)} KB';
    return '${(b / (1024 * 1024)).toStringAsFixed(1)} MB';
  }
}