package comm

import (
	"encoding/json"
	"hkdj-web-server/pub/common/godef"
	"hkdj-web-server/pub/common/lock"
	"hkdj-web-server/pub/common/log"
	"sync"
	"time"

	"github.com/go-redis/redis"
	"go.uber.org/zap"
)

type MessageType int64

const (
	MessageTypeError MessageType = iota // 心跳消息
	MessageTypeAlive
	MessageTypeMediaAdd // 添加一个媒体
	MessageTypeMediaDel // 删除一个媒体
	MessageTypePostAdd  // 添加一个帖子
	MessageTypePostDel  // 删除一个帖子
)

type message struct {
	Type    MessageType
	Payload []byte
}

type OnReceive func(data []byte)

type Communicate struct {
	sendChannel string        // 发送消息的频道
	recvChannel string        // 接收消息的频道
	client      *redis.Client // Redis的客户端
	lock        sync.RWMutex
	callbacks   map[MessageType][]OnReceive
	publish     *lock.Queue[*message]
}

func NewCommunicate(client *redis.Client, sendChannel string, recvChannel string) *Communicate {
	return &Communicate{
		sendChannel: sendChannel,
		recvChannel: recvChannel,
		client:      client,
		callbacks:   make(map[MessageType][]OnReceive),
		publish:     lock.NewQueue[*message](),
	}
}

func (c *Communicate) RegisterReceiver(mType MessageType, receive OnReceive) {
	c.lock.Lock()
	defer c.lock.Unlock()
	c.callbacks[mType] = append(c.callbacks[mType], receive)
}

func (c *Communicate) findEvent(mType MessageType) []OnReceive {
	c.lock.RLock()
	defer c.lock.RUnlock()
	return c.callbacks[mType]
}

func (c *Communicate) dispatchMessage(mType MessageType, payload []byte) {
	events := c.findEvent(mType)
	for _, event := range events {
		event(payload)
	}
}

func (c *Communicate) handleRecv(data *redis.Message) {
	if data.Channel == "error" {
		log.Error("recv from channel %s failed", zap.Any("error", data.Payload))
		return
	}
	if data.Channel == c.recvChannel {
		msg := &message{}
		err := json.Unmarshal([]byte(data.Payload), msg)
		if err != nil {
			log.Error("recv from channel %s failed", zap.Any("error", err))
		}
		c.dispatchMessage(msg.Type, msg.Payload)
	}
}

func (c *Communicate) Send(mType MessageType, payload []byte) error {
	msg := &message{
		Type:    mType,
		Payload: payload,
	}
	c.publish.Enqueue(msg)
	return nil
}

func (c *Communicate) Start() {
	// 启动接收线程
	godef.Go(func() {
		sub := c.client.Subscribe(c.recvChannel)
		defer sub.Close()
		// 循环接收消息
		for {
			msg, err := sub.ReceiveMessage()
			if err != nil {
				msg = &redis.Message{
					Channel: "error",
					Payload: err.Error(),
				}
				c.handleRecv(msg)
				time.Sleep(time.Second)
			} else {
				c.handleRecv(msg)
			}
		}
	})
	// 启动发送线程
	godef.Go(func() {
		for {
			time.Sleep(time.Second)
			for !c.publish.IsEmpty() {
				msg, ok := c.publish.Dequeue()
				if !ok {
					break
				}
				data, err := json.Marshal(msg)
				if err != nil {
					log.Error("Communicate msg marshal failed", zap.Any("error", err))
					break
				}
				err = c.client.Publish(c.sendChannel, data).Err()
				if err != nil {
					c.publish.Enqueue(msg) // 再写回队列
					log.Error("Communicate send msg failed", zap.Any("error", err))
					break
				}
			}
		}
	})
}