import { Loader } from '@tidbcloud/uikit'
import { dayjs } from '@tidbcloud/uikit/utils'
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 { HeroComponent, isHeroQueryReady } from 'dbaas/components/Measurement'
import StepsWizard from 'dbaas/components/StepsWizard'
import { PageAccessGuard } from 'dbaas/contexts/access-control'
import { useIsEditingChangefeed } from 'dbaas/features/Changefeed'
import { CloudProviderConst } from 'dbaas/features/Project'
import { ConfigureTarget } from 'dbaas/screens/Changefeed/CreateOrEditChangefeedPremium/step1'
import { rewriteS3UriToOssUri } from 'dbaas/screens/Changefeed/CreateOrEditChangefeedPremium/step1/payloadBuilders'
import { ConfigureReplica } from 'dbaas/screens/Changefeed/CreateOrEditChangefeedPremium/step2'
import { Review } from 'dbaas/screens/Changefeed/CreateOrEditChangefeedPremium/step3'
import type { CreateOrEditChangefeedFormData } from 'dbaas/screens/Changefeed/CreateOrEditChangefeedPremium/types'
import { useSharedTiDBContext } from 'dbaas/screens/NextGen/tidb-mgmt-container/shared/tidb-context'
import { TiDBPageShell } from 'dbaas/screens/NextGen/tidb-mgmt-container/shared/TiDBPageShell'
import { useNgGetChangefeed } from 'dbaas/services/dataflow'
import {
  ConsoleapiDataFormat,
  ConsoleapiNetworkType,
  ConsoleapiS3AuthType,
  ConsoleapiS3Dataformat,
  ConsoleapiStartMode,
  ConsoleapiStartPosition,
  ConsoleapiStorageFlushConfig,
  ConsoleapiTableConfig,
  ConsoleapiTopicPartitionConfig
} from 'dbaas/services/dataflow/interface'
import { eventTracking } from 'dbaas/utils/tracking'

const defaultKafkaDataformat: ConsoleapiDataFormat = {
  protocol: 'CANAL_JSON',
  enable_tidb_extension: false,
  decimal_handling_mode: 'DECIMAL_HANDLING_MODE_PRECISE',
  bigint_unsigned_handling_mode: 'BIGINT_UNSIGNED_HANDLING_MODE_LONG',
  debezium_config: {
    output_old_value: true,
    disable_schema: false
  }
}

const defaultKafkaTopicPartitionConfig: ConsoleapiTopicPartitionConfig = {
  dispatch_type: 'DISPATCH_TYPE_ONE_TOPIC',
  replication_factor: 2,
  partition_num: 5,
  topic_suffix: '_suffix',
  separator: '_',
  topic_prefix: 'prefix_',
  partition_dispatchers: [
    {
      partition_type: 'TABLE',
      matcher: ['*.*'],
      index_name: '',
      columns: []
    }
  ]
}

const defaultS3DataFormat: ConsoleapiS3Dataformat = {
  protocol: 'S3_PROTOCOL_CSV',
  enable_tidb_extension: false,
  date_separator: 'none',
  delimiter: ',',
  quote: '"',
  null: '\\N',
  include_commit_ts: false,
  binary_encoding_method: 'BASE64'
}

const defaultFlushConfig: ConsoleapiStorageFlushConfig = {
  flush_interval: '60s',
  file_size: String(64 * 1024 * 1024) // 64mb
}

const defaultStartPosition: ConsoleapiStartPosition = {
  mode: 'FROM_NOW',
  utc: new Date().toISOString(),
  tso: ''
}

const defaultTableConfig: ConsoleapiTableConfig = {
  filter_rule: ['*.*'],
  mode: 'FORCE_SYNC',
  event_filter_rule: [],
  case_sensitive: false
}

const normalizeDuplicateStartPosition = (
  startPosition?: ConsoleapiStartPosition
): ConsoleapiStartPosition | undefined => {
  if (startPosition?.mode === ConsoleapiStartMode.FROM_TSO && startPosition.tso === '0') {
    return {
      ...startPosition,
      tso: ''
    }
  }

  return startPosition
}

const getDefaultNetworkType = (cloudProvider: string | undefined, isEssentialV2?: boolean) => {
  if (isEssentialV2) {
    return ConsoleapiNetworkType.NetworkType_Private_Link
  }

  if (cloudProvider === CloudProviderConst.AWS || cloudProvider === CloudProviderConst.GCP) {
    return ConsoleapiNetworkType.NetworkType_VPC_Peering
  } else {
    return ConsoleapiNetworkType.NetworkType_Public
  }
}

/**
 * This component is used to create or edit a changefeed.
 *
 * For editing, there are 3 parts that are not allowed to be updated:
 * 1. network info
 * 2. replication->start replication position
 * 3. specification->rcu
 *
 * For more details, please refer to: https://pingcap.feishu.cn/wiki/Swz9wtI3ii1COek2AjzcdpN0n1g
 */
const CreateChangefeed = () => {
  const [step, setStep] = useState(0)
  const history = useHistory()
  const cluster = useSharedTiDBContext()
  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<CreateOrEditChangefeedFormData>({
    defaultValues: {
      changefeed_type: 'KAFKA',
      ports_str: '',
      network_info: {
        network_type: getDefaultNetworkType(cluster.cloudProvider, cluster.computed.isEssentialV2)
      },
      kafka_info: {
        broker: {
          kafka_version: 'KAFKA_VERSION_3XX',
          compression: 'NONE',
          tls_enable: false
        },
        authentication: {
          auth_type: 'DISABLE'
        },
        data_format: defaultKafkaDataformat,
        topic_partition_config: defaultKafkaTopicPartitionConfig,
        start_position: defaultStartPosition,
        column_selectors: []
      },
      mysql_info: {
        start_position: defaultStartPosition
      },
      s3_info: {
        endpoint: {
          auth_type: ConsoleapiS3AuthType.S3_AUTH_TYPE_ASSUME_ROLE
        },
        data_format: defaultS3DataFormat,
        start_position: defaultStartPosition,
        storage_flush_config: defaultFlushConfig
      },
      gcs_info: {
        data_format: defaultS3DataFormat,
        start_position: defaultStartPosition
      },
      pulsar_info: {
        destination_protocol: 'PULSAR_PROTOCOL_PULSAR',
        compression_type: 'PULSAR_COMPRESSION_NONE',
        authentication: {
          auth_type: 'PULSAR_AUTH_TYPE_BASIC'
        },
        advanced_setting: {
          batching_max_messages: 1000,
          batching_max_publish_delay: 10,
          connection_timeout: 5,
          operation_timeout: 30,
          send_timeout: 30
        },
        start_position: defaultStartPosition,
        data_format: {
          protocol: 'CANAL_JSON',
          enable_tidb_extension: false
        },
        topic_distribution: {
          dispatch_type: 'DISPATCH_TYPE_ONE_TOPIC',
          pulsar_tenant: 'public',
          pulsar_namespace: 'default',
          topic_suffix: '_suffix',
          separator: '_',
          topic_prefix: 'prefix_'
        },
        partition_dispatchers: [
          {
            partition_type: 'TABLE',
            matcher: ['*.*']
          }
        ]
      },
      name: `changefeed-${dayjs().format('YYYYMMDDHHmm')}`,

      table_config: defaultTableConfig,

      // for verify filter in step 2
      db_username: '',
      db_password: ''
    }
  })

  const changefeedQuery = useNgGetChangefeed(cluster?.tidbId!, changefeedIdFromUrl || duplicateFromId, {
    query: {
      enabled: !!cluster?.tidbId && (!!changefeedIdFromUrl || !!duplicateFromId),
      refetchOnWindowFocus: false,
      refetchOnReconnect: false,
      onSuccess(data) {
        if (isEditing) {
          methods.setValue('changefeed_id', data.changefeed_id)
        }
        methods.setValue('changefeed_type', data.changefeed_type)
        methods.setValue('name', !!duplicateFromId ? `${data.name}_copy` : data.name)
        methods.setValue('changefeed_spec', data.changefeed_spec)
        if (data.request?.network_info) {
          methods.setValue('network_info', data.request.network_info)
          methods.setValue('ports_str', data.request.network_info?.ports?.join(',') || '')
        }

        let tableConfig: ConsoleapiTableConfig | undefined = undefined
        if (data.request?.serverless_tidb_info) {
          methods.setValue('serverless_tidb_info', data.request.serverless_tidb_info)
          tableConfig = data.request.serverless_tidb_info.mysql_info?.table_config
        }
        if (data.request?.kafka_info) {
          const startPosition = duplicateFromId
            ? normalizeDuplicateStartPosition(data.request.kafka_info.start_position)
            : data.request.kafka_info.start_position
          methods.setValue('kafka_info', {
            ...data.request.kafka_info,
            start_position: startPosition
          })
          tableConfig = data.request.kafka_info?.table_config
        }
        if (data.request?.mysql_info) {
          const startPosition = duplicateFromId
            ? normalizeDuplicateStartPosition(data.request.mysql_info.start_position)
            : data.request.mysql_info.start_position
          methods.setValue('mysql_info', {
            ...data.request.mysql_info,
            start_position: startPosition
          })
          tableConfig = data.request.mysql_info?.table_config
        }
        if (data.request?.s3_info) {
          const startPosition = duplicateFromId
            ? normalizeDuplicateStartPosition(data.request.s3_info.start_position)
            : data.request.s3_info.start_position
          const isAliCloudOSS =
            cluster.cloudProvider === CloudProviderConst.ALICLOUD ||
            data.request.s3_info.endpoint?.provider === 'S3_PROVIDER_ALIBABA_OSS'
          methods.setValue('s3_info', {
            ...data.request.s3_info,
            endpoint: {
              ...data.request.s3_info.endpoint,
              s3Uri: isAliCloudOSS
                ? rewriteS3UriToOssUri(data.request.s3_info.endpoint?.s3Uri)
                : data.request.s3_info.endpoint?.s3Uri
            },
            start_position: startPosition
          })
          tableConfig = data.request.s3_info?.table_config
        }
        if (data.request?.gcs_info) {
          const startPosition = duplicateFromId
            ? normalizeDuplicateStartPosition(data.request.gcs_info.start_position)
            : data.request.gcs_info.start_position
          methods.setValue('gcs_info', {
            ...data.request.gcs_info,
            start_position: startPosition
          })
          tableConfig = data.request.gcs_info?.table_config
        }
        if (data.request?.pulsar_info) {
          const startPosition = duplicateFromId
            ? normalizeDuplicateStartPosition(data.request.pulsar_info.start_position)
            : data.request.pulsar_info.start_position
          methods.setValue('pulsar_info', {
            ...data.request.pulsar_info,
            start_position: startPosition
          })
          tableConfig = data.request.pulsar_info?.table_config
        }

        if (tableConfig) {
          methods.setValue('table_config', tableConfig)
        }
      }
    }
  })
  const { isFetching, isFetched, isError } = changefeedQuery

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

  const stepItems = [
    {
      name: 'Destination',
      content: <ConfigureTarget cluster={cluster} onFinish={onFinish} onCancel={onCancel} />
    },
    {
      name: 'Replication',
      content: <ConfigureReplica cluster={cluster} onFinish={onFinish} onGoPrev={onGoPrev} onCancel={onCancel} />
    },
    {
      name: 'Specification and Review',
      content: <Review cluster={cluster} onGoPrev={onGoPrev} onCancel={onCancel} />
    }
  ]

  return (
    <TiDBPageShell title={isEditing ? 'Edit Changefeed' : 'Create Changefeed'} headerProps={{ withBack: true }}>
      <HeroComponent ready={(!changefeedIdFromUrl && !duplicateFromId) || isHeroQueryReady({ isFetched, isError })} />
      <PageAccessGuard id="TidbxCDCAction">
        {isFetching ? (
          <Loader size="xs" />
        ) : (
          <FormProvider {...methods}>
            <StepsWizard steps={stepItems} current={step} />
          </FormProvider>
        )}
      </PageAccessGuard>
    </TiDBPageShell>
  )
}

export default observer(CreateChangefeed)