// Models for the Kafka monitor. All data is scraped from the standard Kafka
// CLI tools (`kafka-topics`, `kafka-consumer-groups`, `kafka-get-offsets`,
// `kafka-broker-api-versions`) run over SSH.

/// Result of locating the Kafka tooling on a server.
class KafkaStatus {
  final bool detected;
  final String binDir; // e.g. /opt/kafka/bin
  final String suffix; // ".sh" for Apache tarballs, "" for Confluent packages
  final bool java; // a JVM is on PATH
  final String version; // best-effort version string
  final String detail; // human note (where it was found / why not)

  KafkaStatus({
    required this.detected,
    this.binDir = '',
    this.suffix = '.sh',
    this.java = false,
    this.version = '',
    this.detail = '',
  });
}

class KafkaPartition {
  final int id;
  final int leader;
  final List<int> replicas;
  final List<int> isr;
  int earliest;
  int latest; // high-watermark (log-end offset)

  KafkaPartition({
    required this.id,
    required this.leader,
    required this.replicas,
    required this.isr,
    this.earliest = 0,
    this.latest = 0,
  });

  /// Messages currently retained on this partition.
  int get messages => (latest - earliest).clamp(0, 1 << 62);

  /// Replica set smaller than in-sync means the partition is healthy but the
  /// ISR shrinking below the replica count means it is under-replicated.
  bool get underReplicated => isr.length < replicas.length;

  bool get offline => leader < 0;
}

class KafkaTopic {
  final String name;
  final int replicationFactor;
  final String configs;
  final List<KafkaPartition> partitions;

  KafkaTopic({
    required this.name,
    required this.replicationFactor,
    required this.partitions,
    this.configs = '',
  });

  bool get internal => name.startsWith('_') || name.startsWith('__');
  int get partitionCount => partitions.length;

  /// Total high-watermark across partitions — the number of records ever
  /// produced (ignoring compaction), a good proxy for "messages sent".
  int get highWatermark => partitions.fold(0, (s, p) => s + p.latest);

  /// Messages currently retained on the topic (latest − earliest).
  int get retained => partitions.fold(0, (s, p) => s + p.messages);

  int get underReplicatedCount =>
      partitions.where((p) => p.underReplicated).length;
  int get offlineCount => partitions.where((p) => p.offline).length;
  bool get healthy => underReplicatedCount == 0 && offlineCount == 0;
}

/// One row of `kafka-consumer-groups --describe` (per topic/partition).
class KafkaGroupAssignment {
  final String topic;
  final int partition;
  final int currentOffset; // committed / processed
  final int logEndOffset; // produced
  final int lag;
  final String consumerId;
  final String host;
  final String clientId;

  KafkaGroupAssignment({
    required this.topic,
    required this.partition,
    required this.currentOffset,
    required this.logEndOffset,
    required this.lag,
    this.consumerId = '',
    this.host = '',
    this.clientId = '',
  });
}

class KafkaConsumerGroup {
  final String groupId;
  String state; // Stable | Empty | Dead | PreparingRebalance | ...
  int members;
  final String coordinator;
  final String assignmentStrategy;
  final List<KafkaGroupAssignment> assignments;

  KafkaConsumerGroup({
    required this.groupId,
    this.state = '',
    this.members = 0,
    this.coordinator = '',
    this.assignmentStrategy = '',
    List<KafkaGroupAssignment>? assignments,
  }) : assignments = assignments ?? [];

  int get totalLag =>
      assignments.fold(0, (s, a) => s + (a.lag < 0 ? 0 : a.lag));
  int get processed => assignments.fold(0, (s, a) => s + a.currentOffset);
  Set<String> get topics => assignments.map((a) => a.topic).toSet();
  int get liveMembers => assignments
      .map((a) => a.consumerId)
      .where((c) => c.isNotEmpty && c != '-')
      .toSet()
      .length;

  bool get active => state.toLowerCase() == 'stable';
  bool get idle => state.toLowerCase() == 'empty';
}

class KafkaBroker {
  final int id;
  final String host;
  final int port;
  final String rack;
  bool isController;

  KafkaBroker({
    required this.id,
    required this.host,
    required this.port,
    this.rack = '',
    this.isController = false,
  });

  String get endpoint => '$host:$port';
}

/// Cluster-level snapshot: brokers + health counts.
class KafkaCluster {
  final List<KafkaBroker> brokers;
  final String clusterId;
  final int controllerId;
  final List<String> underReplicated; // "topic-partition" strings
  final List<String> unavailable;

  KafkaCluster({
    required this.brokers,
    this.clusterId = '',
    this.controllerId = -1,
    List<String>? underReplicated,
    List<String>? unavailable,
  }) : underReplicated = underReplicated ?? [],
       unavailable = unavailable ?? [];

  bool get healthy => underReplicated.isEmpty && unavailable.isEmpty;
}