import { Alert, Box, Button, Card, Checkbox, Divider, Group, Loader, Stack, Tabs, Typography } from '@tidbcloud/uikit'
import { FormCheckbox } from '@tidbcloud/uikit/biz'
import { IconAlertCircle } from '@tidbcloud/uikit/icons'
import { useMemoizedFn } from 'ahooks'
import { omit } from 'lodash-es'
import { useRef, useState } from 'react'
import { Controller, FormProvider, useForm } from 'react-hook-form'
import { useQuery } from 'react-query'

import TableFilter from 'dbaas/components/TableFilter'
import { areValidEventFilters, useIsEditingChangefeed } from 'dbaas/features/Changefeed'
import { SharedCluster } from 'dbaas/features/ClusterManagement/cluster'
import { EventFilter } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/step2/EventFilter'
import type { FormData } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/types'
import { describeSchemaTable, useTestConnector } from 'dbaas/services/dataflow'
import {
  DataflowBigintUnsignedHandlingMode,
  DataflowBinaryEncodingMethod,
  DataflowColumnSelector,
  DataflowConnectorType,
  DataflowDecimalHandlingMode,
  DataflowDispatchType,
  DataflowEventFilterRule,
  DataflowPartitionDispatcher,
  DataflowProtocol,
  DataflowS3Dataformat,
  DataflowS3Protocol,
  DataflowStartMode,
  DataflowStorageFlushConfig,
  DataflowTableConfig,
  DataflowTableMode
} from 'dbaas/services/dataflow/interface'
import { getErrorMessage } from 'dbaas/services/errorCodes'
import useStores from 'dbaas/stores/useStores'
import { compareVersion } from 'dbaas/utils/chore'
import { eventTracking, MixpanelEvents } from 'dbaas/utils/tracking'

import { ColumnSelector } from './ColumnSelector'
import { FormKafka } from './FormKafka'
import { FormMySQL } from './FormMySQL'
import { FormPulsar } from './FormPulsar'
import { FormS3OrGCS } from './FormS3OrGCS'
import { FormTiDBServerless } from './FormTiDBServerless'
import { buildConfigureReplicaPayload } from './payloadBuilders'
import type { ConfigureReplicaForm } from './types'

const TableModeOptions = [
  {
    label:
      'I understand the data inconsistency risk and want to replicate all tables without unique keys, including new tables created in the future.',
    value: DataflowTableMode.FORCE_SYNC
  }
  // {
  //   label: 'Ignore ineligible tables, there is no need to replica.',
  //   value: DataflowTableMode.IGNORE_NOT_SUPPORT_TABLE
  // },
]

const DefaultDataFormat: DataflowS3Dataformat = {
  protocol: DataflowS3Protocol.S3_PROTOCOL_CSV,
  enable_tidb_extension: false,
  date_separator: 'none',
  delimiter: ',',
  quote: '"',
  null: '\\N',
  include_commit_ts: false,
  binary_encoding_method: DataflowBinaryEncodingMethod.BASE64
}

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

export const ConfigureReplica = ({
  onFinish,
  cluster,
  formData,
  onCancel,
  onGoPrev
}: {
  cluster: SharedCluster
  onFinish: (data: FormData, toStep?: number) => void
  formData: FormData
  onGoPrev: () => void
  onCancel: () => void
}) => {
  const {
    store: { tenantData, activeProjectId }
  } = useStores()
  const { connector_type } = formData
  const defaultTableConfig: DataflowTableConfig = {
    filter_rule: ['*.*'],
    mode: DataflowTableMode.FORCE_SYNC,
    event_filter_rule: [],
    case_sensitive: false,
    ...(connector_type === DataflowConnectorType.KAFKA ? formData.kafka_info?.table_config : {}),
    ...(connector_type === DataflowConnectorType.MYSQL ? formData.mysql_info?.table_config : {}),
    //prettier-ignore
    ...(connector_type === DataflowConnectorType.SERVERLESS_TIDB ? formData.serverless_tidb_info?.mysql_info?.table_config : {}),
    ...(connector_type === DataflowConnectorType.S3 ? formData.s3_info?.table_config : {}),
    ...(connector_type === DataflowConnectorType.GCS ? formData.gcs_info?.table_config : {}),
    ...(connector_type === DataflowConnectorType.AZURE_BLOB_STORAGE
      ? formData.azure_blob_storage_info?.table_config
      : {}),
    ...(connector_type === DataflowConnectorType.PULSAR ? formData.pulsar_info?.table_config : {})
  }
  const methods = useForm<ConfigureReplicaForm>({
    defaultValues: {
      kafka: {
        data_format: {
          protocol: DataflowProtocol.CANAL_JSON,
          enable_tidb_extension: false,
          decimal_handling_mode: DataflowDecimalHandlingMode.DECIMAL_HANDLING_MODE_PRECISE,
          bigint_unsigned_handling_mode: DataflowBigintUnsignedHandlingMode.BIGINT_UNSIGNED_HANDLING_MODE_LONG,
          debezium_config: {
            output_old_value: true,
            disable_schema: false,
            ...formData.kafka_info?.data_format?.debezium_config
          },
          ...formData.kafka_info?.data_format
        },
        topic_partition_config: {
          dispatch_type: DataflowDispatchType.DISPATCH_TYPE_ONE_TOPIC,
          replication_factor: 2,
          partition_num: 5,
          topic_suffix: '_suffix',
          separator: '_',
          topic_prefix: 'prefix_',
          partition_dispatchers: [
            {
              partition_type: DataflowPartitionDispatcher.TABLE,
              matcher: ['*.*'],
              index_name: '',
              columns: []
            }
          ],
          ...formData.kafka_info?.topic_partition_config
        },
        start_position: {
          mode: DataflowStartMode.FROM_NOW,
          utc: new Date().toISOString(),
          tso: '',
          ...formData.kafka_info?.start_position
        },
        column_selectors: [...(formData.kafka_info?.column_selectors ?? [])]
      },
      mysql: {
        start_position: {
          mode: DataflowStartMode.FROM_NOW,
          utc: new Date().toISOString(),
          tso: '',
          ...formData.mysql_info?.start_position
        }
      },
      tidb: {
        mysql_info: {
          start_position: {
            mode: DataflowStartMode.FROM_TSO,
            tso: '',
            ...formData.serverless_tidb_info?.mysql_info?.start_position
          }
        }
      },
      s3_info: {
        data_format: {
          ...DefaultDataFormat,
          ...formData.s3_info?.data_format,
          delimiter: formData.s3_info?.data_format?.delimiter || DefaultDataFormat.delimiter,
          quote: formData.s3_info?.data_format?.quote || DefaultDataFormat.quote,
          null: formData.s3_info?.data_format?.null || DefaultDataFormat.null
        },
        start_position: {
          mode: DataflowStartMode.FROM_NOW,
          utc: new Date().toISOString(),
          tso: '',
          ...formData.s3_info?.start_position
        },
        storage_flush_config: {
          ...DefaultFlushConfig,
          ...formData.s3_info?.storage_flush_config
        }
      },
      gcs_info: {
        data_format: {
          ...DefaultDataFormat,
          ...formData.gcs_info?.data_format,
          delimiter: formData.gcs_info?.data_format?.delimiter || DefaultDataFormat.delimiter,
          quote: formData.gcs_info?.data_format?.quote || DefaultDataFormat.quote,
          null: formData.gcs_info?.data_format?.null || DefaultDataFormat.null
        },
        start_position: {
          mode: DataflowStartMode.FROM_NOW,
          utc: new Date().toISOString(),
          tso: '',
          ...formData.gcs_info?.start_position
        },
        storage_flush_config: {
          ...DefaultFlushConfig,
          ...formData.gcs_info?.storage_flush_config
        }
      },
      azure_blob_storage_info: {
        data_format: {
          ...DefaultDataFormat,
          ...formData.azure_blob_storage_info?.data_format,
          delimiter: formData.azure_blob_storage_info?.data_format?.delimiter || DefaultDataFormat.delimiter,
          quote: formData.azure_blob_storage_info?.data_format?.quote || DefaultDataFormat.quote,
          null: formData.azure_blob_storage_info?.data_format?.null || DefaultDataFormat.null
        },
        start_position: {
          mode: DataflowStartMode.FROM_NOW,
          utc: new Date().toISOString(),
          tso: '',
          ...formData.azure_blob_storage_info?.start_position
        },
        storage_flush_config: {
          ...DefaultFlushConfig,
          ...formData.azure_blob_storage_info?.storage_flush_config
        }
      },
      pulsar_info: {
        start_position: {
          mode: DataflowStartMode.FROM_NOW,
          utc: new Date().toISOString(),
          tso: '',
          ...formData.pulsar_info?.start_position
        },
        data_format: {
          protocol: DataflowProtocol.CANAL_JSON,
          enable_tidb_extension: false,
          ...formData.pulsar_info?.data_format
        },
        topic_distribution: {
          dispatch_type: DataflowDispatchType.DISPATCH_TYPE_ONE_TOPIC,
          pulsar_tenant: 'public',
          pulsar_namespace: 'default',
          topic_suffix: '_suffix',
          separator: '_',
          topic_prefix: 'prefix_',
          ...formData.pulsar_info?.topic_distribution
        },
        partition_dispatchers: [
          ...(formData.pulsar_info?.partition_dispatchers ?? [
            {
              partition_type: DataflowPartitionDispatcher.TABLE,
              matcher: ['*.*']
            }
          ])
        ]
      },

      table_config: defaultTableConfig,
      filter_rule_value: JSON.stringify(defaultTableConfig?.filter_rule),
      event_filter_rule_value: JSON.stringify(defaultTableConfig.event_filter_rule)
    }
  })

  const containerRef = useRef<HTMLDivElement>(null)
  const [eventFilterValidateError, setEventFilterValidateError] = useState(false)
  const [columnSelectorsValidateError, setColumnSelectorsValidateError] = useState(false)
  const [filterTab, setFilterTab] = useState<'table-filter' | 'event-filter' | 'column-selectors'>('table-filter')
  const { control, handleSubmit, setValue, watch } = methods
  const isEditing = useIsEditingChangefeed()
  const rules_value = watch('filter_rule_value') ?? ''
  const columnSelectors = watch('kafka.column_selectors')
  const caseSensitive = watch('table_config.case_sensitive') ?? false

  const {
    data: schemaTableData,
    error: getTableError,
    isLoading
  } = useQuery(
    ['/describeSchemaTable', rules_value, caseSensitive],
    () =>
      describeSchemaTable(tenantData.id, activeProjectId, cluster.id, {
        filter_rule: JSON.parse(rules_value),
        case_sensitive: caseSensitive
      }),
    {
      refetchOnWindowFocus: false
    }
  )
  const testConnectorMutation = useTestConnector()
  const error = getTableError || testConnectorMutation.error

  const onEventFilterChange = useMemoizedFn((rules: DataflowEventFilterRule[]) => {
    setValue('event_filter_rule_value', JSON.stringify(rules))
  })

  const onColumnSelectorsChange = useMemoizedFn((selectors: DataflowColumnSelector[]) => {
    setValue('kafka.column_selectors', selectors)
  })

  const onSubmit = async (data: ConfigureReplicaForm) => {
    const payload = buildConfigureReplicaPayload(data, formData)
    const rules = JSON.parse(data.event_filter_rule_value) as DataflowEventFilterRule[]

    if (rules && rules.length > 0) {
      if (areValidEventFilters(rules)) {
        setEventFilterValidateError(false)
      } else {
        setFilterTab('event-filter')
        setEventFilterValidateError(true)
        containerRef.current?.scrollIntoView({ behavior: 'smooth' })
        return
      }
    }

    if (payload.connector_type === DataflowConnectorType.KAFKA) {
      const columnSelectors = payload.kafka_info?.column_selectors
      if (columnSelectors && columnSelectors?.length > 0) {
        if (
          columnSelectors.every(
            (i) => i.matcher && i.matcher.length > 0 && i.columns && i.columns.filter(Boolean).length > 0
          )
        ) {
          setColumnSelectorsValidateError(false)
        } else {
          setFilterTab('column-selectors')
          setColumnSelectorsValidateError(true)
          containerRef.current?.scrollIntoView({ behavior: 'smooth' })
          return
        }
      }
    }

    try {
      // Skip test connection in test environment
      const isTestEnv =
        process.env.NODE_ENV === 'test' || window.location.search.includes('e2e-test') || window.E2E_Test
      if (isTestEnv) {
        console.log(payload)
        console.log('[E2E Test]-[CreateChangefeed]-[Step2]-[Payload]-' + JSON.stringify(payload))
        onFinish(payload)
        return
      }

      if (!isEditing || payload.password) {
        await testConnectorMutation.mutateAsync({
          orgId: tenantData.id,
          projectId: activeProjectId,
          clusterId: cluster.id,
          data: {
            ...omit(payload, 'password'),
            connector_type: payload.connector_type!,
            network_info: formData.network_info
          }
        })
      }

      onFinish(payload)
    } catch (_e) {
      eventTracking('Failed when precheck in kafka changefeed', {
        'Cluster Type': cluster.computed.clusterType
      })
    }
  }

  return (
    <FormProvider {...methods}>
      <Card p={0}>
        <form onSubmit={handleSubmit(onSubmit)}>
          <Controller name="filter_rule_value" control={control} render={({ field }) => <input {...field} hidden />} />
          <Controller
            name="event_filter_rule_value"
            control={control}
            render={({ field }) => <input {...field} hidden />}
          />

          {/* scroll to here when filters are invalid */}
          <Box ref={containerRef}></Box>

          <Tabs p={24} value={filterTab} onChange={(value) => setFilterTab(value as any)}>
            <Tabs.List>
              <Tabs.Tab value="table-filter">Table Filter</Tabs.Tab>

              {compareVersion(cluster.version, '6.2.0', '>=') && <Tabs.Tab value="event-filter">Event Filter</Tabs.Tab>}

              {compareVersion(cluster.version, '7.5.0', '>=') &&
                formData.connector_type === DataflowConnectorType.KAFKA && (
                  <Tabs.Tab value="column-selectors">Column Selector</Tabs.Tab>
                )}
            </Tabs.List>

            <Tabs.Panel value="table-filter" pt="xs">
              <Stack>
                <Controller
                  name="table_config.mode"
                  control={control}
                  render={({ field }) => <input {...field} hidden />}
                />
                {/* for table_config.mode, this is display only  */}
                <Group wrap="nowrap" align="flex-start" gap={12}>
                  <Checkbox defaultChecked disabled mt={2} />
                  <Typography variant="body-lg" c="carbon.8">
                    {TableModeOptions[0].label}
                  </Typography>
                </Group>

                <FormCheckbox
                  name="table_config.case_sensitive"
                  label="Case-sensitive database and table names"
                  onChange={(e) => {
                    eventTracking(MixpanelEvents.DedicatedChangefeed.Enable.ButtonClicked, {
                      field: 'case_sensitive',
                      value: e.target.checked
                    })
                  }}
                />

                <TableFilter
                  importTables={schemaTableData?.replicate_allow_schema_table ?? []}
                  ignoredTables={schemaTableData?.replicate_deny_schema_table ?? []}
                  value={JSON.parse(rules_value)}
                  onChange={(rules) => {
                    setValue('filter_rule_value', JSON.stringify(rules))
                  }}
                  leftListTitle={
                    <Group gap={8}>
                      <span>Tables with valid keys</span>
                      {isLoading && <Loader size={14} />}
                    </Group>
                  }
                  rightListTitle="Tables without valid keys"
                />

                <Alert color="peacock" icon={<IconAlertCircle />}>
                  <Typography>
                    <b>Tables without valid keys</b> in the listed tables indicate that they <b>DON'T</b> have any
                    primary keys or unique indexes. Without a unique identifier for the downstream system to handle
                    duplicate events, data consistency may be compromised during replication.
                  </Typography>
                </Alert>
              </Stack>
            </Tabs.Panel>

            <Tabs.Panel value="event-filter" pt="xs">
              <EventFilter
                defaultValue={defaultTableConfig.event_filter_rule!}
                onChange={onEventFilterChange}
                validateError={eventFilterValidateError}
              />
            </Tabs.Panel>

            <Tabs.Panel value="column-selectors" pt="xs">
              <ColumnSelector
                defaultValue={columnSelectors ?? []}
                onChange={onColumnSelectorsChange}
                validateError={columnSelectorsValidateError}
              />
            </Tabs.Panel>
          </Tabs>

          <Divider />

          {formData.connector_type === DataflowConnectorType.KAFKA && (
            <FormKafka cluster={cluster} schemaTableData={schemaTableData} />
          )}
          {formData.connector_type === DataflowConnectorType.MYSQL && <FormMySQL schemaTableData={schemaTableData} />}
          {formData.connector_type === DataflowConnectorType.SERVERLESS_TIDB && (
            <FormTiDBServerless schemaTableData={schemaTableData} />
          )}
          {formData.connector_type === DataflowConnectorType.S3 && (
            <FormS3OrGCS cluster={cluster} schemaTableData={schemaTableData} path="s3_info" />
          )}
          {formData.connector_type === DataflowConnectorType.GCS && (
            <FormS3OrGCS cluster={cluster} schemaTableData={schemaTableData} path="gcs_info" />
          )}
          {formData.connector_type === DataflowConnectorType.AZURE_BLOB_STORAGE && (
            <FormS3OrGCS cluster={cluster} schemaTableData={schemaTableData} path="azure_blob_storage_info" />
          )}
          {formData.connector_type === DataflowConnectorType.PULSAR && <FormPulsar />}

          {error && (
            <Alert icon={<IconAlertCircle />} color="red" m={24}>
              <span dangerouslySetInnerHTML={{ __html: getErrorMessage(error) }} />
            </Alert>
          )}

          <Group m={24} justify="flex-end" wrap="nowrap">
            <Group>
              <Button
                variant="default"
                type="button"
                data-mp-event="Configure Replica Cancel Button Clicked"
                onClick={onCancel}
              >
                Cancel
              </Button>

              <Button
                variant="outline"
                type="button"
                data-mp-event="Configure Replica Previous Button Clicked"
                onClick={onGoPrev}
              >
                Previous
              </Button>

              <Button
                type="submit"
                loading={testConnectorMutation.isLoading}
                data-mp-event="Configure Replica Next Button Clicked"
              >
                Next
              </Button>
            </Group>
          </Group>
        </form>
      </Card>
    </FormProvider>
  )
}