import 'package:flutter/material.dart';
import 'package:stacked/stacked.dart';

import '../../about/external/data/Kafka.dart';
import '../../about/internal/application/ServerProfile.dart';
import '../../about/internal/application/TextType.dart';
import '../../designs/OutputSheet.dart';
import '../../designs/text/Text.dart';
import '../../utils/Colors.dart';
import '../../utils/CommonUtils.dart';
import 'ConnectKafka.dart';
import 'Kafka.dart';
import 'ViewKafka.dart';

class KafkaState extends State<Kafka>
    with SingleTickerProviderStateMixin
    implements ConnectKafka {
  ViewKafka? _model;

  late final TabController _tabs;

  final _bootstrap = TextEditingController(text: 'localhost:9092');
  final _binDir = TextEditingController();
  String _suffix = '.sh';

  KafkaStatus? _status;
  List<KafkaTopic> _topics = [];
  List<KafkaConsumerGroup> _groups = [];
  KafkaCluster? _cluster;

  bool _loading = true;
  bool _detecting = true;
  bool _showInternal = false;
  String? _error;

  ServerProfile get _server => widget.server;

  String get _bootstrapServer => _bootstrap.text.trim().isEmpty
      ? 'localhost:9092'
      : _bootstrap.text.trim();

  bool get _detected => _status?.detected ?? false;

  @override
  void initState() {
    super.initState();
    _tabs = TabController(length: 3, vsync: this);
  }

  @override
  void dispose() {
    _tabs.dispose();
    _bootstrap.dispose();
    _binDir.dispose();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    return ViewModelBuilder<ViewKafka>.reactive(
      viewModelBuilder: () => ViewKafka(context, this),
      onViewModelReady: _onReady,
      builder: (context, viewModel, child) => Scaffold(
        appBar: AppBar(
          title: text(
            'Kafka · ${_server.name}',
            20,
            TextType.Bold,
            color: context.jp.text,
          ),
          actions: [
            if (_detected)
              IconButton(
                tooltip: 'Probe connection',
                onPressed: _onProbe,
                icon: const Icon(Icons.wifi_tethering),
              ),
            IconButton(
              tooltip: 'Refresh',
              onPressed: (_loading || _detecting) ? null : _onScan,
              icon: const Icon(Icons.refresh),
            ),
            const SizedBox(width: 4),
          ],
          bottom: _detected
              ? PreferredSize(
                  preferredSize: const Size.fromHeight(48),
                  child: TabBar(
                    controller: _tabs,
                    labelStyle: const TextStyle(
                      fontWeight: FontWeight.w800,
                      fontSize: 13,
                    ),
                    tabs: [
                      Tab(text: 'Topics · ${_topics.length}'),
                      Tab(text: 'Groups · ${_groups.length}'),
                      Tab(text: 'Cluster · ${_cluster?.brokers.length ?? 0}'),
                    ],
                  ),
                )
              : null,
        ),
        body: _detecting
            ? const Center(child: CircularProgressIndicator())
            : !_detected
            ? _notDetected()
            : Column(
                children: [
                  _connectionBar(),
                  if (_loading) const LinearProgressIndicator(minHeight: 3),
                  if (_error != null) _errorBar(),
                  Expanded(
                    child: TabBarView(
                      controller: _tabs,
                      children: [_topicsTab(), _groupsTab(), _clusterTab()],
                    ),
                  ),
                ],
              ),
      ),
    );
  }

  void _onReady(ViewKafka model) {
    _model = model;
    // onViewModelReady fires mid-build; a synchronous setState from the
    // first load would dirty this State while it is still being built.
    WidgetsBinding.instance.addPostFrameCallback((_) {
      if (mounted) _initiate();
    });
  }

  Future<void> _initiate() async {
    _model?.detectKafka(_server);
  }

  // ── named handlers ─────────────────────────────────────────────────────────

  void _onScan() {
    if (_binDir.text.trim().isEmpty) {
      _notify('Set the Kafka bin directory first.');
      return;
    }
    setState(() {
      _loading = true;
      _error = null;
    });
    _model?.scan(
      _server,
      binDir: _binDir.text.trim(),
      suffix: _suffix,
      bootstrap: _bootstrapServer,
      includeInternal: _showInternal,
    );
  }

  Future<void> _onRefreshPull() async => _onScan();

  void _onScanSubmitted(String _) => _onScan();

  void _onRedetect() {
    setState(() {
      _detecting = true;
      _error = null;
    });
    _model?.detectKafka(_server);
  }

  void _onProbe() => _model?.probeBroker(
    _server,
    binDir: _binDir.text.trim(),
    suffix: _suffix,
    bootstrap: _bootstrapServer,
  );

  /// Accepts a hand-entered bin directory when auto-detection came up empty.
  void _onUseThisPath() {
    if (_binDir.text.trim().isEmpty) {
      _notify('Enter the bin directory.');
      return;
    }
    setState(() {
      _suffix = '.sh';
      _status = KafkaStatus(
        detected: true,
        binDir: _binDir.text.trim(),
        suffix: _suffix,
        java: _status?.java ?? false,
      );
    });
    _onScan();
  }

  void _onToggleInternal(bool value) {
    setState(() => _showInternal = value);
    _onScan();
  }

  void _notify(String message) {
    if (!mounted) return;
    ScaffoldMessenger.of(
      context,
    ).showSnackBar(SnackBar(content: Text(message)));
  }

  // ── ConnectKafka ───────────────────────────────────────────────────────────

  @override
  void onKafkaDetected(KafkaStatus status) {
    if (!mounted) return;
    setState(() {
      _status = status;
      _detecting = false;
      if (status.detected) {
        _binDir.text = status.binDir;
        _suffix = status.suffix;
      }
    });
    if (status.detected) _onScan();
  }

  @override
  void onDetectFailed(String message) {
    if (!mounted) return;
    setState(() {
      _detecting = false;
      _error = message;
    });
  }

  @override
  void onScanned(KafkaScan scan) {
    if (!mounted) return;
    setState(() {
      _topics = scan.topics;
      _groups = scan.groups;
      _cluster = scan.cluster;
      _loading = false;
      _error = scan.isEmpty
          ? 'No data returned. Check the bootstrap server '
                '($_bootstrapServer) is reachable from this host and the broker '
                'is running.'
          : null;
    });
  }

  @override
  void onScanFailed(String message) {
    if (!mounted) return;
    setState(() {
      _loading = false;
      _error = message;
    });
  }

  @override
  void onProbeOutput(String title, String output) {
    if (!mounted) return;
    showOutputSheet(context, title, output);
  }

  // ── screen-only widgets ────────────────────────────────────────────────────

  Widget _connectionBar() {
    final jp = context.jp;
    return Container(
      padding: const EdgeInsets.fromLTRB(16, 12, 16, 12),
      decoration: BoxDecoration(
        color: jp.surface,
        border: Border(bottom: BorderSide(color: jp.border)),
      ),
      child: LayoutBuilder(
        builder: (context, c) {
          final bootstrap = TextField(
            controller: _bootstrap,
            onSubmitted: _onScanSubmitted,
            decoration: const InputDecoration(
              labelText: 'Bootstrap server',
              hintText: 'localhost:9092',
              prefixIcon: Icon(Icons.dns, size: 20),
              isDense: true,
            ),
          );
          final binDir = TextField(
            controller: _binDir,
            onSubmitted: _onScanSubmitted,
            style: const TextStyle(fontFamily: 'monospace', fontSize: 13),
            decoration: const InputDecoration(
              labelText: 'Kafka bin dir',
              hintText: '/opt/kafka/bin',
              prefixIcon: Icon(Icons.folder_outlined, size: 20),
              isDense: true,
            ),
          );
          final scanButton = FilledButton.icon(
            onPressed: _loading ? null : _onScan,
            icon: const Icon(Icons.sync, size: 18),
            label: text('Scan', 13, TextType.Bold, color: colorWhite),
          );

          if (c.maxWidth < 640) {
            return Column(
              children: [
                bootstrap,
                const SizedBox(height: 10),
                Row(
                  children: [
                    Expanded(child: binDir),
                    const SizedBox(width: 10),
                    scanButton,
                  ],
                ),
              ],
            );
          }
          return Row(
            children: [
              Expanded(flex: 3, child: bootstrap),
              const SizedBox(width: 10),
              Expanded(flex: 4, child: binDir),
              const SizedBox(width: 10),
              scanButton,
            ],
          );
        },
      ),
    );
  }

  Widget _errorBar() => Container(
    width: double.infinity,
    margin: const EdgeInsets.fromLTRB(16, 12, 16, 0),
    padding: const EdgeInsets.all(14),
    decoration: BoxDecoration(
      color: PeachTheme.warn.withValues(alpha: 0.10),
      borderRadius: BorderRadius.circular(12),
      border: Border.all(color: PeachTheme.warn.withValues(alpha: 0.5)),
    ),
    child: Row(
      children: [
        const Icon(Icons.info_outline, color: PeachTheme.warn, size: 20),
        const SizedBox(width: 10),
        Expanded(
          child: text(_error!, 12.5, TextType.Regular, color: context.jp.text),
        ),
      ],
    ),
  );

  Widget _notDetected() {
    final jp = context.jp;
    final hasJava = _status?.java ?? false;
    return ListView(
      padding: const EdgeInsets.fromLTRB(16, 20, 16, 40),
      children: [
        Container(
          padding: const EdgeInsets.all(16),
          decoration: PeachTheme.panel(context, glowColor: PeachTheme.warn),
          child: Column(
            crossAxisAlignment: CrossAxisAlignment.start,
            children: [
              Row(
                children: [
                  Container(
                    width: 46,
                    height: 46,
                    decoration: BoxDecoration(
                      color: PeachTheme.warn.withValues(alpha: 0.14),
                      borderRadius: BorderRadius.circular(13),
                    ),
                    child: const Icon(Icons.hub, color: PeachTheme.warn),
                  ),
                  const SizedBox(width: 14),
                  Expanded(
                    child: Column(
                      crossAxisAlignment: CrossAxisAlignment.start,
                      children: [
                        text(
                          'Kafka tools not found',
                          16,
                          TextType.Bold,
                          color: jp.text,
                        ),
                        text(
                          'kafka-topics wasn\'t on PATH or in the common dirs.',
                          12.5,
                          TextType.Regular,
                          color: jp.textDim,
                        ),
                      ],
                    ),
                  ),
                ],
              ),
              const SizedBox(height: 12),
              text(
                _status?.detail ?? '',
                12.5,
                TextType.Regular,
                color: jp.textDim,
              ),
              const SizedBox(height: 10),
              Row(
                children: [
                  Icon(
                    hasJava ? Icons.check_circle : Icons.error_outline,
                    size: 16,
                    color: hasJava ? PeachTheme.seed : PeachTheme.danger,
                  ),
                  const SizedBox(width: 6),
                  text(
                    hasJava
                        ? 'Java runtime present'
                        : 'No Java runtime on PATH',
                    12.5,
                    TextType.Regular,
                    color: jp.textDim,
                  ),
                ],
              ),
              const SizedBox(height: 16),
              text(
                'If Kafka is installed in a custom location, point to its '
                'bin directory and scan:',
                12.5,
                TextType.Regular,
                color: jp.textDim,
              ),
              const SizedBox(height: 10),
              TextField(
                controller: _binDir,
                style: const TextStyle(fontFamily: 'monospace', fontSize: 13),
                decoration: const InputDecoration(
                  labelText: 'Kafka bin dir',
                  hintText: '/opt/kafka/bin',
                  prefixIcon: Icon(Icons.folder_outlined, size: 20),
                  isDense: true,
                ),
              ),
              const SizedBox(height: 10),
              TextField(
                controller: _bootstrap,
                decoration: const InputDecoration(
                  labelText: 'Bootstrap server',
                  hintText: 'localhost:9092',
                  prefixIcon: Icon(Icons.dns, size: 20),
                  isDense: true,
                ),
              ),
              const SizedBox(height: 12),
              Row(
                children: [
                  FilledButton.icon(
                    onPressed: _onUseThisPath,
                    icon: const Icon(Icons.play_arrow, size: 18),
                    label: text(
                      'Use this path',
                      13,
                      TextType.Bold,
                      color: colorWhite,
                    ),
                  ),
                  const SizedBox(width: 10),
                  OutlinedButton.icon(
                    onPressed: _onRedetect,
                    icon: const Icon(Icons.refresh, size: 18),
                    label: text(
                      'Re-detect',
                      13,
                      TextType.Bold,
                      color: jp.textDim,
                    ),
                  ),
                ],
              ),
            ],
          ),
        ),
      ],
    );
  }

  // ── topics ─────────────────────────────────────────────────────────────────

  Widget _topicsTab() {
    final jp = context.jp;
    final produced = _topics.fold<int>(0, (s, t) => s + t.highWatermark);
    final partitions = _topics.fold<int>(0, (s, t) => s + t.partitionCount);
    return RefreshIndicator(
      onRefresh: _onRefreshPull,
      child: ListView(
        padding: const EdgeInsets.fromLTRB(16, 14, 16, 40),
        children: [
          _statRow([
            ('Topics', '${_topics.length}', PeachTheme.seed),
            ('Partitions', '$partitions', PeachTheme.accent),
            ('Messages sent', formatCount(produced), PeachTheme.violet),
          ]),
          const SizedBox(height: 12),
          Row(
            children: [
              text(
                'TOPICS',
                12,
                TextType.Bold,
                color: jp.textDim,
                letterSpacing: 1.2,
              ),
              const Spacer(),
              text('internal', 11.5, TextType.Regular, color: jp.textFaint),
              Switch(value: _showInternal, onChanged: _onToggleInternal),
            ],
          ),
          const SizedBox(height: 4),
          if (_topics.isEmpty)
            _emptyBox(
              Icons.topic_outlined,
              'No topics',
              _loading ? 'Loading…' : 'This cluster has no topics yet.',
            )
          else
            ..._topics.map(_topicCard),
        ],
      ),
    );
  }

  Widget _topicCard(KafkaTopic topic) {
    final jp = context.jp;
    return Container(
      margin: const EdgeInsets.only(bottom: 12),
      decoration: PeachTheme.panel(context, radius: 16),
      clipBehavior: Clip.antiAlias,
      child: Theme(
        data: Theme.of(context).copyWith(dividerColor: Colors.transparent),
        child: ExpansionTile(
          tilePadding: const EdgeInsets.symmetric(horizontal: 14, vertical: 4),
          childrenPadding: const EdgeInsets.fromLTRB(14, 0, 14, 12),
          leading: Icon(
            Icons.topic,
            color: topic.healthy ? PeachTheme.seed : PeachTheme.danger,
          ),
          title: text(
            topic.name,
            14,
            TextType.Bold,
            color: jp.text,
            maxLines: 1,
            overflow: TextOverflow.ellipsis,
          ),
          subtitle: Padding(
            padding: const EdgeInsets.only(top: 6),
            child: Wrap(
              spacing: 6,
              runSpacing: 6,
              children: [
                _chip('${topic.partitionCount}p', PeachTheme.accent),
                _chip('RF ${topic.replicationFactor}', jp.textFaint),
                _chip(
                  '~${formatCount(topic.retained)} msgs',
                  PeachTheme.violet,
                ),
                if (topic.underReplicatedCount > 0)
                  _chip(
                    '${topic.underReplicatedCount} under-repl',
                    PeachTheme.warn,
                  ),
                if (topic.offlineCount > 0)
                  _chip('${topic.offlineCount} offline', PeachTheme.danger),
                if (topic.internal) _chip('internal', jp.textFaint),
              ],
            ),
          ),
          children: [
            _partitionHeader(),
            ...topic.partitions.map(_partitionRow),
            if (topic.configs.trim().isNotEmpty) ...[
              const SizedBox(height: 8),
              text(
                'configs: ${topic.configs}',
                11,
                TextType.Regular,
                color: jp.textFaint,
              ),
            ],
          ],
        ),
      ),
    );
  }

  Widget _partitionHeader() {
    final jp = context.jp;
    return Padding(
      padding: const EdgeInsets.only(bottom: 4, top: 2),
      child: Row(
        children: [
          _cell('P', jp.textFaint, 30, bold: true),
          _cell('LEADER', jp.textFaint, 60, bold: true),
          _cell('ISR/REP', jp.textFaint, 70, bold: true),
          Expanded(child: _cell('OFFSETS', jp.textFaint, null, bold: true)),
          _cell('MSGS', jp.textFaint, 70, bold: true, right: true),
        ],
      ),
    );
  }

  Widget _partitionRow(KafkaPartition p) {
    final jp = context.jp;
    final degraded = p.underReplicated || p.offline;
    return Padding(
      padding: const EdgeInsets.symmetric(vertical: 3),
      child: Row(
        children: [
          _cell('${p.id}', jp.text, 30),
          _cell(
            p.offline ? 'none' : '${p.leader}',
            p.offline ? PeachTheme.danger : jp.text,
            60,
          ),
          _cell(
            '${p.isr.length}/${p.replicas.length}',
            degraded ? PeachTheme.warn : jp.textDim,
            70,
          ),
          Expanded(
            child: _cell(
              '${formatCount(p.earliest)} → ${formatCount(p.latest)}',
              jp.textDim,
              null,
            ),
          ),
          _cell(formatCount(p.messages), jp.text, 70, right: true, bold: true),
        ],
      ),
    );
  }

  // ── consumer groups ────────────────────────────────────────────────────────

  Widget _groupsTab() {
    final jp = context.jp;
    final totalLag = _groups.fold<int>(0, (s, g) => s + g.totalLag);
    final active = _groups.where((g) => g.active).length;
    return RefreshIndicator(
      onRefresh: _onRefreshPull,
      child: ListView(
        padding: const EdgeInsets.fromLTRB(16, 14, 16, 40),
        children: [
          _statRow([
            ('Groups', '${_groups.length}', PeachTheme.seed),
            ('Active', '$active', PeachTheme.accent),
            (
              'Total lag',
              formatCount(totalLag),
              totalLag > 0 ? PeachTheme.warn : PeachTheme.seed,
            ),
          ]),
          const SizedBox(height: 12),
          text(
            'CONSUMER GROUPS',
            12,
            TextType.Bold,
            color: jp.textDim,
            letterSpacing: 1.2,
          ),
          const SizedBox(height: 8),
          if (_groups.isEmpty)
            _emptyBox(
              Icons.groups_outlined,
              'No consumer groups',
              _loading ? 'Loading…' : 'Nothing is consuming from this cluster.',
            )
          else
            ..._groups.map(_groupCard),
        ],
      ),
    );
  }

  Widget _groupCard(KafkaConsumerGroup group) {
    final jp = context.jp;
    final (color, label) = switch (group.state.toLowerCase()) {
      'stable' => (PeachTheme.seed, 'STABLE'),
      'empty' => (jp.textFaint, 'EMPTY'),
      'dead' => (PeachTheme.danger, 'DEAD'),
      'preparingrebalance' => (PeachTheme.warn, 'REBALANCING'),
      'completingrebalance' => (PeachTheme.warn, 'REBALANCING'),
      _ => (
        jp.textFaint,
        group.state.isEmpty ? 'UNKNOWN' : group.state.toUpperCase(),
      ),
    };
    return Container(
      margin: const EdgeInsets.only(bottom: 12),
      decoration: PeachTheme.panel(context, radius: 16),
      clipBehavior: Clip.antiAlias,
      child: Theme(
        data: Theme.of(context).copyWith(dividerColor: Colors.transparent),
        child: ExpansionTile(
          tilePadding: const EdgeInsets.symmetric(horizontal: 14, vertical: 4),
          childrenPadding: const EdgeInsets.fromLTRB(14, 0, 14, 12),
          leading: Icon(
            Icons.groups,
            color: group.active ? PeachTheme.seed : jp.textFaint,
          ),
          title: Row(
            children: [
              Expanded(
                child: text(
                  group.groupId,
                  14,
                  TextType.Bold,
                  color: jp.text,
                  maxLines: 1,
                  overflow: TextOverflow.ellipsis,
                ),
              ),
              Container(
                padding: const EdgeInsets.symmetric(horizontal: 9, vertical: 3),
                decoration: BoxDecoration(
                  color: color.withValues(alpha: 0.16),
                  borderRadius: BorderRadius.circular(20),
                ),
                child: text(label, 10, TextType.Bold, color: color),
              ),
            ],
          ),
          subtitle: Padding(
            padding: const EdgeInsets.only(top: 6),
            child: Wrap(
              spacing: 6,
              runSpacing: 6,
              children: [
                _chip('${group.members} members', PeachTheme.accent),
                _chip('${group.topics.length} topics', jp.textFaint),
                _chip(
                  'lag ${formatCount(group.totalLag)}',
                  group.totalLag > 0 ? PeachTheme.warn : PeachTheme.seed,
                ),
                _chip(
                  '${formatCount(group.processed)} processed',
                  PeachTheme.violet,
                ),
              ],
            ),
          ),
          children: [
            if (group.assignments.isEmpty)
              Padding(
                padding: const EdgeInsets.symmetric(vertical: 8),
                child: text(
                  'No committed offsets (idle group).',
                  12,
                  TextType.Regular,
                  color: jp.textFaint,
                ),
              )
            else ...[
              _assignHeader(),
              ...group.assignments.map(_assignRow),
            ],
          ],
        ),
      ),
    );
  }

  Widget _assignHeader() {
    final jp = context.jp;
    return Padding(
      padding: const EdgeInsets.only(bottom: 4, top: 2),
      child: Row(
        children: [
          Expanded(
            flex: 4,
            child: _cell('TOPIC', jp.textFaint, null, bold: true),
          ),
          _cell('P', jp.textFaint, 26, bold: true),
          _cell('CURRENT', jp.textFaint, 74, bold: true, right: true),
          _cell('END', jp.textFaint, 74, bold: true, right: true),
          _cell('LAG', jp.textFaint, 60, bold: true, right: true),
        ],
      ),
    );
  }

  Widget _assignRow(KafkaGroupAssignment a) {
    final jp = context.jp;
    final live = a.consumerId.isNotEmpty && a.consumerId != '-';
    return Padding(
      padding: const EdgeInsets.symmetric(vertical: 3),
      child: Column(
        crossAxisAlignment: CrossAxisAlignment.start,
        children: [
          Row(
            children: [
              Expanded(flex: 4, child: _cell(a.topic, jp.text, null)),
              _cell('${a.partition}', jp.textDim, 26),
              _cell(formatCount(a.currentOffset), jp.textDim, 74, right: true),
              _cell(formatCount(a.logEndOffset), jp.textDim, 74, right: true),
              _cell(
                formatCount(a.lag),
                a.lag > 0 ? PeachTheme.warn : jp.textDim,
                60,
                right: true,
                bold: true,
              ),
            ],
          ),
          if (live)
            Padding(
              padding: const EdgeInsets.only(left: 2, top: 1, bottom: 2),
              child: text(
                '${a.consumerId}  ${a.host}',
                10,
                TextType.Regular,
                color: jp.textFaint,
                maxLines: 1,
                overflow: TextOverflow.ellipsis,
              ),
            ),
        ],
      ),
    );
  }

  // ── cluster ────────────────────────────────────────────────────────────────

  Widget _clusterTab() {
    final jp = context.jp;
    final cluster = _cluster;
    final healthy = cluster?.healthy ?? false;
    return RefreshIndicator(
      onRefresh: _onRefreshPull,
      child: ListView(
        padding: const EdgeInsets.fromLTRB(16, 14, 16, 40),
        children: [
          _statRow([
            ('Brokers', '${cluster?.brokers.length ?? 0}', PeachTheme.seed),
            (
              'Under-repl',
              '${cluster?.underReplicated.length ?? 0}',
              (cluster?.underReplicated.isEmpty ?? true)
                  ? PeachTheme.seed
                  : PeachTheme.warn,
            ),
            (
              'Offline',
              '${cluster?.unavailable.length ?? 0}',
              (cluster?.unavailable.isEmpty ?? true)
                  ? PeachTheme.seed
                  : PeachTheme.danger,
            ),
          ]),
          const SizedBox(height: 12),
          Container(
            padding: const EdgeInsets.all(14),
            decoration: BoxDecoration(
              color: (healthy ? PeachTheme.seed : PeachTheme.danger).withValues(
                alpha: 0.10,
              ),
              borderRadius: BorderRadius.circular(14),
              border: Border.all(
                color: (healthy ? PeachTheme.seed : PeachTheme.danger)
                    .withValues(alpha: 0.4),
              ),
            ),
            child: Row(
              children: [
                Icon(
                  healthy ? Icons.verified : Icons.warning_amber_rounded,
                  color: healthy ? PeachTheme.seed : PeachTheme.danger,
                ),
                const SizedBox(width: 12),
                Expanded(
                  child: text(
                    healthy
                        ? 'Cluster healthy — all partitions in sync.'
                        : '${cluster?.underReplicated.length ?? 0} under-replicated · '
                              '${cluster?.unavailable.length ?? 0} offline partitions.',
                    14,
                    TextType.Bold,
                    color: healthy ? PeachTheme.seed : PeachTheme.danger,
                  ),
                ),
              ],
            ),
          ),
          const SizedBox(height: 16),
          if (cluster != null &&
              (cluster.clusterId.isNotEmpty || cluster.controllerId >= 0))
            Container(
              margin: const EdgeInsets.only(bottom: 16),
              padding: const EdgeInsets.all(14),
              decoration: PeachTheme.panel(context, radius: 16),
              child: Column(
                crossAxisAlignment: CrossAxisAlignment.start,
                children: [
                  if (cluster.clusterId.isNotEmpty)
                    _keyValue('Cluster ID', cluster.clusterId),
                  if (cluster.controllerId >= 0)
                    _keyValue('Controller', 'broker ${cluster.controllerId}'),
                ],
              ),
            ),
          text(
            'BROKERS',
            12,
            TextType.Bold,
            color: jp.textDim,
            letterSpacing: 1.2,
          ),
          const SizedBox(height: 8),
          if (cluster == null || cluster.brokers.isEmpty)
            _emptyBox(
              Icons.dns_outlined,
              'No brokers',
              _loading
                  ? 'Loading…'
                  : 'Could not reach any broker at $_bootstrapServer.',
            )
          else
            LayoutBuilder(
              builder: (context, c) {
                const gap = 12.0;
                final cols = c.maxWidth >= 620 ? 2 : 1;
                final w = (c.maxWidth - gap * (cols - 1)) / cols;
                return Wrap(
                  spacing: gap,
                  runSpacing: gap,
                  children: [
                    for (final broker in cluster.brokers)
                      SizedBox(width: w, child: _brokerCard(broker)),
                  ],
                );
              },
            ),
          if (cluster != null && cluster.underReplicated.isNotEmpty) ...[
            const SizedBox(height: 16),
            _partitionListCard(
              'Under-replicated partitions',
              cluster.underReplicated,
              PeachTheme.warn,
            ),
          ],
          if (cluster != null && cluster.unavailable.isNotEmpty) ...[
            const SizedBox(height: 16),
            _partitionListCard(
              'Offline partitions',
              cluster.unavailable,
              PeachTheme.danger,
            ),
          ],
        ],
      ),
    );
  }

  Widget _brokerCard(KafkaBroker broker) {
    final jp = context.jp;
    return Container(
      padding: const EdgeInsets.all(14),
      decoration: PeachTheme.panel(context, radius: 16),
      child: Row(
        children: [
          Container(
            width: 42,
            height: 42,
            alignment: Alignment.center,
            decoration: BoxDecoration(
              color: PeachTheme.accent.withValues(alpha: 0.14),
              borderRadius: BorderRadius.circular(12),
            ),
            child: text(
              '${broker.id}',
              15,
              TextType.Bold,
              color: PeachTheme.accent,
            ),
          ),
          const SizedBox(width: 12),
          Expanded(
            child: Column(
              crossAxisAlignment: CrossAxisAlignment.start,
              children: [
                text(
                  broker.endpoint,
                  14,
                  TextType.Bold,
                  color: jp.text,
                  maxLines: 1,
                  overflow: TextOverflow.ellipsis,
                ),
                text(
                  broker.rack.isEmpty
                      ? 'broker ${broker.id}'
                      : 'rack ${broker.rack}',
                  11.5,
                  TextType.Regular,
                  color: jp.textFaint,
                ),
              ],
            ),
          ),
          if (broker.isController)
            Container(
              padding: const EdgeInsets.symmetric(horizontal: 9, vertical: 3),
              decoration: BoxDecoration(
                color: PeachTheme.seed.withValues(alpha: 0.16),
                borderRadius: BorderRadius.circular(20),
              ),
              child: text(
                'CONTROLLER',
                9.5,
                TextType.Bold,
                color: PeachTheme.seed,
              ),
            ),
        ],
      ),
    );
  }

  Widget _partitionListCard(String title, List<String> items, Color color) {
    final jp = context.jp;
    return Container(
      padding: const EdgeInsets.all(14),
      decoration: PeachTheme.panel(context, radius: 16),
      child: Column(
        crossAxisAlignment: CrossAxisAlignment.start,
        children: [
          Row(
            children: [
              Icon(Icons.warning_amber_rounded, size: 18, color: color),
              const SizedBox(width: 8),
              text(title, 14, TextType.Bold, color: jp.text),
              const SizedBox(width: 8),
              text('${items.length}', 14, TextType.Bold, color: color),
            ],
          ),
          const SizedBox(height: 8),
          Wrap(
            spacing: 6,
            runSpacing: 6,
            children: items.map((s) => _chip(s, color)).toList(),
          ),
        ],
      ),
    );
  }

  // ── shared bits ────────────────────────────────────────────────────────────

  Widget _statRow(List<(String, String, Color)> stats) {
    return Row(
      children: [
        for (var i = 0; i < stats.length; i++) ...[
          if (i > 0) const SizedBox(width: 10),
          Expanded(child: _statCard(stats[i])),
        ],
      ],
    );
  }

  Widget _statCard((String, String, Color) stat) {
    final (label, value, color) = stat;
    return Container(
      padding: const EdgeInsets.symmetric(horizontal: 14, vertical: 14),
      decoration: PeachTheme.panel(context, radius: 16),
      child: Column(
        crossAxisAlignment: CrossAxisAlignment.start,
        children: [
          text(
            value,
            22,
            TextType.Bold,
            color: color,
            maxLines: 1,
            overflow: TextOverflow.ellipsis,
          ),
          const SizedBox(height: 2),
          text(
            label.toUpperCase(),
            10.5,
            TextType.Bold,
            color: context.jp.textFaint,
            letterSpacing: 0.6,
          ),
        ],
      ),
    );
  }

  Widget _chip(String label, Color color) => Container(
    padding: const EdgeInsets.symmetric(horizontal: 8, vertical: 3),
    decoration: BoxDecoration(
      color: color.withValues(alpha: 0.13),
      borderRadius: BorderRadius.circular(7),
      border: Border.all(color: color.withValues(alpha: 0.30)),
    ),
    child: text(label, 11, TextType.Bold, color: color),
  );

  /// A fixed-width monospace cell, so the partition and assignment tables line
  /// up as columns rather than drifting with content.
  Widget _cell(
    String label,
    Color color,
    double? width, {
    bool bold = false,
    bool right = false,
  }) {
    final child = Text(
      label,
      maxLines: 1,
      overflow: TextOverflow.ellipsis,
      textAlign: right ? TextAlign.right : TextAlign.left,
      style: TextStyle(
        fontSize: 11.5,
        fontFamily: 'monospace',
        fontWeight: bold ? FontWeight.w800 : FontWeight.w500,
        color: color,
      ),
    );
    if (width == null) return child;
    return SizedBox(width: width, child: child);
  }

  Widget _keyValue(String key, String value) {
    final jp = context.jp;
    return Padding(
      padding: const EdgeInsets.symmetric(vertical: 3),
      child: Row(
        crossAxisAlignment: CrossAxisAlignment.start,
        children: [
          SizedBox(
            width: 90,
            child: text(key, 12.5, TextType.Regular, color: jp.textFaint),
          ),
          Expanded(
            child: SelectableText(
              value,
              style: TextStyle(
                fontSize: 12.5,
                fontFamily: 'monospace',
                color: jp.text,
              ),
            ),
          ),
        ],
      ),
    );
  }

  Widget _emptyBox(IconData icon, String title, String subtitle) {
    final jp = context.jp;
    return Container(
      padding: const EdgeInsets.symmetric(vertical: 40),
      alignment: Alignment.center,
      decoration: PeachTheme.panel(context),
      child: Column(
        children: [
          Icon(icon, size: 40, color: jp.textFaint),
          const SizedBox(height: 10),
          text(title, 14, TextType.Regular, color: jp.textDim),
          const SizedBox(height: 4),
          text(
            subtitle,
            12,
            TextType.Regular,
            color: jp.textFaint,
            align: TextAlign.center,
          ),
        ],
      ),
    );
  }
}