// Package repository add by blacknc <blacknc@163.com> at 2023/02/20 19:38:11
package repository

import (
	"context"
	"dataservices/commons/database"
	"dataservices/commons/errs"
	"dataservices/commons/utils"
	"dataservices/internal/entity"
	"fmt"
	"time"

	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	tierrs "github.com/tidbcloud/tiinf/errs"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

func ListEndpointsByAppIDs(appIds []uint64) ([]*entity.EndpointV1, error) {
	var res []*entity.EndpointV1
	err := database.DBConn.DB.Table(entity.TableEndpoints).Joins("left join (SELECT endpoint_id as label FROM endpoint_publish ) t1 ON endpoints.id = t1.label").Where("is_deleted=0").Clauses(clause.IN{
		Column: "dataapp_id",
		Values: utils.ToGeneralArr(appIds...),
	}).Order("position ASC, created_at DESC").Scan(&res).Error
	if err != nil {
		return nil, errs.WrapGORMError(err, "Failed to list app's endpoint: %v", appIds)
	}
	return res, nil
}

func IsOnlineEndpoint(endpointId uint64) (bool, error) {
	var res *entity.EndpointV1
	err := database.DBConn.DB.Table(entity.TableEndpoints).Joins("left join (SELECT endpoint_id as label FROM endpoint_publish ) t1 ON endpoints.id = t1.label").Where("is_deleted=0").Clauses(clause.Eq{
		Column: "id",
		Value:  endpointId,
	}).Scan(&res).Error
	if err != nil {
		return false, errs.WrapGORMError(err, "Failed to verify endpoint online: %v", endpointId)
	}
	if res != nil && res.Label > 0 {
		return true, nil
	}
	return false, nil
}

func PageEndpointPublish(ctx context.Context, endpointIDs []uint64, offset, limit int) ([]*entity.EndpointPublish, error) {
	var res []*entity.EndpointPublish
	txDB := MustFromContext(ctx).Table(entity.TableEndpointPublish).Where("endpoint!=''").Order("created_at DESC")
	if len(endpointIDs) > 0 {
		txDB.Clauses(clause.IN{Column: "endpoint_id", Values: utils.ToGeneralArr(endpointIDs...)})
	}
	if err := txDB.Offset(offset).Limit(limit).Find(&res).Error; err != nil {
		return nil, errs.WrapGORMError(err, "Failed to list deployed endpoint")
	}
	return res, nil
}

func PageEndpoints(ctx context.Context, endpointIDs []uint64, offset, limit int) ([]*entity.Endpoint, error) {
	var res []*entity.Endpoint
	txDB := MustFromContext(ctx).Table(entity.TableEndpoints).Where("is_deleted=0 and endpoint!=''").Order("created_at DESC")
	if len(endpointIDs) > 0 {
		txDB.Clauses(clause.IN{Column: "id", Values: utils.ToGeneralArr(endpointIDs...)})
	}
	if err := txDB.Offset(offset).Limit(limit).Find(&res).Error; err != nil {
		return nil, errs.WrapGORMError(err, "Failed to list endpoint")
	}
	return res, nil
}

func ListPublishedEndpointByIDs(ctx context.Context, endpointIDs []uint64) ([]*entity.EndpointPublish, error) {
	db := MustFromContext(ctx)
	var res []*entity.EndpointPublish
	err := db.Table(entity.TableEndpointPublish).Where("endpoint_id in ?", endpointIDs).Find(&res).Error
	if err != nil {
		return nil, errs.WrapGORMError(err, "Failed to list deployed endpoint by ids")
	}
	return res, nil
}

func ListEndpointByIDs(ctx context.Context, endpointIDs []uint64) ([]*entity.Endpoint, error) {
	db := MustFromContext(ctx)
	var res []*entity.Endpoint
	err := db.Table(entity.TableEndpoints).Where("id in ?", endpointIDs).Find(&res).Error
	if err != nil {
		return nil, errs.WrapGORMError(err, "Failed to list endpoint by ids")
	}
	return res, nil
}

func ListEndpointByIDsExcludeDeleted(ctx context.Context, appID uint64, endpointIDs []uint64) ([]*entity.Endpoint, error) {
	db := MustFromContext(ctx)
	var res []*entity.Endpoint
	err := db.Table(entity.TableEndpoints).Where("id in ? and dataapp_id=? and is_deleted=0", endpointIDs, appID).Find(&res).Error
	if err != nil {
		return nil, errs.WrapGORMError(err, "Failed to list endpoint by ids")
	}
	return res, nil
}

func ListEndpoints(ctx context.Context, appId uint64, endpointType ...string) ([]*entity.Endpoint, error) {
	db := MustFromContext(ctx)
	var res []*entity.Endpoint
	tx := db.Table(entity.TableEndpoints).Where("dataapp_id=? and is_deleted=0", appId).Order("created_at DESC")
	if len(endpointType) > 0 {
		tx = tx.Where("type IN ?", endpointType)
	}
	err := tx.Find(&res).Error
	if err != nil {
		return nil, errs.WrapGORMError(err, "Failed to list app's endpoint: %d", appId)
	}
	return res, nil
}

func ListEndpointsWithPreID(ctx context.Context, appId uint64, preID uint64, pageSize int) ([]*entity.Endpoint, error) {
	db := MustFromContext(ctx)
	var res []*entity.Endpoint
	err := db.Table(entity.TableEndpoints).
		Where("dataapp_id=? and is_deleted=0 AND id > ?", appId, preID).
		Order("id ASC").Limit(pageSize).Find(&res).Error
	if err != nil {
		return nil, errs.WrapGORMError(err, "Failed to list app's endpoint: %d", appId)
	}
	return res, nil
}

func CreateAutoGenerateEndpoints(ctx context.Context, endpoints []*entity.Endpoint) ([]*entity.Endpoint, error) {
	if err := MustFromContext(ctx).Table(entity.TableEndpoints).Create(endpoints).Error; err != nil {
		return nil, errs.WrapGORMError(err, "Failed to create endpoint")
	}
	endpointIds := make([]*entity.AutoEndpointId, 0)
	for _, endpoint := range endpoints {
		endpointIds = append(endpointIds, &entity.AutoEndpointId{AgeID: endpoint.ID})
	}
	if err := MustFromContext(ctx).Table(entity.TableEndpointExtAttributes).Create(endpointIds).Error; err != nil {
		return nil, errs.WrapGORMError(err, "Failed to create batch endpoint id")
	}
	return endpoints, nil

}

func SaveEndpoint(ctx context.Context, endpoint *entity.Endpoint, selectedKeys map[string]interface{}) (*entity.Endpoint, error) {
	endpoint.Vid = 0
	return CreateEndpoint(ctx, endpoint, selectedKeys)
}

func CreateEndpoint(ctx context.Context, endpoint *entity.Endpoint, selectedKeys map[string]interface{}) (*entity.Endpoint, error) {

	var err error
	tx := MustFromContext(ctx).Table(entity.TableEndpoints)
	for key := range selectedKeys {
		if selectedKeys[key] != nil {
			tx.Statement.Selects = append(tx.Statement.Selects, key)
		} else {
			tx.Statement.Omits = append(tx.Statement.Omits, key)
		}
	}

	var affected int64
	if endpoint.ID > 0 {
		var endpointMap map[string]interface{}
		endpointMap, err = utils.ConvertObjToMap(endpoint)
		if err != nil {
			return nil, tierrs.NewBadRequest("failed to convert endpoint info", err)
		}
		if len(tx.Statement.Selects) > 0 {
			tx.Statement.Selects = append(tx.Statement.Selects, "vid")
		}
		endpointMap["dataapp_id"] = endpointMap["dataappp_id"]
		endpointMap["updated_at"] = time.Now()
		endpointMap["vid"] = gorm.Expr("vid + 1")
		endpointMap["cluster_id"] = endpoint.ClusterId
		delete(endpointMap, "id")
		delete(endpointMap, "created_at")
		delete(endpointMap, "created_by")
		delete(endpointMap, "dataappp_id")
		if endpoint.Vid > 0 {
			tx = tx.Where("vid=?", endpoint.Vid)
		}
		resDB := tx.Where("id=?", endpoint.ID).Updates(endpointMap)
		err = resDB.Error
		affected = resDB.RowsAffected
		endpoint.Vid += 1
	} else {
		endpoint.Vid = 1
		resDB := tx.Create(&endpoint)
		err = resDB.Error
		affected = resDB.RowsAffected
	}
	if err != nil {
		return nil, errs.WrapGORMError(err, "Failed to save endpoint")
	} else if affected == 0 {
		return nil, errs.ErrEndpointVersionConflict
	}
	return endpoint, nil
}

func DeleteEndpoint(ctx context.Context, endpointId uint64, operator string) error {
	data := map[string]interface{}{
		"is_deleted": 1,
		"updated_by": operator,
	}
	if err := MustFromContext(ctx).Table(entity.TableEndpoints).Where("id=?", endpointId).Updates(data).Error; err != nil {
		return errs.WrapGORMError(err, "Failed to delete endpoint")
	}
	return nil
}

func BatchDeleteEndpoint(ctx context.Context, endpointIds []uint64, operator string) error {
	data := map[string]interface{}{
		"is_deleted": 1,
		"updated_by": operator,
	}
	if err := MustFromContext(ctx).Table(entity.TableEndpoints).Where("id in ?", endpointIds).Updates(data).Error; err != nil {
		return errs.WrapGORMError(err, "Failed to delete endpoint")
	}
	return nil
}

func RecoverEndpoint(ctx context.Context, endpointId uint64, operator string) error {
	data := map[string]interface{}{
		"is_deleted": 0,
		"updated_by": operator,
	}
	if err := MustFromContext(ctx).Table(entity.TableEndpoints).Where("id=?", endpointId).Updates(data).Error; err != nil {
		return errs.WrapGORMError(err, "Failed to Recover endpoint")
	}
	return nil
}

func GetEndpointWithoutCheck(ctx context.Context, endpointId uint64) (*entity.Endpoint, error) {
	var res entity.Endpoint
	if err := MustFromContext(ctx).Table(entity.TableEndpoints).Where("id=?", endpointId).First(&res).Error; err != nil {
		return nil, errs.WrapGORMError(err, "Failed to get endpoint")
	}
	return &res, nil
}

func GetEndpointWithCtx(ctx context.Context, endpointId uint64) (*entity.Endpoint, error) {
	return getEndpoint(ctx, endpointId)
}

func GetEndpoint(endpointId uint64) (*entity.Endpoint, error) {
	return getEndpoint(context.Background(), endpointId)
}

// XXX(bing.wang) This interface will also retrieve deleted data endpoints and is intended solely for internal analysis purposes.
func GetEndpointRaw(ctx context.Context, endpointId uint64) (*entity.Endpoint, error) {
	var res entity.Endpoint
	if err := MustFromContext(ctx).Table(entity.TableEndpoints).Where("id=?", endpointId).First(&res).Error; err != nil {
		return nil, errs.WrapGORMError(err, "Failed to get endpoint")
	}
	return &res, nil
}

func getEndpoint(ctx context.Context, endpointId uint64) (*entity.Endpoint, error) {
	var res entity.Endpoint
	if err := MustFromContext(ctx).Table(entity.TableEndpoints).Where("id=? and is_deleted=0", endpointId).First(&res).Error; err != nil {
		return nil, errs.WrapGORMError(err, "Failed to get endpoint")
	}
	return &res, nil
}

func GetAppPublishedEndpointCount(dataappId uint64) (int64, error) {
	var res int64
	if err := database.DBConn.DB.Table(entity.TableEndpointPublish).Where("dataapp_id=?", dataappId).Count(&res).Error; err != nil {
		return 0, errs.WrapGORMError(err, "Failed to get deployed endpoint count")
	}
	return res, nil
}

func GetAppEndpointCount(dataappId uint64) (int64, error) {
	var res int64
	if err := database.DBConn.DB.Table(entity.TableEndpoints).Where("dataapp_id=? and is_deleted=0", dataappId).Count(&res).Error; err != nil {
		return 0, errs.WrapGORMError(err, "Failed to get deployed endpoint count")
	}
	return res, nil
}

func ListAppPublishedEndpoint(ctx context.Context, dataappId uint64) ([]*entity.EndpointPublish, error) {
	var res []*entity.EndpointPublish
	if err := MustFromContext(ctx).Table(entity.TableEndpointPublish).Where("dataapp_id=?", dataappId).Find(&res).Error; err != nil {
		return nil, errs.WrapGORMError(err, "Failed to list app's published endpoint")
	}
	return res, nil
}

func GetPublishedEndpoint(endpointId uint64) (*entity.EndpointPublish, error) {
	var res entity.EndpointPublish
	err := database.DBConn.DB.Table(entity.TableEndpointPublish).Where("endpoint_id=?", endpointId).First(&res).Error
	if err != nil {
		return nil, errs.WrapGORMError(err, "Failed to get deployed endpoint: %d", endpointId)
	}
	return &res, nil
}

func ListEndpointPublishVersions(endpointId uint64) ([]*entity.EndpointPublishVersions, error) {
	var res []*entity.EndpointPublishVersions
	fields := []string{"dataapp_id", "endpoint_id", "version", "created_at", "updated_at", "is_online"}
	if err := database.DBConn.DB.Table(entity.TableEndpointPublishVersions).Select(fields).Where("endpoint_id=?", endpointId).Order("created_at DESC").Find(&res).Error; err != nil {
		return nil, errs.WrapGORMError(err, "Failed to list endpoint versions: %d", endpointId)
	}
	return res, nil
}

func GetLastPublishEvent(endpointId uint64) (*entity.EndpointPublishVersions, error) {
	var res *entity.EndpointPublishVersions
	if err := database.DBConn.DB.Table(entity.TableEndpointPublishVersions).Where("endpoint_id=?", endpointId).Order("created_at DESC").First(&res).Error; err != nil {
		return nil, errs.WrapGORMError(err, "Failed to get last deployed endpoint event: %d", endpointId)
	}
	return res, nil
}

func GetLastPublishVersion(endpointId uint64) (*entity.EndpointPublishVersions, error) {
	var res *entity.EndpointPublishVersions
	if err := database.DBConn.DB.Table(entity.TableEndpointPublishVersions).Where("endpoint_id=? and is_online=1", endpointId).Order("created_at DESC").First(&res).Error; err != nil {
		return nil, errs.WrapGORMError(err, "Failed to get last deployed endpoint versions: %d", endpointId)
	}
	return res, nil
}

func CreateOnlineEndpointWithTX(ctx context.Context, endpoint *entity.Endpoint, version, createdBy string) error {
	return Transaction(ctx, func(ctx context.Context) error {
		return CreateOnlineEndpoint(ctx, endpoint, version, createdBy)
	})
}

func CreateOnlineEndpoint(ctx context.Context, endpoint *entity.Endpoint, version, createdBy string) error {
	db := MustFromContext(ctx)
	endpointPublish := entity.EndpointPublish{
		DataappId:   endpoint.DataappId,
		EndpointId:  endpoint.ID,
		Version:     version,
		Type:        endpoint.Type,
		Endpoint:    endpoint.Endpoint,
		Method:      endpoint.Method,
		ClusterId:   endpoint.ClusterId,
		Database:    endpoint.Database,
		SqlTemplate: endpoint.SqlTemplate,
		Args:        endpoint.Args,
		Settings:    endpoint.Settings,
		CreatedBy:   createdBy,
		UpdatedBy:   createdBy,
		RWMode:      endpoint.RWMode,
		Batch:       endpoint.IsBatch(),
		Tag:         endpoint.Tag,
	}
	endpointVersion := &entity.EndpointPublishVersions{
		DataappId:   endpoint.DataappId,
		EndpointId:  endpoint.ID,
		Name:        endpoint.Name,
		Description: endpoint.Description,
		Type:        endpoint.Type,
		Endpoint:    endpoint.Endpoint,
		Method:      endpoint.Method,
		ClusterId:   endpoint.ClusterId,
		Database:    endpoint.Database,
		SqlTemplate: endpoint.SqlTemplate,
		Settings:    endpoint.Settings,
		Args:        endpoint.Args,
		Version:     version,
		IsOnline:    1,
		CreatedBy:   createdBy,
		RWMode:      endpoint.RWMode,
	}

	if err := db.Table(entity.TableEndpointPublish).Clauses(clause.OnConflict{
		Columns:   []clause.Column{{Name: "endpoint_id"}},
		DoUpdates: clause.AssignmentColumns([]string{"version", "endpoint", "method", "cluster_id", "database", "sql_template", "args", "settings", "updated_by", "type", "rw_mode", "tag", "batch"}),
	}).Create(&endpointPublish).Error; err != nil {
		return errs.WrapGORMError(err, "Failed to deploy endpoint: %d", endpointPublish.EndpointId)
	}
	tilog.Info(ctx, "update online success!", zap.Any("endpoint_publish", endpointPublish))
	if err := db.Table(entity.TableEndpointPublishVersions).Create(endpointVersion).Error; err != nil {
		return errs.WrapGORMError(err, "Failed to record endpoint deploy version: %d", endpointVersion.EndpointId)
	}
	return nil
}

func UpdatePublishedEndpoint(ctx context.Context, endpointID uint64, endpointPublished entity.EndpointPublish) error {
	err := MustFromContext(ctx).Table(entity.TableEndpointPublish).Where("endpoint_id=?", endpointID).Updates(endpointPublished).Error
	if err != nil {
		return errs.WrapGORMError(err, "Failed to update deployed endpoint")
	}
	return nil
}

func OfflineEndpointWithTX(ctx context.Context, endpointId uint64, createdBy string) error {
	return Transaction(ctx, func(ctx context.Context) error {
		return OfflineEndpoint(ctx, endpointId, createdBy)
	})
}

func OfflineEndpoint(ctx context.Context, endpointId uint64, createdBy string) error {
	db := MustFromContext(ctx)
	res := entity.EndpointPublishVersions{}
	if err := db.Table(entity.TableEndpointPublish).Where("endpoint_id=?", endpointId).Delete(&entity.EndpointPublish{}).Error; err != nil {
		return errs.WrapGORMError(err, "Failed to undeploy endpoint: %d", endpointId)
	}
	if err := db.Table(entity.TableEndpointPublishVersions).Where("endpoint_id=?", endpointId).Order("created_at DESC").First(&res).Error; err != nil {
		return errs.WrapGORMError(err, "Failed to get endpoint deploy version")
	}
	if err := db.Table(entity.TableEndpointPublishVersions).Create(&entity.EndpointPublishVersions{
		DataappId:   res.DataappId,
		EndpointId:  res.EndpointId,
		Name:        res.Name,
		Description: res.Description,
		Endpoint:    res.Endpoint,
		Method:      res.Method,
		ClusterId:   res.ClusterId,
		Database:    res.Database,
		SqlTemplate: res.SqlTemplate,
		Args:        res.Args,
		Settings:    res.Settings,
		Version:     res.Version,
		IsOnline:    0,
		CreatedBy:   createdBy,
	}).Error; err != nil {
		return errs.WrapGORMError(err, "Failed to record endpoint undeploy version: %d", endpointId)
	}
	return nil
}

func IsExistEndpoint(appId, endpointId uint64, endpoint, method string) (bool, *entity.Endpoint, error) {
	res := entity.Endpoint{}
	var result *gorm.DB
	if endpointId == 0 {
		result = database.DBConn.DB.Table(entity.TableEndpoints).Where("dataapp_id = ? and endpoint=? and method=? and is_deleted=0", appId, endpoint, method).Find(&res)
	} else {
		result = database.DBConn.DB.Table(entity.TableEndpoints).Where("dataapp_id = ? and endpoint=? and method=? and is_deleted=0 and id!=?", appId, endpoint, method, endpointId).Find(&res)
	}

	err := result.Error
	if err != nil {
		return false, nil, errs.WrapGORMError(err, "Failed to check endpoint exists")
	}
	if result.RowsAffected > 0 {
		return true, &res, nil
	} else {
		return false, nil, nil
	}
}

func ListDraftEndpointByNameAndMethod(dataAppID uint64, endpoint string, method string) ([]*entity.Endpoint, error) {
	var resList []*entity.Endpoint
	endpointList, err := GetDraftEndpointByName(dataAppID, endpoint)
	if err != nil {
		return nil, err
	}
	for _, val := range endpointList {
		if val.Method == method {
			resList = append(resList, val)
		}
	}
	return resList, nil
}

func GetDraftEndpointByNameAndMethod(dataAppID uint64, endpoint string, method string) (*entity.Endpoint, error) {
	resList, err := ListDraftEndpointByNameAndMethod(dataAppID, endpoint, method)
	if err != nil {
		return nil, err
	}
	if len(resList) > 1 {
		return nil, tierrs.NewInternalServerError("More than 1 records are found by name and method", nil)
	} else if len(resList) == 0 {
		return nil, tierrs.NewNotFound("No record is found.", nil)
	}

	return resList[0], nil
}

func GetDraftEndpointByName(dataAppID uint64, endpoint string) ([]*entity.Endpoint, error) {
	res := []*entity.Endpoint{}
	if err := database.DBConn.DB.Table(entity.TableEndpoints).Where("dataapp_id=? and endpoint=? and is_deleted=0", dataAppID, endpoint).Order("created_at DESC").Find(&res).Error; err != nil {
		return nil, errs.WrapGORMError(err, fmt.Sprintf("GetDraftEndpointByName failed, dataAppID: %d, endpoint: %s, err:%v", dataAppID, endpoint, err))
	}

	return res, nil
}

func GetPublishedEndpointByName(dataAppID uint64, endpoint string) ([]*entity.EndpointPublish, error) {
	res := []*entity.EndpointPublish{}
	if err := database.DBConn.DB.Table(entity.TableEndpointPublish).Where("dataapp_id=? and endpoint=?", dataAppID, endpoint).Order("created_at DESC").Find(&res).Error; err != nil {
		return nil, errs.WrapGORMError(err, fmt.Sprintf("GetLatestOnlineEndpointByName failed, dataAppID: %d, endpoint: %s, err:%v", dataAppID, endpoint, err))
	}
	return res, nil
}

func SaveUserEndpointPlayground(ctx context.Context, obj *entity.UserEndpointPlayground) error {
	err := MustFromContext(ctx).Table(entity.TableUserEndpointPlayground).Where(entity.UserEndpointPlayground{
		DataappID:  obj.DataappID,
		EndpointID: obj.EndpointID,
		CreatedBy:  obj.CreatedBy,
	}).Assign(entity.UserEndpointPlayground{
		Content: obj.Content,
	}).FirstOrCreate(obj).Error
	if err != nil {
		return errs.WrapGORMError(err, fmt.Sprintf("Failed to save user endpoint playground content, dataapp:%d, endpoint:%d, user:%s", obj.DataappID, obj.EndpointID, obj.CreatedBy))
	}
	return nil
}

func LoadUserEndpointPlayground(ctx context.Context, dataappID, endpointID uint64, operator string) (*entity.UserEndpointPlayground, error) {
	var result entity.UserEndpointPlayground
	err := MustFromContext(ctx).Table(entity.TableUserEndpointPlayground).Where(entity.UserEndpointPlayground{
		DataappID:  dataappID,
		EndpointID: endpointID,
		CreatedBy:  operator,
	}).First(&result).Error
	if err != nil {
		return nil, errs.WrapGORMError(err, fmt.Sprintf("Failed to load user endpoint playground content, dataapp:%d, endpoint:%d, user:%s", dataappID, endpointID, operator))
	}
	return &result, nil
}

func CountCreateEndpointUsers(ctx context.Context, startSecAt int64, endSecAt int64, limit int) ([]entity.AnalysisStrKVResult, error) {
	startTimeObj := time.Unix(startSecAt, 0)
	endTimeObj := time.Unix(endSecAt, 0)

	var results []entity.AnalysisStrKVResult
	err := MustFromContext(ctx).Table(entity.TableEndpoints).
		Select("created_by as `key`, COUNT(1) as value").
		Where("created_at between ? and ?", startTimeObj, endTimeObj).
		Group("created_by").
		Limit(limit).
		Scan(&results).Error
	if err != nil {
		return results, errs.WrapGORMError(err, "failed to CountCreateEndpointUsers")
	}
	if results != nil {
		tilog.Info(ctx, "CountCreateEndpointUsers", zap.Any("startTimeObj", startTimeObj),
			zap.Any("endTimeObj", endTimeObj), zap.Any("results", results))
	}
	return results, nil
}

func CountPublishEndpointUsers(ctx context.Context, startSecAt int64, endSecAt int64, limit int) ([]entity.AnalysisStrKVResult, error) {
	startTimeObj := time.Unix(startSecAt, 0)
	endTimeObj := time.Unix(endSecAt, 0)

	var results []entity.AnalysisStrKVResult
	err := MustFromContext(ctx).Table(entity.TableEndpointPublish).
		Select("created_by as `key`, COUNT(1) as value").
		Where("created_at between ? and ?", startTimeObj, endTimeObj).
		Group("created_by").
		Limit(limit).
		Scan(&results).Error
	if err != nil {
		return results, errs.WrapGORMError(err, "failed to CountCreateEndpointUsers")
	}
	if results != nil {
		tilog.Info(ctx, "CountCreateEndpointUsers", zap.Any("startTimeObj", startTimeObj),
			zap.Any("endTimeObj", endTimeObj), zap.Any("results", results))
	}
	return results, nil
}

func ListEndpointsByTimeRange(ctx context.Context, startSecAt int64, endSecAt int64) ([]*entity.Endpoint, error) {
	startTimeObj := time.Unix(startSecAt, 0)
	endTimeObj := time.Unix(endSecAt, 0)
	var res []*entity.Endpoint
	txDB := MustFromContext(ctx).Table(entity.TableEndpoints).Where("created_at between ? and ?", startTimeObj, endTimeObj)
	if err := txDB.Find(&res).Error; err != nil {
		return nil, errs.WrapGORMError(err, "Failed to list dataapp")
	}
	return res, nil
}

func ListPublishEndpointsByTimeRange(ctx context.Context, startSecAt int64, endSecAt int64) ([]*entity.EndpointPublish, error) {
	startTimeObj := time.Unix(startSecAt, 0)
	endTimeObj := time.Unix(endSecAt, 0)
	var res []*entity.EndpointPublish
	txDB := MustFromContext(ctx).Table(entity.TableEndpointPublish).Where("created_at between ? and ?", startTimeObj, endTimeObj)
	if err := txDB.Find(&res).Error; err != nil {
		return nil, errs.WrapGORMError(err, "Failed to list dataapp")
	}
	return res, nil
}

type EndpointCluster struct {
	OrgID     uint64 `gorm:"column:org_id"`
	ClusterID uint64 `gorm:"column:cluster_id"`
}

func ListDraftEndpointClusters(ctx context.Context, page, pageSize int) ([]EndpointCluster, error) {
	sql := `SELECT a.org_id as org_id, b.cluster_id as cluster_id FROM dataapps a JOIN endpoints b ON a.id=b.dataapp_id
	 	WHERE (a.is_deleted=0 OR (a.is_deleted=1 AND a.updated_at>=?)) 
			AND (b.is_deleted=0 OR (b.is_deleted=1 AND b.updated_at>=?)) 
			AND b.cluster_id != 0 GROUP BY a.org_id,b.cluster_id LIMIT ?,?`
	deleteTimeline := time.Now().Add(-time.Hour).Format("2006-01-02 15:04:05")
	poffset := (page - 1) * pageSize
	var result []EndpointCluster
	err := MustFromContext(ctx).Raw(sql, deleteTimeline, deleteTimeline, poffset, pageSize).Find(&result).Error
	if err != nil {
		return nil, errs.WrapGORMError(err, "Failed to list draft endpoint org and cluster")
	}
	return result, nil
}

func ListDeployedEndpointClusters(ctx context.Context, page, pageSize int) ([]EndpointCluster, error) {
	sql := `SELECT a.org_id as org_id, b.cluster_id as cluster_id FROM dataapps a JOIN endpoint_publish b ON a.id=b.dataapp_id
	 	WHERE (a.is_deleted=0 OR (a.is_deleted=1 AND a.updated_at>=?)) 
			AND b.cluster_id != 0 GROUP BY a.org_id,b.cluster_id LIMIT ?,?`
	deleteTimeline := time.Now().Format("2006-01-02 15:04:05")
	poffset := (page - 1) * pageSize
	var result []EndpointCluster
	err := MustFromContext(ctx).Raw(sql, deleteTimeline, poffset, pageSize).Find(&result).Error
	if err != nil {
		return nil, errs.WrapGORMError(err, "Failed to list deployed endpoint org and cluster")
	}
	return result, nil
}