import { Divider, Group, Stack, Switch, Typography } from '@tidbcloud/uikit'
import {
  FormNumberInput,
  FormPasswordInput,
  FormRadioGroup,
  FormSelect,
  FormSwitch,
  FormTextInput
} from '@tidbcloud/uikit/biz'
import { 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 { SharedCluster } from 'dbaas/features/ClusterManagement/cluster'
import { renderDescriptiveSelectOption } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/renderDescriptiveSelectOption'
import {
  DataflowDescribeSchemaTableResp,
  DataflowDispatchType,
  DataflowProtocol
} from 'dbaas/services/dataflow/interface'

import { CardSection } from './CardSection'
import { FormEventSplit } from './FormEventSplit'
import { FormPartitionDistribution } from './FormPartitionDistribution'
import { FormStartPosition } from './FormStartPosition'
import { FormTopicFormat } from './FormTopicFormat'
import { FormTopicName } from './FormTopicName'

export function FormKafka({
  cluster,
  schemaTableData
}: {
  cluster: SharedCluster
  schemaTableData?: DataflowDescribeSchemaTableResp
}) {
  const { setValue, watch } = useFormContext()
  const formatOptions = useMemo(() => getFormatOptions(cluster.version).filter((i) => i.enabled), [cluster])
  const protocol = watch('kafka.data_format.protocol') as DataflowProtocol
  const schemaHttpAuth: boolean = watch('kafka.data_format.schema_registry.enableHttpAuth')
  const topicDispatcher = watch('kafka.topic_partition_config.dispatch_type') as DataflowDispatchType
  const disableSchema: boolean = watch('kafka.data_format.debezium_config.disable_schema')

  const topicDistributionModeOptions = useMemo(
    () =>
      getTopicDistributionModeOptions({ clusterVersion: cluster.version, format: protocol, target: 'Kafka' }).filter(
        (i) => i.enabled
      ),
    [cluster, protocol]
  )

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

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

        {protocol === DataflowProtocol.DEBEZIUM && (
          <>
            <FormSwitch
              name="kafka.data_format.debezium_config.output_old_value"
              label="Include Old Values in Debezium Output"
            />

            <Controller
              name="kafka.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.data_format.debezium_config.disable_schema', !e.target.checked)
              }}
            />
          </>
        )}

        {protocol === DataflowProtocol.AVRO && (
          <>
            <FormRadioGroup
              name="kafka.data_format.decimal_handling_mode"
              label="Decimal"
              data={HandleDecimalOptions.map((i) => ({
                value: i.value,
                label: i.text
              }))}
            />

            <FormRadioGroup
              name="kafka.data_format.bigint_unsigned_handling_mode"
              label="Unsigned BigInt"
              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.data_format.enable_tidb_extension"
              label={
                <Typography>
                  Add some TiDB-extension fields to the message body. Also see{' '}
                  <a
                    href={
                      protocol === DataflowProtocol.AVRO
                        ? DocLinks.avroExtension
                        : protocol === DataflowProtocol.CANAL_JSON
                          ? DocLinks.canalJsonExtension
                          : ''
                    }
                    target="_blank"
                    rel="noopener noreferrer"
                    data-mp-event="TiDB Extension Fields Link Clicked"
                  >
                    TiDB extension fields
                  </a>
                  .
                </Typography>
              }
            />
          </Stack>
        )}
      </CardSection>

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

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

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

                  <FormPasswordInput
                    label="Password"
                    name="kafka.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.topic_partition_config.dispatch_type"
          data={topicDistributionModeOptions.map((i) => ({ ...i, description: i.desc }))}
          renderOption={renderDescriptiveSelectOption}
          inputWrapperOrder={['label', 'input', 'description', 'error']}
        />

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

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

      <CardSection title="Partition Distribution">
        <FormPartitionDistribution target="Kafka" name="kafka.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.topic_partition_config.replication_factor"
            min={1}
          />
          <FormNumberInput
            label="Partition Number"
            description="Number of partitions for topics automatically created by this changefeed."
            name="kafka.topic_partition_config.partition_num"
            min={1}
          />
        </Stack>
      </CardSection>

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