import { Loader } from '@tidbcloud/uikit'
import { useMemoizedFn } from 'ahooks'
import { observer } from 'mobx-react'
import { useMemo, useState } from 'react'
import { FormProvider, useForm } from 'react-hook-form'
import { useHistory } from 'react-router-dom'

import StepsWizard from 'dbaas/components/StepsWizard'
import { PageAccessGuard } from 'dbaas/contexts/access-control'
import { useIsEditingChangefeed } from 'dbaas/features/Changefeed'
import { SharedCluster } from 'dbaas/features/ClusterManagement/cluster'
import { ClusterPageShell } from 'dbaas/screens/ClusterV2/components'
import { useChangefeedServiceGetChangefeed } from 'dbaas/services/api-clients/legacy/serverless'
import {
  BigintUnsignedHandlingModeEnum,
  ChangefeedFilter,
  ChangefeedTypeEnum,
  DecimalHandlingModeEnum,
  KafkaAuthTypeEnum,
  KafkaCompressionTypeEnum,
  KafkaDataFormat,
  KafkaNetworkTypeEnum,
  KafkaProtocolEnum,
  KafkaTopicPartitionConfig,
  KafkaVersionEnum,
  MySQLNetworkTypeEnum,
  PartitionDispatcherEnum,
  StartPosition
} from 'dbaas/services/api-clients/legacy/serverless/interface'
import useStores from 'dbaas/stores/useStores'
import { eventTracking } from 'dbaas/utils/tracking'

import { ConfigureTargetEssential } from './step1'
import { ConfigureReplicaEssential } from './step2'
import { ReviewEssential } from './step3'
import type { CreateOrEditChangefeedEssentialFormData } from './types'

const defaultKafkaDataFormat: KafkaDataFormat = {
  protocol: KafkaProtocolEnum.CANAL_JSON,
  enable_tidb_extension: false,
  avro_config: {
    decimal_handling_mode: DecimalHandlingModeEnum.PRECISE,
    bigint_unsigned_handling_mode: BigintUnsignedHandlingModeEnum.LONG
  },
  debezium_config: {
    disable_output_old_value: false,
    disable_schema: false
  }
}

const defaultKafkaTopicPartitionConfig: KafkaTopicPartitionConfig = {
  dispatch_type: 'ONE_TOPIC',
  replication_factor: 2,
  partition_num: 5,
  topic_suffix: '_suffix',
  separator: '_',
  topic_prefix: 'prefix_',
  partition_dispatchers: [
    {
      partition_type: PartitionDispatcherEnum.TABLE,
      matcher: ['*.*']
    }
  ]
}

const defaultStartPosition: StartPosition = {
  mode: 'FROM_NOW'
}

const defaultFilter: ChangefeedFilter = {
  filter_rule: ['*.*'],
  mode: 'IGNORE_NOT_SUPPORT_TABLE',
  event_filter_rule: [],
  case_sensitive: false
}

/**
 * This component is used to create or edit a changefeed for Essential clusters.
 *
 * For editing, there are parts that are not allowed to be updated:
 * 1. network info
 * 2. replication->start replication position
 */
const CreateChangefeedEssential = ({ cluster }: { cluster: SharedCluster }) => {
  const [step, setStep] = useState(0)
  const history = useHistory()
  const {
    store: { tenantData, activeProjectId }
  } = useStores()

  const urlSearchParams = useMemo(() => new URLSearchParams(window.location.search), [])
  const changefeedIdFromUrl = urlSearchParams.get('id') ?? ''
  const duplicateFromId = urlSearchParams.get('from_id') ?? ''

  const isEditing = useIsEditingChangefeed()

  const methods = useForm<CreateOrEditChangefeedEssentialFormData>({
    defaultValues: {
      changefeed_type: ChangefeedTypeEnum.KAFKA,
      display_name: `changefeed-${new Date().toISOString().slice(0, 16).replace(/[-:T]/g, '')}`,
      filter: defaultFilter,
      start_position: defaultStartPosition,
      kafka: {
        network: {
          network_type: KafkaNetworkTypeEnum.PUBLIC
        },
        broker: {
          kafka_version: KafkaVersionEnum.VERSION_3XX,
          compression: KafkaCompressionTypeEnum.NONE
        },
        authentication: {
          auth_type: KafkaAuthTypeEnum.DISABLE,
          enable_tls: false
        },
        data_format: defaultKafkaDataFormat,
        topic_partition_config: defaultKafkaTopicPartitionConfig,
        output_raw_change_event: false
      },
      mysql: {
        network: {
          network_type: MySQLNetworkTypeEnum.PUBLIC
        },
        authentication: {
          enable_tls: false
        }
      }
    }
  })

  const { isFetching } = useChangefeedServiceGetChangefeed(
    tenantData.id,
    activeProjectId,
    cluster.id!,
    changefeedIdFromUrl || duplicateFromId,
    {
      query: {
        enabled: !!cluster.id && (!!changefeedIdFromUrl || !!duplicateFromId),
        refetchOnWindowFocus: false,
        refetchOnReconnect: false,
        onSuccess(data) {
          if (isEditing) {
            methods.setValue('changefeed_id', data.changefeed_id)
          }
          methods.setValue('changefeed_type', data.sink?.type || ChangefeedTypeEnum.KAFKA)
          methods.setValue('display_name', !!duplicateFromId ? `${data.display_name}_copy` : data.display_name || '')
          if (data.rcu != null) {
            methods.setValue('rcu', data.rcu.toString())
          }

          if (data.filter) {
            methods.setValue('filter', data.filter)
          }
          if (data.start_position) {
            methods.setValue('start_position', data.start_position)
          }

          if (data.sink?.kafka) {
            methods.setValue('kafka', data.sink.kafka)
          }
          if (data.sink?.mysql) {
            methods.setValue('mysql', data.sink.mysql)
          }
        }
      }
    }
  )

  const onFinish = useMemoizedFn(() => {
    setStep((s) => s + 1)
  })
  const onGoPrev = useMemoizedFn(() => {
    setStep((s) => s - 1)
  })
  const onCancel = useMemoizedFn(() => {
    eventTracking('Cancel changefeed creating')
    history.push(`/clusters/${cluster.id}/changefeed`)
  })

  const stepItems = [
    {
      name: 'Destination',
      content: <ConfigureTargetEssential cluster={cluster} onFinish={onFinish} onCancel={onCancel} />
    },
    {
      name: 'Replication',
      content: (
        <ConfigureReplicaEssential cluster={cluster} onFinish={onFinish} onGoPrev={onGoPrev} onCancel={onCancel} />
      )
    },
    {
      name: 'Review',
      content: <ReviewEssential cluster={cluster} onGoPrev={onGoPrev} onCancel={onCancel} />
    }
  ]

  return (
    <ClusterPageShell
      title={isEditing ? 'Edit Changefeed' : 'Create Changefeed'}
      bodyProps={{ mb: 80 }}
      headerProps={{ withBack: true }}
    >
      <PageAccessGuard id="TidbxCDCAction" scopeProps={{ tidbId: cluster.id }}>
        {isFetching ? (
          <Loader size="xs" />
        ) : (
          <FormProvider {...methods}>
            <StepsWizard steps={stepItems} current={step} />
          </FormProvider>
        )}
      </PageAccessGuard>
    </ClusterPageShell>
  )
}

export default observer(CreateChangefeedEssential)