package models

import (
	"bytes"
	"fmt"
	"log"
	"sync"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)

type MqttClient struct {
	Client        mqtt.Client
	opts          *mqtt.ClientOptions
	Username      string
	Password      string
	ClientId      string
	Broker        string
	Reconnecting  bool            // 是否正在重连
	Subscriptions map[string]byte // 存储已订阅的主题和QoS
	reconnectLock sync.Mutex
}

func (ggg *MqttClient) Setup() *MqttClient {
	ggg.opts = mqtt.NewClientOptions().AddBroker(ggg.Broker) //tcp://192.168.1.2:1883
	ggg.opts.SetUsername(ggg.Username)
	ggg.opts.SetPassword(ggg.Password)
	ggg.opts.SetClientID(ggg.ClientId)
	ggg.opts.SetDefaultPublishHandler(func(c mqtt.Client, m mqtt.Message) {
		log.Printf("[%s] 收到消息: 主题=%s, 内容=%s\n", ggg.Username, m.Topic(), m.Payload())
	})
	// 设置连接回调
	ggg.opts.OnConnect = func(client mqtt.Client) {
		log.Printf("设备 %s 连接成功\n", ggg.Username)
		ggg.Reconnecting = false
		// 重连后恢复订阅
		ggg.reconnectLock.Lock()
		defer ggg.reconnectLock.Unlock()
		if ggg.Subscriptions != nil {
			for topic, qos := range ggg.Subscriptions {
				token := client.Subscribe(topic, qos, nil)
				token.Wait()
				if token.Error() != nil {
					log.Printf("恢复订阅失败 %s: %v\n", topic, token.Error())
				} else {
					log.Printf("恢复订阅成功 %s\n", topic)
				}
			}
		}
	}

	// 设置断开连接回调
	ggg.opts.OnConnectionLost = func(client mqtt.Client, err error) {
		log.Printf("设备 %s 连接丢失: %v\n", ggg.Username, err)
		ggg.startReconnect()
	}

	ggg.Client = mqtt.NewClient(ggg.opts)
	//连接并等待连接成功
	if token := ggg.Client.Connect(); token.WaitTimeout(3 * time.Second) {
		if token.Error() != nil {
			return nil
		}
	} else {
		return nil
	}

	return ggg
}
func (ggg *MqttClient) Publish(topic string, qos int, retained bool, payload interface{}) error {
	if !ggg.Client.IsConnected() {
		return fmt.Errorf("MQTT客户端未连接")
	}
	// 校验 QoS 合法性
	if qos < 0 || qos > 2 {
		return fmt.Errorf("invalid QoS level: %d", qos)
	}

	var pubPayload interface{}
	switch p := payload.(type) {
	case string:
		pubPayload = p
	case []byte:
		pubPayload = p
	case bytes.Buffer:
		pubPayload = p.Bytes()
	default:
		return fmt.Errorf("unknown payload type")
	}

	// 校验 payload 长度（最大 268,435,455 字节）
	if b, ok := pubPayload.([]byte); ok {
		if len(b) > 268435455 {
			return fmt.Errorf("payload too large for MQTT publish")
		}
	} else if s, ok := pubPayload.(string); ok {
		if len(s) > 268435455 {
			return fmt.Errorf("payload too large for MQTT publish")
		}
	}

	token := ggg.Client.Publish(topic, byte(qos), retained, payload)

	// 等待发布结果，最多等待 3 秒
	if !token.WaitTimeout(3 * time.Second) {
		return fmt.Errorf("publish timeout")
	}

	if err := token.Error(); err != nil {
		return fmt.Errorf("publish failed: %v", err)
	}

	return nil
}
func (ggg *MqttClient) Subscribe(topic string, qos int, callback mqtt.MessageHandler) error {
	if !ggg.Client.IsConnected() {
		return fmt.Errorf("MQTT客户端未连接")
	}
	// 校验 QoS 合法性
	if qos < 0 || qos > 2 {
		return fmt.Errorf("invalid QoS level: %d", qos)
	}
	// 如果没有提供回调函数，使用默认处理
	if callback == nil {
		callback = func(client mqtt.Client, msg mqtt.Message) {
			opts1 := client.OptionsReader()
			log.Printf("[%s] 收到订阅消息: 主题=%s, 内容=%s\n", opts1.Username(), msg.Topic(), msg.Payload())
		}
	}
	if ggg.Subscriptions == nil {
		ggg.Subscriptions = make(map[string]byte)
	}
	// 存储订阅信息以便重连后恢复
	ggg.reconnectLock.Lock()
	ggg.Subscriptions[topic] = byte(qos)
	ggg.reconnectLock.Unlock()

	token := ggg.Client.Subscribe(topic, byte(qos), callback)
	if token.Wait() && token.Error() != nil {
		return fmt.Errorf("订阅失败: %v", token.Error())
	}

	return nil
}
func (m *MqttClient) Unsubscribe(topic string) error {
	m.reconnectLock.Lock()
	delete(m.Subscriptions, topic)
	m.reconnectLock.Unlock()

	token := m.Client.Unsubscribe(topic)
	token.Wait()
	return token.Error()
}
func (ggg *MqttClient) Disconnect() { // 断开连接
	ggg.reconnectLock.Lock()
	defer ggg.reconnectLock.Unlock()
	ggg.Reconnecting = false

	if ggg.Client.IsConnected() {
		ggg.Client.Disconnect(250) // 优雅断开连接，等待250ms
		log.Printf("客户端 %s 已断开连接\n", ggg.Username)
	}
}

// 开始重连逻辑
func (m *MqttClient) startReconnect() {
	m.reconnectLock.Lock()
	defer m.reconnectLock.Unlock()

	// 防止重复启动重连
	if m.Reconnecting {
		return
	}

	m.Reconnecting = true

	// 启动重连协程
	go func() {
		retryCount := 0
		maxRetries := 10

		for {
			// 指数退避策略
			retryCount++
			d := 0
			if retryCount*2 < 30 {
				d = retryCount * 2
			} else {
				d = 30
			}
			backoffTime := time.Second * time.Duration(d)
			log.Printf("重连尝试 %d/%d，%s 后重试\n", retryCount, maxRetries, backoffTime)
			time.Sleep(backoffTime)

			m.reconnectLock.Lock()
			if !m.Reconnecting {
				m.reconnectLock.Unlock()
				return
			}
			m.reconnectLock.Unlock()

			log.Printf("尝试重连到 %s...\n", m.Broker)

			// 创建新客户端实例
			m.Client = mqtt.NewClient(m.opts)
			token := m.Client.Connect()
			token.Wait()

			if token.Error() == nil {
				log.Printf("重连成功\n")
				m.Reconnecting = false
				return
			}

			log.Printf("重连失败: %v\n", token.Error())

			if retryCount >= maxRetries {
				log.Printf("达到最大重连次数，停止重连\n")
				m.Reconnecting = false
				return
			}
		}
	}()
}
