FireLeave_tool/connect/ws_channel.go

package connect

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"time"

	"FireLeave_tool/logger"
	"github.com/gorilla/websocket"
)

// Global variables
var (
	wsClient         *WSClient
	wsMutex          sync.RWMutex
	isConnecting     bool
	wsURL            = "ws://172.17.0.1:18080/ws" // Updated to specified address
	deviceDataMutex  sync.Mutex
	ConfigUpdateChan chan DeviceData
	currentServiceID string // New: current service's server_id
)

type PublishParams struct {
	Topic   string          `json:"topic"`
	Data    json.RawMessage `json:"data,omitempty"`
	Content json.RawMessage `json:"content,omitempty"`
}

const (
	wsFrameRequest byte = 0x01
	wsFrameReply   byte = 0x02
	wsFrameCast    byte = 0x03
)

type InferenceHeartbeat struct {
	DeviceUUID  string  `json:"device_uuid"`
	PersonCount int     `json:"person_count"`
	Temperature float64 `json:"temperature,omitempty"`
	Timestamp   int64   `json:"timestamp"` // 上报时间戳
}

func GetWSClient() *WSClient {
	wsMutex.RLock()
	defer wsMutex.RUnlock()
	return wsClient
}

func (c *WSClient) IsConnected() bool {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	return c.connected
}

// DeviceData struct
type DeviceData struct {
	DeviceUUID           string      `json:"device_uuid"`
	TaskID               string      `json:"task_id"`
	CameraRTSP           string      `json:"camera_rtsp"`
	CameraIP             string      `json:"camera_ip"`
	Confidence           int         `json:"confidence"`
	DetectArea           []Area      `json:"detect_area"`
	DetectAreaStr        string      `json:"detect_area_str,omitempty"`
	PersonCount          int         `json:"person_count"`
	DetectionTime        json.Number `json:"detection_time"` // Using json.Number
	AlarmTime            json.Number `json:"alarm_time"`     // Using json.Number
	TemperatureThreshold float64     `json:"temperature_threshold"`
	Temperature          float64     `json:"temperature"`
	UploadURL            string      `json:"upload_url"`
	WSConn               *websocket.Conn
	CameraChannel        int    `json:"camera_channel"`
	LastAlarmTime        int64  `json:"-"` // Last alarm time
	AlarmPauseUntil      int64  `json:"-"` // Pause alarm untilw
	HostUUID             string `json:"host_uuid"`
	Address              string `json:"address"`
	// New fields for new configuration format
	FireLeaveDetectArea []Area         `json:"fire_leave_detect_area,omitempty"`
	FireCheckDetectArea []Area         `json:"fire_check_detect_area,omitempty"`
	FireLeaveParam      FireLeaveParam `json:"fire_leave_param,omitempty"`
	FireCheckParam      FireCheckParam `json:"fire_check_param,omitempty"`
}

type InferenceParams struct {
	DeviceUUID    string `json:"device_uuid"`
	CameraRTSP    string `json:"camera_rtsp"`
	CameraChannel int    `json:"camera_channel"`
	Confidence    int    `json:"confidence"`
	DetectArea    string `json:"detect_area"` // String format detection area
	Port          int    `json:"port"`        // HTTP server port
}

// WSMessage struct
type WSMessage struct {
	ID     int64           `json:"id,omitempty"`
	Method string          `json:"method"`
	Params json.RawMessage `json:"params,omitempty"`
	Result json.RawMessage `json:"result,omitempty"` // Modified: string -> json.RawMessage
	Error  *WSError        `json:"error,omitempty"`
}

type WSError struct {
	Code    int    `json:"code"`
	Message string `json:"message"`
}

type ServiceReply struct {
	PacketID uint32
	Result   []byte
	Error    *WSError
}

type RegisterParams struct {
	ContainerName string                 `json:"container_name,omitempty"`
	ServiceID     string                 `json:"service_id,omitempty"`
	MetaData      map[string]interface{} `json:"meta_data,omitempty"` // New: metadata field
}

type SubscribeParams struct {
	Topic string `json:"topic"`
}

type WSClient struct {
	Conn          *websocket.Conn
	url           string
	serviceID     string
	serverID      string
	connected     bool
	mutex         sync.Mutex
	messageID     uint32
	callbacks     map[uint32]chan *ServiceReply
	subscribers   map[string]func([]byte)
	closeChan     chan struct{}
	reconnectFlag bool
	reconnecting  bool
}

// NewWSClient creates WebSocket client
func NewWSClient(url string) (*WSClient, error) {
	conn, _, err := websocket.DefaultDialer.Dial(url, nil)
	if err != nil {
		return nil, err
	}

	client := &WSClient{
		Conn:          conn,
		url:           url,
		connected:     true,
		messageID:     1,
		callbacks:     make(map[uint32]chan *ServiceReply),
		subscribers:   make(map[string]func([]byte)),
		closeChan:     make(chan struct{}),
		reconnectFlag: true,
	}

	go client.readLoop()
	go client.heartbeatLoop()
	return client, nil
}

func (c *WSClient) connect() error {
	conn, _, err := websocket.DefaultDialer.Dial(c.url, nil)
	if err != nil {
		return err
	}
	c.Conn = conn
	c.connected = true
	go c.readLoop()
	go c.heartbeatLoop()
	return nil
}

// InitWSChannel initializes WebSocket
func InitWSChannel() error {
	wsMutex.Lock()
	defer wsMutex.Unlock()

	if wsClient != nil && wsClient.connected {
		logger.Logger.Printf("WebSocket connection already exists")
		return nil
	}

	if isConnecting {
		logger.Logger.Printf("WebSocket connection is being established")
		return fmt.Errorf("WebSocket connection is being established")
	}

	isConnecting = true

	logger.Logger.Printf("Starting WebSocket connection initialization: %s", wsURL)

	// Connect WebSocket asynchronously to avoid blocking
	go func() {
		client, err := NewWSClient(wsURL)
		if err != nil {
			logger.Logger.Printf("Failed to create WebSocket client: %v", err)
			wsMutex.Lock()
			isConnecting = false
			wsMutex.Unlock()
			return
		}

		wsMutex.Lock()
		defer wsMutex.Unlock()
		if ConfigUpdateChan == nil {
			ConfigUpdateChan = make(chan DeviceData, 100)
		}

		// Note: No longer calling LoadDeviceData() here to avoid duplicate loading
		wsClient = client
		isConnecting = false
		logger.Logger.Printf("WebSocket connection initialized successfully (asynchronous)")
	}()

	return nil
}

// New: function to generate server_id
func generateServiceID() string {
	return fmt.Sprintf("fireleave-service-%d", time.Now().UnixNano())
}

// New: set current server_id
//func SetServerID(serverID string) {
//	currentServerID = serverID
//}

func GetServiceID() string {
	if currentServiceID == "" {
		currentServiceID = generateServiceID()
	}
	return currentServiceID
}

// RegisterService service registration - modified: now requires providing own server_id
func RegisterService(containerName string) (string, error) {
	client := GetWSClient()
	if client == nil {
		return "", fmt.Errorf("WebSocket client not initialized")
	}

	serviceID := GetServiceID()

	params := RegisterParams{
		ContainerName: containerName,
		ServiceID:     serviceID,
		MetaData: map[string]interface{}{ // Add metadata
			"version": "1.0",
			"type":    "fire_leave",
		},
	}
	logger.Logger.Printf("Starting service registration: ContainerName=%s, ServiceID=%s", containerName, serviceID)
	response, err := client.sendRequest("register", params, true)
	if err != nil {
		logger.Logger.Printf("Service registration failed: %v", err)
		return "", err
	}
	if response.Error != nil {
		logger.Logger.Printf("Service registration error: Code=%d, Message=%s", response.Error.Code, response.Error.Message)
		return "", fmt.Errorf("Service registration error: %s", response.Error.Message)
	}

	if string(response.Result) != "ok" {
		logger.Logger.Printf("Service registration failed: Expected result 'ok', got '%s'", string(response.Result))
		return "", fmt.Errorf("Service registration failed: Server returned abnormal response")
	}

	client.mutex.Lock()
	client.serviceID = serviceID
	client.mutex.Unlock()
	logger.Logger.Printf("Service registration successful: ServiceID=%s", serviceID)
	return serviceID, nil
}

// SubscribeTopic topic subscription
func SubscribeTopic(topic string, callback func([]byte)) error {
	client := GetWSClient()
	if client == nil {
		return fmt.Errorf("WebSocket client not initialized")
	}
	params := SubscribeParams{
		Topic: topic,
	}
	logger.Logger.Printf("Starting topic subscription: Topic=%s", topic)
	response, err := client.sendRequest("subscribe", params, true)
	if err != nil {
		logger.Logger.Printf("Topic subscription failed: %v", err)
		return err
	}
	if response.Error != nil {
		logger.Logger.Printf("Topic subscription error: Code=%d, Message=%s", response.Error.Code, response.Error.Message)
		return fmt.Errorf("Topic subscription error: %s", response.Error.Message)
	}
	client.mutex.Lock()
	client.subscribers[topic] = callback
	client.mutex.Unlock()
	logger.Logger.Printf("Topic subscription successful: Topic=%s", topic)
	return nil
}

func (c *WSClient) SendRawMessage(message interface{}) error {
	messageData, err := json.Marshal(message)
	if err != nil {
		logger.Logger.Printf("[WebSocket Send Full JSON] marshal failed: %v", err)
	} else {
		logger.Logger.Printf("[WebSocket Send Full JSON] %s", string(messageData))
	}

	if metric, ok := extractMetricData(message); ok {
		return c.sendMetricData(metric.routeKey, metric.metric)
	}
	return fmt.Errorf("unsupported websocket message type for protobuf protocol")
}

type metricDataMessage struct {
	routeKey []byte
	metric   []byte
}

func extractMetricData(message interface{}) (metricDataMessage, bool) {
	msg, ok := message.(map[string]interface{})
	if !ok || msg["method"] != "metric_data" {
		return metricDataMessage{}, false
	}
	params, ok := msg["params"].(map[string]interface{})
	if !ok {
		return metricDataMessage{}, false
	}
	routeKey, ok := params["route_key"].(string)
	if !ok || routeKey == "" {
		return metricDataMessage{}, false
	}
	metric, ok := params["metric"].(string)
	if !ok {
		return metricDataMessage{}, false
	}
	return metricDataMessage{
		routeKey: []byte(routeKey),
		metric:   []byte(metric),
	}, true
}

func (c *WSClient) sendBinaryFrame(frameType byte, payload []byte) error {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	if !c.connected {
		return fmt.Errorf("WebSocket connection disconnected")
	}
	frame := make([]byte, 1+len(payload))
	frame[0] = frameType
	copy(frame[1:], payload)
	logger.Logger.Printf("Sending WebSocket binary frame: type=0x%02x, payload=%d bytes", frameType, len(payload))
	err := c.Conn.WriteMessage(websocket.BinaryMessage, frame)
	if err != nil {
		logger.Logger.Printf("WebSocket message sending failed: %v", err)
		c.connected = false
		c.startReconnectLocked()
		return fmt.Errorf("Message sending failed: %v", err)
	}
	return nil
}

func (c *WSClient) sendMetricData(routeKey, metric []byte) error {
	payload := encodeServiceCastMetricData(routeKey, metric)
	return c.sendBinaryFrame(wsFrameCast, payload)
}

// sendRequest sends request
func (c *WSClient) sendRequest(method string, params interface{}, waitResponse bool) (*ServiceReply, error) {
	c.mutex.Lock()
	id := c.messageID
	c.messageID++
	c.mutex.Unlock()

	var payload []byte
	switch method {
	case "register":
		registerParams, ok := params.(RegisterParams)
		if !ok {
			return nil, fmt.Errorf("invalid register params")
		}
		payload = encodeServiceRequestRegister(id, registerParams.ServiceID)
	case "subscribe":
		subscribeParams, ok := params.(SubscribeParams)
		if !ok {
			return nil, fmt.Errorf("invalid subscribe params")
		}
		payload = encodeServiceRequestSubscribe(id, subscribeParams.Topic)
	default:
		return nil, fmt.Errorf("unsupported request method: %s", method)
	}

	if waitResponse {
		callback := make(chan *ServiceReply, 1)
		c.mutex.Lock()
		c.callbacks[id] = callback
		c.mutex.Unlock()

		defer func() {
			c.mutex.Lock()
			delete(c.callbacks, id)
			c.mutex.Unlock()
			logger.Logger.Printf("Cleaning up callback channel: ID=%d", id)
		}()

		if err := c.sendBinaryFrame(wsFrameRequest, payload); err != nil {
			return nil, err
		}

		logger.Logger.Printf("Waiting for response: ID=%d", id)
		select {
		case response := <-callback:
			logger.Logger.Printf("Received response: ID=%d", id)
			return response, nil
		case <-time.After(30 * time.Second):
			logger.Logger.Printf("Response timeout: ID=%d", id)
			return nil, fmt.Errorf("Response timeout")
		case <-c.closeChan:
			logger.Logger.Printf("Connection closed, stopping response wait: ID=%d", id)
			return nil, fmt.Errorf("Connection closed")
		}
	}
	return nil, c.sendBinaryFrame(wsFrameRequest, payload)
}

// readLoop message reading loop
func (c *WSClient) readLoop() {
	logger.Logger.Printf("Starting WebSocket message reading loop")
	defer func() {
		if r := recover(); r != nil {
			logger.Logger.Printf("readLoop panic: %v", r)
		}
		// Important: do not close(closeChan)!
		logger.Logger.Printf("WebSocket read loop exited, connection marked as disconnected")
	}()

	for {
		messageType, messageData, err := c.Conn.ReadMessage()
		if err != nil {
			logger.Logger.Printf("WebSocket message reading failed: %v", err)
			c.mutex.Lock()
			c.connected = false
			c.mutex.Unlock()
			c.startReconnect() // Trigger reconnection
			return
		}

		if messageType != websocket.BinaryMessage {
			logger.Logger.Printf("[WebSocket Received Raw] non-binary type=%d size=%d payload=%s", messageType, len(messageData), string(messageData))
			continue
		}
		if len(messageData) == 0 {
			logger.Logger.Printf("Ignoring empty WebSocket binary frame")
			continue
		}

		if string(messageData) == "pong" {
			logger.Logger.Printf("Received pong response")
			continue
		}

		frameType := messageData[0]
		payload := messageData[1:]
		logger.Logger.Printf("Received WebSocket binary frame: type=0x%02x, payload=%d bytes", frameType, len(payload))

		switch frameType {
		case wsFrameReply:
			reply, err := decodeServiceReply(payload)
			if err != nil {
				logger.Logger.Printf("WebSocket reply parsing failed: %v", err)
				continue
			}
			logServiceReply(reply)
			c.mutex.Lock()
			if callback, exists := c.callbacks[reply.PacketID]; exists {
				select {
				case callback <- reply:
					logger.Logger.Printf("Successfully sent reply to callback channel: ID=%d", reply.PacketID)
				default:
					logger.Logger.Printf("Callback channel full, discarding reply: ID=%d", reply.PacketID)
				}
			} else {
				logger.Logger.Printf("[WebSocket Received Reply] no waiting callback for packet_id=%d", reply.PacketID)
			}
			c.mutex.Unlock()
		case wsFrameCast:
			if err := c.handleServiceCast(payload); err != nil {
				logger.Logger.Printf("WebSocket cast handling failed: %v", err)
			}
		default:
			logger.Logger.Printf("Ignoring unknown WebSocket frame type: 0x%02x", frameType)
			continue
		}
	}
}

// handlePublishMessage handles publish messages
func (c *WSClient) handlePublishMessage(message WSMessage) {
	logger.Logger.Printf("Processing publish message")
	var publishParams PublishParams
	if err := json.Unmarshal(message.Params, &publishParams); err != nil {
		logger.Logger.Printf("Failed to parse publish message parameters: %v", err)
		return
	}
	logger.Logger.Printf("WebSocket 收到 publish 消息 → Topic: %s | 原始数据: %s",
		publishParams.Topic, string(publishParams.Data))

	if publishParams.Topic != "" {
		c.mutex.Lock()
		callback, exists := c.subscribers[publishParams.Topic]
		c.mutex.Unlock()
		if exists && callback != nil {
			callback(message.Params)
		}
	}

	if publishParams.Data != nil {
		logger.Logger.Printf("Ignoring publish.data device configuration payload; service.conf is managed by startup config only")
	}
}

func (c *WSClient) handleServiceCast(payload []byte) error {
	topic, content, ok, err := decodeServiceCastTopicEvent(payload)
	if err != nil {
		return err
	}
	if !ok {
		logger.Logger.Printf("Ignoring unsupported ServiceCast body")
		return nil
	}

	logger.Logger.Printf("WebSocket received topic_event: Topic=%s | Content=%s", topic, string(content))
	c.mutex.Lock()
	callback, exists := c.subscribers[topic]
	c.mutex.Unlock()
	if !exists || callback == nil {
		logger.Logger.Printf("No subscriber for topic_event: Topic=%s", topic)
		return nil
	}

	wrapped, err := wrapTopicEvent(topic, content)
	if err != nil {
		return err
	}
	callback(wrapped)
	return nil
}

func wrapTopicEvent(topic string, content []byte) ([]byte, error) {
	msg := map[string]interface{}{
		"topic": topic,
	}
	if json.Valid(content) {
		msg["content"] = json.RawMessage(content)
	} else {
		msg["content"] = string(content)
	}
	return json.Marshal(msg)
}

// heartbeatLoop heartbeat loop - fixed version
func (c *WSClient) heartbeatLoop() {
	logger.Logger.Printf("Starting WebSocket heartbeat loop")
	ticker := time.NewTicker(25 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			c.mutex.Lock()
			connected := c.connected
			conn := c.Conn
			c.mutex.Unlock()

			if connected && conn != nil {
				err := conn.WriteMessage(websocket.PingMessage, []byte{})
				if err != nil {
					logger.Logger.Printf("Failed to send ping: %v", err)
					c.mutex.Lock()
					c.connected = false
					c.mutex.Unlock()
					c.startReconnect()
				} else {
					logger.Logger.Printf("Ping sent successfully")
				}
			}
		case <-c.closeChan:
			logger.Logger.Printf("Heartbeat loop received close signal, exiting")
			ticker.Stop() // Explicitly stop ticker
			return
		}
	}
}

func (c *WSClient) reRegisterAndResubscribe() {
	// 1. 等待连接稳定
	time.Sleep(1 * time.Second)

	// 2. 重新注册
	containerName := "fireleave-container"
	serviceID, err := RegisterService(containerName)
	if err != nil {
		logger.Logger.Printf("Failed to re-register service after reconnect: %v", err)
		return
	}
	logger.Logger.Printf("Re-registration successful: ServiceID=%s", serviceID)

	// 3. 重新订阅
	c.mutex.Lock()
	subscribers := make(map[string]func([]byte), len(c.subscribers))
	for topic, callback := range c.subscribers {
		subscribers[topic] = callback
	}
	c.mutex.Unlock()
	for topic, callback := range subscribers {
		if err := SubscribeTopic(topic, callback); err != nil {
			logger.Logger.Printf("Failed to re-subscribe topic %s: %v", topic, err)
		} else {
			logger.Logger.Printf("Re-subscribed to topic: %s", topic)
		}
	}
}

func (c *WSClient) startReconnect() {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.startReconnectLocked()
}

func (c *WSClient) startReconnectLocked() {
	if !c.reconnectFlag || c.reconnecting {
		return
	}
	c.reconnecting = true
	go c.reconnect()
}

// connect/wschannel.go → reconnect complete fix
func (c *WSClient) reconnect() {
	defer func() {
		c.mutex.Lock()
		c.reconnecting = false
		c.mutex.Unlock()
	}()

	c.mutex.Lock()
	reconnectFlag := c.reconnectFlag
	c.mutex.Unlock()
	if !reconnectFlag {
		return
	}

	for i := 0; i < 12; i++ {
		delay := time.Duration(1<<uint(i)) * time.Second
		if delay > 30*time.Second {
			delay = 30 * time.Second
		}
		time.Sleep(delay)

		conn, _, err := websocket.DefaultDialer.Dial(c.url, nil)
		if err != nil {
			logger.Logger.Printf("Reconnection attempt %d/12 failed: %v", i+1, err)
			continue
		}

		c.mutex.Lock()
		c.Conn = conn
		c.connected = true
		c.mutex.Unlock()

		logger.Logger.Printf("WebSocket reconnection successful!")

		// 关键：重连后必须先 register！
		go c.reRegisterAndResubscribe()

		// 重启读写循环
		go c.readLoop()
		return
	}

	logger.Logger.Printf("Reconnection failed after 12 attempts")
	c.mutex.Lock()
	c.reconnectFlag = false
	c.mutex.Unlock()
}
func (d *DeviceData) ToInferenceParams(port int) InferenceParams {
	return InferenceParams{
		DeviceUUID:    d.DeviceUUID,
		CameraRTSP:    d.CameraRTSP,
		CameraChannel: d.CameraChannel,
		Confidence:    d.Confidence,
		DetectArea:    ProcessDetectAreaForInference(d.DetectArea),
		Port:          port,
	}
}

func appendVarint(dst []byte, v uint64) []byte {
	for v >= 0x80 {
		dst = append(dst, byte(v)|0x80)
		v >>= 7
	}
	return append(dst, byte(v))
}

func appendBytesField(dst []byte, fieldNumber int, value []byte) []byte {
	dst = appendVarint(dst, uint64(fieldNumber<<3|2))
	dst = appendVarint(dst, uint64(len(value)))
	return append(dst, value...)
}

func appendVarintField(dst []byte, fieldNumber int, value uint64) []byte {
	dst = appendVarint(dst, uint64(fieldNumber<<3))
	return appendVarint(dst, value)
}

func encodeServiceRequestRegister(packetID uint32, serviceID string) []byte {
	register := appendBytesField(nil, 1, []byte(serviceID))
	payload := appendVarintField(nil, 1, uint64(packetID))
	payload = appendBytesField(payload, 10, register)
	return payload
}

func encodeServiceRequestSubscribe(packetID uint32, topic string) []byte {
	subscribe := appendBytesField(nil, 1, []byte(topic))
	payload := appendVarintField(nil, 1, uint64(packetID))
	payload = appendBytesField(payload, 11, subscribe)
	return payload
}

func encodeServiceCastMetricData(routeKey, metric []byte) []byte {
	metricData := appendBytesField(nil, 1, routeKey)
	metricData = appendBytesField(metricData, 2, metric)
	return appendBytesField(nil, 11, metricData)
}

func consumeVarint(data []byte, pos *int) (uint64, error) {
	var value uint64
	var shift uint
	for *pos < len(data) {
		b := data[*pos]
		*pos += 1
		value |= uint64(b&0x7f) << shift
		if b < 0x80 {
			return value, nil
		}
		shift += 7
		if shift >= 64 {
			return 0, fmt.Errorf("protobuf varint overflow")
		}
	}
	return 0, fmt.Errorf("unexpected end of protobuf varint")
}

func consumeBytes(data []byte, pos *int) ([]byte, error) {
	length, err := consumeVarint(data, pos)
	if err != nil {
		return nil, err
	}
	if length > uint64(len(data)-*pos) {
		return nil, fmt.Errorf("protobuf bytes field length out of range")
	}
	value := data[*pos : *pos+int(length)]
	*pos += int(length)
	return value, nil
}

func skipField(data []byte, pos *int, wireType int) error {
	switch wireType {
	case 0:
		_, err := consumeVarint(data, pos)
		return err
	case 2:
		_, err := consumeBytes(data, pos)
		return err
	default:
		return fmt.Errorf("unsupported protobuf wire type: %d", wireType)
	}
}

func decodeServiceReply(payload []byte) (*ServiceReply, error) {
	reply := &ServiceReply{}
	for pos := 0; pos < len(payload); {
		key, err := consumeVarint(payload, &pos)
		if err != nil {
			return nil, err
		}
		fieldNumber := int(key >> 3)
		wireType := int(key & 0x7)
		switch fieldNumber {
		case 1:
			if wireType != 0 {
				return nil, fmt.Errorf("invalid packet_id wire type: %d", wireType)
			}
			packetID, err := consumeVarint(payload, &pos)
			if err != nil {
				return nil, err
			}
			reply.PacketID = uint32(packetID)
		case 10:
			if wireType != 2 {
				return nil, fmt.Errorf("invalid result wire type: %d", wireType)
			}
			result, err := consumeBytes(payload, &pos)
			if err != nil {
				return nil, err
			}
			reply.Result = result
		case 11:
			if wireType != 2 {
				return nil, fmt.Errorf("invalid error wire type: %d", wireType)
			}
			errorPayload, err := consumeBytes(payload, &pos)
			if err != nil {
				return nil, err
			}
			reply.Error = decodeServiceReplyError(errorPayload)
		default:
			if err := skipField(payload, &pos, wireType); err != nil {
				return nil, err
			}
		}
	}
	return reply, nil
}

func logServiceReply(reply *ServiceReply) {
	if reply == nil {
		logger.Logger.Printf("[WebSocket Received Reply] <nil>")
		return
	}
	errorJSON := "null"
	if reply.Error != nil {
		if data, err := json.Marshal(reply.Error); err == nil {
			errorJSON = string(data)
		} else {
			errorJSON = fmt.Sprintf(`{"marshal_error":%q}`, err.Error())
		}
	}
	logger.Logger.Printf("[WebSocket Received Reply] packet_id=%d result=%s error=%s", reply.PacketID, string(reply.Result), errorJSON)
}

func decodeServiceReplyError(payload []byte) *WSError {
	wsErr := &WSError{}
	for pos := 0; pos < len(payload); {
		key, err := consumeVarint(payload, &pos)
		if err != nil {
			return wsErr
		}
		fieldNumber := int(key >> 3)
		wireType := int(key & 0x7)
		switch fieldNumber {
		case 1:
			code, err := consumeVarint(payload, &pos)
			if err != nil {
				return wsErr
			}
			wsErr.Code = int(int32(code))
		case 2:
			msg, err := consumeBytes(payload, &pos)
			if err != nil {
				return wsErr
			}
			wsErr.Message = string(msg)
		default:
			_ = skipField(payload, &pos, wireType)
		}
	}
	return wsErr
}

func decodeServiceCastTopicEvent(payload []byte) (string, []byte, bool, error) {
	for pos := 0; pos < len(payload); {
		key, err := consumeVarint(payload, &pos)
		if err != nil {
			return "", nil, false, err
		}
		fieldNumber := int(key >> 3)
		wireType := int(key & 0x7)
		if fieldNumber != 10 {
			if err := skipField(payload, &pos, wireType); err != nil {
				return "", nil, false, err
			}
			continue
		}
		if wireType != 2 {
			return "", nil, false, fmt.Errorf("invalid topic_event wire type: %d", wireType)
		}
		topicPayload, err := consumeBytes(payload, &pos)
		if err != nil {
			return "", nil, false, err
		}
		topic, content, err := decodeTopicEvent(topicPayload)
		return topic, content, true, err
	}
	return "", nil, false, nil
}

func decodeTopicEvent(payload []byte) (string, []byte, error) {
	var topic string
	var content []byte
	for pos := 0; pos < len(payload); {
		key, err := consumeVarint(payload, &pos)
		if err != nil {
			return "", nil, err
		}
		fieldNumber := int(key >> 3)
		wireType := int(key & 0x7)
		switch fieldNumber {
		case 1:
			value, err := consumeBytes(payload, &pos)
			if err != nil {
				return "", nil, err
			}
			topic = string(value)
		case 2:
			value, err := consumeBytes(payload, &pos)
			if err != nil {
				return "", nil, err
			}
			content = value
		default:
			if err := skipField(payload, &pos, wireType); err != nil {
				return "", nil, err
			}
		}
	}
	return topic, content, nil
}

func (c *WSClient) SendAsync(message interface{}) {
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
		defer cancel()

		done := make(chan error, 1)
		go func() {
			done <- c.SendRawMessage(message)
		}()

		select {
		case err := <-done:
			if err != nil {
				logger.Logger.Printf("Asynchronous reporting failed: %v", err)
			} else {
				logger.Logger.Printf("Asynchronous reporting successful")
			}
		case <-ctx.Done():
			logger.Logger.Printf("Asynchronous reporting timeout (8 seconds no response), continuing business process")
		}
	}()
}