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/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
)

type tiCDCChangefeedResourceRepoImp struct {
	dbProvider db.TransProvider
}

func NewTiCDCChangefeedResourceRepoImp(dbProvider db.TransProvider) repo.TiCDCChangefeedResourceRepo {
	return &tiCDCChangefeedResourceRepoImp{dbProvider: dbProvider}
}

func (r *tiCDCChangefeedResourceRepoImp) Save(
	ctx context.Context,
	resource *entity.TiCDCChangefeedResource,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "tiCDCChangefeedResourceRepo.Save")
	defer func() { tracing.EndSpan(span, nil) }()

	if err := r.dbProvider.WriteDB(ctx).Save(resource).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to save ticdc changefeed resource")
	}
	return nil
}

func (r *tiCDCChangefeedResourceRepoImp) GetByChangefeedID(
	ctx context.Context,
	clusterID, changefeedID string,
) (*entity.TiCDCChangefeedResource, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "tiCDCChangefeedResourceRepo.GetByChangefeedID")
	defer func() { tracing.EndSpan(span, nil) }()

	resource := &entity.TiCDCChangefeedResource{}
	if err := r.dbProvider.ReadDB(ctx).
		Where("cluster_id = ? AND changefeed_id = ?", clusterID, changefeedID).
		First(resource).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to get ticdc changefeed resource")
	}
	return resource, nil
}

func (r *tiCDCChangefeedResourceRepoImp) ListByClusterID(
	ctx context.Context,
	clusterID string,
) ([]*entity.TiCDCChangefeedResource, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "tiCDCChangefeedResourceRepo.ListByClusterID")
	defer func() { tracing.EndSpan(span, nil) }()

	var resources []*entity.TiCDCChangefeedResource
	if err := r.dbProvider.ReadDB(ctx).
		Where("cluster_id = ?", clusterID).
		Find(&resources).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list ticdc changefeed resources by cluster")
	}
	return resources, nil
}

func (r *tiCDCChangefeedResourceRepoImp) ListByShard(
	ctx context.Context,
	provider, region, kubernetes, shareScope, shareKey string,
) ([]*entity.TiCDCChangefeedResource, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "tiCDCChangefeedResourceRepo.ListByShard")
	defer func() { tracing.EndSpan(span, nil) }()

	var resources []*entity.TiCDCChangefeedResource
	if err := r.dbProvider.ReadDB(ctx).
		Where(
			"cloud_provider = ? AND region = ? AND kubernetes = ? AND share_scope = ? AND share_key = ?",
			provider, region, kubernetes, shareScope, shareKey,
		).
		Find(&resources).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list ticdc changefeed resources by shard")
	}
	return resources, nil
}

func (r *tiCDCChangefeedResourceRepoImp) ListNeedingSync(
	ctx context.Context,
) ([]*entity.TiCDCChangefeedResource, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "tiCDCChangefeedResourceRepo.ListNeedingSync")
	defer func() { tracing.EndSpan(span, nil) }()

	var resources []*entity.TiCDCChangefeedResource
	if err := r.dbProvider.ReadDB(ctx).
		Where("expected_specification <> actual_specification").
		Find(&resources).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list ticdc changefeed resources needing sync")
	}
	return resources, nil
}

func (r *tiCDCChangefeedResourceRepoImp) DeleteStaleReleasedRows(
	ctx context.Context, staleAge time.Duration,
) (int64, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "tiCDCChangefeedResourceRepo.DeleteStaleReleasedRows")
	defer func() { tracing.EndSpan(span, nil) }()

	cutoff := time.Now().UTC().Add(-staleAge)
	result := r.dbProvider.WriteDB(ctx).
		Where("expected_specification = '' AND actual_specification = '' AND updated_at < ?", cutoff).
		Delete(&entity.TiCDCChangefeedResource{})
	if result.Error != nil {
		return 0, bizerr.WrapGormError(result.Error, "fail to delete stale ticdc changefeed resources")
	}
	return result.RowsAffected, nil
}

type tiCDCClusterRepoImp struct {
	dbProvider db.TransProvider
}

func NewTiCDCClusterRepoImp(dbProvider db.TransProvider) repo.TiCDCClusterRepo {
	return &tiCDCClusterRepoImp{dbProvider: dbProvider}
}

func (r *tiCDCClusterRepoImp) Save(
	ctx context.Context,
	cluster *entity.TiCDCCluster,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "tiCDCClusterRepo.Save")
	defer func() { tracing.EndSpan(span, nil) }()

	if err := r.dbProvider.WriteDB(ctx).Save(cluster).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to save ticdc cluster")
	}
	return nil
}

func (r *tiCDCClusterRepoImp) GetByShard(
	ctx context.Context,
	provider, region, kubernetes, shareScope, shareKey string,
) (*entity.TiCDCCluster, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "tiCDCClusterRepo.GetByShard")
	defer func() { tracing.EndSpan(span, nil) }()

	cluster := &entity.TiCDCCluster{}
	if err := r.dbProvider.ReadDB(ctx).
		Where(
			"cloud_provider = ? AND region = ? AND kubernetes = ? AND share_scope = ? AND share_key = ?",
			provider, region, kubernetes, shareScope, shareKey,
		).
		First(cluster).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to get ticdc cluster by shard")
	}
	return cluster, nil
}

func (r *tiCDCClusterRepoImp) List(
	ctx context.Context,
	provider, region, state string,
) ([]*entity.TiCDCCluster, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "tiCDCClusterRepo.List")
	defer func() { tracing.EndSpan(span, nil) }()

	q := r.dbProvider.ReadDB(ctx).Model(&entity.TiCDCCluster{})
	if provider != "" {
		q = q.Where("cloud_provider = ?", provider)
	}
	if region != "" {
		q = q.Where("region = ?", region)
	}
	if state != "" {
		q = q.Where("state = ?", state)
	}
	var clusters []*entity.TiCDCCluster
	if err := q.Order("id ASC").Find(&clusters).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list ticdc clusters")
	}
	return clusters, nil
}

type kernelTiCDCChangefeedRepoImp struct {
	dbProvider db.TransProvider
}

func NewKernelTiCDCChangefeedRepoImp(dbProvider db.TransProvider) repo.KernelTiCDCChangefeedRepo {
	return &kernelTiCDCChangefeedRepoImp{dbProvider: dbProvider}
}

func (r *kernelTiCDCChangefeedRepoImp) Save(
	ctx context.Context,
	changefeed *entity.KernelTiCDCChangefeed,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "kernelTiCDCChangefeedRepo.Save")
	defer func() { tracing.EndSpan(span, nil) }()

	if err := r.dbProvider.WriteDB(ctx).Save(changefeed).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to save kernel ticdc changefeed")
	}
	return nil
}

func (r *kernelTiCDCChangefeedRepoImp) GetByChangefeedID(
	ctx context.Context,
	clusterID, changefeedID string,
) (*entity.KernelTiCDCChangefeed, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "kernelTiCDCChangefeedRepo.GetByChangefeedID")
	defer func() { tracing.EndSpan(span, nil) }()

	changefeed := &entity.KernelTiCDCChangefeed{}
	if err := r.dbProvider.ReadDB(ctx).
		Where("cluster_id = ? AND changefeed_id = ?", clusterID, changefeedID).
		First(changefeed).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to get kernel ticdc changefeed")
	}
	return changefeed, nil
}

func (r *kernelTiCDCChangefeedRepoImp) ListByClusterID(
	ctx context.Context,
	clusterID string,
) ([]*entity.KernelTiCDCChangefeed, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "kernelTiCDCChangefeedRepo.ListByClusterID")
	defer func() { tracing.EndSpan(span, nil) }()

	var changefeeds []*entity.KernelTiCDCChangefeed
	if err := r.dbProvider.ReadDB(ctx).
		Where("cluster_id = ?", clusterID).
		Find(&changefeeds).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list kernel ticdc changefeeds by cluster")
	}
	return changefeeds, nil
}