package models

import (
	"encoding/binary"
	"errors"
	"fmt"
	"log"
	"net"
	"sync"
	"time"

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

// MqttServerOrWs
type MqttServerOrWs struct {
	Client     *websocket.Conn
	User       string
	Fd         uint32
	ClientId   string
	MqttClient MqttClientZ
}

var (
	MqttA        = &MqttAnalysis{}
	WsConnetPool = make(map[uint32]*MqttServerOrWs) // WebSocket连接池
	RWlock       = sync.RWMutex{}                   // 读写锁
	MqttHost     = "localhost:1883"
)

func NewMqttServerOrWs(conn *websocket.Conn, fd uint32, mc MqttClientZ) *MqttServerOrWs {
	client := &MqttServerOrWs{
		Client:     conn,
		User:       "",
		Fd:         fd,
		ClientId:   "",
		MqttClient: mc,
	}
	return client
}

func (ggg *MqttServerOrWs) HandleMessages() {
	bytesAdd := make([]byte, 0) //  用于存储读取的二进制数据
	defer func() {
		if r := recover(); r != nil {
			log.Println("Recovered from panic:", r)
			// 确保所有锁都被释放
			if RWlock.TryRLock() {
				RWlock.RUnlock()
			}
			// 清理WebSocket连接池中的资源
			if wsClient, exists := WsConnetPool[ggg.Fd]; exists {
				fmt.Println("清理WebSocket连接")
				if wsClient.Client != nil {
					wsClient.Client.Close()
				}
				delete(WsConnetPool, ggg.Fd)
			}
			// 清理MQTT客户端
			if ggg.MqttClient.Client != nil {
				ggg.MqttClient.Client.Disconnect(250) // 给MQTT客户端250ms时间完成操作
				ggg.MqttClient.Client = nil
			}
			// 标记连接状态为关闭
			ggg.Client = nil
		}
	}()
	for {
		if ggg.Client == nil {
			RWlock.RUnlock()
			log.Println("连接已关闭，退出处理循环")
			break
		}
		// 设置读取超时
		if err := ggg.Client.SetReadDeadline(time.Now().Add(90 * time.Second)); err != nil {
			if errors.Is(err, net.ErrClosed) {
				log.Println("连接已关闭，退出处理循环1")
			} else {
				log.Printf("设置读取超时失败: %v", err)
			}
			ggg.clearup()
			break
		}
		// 读取消息
		RWlock.RLock()
		if ggg.Client == nil {
			RWlock.RUnlock()
			log.Println("连接已关闭，退出处理循环")
			break
		}
		tmpClient := ggg.Client
		RWlock.RUnlock()
		// 读取消息
		_, message, err := tmpClient.ReadMessage()

		if err != nil {
			// 清理连接
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				log.Printf("意外关闭错误: %v", err)
				ggg.clearup()
				break
			}
			continue
		}

		bytesAdd = append(bytesAdd, message...)
		if len(bytesAdd) < 2 {
			//固定报头 log.Println("MQTT报文不完整，继续读取")
			continue
		}
		// 解析报文类型
		header := bytesAdd[0] >> 4
		remainingLength := int(bytesAdd[1])

		// 检查剩余长度是否足够解析可变报头
		if len(bytesAdd) < 2+remainingLength {
			//log.Println("MQTT报文不完整，继续读取")
			continue
		}
		//fmt.Printf("报文读取完整, 报文类型: %d 剩余长度:%d \n", packetType, remainingLength)
		// 解析剩余长度
		remainingLength, lengthBytes, err := MqttA.DecodeRemainingLength(bytesAdd[1:])
		if err != nil {
			log.Printf("解析剩余长度失败: %v", err)
			bytesAdd = make([]byte, 0)
			continue
		}
		payloadStart := 1 + lengthBytes
		// 处理不同类型的MQTT报文
		switch header {
		case 1: // CONNECT
			connectData, err := MqttA.ParseConnectPacket(bytesAdd)
			if err != nil {
				log.Printf("解析CONNECT报文失败: %v", err)
				connack := MqttA.GenerateConnackPacket(false, 0x01) // 不支持的协议版本
				ggg.Client.WriteMessage(websocket.BinaryMessage, connack)
				bytesAdd = make([]byte, 0)
				continue
			}

			//log.Printf("收到CONNECT报文: %v", connectData)
			ggg.ClientId = connectData["ClientID"]
			ggg.User = connectData["Username"]
			password := connectData["Password"]
			//log.Printf("账号:%s 密码:%s", ggg.User, password, WsConnetPool)
			bytesAdd = make([]byte, 0)
			mqttClient := ggg.MqttClient.Connect(MqttHost, ggg.ClientId, ggg.User, password)
			if mqttClient != nil {
				fmt.Println("账号密码错误")
				connack := MqttA.GenerateConnackPacket(false, 0x01) //发送CONNACK响应 连接断开
				ggg.Client.WriteMessage(websocket.BinaryMessage, connack)
				return
			}
			connack := MqttA.GenerateConnackPacket(false, 0x00) //发送CONNACK响应 连接已接受
			if err := ggg.Client.WriteMessage(websocket.BinaryMessage, connack); err != nil {
				fmt.Printf("发送CONNACK失败: %v", err)
				return
			}
			log.Println("MQTT连接成功")
			RWlock.Lock()
			WsConnetPool[ggg.Fd] = ggg
			RWlock.Unlock()
		case 3:
			// 解析PUBLISH报文
			pos := payloadStart
			// 解析主题名长度
			topicLength := binary.BigEndian.Uint16(bytesAdd[pos : pos+2])
			pos += 2
			// 解析主题名
			topic := string(bytesAdd[pos : pos+int(topicLength)])
			pos += int(topicLength)

			// 解析包ID（如果QoS > 0）
			var packetID uint16
			qos := (header >> 1) & 0x03
			if qos > 0 {
				packetID = binary.BigEndian.Uint16(bytesAdd[pos : pos+2])
			}
			// 解析负载
			payload := bytesAdd[pos : payloadStart+remainingLength]
			log.Printf("收到 推送 报文: 主题:%s, QoS:%d, 包ID:%d, 内容=%s", topic, qos, packetID, string(payload))

			// 处理QoS确认
			if qos == 1 {
				// 发送PUBACK
				puback := []byte{
					0x40, // PUBACK报文类型
					0x02, // 剩余长度
					byte(packetID >> 8),
					byte(packetID & 0xFF),
				}
				if err := ggg.Client.WriteMessage(websocket.BinaryMessage, puback); err != nil {
					log.Printf("发送PUBACK失败: %v", err)
				}
			} else if qos == 2 {
				// TODO: 处理QoS 2流程
				log.Println("收到QoS 2的PUBLISH，暂不处理")
			}

			// 转发消息到订阅者
			ggg.forwardMessageToSubscribers(topic, payload, 0)
			bytesAdd = make([]byte, 0)

		case 8: //订阅主题
			// 解析SUBSCRIBE报文
			pos := payloadStart
			// 解析包ID
			packetID := binary.BigEndian.Uint16(bytesAdd[pos : pos+2])
			pos += 2
			topic := ""
			// 解析订阅列表
			for pos < payloadStart+remainingLength {
				// 解析主题过滤器长度
				filterLength := binary.BigEndian.Uint16(bytesAdd[pos : pos+2])
				pos += 2
				// 解析主题过滤器
				topic = string(bytesAdd[pos : pos+int(filterLength)])
				pos += int(filterLength)
				// 解析最大QoS
				maxQoS := bytesAdd[pos] & 0x03
				pos++
				log.Printf("收到订阅报文: 包ID:%d, 主题:%s, QoS=%d",
					packetID, topic, maxQoS)
				if ggg.MqttClient.Topic == "" { // 处理订阅逻辑
					ggg.MqttClient.Topic = topic
					ggg.ServerMsg()
				}
			}

			suback := MqttA.GenerateSubackPacket(packetID, []byte{0x00}) // 发送订阅成功响应
			if err := ggg.Client.WriteMessage(websocket.BinaryMessage, suback); err != nil {
				log.Printf("发送SUBACK失败: %v", err)
			}
			bytesAdd = make([]byte, 0)

		case 12:
			//fmt.Println("PINGREQ", header)
			pingresp := []byte{0xD0, 0x00} //发送PINGRESP响应 PINGRESP报文
			if err := ggg.Client.WriteMessage(websocket.BinaryMessage, pingresp); err != nil {
				log.Printf("发送PINGRESP失败: %v", err)
			}
			bytesAdd = make([]byte, 0)

		case 14: // DISCONNECT
			log.Printf("收到DISCONNECT关闭连接报文: ClientID=%s", ggg.ClientId)
			connack := MqttA.GenerateConnackPacket(false, 0x01) //发送CONNACK响应 连接断开
			RWlock.Lock()
			if err := ggg.Client.WriteMessage(websocket.BinaryMessage, connack); err != nil {
				log.Printf("发送DISCONNECT失败: %v", err)
			}
			RWlock.Unlock()
			ggg.clearup()
			return

		default:
			log.Printf("未知MQTT报文类型: %d", header)
			bytesAdd = make([]byte, 0)
		}
	}
}

// 转发消息到订阅者
func (ggg *MqttServerOrWs) forwardMessageToSubscribers(topic string, payload []byte, qos byte) {
	RWlock.Lock()
	if ggg.Client == nil {
		RWlock.Unlock()
		return
	}
	// 这里应该实现实际的消息转发逻辑
	a, _ := MqttA.BuildPublishMessage(topic, payload, qos, false) // 生成MQTT PUBLISH报文
	err := ggg.Client.WriteMessage(websocket.BinaryMessage, a)
	if err != nil {
		RWlock.Unlock()
		log.Println("转发消息到订阅者失败")
		ggg.clearup()
		return
	}
	err1 := ggg.MqttClient.PublishZ(topic, payload, 0, false)
	if err1 != nil {
		RWlock.Unlock()
		log.Println("推送往MQTT服务器失败", err1.Error())
		ggg.clearup()
		return
	}
	log.Println("推送往MQTT服务器成功")
	RWlock.Unlock()
}

func (ggg *MqttServerOrWs) ServerMsg() {
	// 启动消息订阅处理
	go func(ggg *MqttServerOrWs) {
		// 处理goroutine内的panic
		defer func() {
			if err := recover(); err != nil {
				log.Printf("MQTT订阅goroutine发生panic: %v", err)
				ggg.clearup()
			}
		}()

		// 检查MQTT客户端是否已连接
		if !ggg.MqttClient.Client.IsConnected() {
			log.Println("MQTT客户端未连接，无法订阅主题")
			ggg.clearup()
			return
		}

		// 执行订阅并处理错误
		token := ggg.MqttClient.Client.Subscribe(ggg.MqttClient.Topic, 0, func(client mqtt.Client, msg mqtt.Message) {
			// 处理回调函数内的panic
			defer func() {
				if err := recover(); err != nil {
					log.Printf("MQTT消息处理回调发生panic: %v", err)
					ggg.clearup()
				}
			}()

			// 构建并发送消息到WebSocket客户端
			payload, err := MqttA.BuildPublishMessage(ggg.MqttClient.Topic, msg.Payload(), 0, false)

			if err != nil {
				log.Printf("构建发布消息失败: %v", err)
				ggg.clearup()
				return
			}

			// 写入WebSocket并检查错误
			if err := ggg.Client.WriteMessage(websocket.BinaryMessage, payload); err != nil {
				log.Printf("发送消息到WebSocket失败: %v", err)
				// 通常发生错误时应该关闭连接
				ggg.clearup()
				return
			}

			log.Printf("收到订阅消息: %s", string(msg.Payload()))
		},
		)

		// 等待订阅完成并检查结果
		if token.Wait() && token.Error() != nil {
			log.Printf("订阅主题失败: %v", token.Error())
			ggg.clearup()
			return
		}

		log.Printf("成功订阅主题: %s", ggg.MqttClient.Topic)

	}(ggg)

}
func (ggg *MqttServerOrWs) clearup() {
	RWlock.Lock()
	defer RWlock.Unlock()

	if wsClient, exists := WsConnetPool[ggg.Fd]; exists {
		if wsClient.Client != nil {
			wsClient.Client.Close()
		}
		delete(WsConnetPool, ggg.Fd)
	}
	if ggg.MqttClient.Client != nil {
		ggg.MqttClient.Client.Disconnect(0)
	}
}
