import { Box, Stack, Typography } from '@tidbcloud/uikit'
import { PropertyCard } from '@tidbcloud/uikit/biz'
import { Fragment } from 'react'

import { showTidbExtension } from 'dbaas/features/Changefeed'
import { getConnectionMethodDisplayName } from 'dbaas/features/Changefeed/connection'
import {
  AuthTypeOptionMap,
  getTopicDistributionModeOptions,
  KafkaCompressionOptions,
  KafkaVersionOptions,
  partitionDistributionOptions
} from 'dbaas/features/Changefeed/constants'
import { getEndpointServiceNameLabel } from 'dbaas/features/Changefeed/helpers'
import {
  ConsoleapiChangefeed,
  ConsoleapiSinkEndpoint,
  DataflowDispatchType,
  DataflowNetworkType
} from 'dbaas/services/dataflow/interface'

import { DataFormatProperties } from '../ChangefeedDetail/DataFormatProperties'
import { StartPositionProperty } from '../ChangefeedDetail/StartPositionDisplay'
import { TableConfigProperties } from '../ChangefeedDetail/TableFilterProperties'

interface KafkaPropertiesPremiumProps {
  data: ConsoleapiChangefeed
  sinkEndpointData?: ConsoleapiSinkEndpoint
  cloudProvider: string
}

export const KafkaPropertiesPremium = ({ data, sinkEndpointData, cloudProvider }: KafkaPropertiesPremiumProps) => {
  const networkInfo = data.request?.network_info
  const networkType = networkInfo?.network_type
  const brokerEndpoints = data.request?.kafka_info?.broker?.broker_endpoints
  const advertisedListener = sinkEndpointData?.advertised_listener
  const advertisedRecords = advertisedListener?.records ?? []

  const renderAdvertisedListener = () => {
    if (!advertisedListener) {
      return <Typography>-</Typography>
    }

    return (
      <Stack gap={4}>
        <Typography>{advertisedListener.root_domain_name}</Typography>
        {advertisedRecords.length > 0 ? (
          advertisedRecords.map((record, idx) => {
            const domainParts = [record?.subdomain_name, advertisedListener.root_domain_name].filter(Boolean)
            const domain = domainParts.join('.')
            const key = `${record.zone ?? 'default'}-${record.subdomain_name ?? idx}`
            return (
              <Typography key={key} c="carbon.7">
                {domain}
                {record.zone ? ` (${record.zone})` : ''}
              </Typography>
            )
          })
        ) : (
          <Typography>-</Typography>
        )}
      </Stack>
    )
  }

  return (
    <>
      <PropertyCard.Divider />
      <PropertyCard.Item label="Connection Method">
        {networkType ? getConnectionMethodDisplayName(networkType as DataflowNetworkType) : '-'}
      </PropertyCard.Item>
      <PropertyCard.Item label="Kafka Version">
        {KafkaVersionOptions.find((i) => i.value === data.request?.kafka_info?.broker?.kafka_version)?.label || '-'}
      </PropertyCard.Item>
      <PropertyCard.Item label="Compress">
        {KafkaCompressionOptions.find((i) => i.value === data.request?.kafka_info?.broker?.compression)?.label || '-'}
      </PropertyCard.Item>

      {(networkType === 'NetworkType_Public' || networkType === 'NetworkType_VPC_Peering') && (
        <PropertyCard.Item label="Broker Endpoints">{brokerEndpoints || '-'}</PropertyCard.Item>
      )}

      {networkType === 'NetworkType_Private_Link' && (
        <>
          {sinkEndpointData?.endpoint_name && (
            <PropertyCard.Item label="Private Endpoint Name">{sinkEndpointData.endpoint_name}</PropertyCard.Item>
          )}
          {sinkEndpointData?.endpoint_service_name && (
            <PropertyCard.Item label={getEndpointServiceNameLabel(cloudProvider)}>
              {sinkEndpointData.endpoint_service_name}
            </PropertyCard.Item>
          )}
          <PropertyCard.Item label="Advertised Listener" align="start">
            {renderAdvertisedListener()}
          </PropertyCard.Item>
          <PropertyCard.Item label="Bootstrap Ports">
            {networkInfo?.ports?.length ? networkInfo.ports.join(', ') : '-'}
          </PropertyCard.Item>
        </>
      )}
      <PropertyCard.Item label="TLS Encryption">
        {data.request?.kafka_info?.broker?.tls_enable ? 'Enabled' : 'Disabled'}
      </PropertyCard.Item>
      {data.request?.kafka_info?.broker?.tls_enable && (
        <PropertyCard.Item label="Insecure Skip Verify">
          {data.request?.kafka_info?.broker?.insecure_skip_verify ? 'Enabled' : 'Disabled'}
        </PropertyCard.Item>
      )}
      <PropertyCard.Item label="Authentication">
        {AuthTypeOptionMap[data.request?.kafka_info?.authentication?.auth_type!] || '-'}
      </PropertyCard.Item>
      {data.request?.kafka_info?.authentication?.userName && (
        <PropertyCard.Item label="User Name">{data.request.kafka_info.authentication.userName}</PropertyCard.Item>
      )}

      <PropertyCard.Divider />

      <TableConfigProperties data={data.request?.kafka_info?.table_config} />

      {data.request?.kafka_info?.column_selectors?.length ? (
        <PropertyCard.Item label="Column Selector">
          <Stack gap={4}>
            {data.request.kafka_info.column_selectors.map((selector, idx) => (
              <Typography variant="body-lg" key={idx} sx={{ wordBreak: 'break-all' }}>
                <b>{selector.matcher?.join(', ') ?? '-'}</b>: {selector.columns?.join(', ') ?? '-'}
              </Typography>
            ))}
          </Stack>
        </PropertyCard.Item>
      ) : null}

      <StartPositionProperty data={data.request?.kafka_info?.start_position} />

      <DataFormatProperties data={data.request?.kafka_info?.data_format} />

      {data.request?.kafka_info?.data_format?.protocol &&
        showTidbExtension(data.request.kafka_info.data_format.protocol) && (
          <PropertyCard.Item label="TiDB Extension">
            {data.request?.kafka_info?.data_format?.enable_tidb_extension ? 'Enable' : 'Disable'}
          </PropertyCard.Item>
        )}

      <PropertyCard.Item label="Topic Distribution Mode">
        {getTopicDistributionModeOptions({ target: 'Kafka' }).find(
          (i) => i.value === data.request?.kafka_info?.topic_partition_config?.dispatch_type
        )?.label || '-'}
      </PropertyCard.Item>

      <PropertyCard.Item label="Topic Name">
        {data.request?.kafka_info?.topic_partition_config?.default_topic || '-'}
      </PropertyCard.Item>

      {data.request?.kafka_info?.topic_partition_config?.dispatch_type !==
        DataflowDispatchType.DISPATCH_TYPE_ONE_TOPIC && (
        <>
          {data.request?.kafka_info?.topic_partition_config?.topic_prefix && (
            <PropertyCard.Item label="Topic Prefix">
              {data.request.kafka_info.topic_partition_config.topic_prefix}
            </PropertyCard.Item>
          )}
          {data.request?.kafka_info?.topic_partition_config?.separator &&
            data.request?.kafka_info?.topic_partition_config?.dispatch_type ===
              DataflowDispatchType.DISPATCH_TYPE_BY_TABLE && (
              <PropertyCard.Item label="Topic Separator">
                {data.request.kafka_info.topic_partition_config.separator}
              </PropertyCard.Item>
            )}
          {data.request?.kafka_info?.topic_partition_config?.topic_suffix && (
            <PropertyCard.Item label="Topic Suffix">
              {data.request.kafka_info.topic_partition_config.topic_suffix}
            </PropertyCard.Item>
          )}
        </>
      )}

      {data.request?.kafka_info?.topic_partition_config?.partition_dispatchers && (
        <PropertyCard.Item label="Partition Dispatcher">
          <Box sx={{ display: 'grid', gridTemplateColumns: 'fit-content(50%) 1fr', gap: 4 }}>
            {data.request.kafka_info.topic_partition_config.partition_dispatchers.map((i, index) => (
              <Fragment key={index}>
                <Typography>
                  {i.matcher?.join(',')}
                  {' :'}
                </Typography>

                <Typography>
                  {[
                    partitionDistributionOptions.find((j) => j.value === i.partition_type)?.text,
                    i.index_name,
                    i.columns?.join(',')
                  ]
                    .filter(Boolean)
                    .join(', ')}
                </Typography>
              </Fragment>
            ))}
          </Box>
        </PropertyCard.Item>
      )}

      <PropertyCard.Item label="Replication Factor">
        {data.request?.kafka_info?.topic_partition_config?.replication_factor || '-'}
      </PropertyCard.Item>

      <PropertyCard.Item label="Partition Number">
        {data.request?.kafka_info?.topic_partition_config?.partition_num || '-'}
      </PropertyCard.Item>

      <PropertyCard.Item label="Output Raw Events">
        {data.request?.kafka_info?.data_format?.output_raw_change_event ? 'Yes' : 'No'}
      </PropertyCard.Item>
    </>
  )
}