package biz_changefeed

import (
	"context"

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

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

func NewBizChangefeedRepo(dbProvider db.Provider, idGenerator id_generator.IDGenerator) repo.BizChangefeedRepo {
	return &BizChangefeedRepoImpl{
		dbProvider:  dbProvider,
		idGenerator: idGenerator,
	}
}

func (m *BizChangefeedRepoImpl) Load(ctx context.Context, id uint64) (*entity.BizChangefeed, bizerr.BizError) {
	rdb := m.dbProvider.ReadDB(ctx)
	bizChangefeed := new(entity.BizChangefeed)
	if err := entity.NewBizChangefeedQuerySet(rdb).IDEq(entity.ID(id)).One(bizChangefeed); err != nil {
		return nil, bizerr.WrapGormError(err, "load biz_changefeed error")
	}
	return bizChangefeed, nil
}

func (m *BizChangefeedRepoImpl) List(
	ctx context.Context, ops ...repo.BizChangefeedFilterOp,
) ([]entity.BizChangefeed, bizerr.BizError) {
	rdb := m.dbProvider.ReadDB(ctx)
	qs := entity.NewBizChangefeedQuerySet(rdb)
	for _, op := range ops {
		qs = op(qs)
	}

	var bizChangefeeds []entity.BizChangefeed
	if err := qs.All(&bizChangefeeds); err != nil {
		return nil, bizerr.WrapGormError(err, "list biz_changefeeds error")
	}
	return bizChangefeeds, nil
}

func (m *BizChangefeedRepoImpl) ListWithSoftDelete(
	ctx context.Context, ops ...repo.BizChangefeedFilterOp,
) ([]entity.BizChangefeed, bizerr.BizError) {
	rdb := m.dbProvider.ReadDB(ctx).Unscoped()
	qs := entity.NewBizChangefeedQuerySet(rdb)
	for _, op := range ops {
		qs = op(qs)
	}

	var bizChangefeeds []entity.BizChangefeed
	if err := qs.All(&bizChangefeeds); err != nil {
		return nil, bizerr.WrapGormError(err, "list biz_changefeeds error")
	}
	return bizChangefeeds, nil
}

func (m *BizChangefeedRepoImpl) Count(
	ctx context.Context, ops ...repo.BizChangefeedFilterOp,
) (int64, bizerr.BizError) {
	rdb := m.dbProvider.ReadDB(ctx)
	qs := entity.NewBizChangefeedQuerySet(rdb)
	for _, op := range ops {
		qs = op(qs)
	}

	total, err := qs.Count()
	if err != nil {
		return 0, bizerr.WrapGormError(err, "count biz_changefeeds error")
	}
	return total, nil
}

func (m *BizChangefeedRepoImpl) Create(ctx context.Context, bizChangefeed *entity.BizChangefeed) bizerr.BizError {
	// the migrated cdcTask use the cdc task id as the changefeed id
	if bizChangefeed.ID == 0 {
		bizChangefeed.ID = entity.ID(m.idGenerator.NextID())
	}
	wdb := m.dbProvider.WriteDB(ctx)
	if err := wdb.Create(bizChangefeed).Error; err != nil {
		return bizerr.WrapGormError(err, "create biz_changefeed error")
	}
	return nil
}

func (m *BizChangefeedRepoImpl) Update(ctx context.Context, bizChangefeed *entity.BizChangefeed) bizerr.BizError {
	wdb := m.dbProvider.WriteDB(ctx)
	if err := entity.SaveWithVersioning(wdb, bizChangefeed); err != nil {
		return bizerr.WrapGormError(err, "update biz_changefeed error")
	}
	return nil
}

func (m *BizChangefeedRepoImpl) Delete(ctx context.Context, id uint64) bizerr.BizError {
	wdb := m.dbProvider.WriteDB(ctx)
	if err := wdb.Delete(&entity.BizChangefeed{}, id).Error; err != nil {
		return bizerr.WrapGormError(err, "delete biz_changefeed error")
	}
	return nil
}

func (m *BizChangefeedRepoImpl) Get(
	ctx context.Context, ops ...repo.BizChangefeedFilterOp,
) (*entity.BizChangefeed, bizerr.BizError) {
	rdb := m.dbProvider.ReadDB(ctx)
	qs := entity.NewBizChangefeedQuerySet(rdb)
	for _, op := range ops {
		qs = op(qs)
	}

	bizChangefeed := new(entity.BizChangefeed)
	if err := qs.One(bizChangefeed); err != nil {
		return nil, bizerr.WrapGormError(err, "get biz_changefeeds error")
	}
	return bizChangefeed, nil
}