// Copyright 2022 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 mysql

import (
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"time"

	"github.com/go-gormigrate/gormigrate/v2"
	"github.com/pingcap/log"
	"go.uber.org/zap"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"

	"github.com/pingcap-inc/compute-engine/pkg/database/model"
)

// Migrate do migration with versioning style instead of auto style.
func Migrate(db *gorm.DB) error {
	m := gormigrate.New(db, gormigrate.DefaultOptions, []*gormigrate.Migration{})
	m.InitSchema(initSchema)
	return m.Migrate()
}

const (
	TaskTableName = "task_instance"
	TagsTableName = "task_tags"
	ArgsTableName = "task_args"
	DataTableName = "task_data"
)

type Task struct {
	TaskID    string     `gorm:"column:id;type:varchar(64);PRIMARY_KEY"`
	TaskName  string     `gorm:"column:task_name;type:varchar(64);NOT NULL;DEFAULT:'';index:search_idx"`
	Type      int        `gorm:"column:task_type;type:tinyint(4);NOT NULL;DEFAULT:0"`
	Status    int        `gorm:"column:status;type:tinyint(4);NOT NULL;DEFAULT:0;index:search_idx"`
	Hook      string     `gorm:"column:hook;type:varchar(256);DEFAULT:''"`
	Tags      []TaskTag  `gorm:"foreignKey:TaskID;references:TaskID"`
	Args      []TaskArg  `gorm:"foreignKey:TaskID;references:TaskID"`
	Data      []TaskData `gorm:"foreignKey:TaskID;references:TaskID"`
	CreatedAt time.Time  `gorm:"column:created_at;<-:false;type:timestamp;NOT NULL;DEFAULT:CURRENT_TIMESTAMP;index:search_idx,sort:desc"`
	UpdatedAt time.Time  `gorm:"column:updated_at;<-:false;type:timestamp;NOT NULL;DEFAULT:CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"`
}

func (task *Task) Instance() (*model.TaskInstance, error) {
	tags := map[string]interface{}{}
	for _, t := range task.Tags {
		var v interface{}
		if err := json.Unmarshal([]byte(t.Value), &v); err != nil {
			return nil, err
		}
		tags[t.Key] = v
	}

	args := map[string]interface{}{}
	for _, t := range task.Args {
		var v interface{}
		if err := json.Unmarshal([]byte(t.Value), &v); err != nil {
			return nil, err
		}
		args[t.Key] = v
	}

	data := map[string]interface{}{}
	for _, d := range task.Data {
		var v interface{}
		if err := json.Unmarshal([]byte(d.Value), &v); err != nil {
			return nil, err
		}
		data[d.Key] = v
	}

	return &model.TaskInstance{
		ID:        task.TaskID,
		Task:      task.TaskName,
		Status:    model.TaskStatus(task.Status),
		Hook:      task.Hook,
		Tags:      tags,
		Args:      args,
		Data:      data,
		CreatedAt: model.JSONTime(task.CreatedAt),
		UpdatedAt: model.JSONTime(task.UpdatedAt),
	}, nil
}

type TaskTag struct {
	gorm.Model
	TaskID string `gorm:"column:task_id;type:varchar(64);index:search_idx"`
	Key    string `gorm:"column:key;type:varchar(64);index:search_idx"`
	Value  string `gorm:"column:value;type:text"`
}

type TaskArg struct {
	gorm.Model
	TaskID string `gorm:"column:task_id;type:varchar(64);index:search_idx"`
	Key    string `gorm:"column:key;type:varchar(64);index:search_idx"`
	Value  string `gorm:"column:value;type:text"`
}

type TaskData struct {
	gorm.Model
	TaskID string `gorm:"column:task_id;type:varchar(64);index:search_idx"`
	Key    string `gorm:"column:key;type:varchar(64);index:search_idx"`
	Value  string `gorm:"column:value;type:text"`
}

func (task *Task) TableName() string {
	return TaskTableName
}

func (task *TaskTag) TableName() string {
	return TagsTableName
}

func (task *TaskArg) TableName() string {
	return ArgsTableName
}

func (task *TaskData) TableName() string {
	return DataTableName
}

func initSchema(db *gorm.DB) error {
	if err := db.Set("gorm:table_options", "ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin").AutoMigrate(&Task{}); err != nil {
		log.Error("migrate task table error", zap.Error(err))
		return err
	}
	if err := db.Set("gorm:table_options", "ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin").AutoMigrate(&TaskTag{}); err != nil {
		log.Error("migrate task_tags table error", zap.Error(err))
		return err
	}
	if err := db.Set("gorm:table_options", "ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin").AutoMigrate(&TaskArg{}); err != nil {
		log.Error("migrate task_args table error", zap.Error(err))
		return err
	}
	if err := db.Set("gorm:table_options", "ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin").AutoMigrate(&TaskData{}); err != nil {
		log.Error("migrate task_data table error", zap.Error(err))
		return err
	}
	return nil
}

type TaskDao struct {
	db *gorm.DB
}

func NewModel() (model.TaskModel, error) {
	dsn := os.Getenv("MYSQL_DSN")
	if dsn == "" {
		log.Error("MYSQL_DSN is empty")
		err := errors.New("env invalid, no MYSQL_DSN")
		return nil, err
	}
	log.Info("connecting database...")
	db, err := Dial(&MysqlConfig{DSN: dsn})
	if err != nil {
		log.Error("init model error", zap.Error(err))
		return nil, err
	}
	log.Info("connect database success")

	return &TaskDao{
		db: db,
	}, nil
}

func (dao *TaskDao) PutInstance(instance *model.TaskInstance) error {
	taglist := []TaskTag{}
	for k, v := range instance.Tags {
		bytes, err := json.Marshal(v)
		if err != nil {
			return err
		}
		tag := TaskTag{
			TaskID: instance.ID,
			Key:    k,
			Value:  string(bytes),
		}
		taglist = append(taglist, tag)
	}
	arglist := []TaskArg{}
	for k, v := range instance.Args {
		bytes, err := json.Marshal(v)
		if err != nil {
			return err
		}
		arg := TaskArg{
			TaskID: instance.ID,
			Key:    k,
			Value:  string(bytes),
		}
		arglist = append(arglist, arg)
	}
	datalist := []TaskData{}
	for k, v := range instance.Data {
		log.Debug("marshaling data", zap.String("id", instance.ID), zap.String("key", k))
		bytes, err := json.Marshal(v)
		if err != nil {
			return err
		}
		data := TaskData{
			TaskID: instance.ID,
			Key:    k,
			Value:  string(bytes),
		}
		datalist = append(datalist, data)
	}

	task := &Task{
		TaskID:   instance.ID,
		Type:     int(model.Lambda),
		TaskName: instance.Task,
		Status:   int(instance.Status),
		Hook:     instance.Hook,
		Tags:     taglist,
		Args:     arglist,
		Data:     datalist,
	}
	log.Debug("updating tag list", zap.Int("length", len(taglist)))
	if err := dao.db.Model(task).Association("Tags").Replace(taglist); err != nil {
		return err
	}
	log.Debug("updating arg list", zap.Int("length", len(arglist)))
	if err := dao.db.Model(task).Association("Args").Replace(arglist); err != nil {
		return err
	}
	log.Debug("updating data list", zap.Int("length", len(datalist)))
	if err := dao.db.Model(task).Association("Data").Replace(datalist); err != nil {
		return err
	}
	return dao.db.Clauses(clause.OnConflict{
		UpdateAll: true,
	}).Create(task).Error
}

func (dao *TaskDao) GetInstanceList(ids []string) ([]model.TaskInstance, error) {
	tasks := []Task{}
	if err := dao.db.Order("created_at").Preload("Tags").Preload("Args").Preload("Data").Find(&tasks, "id in ?", ids).Error; err != nil {
		return nil, err
	}

	instances := []model.TaskInstance{}
	for _, task := range tasks {
		instance, err := task.Instance()
		if err != nil {
			return nil, err
		}
		instances = append(instances, *instance)
	}
	return instances, nil
}

func (dao *TaskDao) SearchInstance(
	taskName string,
	status int,
	begin *time.Time,
	end *time.Time,
	tags map[string]interface{},
	args map[string]interface{},
	data map[string]interface{},
) ([]string, error) {
	db := dao.db.Model(&Task{})
	for k, v := range tags {
		vv, err := json.Marshal(v)
		if err != nil {
			return nil, err
		}
		alias := fmt.Sprintf("%s_%s", TagsTableName, k)
		db = db.Joins(fmt.Sprintf("join %s as %s on %s.task_id = %s.id", TagsTableName, alias, alias, TaskTableName))
		db = db.Where(fmt.Sprintf("%s.key = ? and %s.value = ?", alias, alias), k, vv)
	}
	for k, v := range args {
		vv, err := json.Marshal(v)
		if err != nil {
			return nil, err
		}
		alias := fmt.Sprintf("%s_%s", ArgsTableName, k)
		db = db.Joins(fmt.Sprintf("join %s as %s on %s.task_id = %s.id", ArgsTableName, alias, alias, TaskTableName))
		db = db.Where(fmt.Sprintf("%s.key = ? and %s.value = ?", alias, alias), k, vv)
	}
	for k, v := range data {
		vv, err := json.Marshal(v)
		if err != nil {
			return nil, err
		}
		alias := fmt.Sprintf("%s_%s", DataTableName, k)
		db = db.Joins(fmt.Sprintf("join %s as %s on %s.task_id = %s.id", DataTableName, alias, alias, TaskTableName))
		db = db.Where(fmt.Sprintf("%s.key = ? and %s.value = ?", alias, alias), k, vv)
	}
	db = db.Where(fmt.Sprintf("%s.task_name = ? and %s.status = ?", TaskTableName, TaskTableName), taskName, status)
	if begin != nil {
		db = db.Where(fmt.Sprintf("%s.created_at >= ?", TaskTableName), begin)
	}
	if end != nil {
		db = db.Where(fmt.Sprintf("%s.created_at < ?", TaskTableName), end)
	}
	rows, err := db.Select(fmt.Sprintf("%s.id", TaskTableName)).Rows()
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var ids []string
	for rows.Next() {
		id := ""
		if err := rows.Scan(&id); err != nil {
			return nil, err
		}
		ids = append(ids, id)
	}
	return ids, nil
}

func (dao *TaskDao) Close() error {
	db, err := dao.db.DB()
	if err != nil {
		log.Error("get gorm db instance error", zap.Error(err))
		return err
	}
	if err := db.Close(); err != nil {
		log.Error("closing mysql error", zap.Error(err))
		return err
	}
	return nil
}