package kafkax

import (
	"encoding/json"
	"fmt"
	"github.com/Shopify/sarama"
	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/message/models"
	"log"
	"strings"
	"time"
)

// sarame kafka 消息生产
type KafkaMessageProducer struct {
	KafkaHosts string
	LogInfo    models.LogInfo
}

// 同步发送
func (engine *KafkaMessageProducer) Publish(messages []*models.Message, option map[string]interface{}) (*models.MessagePublishResult, error) {
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Retry.Max = 10
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Version = sarama.V0_11_0_0
	brokerList := strings.Split(engine.KafkaHosts, ",")
	producer, err := sarama.NewSyncProducer(brokerList, config)
	if err != nil {
		return nil, err
	}
	defer func() {
		if err := producer.Close(); err != nil {
			log.Println(err)
		}
	}()
	var successMessageIds []int64
	var errMessageIds []int64
	for _, message := range messages {
		if value, err := json.Marshal(message); err == nil {
			msg := &sarama.ProducerMessage{
				Topic:     message.Topic,
				Value:     sarama.StringEncoder(value),
				Timestamp: time.Now(),
			}
			partition, offset, err := producer.SendMessage(msg)
			if err != nil {
				errMessageIds = append(errMessageIds, message.Id)
				log.Println(err)
			} else {
				successMessageIds = append(successMessageIds, message.Id)
				var append = make(map[string]interface{})
				append["topic"] = message.Topic
				append["partition"] = partition
				append["offset"] = offset
				log.Println("kafka消息发送", append)
			}
		}
	}
	return &models.MessagePublishResult{SuccessMessageIds: successMessageIds, ErrorMessageIds: errMessageIds}, nil
}

// 消息调度器
type MessageDispatcher struct {
	notifications     chan struct{}
	messageChan       chan *models.Message
	dispatchTicker    *time.Ticker
	messageRepository models.MessageRepository
	producer          models.MessageProducer
}

func (dispatcher *MessageDispatcher) MessagePublishedNotice() error {
	time.Sleep(time.Second * 2)
	dispatcher.notifications <- struct{}{}
	return nil
}

func (dispatcher *MessageDispatcher) MessagePublish(messages []*models.Message) error {
	for i := range messages {
		dispatcher.messageChan <- messages[i]
	}
	return nil
}

// go dispatcher.Dispatch() 启动一个独立协程
func (dispatcher *MessageDispatcher) Dispatch() {
	for {
		select {
		case <-dispatcher.dispatchTicker.C:
			go func(dispatcher *MessageDispatcher) {
				dispatcher.notifications <- struct{}{}
			}(dispatcher)
		case <-dispatcher.notifications:
			if dispatcher.messageRepository == nil {
				continue
			}
			messages, _ := dispatcher.messageRepository.FindNoPublishedStoredMessages()
			var messagesInProcessIds []int64
			for i := range messages {
				messagesInProcessIds = append(messagesInProcessIds, messages[i].Id)
			}
			if messages != nil && len(messages) > 0 {
				dispatcher.messageRepository.FinishMessagesStatus(messagesInProcessIds, int(models.InProcess))

				reuslt, err := dispatcher.producer.Publish(messages, nil)
				if err == nil && len(reuslt.SuccessMessageIds) > 0 {
					dispatcher.messageRepository.FinishMessagesStatus(reuslt.SuccessMessageIds, int(models.Finished))
				}
				//发送失败的消息ID列表 更新状态 进行中->未开始
				if len(reuslt.ErrorMessageIds) > 0 {
					dispatcher.messageRepository.FinishMessagesStatus(reuslt.ErrorMessageIds, int(models.UnFinished))
				}
			}
		case msg := <-dispatcher.messageChan:
			dispatcher.producer.Publish([]*models.Message{msg}, nil)
		}
	}
}

type MessageDirector struct {
	messageRepository models.MessageRepository
	dispatcher        *MessageDispatcher
}

func (d *MessageDirector) PublishMessages(messages []*models.Message) error {
	if d.dispatcher == nil {
		return fmt.Errorf("dispatcher还没有启动")
	}
	if d.messageRepository == nil {
		d.dispatcher.MessagePublish(messages)
		return nil
	}
	for _, message := range messages {
		if err := d.messageRepository.SaveMessage(message); err != nil {
			return err
		}
	}
	if err := d.dispatcher.MessagePublishedNotice(); err != nil {
		return err
	}
	return nil
}

// 消息发布器
// options["kafkaHosts"]="localhost:9092"
// options["timeInterval"]=time.Second*60*5
func NewMessageDirector(messageRepository models.MessageRepository, options map[string]interface{}) *MessageDirector {
	dispatcher := &MessageDispatcher{
		notifications:     make(chan struct{}),
		messageRepository: messageRepository,
		messageChan:       make(chan *models.Message, 100),
	}

	var hosts string
	if kafkaHosts, ok := options["kafkaHosts"]; ok {
		hosts = kafkaHosts.(string)
	} else {
		hosts = "localhost:9092"
	}
	dispatcher.producer = &KafkaMessageProducer{KafkaHosts: hosts, LogInfo: models.DefaultLog}

	if interval, ok := options["timeInterval"]; ok {
		dispatcher.dispatchTicker = time.NewTicker(interval.(time.Duration))
	} else {
		dispatcher.dispatchTicker = time.NewTicker(time.Second * 60 * 5)
	}
	go dispatcher.Dispatch()

	return &MessageDirector{
		messageRepository: messageRepository,
		dispatcher:        dispatcher,
	}
}