package repo_imp

import (
	"context"
	"time"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

type changefeedRepoImp struct {
	dbProvider db.TransProvider
}

func NewChangefeedRepoImp(dbProvider db.TransProvider) repo.ChangefeedRepo {
	return &changefeedRepoImp{dbProvider}
}

func (r *changefeedRepoImp) ListByFilter(
	ctx context.Context,
	filter entity.ListChangefeedsFilter, //nolint:gocritic
) ([]*entity.CDCChangefeed, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "changefeedRepo.ListByFilter")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	rdb = buildCDCChangefeedFilter(filter, rdb)

	var changefeeds []*entity.CDCChangefeed
	if err := rdb.Find(&changefeeds).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list cdc changefeeds")
	}
	return changefeeds, nil
}

func (r *changefeedRepoImp) CountByFilter(
	ctx context.Context,
	filter entity.ListChangefeedsFilter, //nolint:gocritic
) (int64, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "changefeedRepo.CountByFilter")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	rdb = buildCDCChangefeedCountFilter(filter, rdb)

	var count int64
	if err := rdb.Model(&entity.CDCChangefeed{}).Count(&count).Error; err != nil {
		return 0, bizerr.WrapGormError(err, "fail to count cdc changefeeds")
	}
	return count, nil
}

func (r *changefeedRepoImp) Create(
	ctx context.Context,
	cdc *entity.CDCChangefeed,
) (*entity.CDCChangefeed, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "changefeedRepo.Create")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Create(cdc).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to create cdc changefeed")
	}
	return cdc, nil
}

func (r *changefeedRepoImp) GetById(
	ctx context.Context,
	clusterId, changefeedId string,
) (*entity.CDCChangefeed, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "changefeedRepo.GetById")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	changefeed := &entity.CDCChangefeed{}
	if err := rdb.Where("id = ? and cluster_id = ? and state != ?", changefeedId, clusterId, constant.ChangefeedStateDeleted).First(changefeed).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to get cdc changefeed")
	}
	return changefeed, nil
}

func (r *changefeedRepoImp) GetByID(
	ctx context.Context,
	changefeedID string,
) (*entity.CDCChangefeed, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "changefeedRepo.GetByID")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	changefeed := &entity.CDCChangefeed{}
	if err := rdb.Where("id = ? and state != ?", changefeedID, constant.ChangefeedStateDeleted).
		First(changefeed).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to get cdc changefeed")
	}
	return changefeed, nil
}

// Start Resume a CDCChangefeed by user
func (r *changefeedRepoImp) Start(
	ctx context.Context,
	clusterId, changefeedId string,
) (*entity.CDCChangefeed, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "changefeedRepo.Start")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	var changefeed entity.CDCChangefeed
	if err := wdb.Where("id = ? and cluster_id = ? and state != ?", changefeedId, clusterId, constant.ChangefeedStateDeleted).
		First(&changefeed).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to find cdc changefeed for start")
	}

	allowResumeStates := []string{
		string(constant.ChangefeedStatePaused),
		string(constant.ChangefeedStateFailed),
	}
	result := wdb.Model(&changefeed).
		Where("state in ?", allowResumeStates).
		Update("state", constant.ChangfeedStateRunning)
	if result.Error != nil {
		return nil, bizerr.WrapGormError(result.Error, "fail to update cdc changefeed state to running")
	}
	if result.RowsAffected == 0 {
		return nil, bizerr.New(errorcode.AffectedNoRows, "")
	}
	changefeed.State = constant.ChangfeedStateRunning
	return &changefeed, nil
}

// SyncState sync state according to the ticdc state
func (r *changefeedRepoImp) SyncState(ctx context.Context, changefeed *entity.CDCChangefeed) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "changefeedRepo.SyncState")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)

	var current entity.CDCChangefeed
	if err := wdb.Model(&entity.CDCChangefeed{}).
		Select("state", "error", "failed_time").
		Where("id = ?", changefeed.ID).
		First(&current).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to get current cdc changefeed state")
	}

	updateColumns := map[string]any{
		"state":          changefeed.State,
		"last_sync_at":   time.Now().UTC(),
		"checkpoint_tso": changefeed.CheckpointTSO,
	}
	updateColumns["error"] = changefeed.Error
	syncTime := time.Now().UTC()
	switch changefeed.State {
	case constant.ChangefeedStateFailed, constant.ChangefeedStateWarning, constant.ChangefeedStateCreateFailed:
		if shouldRefreshFailedTime(
			current.State,
			current.Error,
			changefeed.State,
			changefeed.Error,
		) {
			updateColumns["failed_time"] = syncTime
		} else {
			updateColumns["failed_time"] = current.FailedTime
		}
	default:
		updateColumns["failed_time"] = nil
	}
	result := wdb.Model(changefeed).Updates(updateColumns)
	if result.Error != nil {
		return bizerr.WrapGormError(result.Error, "fail to sync cdc changefeed state")
	}
	if result.RowsAffected == 0 {
		return bizerr.WrapGormError(result.Error, "no row affected, fail to sync cdc changefeed state")
	}
	return nil
}

func shouldRefreshFailedTime(
	currentState constant.ChangfeedState,
	currentError string,
	nextState constant.ChangfeedState,
	nextError string,
) bool {
	return currentState != nextState || currentError != nextError
}

// Stop stops a CDCChangefeed
func (r *changefeedRepoImp) Stop(
	ctx context.Context,
	clusterId, changefeedId string,
) (*entity.CDCChangefeed, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "changefeedRepo.Stop")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	var changefeed entity.CDCChangefeed
	if err := wdb.Where("id = ? and cluster_id = ? and state != ?", changefeedId, clusterId, constant.ChangefeedStateDeleted).
		First(&changefeed).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to find cdc changefeed for stop")
	}

	allowPauseStates := []string{
		string(constant.ChangfeedStateRunning),
		string(constant.ChangefeedStateWarning),
		string(constant.ChangefeedStateFailed),
	}
	result := wdb.Model(&changefeed).
		Where("state in ?", allowPauseStates).
		Update("state", constant.ChangefeedStatePaused)
	if result.Error != nil {
		return nil, bizerr.WrapGormError(result.Error, "fail to update cdc changefeed state to paused")
	}
	if result.RowsAffected == 0 {
		return nil, bizerr.New(errorcode.AffectedNoRows, "")
	}
	changefeed.State = constant.ChangefeedStatePaused
	return &changefeed, nil
}

func (r *changefeedRepoImp) ChangeStateFromCreating(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "changefeedRepo.ChangeStateFromCreating")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)

	if changefeed.ID == "" {
		return bizerr.New(bizerr.ErrCodeParams, "changefeed id is required")
	}

	updateColumns := map[string]any{
		"state": changefeed.State,
	}
	if changefeed.State == constant.ChangefeedStateCreateFailed {
		updateColumns["error"] = changefeed.Error
		updateColumns["failed_time"] = time.Now().UTC()
	}
	if changefeed.State == constant.ChangfeedStateRunning {
		updateColumns["running_at"] = time.Now().UTC()
		updateColumns["failed_time"] = nil
		updateColumns["changefeed_id"] = changefeed.ChangefeedID
	}
	result := wdb.Model(changefeed).Where("state = ?", constant.ChangefeedStateCreating).Updates(updateColumns)
	if result.Error != nil {
		return bizerr.WrapGormError(result.Error, "fail to change state from creating")
	}
	if result.RowsAffected == 0 {
		return bizerr.New(errorcode.AffectedNoRows, "changefeed state is not in creating")
	}

	return nil
}

// Delete soft delete a CDCChangefeed
func (r *changefeedRepoImp) Delete(
	ctx context.Context,
	clusterId, changefeedId string,
) (*entity.CDCChangefeed, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "changefeedRepo.Delete")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	var changefeed entity.CDCChangefeed
	if err := wdb.Where("id = ? and cluster_id = ? and state != ?", changefeedId, clusterId, constant.ChangefeedStateDeleted).
		First(&changefeed).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to find cdc changefeed for delete")
	}

	updateColumns := map[string]any{
		"state":      constant.ChangefeedStateDeleted,
		"deleted_at": time.Now().UTC(),
	}

	result := wdb.Model(&changefeed).
		Where("state != ?", constant.ChangefeedStateDeleted).
		Updates(updateColumns)
	if result.Error != nil {
		return nil, bizerr.WrapGormError(result.Error, "fail to delete cdc changefeed")
	}
	if result.RowsAffected == 0 {
		return nil, bizerr.New(errorcode.AffectedNoRows, "")
	}
	changefeed.State = constant.ChangefeedStateDeleted
	return &changefeed, nil
}

//nolint:gocritic
func buildCDCChangefeedFilter(filter entity.ListChangefeedsFilter, db *gorm.DB) *gorm.DB {
	db = db.Where("state != ?", constant.ChangefeedStateDeleted)
	if filter.ClusterID != "" {
		db = db.Where("cluster_id = ?", filter.ClusterID)
	}
	if filter.CloudProvider != "" {
		db = db.Where("cloud_provider = ?", filter.CloudProvider)
	}
	if filter.Region != "" {
		db = db.Where("region = ?", filter.Region)
	}
	if filter.Kubernetes != "" {
		db = db.Where("kubernetes = ?", filter.Kubernetes)
	}
	if filter.Arch != "" {
		db = db.Where("arch = ?", filter.Arch)
	}
	if filter.State != "" {
		db = db.Where("state = ?", filter.State)
	}
	if filter.SinkType != "" {
		db = db.Where("sink_type = ?", filter.SinkType)
	}
	if filter.Offset != nil {
		db = db.Offset(int(*filter.Offset))
	}
	if filter.Limit != nil {
		db = db.Limit(int(*filter.Limit))
	}
	if filter.OrderBy != "" {
		db = db.Order(filter.OrderBy)
	}
	if len(filter.StatesNotIn) > 0 {
		db = db.Where("state NOT IN ?", filter.StatesNotIn)
	}
	if len(filter.StatesIn) > 0 {
		db = db.Where("state IN ?", filter.StatesIn)
	}
	if filter.CheckpointTSOLT != 0 {
		db = db.Where("checkpoint_tso < ?", filter.CheckpointTSOLT)
	}
	if filter.FailedTimeGTE != nil {
		db = db.Where("failed_time >= ?", *filter.FailedTimeGTE)
	}
	return db
}

//nolint:gocritic
func buildCDCChangefeedCountFilter(filter entity.ListChangefeedsFilter, db *gorm.DB) *gorm.DB {
	db = db.Where("state != ?", constant.ChangefeedStateDeleted)
	if filter.ClusterID != "" {
		db = db.Where("cluster_id = ?", filter.ClusterID)
	}
	if filter.CloudProvider != "" {
		db = db.Where("cloud_provider = ?", filter.CloudProvider)
	}
	if filter.Region != "" {
		db = db.Where("region = ?", filter.Region)
	}
	if filter.Kubernetes != "" {
		db = db.Where("kubernetes = ?", filter.Kubernetes)
	}
	if filter.Arch != "" {
		db = db.Where("arch = ?", filter.Arch)
	}
	if filter.State != "" {
		db = db.Where("state = ?", filter.State)
	}
	if filter.SinkType != "" {
		db = db.Where("sink_type = ?", filter.SinkType)
	}
	return db
}

func (r *changefeedRepoImp) Save(
	ctx context.Context,
	cdc *entity.CDCChangefeed,
) (*entity.CDCChangefeed, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "changefeedRepo.Save")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)

	// TODO only allowed to edit in paused state
	if err := wdb.Save(cdc).Error; err != nil {
		return cdc, bizerr.WrapGormError(err, "fail to save cdc changefeed")
	}
	return cdc, nil
}

func (r *changefeedRepoImp) CleanByRetention(ctx context.Context, retentionDays int) (int64, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "changefeedRepo.CleanByRetention")
	defer func() { tracing.EndSpan(span, nil) }()

	curTime := time.Now()
	cleanTime := curTime.AddDate(0, 0, -retentionDays)
	wdb := r.dbProvider.WriteDB(ctx)
	resp := wdb.Where("deleted_at < ? and state = ?", cleanTime, constant.ChangefeedStateDeleted).
		Delete(&entity.CDCChangefeed{})
	if resp.Error != nil {
		return 0, bizerr.WrapGormError(resp.Error, "fail to clean cdc changefeeds")
	}
	return resp.RowsAffected, nil
}

func (r *changefeedRepoImp) GroupByClusterId(
	ctx context.Context,
) ([]*entity.ChangefeedCountByCluster, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "changefeedRepo.GroupByClusterId")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	var result []*entity.ChangefeedCountByCluster
	if err := rdb.Model(&entity.CDCChangefeed{}).
		Select("cluster_id, count(*) as count").
		Where("state != ?", constant.ChangefeedStateDeleted).
		Group("cluster_id").
		Scan(&result).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to count by state")
	}
	return result, nil
}

func (r *changefeedRepoImp) GroupByState(ctx context.Context) (_ []*entity.ChangefeedStateCount, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "changefeedRepo.GroupByState")
	defer func() { tracing.EndSpan(span, err) }()

	rdb := r.dbProvider.ReadDB(ctx)
	var result []*entity.ChangefeedStateCount
	if err := rdb.Model(&entity.CDCChangefeed{}).Select("state, count(*) as count").Group("state").Scan(&result).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to count by state")
	}
	return result, nil
}

type cdcClusterRepoImp struct {
	dbProvider db.TransProvider
}

func NewCDCClusterRepoImp(dbProvider db.TransProvider) repo.CDCClusterRepo {
	return &cdcClusterRepoImp{dbProvider}
}

func (r *cdcClusterRepoImp) Create(ctx context.Context, cluster *entity.CDCCluster) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "cdcClusterRepo.Create")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Create(cluster).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to create cdc cluster")
	}
	return nil
}

func (r *cdcClusterRepoImp) GetById(ctx context.Context, clusterId string) (*entity.CDCCluster, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "cdcClusterRepo.GetById")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	cluster := &entity.CDCCluster{}
	if err := rdb.Where("cluster_id = ?", clusterId).First(cluster).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to get cdc cluster")
	}
	return cluster, nil
}

func (r *cdcClusterRepoImp) Delete(ctx context.Context, clusterId string) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "cdcClusterRepo.Delete")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Where("cluster_id = ? and state = ?", clusterId, constant.CDCClusterStateDeleted).Delete(&entity.CDCCluster{}).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to delete cdc cluster")
	}
	return nil
}

func (r *cdcClusterRepoImp) IncreaseOrCreateChangefeed(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
) (int64, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "cdcClusterRepo.IncreaseOrCreateChangefeed")
	defer func() { tracing.EndSpan(span, nil) }()

	cdcCluster := &entity.CDCCluster{
		ClusterID:       changefeed.ClusterID,
		TenantID:        changefeed.TenantID,
		CloudProvider:   changefeed.CloudProvider,
		Region:          changefeed.Region,
		Kubernetes:      changefeed.Kubernetes,
		KeySpaceID:      changefeed.KeySpaceID,
		ChangefeedCount: 1,
		State:           constant.CDCClusterStateCreating,
		ClusterStatusId: 0,
	}

	wdb := r.dbProvider.WriteDB(ctx)
	result := wdb.Clauses(clause.OnConflict{
		Columns:   []clause.Column{{Name: "cluster_id"}},
		DoUpdates: clause.Assignments(map[string]any{"changefeed_count": gorm.Expr("changefeed_count + 1")}),
	}).Create(&cdcCluster)

	if result.Error != nil {
		return 0, bizerr.WrapGormError(result.Error, "fail to increase or create changefeed_count")
	}
	return result.RowsAffected, nil
}

func (r *cdcClusterRepoImp) DecreaseChangefeed(ctx context.Context, clusterId string) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "cdcClusterRepo.DecreaseChangefeed")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	result := wdb.Model(&entity.CDCCluster{}).
		Where("cluster_id = ? AND changefeed_count > 0", clusterId).
		Update("changefeed_count", gorm.Expr("changefeed_count - 1"))
	if result.Error != nil {
		return bizerr.WrapGormError(result.Error, "fail to decrease changefeed_count")
	}
	if result.RowsAffected == 0 {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "no cdc cluster found to decrease changefeed_count")
	}
	return nil
}

func (r *cdcClusterRepoImp) ListCDCWithoutChangefeed(ctx context.Context) ([]*entity.CDCCluster, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "cdcClusterRepo.ListCDCWithoutChangefeed")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	var clusters []*entity.CDCCluster
	if err := rdb.Where("changefeed_count <= 0 and state != ?", constant.CDCClusterStateDeleted).Find(&clusters).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list cdc clusters without changefeed")
	}
	return clusters, nil
}

func (r *cdcClusterRepoImp) SetClusterStatusId(ctx context.Context, clusterId string, statusId uint64) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "cdcClusterRepo.SetClusterStatusId")
	defer func() { tracing.EndSpan(span, nil) }()

	updateColumns := map[string]any{
		"cluster_status_id": statusId,
	}

	wdb := r.dbProvider.WriteDB(ctx)
	result := wdb.Model(&entity.CDCCluster{}).
		Where("cluster_id = ?", clusterId).
		Updates(updateColumns)
	if result.Error != nil {
		return bizerr.WrapGormError(result.Error, "fail to set cluster status id")
	}
	if result.RowsAffected == 0 {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "no cdc cluster found to set cluster status id")
	}
	return nil
}

func (r *cdcClusterRepoImp) UpdateState(
	ctx context.Context,
	clusterId string,
	state constant.CDCClusterState,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "cdcClusterRepo.UpdateClusterStatus")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	result := wdb.Model(&entity.CDCCluster{}).
		Where("cluster_id = ?", clusterId).
		Update("state", state)
	if result.Error != nil {
		return bizerr.WrapGormError(result.Error, "fail to update state")
	}
	if result.RowsAffected == 0 {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "no cdc cluster found to update state")
	}
	return nil
}

func (r *cdcClusterRepoImp) CleanByRetention(ctx context.Context, retentionDays int) (int64, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "cdcClusterRepo.CleanByRetention")
	defer func() { tracing.EndSpan(span, nil) }()

	curTime := time.Now()
	cleanTime := curTime.AddDate(0, 0, -retentionDays)
	wdb := r.dbProvider.WriteDB(ctx)
	resp := wdb.Where("state = ? and updated_at < ? and changefeed_count = 0", constant.CDCClusterStateDeleted, cleanTime).
		Delete(&entity.CDCCluster{})
	if resp.Error != nil {
		return 0, bizerr.WrapGormError(resp.Error, "fail to clean cdc cluster")
	}
	return resp.RowsAffected, nil
}