package impl

import (
	"context"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/mock"
	center_mock "github.com/tidbcloud/dataflow-service/app/event_bus/publisher/center/mocks"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/eventbus-datastruct/event"
)

func TestHandleBizChangefeedEvent(t *testing.T) {
	testcases := []struct {
		event            entity.DataflowEventType
		changefeedStatus entity.BizChangefeedOperationStatus
		subSummary       string
		displayStatus    event.EventStatus
		displayEventType string
	}{
		{
			event:            entity.CreateChangefeed,
			changefeedStatus: entity.BizChangefeedOperationStatusCreating,
			subSummary:       "Start to Create changefeed",
			displayStatus:    event.EventStatusStarted,
			displayEventType: string(entity.CreateChangefeed),
		},
		{
			event:            entity.CreateChangefeed,
			changefeedStatus: entity.BizChangefeedOperationStatusNormal,
			subSummary:       "Changefeed successfully Create",
			displayStatus:    event.EventStatusSucceeded,
			displayEventType: string(entity.CreateChangefeed),
		},
		{
			event:            entity.CreateChangefeed,
			changefeedStatus: entity.BizChangefeedOperationStatusFailed,
			subSummary:       "Failed to run changefeed",
			displayStatus:    event.EventStatusFailed,
			displayEventType: string(entity.CreateChangefeed),
		},
		{
			event:            entity.ScaleChangefeed,
			changefeedStatus: entity.BizChangefeedOperationStatusScaling,
			subSummary:       "Start to Scale changefeed",
			displayStatus:    event.EventStatusStarted,
			displayEventType: string(entity.ScaleChangefeed),
		},
		{
			event:            entity.ScaleChangefeed,
			changefeedStatus: entity.BizChangefeedOperationStatusNormal,
			subSummary:       "Changefeed successfully Scale",
			displayStatus:    event.EventStatusSucceeded,
			displayEventType: string(entity.ScaleChangefeed),
		},
		{
			event:            entity.DeleteChangefeed,
			changefeedStatus: entity.BizChangefeedOperationStatusDeleting,
			subSummary:       "Start to Delete changefeed",
			displayStatus:    event.EventStatusStarted,
			displayEventType: string(entity.DeleteChangefeed),
		},
		{
			event:            entity.DeleteChangefeed,
			changefeedStatus: entity.BizChangefeedOperationStatusDeleted,
			subSummary:       "Changefeed successfully Delete",
			displayStatus:    event.EventStatusSucceeded,
			displayEventType: string(entity.DeleteChangefeed),
		},
		{
			event:            entity.PauseChangefeed,
			changefeedStatus: entity.BizChangefeedOperationStatusPaused,
			subSummary:       "Changefeed successfully Pause",
			displayStatus:    event.EventStatusSucceeded,
			displayEventType: string(entity.PauseChangefeed),
		},
		{
			event:            entity.ResumeChangefeed,
			changefeedStatus: entity.BizChangefeedOperationStatusNormal,
			subSummary:       "Changefeed successfully Resume",
			displayStatus:    event.EventStatusSucceeded,
			displayEventType: string(entity.ResumeChangefeed),
		},
		{
			event:            entity.EditChangefeed,
			changefeedStatus: entity.BizChangefeedOperationStatusPaused,
			subSummary:       "Changefeed successfully Edit",
			displayStatus:    event.EventStatusSucceeded,
			displayEventType: string(entity.EditChangefeed),
		},
	}

	ctx := context.Background()
	now := time.Now()
	mockBizChangefeedEntity := &entity.BizChangefeed{
		ID:             111,
		DownstreamType: entity.TiCDCDownStreamTypeMySQL,
		UpdatedAt:      &now,
	}

	for _, tc := range testcases {
		mockPublisher := center_mock.NewEventCenterPublisher(t)
		hook := &bizChangefeedEventBusHooker{
			publisher: mockPublisher,
		}
		mockBizChangefeedEntity.OperationStatus = tc.changefeedStatus
		matcher := func(displayEvent *event.DisplayEvent) bool {
			return strings.Contains(displayEvent.Summary, tc.subSummary) &&
				displayEvent.DisplayStatus == tc.displayStatus &&
				displayEvent.DisplayEventType == tc.displayEventType
		}

		mockPublisher.On("PublishAsync", mock.MatchedBy(matcher)).Return()
		hook.HandleBizChangefeedEvent(ctx, tc.event, 0, mockBizChangefeedEntity)
	}
}

func TestBuildChangefeedEventMsg(t *testing.T) {
	tests := map[string]struct {
		event            entity.DataflowEventType
		operationStatus  entity.BizChangefeedOperationStatus
		errMsg           string
		expectedContains string
		notContains      string
	}{
		"create changefeed - creating": {
			event:            entity.CreateChangefeed,
			operationStatus:  entity.BizChangefeedOperationStatusCreating,
			expectedContains: "Start to Create changefeed",
		},
		"create changefeed - normal": {
			event:            entity.CreateChangefeed,
			operationStatus:  entity.BizChangefeedOperationStatusNormal,
			expectedContains: "Changefeed successfully Create",
		},
		"create changefeed - failed": {
			event:            entity.CreateChangefeed,
			operationStatus:  entity.BizChangefeedOperationStatusFailed,
			errMsg:           "connection timeout",
			expectedContains: "Failed to run changefeed",
		},
		"create changefeed - failed includes reason": {
			event:            entity.CreateChangefeed,
			operationStatus:  entity.BizChangefeedOperationStatusFailed,
			errMsg:           "[CDC:ErrMySQLConnectionError]connection failed",
			expectedContains: "Reason: [CDC:ErrMySQLConnectionError]connection failed",
		},
		"scale changefeed - scaling": {
			event:            entity.ScaleChangefeed,
			operationStatus:  entity.BizChangefeedOperationStatusScaling,
			expectedContains: "Start to Scale changefeed",
		},
		"scale changefeed - normal": {
			event:            entity.ScaleChangefeed,
			operationStatus:  entity.BizChangefeedOperationStatusNormal,
			expectedContains: "Changefeed successfully Scale",
		},
		"delete changefeed - deleting": {
			event:            entity.DeleteChangefeed,
			operationStatus:  entity.BizChangefeedOperationStatusDeleting,
			expectedContains: "Start to Delete changefeed",
		},
		"delete changefeed - deleted": {
			event:            entity.DeleteChangefeed,
			operationStatus:  entity.BizChangefeedOperationStatusDeleted,
			expectedContains: "Changefeed successfully Delete",
		},
		"pause changefeed - paused": {
			event:            entity.PauseChangefeed,
			operationStatus:  entity.BizChangefeedOperationStatusPaused,
			expectedContains: "Changefeed successfully Pause",
		},
		"resume changefeed - normal": {
			event:            entity.ResumeChangefeed,
			operationStatus:  entity.BizChangefeedOperationStatusNormal,
			expectedContains: "Changefeed successfully Resume",
		},
		"edit changefeed - normal": {
			event:            entity.EditChangefeed,
			operationStatus:  entity.BizChangefeedOperationStatusNormal,
			expectedContains: "Changefeed successfully Edit",
		},
		"failed with empty error message": {
			event:            entity.CreateChangefeed,
			operationStatus:  entity.BizChangefeedOperationStatusFailed,
			errMsg:           "",
			expectedContains: "Failed to run changefeed",
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			bizChangefeedEntity := &entity.BizChangefeed{
				OperationStatus: tc.operationStatus,
				ErrMsg:          tc.errMsg,
			}

			result := buildChangefeedEventMsg(tc.event, bizChangefeedEntity)

			if !strings.Contains(result, tc.expectedContains) {
				t.Errorf("expected message to contain %q, got: %q", tc.expectedContains, result)
			}

			if tc.notContains != "" && strings.Contains(result, tc.notContains) {
				t.Errorf("expected message not to contain %q, got: %q", tc.notContains, result)
			}
		})
	}
}