import {
  Alert,
  Box,
  Button,
  Card,
  Checkbox,
  Divider,
  Group,
  HoverCard,
  Input,
  Loader,
  notifier,
  Radio,
  Stack,
  Tooltip,
  Typography
} from '@tidbcloud/uikit'
import {
  CodeBlock,
  FormErrorMessage,
  FormRadioGroup,
  FormSegmentedControl,
  FormSelect,
  FormTextInput
} from '@tidbcloud/uikit/biz'
import { IconAlertCircle, IconCheckCircle, IconCopy01, IconInfoCircle, IconXCircle } from '@tidbcloud/uikit/icons'
import { useEffect, useState } from 'react'
import { Controller, FormProvider, useForm, useFormContext } from 'react-hook-form'

import { getEndpointServiceNameLabel } from 'dbaas/features/Changefeed/helpers'
import { CloudProviderConst } from 'dbaas/features/Project'
import { getDomainType } from 'dbaas/screens/ClusterV2/features/Network/dedicated/PrivateEndpointChangefeed/Detail'
import { removeRootDomain } from 'dbaas/screens/ClusterV2/features/Network/dedicated/PrivateEndpointChangefeed/helper'
import { SharedTiDB } from 'dbaas/screens/NextGen/shared/model/shared-tidb-state'
import {
  useNgCreateSinkEndpoint,
  useNgDescribeEndpointServiceAvailabilityZones,
  useNgGenerateSinkAdvertisedListener,
  useNgGetSinkEndpoint,
  useNgGetSinkEndpointAuthInfo,
  useNgListSinkAvailabilityZones,
  useNgValidateSinkAdvertisedListenerDomain
} from 'dbaas/services/dataflow'
import {
  ConsoleapiSinkEndpointType,
  NgListSinkEndpointsStatus,
  SinkAdvertisedListenerDomainType,
  SinkAdvertisedListenerRecord
} from 'dbaas/services/dataflow/interface'
import { getErrorMessage } from 'dbaas/services/errorCodes'

import { shouldShowEssentialV2AwsPrincipal } from './auth-info'

const composeFullDomain = (subdomain?: string, rootDomain?: string) => {
  if (!subdomain || !rootDomain) {
    return `${subdomain ?? ''}${rootDomain ?? ''}`
  }

  // Ensure subdomain ends with '.'
  const normalizedSubdomain = subdomain.endsWith('.') ? subdomain : `${subdomain}.`

  // Ensure root domain does not start with '.'
  const normalizedRootDomain = rootDomain.startsWith('.') ? rootDomain.slice(1) : rootDomain

  return `${normalizedSubdomain}${normalizedRootDomain}`
}

interface CreatePrivateEndpointForChangefeedFormData {
  endpoint_name: string
  connection_type: ConsoleapiSinkEndpointType
  endpoint_service_name: string
  msk_cluster_arn: string
  az_type: '3_azs' | 'single_az'
  endpoint_zones: string[]
  with_advertised_listener: boolean
  domain_type: SinkAdvertisedListenerDomainType
  downstream_id: string
  root_domain_name: string
  advertised_listener_records?: SinkAdvertisedListenerRecord[]
}

const AWS_CONNECTION_TYPE_OPTIONS = [
  {
    label: 'AWS Endpoint Service',
    value: 'VPC_ENDPOINT_SERVICE'
  },
  {
    label: 'AWS MSK Provisioned',
    value: 'AMAZON_MSK'
  }
] satisfies Array<{ label: string; value: ConsoleapiSinkEndpointType }>

interface CreatePrivateEndpointTrackingEvents {
  cancel: string
  submit: string
}

const defaultTrackingEvents: CreatePrivateEndpointTrackingEvents = {
  cancel: 'Cancel creating private endpoint for changefeed',
  submit: 'Submit creating private endpoint for changefeed'
}

function Reminder({ cluster }: { cluster: SharedTiDB }) {
  const isAws = cluster.cloudProvider === CloudProviderConst.AWS
  const isAzure = cluster.cloudProvider === CloudProviderConst.AZURE
  const isGcp = cluster.cloudProvider === CloudProviderConst.GCP
  const isAli = cluster.cloudProvider === CloudProviderConst.ALICLOUD

  const privateLinkAuthInfo = useNgGetSinkEndpointAuthInfo(cluster.tidbId!, {
    query: {
      refetchOnWindowFocus: false,
      refetchOnMount: false
    }
  })

  return (
    <Alert icon={<IconAlertCircle size={16} />}>
      {isAws && (
        <>
          Authorize the{' '}
          <b>
            {privateLinkAuthInfo.isLoading
              ? 'loading...'
              : `AWS Principal (${privateLinkAuthInfo.data?.aws_principal_id ?? 'N/A'})`}
          </b>{' '}
          of TiDB Cloud to create endpoint for your Private Link service.
        </>
      )}
      {isAzure && (
        <>
          <Typography>
            Suggest your downstream service is deployed in the same zones or part of zones where the TiDB instance is
            deployed to reduce cross-zone traffic.
          </Typography>
          <Typography>
            Authorize the subscription (
            {privateLinkAuthInfo.isLoading
              ? 'loading...'
              : privateLinkAuthInfo.error
                ? getErrorMessage(privateLinkAuthInfo.error)
                : privateLinkAuthInfo.data?.cloud_account_id}
            ) of TiDB Cloud or allow anyone with your alias to access your Private Link service before creating the
            private endpoint for external services.
          </Typography>
        </>
      )}
      {isGcp && (
        <>
          <Typography>
            Suggest your downstream service is deployed in the same zones or part of zones where the TiDB instance is
            deployed to reduce cross-zone traffic.
          </Typography>
          <Typography>
            Add{' '}
            <b>
              {privateLinkAuthInfo.isLoading
                ? 'loading...'
                : `Google Cloud project (${privateLinkAuthInfo.data?.cloud_account_id})`}
            </b>{' '}
            of TiDB Cloud to accept lists of the Private Service Connect in advance or manually accept the endpoint
            creation request later.
          </Typography>
        </>
      )}
      {isAli && (
        <>
          <Typography>
            It is necessary to whitelist{' '}
            <b>
              {privateLinkAuthInfo.isLoading
                ? 'loading...'
                : `TiDB Cloud's Alibaba Cloud account ID (${privateLinkAuthInfo.data?.cloud_account_id})`}
            </b>{' '}
            for your endpoint service to grant the TiDB Cloud VPC access. Refer to{' '}
            <a
              href="https://www.alibabacloud.com/help/en/privatelink/user-guide/add-and-manage-service-whitelists"
              target="_blank"
              rel="noopener noreferrer"
            >
              documentation
            </a>{' '}
            for more details.
          </Typography>
        </>
      )}
    </Alert>
  )
}

function EssentialV2PrincipalInfo({ cluster }: { cluster: SharedTiDB }) {
  const privateLinkAuthInfo = useNgGetSinkEndpointAuthInfo(cluster.tidbId!, {
    query: {
      refetchOnWindowFocus: false,
      refetchOnMount: false
    }
  })
  const awsPrincipal = privateLinkAuthInfo.isLoading
    ? 'loading...'
    : (privateLinkAuthInfo.data?.aws_principal_id ?? 'N/A')

  return (
    <>
      <Group gap={16}>
        <Typography>TiDB Cloud AWS Principal:</Typography>
        <Typography fw={600}>{awsPrincipal}</Typography>
        <Tooltip label="Copy" withArrow position="right">
          <Button
            type="button"
            variant="subtle"
            size="xs"
            p={0}
            disabled={!awsPrincipal || awsPrincipal === 'loading...' || awsPrincipal === 'N/A'}
            onClick={() => {
              void navigator.clipboard.writeText(awsPrincipal)
            }}
          >
            <IconCopy01 size={16} />
          </Button>
        </Tooltip>
      </Group>
      <Alert color="peacock" icon={<IconInfoCircle size={16} />}>
        Copy the Principal of TiDB Cloud above and allow it in AWS for your endpoint service.
      </Alert>
    </>
  )
}

export function CreatePrivateEndpointForChangefeed({
  cluster,
  onCancel,
  onSuccess,
  trackingEvents = defaultTrackingEvents
}: {
  cluster: SharedTiDB
  onCancel: () => void
  onSuccess?: (endpointId: string, endpointName?: string) => void
  trackingEvents?: CreatePrivateEndpointTrackingEvents
}) {
  const isAws = cluster.cloudProvider === CloudProviderConst.AWS
  const isAzure = cluster.cloudProvider === CloudProviderConst.AZURE
  const isGcp = cluster.cloudProvider === CloudProviderConst.GCP
  const methods = useForm<CreatePrivateEndpointForChangefeedFormData>({
    defaultValues: {
      connection_type: 'VPC_ENDPOINT_SERVICE',
      az_type: '3_azs',
      endpoint_zones: [],
      with_advertised_listener: false,
      domain_type: SinkAdvertisedListenerDomainType.GENERATED
    }
  })
  const name = methods.watch('endpoint_name')
  const connectionType = methods.watch('connection_type')
  const endpointServiceName = methods.watch('endpoint_service_name')
  const mskClusterArn = methods.watch('msk_cluster_arn')
  const endpointZones = methods.watch('endpoint_zones') ?? []
  const withAdvertisedListener = methods.watch('with_advertised_listener')
  const domainType = methods.watch('domain_type')
  const rootDomainName = methods.watch('root_domain_name')
  const create = useNgCreateSinkEndpoint()
  const createdEndpointId = create.data?.id || create.data?.endpoint_id || ''

  const { data: sinkEndpointData, isLoading: isLoadingSinkEndpoint } = useNgGetSinkEndpoint(
    cluster.tidbId!,
    createdEndpointId,
    {
      query: {
        enabled: create.isSuccess && !!createdEndpointId,
        refetchOnWindowFocus: false,
        onSuccess(data) {
          if (data.status !== NgListSinkEndpointsStatus.SINK_ENDPOINT_STATUS_AVAILABLE) {
            return
          }

          const endpointId = data.id ?? data.endpoint_id
          if (endpointId) {
            onSuccess?.(endpointId, data.endpoint_name)
          }
        },
        refetchInterval(data) {
          if (
            data?.status === NgListSinkEndpointsStatus.SINK_ENDPOINT_STATUS_CREATING ||
            data?.status === NgListSinkEndpointsStatus.SINK_ENDPOINT_STATUS_PENDING_ACCEPTANCE
          ) {
            return 5000
          }

          return false
        }
      }
    }
  )

  const privateLinkAuthInfo = useNgGetSinkEndpointAuthInfo(cluster.tidbId!, {
    query: {
      refetchOnWindowFocus: false,
      refetchOnMount: false
    }
  })

  const onSubmit = async (data: CreatePrivateEndpointForChangefeedFormData) => {
    const isAmazonMsk = data.connection_type === 'AMAZON_MSK'

    await create.mutateAsync({
      clusterId: cluster.tidbId!,
      data: {
        sink_endpoint: {
          endpoint_name: data.endpoint_name,
          endpoint_type: data.connection_type,
          endpoint_service_name: isAmazonMsk ? undefined : data.endpoint_service_name,
          msk_arn: isAmazonMsk ? data.msk_cluster_arn : undefined,
          auth_type: isAmazonMsk ? 'SASL_SCRAM' : undefined,
          provider: cluster.cloudProvider ?? '',
          region: cluster.regionId ?? '',
          endpoint_zones: isAmazonMsk ? undefined : data.endpoint_zones,
          service_region: isAmazonMsk ? undefined : (cluster.regionId ?? ''),
          advertised_listener:
            !isAmazonMsk && data.with_advertised_listener
              ? {
                  domain_type: data.domain_type,
                  root_domain_name: data.root_domain_name,
                  records: isAws
                    ? data.advertised_listener_records?.map((i) => ({
                        ...i,
                        // Ensure the final domain ends with root domain name
                        subdomain_name: composeFullDomain(i.subdomain_name, data.root_domain_name)
                      }))
                    : undefined
                }
              : undefined
        }
      }
    })
  }

  const isCreatingPrivateLink =
    isLoadingSinkEndpoint || sinkEndpointData?.status === NgListSinkEndpointsStatus.SINK_ENDPOINT_STATUS_CREATING

  const successMessage = (() => {
    if (!create.isSuccess) {
      return null
    }

    if (!sinkEndpointData || sinkEndpointData?.status === NgListSinkEndpointsStatus.SINK_ENDPOINT_STATUS_CREATING) {
      return <Typography>Creating the private endpoint for external services...</Typography>
    }

    if (sinkEndpointData?.status === NgListSinkEndpointsStatus.SINK_ENDPOINT_STATUS_FAILED) {
      return <Typography>Failed to create the private endpoint for external services.</Typography>
    }

    if (sinkEndpointData?.status === NgListSinkEndpointsStatus.SINK_ENDPOINT_STATUS_PENDING_ACCEPTANCE) {
      if (isAws) {
        if (connectionType === 'AMAZON_MSK') {
          return (
            <Typography>
              Please go to AWS console and verify the corresponding Amazon MSK VPC Connection status is accepted and
              available.
            </Typography>
          )
        }

        return (
          <Typography>
            Please go to your AWS console, locate the Private Link service(<b>{create.data.endpoint_service_name}</b>),
            and accept the request from VPC Endpoint (<b>{sinkEndpointData?.endpoint_id ?? 'N/A'}</b>).
          </Typography>
        )
      }

      if (isAzure) {
        return (
          <Typography>
            Please go to Microsoft Azure portal, locate the Private Link service (<b>{endpointServiceName}</b>) and
            accept the request from the Private Endpoint (<b>{sinkEndpointData?.endpoint_id ?? 'N/A'}</b>) in{' '}
            {'"Settings -> Private endpoint connections"'} of the Private Link service.
          </Typography>
        )
      }

      if (isGcp) {
        return (
          <Typography>
            Please go to your Google Cloud console, locate the Private Service Connect service (
            <b>{endpointServiceName}</b>) and accept the request from project (
            <b>{privateLinkAuthInfo.data?.cloud_account_id ?? 'N/A'}</b>).
          </Typography>
        )
      }
    }

    if (sinkEndpointData?.status === NgListSinkEndpointsStatus.SINK_ENDPOINT_STATUS_AVAILABLE) {
      return <Typography>The private endpoint for external services is created successfully!</Typography>
    }

    return null
  })()

  const successAlert =
    create.isSuccess && successMessage ? (
      <Alert
        color={sinkEndpointData?.status === NgListSinkEndpointsStatus.SINK_ENDPOINT_STATUS_FAILED ? 'red' : 'green'}
        icon={
          isCreatingPrivateLink ? (
            <Loader size="xs" />
          ) : sinkEndpointData?.status === NgListSinkEndpointsStatus.SINK_ENDPOINT_STATUS_FAILED ? (
            <IconXCircle size={16} />
          ) : (
            <IconCheckCircle size={16} />
          )
        }
      >
        <Stack gap={8}>{successMessage}</Stack>
      </Alert>
    ) : null

  const endpointServiceNameLabel = getEndpointServiceNameLabel(cluster.cloudProvider!)
  const isAmazonMsk = isAws && connectionType === 'AMAZON_MSK'
  const essentialV2ZoneCountValid =
    !cluster.computed.isEssentialV2 || (endpointZones.length >= 1 && endpointZones.length <= 2)
  const showEssentialV2AwsPrincipal = shouldShowEssentialV2AwsPrincipal(
    cluster.computed.isEssentialV2,
    cluster.cloudProvider
  )
  const requiredFieldsValid = isAmazonMsk
    ? !!name && !!mskClusterArn
    : !!name && !!endpointServiceName && essentialV2ZoneCountValid

  return (
    <FormProvider {...methods}>
      <Stack component="form" onSubmit={methods.handleSubmit(onSubmit)}>
        <FormTextInput
          name="endpoint_name"
          label={
            isAmazonMsk
              ? 'Private Link Connection Name'
              : cluster.computed.isEssentialV2
                ? 'Private Endpoint Name'
                : 'Name'
          }
          placeholder={isAmazonMsk ? 'Enter your private link connection name' : 'Enter your private endpoint name'}
          rules={{ required: 'Required' }}
        />

        {isAws && (
          <FormSelect
            name="connection_type"
            label="Connection Type"
            data={AWS_CONNECTION_TYPE_OPTIONS}
            allowDeselect={false}
          />
        )}

        {isAmazonMsk ? (
          <AmazonMskProvisionedFields cluster={cluster} />
        ) : (
          <>
            {showEssentialV2AwsPrincipal ? (
              <EssentialV2PrincipalInfo cluster={cluster} />
            ) : (
              <Reminder cluster={cluster} />
            )}

            {cluster.computed.isEssentialV2 ? (
              <EssentialV2EndpointServiceAvailabilityZones cluster={cluster} />
            ) : (
              <>
                <FormTextInput
                  name="endpoint_service_name"
                  label={endpointServiceNameLabel}
                  placeholder={`Enter your ${endpointServiceNameLabel.toLowerCase()}`}
                  rules={{ required: 'Required' }}
                />
                <FormAvailabilityZone cluster={cluster} />
              </>
            )}

            <FormAdvertisedListener cluster={cluster} />
          </>
        )}

        <Divider mx={-24} />

        {create.isError && (
          <FormErrorMessage
            message={getErrorMessage(create.error, true) || 'Failed to create private endpoint for external services'}
          />
        )}

        {successAlert}

        {!create.isSuccess && (
          <Group justify="flex-end">
            <Button type="button" variant="default" data-mp-event={trackingEvents.cancel} onClick={onCancel}>
              Cancel
            </Button>
            <Button
              type="submit"
              disabled={!requiredFieldsValid}
              loading={create.isLoading}
              data-mp-event={trackingEvents.submit}
              data-mp-with-advertised-listener={withAdvertisedListener}
              data-mp-domain-type={domainType}
              data-mp-root-domain-name={rootDomainName}
            >
              Create
            </Button>
          </Group>
        )}

        {create.isSuccess && (
          <Group justify="flex-end">
            <Button type="button" variant="default" onClick={onCancel}>
              Close
            </Button>
          </Group>
        )}
      </Stack>
    </FormProvider>
  )
}

function AmazonMskProvisionedFields({ cluster }: { cluster: SharedTiDB }) {
  const isEssentialV2 = cluster.computed.isEssentialV2
  const authInfo = useNgGetSinkEndpointAuthInfo(cluster.tidbId!, {
    query: {
      enabled: !!cluster.tidbId,
      refetchOnWindowFocus: false,
      refetchOnMount: false
    }
  })
  const availabilityZones = useNgListSinkAvailabilityZones(cluster.tidbId!, {
    query: {
      enabled: !!cluster.tidbId && !isEssentialV2,
      refetchOnWindowFocus: false,
      select(data) {
        return [...(data.availability_zones ?? [])].sort()
      }
    }
  })
  const accountId = authInfo.data?.cloud_account_id || authInfo.data?.aws_principal_id?.match(/::(\d+):/)?.[1] || 'N/A'
  const zoneIds = availabilityZones.data?.join(', ') || 'N/A'
  const documentationPlan = isEssentialV2 ? 'essential' : 'premium'

  return (
    <>
      <FormTextInput
        name="msk_cluster_arn"
        label="MSK Cluster ARN"
        placeholder="Enter your MSK Cluster ARN"
        rules={{ required: 'Required' }}
      />

      <Input.Description>
        Don't know how to set up the AWS MSK cluster? See{' '}
        <a
          href={`https://docs.pingcap.com/tidbcloud/serverless-private-link-connection-to-amazon-msk/?plan=${documentationPlan}`}
          target="_blank"
          rel="noreferrer"
        >
          Connect to AWS MSK Provisioned via a Private Link Connection
        </a>
        .
      </Input.Description>

      <Alert color="peacock" icon={<IconInfoCircle size={16} />}>
        Please make sure the Amazon MSK is attached with a cluster policy for account{' '}
        <b>{authInfo.isLoading ? 'loading...' : accountId}</b>
        {isEssentialV2 ? (
          '. The Amazon MSK cluster can be deployed in any Availability Zone.'
        ) : (
          <> and deployed in the same AZs as listed below: {availabilityZones.isLoading ? 'loading...' : zoneIds}.</>
        )}
      </Alert>
    </>
  )
}

function FormAvailabilityZone({ cluster }: { cluster: SharedTiDB }) {
  const methods = useFormContext<CreatePrivateEndpointForChangefeedFormData>()
  const azType = methods.watch('az_type')
  const { data: availabilityZones = [], isLoading: isLoadingZones } = useNgListSinkAvailabilityZones(cluster.tidbId!, {
    query: {
      enabled: !!cluster.tidbId,
      refetchOnWindowFocus: false,
      select(data) {
        data?.availability_zones?.sort()
        return data.availability_zones ?? []
      },
      onSuccess(data) {
        if (azType === '3_azs') {
          methods.setValue('endpoint_zones', data ?? [])
        } else if (azType === 'single_az') {
          methods.setValue('endpoint_zones', data ? [data[0]] : [])
        }
      }
    }
  })
  const isAws = cluster.cloudProvider === CloudProviderConst.AWS

  // Only AWS support to choose availability zones
  if (!isAws) {
    return (
      <Stack gap={8}>
        <Input.Label>Zones of TiDB Instance</Input.Label>
        <Group>{isLoadingZones ? 'Loading...' : availabilityZones.length ? availabilityZones.join(', ') : 'N/A'}</Group>
      </Stack>
    )
  }

  return (
    <Group>
      <FormRadioGroup
        name="az_type"
        label="Number of AZs"
        data={[
          { label: '3 AZs', value: '3_azs' },
          { label: 'Single AZ', value: 'single_az' }
        ]}
        onChange={(val) => {
          if (val === 'single_az') {
            methods.setValue('endpoint_zones', availabilityZones.slice(0, 1))
          } else {
            methods.setValue('endpoint_zones', availabilityZones)
          }
        }}
      />

      <Divider orientation="vertical" />

      {azType === '3_azs' && (
        <Checkbox.Group label="AZ IDs" value={availabilityZones}>
          <Group mt={6}>
            {isLoadingZones && <Loader size="xs" />}
            {availabilityZones.map((zone) => (
              <Checkbox key={zone} value={zone} label={zone} disabled checked />
            ))}
          </Group>
        </Checkbox.Group>
      )}
      {azType === 'single_az' && (
        <Controller
          name="endpoint_zones"
          render={({ field }) => (
            <Radio.Group
              label="AZ IDs"
              value={field.value[0]}
              onChange={(value) => methods.setValue('endpoint_zones', [value])}
            >
              <Group mt={10}>
                {availabilityZones.map((zone) => (
                  <Radio key={zone} value={zone} label={zone} />
                ))}
              </Group>
            </Radio.Group>
          )}
        />
      )}
    </Group>
  )
}

function EssentialV2FormAvailabilityZone({ availabilityZones }: { availabilityZones: string[] }) {
  const methods = useFormContext<CreatePrivateEndpointForChangefeedFormData>()
  const selectedZones = methods.watch('endpoint_zones') ?? []

  return (
    <Checkbox.Group
      label="AZ IDs"
      description="Choose 1 or 2 AZs where your target endpoint service is available"
      value={selectedZones}
      onChange={(zones) => {
        methods.setValue('endpoint_zones', zones.slice(0, 2), { shouldDirty: true, shouldValidate: true })
      }}
    >
      <Group mt={6}>
        {availabilityZones.map((zone) => (
          <Checkbox
            key={zone}
            value={zone}
            label={zone}
            disabled={selectedZones.length >= 2 && !selectedZones.includes(zone)}
          />
        ))}
      </Group>
    </Checkbox.Group>
  )
}

function EssentialV2EndpointServiceAvailabilityZones({ cluster }: { cluster: SharedTiDB }) {
  const methods = useFormContext<CreatePrivateEndpointForChangefeedFormData>()
  const { setValue } = methods
  const endpointServiceName = methods.watch('endpoint_service_name')?.trim() ?? ''
  const [fetchedEndpointServiceName, setFetchedEndpointServiceName] = useState<string>()
  const endpointServiceZones = useNgDescribeEndpointServiceAvailabilityZones(
    cluster.tidbId!,
    {
      endpoint_service_name: endpointServiceName,
      service_region: cluster.regionId ?? ''
    },
    {
      query: {
        enabled: false,
        refetchOnWindowFocus: false,
        select(data) {
          return (
            data.zones
              ?.map((zone) => zone.availability_zone_id ?? zone.availability_zone)
              .filter((zone): zone is string => !!zone)
              .sort() ?? []
          )
        }
      }
    }
  )
  const availabilityZones = endpointServiceZones.data ?? []
  const canFetchZones = !!cluster.tidbId && !!endpointServiceName && !!cluster.regionId
  const shouldShowFetchResult = fetchedEndpointServiceName === endpointServiceName && !endpointServiceZones.isFetching

  useEffect(() => {
    setValue('endpoint_zones', [])
    setFetchedEndpointServiceName(undefined)
  }, [endpointServiceName, setValue])

  return (
    <>
      <Group align="flex-end">
        <Box sx={{ flex: 1 }}>
          <FormTextInput
            name="endpoint_service_name"
            label="Endpoint Service Name"
            placeholder="Enter your endpoint service name"
            rules={{ required: 'Required' }}
          />
        </Box>
        <Button
          type="button"
          variant="default"
          disabled={!canFetchZones}
          loading={endpointServiceZones.isFetching}
          onClick={async () => {
            setValue('endpoint_zones', [])
            setFetchedEndpointServiceName(undefined)
            await endpointServiceZones.refetch()
            setFetchedEndpointServiceName(endpointServiceName)
          }}
        >
          Fetch Availability Zones
        </Button>
      </Group>

      {shouldShowFetchResult && endpointServiceZones.isError && (
        <Alert color="red" icon={<IconXCircle size={16} />}>
          Unable to retrieve accessible Availability Zones (AZs). Please check that the endpoint service name is correct
          and the principal has been allowed.
        </Alert>
      )}

      {shouldShowFetchResult && endpointServiceZones.isFetched && !endpointServiceZones.isError && (
        <>
          {availabilityZones.length ? (
            <EssentialV2FormAvailabilityZone availabilityZones={availabilityZones} />
          ) : (
            <Alert color="red" icon={<IconXCircle size={16} />}>
              Unable to retrieve accessible Availability Zones (AZs). Please check that the endpoint service name is
              correct and the principal has been allowed.
            </Alert>
          )}
        </>
      )}
    </>
  )
}

const awsKafkaAdvertisedListenerPatternTip = [
  'Find a unique pattern before setup Private Link service.',
  'Setup Kafka Private Link service by the unique pattern.',
  'After the Kafka Private Link service setup, use the same pattern to connect to Kafka Private Link service.'
]

const gcpKafkaAdvertisedListenerPatternTip = [
  'Find a unique pattern before setting up a Private Service Connect service.',
  'Setup Kafka Private Service Connect service by the unique pattern.',
  'After the Kafka Private Service Connect service setup, use the same pattern to connect to Kafka Private Service Connect service.'
]

function FormAdvertisedListener({ cluster }: { cluster: SharedTiDB }) {
  const methods = useFormContext<CreatePrivateEndpointForChangefeedFormData>()
  const withAdvertisedListener = methods.watch('with_advertised_listener')
  const zones = methods.watch('endpoint_zones')
  const domainType = methods.watch('domain_type')
  const rootDomainName = methods.watch('root_domain_name')
  const downstreamId = methods.watch('downstream_id')
  const generateDomain = useNgGenerateSinkAdvertisedListener()
  const validateDomain = useNgValidateSinkAdvertisedListenerDomain()
  const isGcp = cluster.cloudProvider === CloudProviderConst.GCP
  const isAws = cluster.cloudProvider === CloudProviderConst.AWS

  const handleGenerateDomain = async () => {
    try {
      if (domainType === 'CUSTOM') {
        const result = await validateDomain.mutateAsync({
          clusterId: cluster.tidbId!,
          data: {
            endpoint_service_name: methods.getValues('endpoint_service_name'),
            root_domain_name: methods.getValues('root_domain_name'),
            zones: methods.getValues('endpoint_zones')
          }
        })
        methods.setValue(
          'advertised_listener_records',
          result.advertised_listener?.records?.map((i) => ({
            ...i,
            subdomain_name: removeRootDomain(i.subdomain_name!, rootDomainName) || '*'
          })) ?? []
        )
      } else {
        const result = await generateDomain.mutateAsync({
          clusterId: cluster.tidbId!,
          data: {
            endpoint_service_name: methods.getValues('endpoint_service_name'),
            zones,
            downstream_id: downstreamId
          }
        })
        methods.setValue('advertised_listener_records', result.advertised_listener?.records || [])
      }
    } catch (error) {
      notifier.error(getErrorMessage(error, true))
    }
  }

  return (
    <>
      <Controller
        name="with_advertised_listener"
        render={({ field }) => (
          <Checkbox
            checked={field.value}
            onChange={(event) => field.onChange(event.currentTarget.checked)}
            label={
              <Group align="center" gap={8}>
                <span>Configure Advertised Listener for Kafka</span>
                <HoverCard>
                  <HoverCard.Target>
                    <IconInfoCircle />
                  </HoverCard.Target>
                  <HoverCard.Dropdown sx={{ maxWidth: 400 }}>
                    This configuration is only needed for Kafka sink. It supports both TiDB managed and custom domains.
                    Set the advertised listener to ensure clients connect using the correct domain for your Private
                    Endpoint.
                  </HoverCard.Dropdown>
                </HoverCard>
              </Group>
            }
          />
        )}
      />

      {withAdvertisedListener && (
        <>
          <Group align="flex-start">
            <Stack gap={6}>
              <Input.Label>Domain Type</Input.Label>
              <FormSegmentedControl
                name="domain_type"
                data={[
                  {
                    label: getDomainType(SinkAdvertisedListenerDomainType.GENERATED),
                    value: SinkAdvertisedListenerDomainType.GENERATED
                  },
                  {
                    label: getDomainType(SinkAdvertisedListenerDomainType.CUSTOM),
                    value: SinkAdvertisedListenerDomainType.CUSTOM
                  }
                ]}
                onChange={() => {
                  if (generateDomain.data) {
                    generateDomain.reset()
                  }
                  if (validateDomain.data) {
                    validateDomain.reset()
                  }
                }}
              />
            </Stack>

            {domainType === SinkAdvertisedListenerDomainType.GENERATED && (
              <FormTextInput
                name="downstream_id"
                label={
                  <Input.Label mb={0}>
                    <Group gap={8}>
                      <span>Domain Pattern</span>

                      <HoverCard shadow="md" withArrow withinPortal width={400} position="top">
                        <HoverCard.Target>
                          <IconInfoCircle size={14} />
                        </HoverCard.Target>
                        <HoverCard.Dropdown>
                          <Box component="ol" m={0} pl={16}>
                            {(() => {
                              let tips = awsKafkaAdvertisedListenerPatternTip
                              if (isGcp) {
                                tips = gcpKafkaAdvertisedListenerPatternTip
                              }

                              return tips.map((tip) => (
                                <li key={tip}>
                                  <Typography variant="body-lg">{tip}</Typography>
                                </li>
                              ))
                            })()}
                          </Box>
                        </HoverCard.Dropdown>
                      </HoverCard>
                    </Group>
                  </Input.Label>
                }
                placeholder="Enter your domain pattern"
                rules={{ required: 'Required' }}
                sx={{ flex: 1 }}
              />
            )}

            {domainType === SinkAdvertisedListenerDomainType.CUSTOM && (
              <FormTextInput
                name="root_domain_name"
                label="Custom Domain"
                placeholder="Enter your custom domain"
                rules={{ required: 'Required' }}
                sx={{ flex: 1 }}
              />
            )}

            <Button
              variant="default"
              mt={26}
              onClick={handleGenerateDomain}
              loading={generateDomain.isLoading || validateDomain.isLoading}
              disabled={(domainType === 'GENERATED' && !downstreamId) || (domainType === 'CUSTOM' && !rootDomainName)}
            >
              {domainType === 'GENERATED' ? 'Generate' : 'Check'}
            </Button>
          </Group>

          {generateDomain.isSuccess && generateDomain.data && (
            <Card withBorder bg="carbon.2">
              <Stack gap={8}>
                <Typography>
                  Configure the broker address for your Kafka advertised listener in the field corresponding to each
                  availability zone.
                </Typography>

                {generateDomain.data.advertised_listener?.records?.map((item) => (
                  <Box key={item.subdomain_name}>
                    <Group>
                      <Typography miw={60}>
                        <b>{item.zone}</b>
                      </Typography>

                      <CodeBlock sx={{ flex: 1 }}>{item.subdomain_name!}</CodeBlock>
                    </Group>
                  </Box>
                ))}
              </Stack>
            </Card>
          )}

          {validateDomain.isSuccess && validateDomain.data && (
            <Card withBorder bg="carbon.2">
              <Stack gap={8}>
                <Typography>
                  Make sure brokers are configured as following pattern, then we can guarantee network connectivity.
                </Typography>

                {validateDomain.data.advertised_listener?.records?.map((record, index) => (
                  <Box key={record.zone}>
                    <Group>
                      <Typography miw={60}>
                        <b>{record.zone}</b>
                      </Typography>

                      <FormTextInput
                        name={`advertised_listener_records.${index}.subdomain_name`}
                        rightAddon={`.${rootDomainName}`}
                        placeholder={record.zone}
                        disabled={removeRootDomain(record.subdomain_name!, rootDomainName) === ''}
                        sx={{ flex: 1 }}
                        rules={isAws ? { required: 'Required' } : undefined}
                      />
                    </Group>
                  </Box>
                ))}
              </Stack>
            </Card>
          )}
        </>
      )}
    </>
  )
}