/*
 * 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 event

import (
	"context"
	"fmt"
	"hash/fnv"

	"github.com/aws/aws-sdk-go/aws"

	"go.uber.org/zap"

	"github.com/pingcap/log"
	"k8s.io/api/batch/v1beta1"
	v13 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/types"

	v1 "k8s.io/api/batch/v1"
	v12 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	"github.com/pingcap/mimir/pkg/protos"

	k8serr "k8s.io/apimachinery/pkg/api/errors"
)

type TableCheckerSubscriber struct {
	Cli               client.Client
	DbSecretName      string
	DbSecretNamespace string
	Image             string
}

func (tcs *TableCheckerSubscriber) SubscribeAddSink(event AddSinkEvent) error {
	if event.TableName == "" || event.Schedule == "" {
		log.Info("empty tableName or schedule", zap.String("tableName", event.TableName), zap.String("schedule", event.Schedule))
		return nil
	}

	return tcs.CreateOrUpdateJob(event.Database, event.TableName, event.Schedule)
}

func (tcs *TableCheckerSubscriber) SubscribeUpdateSink(event UpdateSinkEvent) error {
	if event.TableName == "" || event.Schedule == "" {
		log.Info("empty tableName or schedule", zap.String("tableName", event.TableName), zap.String("schedule", event.Schedule))
		return nil
	}

	return tcs.CreateOrUpdateJob(event.Database, event.TableName, event.Schedule)
}

func (tcs *TableCheckerSubscriber) SubscribeDelete(event DeleteSinkEvent) error {
	if event.TableName == "" {
		log.Info("empty tableName", zap.String("tableName", event.TableName))
		return nil
	}

	job := v1.CronJob{}

	err := tcs.Cli.Get(context.TODO(), types.NamespacedName{
		Namespace: protos.DefaultNamespace,
		Name:      jobName(event.Database, event.TableName),
	}, &job)
	if err != nil {
		if k8serr.IsNotFound(err) {
			return nil
		}
		return err
	}

	return tcs.Cli.Delete(context.TODO(), &job)
}

func (tcs *TableCheckerSubscriber) getDBDSN(db string) (string, error) {
	var secret v13.Secret

	err := tcs.Cli.Get(context.TODO(), types.NamespacedName{
		Namespace: tcs.DbSecretNamespace,
		Name:      tcs.DbSecretName,
	}, &secret)
	if err != nil {
		return "", nil
	}

	return fmt.Sprintf("%s:%s@(%s:%s)/%s?%s", string(secret.Data["user"]), string(secret.Data["password"]), string(secret.Data["host"]), string(secret.Data["port"]), db, "charset=utf8mb4&parseTime=True&loc=UTC"), nil
}

func (tcs *TableCheckerSubscriber) CreateOrUpdateJob(dbName string, tableName string, schedule string) error {
	dsn, err := tcs.getDBDSN(dbName)
	if err != nil {
		return err
	}

	job := &v1beta1.CronJob{
		ObjectMeta: v12.ObjectMeta{
			Name:      jobName(dbName, tableName),
			Namespace: protos.DefaultNamespace,
		},
		Spec: v1beta1.CronJobSpec{
			Schedule: schedule,
			JobTemplate: v1beta1.JobTemplateSpec{
				Spec: v1.JobSpec{
					BackoffLimit: aws.Int32(1),
					Template: v13.PodTemplateSpec{
						Spec: v13.PodSpec{
							RestartPolicy: "Never",
							Containers: []v13.Container{
								{
									Name:  "checker",
									Image: tcs.Image,
									Command: []string{
										"/checker",
										"--push-gateway-url",
										"pushgateway:9091",
										"--table",
										tableName,
										"--dsn",
										dsn,
									},
								},
							},
						},
					},
				},
			},
		},
	}

	_, err = controllerutil.CreateOrUpdate(context.TODO(), tcs.Cli, job, func() error {
		job.Spec.Schedule = schedule

		return nil
	})

	return err
}

func jobName(dbName string, tableName string) string {
	return fnvHash(fnvHash(fmt.Sprintf("%s-%s", dbName, tableName)))
}

func fnvHash(raw string) string {
	hash := fnv.New32()
	if _, err := hash.Write([]byte(raw)); err != nil {
		// impossible path
		log.Error("failed to write fnv hash", zap.Error(err))
	}
	return fmt.Sprintf("%x", hash.Sum32())
}