import { Box, Button, Card, Divider, Group, Stack, Typography } from '@tidbcloud/uikit'
import { FormErrorMessage, FormSelect, FormTextInput } from '@tidbcloud/uikit/biz'
import { Fragment, ReactNode, useMemo } from 'react'
import { FormProvider, useFormContext } from 'react-hook-form'
import { useHistory } from 'react-router-dom'

import { ChangefeedNames, useIsEditingChangefeed } from 'dbaas/features/Changefeed'
import { partitionDistributionOptions } from 'dbaas/features/Changefeed/constants'
import { canConfigureEventSplitting } from 'dbaas/features/Changefeed/helpers'
import { SharedCluster } from 'dbaas/features/ClusterManagement/cluster'
import { getProtocolDisplayName } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/getProtocolDisplayName'
import { renderDescriptiveSelectOption } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/renderDescriptiveSelectOption'
import HeadLabel from 'dbaas/screens/ClustersV2/new/components/HeadLabel'
import {
  useChangefeedServiceCreateChangefeed,
  useChangefeedServiceEditChangefeed,
  useChangefeedServiceListChangefeedSpecifications
} from 'dbaas/services/api-clients/legacy/serverless'
import {
  ChangefeedTypeEnum,
  EventFilterRule,
  KafkaAuthTypeEnum,
  KafkaNetworkTypeEnum,
  MySQLNetworkTypeEnum,
  StartModeEnum,
  TopicDispatchTypeEnum
} from 'dbaas/services/api-clients/legacy/serverless/interface'
import { getErrorMessage } from 'dbaas/services/errorCodes'
import useStores from 'dbaas/stores/useStores'
import { eventTracking } from 'dbaas/utils/tracking'

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

import { buildCreateChangefeedPayload, buildEditChangefeedPayload } from './payloadBuilders'

function ReviewProperty({ label, value }: { label: ReactNode; value: ReactNode }) {
  return (
    <Group align="flex-start" wrap="nowrap">
      <Typography variant="body-lg" w={240} c="carbon.7" sx={{ flexShrink: 0 }}>
        {label}
      </Typography>
      <Typography c="carbon.9" sx={{ wordBreak: 'break-all', flexGrow: 1 }}>
        {value}
      </Typography>
    </Group>
  )
}

function EventFilterSummary({ rules }: { rules?: EventFilterRule[] }) {
  if (!rules?.length) {
    return null
  }

  return (
    <Stack gap={4}>
      {rules.map((rule, index) => (
        <Typography key={index} variant="body-lg" sx={{ wordBreak: 'break-all' }}>
          <b>{rule.matcher?.join(', ') || '-'}</b>
          {rule.ignore_event?.length ? `: Ignored events: ${rule.ignore_event.join(', ')}` : ''}
        </Typography>
      ))}
    </Stack>
  )
}

const KafkaNetworkTypeLabels: Record<KafkaNetworkTypeEnum, string> = {
  [KafkaNetworkTypeEnum.PUBLIC]: 'Public',
  [KafkaNetworkTypeEnum.PRIVATE_LINK]: 'Private Link'
}

const MySQLNetworkTypeLabels: Record<MySQLNetworkTypeEnum, string> = {
  [MySQLNetworkTypeEnum.PUBLIC]: 'Public',
  [MySQLNetworkTypeEnum.PRIVATE_LINK]: 'Private Link'
}

const AuthTypeLabels: Record<KafkaAuthTypeEnum, string> = {
  [KafkaAuthTypeEnum.DISABLE]: 'Disabled',
  [KafkaAuthTypeEnum.SASL_PLAIN]: 'SASL/PLAIN',
  [KafkaAuthTypeEnum.SASL_SCRAM_SHA_256]: 'SASL/SCRAM-SHA-256',
  [KafkaAuthTypeEnum.SASL_SCRAM_SHA_512]: 'SASL/SCRAM-SHA-512'
}

const TopicDispatchLabels: Record<TopicDispatchTypeEnum, string> = {
  [TopicDispatchTypeEnum.ONE_TOPIC]: 'Single Topic',
  [TopicDispatchTypeEnum.BY_TABLE]: 'By Table',
  [TopicDispatchTypeEnum.BY_DATABASE]: 'By Database'
}

const DecimalHandlingLabels: Record<'PRECISE' | 'STRING', string> = {
  PRECISE: 'Precise',
  STRING: 'String'
}

const UnsignedBigIntLabels: Record<'LONG' | 'STRING', string> = {
  LONG: 'Long',
  STRING: 'String'
}

const SchemaRegistryTypeLabels: Record<'CONFLUENT' | 'AWS_GLUE', string> = {
  CONFLUENT: 'Confluent Schema Registry',
  AWS_GLUE: 'AWS Glue Schema Registry'
}

const StartPositionModeLabels: Record<StartModeEnum, string> = {
  [StartModeEnum.FROM_NOW]: 'From now on',
  [StartModeEnum.FROM_TSO]: 'From a specific TSO',
  [StartModeEnum.FROM_TIME]: 'From a specific time'
}

const renderStartPosition = (data?: CreateOrEditChangefeedEssentialFormData['start_position']) => {
  if (!data?.mode) {
    return '-'
  }
  const label = StartPositionModeLabels[data.mode] ?? data.mode
  if (data.mode === StartModeEnum.FROM_TSO && data.tso) {
    return (
      <Group>
        <span>{label}</span>
        <span>{data.tso}</span>
      </Group>
    )
  }
  if (data.mode === StartModeEnum.FROM_TIME && data.time) {
    return (
      <Group>
        <span>{label}</span>
        <span>{data.time}</span>
      </Group>
    )
  }
  return label
}

export const ReviewEssential = ({
  onGoPrev,
  onCancel,
  cluster
}: {
  cluster: SharedCluster
  onGoPrev: () => void
  onCancel: () => void
}) => {
  const isEditing = useIsEditingChangefeed()
  const history = useHistory()
  const {
    store: { tenantData, activeProjectId }
  } = useStores()

  const createChangefeed = useChangefeedServiceCreateChangefeed()
  const editChangefeed = useChangefeedServiceEditChangefeed()
  const methods = useFormContext<CreateOrEditChangefeedEssentialFormData>()
  const { handleSubmit, setValue, watch } = methods
  const selectedRcu = watch('rcu')

  const formData = methods.getValues()
  const error = isEditing ? editChangefeed.error : createChangefeed.error

  const { data: specs, isLoading: isSpecLoading } = useChangefeedServiceListChangefeedSpecifications(
    tenantData.id,
    activeProjectId,
    cluster.id!,
    {
      query: {
        enabled: Boolean(tenantData.id && activeProjectId && cluster.id),
        refetchOnWindowFocus: false,
        retry: false,
        onSuccess(response) {
          const availableSpecs = response?.items?.filter((item) => item.rcu != null) ?? []
          if (!selectedRcu && availableSpecs.length > 0) {
            setValue('rcu', availableSpecs[0].rcu!.toString())
          }
        }
      }
    }
  )

  const specOptions = useMemo(() => {
    return (
      specs?.items
        ?.filter((item) => item.rcu != null)
        .map((item) => ({
          label: `${item.rcu}`,
          description: `The maximum replication performance is ${item.rps_limit ?? 0} rows/s`,
          value: item.rcu!.toString()
        })) ?? []
    )
  }, [specs?.items])

  const onCreate = async (data: CreateOrEditChangefeedEssentialFormData) => {
    try {
      const isTestEnv =
        process.env.NODE_ENV === 'test' || window.location.search.includes('e2e-test') || window.E2E_Test
      if (isTestEnv) {
        const payload = isEditing ? buildEditChangefeedPayload(data) : buildCreateChangefeedPayload(data)
        console.log(payload)
        console.log('[E2E Test]-[CreateChangefeed]-[Step3]-[Payload]-' + JSON.stringify(payload))
        return
      }
      if (isEditing) {
        const payload = buildEditChangefeedPayload(data)
        await editChangefeed.mutateAsync({
          orgId: tenantData.id,
          projectId: activeProjectId,
          clusterId: cluster.id!,
          changefeedId: data.changefeed_id!,
          data: payload
        })
        eventTracking(`Update changefeed successful`, {
          'Cluster Type': 'Essential'
        })
      } else {
        const payload = buildCreateChangefeedPayload(data)
        await createChangefeed.mutateAsync({
          orgId: tenantData.id,
          projectId: activeProjectId,
          clusterId: cluster.id!,
          data: payload
        })
        eventTracking(`Create changefeed successful`, {
          'Cluster Type': 'Essential',
          'Changefeed Type': data.changefeed_type
        })
      }

      history.push(`/clusters/${cluster.id}/changefeed`)
    } catch (_e) {
      eventTracking('Failed when create changefeed', {
        'Cluster Type': 'Essential'
      })
    }
  }

  const renderKafkaReview = () => {
    const kafka = formData.kafka
    const dataFormat = kafka?.data_format
    const avroConfig = dataFormat?.avro_config
    const confluentSchemaRegistry = avroConfig?.confluent_schema_registry
    const glueSchemaRegistry = avroConfig?.aws_glue_schema_registry
    const schemaRegistryType = confluentSchemaRegistry ? 'CONFLUENT' : glueSchemaRegistry ? 'AWS_GLUE' : undefined
    return (
      <>
        <ReviewProperty
          label="Network Type"
          value={kafka?.network?.network_type ? KafkaNetworkTypeLabels[kafka.network.network_type] : '-'}
        />
        {kafka?.network?.network_type === KafkaNetworkTypeEnum.PUBLIC && (
          <ReviewProperty label="Broker Endpoints" value={kafka?.network?.public_endpoints || '-'} />
        )}
        {kafka?.network?.network_type === KafkaNetworkTypeEnum.PRIVATE_LINK && (
          <>
            <ReviewProperty
              label="Private Link Connection ID"
              value={kafka?.network?.private_link?.private_link_connection_id || '-'}
            />
            {kafka?.network?.private_link?.port != null && (
              <ReviewProperty label="Port" value={kafka.network.private_link.port} />
            )}
          </>
        )}
        <ReviewProperty label="TLS Encryption" value={kafka?.authentication?.enable_tls ? 'Enabled' : 'Disabled'} />
        <ReviewProperty
          label="Authentication"
          value={kafka?.authentication?.auth_type ? AuthTypeLabels[kafka.authentication.auth_type] : '-'}
        />
        <ReviewProperty label="Protocol" value={getProtocolDisplayName(dataFormat?.protocol) || '-'} />
        <ReviewProperty label="TiDB Extension" value={dataFormat?.enable_tidb_extension ? 'Enable' : 'Disable'} />

        {dataFormat?.protocol === 'DEBEZIUM' && (
          <>
            <ReviewProperty
              label="Include Old Values in Output"
              value={dataFormat?.debezium_config?.disable_output_old_value ? 'No' : 'Yes'}
            />
            <ReviewProperty
              label="Include Schema Description in Output"
              value={dataFormat?.debezium_config?.disable_schema ? 'No' : 'Yes'}
            />
          </>
        )}

        {dataFormat?.protocol === 'AVRO' && (
          <>
            <ReviewProperty
              label="Decimal Format"
              value={avroConfig?.decimal_handling_mode ? DecimalHandlingLabels[avroConfig.decimal_handling_mode] : '-'}
            />
            <ReviewProperty
              label="Unsigned BigInt Format"
              value={
                avroConfig?.bigint_unsigned_handling_mode
                  ? UnsignedBigIntLabels[avroConfig.bigint_unsigned_handling_mode]
                  : '-'
              }
            />
            <ReviewProperty
              label="Schema Registry Type"
              value={schemaRegistryType ? SchemaRegistryTypeLabels[schemaRegistryType] : '-'}
            />

            {schemaRegistryType === 'CONFLUENT' && (
              <>
                <ReviewProperty label="Schema Registry Endpoint" value={confluentSchemaRegistry?.endpoint || '-'} />
                <ReviewProperty
                  label="HTTP Authentication"
                  value={confluentSchemaRegistry?.enable_http_auth ? 'Enable' : 'Disable'}
                />
                {confluentSchemaRegistry?.enable_http_auth && (
                  <ReviewProperty label="Schema Registry Username" value={confluentSchemaRegistry?.username || '-'} />
                )}
              </>
            )}

            {schemaRegistryType === 'AWS_GLUE' && (
              <>
                <ReviewProperty label="AWS Region" value={glueSchemaRegistry?.region || '-'} />
                <ReviewProperty label="Registry Name" value={glueSchemaRegistry?.name || '-'} />
                <ReviewProperty label="Access Key ID" value={glueSchemaRegistry?.access_key_id || '-'} />
              </>
            )}
          </>
        )}

        <ReviewProperty
          label="Topic Distribution"
          value={
            kafka?.topic_partition_config?.dispatch_type
              ? TopicDispatchLabels[kafka.topic_partition_config.dispatch_type]
              : '-'
          }
        />
        {!!kafka?.column_selectors?.length && (
          <ReviewProperty
            label="Column Selector"
            value={
              <Stack gap={4}>
                {kafka.column_selectors.map((selector, index) => (
                  <span key={index}>
                    {selector.matcher?.join(', ')}: {selector.columns?.join(', ')}
                  </span>
                ))}
              </Stack>
            }
          />
        )}
        <ReviewProperty label="Topic Name" value={kafka?.topic_partition_config?.default_topic || '-'} />
        {kafka?.topic_partition_config?.dispatch_type !== TopicDispatchTypeEnum.ONE_TOPIC && (
          <>
            {kafka?.topic_partition_config?.topic_prefix && (
              <ReviewProperty label="Topic Prefix" value={kafka.topic_partition_config.topic_prefix} />
            )}
            {kafka?.topic_partition_config?.separator &&
              kafka?.topic_partition_config?.dispatch_type === TopicDispatchTypeEnum.BY_TABLE && (
                <ReviewProperty label="Topic Separator" value={kafka.topic_partition_config.separator} />
              )}
            {kafka?.topic_partition_config?.topic_suffix && (
              <ReviewProperty label="Topic Suffix" value={kafka.topic_partition_config.topic_suffix} />
            )}
          </>
        )}
        {kafka?.topic_partition_config?.partition_dispatchers?.length && (
          <ReviewProperty
            label="Partition Dispatcher"
            value={
              <Box sx={{ display: 'grid', gridTemplateColumns: 'fit-content(50%) 1fr', gap: 4 }}>
                {kafka.topic_partition_config.partition_dispatchers?.map((item, index) => (
                  <Fragment key={index}>
                    <Typography>
                      {item.matcher?.join(',')}
                      {' :'}
                    </Typography>
                    <Typography>
                      {[
                        partitionDistributionOptions.find((option) => option.value === item.partition_type)?.text,
                        item.index_name,
                        item.columns?.join(',')
                      ]
                        .filter(Boolean)
                        .join(', ')}
                    </Typography>
                  </Fragment>
                ))}
              </Box>
            }
          />
        )}
        <ReviewProperty label="Replication Factor" value={kafka?.topic_partition_config?.replication_factor || '-'} />
        <ReviewProperty label="Partition Number" value={kafka?.topic_partition_config?.partition_num || '-'} />
        {cluster.version && canConfigureEventSplitting(cluster.version) && (
          <ReviewProperty label="Output Raw Events" value={kafka?.output_raw_change_event ? 'Yes' : 'No'} />
        )}
      </>
    )
  }

  const renderMySQLReview = () => {
    const mysql = formData.mysql
    return (
      <>
        <ReviewProperty
          label="Network Type"
          value={mysql?.network?.network_type ? MySQLNetworkTypeLabels[mysql.network.network_type] : '-'}
        />
        {mysql?.network?.network_type === MySQLNetworkTypeEnum.PUBLIC && (
          <ReviewProperty label="MySQL Endpoint" value={mysql?.network?.public_endpoint || '-'} />
        )}
        {mysql?.network?.network_type === MySQLNetworkTypeEnum.PRIVATE_LINK && (
          <>
            <ReviewProperty
              label="Private Link Connection ID"
              value={mysql?.network?.private_link?.private_link_connection_id || '-'}
            />
            <ReviewProperty label="Port" value={mysql?.network?.private_link?.port || '-'} />
          </>
        )}
        <ReviewProperty label="Username" value={mysql?.authentication?.user_name || '-'} />
        <ReviewProperty label="TLS Encryption" value={mysql?.authentication?.enable_tls ? 'Enabled' : 'Disabled'} />
      </>
    )
  }

  return (
    <FormProvider {...methods}>
      <Card p={0}>
        <form onSubmit={handleSubmit(onCreate)}>
          <Stack m={24}>
            <HeadLabel label="Specification and Name" />
            <FormSelect
              rules={{ required: 'Required' }}
              name="rcu"
              data={specOptions}
              placeholder="Select the number of changefeed capacity unit"
              renderOption={renderDescriptiveSelectOption}
              disabled={isEditing || isSpecLoading}
              label="Specify the number of changefeed capacity unit used by this changefeed"
              data-testid="essentialv1-changefeed-rcu-select"
              mb={16}
            />
            <FormTextInput
              name="display_name"
              label="Changefeed Name"
              placeholder="Enter the changefeed name"
              rules={{ required: 'Required' }}
              disabled={isEditing}
            />
          </Stack>

          <Divider />

          <Stack m={24} gap={16}>
            <HeadLabel label="Configuration Summary" />

            <ReviewProperty label="Destination" value={ChangefeedNames[formData.changefeed_type]} />

            {formData.changefeed_type === ChangefeedTypeEnum.KAFKA && renderKafkaReview()}
            {formData.changefeed_type === ChangefeedTypeEnum.MYSQL && renderMySQLReview()}

            <Divider />

            <ReviewProperty label="Table Filter" value={formData.filter?.filter_rule?.join(', ') || '*.*'} />
            <ReviewProperty label="Case Sensitive" value={formData.filter?.case_sensitive ? 'Yes' : 'No'} />
            {!!formData.filter?.event_filter_rule?.length && (
              <ReviewProperty
                label="Event Filter"
                value={<EventFilterSummary rules={formData.filter.event_filter_rule} />}
              />
            )}
            <ReviewProperty label="Start Position" value={renderStartPosition(formData.start_position)} />
          </Stack>

          {error && <FormErrorMessage message={getErrorMessage(error, true)} />}

          <Divider />

          <Group m={24} justify="flex-end">
            <Button
              variant="default"
              type="button"
              onClick={onCancel}
              data-mp-event="Changefeed Review Cancel Button Clicked"
            >
              Cancel
            </Button>
            <Button
              variant="default"
              type="button"
              onClick={onGoPrev}
              data-mp-event="Changefeed Review Previous Button Clicked"
            >
              Previous
            </Button>
            <Button
              type="submit"
              loading={createChangefeed.isLoading || editChangefeed.isLoading}
              data-mp-event="Changefeed Review Submit Button Clicked"
            >
              Submit
            </Button>
          </Group>
        </form>
      </Card>
    </FormProvider>
  )
}