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

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"time"

	"gorm.io/gorm/clause"

	"gorm.io/gorm"
)

type JobType int
type InputType int

const (
	JobTableName = "job"
)
const (
	JobTypeFlinkStreaming JobType = 1
	JobTypeFlinkBatch     JobType = 2

	InputTypeSQL InputType = 1
	InputTypeJar InputType = 2
)

// type Job struct {
// 	ID         uint      `gorm:"column:id;type:int(11) unsigned NOT NULL AUTO_INCREMENT;PRIMARY_KEY"`
// 	UserID     string    `gorm:"column:user_id;type:varchar(64);uniqueIndex:job_user_id_job_id_index,priority:2;NOT NULL;DEFAULT:''"`
// 	JobID      string    `gorm:"column:job_id;type:varchar(64);uniqueIndex:job_user_id_job_id_index,priority:1;NOT NULL;DEFAULT:''"`
// 	JMEndpoint string    `gorm:"column:jm_endpoint;type:varchar(128);NOT NULL;DEFAULT:''"`
// 	Namespace  string    `gorm:"column:namespace;type:varchar(64);NOT NULL;DEFAULT:''"`
// 	JobType    JobType   `gorm:"column:job_type;type:tinyint(4);NOT NULL;DEFAULT:0"`
// 	InputType  InputType `gorm:"column:input_type;type:tinyint(4);NOT NULL;DEFAULT:0"`
// 	InputKey   string    `gorm:"column:input_key;type:varchar(128);NOT NULL;DEFAULT:''"`
// 	// DeletedAt  gorm.DeletedAt `gorm:"column:deleted_at"`

// 	// Disallow gorm to set the `created_at`
// 	CreatedAt time.Time `gorm:"column:created_at;<-:false;type:timestamp;NOT NULL;DEFAULT:CURRENT_TIMESTAMP"`
// 	UpdatedAt time.Time `gorm:"column:updated_at;<-:false;type:timestamp;NOT NULL;DEFAULT:CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"`
// }

type Job struct {
	ID         uint      `gorm:"column:id;type:int(11) unsigned NOT NULL AUTO_INCREMENT;PRIMARY_KEY"`
	UserID     string    `gorm:"column:user_id;type:varchar(64);uniqueIndex:job_user_id_job_id_index,priority:2;NOT NULL;DEFAULT:''"`
	JobID      string    `gorm:"column:job_id;type:varchar(64);uniqueIndex:job_user_id_job_id_index,priority:1;NOT NULL;DEFAULT:''"`
	JMEndpoint string    `gorm:"column:jm_endpoint;type:varchar(128);NOT NULL;DEFAULT:''"`
	IngressId  string    `gorm:"column:ingress_id;type:varchar(64);NOT NULL;DEFAULT:''"`
	SinkId     string    `gorm:"column:sink_id;type:varchar(64);NOT NULL;DEFAULT:''"`
	Namespace  string    `gorm:"column:namespace;type:varchar(64);NOT NULL;DEFAULT:''"`
	JobType    JobType   `gorm:"column:job_type;type:tinyint(4);NOT NULL;DEFAULT:0"`
	InputType  InputType `gorm:"column:input_type;type:tinyint(4);NOT NULL;DEFAULT:0"`
	InputKey   string    `gorm:"column:input_key;type:varchar(128);NOT NULL;DEFAULT:''"`
	// DeletedAt  gorm.DeletedAt `gorm:"column:deleted_at"`

	// Disallow gorm to set the `created_at`
	CreatedAt time.Time `gorm:"column:created_at;<-:false;type:timestamp;NOT NULL;DEFAULT:CURRENT_TIMESTAMP"`
	UpdatedAt time.Time `gorm:"column:updated_at;<-:false;type:timestamp;NOT NULL;DEFAULT:CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"`
}

func (j *Job) TableName() string {
	return JobTableName
}

func (j *Job) IsFlink() bool {
	return j.JobType == JobTypeFlinkBatch || j.JobType == JobTypeFlinkStreaming
}

type JobDao interface {
	Insert(ctx context.Context, job *Job) error
	List(ctx context.Context, cond *ListJobCond) ([]Job, error)
	Find(ctx context.Context, userID string, jobID string) (*Job, bool, error)
	Select(ctx context.Context, fields ...interface{}) (*Job, bool, error)
	Delete(ctx context.Context, userID string, jobID string) error
	Update(ctx context.Context, job *Job) error
}

type jobDao struct {
	db *gorm.DB
}

func NewJobDao(options ...DaoOption) (JobDao, error) {
	dao := new(jobDao)
	for _, option := range options {
		if err := option(dao); err != nil {
			return nil, err
		}
	}
	return dao, nil
}

func (dao *jobDao) SetDB(db *gorm.DB) {
	dao.db = db
}

func (dao *jobDao) Insert(ctx context.Context, job *Job) error {
	return dao.db.WithContext(ctx).
		Table(job.TableName()).
		Clauses(clause.OnConflict{
			Columns: []clause.Column{
				{Name: "user_id"},
				{Name: "job_id"},
			},
			UpdateAll: true,
		}).Create(job).Error
}

func (dao *jobDao) Delete(ctx context.Context, userID string, jobID string) error {
	if err := dao.db.WithContext(ctx).Table(JobTableName).Where(
		"user_id = ? && job_id = ?", userID, jobID).Delete(&Job{}).Error; err != nil {
		return err
	}
	return nil
}

type ListJobCond struct {
	UserID       string
	ReverseOrder bool
	Limit        int
}

func (dao *jobDao) List(ctx context.Context, cond *ListJobCond) ([]Job, error) {
	q := dao.db.WithContext(ctx).Table(JobTableName)
	if len(cond.UserID) != 0 {
		q = q.Where("user_id = ?", cond.UserID)
	}
	if cond.ReverseOrder {
		q = q.Order("created_at DESC")
	}
	if cond.Limit != 0 {
		q.Limit(cond.Limit)
	}
	rows, err := q.Rows()
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	data := make([]Job, 0)
	for rows.Next() {
		job := Job{}
		if err := dao.db.ScanRows(rows, &job); err != nil {
			return nil, err
		}
		data = append(data, job)
	}
	return data, nil
}

func (dao *jobDao) Find(ctx context.Context, userID string, jobID string) (*Job, bool, error) {
	job := new(Job)
	if err := dao.db.WithContext(ctx).Table(job.TableName()).Where("user_id = ? && job_id = ?", userID, jobID).First(job).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, false, nil
		}
		return nil, false, err
	}
	return job, true, nil
}

func (dao *jobDao) Select(ctx context.Context, fields ...interface{}) (*Job, bool, error) {
	whereClause := make([]string, 0)
	values := make([]interface{}, 0)
	var flag bool = true
	for _, field := range fields {
		if flag {
			refvalue := reflect.ValueOf(field).Elem()
			whereClause = append(whereClause, fmt.Sprintf("%s = ?", refvalue))
		} else {
			values = append(values, field)
		}
		flag = !flag
	}
	job := new(Job)
	if err := dao.db.WithContext(ctx).Table(job.TableName()).Where(strings.Join(whereClause, " && "), values...).First(job).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, false, nil
		}
		return nil, false, err
	}
	return job, true, nil
}

func (dao *jobDao) Update(ctx context.Context, job *Job) error {
	return dao.db.WithContext(ctx).Save(job).Error
}