package mqttx

import (
	"errors"
	"fmt"
	"gitee.com/fmpt/fmgo/logx"
	"gitee.com/fmpt/fmgo/stringx"
	"github.com/yosssi/gmq/mqtt"
	"github.com/yosssi/gmq/mqtt/client"
	"strconv"
	"strings"
	"time"
)

type HandFunc func(topic, reply string, message []byte)

type Client struct {
	client *client.Client
	addr   string
	id     string
	topics map[string]HandFunc
}

func New(addr, id string) *Client {
	cl := &Client{
		addr: addr,
		id:   id,
	}
	cl.client = client.New(&client.Options{
		// Define the processing of the error handler.
		ErrorHandler: func(err error) {
			cl.errorHandler(err)
		},
	})
	return cl
}

func (c *Client) errorHandler(err error) {
	if err.Error() == "EOF" {
		_ = c.Connect()
	}
}

func (c *Client) AddTopic(topic string, hFunc HandFunc) {
	c.topics[topic] = hFunc
}

func (c *Client) AddTopics(topics map[string]HandFunc) {
	c.topics = topics
}

func (c *Client) Connect() error {
	err := c.client.Connect(&client.ConnectOptions{
		Network:  "tcp",
		Address:  c.addr,
		ClientID: []byte(c.id),
	})
	if err != nil {
		time.AfterFunc(5*time.Second, func() {
			_ = c.Connect()
		})
		return err
	}
	if len(c.topics) > 0 {
		err = c.Subscribe(c.topics)
		if err != nil {
			return err
		}
	}
	return nil
}

func (c *Client) ConnectWithAuthentication(username, password string) error {
	err := c.client.Connect(&client.ConnectOptions{
		Network:      "tcp",
		Address:      c.addr,
		ClientID:     []byte(c.id),
		UserName:     []byte(username),
		Password:     []byte(password),
		CleanSession: true,
	})
	if err != nil {
		time.AfterFunc(5*time.Second, func() {
			_ = c.Connect()
		})
		return err
	}
	if len(c.topics) > 0 {
		err = c.Subscribe(c.topics)
		if err != nil {
			return err
		}
	}
	return nil
}

func (c *Client) Stop() error {
	err := c.client.Disconnect()
	return err
}

func (c *Client) Subscribe(topic map[string]HandFunc) error {
	var subReqs []*client.SubReq
	for k, v := range topic {
		t, h := k, v
		subReqs = append(subReqs, &client.SubReq{
			TopicFilter: []byte(t + "/#"),
			QoS:         mqtt.QoS1,
			Handler: func(topicName, message []byte) {
				topics := strings.Split(string(topicName), "/")
				if len(topics) == 2 {
					h(topics[0], topics[1], message)
					return
				}
				go func() {
					defer logx.Recovery()
					h(topics[0], "", message)
				}()
			},
		})
	}
	err := c.client.Subscribe(&client.SubscribeOptions{SubReqs: subReqs})
	if err != nil {
		return err
	}
	return nil
}

func (c *Client) Publish(topic string, data []byte) error {
	if err := c.client.Publish(&client.PublishOptions{
		QoS:       mqtt.QoS1,
		TopicName: []byte(topic),
		Message:   data,
	}); err != nil {
		return err
	}
	return nil
}

func (c *Client) Request(topic string, data []byte) ([]byte, error) {
	respTopic := c.randTopic("Request")
	mc := make(chan []byte)
	var msg []byte
	if err := c.client.Subscribe(&client.SubscribeOptions{
		SubReqs: []*client.SubReq{
			{
				TopicFilter: []byte(respTopic),
				QoS:         mqtt.QoS1,
				Handler: func(topicName, message []byte) {
					mc <- message
				},
			},
		},
	}); err != nil {
		return nil, err
	}
	defer func() {
		_ = c.client.Unsubscribe(&client.UnsubscribeOptions{
			TopicFilters: [][]byte{
				[]byte(respTopic),
			},
		})
	}()

	if err := c.client.Publish(&client.PublishOptions{
		QoS:       mqtt.QoS1,
		TopicName: []byte(fmt.Sprintf("%s/%s", topic, respTopic)),
		Message:   data,
	}); err != nil {
		return nil, err
	}
	select {
	case msg = <-mc:
		return msg, nil
	case <-time.After(time.Second * 5):
		return nil, errors.New("timeout")
	}
}

func (c *Client) randTopic(customStr string) string {
	randString := stringx.RandString(32)
	return strings.Join([]string{strconv.FormatInt(time.Now().UnixNano(), 10), randString, customStr}, "/")
}