/*
 * Copyright 2021 PingCAP, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package service

import (
	"context"
	"crypto/md5"

	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/httputil"
	"net/url"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/pingcap/log"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	k8serr "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	k8s "k8s.io/client-go/kubernetes"

	ctlCli "sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/pingcap/mimir/config"
	"github.com/pingcap/mimir/errors"
	"github.com/pingcap/mimir/model"
	"github.com/pingcap/mimir/pkg/protos"
)

type APIService interface {
	CreateSink(ctx context.Context, param *protos.CreateSinkParam) (*protos.CreateSinkData, error)
	ListSink(ctx context.Context) (*protos.ListSinkData, error)
	RmSink(ctx context.Context, param *protos.RmSinkParam) (*protos.RmSinkData, error)
	SubmitJob(ctx context.Context, param *protos.SubmitJobParam) (*protos.SubmitJobData, error)
	ListJob(ctx context.Context, param *protos.ListJobParam) (*protos.ListJobData, error)
	StopJob(ctx context.Context, param *protos.StopJobParam) (*protos.StopJobData, error)
	ResumeJob(ctx context.Context, param *protos.ResumeJobParam) (*protos.ResumeJobData, error)
	GetLogs(ctx context.Context, param *protos.GetLogsParam) (io.ReadCloser, error)
	RmJob(ctx context.Context, param *protos.RmJobParam) (*protos.RmJobData, error)
	ShowSql(ctx context.Context, param *protos.ShowJobSqlParam) (*protos.ShowJobSqlData, error)
	CreateIngress(ctx context.Context, param *protos.CreateIngressParam) (*protos.CreateIngressData, error)
	RmIngress(ctx context.Context, param *protos.RmIngressParam) (*protos.RmIngressData, error)
	SubmitJobV2(ctx context.Context, param *protos.SubmitJobParamV2) (*protos.SubmitJobData, error)
	ListIngress(ctx context.Context) (*protos.ListIngressData, error)
	FlinkSessionProxy(prefix string) http.Handler
}

// apiservice: inner APIServer implementation
// TODO: refactor to be structural
type apiService struct {
	k8sConfigFile              string
	k8sClientSet               k8s.Interface
	flinkSessionEndpoint       *url.URL
	flinkClientPath            string
	flinkImage                 string
	flinkRestExposedType       string
	flinkClusterServiceAccount string
	flinkPersistenceConfig     config.FlinkPersistenceConfig
	executorCMD                []string
	jobDao                     model.JobDao
	sinkDao                    model.SinkDao
	savepointDao               model.SavepointDao
	Kafkaconf                  config.KafkaConf
	ingressDao                 model.IngressDao
	bizDBUser                  string
	bizDBPassword              string
	bizDBHost                  string
	bizDBPort                  string
	defaultmapper              config.Mapper
}

type Option func(service *apiService) error

func FlinkClientPathOpt(path string) Option {
	return func(service *apiService) error {
		service.flinkClientPath = path
		return nil
	}
}

func FlinkImageOpt(image string) Option {
	return func(service *apiService) error {
		service.flinkImage = image
		return nil
	}
}

func FlinkSessionOpt(endpoint string) Option {
	return func(service *apiService) error {
		var err error
		service.flinkSessionEndpoint, err = url.Parse(endpoint)
		return err
	}
}

func FlinkRestExposeType(typ string) Option {
	return func(service *apiService) error {
		allow := map[string]struct{}{
			"":             {},
			"ClusterIP":    {},
			"NodePort":     {},
			"LoadBalancer": {},
		}
		if _, ok := allow[typ]; !ok {
			return fmt.Errorf("%s is not support", typ)
		}
		service.flinkRestExposedType = typ
		return nil
	}
}

func FlinkClusterServiceAccountOpt(sa string) Option {
	return func(service *apiService) error {
		service.flinkClusterServiceAccount = sa
		return nil
	}
}

func FlinkPersistenceOpt(config config.FlinkPersistenceConfig) Option {
	return func(service *apiService) error {
		service.flinkPersistenceConfig = config
		return nil
	}
}

func ExecutorCMDOpt(cmd []string) Option {
	return func(service *apiService) error {
		service.executorCMD = cmd
		return nil
	}
}

func K8sConfigFileOpt(file string) Option {
	return func(service *apiService) error {
		service.k8sConfigFile = file
		return nil
	}
}

func K8sClientOpt(clientSet k8s.Interface) Option {
	return func(service *apiService) error {
		service.k8sClientSet = clientSet
		return nil
	}
}

func JobDaoOpt(dao model.JobDao) Option {
	return func(service *apiService) error {
		service.jobDao = dao
		return nil
	}
}

func SavepointDaoOpt(dao model.SavepointDao) Option {
	return func(service *apiService) error {
		service.savepointDao = dao
		return nil
	}
}

func KafkaOpt(kc *config.KafkaConf) Option {
	return func(service *apiService) error {
		if kc != nil {
			service.Kafkaconf = *kc
		}
		return nil
	}
}

func MapperOpt(mapper []config.Mapper) Option {
	return func(service *apiService) error {
		if len(mapper) != 0 {
			service.defaultmapper = mapper[0]
		}
		return nil
	}
}

func BizDsnOpt(client ctlCli.Client, secretName string, namespace string) Option {
	return func(service *apiService) error {
		var secret corev1.Secret

		err := client.Get(context.TODO(), types.NamespacedName{
			Namespace: namespace,
			Name:      secretName,
		}, &secret)
		if err != nil {
			return nil
		}
		service.bizDBUser = string(secret.Data["user"])
		service.bizDBPassword = string(secret.Data["password"])
		service.bizDBHost = string(secret.Data["host"])
		service.bizDBPort = string(secret.Data["port"])
		return nil
	}
}

func IngressDaoOpt(dao model.IngressDao) Option {
	return func(service *apiService) error {
		service.ingressDao = dao
		return nil
	}
}

// New return a APIService
func New(opts ...Option) (*apiService, error) {
	svc := new(apiService)
	for _, opt := range opts {
		if err := opt(svc); err != nil {
			return nil, err
		}
	}
	return svc, nil
}

// FlinkSessionProxy returns a proxy that proxy requests from /flink-api/xxx to job manager in flink session cluster
func (svc *apiService) FlinkSessionProxy(prefix string) http.Handler {
	proxy := httputil.NewSingleHostReverseProxy(svc.flinkSessionEndpoint)
	originDirector := proxy.Director
	director := func(req *http.Request) {
		// before calling the real director, we need remove the prefix
		req.URL.Path = strings.TrimPrefix(req.URL.Path, prefix)
		req.URL.RawPath = strings.TrimPrefix(req.URL.RawPath, prefix)
		originDirector(req)
	}
	proxy.Director = director
	return proxy
}

// SubmitJob use the param to submit a flink job in application mode
func (svc *apiService) SubmitJob(ctx context.Context, param *protos.SubmitJobParam) (*protos.SubmitJobData, error) {
	// 1. check if cluster is already exist.
	// 2. create secret which contains the input sql.
	// 3. exec flink-client run-application with customized arguments.
	// 4. save record to db.
	// 5. set ownerRef for created configMap.

	// TODO: Refactor the hardcode namespace.
	namespace := protos.DefaultNamespace

	if len(param.ClusterID) != 0 {
		// TODO: check if exists in database
		_, err := svc.k8sClientSet.AppsV1().Deployments(namespace).Get(ctx, param.ClusterID, metav1.GetOptions{})
		if err == nil {
			return nil, errors.ErrAlreadyExisted(fmt.Sprintf("cluster %v already exist", param.ClusterID))
		} else if status, ok := err.(k8serr.APIStatus); ok && status.Status().Reason != metav1.StatusReasonNotFound {
			log.Error("get deploy error", zap.Error(err))
		}
	}
	secretName := fmt.Sprintf("%x", md5.Sum([]byte(param.SQL)))
	data := new(protos.SubmitJobData)
	_, err := svc.createSecret(ctx, namespace, secretName, param.SQL)
	if err != nil {
		log.Error("create secret error", zap.Error(err))
		return nil, errors.ErrInternal(err.Error())
	}

	property := make(map[string]string)
	if param.CheckpointOn {
		if param.CheckpointInterval != 0 {
			property["execution.checkpointing.interval"] = fmt.Sprintf("%d", param.CheckpointInterval)
		} else {
			property["execution.checkpointing.interval"] = fmt.Sprintf("%d", svc.flinkPersistenceConfig.DefaultCheckpointInterval)
		}
	}

	clusterID, jmEndpoint, err := svc.runApplication(ctx, namespace, param.ClusterID, secretName, property)
	if err != nil {
		log.Error("run application error", zap.Error(err))
		return nil, errors.ErrInternal(err.Error())
	}
	job := new(model.Job)
	job.JobID = clusterID
	// TODO: Refactor the hardcode UserID.
	job.UserID = protos.DefaultUserID
	job.Namespace = namespace
	job.JMEndpoint = jmEndpoint
	job.JobType = model.JobTypeFlinkStreaming
	job.InputType = model.InputTypeSQL
	job.InputKey = secretName
	if err := svc.jobDao.Insert(ctx, job); err != nil {
		log.Error("insert job error", zap.Error(err))
		return nil, errors.ErrInternal(fmt.Sprintf("insert job %s failed", clusterID))
	}

	data.ClusterID = clusterID
	return data, nil
}

func (svc *apiService) ShowSql(ctx context.Context, param *protos.ShowJobSqlParam) (*protos.ShowJobSqlData, error) {
	log.Info("into show job sql")
	const key = "input.sql"
	job, exist, err := svc.jobDao.Find(ctx, protos.DefaultUserID, param.ClusterID)
	if err != nil {
		return nil, errors.ErrInternal(fmt.Sprintf("select job failed %s", err))
	}
	if !exist {
		return nil, errors.ErrNotFound("Job dont exist")
	}
	secretName := job.InputKey
	namespace := protos.DefaultNamespace
	secret, err := svc.k8sClientSet.CoreV1().Secrets(namespace).Get(ctx, secretName, metav1.GetOptions{})
	if err != nil {
		return nil, errors.ErrNotFound(fmt.Sprintf("failed get secret, %s", err))
	}
	sql, ok := secret.Data[key]
	if !ok {
		return nil, errors.ErrNotFound(fmt.Sprintf("no such secret: %s:%s", secretName, key))
	}
	return &protos.ShowJobSqlData{Sql: string(sql)}, nil
}

// flink status refer: https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/common/JobStatus.java
// + pending
func (svc *apiService) ListJob(ctx context.Context, _ *protos.ListJobParam) (*protos.ListJobData, error) {
	// TODO: Refactor the hardcode userID.
	userID := protos.DefaultUserID
	// TODO: Refactor the hardcode Limit.
	cond := model.ListJobCond{
		UserID: userID,
		Limit:  100,
	}
	jobs, err := svc.jobDao.List(ctx, &cond)
	if err != nil {
		log.Error("list job error", zap.Error(err))
		return nil, errors.ErrInternal("list job failed, caused by dao connection")
	}

	// TODO: configurable timeout
	queryCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()

	wg := new(sync.WaitGroup)
	jobStatusMap := make(map[string]string)
	mutex := new(sync.RWMutex)

	for _, job := range jobs {
		wg.Add(1)
		go func(ctx context.Context, jobID string, jmEndpoint string) {
			defer wg.Done()
			var status string
			var exist bool
			var available bool
			var err error

			err = nil
			status = "STOPPED"
			if exist, err = svc.ifFlinkDeploymentExist(ctx, jobID); exist {
				status = "FAILING"
				if available, err = svc.ifFlinkDeploymentAvaliable(ctx, jobID); available {
					status, err = svc.getFlinkJobStatus(jmEndpoint)
				}
			}
			if err != nil {
				log.Info("get flink job status:", zap.Error(err))
				status = "UNKNOWN"
			}
			mutex.Lock()
			jobStatusMap[jobID] = status
			mutex.Unlock()
		}(queryCtx, job.JobID, job.JMEndpoint)
	}

	data := new(protos.ListJobData)
	data.Items = make([]protos.ListJobItem, len(jobs))

	wg.Wait()
	for i, job := range jobs {
		data.Items[i] = protos.ListJobItem{
			UserID:    job.UserID,
			JobID:     job.JobID,
			Namespace: job.Namespace,
			JobType:   int(job.JobType),
		}
		if status, ok := jobStatusMap[job.JobID]; ok {
			data.Items[i].Status = status
		} else {
			data.Items[i].Status = "UNKNOWN"
		}
	}
	return data, nil
}

func (svc *apiService) StopJob(ctx context.Context, param *protos.StopJobParam) (*protos.StopJobData, error) {
	// 1. check job existence
	// 2. check job status
	// 3a. if savepointing, call flink stop (and parse returned savepoint path)
	// 3b. if not, call flink cancel

	// TODO: remove hardcoded logic
	userID := protos.DefaultUserID
	namespace := protos.DefaultNamespace

	job, exist, err := svc.jobDao.Find(ctx, userID, param.JobID)
	if err != nil {
		log.Error("find job error", zap.Error(err))
		return nil, errors.ErrInternal(fmt.Sprintf("find job error: %v", err))
	}
	if !exist {
		return nil, errors.ErrNotFound(fmt.Sprintf("job %s not found", param.JobID))
	}

	kDeployment, err := svc.k8sClientSet.AppsV1().Deployments(namespace).Get(ctx, job.JobID, metav1.GetOptions{})
	if err != nil {
		return nil, errors.ErrNotFound(fmt.Sprintf("deployment %s not found", job.JobID))
	}
	// if the job is not flink, delete deployment and returen
	if !job.IsFlink() {
		err = svc.k8sClientSet.AppsV1().Deployments(namespace).Delete(ctx, kDeployment.Name, metav1.DeleteOptions{})
		if err != nil {
			return nil, err
		}
		return &protos.StopJobData{JobID: job.JobID, Legal: true}, nil
	}

	jobStatus, err := svc.getFlinkJobStatus(job.JMEndpoint)
	if jobStatus != "RUNNING" || err != nil {
		if param.Force {
			err = svc.k8sClientSet.AppsV1().Deployments(namespace).Delete(ctx, kDeployment.Name, metav1.DeleteOptions{})
			if err != nil {
				return nil, err
			}
			return &protos.StopJobData{JobID: job.JobID, Legal: true}, nil
		} else {
			return &protos.StopJobData{JobID: job.JobID, Legal: false, Info: fmt.Sprintf("Job status: %s\nError: %v", jobStatus, err)}, nil
		}
	}

	flinkJobIDs, err := svc.queryFlinkJobs(ctx, namespace, job.JobID)
	if err != nil {
		log.Error(err.Error())
		return nil, errors.ErrInternal(err.Error())
	}
	if len(flinkJobIDs) == 0 {
		return nil, errors.ErrNotFound("no flink jobs found")
	}
	log.Debug("StopJob", zap.Strings("flink jobs", flinkJobIDs))

	if param.Savepointing {
		savePointPaths := make([]string, 0)
		for _, flinkJobID := range flinkJobIDs {
			savePointPath, err := svc.stopFlinkJob(ctx, namespace, job.JobID, flinkJobID)
			if err != nil {
				log.Error(err.Error())
				return nil, errors.ErrInternal(err.Error())
			}
			savePointPaths = append(savePointPaths, savePointPath)
		}
		// REVIEW: if an application contains multiple jobs, only one could be restored from savepoint
		// Thus one-job per application should be GUARANTEED
		pathList := strings.Join(savePointPaths, ",")
		savepoint := &model.Savepoint{
			Path:   pathList,
			JobRef: job.ID,
		}
		err = svc.savepointDao.Insert(ctx, savepoint)
	} else {
		err = svc.cancelFlinkJob(ctx, namespace, job.JobID, flinkJobIDs[0])
	}
	if err != nil {
		log.Error(err.Error())
		return nil, errors.ErrInternal(err.Error())
	}

	return &protos.StopJobData{JobID: job.JobID, Legal: true}, nil
}

// ResumeJob
// REVIEW: Is listing & selecting savepoint needed?
func (svc *apiService) ResumeJob(ctx context.Context, param *protos.ResumeJobParam) (*protos.ResumeJobData, error) {
	// 1. check job existence
	// 2. check job status
	// 3. check job savepoint existence
	// 4a. if savepoints exist, set savepoint in property

	// TODO: remove hardcoded logic
	userID := protos.DefaultUserID
	namespace := protos.DefaultNamespace

	job, exist, err := svc.jobDao.Find(ctx, userID, param.JobID)
	if err != nil {
		log.Error("find job error", zap.Error(err))
		return nil, errors.ErrInternal(fmt.Sprintf("find job error: %v", err))
	}
	if !exist {
		return nil, errors.ErrNotFound(fmt.Sprintf("job %s not found", param.JobID))
	}

	exist, err = svc.ifFlinkDeploymentExist(ctx, job.JobID)
	if err != nil {
		log.Error("check flink deployment error", zap.Error(err))
		return nil, errors.ErrInternal(fmt.Sprintf("check flink deployment error: %v", err))
	}
	if exist {
		return nil, errors.ErrInvalidParameter("only resuming stopped job is allowed")
	}

	savepoints, err := svc.savepointDao.List(ctx, job.ID)
	if err != nil {
		log.Error("list savepoint error", zap.Error(err))
		return nil, errors.ErrInternal(fmt.Sprintf("list savepoint error: %v", err))
	}

	var info string
	property := make(map[string]string)
	if len(savepoints) != 0 && len(savepoints[0].Path) != 0 {
		property["execution.savepoint.path"] = strings.Split(savepoints[0].Path, ",")[0]
		property["execution.savepoint.ignore-unclaimed-state"] = "true"
	} else {
		info = "no savepoint found, resuming without savepoints"
	}

	if param.CheckpointOn {
		if param.CheckpointInterval != 0 {
			property["execution.checkpointing.interval"] = fmt.Sprintf("%d", param.CheckpointInterval)
		} else {
			property["execution.checkpointing.interval"] = fmt.Sprintf("%d", svc.flinkPersistenceConfig.DefaultCheckpointInterval)
		}
	}

	clusterID, jmEndpoint, err := svc.runApplication(ctx, namespace, job.JobID, job.InputKey, property)
	if err != nil {
		log.Error("run application error", zap.Error(err))
		return nil, errors.ErrInternal(fmt.Sprintf("run application error: %v", err))
	}

	job.JobID = clusterID
	job.JMEndpoint = jmEndpoint
	if err = svc.jobDao.Update(ctx, job); err != nil {
		return nil, errors.ErrInternal(fmt.Sprintf("update job error: %v", err))
	}

	return &protos.ResumeJobData{JobID: job.JobID, Info: info}, nil
}

func (svc *apiService) GetLogs(ctx context.Context, param *protos.GetLogsParam) (io.ReadCloser, error) {
	jmList, err := svc.k8sClientSet.CoreV1().Pods(protos.DefaultNamespace).List(ctx, metav1.ListOptions{
		LabelSelector: "component=jobmanager",
	})
	if err != nil {
		log.Error("list pods", zap.Error(err))
		return nil, err
	}

	// Sort the jm list by pod name length so that we can found the right pod by name prefix
	// For example, if we want to find the pod `foo-bar` in the list:
	//   - foo-bar-baz-suffix
	//   - foo-bar-suffix
	// We will get `foo-bar-baz-suffix` finnaly, which is the pod of `foo-bar-baz`, after sorting,
	// we can get the right one because `foo-bar-suffix` is before `foo-bar-baz-suffix`
	sort.Slice(jmList.Items, func(i, j int) bool {
		return len(jmList.Items[i].Name) < len(jmList.Items[j].Name)
	})
	for _, pod := range jmList.Items {
		if strings.HasPrefix(pod.Name, param.JobID) {
			opts := &corev1.PodLogOptions{
				Follow:       param.Follow,
				Timestamps:   param.Timestamps,
				Previous:     param.Previous,
				SinceSeconds: param.SinceSeconds,
				LimitBytes:   param.LimitBytes,
				TailLines:    param.TailLines,
			}
			if param.SinceTime != nil {
				opts.SinceTime = &metav1.Time{Time: *param.SinceTime}
			}
			request := svc.k8sClientSet.CoreV1().Pods(protos.DefaultNamespace).GetLogs(pod.Name, opts)
			stream, err := request.Stream(ctx)
			if err != nil {
				log.Error("get logs from pod", zap.String("pod", pod.Name), zap.Error(err))
				return nil, err
			}
			return stream, nil
		}
	}
	log.Error("pod of job not found", zap.String("job_id", param.JobID))
	return nil, errors.ErrNotFound("pod of target job not found")
}

func (svc *apiService) RmJob(ctx context.Context, param *protos.RmJobParam) (*protos.RmJobData, error) {
	job, exist, err := svc.jobDao.Find(ctx, protos.DefaultUserID, param.JobID)
	if err != nil {
		log.Error(err.Error())
		return nil, errors.ErrInternal(err.Error())
	}
	if !exist {
		return nil, errors.ErrNotFound("job not found")
	}

	deploymentsClient := svc.k8sClientSet.AppsV1().Deployments(protos.DefaultNamespace)
	deletePolicy := metav1.DeletePropagationForeground
	deployment, err := deploymentsClient.Get(ctx, param.JobID, metav1.GetOptions{})
	if err != nil {
		if !k8serr.IsNotFound(err) {
			log.Error("get deployment", zap.Error(err))
			return nil, errors.ErrInternal(err.Error())
		}
	} else if deployment != nil {
		if !param.Force && deployment.Status.AvailableReplicas >= 1 {
			return &protos.RmJobData{JobID: param.JobID, Legal: false}, nil
		}
		err = deploymentsClient.Delete(ctx, deployment.Name, metav1.DeleteOptions{
			PropagationPolicy: &deletePolicy,
		})
		if err != nil && !k8serr.IsNotFound(err) {
			log.Error("delete deployment", zap.Error(err))
			return nil, errors.ErrInternal(err.Error())
		}
	}
	// delete the secret only when the job is a flink job
	if job.IsFlink() {
		secretsClient := svc.k8sClientSet.CoreV1().Secrets(protos.DefaultNamespace)
		secret, err := secretsClient.Get(ctx, job.InputKey, metav1.GetOptions{})
		if err != nil {
			if !k8serr.IsNotFound(err) {
				log.Error("get secret", zap.Error(err))
				return nil, errors.ErrInternal(err.Error())
			}
		} else if secret != nil {
			err = secretsClient.Delete(ctx, secret.Name, metav1.DeleteOptions{})
			if err != nil && !k8serr.IsNotFound(err) {
				log.Error("delete secret", zap.Error(err))
				return nil, errors.ErrInternal(err.Error())
			}
		}
	}

	err = svc.jobDao.Delete(ctx, protos.DefaultUserID, param.JobID)
	if err != nil {
		log.Error("delete job", zap.Error(err))
		return nil, errors.ErrInternal(err.Error())
	}
	return &protos.RmJobData{JobID: param.JobID, Legal: true}, nil
}

func (svc *apiService) CreateIngress(ctx context.Context, param *protos.CreateIngressParam) (*protos.CreateIngressData, error) {
	var endpoint string
	var err error
	log.Debug(fmt.Sprintf("CreateIngress start, %v", param))
	switch param.Type {
	case "kafka":
		endpoint, err = svc.CreateKafkaTopic(KafkaTopicParam{TopicName: param.Name, NumPartitions: 1, ReplicationFactor: 1})
		if err != nil {
			return nil, err
		}
	case "sqs":
		endpoint, err = svc.CreateSqs(param.Name)
		if err != nil {
			return nil, err
		}
	}
	ingress := new(model.Ingress)
	ingress.UserID = protos.DefaultUserID
	ingress.Name = param.Name
	ingress.Endpoint = endpoint
	ingress.IngressType = param.Type

	if err := svc.ingressDao.Insert(ctx, ingress); err != nil {
		log.Error("insert ingress error", zap.Error(err))
		return nil, errors.ErrInternal(fmt.Sprintf("insert ingress %s failed", param.Name))
	}
	return &protos.CreateIngressData{Endpoint: endpoint}, nil
}

func (svc *apiService) RmIngress(ctx context.Context, param *protos.RmIngressParam) (*protos.RmIngressData, error) {
	log.Debug(fmt.Sprintf("RmIngress start, %v", param))
	isExist, err := svc.CheckExist(ctx, "ingress_id", param.IngressID)
	if err != nil {
		log.Error(fmt.Sprintf("check exist failed, %s", err))
		return nil, err
	}
	if isExist {
		return nil, fmt.Errorf("please rm related job first")
	}
	switch param.Type {
	case "kafka":
		err := svc.DeleteKafkaTopic(param.IngressID)
		if err != nil {
			return nil, err
		}
	case "sqs":
		ingress, isExisted, err := svc.ingressDao.Find(ctx, protos.DefaultUserID, param.IngressID)
		if err != nil {
			return nil, err
		}
		if !isExisted {
			return nil, errors.ErrNotFound("this ingress don't existed")
		}
		err = svc.DeleteSqs(ingress.Endpoint)
		if err != nil {
			return nil, err
		}
	}
	err = svc.ingressDao.Delete(ctx, protos.DefaultUserID, param.IngressID, param.Type)
	if err != nil {
		return nil, errors.ErrInternal("delete ingress failed")
	}
	return &protos.RmIngressData{IngressID: param.IngressID}, nil
}

func (svc *apiService) ListIngress(ctx context.Context) (*protos.ListIngressData, error) {
	// TODO: Refactor the hardcode userID.
	userID := protos.DefaultUserID
	// TODO: Refactor the hardcode Limit.
	cond := model.ListIngressCond{
		UserID: userID,
		Limit:  100,
	}
	ingress, err := svc.ingressDao.List(ctx, &cond)
	if err != nil {
		log.Error("list sink error", zap.Error(err))
		return nil, errors.ErrInternal("list sink failed, caused by dao connection")
	}
	data := &protos.ListIngressData{Items: []protos.ListIngressItem{}}
	for _, ins := range ingress {
		data.Items = append(data.Items, protos.ListIngressItem{
			UserID:    ins.IngressType,
			IngressID: ins.Name,
			Type:      ins.IngressType, // only tidb is supported now
			Endpoint:  ins.Endpoint,
		})
	}
	return data, nil
}

func (svc *apiService) createSecret(ctx context.Context, namespace string, name string, sql string) (*corev1.Secret, error) {
	sct := corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{},
		StringData: make(map[string]string),
	}
	sct.ObjectMeta.Namespace = namespace
	sct.ObjectMeta.Name = name
	sct.StringData["input.sql"] = sql
	ret, err := svc.k8sClientSet.CoreV1().Secrets(namespace).Create(ctx, &sct, metav1.CreateOptions{})
	if err == nil {
		return ret, nil
	} else if status, ok := err.(k8serr.APIStatus); ok && status.Status().Reason == metav1.StatusReasonAlreadyExists {
		return svc.k8sClientSet.CoreV1().Secrets(namespace).Get(ctx, name, metav1.GetOptions{})
	}
	return nil, err
}

func (svc *apiService) createConfigMap(ctx context.Context, namespace string, name string, sql string) (*corev1.ConfigMap, error) {
	cm := corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{}, Data: make(map[string]string)}
	cm.ObjectMeta.Namespace = namespace
	cm.ObjectMeta.Name = name
	cm.Data["sql"] = sql
	ret, err := svc.k8sClientSet.CoreV1().ConfigMaps(namespace).Create(ctx, &cm, metav1.CreateOptions{})
	if err == nil {
		return ret, nil
	} else if status, ok := err.(k8serr.APIStatus); ok && status.Status().Reason == metav1.StatusReasonAlreadyExists {
		return svc.k8sClientSet.CoreV1().ConfigMaps(namespace).Get(ctx, name, metav1.GetOptions{})
	}
	return nil, err
}

func (svc *apiService) createNameSpace(ctx context.Context, namespace string) error {
	ns := corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}
	if _, err := svc.k8sClientSet.CoreV1().Namespaces().Create(ctx, &ns, metav1.CreateOptions{}); err != nil {
		if status, ok := err.(k8serr.APIStatus); ok && status.Status().Reason == metav1.StatusReasonAlreadyExists {
			return nil
		}
		log.Error("create namespace error", zap.Error(err))
		return err
	}
	return nil
}

func (svc *apiService) setOwnerReference(ctx context.Context, namespace string, sct *corev1.Secret, deployment string) error {
	deploy, err := svc.k8sClientSet.AppsV1().Deployments(namespace).Get(ctx, deployment, metav1.GetOptions{})
	if err != nil {
		return err
	}
	// we should use patch instead of  update
	truVar := true
	falseVar := false
	ownerRef := metav1.OwnerReference{
		APIVersion:         "apps/v1",
		Kind:               "Deployment",
		Name:               deploy.GetName(),
		UID:                deploy.GetUID(),
		Controller:         &falseVar,
		BlockOwnerDeletion: &truVar,
	}

	patch := corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{OwnerReferences: []metav1.OwnerReference{ownerRef}},
	}
	bs, err := json.Marshal(patch)
	if err != nil {
		return err
	}
	_, err = svc.k8sClientSet.CoreV1().Secrets(namespace).Patch(ctx, sct.GetName(), types.StrategicMergePatchType, bs, metav1.PatchOptions{})
	return err
}

type jobList struct {
	Jobs []struct {
		ID     string `json:"id"`
		Status string `json:"status"`
	} `json:"jobs"`
}

func (svc *apiService) getFlinkJobStatus(jmEndpoint string) (string, error) {
	path := "jobs"
	// TODO: configurable connection timeout
	const ConnectionTimeout = 10 * time.Second
	ctx, cancel := context.WithTimeout(context.Background(), ConnectionTimeout)
	defer cancel()

	req, err := http.NewRequestWithContext(ctx, "GET", fmt.Sprintf("%s/%s", jmEndpoint, path), nil)
	if err != nil {
		log.Error(err.Error())
		return "", err
	}
	resp, err := http.DefaultClient.Do(req)
	httperr, _ := err.(*url.Error)
	if err != nil {
		if httperr.Timeout() {
			return "PENDING", nil
		}
		log.Error(err.Error())
		return "", err
	}
	defer resp.Body.Close()
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		log.Error(err.Error())
		return "", err
	}
	joblist := &jobList{}
	if err := json.Unmarshal(body, joblist); err != nil {
		log.Error(err.Error())
		return "", err
	}
	if len(joblist.Jobs) == 0 {
		log.Error("no job")
		return "", fmt.Errorf("no job")
	}
	return joblist.Jobs[0].Status, nil
}

func (svc *apiService) ifFlinkDeploymentExist(ctx context.Context, jobID string) (bool, error) {
	deploymentsClient := svc.k8sClientSet.AppsV1().Deployments(protos.DefaultNamespace)
	_, err := deploymentsClient.Get(ctx, jobID, metav1.GetOptions{})
	if err == nil {
		return true, nil
	}
	if k8serr.IsNotFound(err) {
		return false, nil
	}
	return false, err
}

func (svc *apiService) ifFlinkDeploymentAvaliable(ctx context.Context, jobID string) (bool, error) {
	exist, err := svc.ifFlinkDeploymentExist(ctx, jobID)
	if err != nil {
		return false, err
	}
	if !exist {
		return false, nil
	}

	deploymentsClient := svc.k8sClientSet.AppsV1().Deployments(protos.DefaultNamespace)
	result, err := deploymentsClient.Get(ctx, jobID, metav1.GetOptions{})
	if err != nil {
		return false, err
	}
	if result.Status.AvailableReplicas < 1 {
		return false, nil
	}
	return true, nil
}

func (svc *apiService) getFlinkRestIP(ctx context.Context, kService corev1.Service) (string, error) {
	switch svc.flinkRestExposedType {
	case "ClusterIP":
		return kService.Spec.ClusterIP, nil
	case "NodePort":
		addressList, err := svc.getKubernetesNodeIPs(ctx)
		if err != nil {
			return "", err
		}
		if len(addressList) == 0 {
			return "", errors.ErrNotFound("kubernetes node not found")
		}
		return addressList[0], nil
	case "LoadBalancer":
		addressList := svc.getEKSLoadBalancerIngress(kService)
		if len(addressList) == 0 {
			return "", errors.ErrNotFound("loadbalancer hostname not found")
		}
		return addressList[0], nil
	}
	return "", nil
}

func (svc *apiService) getFlinkRestPort(name string, ports []corev1.ServicePort) (int32, error) {
	for _, port := range ports {
		if port.Name == name {
			return port.Port, nil
		}
	}
	log.Error("not rest port")
	return 0, errors.ErrNotFound("flink port not found")
}

func (svc *apiService) getKubernetesNodeIPs(ctx context.Context) (addressList []string, err error) {
	nodeList, err := svc.k8sClientSet.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
	if err != nil {
		return
	}
	for _, node := range nodeList.Items {
		externalIP, internalIP := "", ""
		for _, addr := range node.Status.Addresses {
			switch addr.Type {
			case "ExternalIP":
				externalIP = addr.Address
			case "InternalIP":
				internalIP = addr.Address
			}
		}
		if externalIP != "" {
			addressList = append(addressList, externalIP)
		} else if internalIP != "" {
			addressList = append(addressList, internalIP)
		}
	}
	return
}

func (svc *apiService) getEKSLoadBalancerIngress(kService corev1.Service) []string {
	var addressList []string
	ingressList := kService.Status.LoadBalancer.Ingress
	for _, ingress := range ingressList {
		addressList = append(addressList, ingress.Hostname)
	}
	return addressList
}

func (svc *apiService) SubmitJobV2(ctx context.Context, param *protos.SubmitJobParamV2) (*protos.SubmitJobData, error) {
	_, isExisted, err := svc.jobDao.Find(ctx, protos.DefaultUserID, param.ClusterID)
	if err != nil {
		log.Error(fmt.Sprintf("Fetch job failed, %s", err))
		return nil, err
	}
	if isExisted {
		return nil, fmt.Errorf("job have existed")
	}

	ingress, isExisted, err := svc.ingressDao.Find(ctx, protos.DefaultUserID, param.IngressName)
	if err != nil {
		log.Error(fmt.Sprintf("Fetch ingress failed, %s", err))
		return nil, err
	}
	if !isExisted {
		return nil, errors.ErrNotFound("ingress dont exist")
	}
	sink, isExisted, err := svc.sinkDao.Find(ctx, protos.DefaultUserID, param.SinkName)
	if err != nil {
		log.Error(fmt.Sprintf("Fetch sink failed, %s", err))
		return nil, err
	}
	if !isExisted {
		return nil, errors.ErrNotFound("sink dont exist")
	}
	var topic string
	if ingress.IngressType == "kafka" {
		temp := strings.Split(ingress.Endpoint, "/")
		if len(temp) != 2 {
			log.Error(fmt.Sprintf("empty broker list %s", ingress.Endpoint))
			return nil, errors.ErrInternal("empty broker list")
		}
		topic = temp[1]
		ingress.Endpoint = temp[0]
	}

	Job := JobCore{
		Id:           param.ClusterID,
		Dsn:          svc.getBizDSN(sink.Database),
		Ingress:      *ingress,
		Sink:         *sink,
		K8sClientSet: svc.k8sClientSet,
		Mapper:       svc.defaultmapper,
		KafkaConf:    svc.Kafkaconf,
		Topic:        topic,
	}
	if err := Job.Execute(); err != nil {
		log.Error(err.Error())
		return nil, errors.ErrInternal("execute job failed")
	}
	if err := Job.Land(svc.jobDao); err != nil {
		log.Error(err.Error())
	}

	return &protos.SubmitJobData{ClusterID: Job.Id}, nil
}

func (svc *apiService) getBizDSN(db string) string {
	return fmt.Sprintf("%s:%s@(%s:%s)/%s?%s", svc.bizDBUser, svc.bizDBPassword, svc.bizDBHost, svc.bizDBPort, db, "charset=utf8mb4&parseTime=True&loc=UTC")
}

func (svc *apiService) CheckExist(ctx context.Context, fieldname, fieldvalue string) (bool, error) {
	_, isExist, err := svc.jobDao.Select(ctx, &fieldname, &fieldvalue)
	if err != nil {
		return false, err
	}
	return isExist, nil
}