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 } from 'react'
import { Controller, 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 { SharedCluster } from 'dbaas/features/ClusterManagement/cluster'
import { renderDescriptiveSelectOption } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/renderDescriptiveSelectOption'
import { HeadLabel } from 'dbaas/screens/ClustersV2/new/components'
import { CreatePrivateEndpointForChangefeed } from 'dbaas/screens/ClusterV2/features/Network/dedicated/PrivateEndpointChangefeed/Create'
import { useListSinkEndpoints } from 'dbaas/services/dataflow'
import { DataflowKafkaAuthType, DataflowNetworkType } from 'dbaas/services/dataflow/interface'
import { compareVersion } from 'dbaas/utils/chore'

import {
  buildKafkaPrivateEndpointOptions,
  validateKafkaPrivateEndpointValue
} from './kafkaPrivateEndpointOptionHelpers'
import { renderKafkaPrivateEndpointOption } from './kafkaPrivateEndpointOptions'
import { ConfigureTargetForm } from './types'

export function FormKafka({ cluster, orgId, projectId }: { cluster: SharedCluster; orgId: string; projectId: string }) {
  const { watch, setValue, control, unregister } = useFormContext<ConfigureTargetForm>()
  const showTlsToggle = useMemo(() => compareVersion(cluster.version, '6.1.0', '>='), [cluster])
  const showAuthOptions = showTlsToggle
  const networkType = watch('network_info.network_type')
  const isPrivateLink =
    networkType === DataflowNetworkType.NetworkType_Private_Link ||
    networkType === DataflowNetworkType.NetworkType_Private_Service_Connect
  const authType = watch('kafka.authentication.auth_type')
  const tlsEnabled = watch('kafka.broker.tls_enable')
  const showUsernameAndPassword = authType !== DataflowKafkaAuthType.DISABLE
  const isEditing = useIsEditingChangefeed()
  const connectionMethods = getChangefeedConnectionMethods(cluster.provider)
  const [privateEndpointOpened, privateEndpointHandler] = useDisclosure()
  const { data, isLoading, refetch } = useListSinkEndpoints(
    orgId,
    projectId,
    {
      provider: cluster.provider,
      region: cluster.region
    },
    {
      query: {
        enabled: cluster?.computed.isDedicated && isPrivateLink,
        refetchOnMount: false,
        refetchOnWindowFocus: false
      }
    }
  )
  const privateEndpointOptions = useMemo(() => {
    return buildKafkaPrivateEndpointOptions(data?.sink_endpoints)
  }, [data])

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

  return (
    <>
      <Divider />

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

        <Stack>
          <Controller name="kafka.provider" control={control} render={({ field }) => <input {...field} hidden />} />

          <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#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="network_info.ports_str"
                label="Bootstrap Ports"
                disabled={isEditing}
                rules={{
                  required: 'Required'
                }}
                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.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 Changefeed"
          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>
          {showAuthOptions && <FormRadioGroup name="kafka.authentication.auth_type" data={AuthTypeOptions} />}

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

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

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

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

          <Group gap={32}>
            {showTlsToggle && <FormSwitch name="kafka.broker.tls_enable" label="TLS Encryption" labelPosition="left" />}

            {showTlsToggle && tlsEnabled && (
              <FormSwitch
                name="kafka.broker.insecure_skip_verify"
                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>
                }
                labelPosition="left"
              />
            )}
          </Group>
        </Stack>
      </Stack>
    </>
  )
}