import '../../about/internal/application/ServerProfile.dart';
import '../parent/ParentViewModel.dart';
import 'ConnectKafka.dart';

class ViewKafka extends ParentViewModel {
  ConnectKafka connection;

  ViewKafka(super.context, this.connection);

  /// Looks for the Kafka CLI tools on PATH and in the usual install dirs.
  void detectKafka(ServerProfile server) async {
    if (!await canReach(server, () => detectKafka(server))) return;
    try {
      final status = await getDataManager().kafkaDetect(server);
      connection.onKafkaDetected(status);
    } catch (error) {
      connection.onDetectFailed(error.toString());
    }
  }

  /// Reads topics, consumer groups and cluster health in one pass. They're
  /// three separate CLI invocations, so this is where they're sequenced.
  void scan(
    ServerProfile server, {
    required String binDir,
    required String suffix,
    required String bootstrap,
    required bool includeInternal,
  }) async {
    try {
      final topics = await getDataManager().kafkaTopics(
        server,
        binDir: binDir,
        suffix: suffix,
        bootstrap: bootstrap,
        includeInternal: includeInternal,
      );
      final groups = await getDataManager().kafkaGroups(
        server,
        binDir: binDir,
        suffix: suffix,
        bootstrap: bootstrap,
      );
      final cluster = await getDataManager().kafkaCluster(
        server,
        binDir: binDir,
        suffix: suffix,
        bootstrap: bootstrap,
      );

      connection.onScanned(
        KafkaScan(topics: topics, groups: groups, cluster: cluster),
      );
    } catch (error) {
      connection.onScanFailed(error.toString());
    }
  }

  /// Runs `kafka-broker-api-versions` so the user can see exactly what the
  /// broker says when a scan comes back empty.
  void probeBroker(
    ServerProfile server, {
    required String binDir,
    required String suffix,
    required String bootstrap,
  }) async {
    showLoading("Probing $bootstrap …");
    try {
      final output = await getDataManager().kafkaTest(
        server,
        binDir: binDir,
        suffix: suffix,
        bootstrap: bootstrap,
      );
      closeLoading();
      connection.onProbeOutput(
        'kafka-broker-api-versions · $bootstrap',
        output.isEmpty ? '(no output — broker unreachable?)' : output,
      );
    } catch (error) {
      handleError(
        error,
        () => probeBroker(
          server,
          binDir: binDir,
          suffix: suffix,
          bootstrap: bootstrap,
        ),
        () {},
        "Retry",
      );
    }
  }
}