package dbutil

import (
	"strconv"
	"time"

	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	"gorm.io/gorm"
)

func GetImportByID(tx *gorm.DB, clusterID, id entity.ID) (*entity.Import, error) {
	var imp entity.Import
	err := entity.NewImportQuerySet(tx).IDEq(id).ClusterIDEq(clusterID).One(&imp)
	if err != nil {
		return nil, TagNotFound(err)
	}
	return &imp, nil
}

func ListImport(tx *gorm.DB, clusterID entity.ID, pager utils.Pager) ([]entity.Import, error) {
	var imports []entity.Import
	err := entity.NewImportQuerySet(tx).
		ClusterIDEq(clusterID).
		CreatedByDMEq(false).
		OrderDescByCreatedAt().
		Offset(int(pager.Offset)).
		Limit(int(pager.PageSize)).
		All(&imports)
	if err != nil {
		return nil, err
	}
	return imports, nil
}

func ListImportByStatuses(tx *gorm.DB, clusterID entity.ID, pager utils.Pager, statuses []entity.ImportStatus) ([]entity.Import, error) {
	var imports []entity.Import
	err := entity.NewImportQuerySet(tx).
		ClusterIDEq(clusterID).
		CreatedByDMEq(false).
		StatusIn(statuses...).
		OrderDescByCreatedAt().
		Offset(int(pager.Offset)).
		Limit(int(pager.PageSize)).
		All(&imports)
	if err != nil {
		return nil, err
	}
	return imports, nil
}

// select a.* from imports as a inner join (select name,max(created_at) as 'max_created_at' from imports where created_by_dm=true group by name) b on a.name=b.name and a.created_at = b.max_created_at order by id desc limit 100;
func ListImportCreatedByDM(tx *gorm.DB, clusterID entity.ID, pager utils.Pager, statuses []entity.ImportStatus) ([]entity.Import, error) {
	var imports []entity.Import
	statusClause := ""
	if len(statuses) > 0 {
		// FIXME: this interface is not verified
		statusClause = " WHERE status IN (?)"
	}
	joinSQL := "INNER JOIN (SELECT name AS 'max_name',max(created_at) AS 'max_created_at' FROM imports WHERE cluster_id=" +
		strconv.FormatUint(clusterID.Uint64(), 10) + " and created_by_dm=true" + statusClause +
		" GROUP BY max_name) b ON a.name=b.max_name and a.created_at=b.max_created_at"
	var sql *gorm.DB
	if len(statuses) > 0 {
		statusesArgs := make([]interface{}, len(statuses))
		for i := range statuses {
			statusesArgs[i] = statuses[i]
		}
		sql = tx.Table("imports a").
			InnerJoins(joinSQL, statusesArgs...)
	} else {
		sql = tx.Table("imports a").
			InnerJoins(joinSQL)
	}
	err := sql.Order("a.id DESC").
		Offset(int(pager.Offset)).
		Limit(int(pager.PageSize)).
		Find(&imports).Error
	return imports, err
}

func ListActiveImportsCreatedAfter(tx *gorm.DB, createdAfter time.Time) ([]entity.Import, error) {
	var imports []entity.Import
	err := entity.NewImportQuerySet(tx).
		CleanUpEq(false).
		CreatedAtGte(createdAfter).
		All(&imports)
	if err != nil {
		return nil, err
	}
	return imports, nil
}

func ListActiveImports(tx *gorm.DB) ([]entity.Import, error) {
	var imports []entity.Import
	err := entity.NewImportQuerySet(tx).
		CleanUpEq(false).
		All(&imports)
	if err != nil {
		return nil, err
	}
	return imports, nil
}

func ListFailedImportsFinishedAfter(tx *gorm.DB, finishedAfter time.Time) ([]entity.Import, error) {
	var imports []entity.Import
	err := entity.NewImportQuerySet(tx).
		StatusEq(entity.ImportStatusFailed).
		FinishedAtGt(finishedAfter).
		All(&imports)
	if err != nil {
		return nil, err
	}
	return imports, nil
}

func CountImport(tx *gorm.DB, clusterID entity.ID) (int64, error) {
	return entity.NewImportQuerySet(tx).ClusterIDEq(clusterID).Count()
}

func CountRunningImports(tx *gorm.DB, clusterID entity.ID) (int64, error) {
	statuses := []entity.ImportStatus{entity.ImportStatusPreparing, entity.ImportStatusImporting, entity.ImportStatusCanceling}
	return entity.NewImportQuerySet(tx).ClusterIDEq(clusterID).StatusIn(statuses...).Count()
}