import { Divider, Group, Input, Loader, Stack } from '@tidbcloud/uikit'
import {
  FormNumberInput,
  FormPasswordInput,
  FormRadioGroup,
  FormSelect,
  FormSwitch,
  FormTextInput
} from '@tidbcloud/uikit/biz'
import { IconLinkExternal02 } from '@tidbcloud/uikit/icons'
import { useEffect, useMemo } from 'react'
import { useFormContext } from 'react-hook-form'

import { SharedCluster } from 'dbaas/features/ClusterManagement/cluster'
import { useOrgProjectClusterIds } from 'dbaas/features/Common'
import HeadLabel from 'dbaas/screens/ClustersV2/new/components/HeadLabel'
import { usePrivateLinkConnectionServiceListPrivateLinkConnections } from 'dbaas/services/api-clients/legacy/serverless'
import {
  KafkaAuthTypeEnum,
  KafkaCompressionTypeEnum,
  KafkaNetworkTypeEnum,
  KafkaVersionEnum,
  PrivateLinkConnectionTypeEnum
} from 'dbaas/services/api-clients/legacy/serverless/interface'

import type { CreateOrEditChangefeedEssentialFormData } from '../types'

const NETWORK_TYPE_OPTIONS = [
  { label: 'Public', value: KafkaNetworkTypeEnum.PUBLIC },
  { label: 'Private Link', value: KafkaNetworkTypeEnum.PRIVATE_LINK }
]

const KAFKA_VERSION_OPTIONS = [
  { label: 'Kafka v2', value: KafkaVersionEnum.VERSION_2XX },
  { label: 'Kafka v3', value: KafkaVersionEnum.VERSION_3XX }
]

const KAFKA_COMPRESSION_OPTIONS = [
  { label: 'None', value: KafkaCompressionTypeEnum.NONE },
  { label: 'Gzip', value: KafkaCompressionTypeEnum.GZIP },
  { label: 'Snappy', value: KafkaCompressionTypeEnum.SNAPPY },
  { label: 'Lz4', value: KafkaCompressionTypeEnum.LZ4 },
  { label: 'Zstd', value: KafkaCompressionTypeEnum.ZSTD }
]

const KAFKA_AUTH_OPTIONS = [
  { label: 'Disable', value: KafkaAuthTypeEnum.DISABLE },
  { label: 'SASL/PLAIN', value: KafkaAuthTypeEnum.SASL_PLAIN },
  { label: 'SASL/SCRAM-SHA-256', value: KafkaAuthTypeEnum.SASL_SCRAM_SHA_256 },
  { label: 'SASL/SCRAM-SHA-512', value: KafkaAuthTypeEnum.SASL_SCRAM_SHA_512 }
]

export const FormKafkaEssential = ({ cluster }: { cluster: SharedCluster }) => {
  const { orgId, projectId } = useOrgProjectClusterIds()
  const { clearErrors, setValue, watch } = useFormContext<CreateOrEditChangefeedEssentialFormData>()
  const networkType = watch('kafka.network.network_type')
  const authType = watch('kafka.authentication.auth_type')
  const enableTls = watch('kafka.authentication.enable_tls')
  const privateLinkConnectionId = watch('kafka.network.private_link.private_link_connection_id')

  const needsCredentials = authType !== KafkaAuthTypeEnum.DISABLE

  const { data: privateLinkConnections, isLoading: isLoadingPrivateLinkConnections } =
    usePrivateLinkConnectionServiceListPrivateLinkConnections(
      orgId,
      projectId,
      cluster.id!,
      {
        page_size: 50,
        page_token: '1',
        state: 'ACTIVE'
      },
      {
        query: {
          enabled: networkType === 'PRIVATE_LINK' && !!orgId && !!projectId && !!cluster.id,
          refetchOnWindowFocus: false
        }
      }
    )

  const privateLinkConnectionType = useMemo(() => {
    return privateLinkConnections?.private_link_connections?.find(
      (connection) => connection.private_link_connection_id === privateLinkConnectionId
    )?.type
  }, [privateLinkConnectionId, privateLinkConnections?.private_link_connections])
  const isAmazonMskPrivateLink =
    networkType === KafkaNetworkTypeEnum.PRIVATE_LINK &&
    privateLinkConnectionType === PrivateLinkConnectionTypeEnum.AMAZON_MSK_PROVISIONED
  const authOptions = isAmazonMskPrivateLink
    ? KAFKA_AUTH_OPTIONS.filter((option) => option.value === KafkaAuthTypeEnum.SASL_SCRAM_SHA_512)
    : KAFKA_AUTH_OPTIONS
  const shouldShowBootstrapPort =
    networkType === KafkaNetworkTypeEnum.PRIVATE_LINK && !!privateLinkConnectionId && !isAmazonMskPrivateLink

  useEffect(() => {
    if (!shouldShowBootstrapPort) {
      setValue('kafka.network.private_link.port', undefined)
      clearErrors('kafka.network.private_link.port')
    }
  }, [clearErrors, setValue, shouldShowBootstrapPort])

  useEffect(() => {
    if (isAmazonMskPrivateLink) {
      setValue('kafka.authentication.auth_type', KafkaAuthTypeEnum.SASL_SCRAM_SHA_512)
      setValue('kafka.authentication.enable_tls', true)
      setValue('kafka.authentication.server_name', undefined)
      clearErrors(['kafka.authentication.auth_type', 'kafka.authentication.server_name'])
    }
  }, [clearErrors, isAmazonMskPrivateLink, setValue])

  return (
    <Stack p={24} pt={0}>
      <Divider mx={-24} />

      <HeadLabel label="Connection" />

      <FormRadioGroup
        name="kafka.network.network_type"
        label="Connectivity Method"
        data={NETWORK_TYPE_OPTIONS}
        rules={{ required: 'Required' }}
      />

      {networkType === KafkaNetworkTypeEnum.PUBLIC && (
        <FormTextInput
          name="kafka.network.public_endpoints"
          label="Bootstrap Servers"
          description="Enter the Kafka servers in the format of host:port, separated by commas."
          placeholder="broker1:9092,broker2:9092"
          rules={{ required: 'Required' }}
        />
      )}

      {networkType === KafkaNetworkTypeEnum.PRIVATE_LINK && (
        <Group grow wrap="nowrap" align="flex-start">
          <FormSelect
            name="kafka.network.private_link.private_link_connection_id"
            label="Private Link Connection"
            placeholder="Select a Private Link Connection"
            data={
              privateLinkConnections?.private_link_connections?.map((connection) => ({
                label: connection.display_name,
                value: connection.private_link_connection_id!
              })) || []
            }
            rightSection={isLoadingPrivateLinkConnections ? <Loader size="xs" /> : null}
            inputWrapperOrder={['label', 'input', 'description', 'error']}
            description={
              <span>
                Didn't find one?{' '}
                <a href={`/clusters/${cluster.id}/networking`} target="_blank" rel="noopener noreferrer">
                  create a Private Link Connection here <IconLinkExternal02 sx={{ position: 'relative', top: 1 }} />
                </a>
              </span>
            }
            rules={{ required: 'Required' }}
          />
          {shouldShowBootstrapPort && (
            <FormNumberInput
              name="kafka.network.private_link.port"
              label="Bootstrap Port"
              placeholder="Enter the port number"
              rules={{ required: 'Required' }}
            />
          )}
        </Group>
      )}

      <Divider mx={-24} />

      <HeadLabel label="Authentication" />

      <FormRadioGroup
        name="kafka.authentication.auth_type"
        data={authOptions}
        mt={-8}
        rules={{ required: 'Required' }}
      />

      {needsCredentials && (
        <Group grow align="flex-start">
          <FormTextInput
            name="kafka.authentication.username"
            label="Username"
            placeholder="Enter the username"
            rules={{ required: 'Required' }}
          />
          <FormPasswordInput
            name="kafka.authentication.password"
            label="Password"
            type="password"
            placeholder="Enter the password"
            rules={{ required: 'Required' }}
          />
        </Group>
      )}

      <Divider mx={-24} />

      <HeadLabel label="Advanced Settings" />

      <FormSelect
        name="kafka.broker.kafka_version"
        label="Kafka Version"
        data={KAFKA_VERSION_OPTIONS}
        placeholder="Select the Kafka version"
        rules={{ required: 'Required' }}
      />
      <FormSelect
        name="kafka.broker.compression"
        label="Compression"
        data={KAFKA_COMPRESSION_OPTIONS}
        placeholder="Select the compression type"
        rules={{ required: 'Required' }}
      />

      {!isAmazonMskPrivateLink && (
        <Stack gap={8}>
          <Input.Label>TLS Encryption</Input.Label>
          <FormSwitch name="kafka.authentication.enable_tls" />
        </Stack>
      )}

      {!isAmazonMskPrivateLink && networkType !== 'PUBLIC' && enableTls && (
        <FormTextInput
          name="kafka.authentication.server_name"
          label="TLS Server Name (optional)"
          description="Specify this field when your server relies on TLS SNI, for example, Confluent Cloud Dedicated clusters."
          placeholder="Enter the TLS server name"
        />
      )}
    </Stack>
  )
}