package server

import (
	"context"
	"fmt"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	executormocks "github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor/mocks"
	repomocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

func TestApplySemConfig(t *testing.T) {
	t.Run("should fail when shared pool id is empty", func(t *testing.T) {
		s := &serverImpl{}
		err := s.ApplySemConfig(context.Background(), "")
		require.Error(t, err)
		st, ok := status.FromError(err)
		assert.True(t, ok)
		assert.Equal(t, codes.InvalidArgument, st.Code())
	})

	t.Run("should fail when a duplicate dag is already running", func(t *testing.T) {
		dagExecutor := executormocks.NewRemoteExecutor(t)

		running := executor.NewExecTaskIdStatus()
		running["existing-exec-id"] = &executor.ExecTaskStatusEx{}
		dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanApplySemConfig, mock.Anything).Return(running, nil).Once()

		s := &serverImpl{
			dagExecutor: dagExecutor,
		}
		err := s.ApplySemConfig(context.Background(), "sp-1")
		require.Error(t, err)
		st, ok := status.FromError(err)
		assert.True(t, ok)
		assert.Equal(t, codes.AlreadyExists, st.Code())
	})

	t.Run("should fail when shared pool does not exist", func(t *testing.T) {
		dagExecutor := executormocks.NewRemoteExecutor(t)
		sharedPoolRepo := repomocks.NewSharedPoolRepo(t)

		dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanApplySemConfig, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Once()
		sharedPoolRepo.On("Load", mock.Anything, "sp-not-found", false).Return(nil, fmt.Errorf("not found")).Once()

		s := &serverImpl{
			dagExecutor:    dagExecutor,
			sharedPoolRepo: sharedPoolRepo,
		}
		err := s.ApplySemConfig(context.Background(), "sp-not-found")
		require.Error(t, err)
		assert.Contains(t, err.Error(), "failed to load shared pool")
	})

	t.Run("should succeed and submit dag", func(t *testing.T) {
		dagExecutor := executormocks.NewRemoteExecutor(t)
		sharedPoolRepo := repomocks.NewSharedPoolRepo(t)

		dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanApplySemConfig, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Once()
		sharedPoolRepo.On("Load", mock.Anything, "sp-1", false).Return(&entity.SharedPool{}, nil).Once()
		dagExecutor.On("Submit", mock.Anything, dag.PlanApplySemConfig, mock.MatchedBy(func(args actions.ActionArgs) bool {
			spId, err := args.GetString(workflow.SharedPoolIdKey)
			return err == nil && spId == "sp-1"
		}), mock.AnythingOfType("string")).Return(nil).Once()

		s := &serverImpl{
			dagExecutor:    dagExecutor,
			sharedPoolRepo: sharedPoolRepo,
		}
		err := s.ApplySemConfig(context.Background(), "sp-1")
		require.NoError(t, err)
		dagExecutor.AssertExpectations(t)
	})

	t.Run("should fail when dag submit fails", func(t *testing.T) {
		dagExecutor := executormocks.NewRemoteExecutor(t)
		sharedPoolRepo := repomocks.NewSharedPoolRepo(t)

		dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanApplySemConfig, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Once()
		sharedPoolRepo.On("Load", mock.Anything, "sp-1", false).Return(&entity.SharedPool{}, nil).Once()
		dagExecutor.On("Submit", mock.Anything, dag.PlanApplySemConfig, mock.MatchedBy(func(args actions.ActionArgs) bool {
			spId, err := args.GetString(workflow.SharedPoolIdKey)
			return err == nil && spId == "sp-1"
		}), mock.AnythingOfType("string")).Return(fmt.Errorf("submit error")).Once()

		s := &serverImpl{
			dagExecutor:    dagExecutor,
			sharedPoolRepo: sharedPoolRepo,
		}
		err := s.ApplySemConfig(context.Background(), "sp-1")
		require.Error(t, err)
		assert.Contains(t, err.Error(), "failed to submit apply sem config dag")
	})
}