package service

import (
	"context"
	"fmt"

	"github.com/pingcap/log"
	appsv1 "k8s.io/api/apps/v1"
	apiv1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	k8s "k8s.io/client-go/kubernetes"

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

type JobCore struct {
	Id           string
	Dsn          string
	Sink         model.Sink
	Ingress      model.Ingress // topic
	K8sClientSet k8s.Interface
	Mapper       config.Mapper
	KafkaConf    config.KafkaConf
	Topic        string
}

// init a default mapper pod
func (j *JobCore) Execute() error {
	deploymentsClient := j.K8sClientSet.AppsV1().Deployments(protos.DefaultNamespace)
	deployment := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name: j.Id,
		},
		Spec: appsv1.DeploymentSpec{
			Replicas: int32Ptr(1),
			Selector: &metav1.LabelSelector{
				MatchLabels: map[string]string{
					"app": "default-mapper",
				},
			},
			Template: apiv1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						"app": "default-mapper",
					},
				},
				Spec: apiv1.PodSpec{
					Containers: []apiv1.Container{
						{
							Name:            "default-mapper",
							Image:           j.Mapper.Image,
							ImagePullPolicy: apiv1.PullIfNotPresent,
							Command: []string{
								"/default_mapper",
								"--dsn",
								j.Dsn,
								"--source-type",
								j.Ingress.IngressType,
								"--sqs-url",
								j.Ingress.Endpoint,
								"--table",
								j.Sink.Table,
								"--broker",
								j.Ingress.Endpoint,
								"--topic",
								j.Topic,
								"--group-id",
								protos.DefaultGroupID,
								"--username",
								j.KafkaConf.User,
								"--password",
								j.KafkaConf.Passwd,
							},
						},
					},
				},
			},
		},
	}
	_, err := deploymentsClient.Create(context.TODO(), deployment, metav1.CreateOptions{})
	if err != nil {
		log.Error(err.Error())
		return fmt.Errorf("create Job Pod failed, %s", err)

	}

	return nil
}

func (j *JobCore) Land(dao model.JobDao) error {
	ctx := context.Background()
	job := &model.Job{
		UserID:    protos.DefaultUserID,
		JobID:     j.Id,
		IngressId: j.Ingress.Name,
		SinkId:    j.Sink.SinkID,
		Namespace: protos.DefaultNamespace,
	}
	if err := dao.Insert(ctx, job); err != nil {
		log.Error(fmt.Sprintf("Insert Job failed, %s", err))
		return err
	}
	return nil
}

func int32Ptr(i int32) *int32 { return &i }