package backup

import (
	"context"
	"strings"
	"time"

	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/pkgv2/db"
	"gorm.io/gorm"
)

const batchAddSize = 500

type ServerlessBackupRepoImpl struct {
	dbProvider db.Provider
}

var _ ServerlessBackupRepo = (*ServerlessBackupRepoImpl)(nil)

func NewServerlessBackupRepo(dbProvider db.Provider) ServerlessBackupRepo {
	return &ServerlessBackupRepoImpl{
		dbProvider: dbProvider,
	}
}

func (b *ServerlessBackupRepoImpl) Load(ctx context.Context, id entity.ID) (*entity.ServerlessBackup, error) {
	rdb := b.dbProvider.ReadDB(ctx)
	backup := new(entity.ServerlessBackup)
	if err := entity.NewServerlessBackupQuerySet(rdb).IDEq(id).One(backup); err != nil {
		return nil, err
	}
	return backup, nil
}

func (b *ServerlessBackupRepoImpl) GetByFilter(
	ctx context.Context, filter *ServerlessBackupFilter,
) (*entity.ServerlessBackup, error) {
	rdb := b.dbProvider.ReadDB(ctx)
	rdb = b.buildTxServerlessBackupFilter(rdb, filter)
	backup := new(entity.ServerlessBackup)
	if result := rdb.First(backup); result.Error != nil {
		return nil, result.Error
	}
	return backup, nil
}

func (b *ServerlessBackupRepoImpl) ListByFilter(
	ctx context.Context, filter *ServerlessBackupFilter,
) ([]*entity.ServerlessBackup, error) {
	rdb := b.dbProvider.ReadDB(ctx)
	rdb = b.buildTxServerlessBackupFilter(rdb, filter)
	var backups []*entity.ServerlessBackup
	if result := rdb.Find(&backups); result.Error != nil {
		return nil, result.Error
	}
	return backups, nil
}

func (b *ServerlessBackupRepoImpl) Add(ctx context.Context, backup *entity.ServerlessBackup) error {
	wdb := b.dbProvider.WriteDB(ctx)
	if result := wdb.Create(backup); result.Error != nil {
		return result.Error
	}
	return nil
}

func (b *ServerlessBackupRepoImpl) BatchAdd(ctx context.Context, backups []*entity.ServerlessBackup) error {
	wdb := b.dbProvider.WriteDB(ctx)
	if result := wdb.CreateInBatches(backups, batchAddSize); result.Error != nil {
		return result.Error
	}
	return nil
}

func (b *ServerlessBackupRepoImpl) Delete(ctx context.Context, backup *entity.ServerlessBackup) error {
	deletedAt := time.Now()
	backup.DeletedAt = &deletedAt
	wdb := b.dbProvider.WriteDB(ctx)
	if result := wdb.Save(backup); result.Error != nil {
		return result.Error
	}
	return nil
}

func (b *ServerlessBackupRepoImpl) buildTxServerlessBackupFilter(
	tx *gorm.DB, filter *ServerlessBackupFilter,
) *gorm.DB {
	if filter.BackupID != nil {
		tx = tx.Where("backup_id = ?", *filter.BackupID)
	}
	if filter.Region != nil {
		tx = tx.Where("region = ?", *filter.Region)
	}
	if filter.BackupTimeGte != nil {
		tx = tx.Where("backup_time >= ?", *filter.BackupTimeGte)
	}
	if filter.OrderByBackupTime != nil {
		if *filter.OrderByBackupTime == OrderByAsc {
			tx = tx.Order("backup_time asc")
		} else if *filter.OrderByBackupTime == OrderByDesc {
			tx = tx.Order("backup_time desc")
		}
	}
	return tx
}

type ServerlessUserClusterBackupRepoImpl struct {
	dbProvider db.Provider
}

var _ ServerlessUserClusterBackupRepo = (*ServerlessUserClusterBackupRepoImpl)(nil)

func NewServerlessUserClusterBackupRepo(dbProvider db.Provider) ServerlessUserClusterBackupRepo {
	return &ServerlessUserClusterBackupRepoImpl{
		dbProvider: dbProvider,
	}
}

func (b *ServerlessUserClusterBackupRepoImpl) Load(
	ctx context.Context, id entity.ID,
) (*entity.ServerlessUserClusterBackup, error) {
	rdb := b.dbProvider.ReadDB(ctx)
	backup := new(entity.ServerlessUserClusterBackup)
	if err := entity.NewServerlessUserClusterBackupQuerySet(rdb).
		IDEq(id).PreloadServerlessBackup().One(backup); err != nil {
		return nil, err
	}
	return backup, nil
}

func (b *ServerlessUserClusterBackupRepoImpl) GetByFilter(
	ctx context.Context, filter *ServerlessUserClusterBackupFilter,
) (*entity.ServerlessUserClusterBackup, error) {
	rdb := b.dbProvider.ReadDB(ctx)
	rdb = b.buildTxServerlessUserClusterBackupRepoFilter(rdb, filter)
	backup := new(entity.ServerlessUserClusterBackup)
	if result := rdb.Preload("ServerlessBackup").First(backup); result.Error != nil {
		return nil, result.Error
	}
	return backup, nil
}

func (b *ServerlessUserClusterBackupRepoImpl) ListByFilter(
	ctx context.Context, filter *ServerlessUserClusterBackupFilter,
) ([]*entity.ServerlessUserClusterBackup, error) {
	rdb := b.dbProvider.ReadDB(ctx)
	rdb = b.buildTxServerlessUserClusterBackupRepoFilter(rdb, filter)
	var backups []*entity.ServerlessUserClusterBackup
	if result := rdb.Preload("ServerlessBackup").Find(&backups); result.Error != nil {
		return nil, result.Error
	}
	return backups, nil
}

func (b *ServerlessUserClusterBackupRepoImpl) CountByFilter(
	ctx context.Context, filter *ServerlessUserClusterBackupFilter,
) (uint64, error) {
	rdb := b.dbProvider.ReadDB(ctx)
	rdb = b.buildTxServerlessUserClusterBackupRepoFilter(rdb, filter)
	var total int64
	if result := rdb.Model(&entity.ServerlessUserClusterBackup{}).Count(&total); result.Error != nil {
		return 0, result.Error
	}
	return uint64(total), nil
}

func (b *ServerlessUserClusterBackupRepoImpl) Add(
	ctx context.Context, backup *entity.ServerlessUserClusterBackup,
) error {
	wdb := b.dbProvider.WriteDB(ctx)
	if result := wdb.Create(backup); result.Error != nil {
		return result.Error
	}
	return nil
}

func (b *ServerlessUserClusterBackupRepoImpl) Save(
	ctx context.Context, backup *entity.ServerlessUserClusterBackup,
) error {
	wdb := b.dbProvider.WriteDB(ctx)
	updatedAt := time.Now()
	backup.UpdatedAt = &updatedAt
	if result := wdb.Save(backup); result.Error != nil {
		return result.Error
	}
	return nil
}

func (b *ServerlessUserClusterBackupRepoImpl) BatchSave(
	ctx context.Context, ids []entity.ID, updateFields map[string]interface{},
) error {
	wdb := b.dbProvider.WriteDB(ctx)
	batchNum := 1000
	for i := 0; i < len(ids); i += batchNum {
		l := i
		r := i + batchNum
		if r > len(ids) {
			r = len(ids)
		}

		updateIDs := ids[l:r]
		result := wdb.Model(entity.ServerlessUserClusterBackup{}).
			Where("id in ?", updateIDs).
			Updates(updateFields)
		if result.Error != nil {
			return result.Error
		}
	}
	return nil
}

func (b *ServerlessUserClusterBackupRepoImpl) Delete(
	ctx context.Context, backup *entity.ServerlessUserClusterBackup,
) error {
	wdb := b.dbProvider.WriteDB(ctx)
	if result := wdb.Delete(backup); result.Error != nil {
		return result.Error
	}
	return nil
}

func (b *ServerlessUserClusterBackupRepoImpl) buildTxServerlessUserClusterBackupRepoFilter(
	tx *gorm.DB, filter *ServerlessUserClusterBackupFilter,
) *gorm.DB {
	if filter.ClusterID != nil {
		tx = tx.Where("cluster_id = ?", *filter.ClusterID)
	}
	if filter.ProjectID != nil {
		tx = tx.Where("project_id = ?", *filter.ProjectID)
	}
	if filter.TenantID != nil {
		tx = tx.Where("tenant_id = ?", *filter.TenantID)
	}
	if filter.ServerlessBackupID != nil {
		tx = tx.Where("serverless_backup_id = ?", *filter.ServerlessBackupID)
	}
	if len(filter.Statuses) > 0 {
		tx = tx.Where("status in ?", filter.Statuses)
	}
	if filter.CreatedAtGt != nil {
		tx = tx.Where("created_at > ?", *filter.CreatedAtGt)
	}
	if filter.Offset != nil {
		tx = tx.Offset(int(*filter.Offset))
	}
	if filter.Limit != nil {
		tx = tx.Limit(int(*filter.Limit))
	}
	b.buildTxServerlessUserClusterBackupRepoOrderBy(tx, filter)
	return tx
}

func (b *ServerlessUserClusterBackupRepoImpl) buildTxServerlessUserClusterBackupRepoOrderBy(
	tx *gorm.DB, filter *ServerlessUserClusterBackupFilter,
) *gorm.DB {
	if filter.OrderByCreatedAt != nil {
		if *filter.OrderByCreatedAt == OrderByAsc {
			tx = tx.Order("created_at asc")
		} else if *filter.OrderByCreatedAt == OrderByDesc {
			tx = tx.Order("created_at desc")
		}
	}
	return tx
}

func (b *ServerlessUserClusterBackupRepoImpl) GetCreateTableSql(ctx context.Context) (string, error) {
	rdb := b.dbProvider.ReadDB(ctx)
	row := rdb.Raw("show create table serverless_user_cluster_backups").Row()
	var tableName, tableSql string
	if err := row.Scan(&tableName, &tableSql); err != nil {
		return "", err
	}
	return tableSql, nil
}

func (b *ServerlessUserClusterBackupRepoImpl) ReorgMaxValuePartition(
	ctx context.Context, additionalPartitions []string,
) error {
	if len(additionalPartitions) == 0 {
		return nil
	}

	builder := strings.Builder{}
	builder.WriteString("ALTER TABLE serverless_user_cluster_backups REORGANIZE PARTITION pMax INTO (\n")
	for _, partition := range additionalPartitions {
		builder.WriteString(partition)
		builder.WriteString(",\n")
	}
	builder.WriteString("PARTITION pMax VALUES LESS THAN (MAXVALUE))")

	wdb := b.dbProvider.WriteDB(ctx)
	return wdb.Exec(builder.String()).Error
}

func (b *ServerlessUserClusterBackupRepoImpl) DropPartition(ctx context.Context, partitionName string) error {
	dropSql := "ALTER TABLE serverless_user_cluster_backups DROP PARTITION " + partitionName
	wdb := b.dbProvider.WriteDB(ctx)
	return wdb.Exec(dropSql).Error
}