use crate::argument_parsers::RangedValueParser;
use crate::arguments::topic_id_argument;
use crate::capability::{
  Capability, CommandExecutor, CREATE_COMMAND, CREATE_COMMAND_ALIAS, DELETE_COMMAND, DELETE_COMMAND_ALIAS, LIST_COMMAND, LIST_COMMAND_ALIAS, SHOW_COMMAND, SHOW_COMMAND_ALIAS,
};
use crate::capability_builder::CapabilityBuilder;
use crate::context::Context;
use crate::flags::FlagType;
use crate::formatters::ids_formatter::IdsFormatter;
use crate::formatters::list_formatter::ListFormatter;
use crate::formatters::unit_formatter::UnitFormatter;
use crate::formatters::{hashmap_to_table, Value};
use crate::formatters::{ColumnAlignment, OutputFormat};
use crate::formatters::{Label, SubjectFormatter};
use crate::subject::{Requirements, Subject};
use crate::subjects::{DEFAULT_ALLOCATION_STATUS_LABELS, DEPENDANT_LABELS, DEPENDANT_LABELS_LIST};
use crate::{err, DshCliResult};
use async_trait::async_trait;
use clap::builder::PossibleValue;
use clap::{builder, Arg, ArgAction, ArgMatches};
use dsh_api::dsh_api_client::DshApiClient;
use dsh_api::topic::TopicInjection;
use dsh_api::types::{Topic, TopicStatus};
use dsh_api::Dependant;
use futures::future::try_join_all;
use itertools::Itertools;
use lazy_static::lazy_static;
use serde::Serialize;
use std::collections::HashMap;

struct TopicSubject {}

const TOPIC_SUBJECT_TARGET: &str = "topic";

lazy_static! {
  pub(crate) static ref TOPIC_SUBJECT: Box<dyn Subject + Send + Sync> = Box::new(TopicSubject {});
}

#[async_trait]
impl Subject for TopicSubject {
  fn subject(&self) -> &'static str {
    TOPIC_SUBJECT_TARGET
  }

  fn subject_command_about(&self) -> String {
    "Show, manage and list DSH scratch topics.".to_string()
  }

  fn subject_command_long_about(&self) -> String {
    "Show, manage and list scratch topics deployed on the DSH.".to_string()
  }

  fn subject_command_alias(&self) -> Option<&str> {
    Some("t")
  }

  fn capability(&self, capability_command: &str) -> Option<&(dyn Capability + Send + Sync)> {
    match capability_command {
      CREATE_COMMAND => Some(TOPIC_CREATE_CAPABILITY.as_ref()),
      DELETE_COMMAND => Some(TOPIC_DELETE_CAPABILITY.as_ref()),
      LIST_COMMAND => Some(TOPIC_LIST_CAPABILITY.as_ref()),
      SHOW_COMMAND => Some(TOPIC_SHOW_CAPABILITY.as_ref()),
      _ => None,
    }
  }

  fn capabilities(&self) -> &Vec<&(dyn Capability + Send + Sync)> {
    &TOPIC_CAPABILITIES
  }
}

lazy_static! {
  static ref TOPIC_CREATE_CAPABILITY: Box<(dyn Capability + Send + Sync)> = Box::new(
    CapabilityBuilder::new(CREATE_COMMAND, Some(CREATE_COMMAND_ALIAS), &TopicCreate {}, "Create new scratch topic")
      .add_target_argument(topic_id_argument().required(true))
      .add_extra_arguments(vec![
        cleanup_policy_flag(),
        compression_type_flag(),
        delete_retention_ms_flag(),
        max_message_size_flag(),
        message_timestamp_type_flag(),
        partitions_flag(),
        retention_bytes_flag(),
        retention_ms_flag(),
        segment_bytes_flag(),
      ])
  );
  static ref TOPIC_DELETE_CAPABILITY: Box<(dyn Capability + Send + Sync)> = Box::new(
    CapabilityBuilder::new(DELETE_COMMAND, Some(DELETE_COMMAND_ALIAS), &TopicDelete {}, "Delete scratch topic")
      .set_long_about("Delete a scratch topic.")
      .add_target_argument(topic_id_argument().required(true))
  );
  static ref TOPIC_LIST_CAPABILITY: Box<(dyn Capability + Send + Sync)> = Box::new(
    CapabilityBuilder::new(LIST_COMMAND, Some(LIST_COMMAND_ALIAS), &TopicList {}, "List scratch topics")
      .set_long_about("Lists all available scratch topics.")
      .add_command_executors(vec![
        (FlagType::AllocationStatus, &TopicListAllocationStatus {}, None),
        (FlagType::Ids, &TopicListIds {}, None),
        (FlagType::Usage, &TopicListUsage {}, None),
      ])
  );
  static ref TOPIC_SHOW_CAPABILITY: Box<(dyn Capability + Send + Sync)> = Box::new(
    CapabilityBuilder::new(SHOW_COMMAND, Some(SHOW_COMMAND_ALIAS), &TopicShow {}, "Show scratch topic configuration")
      .add_command_executors(vec![
        (FlagType::AllocationStatus, &TopicShowAllocationStatus {}, None),
        (FlagType::Properties, &TopicShowProperties {}, None),
        (FlagType::Usage, &TopicShowUsage {}, None),
      ])
      .add_target_argument(topic_id_argument().required(true))
  );
  static ref TOPIC_CAPABILITIES: Vec<&'static (dyn Capability + Send + Sync)> =
    vec![TOPIC_CREATE_CAPABILITY.as_ref(), TOPIC_DELETE_CAPABILITY.as_ref(), TOPIC_LIST_CAPABILITY.as_ref(), TOPIC_SHOW_CAPABILITY.as_ref()];
}

const CLEANUP_POLICY_FLAG: &str = "cleanup-policy";

pub(crate) fn cleanup_policy_flag() -> Arg {
  Arg::new(CLEANUP_POLICY_FLAG)
    .long(CLEANUP_POLICY_FLAG)
    .action(ArgAction::Set)
    .value_parser(builder::PossibleValuesParser::new(vec![PossibleValue::new("compact")]))
    .value_name("POLICY")
    .help("Cleanup policy")
    .long_help("Cleanup policy for the new scratch topic.")
}

const COMPRESSION_TYPE_FLAG: &str = "compression-type";

pub(crate) fn compression_type_flag() -> Arg {
  Arg::new(COMPRESSION_TYPE_FLAG)
    .long(COMPRESSION_TYPE_FLAG)
    .action(ArgAction::Set)
    .value_parser(builder::PossibleValuesParser::new(vec![
      PossibleValue::new("gzip"),
      PossibleValue::new("lz4"),
      PossibleValue::new("producer"),
      PossibleValue::new("snappy"),
      PossibleValue::new("uncompressed"),
      PossibleValue::new("zstd"),
    ]))
    .value_name("TYPE")
    .help("Compression type")
    .long_help("Compression type for the new scratch topic.")
}

const DELETE_RETENTION_MS_FLAG: &str = "delete-retention-ms";

pub(crate) fn delete_retention_ms_flag() -> Arg {
  Arg::new(DELETE_RETENTION_MS_FLAG)
    .long(DELETE_RETENTION_MS_FLAG)
    .action(ArgAction::Set)
    .value_parser(RangedValueParser::<u64>::with_lower(0))
    .value_name("MS")
    .help("Delete retention")
    .long_help("Delete retention time in milliseconds.")
}

const MAX_MESSAGE_BYTES_FLAG: &str = "max-message-bytes";

pub(crate) fn max_message_size_flag() -> Arg {
  Arg::new(MAX_MESSAGE_BYTES_FLAG)
    .long(MAX_MESSAGE_BYTES_FLAG)
    .action(ArgAction::Set)
    .value_parser(RangedValueParser::<u64>::new(1024, 1048576))
    .value_name("BYTES")
    .help("Max message size")
    .long_help(
      "Maximum message size (in bytes) for the new scratch topic. \
      The minimum message size is 1024 bytes and the maximum size is 1048576 bytes.",
    )
}

const MESSAGE_TIMESTAMP_TYPE_FLAG: &str = "message-timestamp-type";
const TIMESTAMP_CREATE_TIME: &str = "create-time";
const TIMESTAMP_LOG_APPEND_TIME: &str = "log-append-time";

pub(crate) fn message_timestamp_type_flag() -> Arg {
  Arg::new(MESSAGE_TIMESTAMP_TYPE_FLAG)
    .long(MESSAGE_TIMESTAMP_TYPE_FLAG)
    .action(ArgAction::Set)
    .value_parser(builder::PossibleValuesParser::new(vec![
      PossibleValue::new(TIMESTAMP_CREATE_TIME),
      PossibleValue::new(TIMESTAMP_LOG_APPEND_TIME),
    ]))
    .value_name("TYPE")
    .help("Message timestamp type")
    .long_help(
      "Message timestamp type for the new scratch topic. \
        The allowed values are 'create-time' and 'log-append-time'.",
    )
}

const PARTITIONS_FLAG: &str = "partitions";

pub(crate) fn partitions_flag() -> Arg {
  Arg::new(PARTITIONS_FLAG)
    .long(PARTITIONS_FLAG)
    .action(ArgAction::Set)
    .value_parser(RangedValueParser::<u32>::new(1, 128))
    .value_name("PARTITIONS")
    .help("Number of partitions")
    .long_help(
      "Number of partitions for the new scratch topic. \
          If this option is not specified the created topic will have only 1 partition.",
    )
}

const RETENTION_BYTES_FLAG: &str = "retention-bytes";

pub(crate) fn retention_bytes_flag() -> Arg {
  Arg::new(RETENTION_BYTES_FLAG)
    .long(RETENTION_BYTES_FLAG)
    .action(ArgAction::Set)
    .value_parser(RangedValueParser::<i64>::with_lower(-1))
    .value_name("BYTES")
    .help("Retention bytes")
    .long_help(
      "Retention bytes for the new scratch topic. \
      If minimum retention bytes value is set to -1, no limit is applied.",
    )
}

const RETENTION_MS_FLAG: &str = "retention-ms";

pub(crate) fn retention_ms_flag() -> Arg {
  Arg::new(RETENTION_MS_FLAG)
    .long(RETENTION_MS_FLAG)
    .action(ArgAction::Set)
    .value_parser(RangedValueParser::<u64>::new(3600000, 31536000000))
    .value_name("MS")
    .help("Retention milliseconds")
    .long_help(
      "Retention time in milliseconds for the new scratch topic. \
      The minimum retention time value is 3600000 milliseconds (1 hour) \
      and the maximum retention time is 31536000000 milliseconds (1 year).",
    )
}

const SEGMENT_BYTES_FLAG: &str = "segment-bytes";

pub(crate) fn segment_bytes_flag() -> Arg {
  Arg::new(SEGMENT_BYTES_FLAG)
    .long(SEGMENT_BYTES_FLAG)
    .action(ArgAction::Set)
    .value_parser(RangedValueParser::<u64>::with_lower(52428800))
    .value_name("BYTES")
    .help("Segment bytes")
    .long_help(
      "Segment bytes for the new scratch topic. \
          The minimum segment bytes value is 52428800 bytes",
    )
}

struct TopicCreate {}

#[async_trait]
impl CommandExecutor for TopicCreate {
  async fn execute_with_client(&self, target: Option<String>, _: Option<String>, matches: &ArgMatches, client: &DshApiClient, context: &Context) -> DshCliResult<()> {
    const REPLICATION_FACTOR: u32 = 3;
    let topic_id = target.unwrap_or_else(|| unreachable!());
    if client.get_topic_configuration(&topic_id).await.is_ok() {
      return err!("scratch topic '{}' already exists", topic_id);
    }
    let topic = create_topic(matches)?;
    context.print_explanation(format!(
      "create new scratch topic '{}', number of partitions {}, replication factor {}",
      topic_id, topic.partitions, REPLICATION_FACTOR
    ));
    if context.dry_run() {
      context.print_warning("dry-run mode, scratch topic not created");
    } else {
      client.put_topic_configuration(&topic_id, &topic).await?;
      context.print_outcome(format!("scratch topic '{}' created", topic_id));
    }
    Ok(())
  }

  fn requirements(&self, _: &ArgMatches) -> Requirements {
    Requirements::standard_with_api()
  }
}

pub(crate) const CLEANUP_POLICY_PROPERTY: &str = "cleanup.policy";
pub(crate) const COMPRESSION_TYPE_PROPERTY: &str = "compression.type";
pub(crate) const DELETE_RETENTION_MS_PROPERTY: &str = "delete.retention.ms";
pub(crate) const MAX_MESSAGE_BYTES_PROPERTY: &str = "max.message.bytes";
pub(crate) const MESSAGE_TIMESTAMP_PROPERTY: &str = "message.timestamp.type";
pub(crate) const RETENTION_BYTES_PROPERTY: &str = "retention.bytes";
pub(crate) const RETENTION_MS_PROPERTY: &str = "retention.ms";
pub(crate) const SEGMENT_BYTES_PROPERTY: &str = "segment.bytes";

pub(crate) fn create_topic(matches: &ArgMatches) -> DshCliResult<Topic> {
  const REPLICATION_FACTOR: u32 = 3;
  let replication_factor = REPLICATION_FACTOR as i64;
  let partitions = matches.get_one::<u32>(PARTITIONS_FLAG).cloned().unwrap_or(1) as i64;
  let mut kafka_properties = HashMap::new();
  if let Some(cleanup_policy) = matches.get_one::<String>(CLEANUP_POLICY_FLAG) {
    kafka_properties.insert(CLEANUP_POLICY_PROPERTY.to_string(), cleanup_policy.to_string());
  }
  if let Some(compression_type) = matches.get_one::<String>(COMPRESSION_TYPE_FLAG) {
    kafka_properties.insert(COMPRESSION_TYPE_PROPERTY.to_string(), compression_type.to_string());
  }
  if let Some(delete_retention_ms) = matches.get_one::<u64>(DELETE_RETENTION_MS_FLAG) {
    kafka_properties.insert(DELETE_RETENTION_MS_PROPERTY.to_string(), delete_retention_ms.to_string());
  }
  if let Some(max_message_bytes) = matches.get_one::<u64>(MAX_MESSAGE_BYTES_FLAG) {
    kafka_properties.insert(MAX_MESSAGE_BYTES_PROPERTY.to_string(), max_message_bytes.to_string());
  }
  if let Some(message_timestamp_type) = matches.get_one::<String>(MESSAGE_TIMESTAMP_TYPE_FLAG) {
    if message_timestamp_type == TIMESTAMP_LOG_APPEND_TIME {
      kafka_properties.insert(MESSAGE_TIMESTAMP_PROPERTY.to_string(), "LogAppendTime".to_string());
    } else if message_timestamp_type == TIMESTAMP_CREATE_TIME {
      kafka_properties.insert(MESSAGE_TIMESTAMP_PROPERTY.to_string(), "CreateTime".to_string());
    }
  }
  if let Some(retention_bytes) = matches.get_one::<i64>(RETENTION_BYTES_FLAG) {
    kafka_properties.insert(RETENTION_BYTES_PROPERTY.to_string(), retention_bytes.to_string());
  }
  if let Some(retention_ms) = matches.get_one::<u64>(RETENTION_MS_FLAG) {
    kafka_properties.insert(RETENTION_MS_PROPERTY.to_string(), retention_ms.to_string());
  }
  if let Some(segment_bytes) = matches.get_one::<u64>(SEGMENT_BYTES_FLAG) {
    kafka_properties.insert(SEGMENT_BYTES_PROPERTY.to_string(), segment_bytes.to_string());
  }
  Ok(Topic { kafka_properties, partitions, replication_factor })
}

struct TopicDelete {}

#[async_trait]
impl CommandExecutor for TopicDelete {
  async fn execute_with_client(&self, target: Option<String>, _: Option<String>, _: &ArgMatches, client: &DshApiClient, context: &Context) -> DshCliResult<()> {
    let topic_id = target.unwrap_or_else(|| unreachable!());
    if client.get_topic(&topic_id).await.is_err() {
      return err!("scratch topic '{}' does not exists", topic_id);
    }
    let topic_dependants = client.topic_dependants(&topic_id).await?;
    if context.dependencies_warning("topic", topic_dependants, &topic_id) && !context.confirmed("do you want to continue?")? {
      context.print_outcome(format!("cancelled, topic '{}' not deleted", topic_id));
      return Ok(());
    }
    if context.confirmed(format!("delete scratch topic '{}'?", topic_id))? {
      if context.dry_run() {
        context.print_warning("dry-run mode, scratch topic not deleted");
      } else {
        client.delete_topic_configuration(&topic_id).await?;
        context.print_outcome(format!("scratch topic '{}' deleted", topic_id));
      }
    } else {
      context.print_outcome(format!("cancelled, scratch topic '{}' not deleted", topic_id));
    }
    Ok(())
  }

  fn requirements(&self, _: &ArgMatches) -> Requirements {
    Requirements::standard_with_api()
  }
}

pub(crate) static TOPIC_LABELS: [TopicLabel; 9] = [
  TopicLabel::Target,
  TopicLabel::NumberOfPartitions,
  TopicLabel::ReplicationFactor,
  TopicLabel::CleanupPolicy,
  TopicLabel::TimestampType,
  TopicLabel::MaxMessageBytes,
  TopicLabel::SegmentBytes,
  TopicLabel::Notifications,
  TopicLabel::Provisioned,
];

struct TopicList {}

#[async_trait]
impl CommandExecutor for TopicList {
  async fn execute_with_client(&self, _: Option<String>, _: Option<String>, _: &ArgMatches, client: &DshApiClient, context: &Context) -> DshCliResult<()> {
    context.print_explanation("list all scratch topics with their configurations");
    let start_instant = context.now();
    let topic_ids = client.get_topic_ids().await?;
    let configurations = try_join_all(topic_ids.iter().map(|topic_id| client.get_topic(topic_id))).await?;
    context.print_execution_time(start_instant);
    let mut formatter = ListFormatter::new(&TOPIC_LABELS, context);
    formatter.push_target_ids_and_values(topic_ids.as_slice(), configurations.as_slice());
    formatter.print(None)?;
    Ok(())
  }

  fn requirements(&self, _: &ArgMatches) -> Requirements {
    Requirements::standard_with_api()
  }
}

struct TopicListAllocationStatus {}

#[async_trait]
impl CommandExecutor for TopicListAllocationStatus {
  async fn execute_with_client(&self, _: Option<String>, _: Option<String>, _: &ArgMatches, client: &DshApiClient, context: &Context) -> DshCliResult<()> {
    context.print_explanation("list all scratch topics with their allocation status");
    let start_instant = context.now();
    let topic_ids = client.get_topic_ids().await?;
    let allocation_statuses = try_join_all(topic_ids.iter().map(|topic_id| client.get_topic_status(topic_id))).await?;
    context.print_execution_time(start_instant);
    let mut formatter = ListFormatter::new(&DEFAULT_ALLOCATION_STATUS_LABELS, context);
    formatter.push_target_ids_and_values(topic_ids.as_slice(), allocation_statuses.as_slice());
    formatter.print(None)?;
    Ok(())
  }

  fn requirements(&self, _: &ArgMatches) -> Requirements {
    Requirements::standard_with_api()
  }
}

struct TopicListIds {}

#[async_trait]
impl CommandExecutor for TopicListIds {
  async fn execute_with_client(&self, _: Option<String>, _: Option<String>, _: &ArgMatches, client: &DshApiClient, context: &Context) -> DshCliResult<()> {
    context.print_explanation("list all scratch topic ids");
    let start_instant = context.now();
    let topic_ids = client.get_topic_ids().await?;
    context.print_execution_time(start_instant);
    let mut formatter = IdsFormatter::new("topic id", context);
    formatter.push_target_ids(&topic_ids);
    formatter.print(Some(OutputFormat::Plain))?;
    Ok(())
  }

  fn requirements(&self, _: &ArgMatches) -> Requirements {
    Requirements::standard_with_api()
  }
}

struct TopicListUsage {}

#[async_trait]
impl CommandExecutor for TopicListUsage {
  async fn execute_with_client(&self, _: Option<String>, _: Option<String>, _: &ArgMatches, client: &DshApiClient, context: &Context) -> DshCliResult<()> {
    context.print_explanation("list all scratch topics with the services and apps that use them");
    let start_instant = context.now();
    let topics_with_dependants: Vec<(String, Vec<Dependant<TopicInjection>>)> = client.topics_with_dependants().await?;
    context.print_execution_time(start_instant);
    if topics_with_dependants.is_empty() {
      context.print_outcome("no scratch topics found in services");
    } else {
      let mut formatter = ListFormatter::new(&DEPENDANT_LABELS_LIST, context);
      for (topic_id, dependants) in &topics_with_dependants {
        for dependant in dependants {
          formatter.push_target_id_value(topic_id.clone(), dependant);
        }
      }
      formatter.print(None)?;
    }
    Ok(())
  }

  fn requirements(&self, _: &ArgMatches) -> Requirements {
    Requirements::standard_with_api()
  }
}

static TOPIC_STATUS_LABELS: [TopicLabel; 14] = [
  TopicLabel::Target,
  TopicLabel::NumberOfPartitions,
  TopicLabel::ReplicationFactor,
  TopicLabel::CleanupPolicy,
  TopicLabel::CompressionType,
  TopicLabel::DeleteRetentionMs,
  TopicLabel::TimestampType,
  TopicLabel::MaxMessageBytes,
  TopicLabel::SegmentBytes,
  TopicLabel::RetentionBytes,
  TopicLabel::RetentionMs,
  TopicLabel::Notifications,
  TopicLabel::Provisioned,
  TopicLabel::KafkaProperties,
];

struct TopicShow {}

#[async_trait]
impl CommandExecutor for TopicShow {
  async fn execute_with_client(&self, target: Option<String>, _: Option<String>, _: &ArgMatches, client: &DshApiClient, context: &Context) -> DshCliResult<()> {
    let topic_id = target.unwrap_or_else(|| unreachable!());
    context.print_explanation(format!("show the configuration for scratch topic '{}'", topic_id));
    let start_instant = context.now();
    let topic = client.get_topic(&topic_id).await?;
    context.print_execution_time(start_instant);
    context.print_allocation_status(&Ok(topic.status.clone()), TOPIC_SUBJECT_TARGET);
    UnitFormatter::new(topic_id, &TOPIC_STATUS_LABELS, context).print(&topic, None)
  }

  fn requirements(&self, _: &ArgMatches) -> Requirements {
    Requirements::standard_with_api()
  }
}

struct TopicShowAllocationStatus {}

#[async_trait]
impl CommandExecutor for TopicShowAllocationStatus {
  async fn execute_with_client(&self, target: Option<String>, _: Option<String>, _: &ArgMatches, client: &DshApiClient, context: &Context) -> DshCliResult<()> {
    let topic_id = target.unwrap_or_else(|| unreachable!());
    context.print_explanation(format!("show the allocation status for scratch topic '{}'", topic_id));
    let start_instant = context.now();
    let allocation_status = client.get_topic_status(&topic_id).await?;
    context.print_execution_time(start_instant);
    UnitFormatter::new(topic_id, &DEFAULT_ALLOCATION_STATUS_LABELS, context).print(&allocation_status, None)
  }

  fn requirements(&self, _: &ArgMatches) -> Requirements {
    Requirements::standard_with_api()
  }
}

static PROPERTY_LABELS: [PropertyLabel; 2] = [PropertyLabel::Property, PropertyLabel::Value];

struct TopicShowProperties {}

#[async_trait]
impl CommandExecutor for TopicShowProperties {
  async fn execute_with_client(&self, target: Option<String>, _: Option<String>, _: &ArgMatches, client: &DshApiClient, context: &Context) -> DshCliResult<()> {
    let topic_id = target.unwrap_or_else(|| unreachable!());
    context.print_explanation(format!("show the properties for scratch topic '{}'", topic_id));
    let start_instant = context.now();
    let topic_status = client.get_topic(&topic_id).await?;
    context.print_execution_time(start_instant);
    match topic_status.actual {
      Some(actual_topic) => {
        let mut pairs: Vec<(String, String)> = actual_topic.kafka_properties.into_iter().collect_vec();
        pairs.sort_by(|(key_a, _), (key_b, _)| key_a.cmp(key_b));
        let (properties, values): (Vec<String>, Vec<String>) = pairs.into_iter().unzip();
        let mut formatter = ListFormatter::new(&PROPERTY_LABELS, context);
        formatter.push_target_ids_and_values(&properties, &values);
        formatter.print(None)?;
      }
      None => context.print_outcome("no actual topics"),
    }
    Ok(())
  }

  fn requirements(&self, _: &ArgMatches) -> Requirements {
    Requirements::standard_with_api()
  }
}

struct TopicShowUsage {}

#[async_trait]
impl CommandExecutor for TopicShowUsage {
  async fn execute_with_client(&self, target: Option<String>, _: Option<String>, _: &ArgMatches, client: &DshApiClient, context: &Context) -> DshCliResult<()> {
    let topic_id = target.unwrap_or_else(|| unreachable!());
    context.print_explanation(format!("show the services that use scratch topic '{}'", topic_id));
    let start_instant = context.now();
    let dependants: Vec<Dependant<TopicInjection>> = client.topic_dependants(&topic_id).await?;
    context.print_execution_time(start_instant);
    if dependants.is_empty() {
      context.print_outcome("topic not used")
    } else {
      let mut formatter = ListFormatter::new(&DEPENDANT_LABELS, context);
      formatter.push_values(&dependants);
      formatter.print(None)?;
    }
    Ok(())
  }

  fn requirements(&self, _: &ArgMatches) -> Requirements {
    Requirements::standard_with_api()
  }
}

#[derive(Eq, Hash, PartialEq, Serialize)]
pub(crate) enum TopicLabel {
  CleanupPolicy,
  CompressionType,
  DeleteRetentionMs,
  #[allow(dead_code)]
  DerivedFrom,
  KafkaProperties,
  MaxMessageBytes,
  Notifications,
  NumberOfPartitions,
  Provisioned,
  ReplicationFactor,
  RetentionBytes,
  RetentionMs,
  SegmentBytes,
  Target,
  TimestampType,
}

impl Label for TopicLabel {
  fn as_str(&self) -> &str {
    match self {
      Self::CleanupPolicy => "cleanup policy",
      Self::CompressionType => "compression type",
      Self::DeleteRetentionMs => "delete retention",
      Self::DerivedFrom => "derived from",
      Self::KafkaProperties => "kafka properties",
      Self::MaxMessageBytes => "max message bytes",
      Self::Notifications => "notifications",
      Self::NumberOfPartitions => "number of partitions",
      Self::Provisioned => "provisioned",
      Self::ReplicationFactor => "replication factor",
      Self::RetentionBytes => "retention bytes",
      Self::RetentionMs => "retention ms",
      Self::SegmentBytes => "segment bytes",
      Self::Target => "topic id",
      Self::TimestampType => "timestamp type",
    }
  }

  fn as_str_for_list(&self) -> &str {
    match self {
      Self::CleanupPolicy => "cleanup",
      Self::CompressionType => "compr",
      Self::DeleteRetentionMs => "del ret",
      Self::DerivedFrom => "derived",
      Self::KafkaProperties => "props",
      Self::MaxMessageBytes => "max bytes",
      Self::Notifications => "not",
      Self::NumberOfPartitions => "part",
      Self::Provisioned => "prov",
      Self::ReplicationFactor => "repl",
      Self::RetentionBytes => "ret bytes",
      Self::RetentionMs => "ret ms",
      Self::SegmentBytes => "seg bytes",
      Self::Target => "topic id",
      Self::TimestampType => "ts",
    }
  }

  fn is_target_label(&self) -> bool {
    matches!(self, Self::Target)
  }

  fn column_alignment(&self) -> ColumnAlignment {
    match self {
      TopicLabel::DeleteRetentionMs
      | TopicLabel::MaxMessageBytes
      | TopicLabel::NumberOfPartitions
      | TopicLabel::ReplicationFactor
      | TopicLabel::RetentionBytes
      | TopicLabel::RetentionMs
      | TopicLabel::SegmentBytes => ColumnAlignment::Right,
      _ => ColumnAlignment::default(),
    }
  }
}

impl SubjectFormatter<TopicLabel> for Topic {
  fn value(&self, label: &TopicLabel, target_id: &str) -> Value {
    match label {
      TopicLabel::CleanupPolicy => Value::some_or_hide(self.kafka_properties.get(CLEANUP_POLICY_PROPERTY)),
      TopicLabel::DerivedFrom => Value::unreachable(),
      TopicLabel::KafkaProperties => Value::plain(hashmap_to_table(&get_implicit_properties(&self.kafka_properties))),
      TopicLabel::MaxMessageBytes => Value::some_or_hide(self.kafka_properties.get(MAX_MESSAGE_BYTES_PROPERTY)),
      TopicLabel::Notifications => Value::unreachable(),
      TopicLabel::NumberOfPartitions => Value::plain(self.partitions),
      TopicLabel::Provisioned => Value::unreachable(),
      TopicLabel::ReplicationFactor => Value::plain(self.replication_factor),
      TopicLabel::SegmentBytes => Value::some_or_hide(self.kafka_properties.get(SEGMENT_BYTES_PROPERTY)),
      TopicLabel::Target => Value::target(target_id),
      TopicLabel::TimestampType => Value::some_or_hide(self.kafka_properties.get(MESSAGE_TIMESTAMP_PROPERTY)),
      TopicLabel::CompressionType => Value::some_or_hide(self.kafka_properties.get(COMPRESSION_TYPE_PROPERTY)),
      TopicLabel::DeleteRetentionMs => Value::some_or_hide(self.kafka_properties.get(DELETE_RETENTION_MS_PROPERTY)),
      TopicLabel::RetentionBytes => Value::some_or_hide(self.kafka_properties.get(RETENTION_BYTES_PROPERTY)),
      TopicLabel::RetentionMs => Value::some_or_hide(self.kafka_properties.get(RETENTION_MS_PROPERTY)),
    }
  }
}

pub(crate) fn get_implicit_properties(kafka_properties: &HashMap<String, String>) -> HashMap<&String, &String> {
  const EXPLICIT_VALUES: [&str; 8] = [
    CLEANUP_POLICY_PROPERTY,
    COMPRESSION_TYPE_PROPERTY,
    DELETE_RETENTION_MS_PROPERTY,
    MAX_MESSAGE_BYTES_PROPERTY,
    MESSAGE_TIMESTAMP_PROPERTY,
    RETENTION_BYTES_PROPERTY,
    RETENTION_MS_PROPERTY,
    SEGMENT_BYTES_PROPERTY,
  ];
  kafka_properties
    .iter()
    .filter(|(key, _)| !EXPLICIT_VALUES.contains(&key.as_str()))
    .collect::<HashMap<_, _>>()
}

impl SubjectFormatter<TopicLabel> for TopicStatus {
  fn value(&self, label: &TopicLabel, target_id: &str) -> Value {
    match label {
      TopicLabel::CleanupPolicy => Value::some_or_hide(self.actual.as_ref().and_then(|topic| topic.kafka_properties.get(CLEANUP_POLICY_PROPERTY))),
      TopicLabel::CompressionType => Value::some_or_hide(self.actual.as_ref().and_then(|topic| topic.kafka_properties.get(COMPRESSION_TYPE_PROPERTY))),
      TopicLabel::DeleteRetentionMs => Value::some_or_hide(self.actual.as_ref().and_then(|topic| topic.kafka_properties.get(DELETE_RETENTION_MS_PROPERTY))),
      TopicLabel::DerivedFrom => Value::some_or_hide(self.status.derived_from.as_ref()),
      TopicLabel::KafkaProperties => Value::some_or_hide(
        self
          .actual
          .as_ref()
          .map(|topic| hashmap_to_table(&get_implicit_properties(&topic.kafka_properties))),
      ),
      TopicLabel::MaxMessageBytes => Value::some_or_hide(self.actual.as_ref().and_then(|topic| topic.kafka_properties.get(MAX_MESSAGE_BYTES_PROPERTY))),
      TopicLabel::Notifications => Value::warn(self.status.notifications.iter().map(|notification| notification.to_string()).join("\n")),
      TopicLabel::NumberOfPartitions => Value::some_or_hide(self.actual.as_ref().map(|topic| topic.partitions)),
      TopicLabel::Provisioned => Value::plain(self.status.provisioned),
      TopicLabel::ReplicationFactor => Value::some_or_hide(self.actual.as_ref().map(|topic| topic.replication_factor)),
      TopicLabel::RetentionBytes => Value::some_or_hide(self.actual.as_ref().and_then(|topic| topic.kafka_properties.get(RETENTION_BYTES_PROPERTY))),
      TopicLabel::RetentionMs => Value::some_or_hide(self.actual.as_ref().and_then(|topic| topic.kafka_properties.get(RETENTION_MS_PROPERTY))),
      TopicLabel::SegmentBytes => Value::some_or_hide(self.actual.as_ref().and_then(|topic| topic.kafka_properties.get(SEGMENT_BYTES_PROPERTY))),
      TopicLabel::Target => Value::target(target_id),
      TopicLabel::TimestampType => Value::some_or_hide(self.actual.as_ref().and_then(|topic| topic.kafka_properties.get(MESSAGE_TIMESTAMP_PROPERTY))),
    }
  }
}

#[derive(Eq, Hash, PartialEq, Serialize)]
enum PropertyLabel {
  Property,
  Value,
}

impl Label for PropertyLabel {
  fn as_str(&self) -> &str {
    match self {
      PropertyLabel::Property => "property",
      PropertyLabel::Value => "value",
    }
  }

  fn is_target_label(&self) -> bool {
    matches!(self, Self::Property)
  }
}
