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

import { showTidbExtension } from 'dbaas/features/Changefeed'
import { DocLinks, getFormatOptions } from 'dbaas/features/Changefeed/constants'
import { canConfigureEventSplitting } from 'dbaas/features/Changefeed/helpers'
import { SharedCluster } from 'dbaas/features/ClusterManagement/cluster'
import { renderDescriptiveSelectOption } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/renderDescriptiveSelectOption'
import { FormEventSplit } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/step2/FormEventSplit'
import { FormPartitionDistribution } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/step2/FormPartitionDistribution'
import HeadLabel from 'dbaas/screens/ClustersV2/new/components/HeadLabel'
import {
  DescribeSchemaTableResp,
  KafkaProtocolEnum,
  TopicDispatchTypeEnum
} from 'dbaas/services/api-clients/legacy/serverless/interface'

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

import { FormStartPositionEssential } from './FormStartPosition'

const getTopicDistributionModeOptions = (options?: { format?: KafkaProtocolEnum }) => {
  const { format } = options || {}
  return [
    {
      label: `Send all changelogs to one specified Kafka Topic`,
      value: TopicDispatchTypeEnum.ONE_TOPIC,
      enabled: !format || format !== KafkaProtocolEnum.AVRO,
      desc: 'All tables will be distributed to the same topic.'
    },
    {
      label: `Distribute changelogs by table to Kafka Topics`,
      value: TopicDispatchTypeEnum.BY_TABLE,
      enabled: true,
      desc: `Each table's changelogs are sent to a separate topic. In other words, each topic contains only one table's changelogs.`
    },
    {
      label: `Distribute changelogs by database to Kafka Topics`,
      value: TopicDispatchTypeEnum.BY_DATABASE,
      enabled: !format || format !== KafkaProtocolEnum.AVRO,
      desc: 'Multiple tables under one database will be distributed to the same topic.'
    }
  ]
}

const DECIMAL_HANDLING_OPTIONS = [
  { label: 'Precise', value: 'PRECISE' },
  { label: 'String', value: 'STRING' }
]

const UNSIGNED_BIGINT_OPTIONS = [
  { label: 'Long', value: 'LONG' },
  { label: 'String', value: 'STRING' }
]

export const FormKafkaReplicationEssential = ({
  cluster,
  schemaTableData
}: {
  cluster: SharedCluster
  schemaTableData?: DescribeSchemaTableResp
}) => {
  const { watch, control, setValue } = useFormContext<CreateOrEditChangefeedEssentialFormData>()
  const dispatchType = watch('kafka.topic_partition_config.dispatch_type')
  const protocol = watch('kafka.data_format.protocol')
  const schemaHttpAuth = watch('kafka.data_format.avro_config.confluent_schema_registry.enable_http_auth')
  const partitionDispatchers = watch('kafka.topic_partition_config.partition_dispatchers')
  const formatOptions = useMemo(() => getFormatOptions('8').filter((i) => i.enabled), [])
  const topicDistributionModeOptions = useMemo(
    () =>
      getTopicDistributionModeOptions({
        format: protocol
      }).filter((i) => i.enabled),
    [protocol]
  )
  const showEventSplitting = cluster.version ? canConfigureEventSplitting(cluster.version) : false
  const isOneTopic = dispatchType === TopicDispatchTypeEnum.ONE_TOPIC
  const isByTable = dispatchType === TopicDispatchTypeEnum.BY_TABLE
  const isByDatabase = dispatchType === TopicDispatchTypeEnum.BY_DATABASE
  const topicPrefix = watch('kafka.topic_partition_config.topic_prefix')
  const topicSeparator = watch('kafka.topic_partition_config.separator')
  const topicSuffix = watch('kafka.topic_partition_config.topic_suffix')

  useEffect(() => {
    if (!partitionDispatchers || partitionDispatchers.length === 0) {
      setValue('kafka.topic_partition_config.partition_dispatchers', [
        {
          matcher: ['*.*'],
          partition_type: 'TABLE'
        }
      ])
    }
  }, [partitionDispatchers, setValue])

  return (
    <>
      <Stack p={24}>
        <HeadLabel label="Start Replication Position" />
        <FormStartPositionEssential schemaTableData={schemaTableData} />
      </Stack>

      <Divider />

      <Stack p={24}>
        <HeadLabel label="Data Format" />

        <Group grow align="flex-start">
          <FormSelect
            name="kafka.data_format.protocol"
            data={formatOptions.map((i) => ({
              value: i.value,
              label: i.text
            }))}
            description={
              <span
                dangerouslySetInnerHTML={{
                  __html: formatOptions.find((i) => i.value === protocol)?.description ?? ''
                }}
              />
            }
            inputWrapperOrder={['label', 'input', 'description', 'error']}
            rules={{ required: 'Required' }}
          />
        </Group>

        {protocol === 'DEBEZIUM' && (
          <>
            <Controller
              name="kafka.data_format.debezium_config.disable_output_old_value"
              control={control}
              render={({ field }) => (
                <Switch
                  label="Include Old Values in Debezium Output"
                  checked={!field.value}
                  onChange={(event) => field.onChange(!event.target.checked)}
                />
              )}
            />

            <Controller
              name="kafka.data_format.debezium_config.disable_schema"
              control={control}
              render={({ field }) => (
                <Switch
                  label="Include Schema Description in Debezium Output"
                  checked={!field.value}
                  onChange={(event) => field.onChange(!event.target.checked)}
                />
              )}
            />
          </>
        )}

        {protocol === 'AVRO' && (
          <>
            <FormRadioGroup
              name="kafka.data_format.avro_config.decimal_handling_mode"
              label="Decimal"
              data={DECIMAL_HANDLING_OPTIONS}
              rules={{ required: 'Required' }}
            />
            <FormRadioGroup
              name="kafka.data_format.avro_config.bigint_unsigned_handling_mode"
              label="Unsigned BigInt"
              data={UNSIGNED_BIGINT_OPTIONS}
              rules={{ required: 'Required' }}
            />
          </>
        )}

        {showTidbExtension(protocol) && (
          <Stack gap={8}>
            <Typography variant="label-lg">TiDB Extension</Typography>

            <FormSwitch
              name="kafka.data_format.enable_tidb_extension"
              label={
                <Typography>
                  Add some TiDB-extension fields to the message body. Also see{' '}
                  <a
                    href={
                      protocol === KafkaProtocolEnum.AVRO
                        ? DocLinks.avroExtension
                        : protocol === KafkaProtocolEnum.CANAL_JSON
                          ? DocLinks.canalJsonExtension
                          : ''
                    }
                    target="_blank"
                    rel="noopener noreferrer"
                    data-mp-event="TiDB Extension Fields Link Clicked"
                  >
                    TiDB extension fields
                  </a>
                  .
                </Typography>
              }
            />
          </Stack>
        )}

        {protocol === 'AVRO' && (
          <>
            <Divider mx={-24} />

            <HeadLabel label="Schema Registry" />

            <FormTextInput
              label="Schema Registry Endpoint"
              name="kafka.data_format.avro_config.confluent_schema_registry.endpoint"
              rules={{ required: 'Required' }}
              placeholder="Enter your schema registry endpoint"
            />

            <FormSwitch
              name="kafka.data_format.avro_config.confluent_schema_registry.enable_http_auth"
              label="HTTP Authentication"
            />

            {schemaHttpAuth && (
              <Group grow>
                <FormTextInput
                  label="User Name"
                  name="kafka.data_format.avro_config.confluent_schema_registry.username"
                  rules={{ required: 'Required' }}
                  placeholder="Enter your schema registry user name"
                  w={300}
                />

                <FormPasswordInput
                  label="Password"
                  name="kafka.data_format.avro_config.confluent_schema_registry.password"
                  rules={{ required: 'Required' }}
                  placeholder="Enter your schema registry password"
                  w={300}
                />
              </Group>
            )}
          </>
        )}
      </Stack>

      <Divider />

      <Stack p={24}>
        <HeadLabel label="Topic Distribution" />

        <FormSelect
          name="kafka.topic_partition_config.dispatch_type"
          label="Distribution Mode"
          data={topicDistributionModeOptions.map((i) => ({ ...i, description: i.desc }))}
          renderOption={renderDescriptiveSelectOption}
          inputWrapperOrder={['label', 'input', 'description', 'error']}
          rules={{ required: 'Required' }}
        />

        <FormTextInput
          label={
            <Input.Label mb={0}>
              <Group gap={8}>
                <span>{isOneTopic ? 'Topic Name' : `Default Topic Name`}</span>
                {!isOneTopic && (
                  <HoverCard>
                    <HoverCard.Target>
                      <IconInfoCircle size={14} />
                    </HoverCard.Target>
                    <HoverCard.Dropdown maw={400}>
                      {`The changelogs of non-row events, such as ${
                        isByDatabase ? `Resolved Event` : `Create Schema Event`
                      }, are sent to the default topic.`}
                    </HoverCard.Dropdown>
                  </HoverCard>
                )}
              </Group>
            </Input.Label>
          }
          name="kafka.topic_partition_config.default_topic"
          rules={{
            required: 'Required'
          }}
          placeholder={isOneTopic ? `Enter the topic name` : `Enter the default topic name`}
        />

        {dispatchType !== TopicDispatchTypeEnum.ONE_TOPIC && (
          <>
            <Group justify="space-between">
              <FormTextInput
                label="Topic Prefix"
                name="kafka.topic_partition_config.topic_prefix"
                placeholder="Enter prefix"
                sx={{ flex: 1 }}
              />

              {isByTable && (
                <FormTextInput
                  label="Separator"
                  name="kafka.topic_partition_config.separator"
                  placeholder="Enter separator"
                  sx={{ flex: 1 }}
                />
              )}

              <FormTextInput
                label="Topic Suffix"
                name="kafka.topic_partition_config.topic_suffix"
                placeholder="Enter separator"
                sx={{ flex: 1 }}
              />
            </Group>

            <Typography c="carbon.7" variant="body-lg">
              For example, a table in `db1` named `tbl2` will have its row change events sent to topic:
              <Typography span ml={4} fw={500}>
                {[topicPrefix, ...(isByDatabase ? ['db1'] : ['db1', topicSeparator, 'tbl2']), topicSuffix]
                  .filter(Boolean)
                  .map((i, index) => {
                    if (index % 2 === 0) {
                      return (
                        <Typography span c="cyan.6" key={`${i}-${index}`}>
                          {i}
                        </Typography>
                      )
                    }
                    return <span key={`${i}-${index}`}>{i}</span>
                  })}
              </Typography>
            </Typography>
          </>
        )}
      </Stack>

      <Divider />

      <Stack p={24}>
        <HeadLabel label="Partition Distribution" />
        <FormPartitionDistribution target="Kafka" name="kafka.topic_partition_config.partition_dispatchers" />
      </Stack>

      <Divider />

      <Stack p={24}>
        <HeadLabel label="Topic Configuration" />
        <FormTextInput
          name="kafka.topic_partition_config.replication_factor"
          label="Replication Factor"
          type="number"
          placeholder="Enter the replication factor"
          rules={{ required: 'Required' }}
        />
        <FormNumberInput
          name="kafka.topic_partition_config.partition_num"
          label="Partition Number"
          rules={{ required: 'Required' }}
        />
      </Stack>

      {showEventSplitting && (
        <>
          <Divider />
          <Stack p={24}>
            <HeadLabel label="Event Splitting" />
            <FormEventSplit name="kafka.output_raw_change_event" />
          </Stack>
        </>
      )}
    </>
  )
}