package data_pipeline

import (
	"context"

	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/pkg/id_generator"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
)

var _ DataPipelineRepo = (*dataPipelineRepoImpl)(nil)

type dataPipelineRepoImpl struct {
	dbProvider  db.Provider
	idGenerator id_generator.IDGenerator
}

func NewDataPipelineRepo(
	dbProvider db.Provider,
	idGenerator id_generator.IDGenerator,
) DataPipelineRepo {
	return &dataPipelineRepoImpl{
		dbProvider:  dbProvider,
		idGenerator: idGenerator,
	}
}

func (r *dataPipelineRepoImpl) AllocateID() entity.ID {
	return entity.ID(r.idGenerator.NextID())
}

func (r *dataPipelineRepoImpl) Create(
	ctx context.Context,
	pipeline *entity.DataPipeline,
) bizerr.BizError {
	if pipeline.ID == 0 {
		pipeline.ID = r.AllocateID()
	}
	if err := r.dbProvider.WriteDB(ctx).Create(pipeline).Error; err != nil {
		return bizerr.WrapGormError(err, "create data_pipeline")
	}
	return nil
}

func (r *dataPipelineRepoImpl) Load(
	ctx context.Context,
	id uint64,
) (*entity.DataPipeline, bizerr.BizError) {
	pipeline := new(entity.DataPipeline)
	if err := entity.NewDataPipelineQuerySet(r.dbProvider.ReadDB(ctx)).
		IDEq(entity.ID(id)).
		One(pipeline); err != nil {
		return nil, bizerr.WrapGormError(err, "load data_pipeline %d", id)
	}
	return pipeline, nil
}

func (r *dataPipelineRepoImpl) Get(
	ctx context.Context,
	ops ...FilterOp,
) (*entity.DataPipeline, bizerr.BizError) {
	querySet := applyFilterOps(entity.NewDataPipelineQuerySet(r.dbProvider.ReadDB(ctx)), ops)
	pipeline := new(entity.DataPipeline)
	if err := querySet.One(pipeline); err != nil {
		return nil, bizerr.WrapGormError(err, "get data_pipeline")
	}
	return pipeline, nil
}

func (r *dataPipelineRepoImpl) List(
	ctx context.Context,
	ops ...FilterOp,
) ([]entity.DataPipeline, bizerr.BizError) {
	querySet := applyFilterOps(entity.NewDataPipelineQuerySet(r.dbProvider.ReadDB(ctx)), ops)
	var pipelines []entity.DataPipeline
	if err := querySet.All(&pipelines); err != nil {
		return nil, bizerr.WrapGormError(err, "list data_pipelines")
	}
	return pipelines, nil
}

func (r *dataPipelineRepoImpl) Count(
	ctx context.Context,
	ops ...FilterOp,
) (int64, bizerr.BizError) {
	querySet := applyFilterOps(entity.NewDataPipelineQuerySet(r.dbProvider.ReadDB(ctx)), ops)
	count, err := querySet.Offset(-1).Limit(-1).Count()
	if err != nil {
		return 0, bizerr.WrapGormError(err, "count data_pipelines")
	}
	return count, nil
}

func (r *dataPipelineRepoImpl) Save(
	ctx context.Context,
	pipeline *entity.DataPipeline,
) bizerr.BizError {
	if err := entity.SaveWithVersioning(r.dbProvider.WriteDB(ctx), pipeline); err != nil {
		return err.Chain("save data_pipeline %d", pipeline.ID)
	}
	return nil
}

func (r *dataPipelineRepoImpl) Delete(ctx context.Context, id uint64) bizerr.BizError {
	if err := r.dbProvider.WriteDB(ctx).Delete(&entity.DataPipeline{}, entity.ID(id)).Error; err != nil {
		return bizerr.WrapGormError(err, "delete data_pipeline %d", id)
	}
	return nil
}

func applyFilterOps(
	querySet entity.DataPipelineQuerySet,
	ops []FilterOp,
) entity.DataPipelineQuerySet {
	for _, op := range ops {
		querySet = op(querySet)
	}
	return querySet
}