package biz_changefeed

import (
	"context"
	"testing"

	"github.com/stretchr/testify/require"
	apprepo "github.com/tidbcloud/dataflow-service-ng/app/repo"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/pkgv2/db"
	"gorm.io/gorm"
)

type fixedIDGenerator struct{ next uint64 }

func (g *fixedIDGenerator) NextID() uint64 {
	g.next++
	return g.next
}

func prepareRepoTest(t *testing.T) (*gorm.DB, *fixedIDGenerator, *BizChangefeedRepoImpl) {
	t.Helper()
	testDB, err := apprepo.GetFakeDB()
	require.NoError(t, err)
	require.NoError(t, testDB.Exec("DELETE FROM biz_changefeeds").Error)
	t.Cleanup(func() {
		require.NoError(t, testDB.Exec("DELETE FROM biz_changefeeds").Error)
	})
	generator := &fixedIDGenerator{next: 9000}
	return testDB, generator, &BizChangefeedRepoImpl{
		dbProvider: db.NewProvider(testDB), idGenerator: generator,
	}
}

// TestCreateKeepsACallerChosenIdentity covers what makes an orchestrated create
// idempotent: the caller picks the id before the write, so a create whose
// response was lost is resolvable with a Get on a key it already holds. If the
// repo allocated its own instead, that key would name nothing.
func TestCreateKeepsACallerChosenIdentity(t *testing.T) {
	_, generator, changefeedRepo := prepareRepoTest(t)
	ctx := context.Background()

	const chosen = uint64(9007199254740993)
	chosenChangefeed := &entity.BizChangefeed{
		ID: entity.ID(chosen), Name: "chosen", ClusterID: 1, TenantID: 2,
		DownstreamType:  entity.TiCDCDownStreamTypeS3,
		NetworkType:     entity.ChangefeedDownstreamNetworkTypeInternet,
		OperationStatus: entity.BizChangefeedOperationStatusCreating,
	}
	require.NoError(t, changefeedRepo.Create(ctx, chosenChangefeed))
	require.Equal(t, chosen, chosenChangefeed.ID.Uint64())

	loaded, err := changefeedRepo.Load(ctx, chosen)
	require.NoError(t, err)
	require.Equal(t, "chosen", loaded.Name)
	require.Equal(t, uint64(9000), generator.next, "the generator must not have been consulted")

	// Replaying the same create collides on the primary key rather than
	// quietly producing a second changefeed.
	require.Error(t, changefeedRepo.Create(ctx, &entity.BizChangefeed{
		ID: entity.ID(chosen), Name: "replay", ClusterID: 1, TenantID: 2,
		DownstreamType:  entity.TiCDCDownStreamTypeS3,
		NetworkType:     entity.ChangefeedDownstreamNetworkTypeInternet,
		OperationStatus: entity.BizChangefeedOperationStatusCreating,
	}))

	// Leaving it unset still allocates, which is what every other caller does.
	allocated := &entity.BizChangefeed{
		Name: "allocated", ClusterID: 1, TenantID: 2,
		DownstreamType:  entity.TiCDCDownStreamTypeS3,
		NetworkType:     entity.ChangefeedDownstreamNetworkTypeInternet,
		OperationStatus: entity.BizChangefeedOperationStatusCreating,
	}
	require.NoError(t, changefeedRepo.Create(ctx, allocated))
	require.Equal(t, uint64(9001), allocated.ID.Uint64())
}