package tieventbus

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"testing"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/tiinf/transport/tieventbus/test"
)

const (
	TxPubTable = "msg_store"
)

type Action string

const (
	Open Action = "open"
)

type EventData struct {
	ID uint64
	Action
}

type EventbusTestSuite struct {
	suite.Suite
	eventbus Eventbus
	db       *sql.DB
}

func (s *EventbusTestSuite) newTestEventbus() (Eventbus, error) {
	return NewEventbusMock(EventbusConfig{
		AppName:    "app",
		DB:         s.db,
		TxPubTable: TxPubTable,
	})
}

func (s *EventbusTestSuite) initTestEventbus() {
	assert := s.Assert()
	var err error
	s.eventbus, err = s.newTestEventbus()
	assert.NoError(err)
}

func (s *EventbusTestSuite) runTestEventbus() {
	assert := s.Assert()
	err := s.eventbus.Run()
	assert.NoError(err)
}

func (s *EventbusTestSuite) closeTestEventbus() {
	s.eventbus.Close()
}

func (s *EventbusTestSuite) SetupTest() {
	s.db = test.InitMysql()
	s.initTestEventbus()
}

func (s *EventbusTestSuite) TearDownTest() {
	s.eventbus.Close()
}

func TestEventbusTestSuite(t *testing.T) {
	suite.Run(t, new(EventbusTestSuite))
}

func (s *EventbusTestSuite) TestEventbusHappyPath() {
	assert := s.Assert()
	eventType := "event-type-1"
	ch := make(chan *EventData)
	eventbus := s.eventbus
	eventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "1",
		HandleFunc: func(event *Event) error {
			ch <- event.Data.(*EventData)
			return nil
		},
		DataStruct: EventData{},
	})
	s.runTestEventbus()
	sendData := &EventData{
		ID:     1,
		Action: Open,
	}
	sendEvent := &Event{
		ID:   "1",
		Type: eventType,
		Data: sendData,
	}
	_ = eventbus.PublishAsync(sendEvent)
	select {
	case eventData := <-ch:
		assert.Equal(sendData.ID, eventData.ID)
		assert.Equal(sendData.Action, eventData.Action)
	case <-time.After(time.Second):
		panic(fmt.Errorf("wait publish timeout"))
	}
}

func (s *EventbusTestSuite) TestEventbusDuplicateGroup() {
	assert := s.Assert()
	eventType := "event-type-1"
	eventbus := s.eventbus
	hc := HandlerConfig{
		EventType: eventType,
		Group:     "1",
		HandleFunc: func(event *Event) error {
			return nil
		},
		RetryTimes: 3,
		DataStruct: EventData{},
	}
	eventbus.RegisterHandler(hc)
	eventbus.RegisterHandler(hc)
	err := eventbus.Run()
	assert.Errorf(err, DupGroupEventTypeConsuming)
}

func (s *EventbusTestSuite) TestNoEventbusHandlerRegistered() {
	assert := s.Assert()
	eventbus := s.eventbus
	s.runTestEventbus()
	sendData := &EventData{
		ID:     1,
		Action: Open,
	}
	eventType := "event-type-1"
	sendEvent := &Event{
		ID:   "1",
		Type: eventType,
		Data: sendData,
	}
	sub, err := eventbus.(*EventbusImpl).pubSub.GetMQSubscriber("!")
	assert.NoError(err)
	msgChan, err := sub.Subscribe(context.Background(), eventType)
	assert.NoError(err)
	tx, err := s.db.Begin()
	assert.NoError(err)
	err = eventbus.PublishWithDBTx(tx, sendEvent)
	assert.NoError(err)
	select {
	case <-msgChan:
		panic(fmt.Errorf("unexcepted"))
	case <-time.After(200 * time.Millisecond):
	}
	err = tx.Commit()
	assert.NoError(err)
	select {
	case <-msgChan:
	case <-time.After(2 * time.Second):
		panic(fmt.Errorf("unexcepted"))
	}
}

func (s *EventbusTestSuite) TestEventbusRetry() {
	assert := s.Assert()
	eventType := "event-type-1"
	eventbus := s.eventbus
	failedFactor := 0
	ch := make(chan *EventData)
	eventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "1",
		HandleFunc: func(event *Event) error {
			if failedFactor == 0 {
				failedFactor++
				return fmt.Errorf("1")
			}
			ch <- event.Data.(*EventData)
			return nil
		},
		RetryTimes: 3,
		DataStruct: EventData{},
	})
	s.runTestEventbus()
	sendData := &EventData{
		ID:     1,
		Action: Open,
	}
	sendEvent := &Event{
		ID:   "1",
		Type: eventType,
		Data: sendData,
	}
	_ = eventbus.PublishAsync(sendEvent)
	select {
	case eventData := <-ch:
		assert.Equal(sendData.ID, eventData.ID)
		assert.Equal(sendData.Action, eventData.Action)
		assert.Equal(failedFactor, 1)
	case <-time.After(time.Second):
		panic(fmt.Errorf("wait publish timeout"))
	}
}

func (s *EventbusTestSuite) TestEventbusDLQ() {
	assert := s.Assert()
	eventType := "event-type-1"
	eventbus := s.eventbus
	ch := make(chan *EventData)
	eventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "1",
		HandleFunc: func(event *Event) error {
			ch <- event.Data.(*EventData)
			return fmt.Errorf("1")
		},
		DataStruct: EventData{},
	})
	s.runTestEventbus()
	sendData := &EventData{
		ID:     1,
		Action: Open,
	}
	sendEvent := &Event{
		ID:   "1",
		Type: eventType,
		Data: sendData,
	}
	_ = eventbus.PublishAsync(sendEvent)
	select {
	case eventData := <-ch:
		assert.Equal(sendData.ID, eventData.ID)
		assert.Equal(sendData.Action, eventData.Action)
	case <-time.After(time.Second):
		panic(fmt.Errorf("wait publish timeout"))
	}
	fixDLQEventbus, err := s.newTestEventbus()
	assert.NoError(err)
	fixDLQEventbus.(*EventbusImpl).pubSub = eventbus.(*EventbusImpl).pubSub
	fixed := false
	fixDLQEventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "2",
		HandleFunc: func(event *Event) error {
			fixed = true
			return nil
		},
		DataStruct: EventData{},
	})
	over := make(chan struct{})
	go func() {
		err = fixDLQEventbus.FixDLQ(1 * time.Second)
		close(over)
	}()
	select {
	case <-over:
	case <-time.After(10 * time.Second):
		panic(fmt.Errorf("wait processDLQ timeout"))
	}
	assert.NoError(err)
	assert.True(fixed)
	fixDLQEventbus.Close()
}

func (s *EventbusTestSuite) TestEventbusPublishWithTx() {
	assert := s.Assert()
	eventType := "event-type-1"
	eventbus := s.eventbus
	ch := make(chan *EventData)
	eventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "1",
		HandleFunc: func(event *Event) error {
			ch <- event.Data.(*EventData)
			return nil
		},
		DataStruct: EventData{},
	})
	s.runTestEventbus()
	sendData := &EventData{
		ID:     1,
		Action: Open,
	}
	sendEvent := &Event{
		ID:   "1",
		Type: eventType,
		Data: sendData,
	}
	tx, err := s.db.Begin()
	assert.NoError(err)
	err = eventbus.PublishWithDBTx(tx, sendEvent)
	assert.NoError(err)
	select {
	case <-ch:
		panic(fmt.Errorf("unexcepted data recieved"))
	case <-time.After(time.Millisecond * 100):
	}
	err = tx.Commit()
	assert.NoError(err)
	select {
	case eventData := <-ch:
		assert.Equal(sendData.ID, eventData.ID)
		assert.Equal(sendData.Action, eventData.Action)
	case <-time.After(2 * time.Second):
		panic(fmt.Errorf("wait event timeout"))
	}
}

func (s *EventbusTestSuite) TestEventbusMetrics() {
	assert := s.Assert()
	eventType := "event-type-1"
	eventbus := s.eventbus
	prometheusRegistry := prometheus.NewRegistry()
	eventbus.EnableMetrics(prometheusRegistry, "default", "default")
	ch := make(chan *EventData)
	failedFactor := 0
	eventbus.RegisterHandler(HandlerConfig{
		EventType:  eventType,
		Group:      "1",
		RetryTimes: 2,
		HandleFunc: func(event *Event) error {
			if failedFactor != 0 {
				return errors.New("1")
			}
			ch <- event.Data.(*EventData)
			failedFactor++
			return nil
		},
		DataStruct: EventData{},
	})
	s.runTestEventbus()
	group := "1"
	sendData := &EventData{
		ID:     1,
		Action: Open,
	}
	sendEvent := &Event{
		ID:   group,
		Type: eventType,
		Data: sendData,
	}
	_ = eventbus.PublishAsync(sendEvent)
	select {
	case eventData := <-ch:
		assert.Equal(sendData.ID, eventData.ID)
		assert.Equal(sendData.Action, eventData.Action)
	case <-time.After(time.Second):
		panic(fmt.Errorf("wait publish timeout"))
	}
	tx, err := s.db.Begin()
	assert.NoError(err)
	err = eventbus.PublishWithDBTx(tx, sendEvent)
	assert.NoError(err)
	err = tx.Commit()
	assert.NoError(err)
	select {
	case <-ch:
		panic(fmt.Errorf("unexcpeted"))
	case <-time.After(2 * time.Second):
	}
	eventbus.(*EventbusImpl).pubSub = &disablePubSub{}
	err = eventbus.Publish(sendEvent)
	assert.Error(err)
	mfs, err := prometheusRegistry.Gather()
	assert.NoError(err)
	assert.Len(mfs, 5)
	for _, mf := range mfs {
		switch mf.GetName() {
		case "default_default_events_published_total":
			counter := make(map[string]map[string]int)
			for _, m := range mf.Metric {
				for _, l := range m.Label {
					if _, ok := counter[*l.Name]; !ok {
						counter[*l.Name] = make(map[string]int)
					}
					if _, ok := counter[*l.Name][*l.Value]; !ok {
						counter[*l.Name][*l.Value] = 0
					}
					counter[*l.Name][*l.Value] += int(*m.Counter.Value)
				}
			}
			assert.Equal(5, counter[eventTypeKey][eventType])
			assert.Equal(3, counter[sourceKey][SourcePublisher])
			assert.Equal(1, counter[sourceKey][SourceConsumer])
			assert.Equal(1, counter[sourceKey][SourceDB])
			assert.Equal(4, counter[successKey]["true"])
			assert.Equal(1, counter[successKey]["false"])
			assert.Equal(3, counter[targetKey][TargetMQ])
			assert.Equal(1, counter[targetKey][TargetDB])
			assert.Equal(1, counter[targetKey][TargetDLQ])
			break
		case "default_default_events_received_total":
			counter := make(map[string]map[string]int)
			for _, m := range mf.Metric {
				for _, l := range m.Label {
					if _, ok := counter[*l.Name]; !ok {
						counter[*l.Name] = make(map[string]int)
					}
					if _, ok := counter[*l.Name][*l.Value]; !ok {
						counter[*l.Name][*l.Value] = 0
					}
					counter[*l.Name][*l.Value] += int(*m.Counter.Value)
				}
			}
			assert.Equal(2, counter[eventTypeKey][eventType])
			assert.Equal(1, counter[eventTypeKey][TxPubTable])
			assert.Equal(1, counter[groupKey]["nil"])
			assert.Equal(2, counter[groupKey][group])
			assert.Equal(2, counter[retryTimesKey]["0"])
			assert.Equal(1, counter[retryTimesKey]["2"])
			assert.Equal(1, counter[sourceKey][SourceDB])
			assert.Equal(2, counter[sourceKey][SourceMQ])
			assert.Equal(2, counter[successKey]["true"])
			assert.Equal(1, counter[successKey]["false"])
			break
		case "default_default_execution_time_seconds":
			counter := make(map[string]map[string]int)
			for _, m := range mf.Metric {
				for _, l := range m.Label {
					if _, ok := counter[*l.Name]; !ok {
						counter[*l.Name] = make(map[string]int)
					}
					if _, ok := counter[*l.Name][*l.Value]; !ok {
						counter[*l.Name][*l.Value] = 0
					}
					counter[*l.Name][*l.Value] += int(*m.Histogram.SampleCount)
				}
			}
			assert.Equal(2, counter[eventTypeKey][eventType])
			assert.Equal(1, counter[eventTypeKey][TxPubTable])
			assert.Equal(1, counter[groupKey]["nil"])
			assert.Equal(2, counter[groupKey][group])
			assert.Equal(2, counter[retryTimesKey]["0"])
			assert.Equal(1, counter[retryTimesKey]["2"])
			assert.Equal(1, counter[sourceKey][SourceDB])
			assert.Equal(2, counter[sourceKey][SourceMQ])
			assert.Equal(2, counter[successKey]["true"])
			assert.Equal(1, counter[successKey]["false"])
			break
		case "default_default_publish_time_seconds":
			counter := make(map[string]map[string]int)
			for _, m := range mf.Metric {
				for _, l := range m.Label {
					if _, ok := counter[*l.Name]; !ok {
						counter[*l.Name] = make(map[string]int)
					}
					if _, ok := counter[*l.Name][*l.Value]; !ok {
						counter[*l.Name][*l.Value] = 0
					}
					counter[*l.Name][*l.Value] += int(*m.Histogram.SampleCount)
				}
			}
			assert.Equal(5, counter[eventTypeKey][eventType])
			assert.Equal(3, counter[sourceKey][SourcePublisher])
			assert.Equal(1, counter[sourceKey][SourceConsumer])
			assert.Equal(1, counter[sourceKey][SourceDB])
			assert.Equal(4, counter[successKey]["true"])
			assert.Equal(1, counter[successKey]["false"])
			assert.Equal(3, counter[targetKey][TargetMQ])
			assert.Equal(1, counter[targetKey][TargetDB])
			assert.Equal(1, counter[targetKey][TargetDLQ])
			break
		case "default_default_events_received_latency_seconds":
			counter := make(map[string]map[string]int)
			for _, m := range mf.Metric {
				for _, l := range m.Label {
					if _, ok := counter[*l.Name]; !ok {
						counter[*l.Name] = make(map[string]int)
					}
					if _, ok := counter[*l.Name][*l.Value]; !ok {
						counter[*l.Name][*l.Value] = 0
					}
					counter[*l.Name][*l.Value] += int(*m.Histogram.SampleCount)
				}
				val := m.Histogram.GetSampleSum() / float64(m.Histogram.GetSampleCount())
				assert.Greater(val, .0)
				assert.Less(val, 1.5)
			}
			assert.Equal(2, counter[eventTypeKey][eventType])
			assert.Equal(1, counter[eventTypeKey][TxPubTable])
			assert.Equal(1, counter[groupKey]["nil"])
			assert.Equal(2, counter[groupKey][group])
			assert.Equal(2, counter[retryTimesKey]["0"])
			assert.Equal(1, counter[retryTimesKey]["2"])
			assert.Equal(1, counter[sourceKey][SourceDB])
			assert.Equal(2, counter[sourceKey][SourceMQ])
			assert.Equal(2, counter[successKey]["true"])
			assert.Equal(1, counter[successKey]["false"])
			break
		}

	}
}

func (s *EventbusTestSuite) TestEventbusRegisterDataStructSuccess() {
	assert := s.Assert()
	eventType := "event-type-1"
	ch1 := make(chan *test.D1)
	ch2 := make(chan *test.D1)
	ch3 := make(chan *test.D2)
	ch4 := make(chan *test.D2)
	eventbus := s.eventbus
	eventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "1",
		HandleFunc: func(event *Event) error {
			ch1 <- event.Data.(*test.D1)
			return nil
		},
		DataStruct: test.D1{},
	})
	eventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "2",
		HandleFunc: func(event *Event) error {
			ch2 <- event.Data.(*test.D1)
			return nil
		},
		DataStruct: &test.D1{},
	})
	eventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "3",
		HandleFunc: func(event *Event) error {
			ch3 <- event.Data.(*test.D2)
			return nil
		},
		DataStruct: &test.D2{},
	})
	eventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "4",
		HandleFunc: func(event *Event) error {
			ch4 <- event.Data.(*test.D2)
			return nil
		},
		DataStruct: test.D2{},
	})
	s.runTestEventbus()
	sendData := &test.D1{ID: 666}
	sendEvent := &Event{
		ID:   "1",
		Type: eventType,
		Data: sendData,
	}
	_ = eventbus.PublishAsync(sendEvent)
	for cnt := 0; cnt != 4; cnt++ {
		select {
		case eventData := <-ch1:
			assert.Equal(sendData.ID, eventData.ID)
		case eventData := <-ch2:
			assert.Equal(sendData.ID, eventData.ID)
		case eventData := <-ch3:
			assert.Equal(sendData.ID, eventData.ID)
		case eventData := <-ch4:
			assert.Equal(sendData.ID, eventData.ID)
		case <-time.After(time.Second):
			panic(fmt.Errorf("wait publish timeout"))
		}
	}
}

func (s *EventbusTestSuite) TestEventbusRegisterDataStructFailed() {
	eventType := "event-type-1"
	ch1 := make(chan *test.D1)
	eventbus := s.eventbus
	dataStruct := &test.D1{}
	eventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "1",
		HandleFunc: func(event *Event) error {
			ch1 <- event.Data.(*test.D1)
			return nil
		},
		DataStruct: &dataStruct,
	})
	assert := s.Assert()
	err := s.eventbus.Run()
	assert.Errorf(err, "eventbus data do not support pointer to pointer")
}

func (s *EventbusTestSuite) TestEventbusProto() {
	assert := s.Assert()
	eventType := "event-type-1"
	ch1 := make(chan *test.D1)
	ch2 := make(chan *test.D2)
	eventbus := s.eventbus
	eventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "1",
		HandleFunc: func(event *Event) error {
			ch1 <- event.Data.(*test.D1)
			return nil
		},
		DataStruct: &test.D1{},
	})
	eventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "2",
		HandleFunc: func(event *Event) error {
			ch2 <- event.Data.(*test.D2)
			return nil
		},
		DataStruct: &test.D2{},
	})
	s.runTestEventbus()
	sendData1 := &test.D1{ID: 666}
	sendEvent1 := &Event{
		ID:   "1",
		Type: eventType,
		Data: sendData1,
	}
	_ = eventbus.PublishAsync(sendEvent1)
	for cnt := 0; cnt != 2; cnt++ {
		select {
		case eventData := <-ch1:
			assert.Equal(sendData1.ID, eventData.ID)
		case eventData := <-ch2:
			assert.Equal(sendData1.ID, eventData.ID)
		case <-time.After(2 * time.Second):
			panic(fmt.Errorf("wait publish timeout"))
		}
	}
	sendData2 := &test.D2{ID: 888}
	sendEvent2 := &Event{
		ID:   "1",
		Type: eventType,
		Data: sendData2,
	}
	_ = eventbus.PublishAsync(sendEvent2)
	for cnt := 0; cnt != 2; cnt++ {
		select {
		case eventData := <-ch1:
			assert.Equal(sendData2.ID, eventData.ID)
		case eventData := <-ch2:
			assert.Equal(sendData2.ID, eventData.ID)
		case <-time.After(2 * time.Second):
			panic(fmt.Errorf("wait publish timeout"))
		}
	}
}

func (s *EventbusTestSuite) TestEventbusJSON() {
	assert := s.Assert()
	eventType := "event-type-1"
	ch1 := make(chan *map[string]interface{})
	ch2 := make(chan *EventData)
	eventbus := s.eventbus
	eventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "1",
		HandleFunc: func(event *Event) error {
			ch1 <- event.Data.(*map[string]interface{})
			return nil
		},
		DataStruct: make(map[string]interface{}),
	})
	eventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "2",
		HandleFunc: func(event *Event) error {
			ch2 <- event.Data.(*EventData)
			return nil
		},
		DataStruct: &EventData{},
	})
	s.runTestEventbus()
	sendData1 := &EventData{ID: 0xffffffffffffffff}
	sendEvent1 := &Event{
		ID:   "1",
		Type: eventType,
		Data: sendData1,
	}
	_ = eventbus.PublishAsync(sendEvent1)
	for cnt := 0; cnt != 2; cnt++ {
		select {
		case eventData := <-ch1:
			ID, err := strconv.ParseUint((*eventData)["ID"].(json.Number).String(), 10, 64)
			assert.NoError(err)
			assert.Equal(sendData1.ID, ID)
		case eventData := <-ch2:
			assert.Equal(sendData1.ID, eventData.ID)
		case <-time.After(2 * time.Second):
			panic(fmt.Errorf("wait publish timeout"))
		}
	}
}

func (s *EventbusTestSuite) TestEventbusRunUntilClosed() {
	var err error
	assert := s.Assert()
	ch1 := make(chan struct{})
	go func() {
		err = s.eventbus.RunUtilClosed()
		close(ch1)
	}()
	select {
	case <-s.eventbus.Running():
	case <-time.After(time.Second):
		panic(fmt.Errorf("wait for running timeout"))
	}
	s.eventbus.Close()
	select {
	case <-ch1:
	case <-time.After(time.Second):
		panic(fmt.Errorf("wait for closing timeout"))
	}
	assert.NoError(err)
	s.initTestEventbus()
	s.eventbus.RegisterHandler(HandlerConfig{
		EventType: "1",
		Group:     "2",
		HandleFunc: func(event *Event) error {
			return nil
		},
		DataStruct: &test.D2{},
	})
	ch2 := make(chan struct{})
	go func() {
		err = s.eventbus.RunUtilClosed()
		close(ch2)
	}()
	select {
	case <-s.eventbus.Running():
	case <-time.After(time.Second):
		panic(fmt.Errorf("wait for running timeout"))
	}
	s.eventbus.Close()
	select {
	case <-ch2:
	case <-time.After(time.Second):
		panic(fmt.Errorf("wait for closing timeout"))
	}
}

func (s *EventbusTestSuite) TestHandlerPanic() {
	assert := s.Assert()
	eventType := "event-type-1"
	eventbus := s.eventbus
	dataStruct := &EventData{}
	eventbus.RegisterHandler(HandlerConfig{
		EventType: eventType,
		Group:     "1",
		HandleFunc: func(event *Event) error {
			panic(errors.New("test"))
		},
		DataStruct: dataStruct,
	})
	assert.NoError(eventbus.Run())
	sendData1 := &EventData{ID: 0xffffffffffffffff}
	sendEvent1 := &Event{
		ID:   "1",
		Type: eventType,
		Data: sendData1,
	}
	msgCh, err := eventbus.(*EventbusImpl).pubSub.(*pubSubMock).channel.Subscribe(context.Background(), getDLQTopic(eventType))
	assert.NoError(err)
	assert.NoError(eventbus.Publish(sendEvent1))
	select {
	case <-msgCh:
	case <-time.After(100 * time.Millisecond):
		panic("timeout")
	}
}