package handler

import (
	"context"
	"testing"
	"time"

	"github.com/cockroachdb/errors"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"google.golang.org/grpc"

	"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
	"github.com/milvus-io/milvus/internal/mocks/streamingnode/client/handler/mock_assignment"
	"github.com/milvus-io/milvus/internal/mocks/streamingnode/client/handler/mock_consumer"
	"github.com/milvus-io/milvus/internal/mocks/streamingnode/client/handler/mock_producer"
	"github.com/milvus-io/milvus/internal/mocks/util/streamingutil/service/mock_lazygrpc"
	"github.com/milvus-io/milvus/internal/mocks/util/streamingutil/service/mock_resolver"
	"github.com/milvus-io/milvus/internal/streamingnode/client/handler/consumer"
	"github.com/milvus-io/milvus/internal/streamingnode/client/handler/producer"
	"github.com/milvus-io/milvus/internal/util/streamingutil/service/contextutil"
	"github.com/milvus-io/milvus/internal/util/streamingutil/status"
	"github.com/milvus-io/milvus/pkg/v3/mocks/proto/mock_streamingpb"
	"github.com/milvus-io/milvus/pkg/v3/mocks/streaming/util/mock_types"
	"github.com/milvus-io/milvus/pkg/v3/proto/streamingpb"
	"github.com/milvus-io/milvus/pkg/v3/streaming/util/message/adaptor"
	"github.com/milvus-io/milvus/pkg/v3/streaming/util/options"
	"github.com/milvus-io/milvus/pkg/v3/streaming/util/types"
	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
)

func TestHandlerClient(t *testing.T) {
	assignment := &types.PChannelInfoAssigned{
		Channel: types.PChannelInfo{Name: "pchannel", Term: 1},
		Node:    types.StreamingNodeInfo{ServerID: 1, Address: "localhost"},
	}

	service := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeHandlerServiceClient](t)
	handlerServiceClient := mock_streamingpb.NewMockStreamingNodeHandlerServiceClient(t)
	handlerServiceClient.EXPECT().GetReplicateCheckpoint(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, grcr *streamingpb.GetReplicateCheckpointRequest, co ...grpc.CallOption) (*streamingpb.GetReplicateCheckpointResponse, error) {
		serverID, ok := contextutil.GetPickServerID(ctx)
		assert.True(t, ok)
		assert.Equal(t, serverID, assignment.Node.ServerID)
		return &streamingpb.GetReplicateCheckpointResponse{
			Checkpoint: &commonpb.ReplicateCheckpoint{
				ClusterId: "pchannel",
				Pchannel:  "pchannel",
				MessageId: nil,
				TimeTick:  0,
			},
		}, nil
	})
	service.EXPECT().GetService(mock.Anything).Return(handlerServiceClient, nil)
	rb := mock_resolver.NewMockBuilder(t)
	rb.EXPECT().Close().Run(func() {})
	w := mock_assignment.NewMockWatcher(t)
	w.EXPECT().Close().Run(func() {})

	p := mock_producer.NewMockProducer(t)
	p.EXPECT().Register(mock.Anything).Return()
	p.EXPECT().Close().RunAndReturn(func() {})
	c := mock_consumer.NewMockConsumer(t)
	c.EXPECT().Close().RunAndReturn(func() error { return nil })

	rebalanceTrigger := mock_types.NewMockAssignmentRebalanceTrigger(t)
	rebalanceTrigger.EXPECT().ReportAssignmentError(mock.Anything, mock.Anything, mock.Anything).Return(nil)

	pK := 0
	handler := &handlerClientImpl{
		lifetime:         typeutil.NewLifetime(),
		service:          service,
		rb:               rb,
		watcher:          w,
		rebalanceTrigger: rebalanceTrigger,
		newProducer: func(ctx context.Context, opts *producer.ProducerOptions, handler streamingpb.StreamingNodeHandlerServiceClient) (Producer, error) {
			serverID, ok := contextutil.GetPickServerID(ctx)
			assert.True(t, ok)
			assert.Equal(t, serverID, assignment.Node.ServerID)
			if pK == 0 {
				pK++
				return nil, status.NewUnmatchedChannelTerm("pchannel", 1, 2)
			}
			return p, nil
		},
		newConsumer: func(ctx context.Context, opts *consumer.ConsumerOptions, handlerClient streamingpb.StreamingNodeHandlerServiceClient) (Consumer, error) {
			serverID, ok := contextutil.GetPickServerID(ctx)
			assert.True(t, ok)
			assert.Equal(t, serverID, assignment.Node.ServerID)
			return c, nil
		},
	}
	ctx := context.Background()

	k := 0
	w.EXPECT().Get(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, s string) *types.PChannelInfoAssigned {
		if k == 0 {
			k++
			return nil
		}
		return assignment
	})
	w.EXPECT().Watch(mock.Anything, mock.Anything, mock.Anything).Return(nil)

	producer, err := handler.CreateProducer(ctx, &ProducerOptions{PChannel: "pchannel"})
	assert.NoError(t, err)
	assert.NotNil(t, producer)
	producer2, err := handler.CreateProducer(ctx, &ProducerOptions{PChannel: "pchannel"})
	assert.NoError(t, err)
	assert.NotNil(t, producer)
	producer3, err := handler.CreateProducer(ctx, &ProducerOptions{PChannel: "pchannel"})
	assert.NoError(t, err)
	assert.NotNil(t, producer3)
	producer.Close()
	producer2.Close()
	producer3.Close()

	rcp, err := handler.GetReplicateCheckpoint(ctx, "pchannel")
	assert.NoError(t, err)
	assert.NotNil(t, rcp)

	handler.GetLatestMVCCTimestampIfLocal(ctx, "pchannel")
	producer4, err := handler.CreateProducer(ctx, &ProducerOptions{PChannel: "pchannel"})
	assert.NoError(t, err)
	assert.NotNil(t, producer4)
	producer4.Close()

	consumer, err := handler.CreateConsumer(ctx, &ConsumerOptions{
		PChannel:      "pchannel",
		VChannel:      "vchannel",
		DeliverPolicy: options.DeliverPolicyAll(),
		DeliverFilters: []options.DeliverFilter{
			options.DeliverFilterTimeTickGT(10),
			options.DeliverFilterTimeTickGTE(10),
		},
		MessageHandler: make(adaptor.ChanMessageHandler),
	})
	assert.NoError(t, err)
	assert.NotNil(t, consumer)
	consumer.Close()

	service.EXPECT().Close().Return()
	handler.Close()
	producer, err = handler.CreateProducer(ctx, nil)
	assert.Error(t, err)
	assert.ErrorIs(t, err, ErrClientClosed)
	assert.Nil(t, producer)

	consumer, err = handler.CreateConsumer(ctx, nil)
	assert.Error(t, err)
	assert.ErrorIs(t, err, ErrClientClosed)
	assert.Nil(t, consumer)

	handler.GetLatestMVCCTimestampIfLocal(ctx, "pchannel")
}

func TestHandlerClient_GetSalvageCheckpoint(t *testing.T) {
	assignment := &types.PChannelInfoAssigned{
		Channel: types.PChannelInfo{Name: "pchannel", Term: 1},
		Node:    types.StreamingNodeInfo{ServerID: 1, Address: "localhost"},
	}

	service := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeHandlerServiceClient](t)
	rb := mock_resolver.NewMockBuilder(t)
	rb.EXPECT().Close().Run(func() {})
	w := mock_assignment.NewMockWatcher(t)
	w.EXPECT().Close().Run(func() {})
	// Always return the assignment so the create func is invoked.
	w.EXPECT().Get(mock.Anything, mock.Anything).Return(assignment)
	// Watch returns context.Canceled to break the backoff retry loop.
	w.EXPECT().Watch(mock.Anything, mock.Anything, mock.Anything).Return(context.Canceled)
	rebalanceTrigger := mock_types.NewMockAssignmentRebalanceTrigger(t)

	handler := &handlerClientImpl{
		lifetime:         typeutil.NewLifetime(),
		service:          service,
		rb:               rb,
		watcher:          w,
		rebalanceTrigger: rebalanceTrigger,
	}
	ctx := context.Background()

	// Remote WAL returns "not implemented"; Watch returns Canceled to exit the loop.
	service.EXPECT().GetService(mock.Anything).Return(nil, errors.New("not implemented"))
	cps, err := handler.GetSalvageCheckpoint(ctx, "pchannel")
	assert.Error(t, err)
	assert.Nil(t, cps)

	// After close, GetSalvageCheckpoint returns ErrClientClosed immediately.
	service.EXPECT().Close().Return()
	handler.Close()
	cps, err = handler.GetSalvageCheckpoint(ctx, "pchannel")
	assert.ErrorIs(t, err, ErrClientClosed)
	assert.Nil(t, cps)
}

func TestHandlerClientReadOnlyAssignmentWaitsForNextAssignment(t *testing.T) {
	assignment := &types.PChannelInfoAssigned{
		Channel: types.PChannelInfo{Name: "pchannel", Term: 1, AccessMode: types.AccessModeRO},
		Node:    types.StreamingNodeInfo{ServerID: 1, Address: "localhost"},
	}

	testcases := []struct {
		name string
		run  func(*handlerClientImpl) error
	}{
		{
			name: "replicate checkpoint",
			run: func(handler *handlerClientImpl) error {
				cp, err := handler.GetReplicateCheckpoint(context.Background(), "pchannel")
				assert.Nil(t, cp)
				return err
			},
		},
		{
			name: "salvage checkpoint",
			run: func(handler *handlerClientImpl) error {
				cps, err := handler.GetSalvageCheckpoint(context.Background(), "pchannel")
				assert.Nil(t, cps)
				return err
			},
		},
		{
			name: "producer",
			run: func(handler *handlerClientImpl) error {
				producer, err := handler.CreateProducer(context.Background(), &ProducerOptions{PChannel: "pchannel"})
				assert.Nil(t, producer)
				return err
			},
		},
	}

	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			w := mock_assignment.NewMockWatcher(t)
			w.EXPECT().Get(mock.Anything, "pchannel").Return(assignment)
			w.EXPECT().Watch(mock.Anything, "pchannel", assignment).Return(context.Canceled)

			handler := &handlerClientImpl{
				lifetime: typeutil.NewLifetime(),
				watcher:  w,
			}

			err := tc.run(handler)
			assert.ErrorIs(t, err, context.Canceled)
		})
	}
}

func TestHandlerClient_PrepareReleaseManualFlush(t *testing.T) {
	assignment := &types.PChannelInfoAssigned{
		Channel: types.PChannelInfo{Name: "pchannel", Term: 1, AccessMode: types.AccessModeRO},
		Node:    types.StreamingNodeInfo{ServerID: 1, Address: "localhost"},
	}
	vchannel := "pchannel_100v0"
	releaseSegmentIDs := []int64{1001}

	service := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeHandlerServiceClient](t)
	service.EXPECT().Close().Return()

	rb := mock_resolver.NewMockBuilder(t)
	rb.EXPECT().Close().Run(func() {})
	w := mock_assignment.NewMockWatcher(t)
	w.EXPECT().Get(mock.Anything, "pchannel").Return(assignment)
	w.EXPECT().Close().Run(func() {})

	handler := &handlerClientImpl{
		lifetime: typeutil.NewLifetime(),
		service:  service,
		rb:       rb,
		watcher:  w,
	}

	prepared, err := handler.PrepareReleaseManualFlush(context.Background(), 100, vchannel, releaseSegmentIDs)
	assert.NoError(t, err)
	assert.False(t, prepared)

	handler.Close()
	prepared, err = handler.PrepareReleaseManualFlush(context.Background(), 100, vchannel, releaseSegmentIDs)
	assert.ErrorIs(t, err, ErrClientClosed)
	assert.False(t, prepared)
}

func TestHandlerClient_GetReplicateCheckpointReplicateViolation(t *testing.T) {
	assignment := &types.PChannelInfoAssigned{
		Channel: types.PChannelInfo{Name: "pchannel", Term: 1},
		Node:    types.StreamingNodeInfo{ServerID: 1, Address: "localhost"},
	}

	service := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeHandlerServiceClient](t)
	handlerServiceClient := mock_streamingpb.NewMockStreamingNodeHandlerServiceClient(t)
	// Remote WAL reports a replicate violation: the target is no longer a secondary
	// cluster (e.g. after force_promote). This is unrecoverable for the current WAL
	// role, so it must be returned immediately rather than retried to the deadline.
	handlerServiceClient.EXPECT().GetReplicateCheckpoint(mock.Anything, mock.Anything).Return(
		nil, status.NewReplicateViolation("wal is not a secondary cluster in replicating topology"))
	service.EXPECT().GetService(mock.Anything).Return(handlerServiceClient, nil)

	w := mock_assignment.NewMockWatcher(t)
	// Always return the assignment so the create func is invoked.
	w.EXPECT().Get(mock.Anything, mock.Anything).Return(assignment)
	// Watch is intentionally NOT expected: an immediate return must not enter the
	// backoff retry loop. If it did, the mock would fail on an unexpected Watch call.
	rebalanceTrigger := mock_types.NewMockAssignmentRebalanceTrigger(t)

	handler := &handlerClientImpl{
		lifetime:         typeutil.NewLifetime(),
		service:          service,
		watcher:          w,
		rebalanceTrigger: rebalanceTrigger,
	}

	cp, err := handler.GetReplicateCheckpoint(context.Background(), "pchannel")
	assert.Error(t, err)
	assert.Nil(t, cp)
	assert.True(t, status.AsStreamingError(err).IsReplicateViolation())
}

func TestDial(t *testing.T) {
	paramtable.Init()

	w := mock_types.NewMockAssignmentDiscoverWatcher(t)
	w.EXPECT().AssignmentDiscover(mock.Anything, mock.Anything).RunAndReturn(
		func(ctx context.Context, f func(*types.VersionedStreamingNodeAssignments) error) error {
			return context.Canceled
		},
	)
	handler := NewHandlerClient(w)
	assert.NotNil(t, handler)
	time.Sleep(100 * time.Millisecond)
	handler.Close()
}