import { Alert, Box, Button, Card, Divider, Group, Loader, Stack, Typography } from '@tidbcloud/uikit'
import { FormRadioGroup, FormRadioGroupProps } from '@tidbcloud/uikit/biz'
import { IconAlertCircle } from '@tidbcloud/uikit/icons'
import { omit } from 'lodash-es'
import { useMemo } from 'react'
import { FormProvider, useForm } from 'react-hook-form'

import {
  ChangefeedNames,
  useIsEditingChangefeed,
  useSupportChangefeedToServerlessTiDB,
  useSupportPulsarChangefeed
} from 'dbaas/features/Changefeed'
import { getChangefeedConnectionMethods } from 'dbaas/features/Changefeed/connection'
import { SharedCluster } from 'dbaas/features/ClusterManagement/cluster'
import { CloudProviderConst } from 'dbaas/features/Project'
import type { FormData } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/types'
import HeadLabel from 'dbaas/screens/ClustersV2/new/components/HeadLabel'
import { useTiDBServiceListTidbs } from 'dbaas/services/api-clients/v2023/cluster-management/v1beta2'
import { TidbServiceListTidbsServicePlan } from 'dbaas/services/api-clients/v2025/tidb-management/v1beta2/interface'
import { useConnectorServiceDescribeAdminProfile, useTestConnector } from 'dbaas/services/dataflow'
import {
  DataflowCompressionType,
  DataflowConnectorType,
  DataflowKafkaAuthType,
  DataflowKafkaType,
  DataflowKafkaVersion,
  DataflowPulsarAuthType,
  DataflowPulsarCompressionType,
  DataflowPulsarDestinationProtocol,
  DataflowS3AuthType
} from 'dbaas/services/dataflow/interface'
import { getErrorMessage } from 'dbaas/services/errorCodes'
import useStores from 'dbaas/stores/useStores'
import { sleep } from 'dbaas/utils/chore'
import { eventTracking } from 'dbaas/utils/tracking'

import { FormAzureBlob } from './FormAzureBlob'
import { FormGCS } from './FormGCS'
import { FormKafka } from './FormKafka'
import { FormMySQL } from './FormMySQL'
import { FormPulsar } from './FormPulsar'
import { FormS3 } from './FormS3'
import { FormTiDBServerless, useServerlessSinkEndpointFlow } from './FormTiDBServerless'
import { getPayload } from './payloadBuilders'
import { buildActiveServerlessClusterOptions } from './serverlessClusterOptions'
import type { ConfigureTargetForm } from './types'

const isValidType = (t: string): boolean =>
  (
    [
      DataflowConnectorType.KAFKA,
      DataflowConnectorType.MYSQL,
      DataflowConnectorType.TIDB,
      DataflowConnectorType.SERVERLESS_TIDB,
      DataflowConnectorType.S3,
      DataflowConnectorType.GCS,
      DataflowConnectorType.PULSAR,
      DataflowConnectorType.AZURE_BLOB_STORAGE
    ] as string[]
  ).includes(t)

export const ConfigureTarget = ({
  formData,
  cluster,
  onFinish,
  onCancel
}: {
  cluster: SharedCluster
  formData: FormData
  onFinish: (data: FormData) => void
  onCancel: () => void
}) => {
  const isEditing = useIsEditingChangefeed()
  const {
    store: { tenantData, activeProjectId }
  } = useStores()

  const params = useMemo(() => new URLSearchParams(window.location.search), [])
  const typeFromUrl = params.get('type') as DataflowConnectorType
  const connectionMethods = getChangefeedConnectionMethods(cluster.provider!)
  const methods = useForm<ConfigureTargetForm>({
    shouldUnregister: true,
    defaultValues: {
      connector_type: formData.connector_type ?? (isValidType(typeFromUrl) ? typeFromUrl : DataflowConnectorType.KAFKA),
      network_info: {
        network_type: connectionMethods.at(0)?.value,
        private_link_service_name: '',
        ports: [],
        ports_str: formData.network_info?.ports?.join(',') ?? '',
        ...formData.network_info
      },
      kafka: {
        provider: DataflowKafkaType.KafkaType_Self_Hosted,
        broker: {
          kafka_version: DataflowKafkaVersion.KAFKA_VERSION_2XX,
          broker_endpoints: '',
          compression: DataflowCompressionType.NONE,
          tls_enable: false,
          ...formData?.kafka_info?.broker
        },
        authentication: {
          auth_type: DataflowKafkaAuthType.DISABLE,
          userName: '',
          password: '',
          ...formData?.kafka_info?.authentication
        }
      },
      mysql: {
        endpoint: formData.mysql_info?.connection?.endpoint ?? '',
        username: formData.mysql_info?.connection?.username ?? '',
        password: formData.mysql_info?.connection?.password ?? ''
      },
      tidb: {
        cluster_id: formData.serverless_tidb_info?.cluster_id ?? '',
        mysql_info: {
          ...formData.serverless_tidb_info?.mysql_info,
          privatelink_info: {
            df_privatelink_endpoint_id: '',
            ...formData.serverless_tidb_info?.mysql_info?.privatelink_info
          },
          connection: {
            username: '',
            password: '',
            ...formData.serverless_tidb_info?.mysql_info?.connection
          }
        }
      },
      s3_info: {
        endpoint: {
          auth_type: DataflowS3AuthType.S3_AUTH_TYPE_ASSUME_ROLE,
          ...formData.s3_info?.endpoint
        },
        ...formData?.s3_info
      },
      gcs_info: {
        endpoint: {
          gcsUri: '',
          ...formData.gcs_info?.endpoint
        },
        ...formData?.gcs_info
      },
      azure_blob_storage_info: {
        endpoint: {
          blob_uri: '',
          sas_token: '',
          ...formData.azure_blob_storage_info?.endpoint
        },
        ...formData?.azure_blob_storage_info
      },
      pulsar_info: {
        destination_protocol: DataflowPulsarDestinationProtocol.PULSAR_PROTOCOL_PULSAR,
        compression_type: DataflowPulsarCompressionType.PULSAR_COMPRESSION_NONE,
        authentication: {
          auth_type: DataflowPulsarAuthType.PULSAR_AUTH_TYPE_BASIC,
          user_name: '',
          password: '',
          token: '',
          ...formData.pulsar_info?.authentication
        },
        advanced_setting: {
          batching_max_messages: 1000,
          batching_max_publish_delay: 10,
          connection_timeout: 5,
          operation_timeout: 30,
          send_timeout: 30,
          ...formData.pulsar_info?.advanced_setting
        },
        ...formData.pulsar_info
      }
    }
  })
  const { handleSubmit, watch, getValues, setValue } = methods
  const type = watch('connector_type')
  const targetServerlessClusterId = watch('tidb.cluster_id') as string
  // const tidbid = watch('tidb.pro')
  const sinkEndpointId = (watch('tidb.mysql_info.privatelink_info.df_privatelink_endpoint_id') as string) || ''
  const testConnectorMutation = useTestConnector({ mutation: { retry: false }, request: { timeout: 120 * 1000 } })
  const { data: gcsProfile } = useConnectorServiceDescribeAdminProfile(tenantData.id, activeProjectId, cluster.id, {
    query: {
      enabled: type === DataflowConnectorType.GCS,
      refetchOnWindowFocus: false
    }
  })
  const supportSinkToServerless = useSupportChangefeedToServerlessTiDB(cluster)
  const supportPulsarChangefeed = useSupportPulsarChangefeed(cluster)

  const { data: serverlessClusters } = useTiDBServiceListTidbs(
    {
      orgId: tenantData.id,
      skip: 0,
      pageSize: 100,
      servicePlans: [
        TidbServiceListTidbsServicePlan.Starter,
        TidbServiceListTidbsServicePlan.Essential
        // TidbServiceListTidbsServicePlan.Essential_V2
      ],
      regionIds: [cluster.regionId]
    },
    {
      query: {
        refetchOnWindowFocus: false
      }
    }
  )
  const hasServerlessCluster = buildActiveServerlessClusterOptions(serverlessClusters?.tidbs).length > 0
  const targetServerlessProjectId = useMemo(() => {
    const targetCluster = serverlessClusters?.tidbs?.find(
      (tidb) =>
        tidb.starterInstance?.clusterId === targetServerlessClusterId ||
        tidb.essentialInstance?.clusterId === targetServerlessClusterId ||
        tidb.essentialV2Instance?.tidbId === targetServerlessClusterId
    )

    return targetCluster?.projectId || ''
  }, [serverlessClusters, targetServerlessClusterId])

  const TARGETS: FormRadioGroupProps['data'] = [
    {
      label: ChangefeedNames.KAFKA,
      value: DataflowConnectorType.KAFKA,
      disabled: isEditing
    },
    {
      label: ChangefeedNames.PULSAR,
      value: DataflowConnectorType.PULSAR,
      disabled: isEditing,
      show: supportPulsarChangefeed
    },
    {
      label: ChangefeedNames.MYSQL,
      value: DataflowConnectorType.MYSQL,
      disabled: isEditing
    },
    {
      label: ChangefeedNames.S3,
      value: DataflowConnectorType.S3,
      disabled: isEditing,
      show: cluster.provider === CloudProviderConst.AWS
    },
    {
      label: ChangefeedNames.GCS,
      value: DataflowConnectorType.GCS,
      disabled: isEditing,
      show: cluster.provider === CloudProviderConst.GCP
    },
    {
      label: ChangefeedNames.AZURE_BLOB_STORAGE,
      value: DataflowConnectorType.AZURE_BLOB_STORAGE,
      disabled: isEditing,
      show: cluster.provider === CloudProviderConst.AZURE
    },
    {
      label: ChangefeedNames.SERVERLESS_TIDB,
      value: DataflowConnectorType.SERVERLESS_TIDB,
      show: cluster.provider !== CloudProviderConst.GCP,
      disabled: isEditing || !hasServerlessCluster || !supportSinkToServerless,
      tooltipProps: {
        position: 'top' as const,
        shadow: 'md'
      },
      tooltip: !hasServerlessCluster ? (
        <Typography variant="body-md">You need to create a Starter instance in the same region first.</Typography>
      ) : supportSinkToServerless ? null : (
        <Typography variant="body-md">
          Not supported in current region. You can find the supported regions at this{' '}
          <a
            href="https://docs.pingcap.com/tidbcloud/changefeed-sink-to-tidb-cloud#restrictions"
            target="_blank"
            rel="noreferrer"
          >
            link
          </a>
          .
        </Typography>
      )
    }
  ]
    .filter((i) => i?.show !== false)
    .map((i) => omit(i, 'show'))

  // test connection
  const onNext = async (data: ConfigureTargetForm) => {
    const payload = getPayload(data, formData, isEditing)

    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]-[Step1]-[Payload]-' + JSON.stringify(payload))
        onFinish(payload)
        return
      }

      // test connection when create or edited password/secrets
      if (
        !isEditing ||
        payload.password ||
        payload.s3_info?.endpoint?.secret_key ||
        payload.s3_info?.endpoint?.iam_assumed_role_arn ||
        payload.azure_blob_storage_info?.endpoint?.sas_token
      ) {
        await testConnectorMutation.mutateAsync({
          orgId: tenantData.id,
          projectId: activeProjectId,
          clusterId: cluster.id,
          data: {
            ...omit(payload, 'password'),
            connector_type: payload.connector_type!
          }
        })
      }

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

  const {
    createServerlessSinkEndpoint,
    error: serverlessSinkEndpointError,
    isCreatingServerlessSinkEndpoint,
    isLoading: isLoadingServerlessSinkEndpoint,
    serverlessSinkEndpointData
  } = useServerlessSinkEndpointFlow({
    orgId: tenantData.id,
    projectId: activeProjectId,
    targetClusterId: targetServerlessClusterId,
    targetProjectId: targetServerlessProjectId,
    serverlessProvider: cluster.provider,
    serverlessRegion: cluster.region,
    sinkEndpointId,
    enabled: type === DataflowConnectorType.SERVERLESS_TIDB,
    onAvailable: () => onNext(getValues()),
    setValue
  })
  const serverlessSinkEndpointStatus = serverlessSinkEndpointData?.status
  const isServerlessSinkEndpointCreating =
    isCreatingServerlessSinkEndpoint ||
    serverlessSinkEndpointStatus === 'SINK_ENDPOINT_STATUS_CREATING' ||
    serverlessSinkEndpointStatus === 'SINK_ENDPOINT_STATUS_UNSPECIFIED'
  const isServerlessSinkEndpointPendingAcceptance =
    serverlessSinkEndpointStatus === 'SINK_ENDPOINT_STATUS_PENDING_ACCEPTANCE'
  const isServerlessSinkEndpointFailed = serverlessSinkEndpointStatus === 'SINK_ENDPOINT_STATUS_FAILED'
  const progressMessage = testConnectorMutation.isLoading
    ? 'Testing downstream connection for changefeed: verifying data link and responsiveness.'
    : type === DataflowConnectorType.SERVERLESS_TIDB && isServerlessSinkEndpointCreating
      ? 'Initializing changefeed connection: preparing resources and configuration.'
      : type === DataflowConnectorType.SERVERLESS_TIDB && isServerlessSinkEndpointPendingAcceptance
        ? 'Awaiting network availability to establish changefeed connection... This step may take a couple of minutes. Please be patient.'
        : type === DataflowConnectorType.SERVERLESS_TIDB && isServerlessSinkEndpointFailed
          ? 'Failed to initialize changefeed connection. Please retry or contact technical support.'
          : ''
  const progressAlertIcon =
    type === DataflowConnectorType.SERVERLESS_TIDB && isServerlessSinkEndpointFailed ? (
      <IconAlertCircle size={16} />
    ) : (
      <Loader size="xs" />
    )
  const shouldWaitForServerlessSinkEndpoint =
    type === DataflowConnectorType.SERVERLESS_TIDB &&
    !!sinkEndpointId &&
    serverlessSinkEndpointStatus !== 'SINK_ENDPOINT_STATUS_AVAILABLE' &&
    serverlessSinkEndpointStatus !== 'SINK_ENDPOINT_STATUS_FAILED'

  // create sink endpoint first if needed
  const onSubmit = async (data: ConfigureTargetForm) => {
    if (isEditing) {
      return await onNext(data)
    }

    const payload = getPayload(data, formData, isEditing)
    if (payload.connector_type === DataflowConnectorType.SERVERLESS_TIDB) {
      await createServerlessSinkEndpoint()
      return
    }

    await onNext(data)
  }

  const error = testConnectorMutation.error || serverlessSinkEndpointError

  return (
    <FormProvider {...methods}>
      <Card p={0}>
        <form onSubmit={handleSubmit(onSubmit)}>
          <Stack m={24} gap={6}>
            <HeadLabel label="Destination" />

            <FormRadioGroup name="connector_type" data={TARGETS} sx={{ marginTop: 0 }} />
          </Stack>

          {type === DataflowConnectorType.KAFKA && (
            <FormKafka cluster={cluster} orgId={tenantData.id} projectId={activeProjectId} />
          )}
          {type === DataflowConnectorType.MYSQL && <FormMySQL cluster={cluster} />}
          {type === DataflowConnectorType.SERVERLESS_TIDB && (
            <FormTiDBServerless orgId={tenantData.id} cluster={cluster} />
          )}
          {type === DataflowConnectorType.S3 && <FormS3 />}
          {type === DataflowConnectorType.GCS && <FormGCS gcp_account_id={gcsProfile?.gcp_account_id} />}
          {type === DataflowConnectorType.AZURE_BLOB_STORAGE && <FormAzureBlob />}
          {type === DataflowConnectorType.PULSAR && <FormPulsar />}

          <Divider />

          {type === DataflowConnectorType.SERVERLESS_TIDB && progressMessage && (
            <Alert icon={progressAlertIcon} color="peacock" m={24}>
              {progressMessage}
            </Alert>
          )}
          {error && (
            <Alert icon={<IconAlertCircle size={16} />} color="red" m={24}>
              <Box
                sx={{ wordBreak: 'break-all' }}
                dangerouslySetInnerHTML={{
                  __html: getErrorMessage(error, true)
                }}
              />
            </Alert>
          )}

          <Group m={24} justify="flex-end">
            <Button
              variant="default"
              type="button"
              data-mp-event="Configure Target Cancel Button Clicked"
              onClick={onCancel}
            >
              Cancel
            </Button>
            <Button
              type="submit"
              data-mp-event="Configure Target Next Button Clicked"
              loading={
                testConnectorMutation.isLoading ||
                isLoadingServerlessSinkEndpoint ||
                shouldWaitForServerlessSinkEndpoint
              }
            >
              Next
            </Button>
          </Group>
        </form>
      </Card>
    </FormProvider>
  )
}