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

import { showTidbExtension } from 'dbaas/features/Changefeed'
import {
  DocLinks,
  getFormatOptions,
  getTopicDistributionModeOptions,
  HandleDecimalOptions,
  HandleUnsignedBigIntOptions
} from 'dbaas/features/Changefeed/constants'
import { canConfigureEventSplitting } from 'dbaas/features/Changefeed/helpers'
import { getChangefeedClusterMetadata } from 'dbaas/features/Changefeed/types'
import { renderDescriptiveSelectOption } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/renderDescriptiveSelectOption'
import { CardSection } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/step2/CardSection'
import { FormEventSplit } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/step2/FormEventSplit'
import { FormPartitionDistribution } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/step2/FormPartitionDistribution'
import { FormTopicFormat } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/step2/FormTopicFormat'
import { FormTopicName } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/step2/FormTopicName'
import type { SharedTiDB } from 'dbaas/screens/NextGen/shared/model/shared-tidb-state'
import { useGetExpWhiteList } from 'dbaas/services/api'
import {
  ConsoleapiBigintUnsignedHandlingMode,
  ConsoleapiDecimalHandlingMode,
  ConsoleapiDescribeSchemaTableResp,
  ConsoleapiDispatchType,
  ConsoleapiProtocol
} from 'dbaas/services/dataflow/interface'
import { ExpFeatHashmap } from 'dbaas/stores/localStorage'
import useStores from 'dbaas/stores/useStores'
import { compareVersion } from 'dbaas/utils/chore'

import { CreateOrEditChangefeedFormData } from '../types'

import { FormStartPosition } from './FormStartPosition'

export function FormKafka({
  cluster,
  schemaTableData
}: {
  cluster: SharedTiDB
  schemaTableData?: ConsoleapiDescribeSchemaTableResp
}) {
  const { setValue, watch } = useFormContext<CreateOrEditChangefeedFormData>()
  const {
    store: { tenantData }
  } = useStores()
  const { version } = useMemo(() => getChangefeedClusterMetadata(cluster), [cluster])
  const { data: isDebeziumAvroEnabled = false } = useGetExpWhiteList(tenantData.id, undefined, {
    query: {
      enabled: !!tenantData.id,
      select: (data) => data?.exp_names?.includes(ExpFeatHashmap.PremiumDebeziumAvro) ?? false,
      refetchOnMount: false,
      refetchOnWindowFocus: false
    }
  })

  const formatOptions = useMemo(
    () =>
      [
        ...getFormatOptions(version),
        {
          text: 'Debezium-Avro',
          value: ConsoleapiProtocol.DEBEZIUM_AVRO,
          enabled: isDebeziumAvroEnabled && (version ? compareVersion(version, '6.1.0', '>=') : true),
          description:
            'Debezium-Avro is a Debezium-compatible, schema-registered, binary CDC protocol with a structured message envelope. Widely used in Confluent-based data streaming ecosystems. Also see <a href="https://docs.pingcap.com/tidb/stable/ticdc-debezium-avro/" target="_blank" rel="noopener noreferrer">Debezium-Avro Protocol</a>.'
        }
      ].filter((i) => i.enabled),
    [isDebeziumAvroEnabled, version]
  )
  const protocol = watch('kafka_info.data_format.protocol')
  const schemaHttpAuth: boolean = watch('kafka_info.data_format.schema_registry.enableHttpAuth')!
  const topicDispatcher = watch('kafka_info.topic_partition_config.dispatch_type')
  const disableSchema: boolean = watch('kafka_info.data_format.debezium_config.disable_schema')!
  const isDebeziumAvroProtocol = protocol === ConsoleapiProtocol.DEBEZIUM_AVRO
  const isAvroProtocol = protocol === ConsoleapiProtocol.AVRO || protocol === ConsoleapiProtocol.DEBEZIUM_AVRO
  const isDebeziumProtocol = protocol === ConsoleapiProtocol.DEBEZIUM

  useEffect(() => {
    if (isDebeziumAvroProtocol) {
      setValue(
        'kafka_info.data_format.decimal_handling_mode',
        ConsoleapiDecimalHandlingMode.DECIMAL_HANDLING_MODE_PRECISE
      )
      setValue(
        'kafka_info.data_format.bigint_unsigned_handling_mode',
        ConsoleapiBigintUnsignedHandlingMode.BIGINT_UNSIGNED_HANDLING_MODE_LONG
      )
      setValue('kafka_info.topic_partition_config.dispatch_type', ConsoleapiDispatchType.DISPATCH_TYPE_BY_TABLE)
    }
  }, [isDebeziumAvroProtocol, setValue])

  const topicDistributionModeOptions = useMemo(
    () =>
      getTopicDistributionModeOptions({ clusterVersion: version, format: protocol, target: 'Kafka' }).filter(
        (i) =>
          i.enabled ||
          (isDebeziumAvroProtocol &&
            (i.value === ConsoleapiDispatchType.DISPATCH_TYPE_ONE_TOPIC ||
              i.value === ConsoleapiDispatchType.DISPATCH_TYPE_BY_DATABASE))
      ),
    [isDebeziumAvroProtocol, protocol, version]
  )

  return (
    <>
      <CardSection title="Start Replication Position">
        <FormStartPosition name="kafka_info.start_position" schemaTableData={schemaTableData} />
      </CardSection>

      <CardSection title="Data Format">
        <FormSelect
          name="kafka_info.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']}
        />

        {isDebeziumProtocol && (
          <>
            <FormSwitch
              name="kafka_info.data_format.debezium_config.output_old_value"
              label="Include Old Values in Debezium Output"
            />

            <Controller
              name="kafka_info.data_format.debezium_config.disable_schema"
              render={({ field }) => <input hidden {...field} />}
            />

            <Switch
              label="Include Schema Description in Debezium Output"
              defaultChecked={!disableSchema}
              onChange={(e) => {
                setValue('kafka_info.data_format.debezium_config.disable_schema', !e.target.checked)
              }}
            />
          </>
        )}

        {isAvroProtocol && (
          <>
            <FormRadioGroup
              name="kafka_info.data_format.decimal_handling_mode"
              label="Decimal Handling"
              data={HandleDecimalOptions.map((i) => ({
                value: i.value,
                label: i.text
              }))}
            />

            <FormRadioGroup
              name="kafka_info.data_format.bigint_unsigned_handling_mode"
              label="Unsigned BigInt Handling"
              data={HandleUnsignedBigIntOptions.map((i) => ({
                value: i.value,
                label: i.text
              }))}
            />
          </>
        )}

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

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

      {isAvroProtocol && (
        <>
          <CardSection title="Schema Registry">
            <FormTextInput
              label="Schema Registry Endpoint"
              name="kafka_info.data_format.schema_registry.schema_registry_endpoints"
              rules={{ required: 'Required' }}
              placeholder="Enter you schema registry endpoint"
            />

            <FormSwitch name="kafka_info.data_format.schema_registry.enableHttpAuth" label="HTTP Authentication" />

            {schemaHttpAuth && (
              <>
                <Group>
                  <FormTextInput
                    label="User Name"
                    name="kafka_info.data_format.schema_registry.userName"
                    rules={{ required: 'Required' }}
                    placeholder="Enter your schema registry user name"
                    w={300}
                  />

                  <FormPasswordInput
                    label="Password"
                    name="kafka_info.data_format.schema_registry.password"
                    rules={{ required: 'Required' }}
                    placeholder="Enter your schema registry password"
                    w={300}
                  />
                </Group>
              </>
            )}
          </CardSection>
          <Divider />
        </>
      )}

      <CardSection title="Topic Distribution">
        <FormSelect
          label="Distribution Mode"
          name="kafka_info.topic_partition_config.dispatch_type"
          data={topicDistributionModeOptions.map((i) => ({ ...i, description: i.desc, disabled: !i.enabled }))}
          renderOption={renderDescriptiveSelectOption}
          inputWrapperOrder={['label', 'input', 'description', 'error']}
        />

        <FormTopicName
          name="kafka_info.topic_partition_config"
          tooltip={`The changelogs of non-row events, such as ${
            topicDispatcher === ConsoleapiDispatchType.DISPATCH_TYPE_BY_DATABASE
              ? `Resolved Event`
              : `Create Schema Event`
          }, are sent to the default topic.`}
        />

        {topicDispatcher !== ConsoleapiDispatchType.DISPATCH_TYPE_ONE_TOPIC && (
          <FormTopicFormat name="kafka_info.topic_partition_config" />
        )}
      </CardSection>

      <CardSection title="Partition Distribution">
        <FormPartitionDistribution target="Kafka" name="kafka_info.topic_partition_config.partition_dispatchers" />
      </CardSection>

      <CardSection title="Topic Configuration">
        <Stack>
          <FormNumberInput
            label="Replication Factor"
            description="Number of servers that will replicate each written message."
            name="kafka_info.topic_partition_config.replication_factor"
            min={1}
          />
          <FormNumberInput
            label="Partition Number"
            description="Number of partitions for topics automatically created by this changefeed."
            name="kafka_info.topic_partition_config.partition_num"
            min={1}
          />
        </Stack>
      </CardSection>

      {canConfigureEventSplitting(version) && (
        <CardSection title="Event Splitting">
          <FormEventSplit name={`kafka_info.data_format.output_raw_change_event`} />
        </CardSection>
      )}
    </>
  )
}