package ticdc

import (
	"context"
	"fmt"
	"unicode/utf8"

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

type tiCDCChangefeedRepoImpl struct {
	dbProvider db.Provider
}

const (
	ticdcChangefeedKernelErrorMessageHeadBytes = 2048
	ticdcChangefeedKernelErrorMessageTailBytes = 2048
)

func NewTiCDCChangefeedRepo(dbProvider db.Provider) repo.TiCDCChangefeedRepo {
	return &tiCDCChangefeedRepoImpl{dbProvider: dbProvider}
}

func (b *tiCDCChangefeedRepoImpl) Load(
	ctx context.Context, id uint64,
) (*entity.TiCDCChangefeed, bizerr.BizError) {
	rdb := b.dbProvider.ReadDB(ctx)
	record := &entity.TiCDCChangefeed{}
	err := entity.NewTiCDCChangefeedQuerySet(rdb).
		IDEq(entity.ID(id)).
		One(record)
	if err != nil {
		return nil, bizerr.WrapGormError(err, "load ticdc cluster changefeed record occur error")
	}
	return record, nil
}

func (b *tiCDCChangefeedRepoImpl) Save(
	ctx context.Context, record *entity.TiCDCChangefeed,
) bizerr.BizError {
	truncateTiCDCChangefeedKernelErrorMessage(record)
	wdb := b.dbProvider.WriteDB(ctx)
	if err := entity.SaveWithVersioning(wdb, record); err != nil {
		return err.Chain("save record, id=%d", record.ID)
	}
	return nil
}

func truncateTiCDCChangefeedKernelErrorMessage(record *entity.TiCDCChangefeed) {
	if record == nil || record.KernelMetrics.RunningError == nil {
		return
	}
	message := record.KernelMetrics.RunningError.Message
	if len(message) <= ticdcChangefeedKernelErrorMessageHeadBytes+ticdcChangefeedKernelErrorMessageTailBytes {
		return
	}
	record.KernelMetrics.RunningError.Message = truncateStringByHeadAndTailBytes(message)
}

func truncateStringByHeadAndTailBytes(s string) string {
	head := truncateStringHeadByBytes(s, ticdcChangefeedKernelErrorMessageHeadBytes)
	tail := truncateStringTailByBytes(s, ticdcChangefeedKernelErrorMessageTailBytes)
	return fmt.Sprintf("%s\n...[truncated %d bytes]...\n%s", head, len(s)-len(head)-len(tail), tail)
}

func truncateStringHeadByBytes(s string, maxBytes int) string {
	if len(s) <= maxBytes {
		return s
	}
	end := 0
	for i, r := range s {
		next := i + utf8.RuneLen(r)
		if next > maxBytes {
			break
		}
		end = next
	}
	return s[:end]
}

func truncateStringTailByBytes(s string, maxBytes int) string {
	if len(s) <= maxBytes {
		return s
	}
	for i := range s {
		if len(s)-i <= maxBytes {
			return s[i:]
		}
	}
	return ""
}

func (b *tiCDCChangefeedRepoImpl) List(
	ctx context.Context, ops ...repo.TiCDCChangefeedFilterOp,
) ([]entity.TiCDCChangefeed, bizerr.BizError) {
	rdb := b.dbProvider.ReadDB(ctx)

	qs := entity.NewTiCDCChangefeedQuerySet(rdb)
	for _, op := range ops {
		qs = op(qs)
	}
	var records []entity.TiCDCChangefeed
	if err := qs.All(&records); err != nil {
		return nil, bizerr.WrapGormError(err, "list ticdc cluster changefeed records occur error")
	}
	return records, nil
}

func (b *tiCDCChangefeedRepoImpl) Get(
	ctx context.Context, ops ...repo.TiCDCChangefeedFilterOp,
) (*entity.TiCDCChangefeed, bizerr.BizError) {
	rdb := b.dbProvider.ReadDB(ctx)

	qs := entity.NewTiCDCChangefeedQuerySet(rdb)
	for _, op := range ops {
		qs = op(qs)
	}
	record := &entity.TiCDCChangefeed{}
	if err := qs.One(record); err != nil {
		return nil, bizerr.WrapGormError(err, "get ticdc cluster changefeed record occur error")
	}
	return record, nil
}

func (b *tiCDCChangefeedRepoImpl) Count(
	ctx context.Context, ops ...repo.TiCDCChangefeedFilterOp,
) (int64, bizerr.BizError) {
	rdb := b.dbProvider.ReadDB(ctx)

	qs := entity.NewTiCDCChangefeedQuerySet(rdb)
	for _, op := range ops {
		qs = op(qs)
	}
	count, err := qs.Offset(-1).Limit(-1).Count()
	if err != nil {
		return 0, bizerr.WrapGormError(err, "count ticdc cluster changefeed record occur error")
	}
	return count, nil
}