import { Box, Divider, Group, HoverCard, Input, Loader, Modal, Stack, Typography } from '@tidbcloud/uikit'
import { FormPasswordInput, FormRadioGroup, FormSelect, FormSwitch, FormTextInput } from '@tidbcloud/uikit/biz'
import { useDisclosure } from '@tidbcloud/uikit/hooks'
import { IconInfoCircle } from '@tidbcloud/uikit/icons'
import { useEffect, useMemo, useRef } from 'react'
import { useFormContext } from 'react-hook-form'

import { useIsEditingChangefeed } from 'dbaas/features/Changefeed'
import { endpointValidator, getChangefeedConnectionMethods } from 'dbaas/features/Changefeed/connection'
import { AuthTypeOptions, KafkaCompressionOptions, KafkaVersionOptions } from 'dbaas/features/Changefeed/constants'
import { renderDescriptiveSelectOption } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/renderDescriptiveSelectOption'
import type { CreateOrEditChangefeedFormData } from 'dbaas/screens/Changefeed/CreateOrEditChangefeedPremium/types'
import { HeadLabel } from 'dbaas/screens/ClustersV2/new/components'
import { CreatePrivateEndpointForChangefeed } from 'dbaas/screens/NextGen/premium/networking/private-endpoint-changefeed/Create'
import { SharedTiDB } from 'dbaas/screens/NextGen/shared/model/shared-tidb-state'
import { useNgListSinkEndpoints } from 'dbaas/services/dataflow'
import { ConsoleapiKafkaAuthType, ConsoleapiNetworkType } from 'dbaas/services/dataflow/interface'

import {
  buildKafkaPrivateEndpointOptions,
  getBootstrapPortsValidationRules,
  isAmazonMskPrivateEndpoint,
  validateKafkaPrivateEndpointValue
} from '../../CreateOrEditChangefeed/step1/kafkaPrivateEndpointOptionHelpers'
import { renderKafkaPrivateEndpointOption } from '../../CreateOrEditChangefeed/step1/kafkaPrivateEndpointOptions'

const getConnectionMethods = (cluster: SharedTiDB) => {
  const connectionMethods = getChangefeedConnectionMethods(cluster.cloudProvider!)

  if (cluster.computed.isEssentialV2) {
    return connectionMethods.filter((method) => method.value !== ConsoleapiNetworkType.NetworkType_VPC_Peering)
  }

  return connectionMethods
}

export function FormKafka({ cluster }: { cluster: SharedTiDB }) {
  const { watch, setValue, unregister, clearErrors } = useFormContext<CreateOrEditChangefeedFormData>()
  const networkType = watch('network_info.network_type')
  const sinkEndpointId = watch('network_info.sink_endpoint_id')
  const isPrivateLink = networkType === ConsoleapiNetworkType.NetworkType_Private_Link
  const authType = watch('kafka_info.authentication.auth_type')
  const tlsEnabled = watch('kafka_info.broker.tls_enable')
  const showUsernameAndPassword = authType !== ConsoleapiKafkaAuthType.DISABLE
  const isEditing = useIsEditingChangefeed()
  const connectionMethods = getConnectionMethods(cluster)
  const [privateEndpointOpened, privateEndpointHandler] = useDisclosure()
  const canUseSinkEndpoints = cluster?.computed.isPremium || cluster?.computed.isBYOC || cluster?.computed.isEssentialV2
  const { data, isLoading, refetch } = useNgListSinkEndpoints(cluster.tidbId!, undefined, {
    query: {
      enabled: canUseSinkEndpoints && !!cluster.tidbId && isPrivateLink,
      refetchOnMount: false,
      refetchOnWindowFocus: false
    }
  })
  const privateEndpointOptions = useMemo(() => {
    return buildKafkaPrivateEndpointOptions(data?.sink_endpoints)
  }, [data])
  const isAmazonMskEndpoint = isPrivateLink && isAmazonMskPrivateEndpoint(data?.sink_endpoints, sinkEndpointId)
  const authTypeOptions = isAmazonMskEndpoint
    ? AuthTypeOptions.filter((option) => option.value === ConsoleapiKafkaAuthType.SASL_SCRAM_SHA_512)
    : AuthTypeOptions
  const previousNetworkType = useRef(networkType)

  useEffect(() => {
    if (previousNetworkType.current === networkType) {
      return
    }

    previousNetworkType.current = networkType
    setValue('network_info.sink_endpoint_id', undefined)
    setValue('network_info.ports', undefined)
    setValue('ports_str', '')
    setValue('kafka_info.broker.broker_endpoints', '')
    setValue('kafka_info.authentication.auth_type', ConsoleapiKafkaAuthType.DISABLE)
    setValue('kafka_info.authentication.userName', '')
    setValue('kafka_info.authentication.password', '')
    setValue('kafka_info.broker.tls_enable', false)
    setValue('kafka_info.broker.insecure_skip_verify', false)
    clearErrors([
      'network_info.sink_endpoint_id',
      'ports_str',
      'kafka_info.broker.broker_endpoints',
      'kafka_info.authentication.auth_type',
      'kafka_info.authentication.userName',
      'kafka_info.authentication.password'
    ])
  }, [clearErrors, networkType, setValue])

  useEffect(() => {
    if (!tlsEnabled) {
      unregister('kafka_info.broker.insecure_skip_verify')
    }
  }, [tlsEnabled, unregister])

  useEffect(() => {
    if (isAmazonMskEndpoint) {
      setValue('ports_str', '')
      setValue('network_info.ports', undefined)
      setValue('kafka_info.authentication.auth_type', ConsoleapiKafkaAuthType.SASL_SCRAM_SHA_512)
      setValue('kafka_info.broker.tls_enable', true)
      setValue('kafka_info.broker.insecure_skip_verify', false)
      clearErrors('ports_str')
    }
  }, [clearErrors, isAmazonMskEndpoint, setValue])

  return (
    <>
      <Divider />

      <Stack m={24} maw={1000}>
        <HeadLabel label="Connection" />

        <Stack>
          <FormSelect
            name="network_info.network_type"
            label={
              <Input.Label mb={0}>
                <Group gap={8}>
                  <span>Connectivity Method</span>
                  <HoverCard shadow="md" withArrow withinPortal width={400} position="top">
                    <HoverCard.Target>
                      <IconInfoCircle size={14} />
                    </HoverCard.Target>
                    <HoverCard.Dropdown>
                      <Typography variant="body-lg">
                        Set the Network so that TiDB Cloud can connect to your Kafka server. Refer to{' '}
                        <a
                          href="https://docs.pingcap.com/tidbcloud/changefeed-sink-to-apache-kafka/?plan=premium#network"
                          target="_blank"
                          rel="noreferrer"
                        >
                          documentation
                        </a>{' '}
                        for more details.
                      </Typography>
                    </HoverCard.Dropdown>
                  </HoverCard>
                </Group>
              </Input.Label>
            }
            data={connectionMethods}
            disabled={isEditing}
          />

          {isPrivateLink ? (
            <Group align="flex-start">
              <FormSelect
                label="Private Endpoint"
                data={privateEndpointOptions}
                disabled={isEditing}
                rules={{
                  required: 'Required',
                  validate: (value) => validateKafkaPrivateEndpointValue(value, privateEndpointOptions)
                }}
                name="network_info.sink_endpoint_id"
                placeholder="Select a Private Endpoint"
                searchable
                nothingFoundMessage="No available Private Endpoint found."
                renderOption={renderKafkaPrivateEndpointOption}
                description={
                  <Input.Description>
                    Didn't find one?{' '}
                    <Box component="a" onClick={privateEndpointHandler.open}>
                      Create a Private Endpoint here
                    </Box>
                  </Input.Description>
                }
                rightSection={isLoading ? <Loader size="xs" /> : null}
                inputWrapperOrder={['label', 'input', 'description', 'error']}
                sx={{ flex: 1 }}
              />

              <FormTextInput
                name="ports_str"
                label="Bootstrap Ports"
                disabled={isEditing || isAmazonMskEndpoint}
                rules={getBootstrapPortsValidationRules(isAmazonMskEndpoint)}
                placeholder="Multiple ports separated by commas, i.e. 9002, 9003"
                inputWrapperOrder={['label', 'input', 'description', 'error']}
                onChange={(e) => {
                  const ports = Array.from(
                    new Set(
                      e.target.value
                        .split(',')
                        .map((i: string) => Number(i.trim()))
                        .filter((i) => !isNaN(i))
                    )
                  )
                  setValue('network_info.ports', ports)
                }}
                sx={{ flex: 1 }}
              />
            </Group>
          ) : (
            <FormTextInput
              label="Brokers Endpoints"
              name="kafka_info.broker.broker_endpoints"
              rules={{
                required: 'Required',
                validate: (val) => endpointValidator(val)
              }}
              disabled={isEditing}
              placeholder="Enter the broker endpoints. eg: 10.0.0.1:9092,10.0.0.2:9092"
            />
          )}
        </Stack>

        <Modal
          title="Create Private Endpoint for External Services"
          opened={privateEndpointOpened}
          onClose={privateEndpointHandler.close}
          size="xl"
          closeOnClickOutside={false}
        >
          <CreatePrivateEndpointForChangefeed
            cluster={cluster}
            onSuccess={(id) => {
              setValue('network_info.sink_endpoint_id', id)
            }}
            onCancel={() => {
              privateEndpointHandler.close()
              refetch()
            }}
          />
        </Modal>
      </Stack>

      <Divider />

      <Stack m={24} gap={8} maw={1000}>
        <HeadLabel label="Authentication" />

        <Stack>
          <FormRadioGroup
            name="kafka_info.authentication.auth_type"
            data={authTypeOptions}
            rules={{
              validate: (value) =>
                !isAmazonMskEndpoint || value === ConsoleapiKafkaAuthType.SASL_SCRAM_SHA_512 || 'Required'
            }}
          />

          {showUsernameAndPassword && (
            <Group w="100%">
              <FormTextInput
                name="kafka_info.authentication.userName"
                label="User Name"
                rules={{ required: true }}
                placeholder="Enter the user name of your Kafka account"
                styles={() => ({ root: { flex: 1 } })}
              />

              <FormPasswordInput
                name="kafka_info.authentication.password"
                label="Password"
                rules={{ required: true }}
                placeholder="Enter the password of your Kafka account"
                styles={() => ({ root: { flex: 1 } })}
              />
            </Group>
          )}
        </Stack>
      </Stack>

      <Divider />

      <Stack m={24} maw={1000}>
        <HeadLabel label="Advanced Settings" />

        <FormSelect
          label="Kafka Version"
          name="kafka_info.broker.kafka_version"
          data={KafkaVersionOptions}
          renderOption={renderDescriptiveSelectOption}
        />

        <FormSelect label="Compression" name="kafka_info.broker.compression" data={KafkaCompressionOptions} />

        {!isAmazonMskEndpoint && (
          <Group gap={32}>
            <Group gap={8}>
              <Input.Label>TLS Encryption</Input.Label>
              <FormSwitch name="kafka_info.broker.tls_enable" />
            </Group>

            {tlsEnabled && (
              <Group gap={8}>
                <Input.Label>
                  <Group gap={2}>
                    Insecure Skip Verify
                    <HoverCard width={400}>
                      <HoverCard.Target>
                        <IconInfoCircle size={14} />
                      </HoverCard.Target>
                      <HoverCard.Dropdown>
                        <Typography variant="body-lg">
                          Determines whether the server's identity is verified during the TLS connection.
                        </Typography>
                      </HoverCard.Dropdown>
                    </HoverCard>
                  </Group>
                </Input.Label>
                <FormSwitch name="kafka_info.broker.insecure_skip_verify" />
              </Group>
            )}
          </Group>
        )}
      </Stack>
    </>
  )
}