package event

import (
	"errors"
	"fmt"
	"sync"
	"time"

	"gitlab.galaxy-immi.com/Backend-group/go-com/tool"
	"gitlab.galaxy-immi.com/Backend-group/go-com/tool/log"
)

// Broadcast 广播器接口
type Broadcast interface {
	publish()
	Subscribe(chan<- any)
	Send(any)
	Close()
}

// Broadcaster 广播器
type Broadcaster struct {
	subscribers []chan<- any
	broadcast   chan any
	mux         sync.Mutex
	eventName   string
	closeChan   chan struct{}
	closed      bool
}

// BroadcasterMap 广播器映射
var BroadcasterMap sync.Map

// NewBroadcaster 创建一个新的 Broadcaster
func NewBroadcaster(event string) *Broadcaster {
	if !tool.InSlice(event, Events) {
		log.ErrorP(fmt.Sprintf("invalid event: %s", event))
		return nil
	}

	if _, exist := BroadcasterMap.Load(event); exist {
		log.ErrorP(fmt.Sprintf("event %s already exists", event))
		return nil
	}

	b := &Broadcaster{
		subscribers: make([]chan<- any, 0),
		broadcast:   make(chan any),
		eventName:   event,
		closeChan:   make(chan struct{}),
		closed:      false,
	}
	BroadcasterMap.Store(event, b)
	b.publish() // 自动开始发布
	return b
}

// Close 停止广播并关闭所有订阅者通道
func (b *Broadcaster) Close() {
	if nilObjReturn(b) != nil {
		return
	}
	b.mux.Lock()
	defer b.mux.Unlock()
	if !b.closed {
		b.closed = true
		close(b.closeChan) // 向goroutine发出关闭信号
		for _, sub := range b.subscribers {
			close(sub) // 主动关闭所有订阅者通道
		}
		BroadcasterMap.Delete(b.eventName)                     // 从map中移除
		log.InfoP(fmt.Sprintf("closed %s event", b.eventName)) // nolint:gosimple
	}
}

// CloseAll 关闭所有事件广播和订阅者
func CloseAll() {
	BroadcasterMap.Range(func(key, value interface{}) bool {
		value.(*Broadcaster).Close()
		return true
	})

	log.InfoP("closed all events")
}

// GetBroadcaster 获取一个 Broadcaster
func GetBroadcaster(event string) *Broadcaster {
	if v, ok := BroadcasterMap.Load(event); ok {
		return v.(*Broadcaster)
	}
	return nil
}

// publish 发布订阅事件
func (b *Broadcaster) publish() {
	go func() {
		for {
			select {
			case msg, ok := <-b.broadcast:
				if !ok {
					return // 如果 broadcast 被关闭，结束 goroutine
				}
				// 发送消息到订阅者
				b.mux.Lock()
				for _, sub := range b.subscribers {
					sub <- msg
				}
				b.mux.Unlock()
			case <-b.closeChan:
				return // 收到关闭信号，结束 goroutine
			}
		}
	}()
	// 给一个 100ms 的延时，保证发布事件 goroutine 已经启动
	time.Sleep(100 * time.Millisecond)
	log.InfoP(fmt.Sprintf("publish event: %v", b.eventName)) // nolint:gosimple
}

// Subscribe 注册一个订阅者的通道
func (b *Broadcaster) Subscribe(sub chan<- any) {
	if err := nilObjReturn(b); err != nil {
		return
	}
	b.mux.Lock()
	defer b.mux.Unlock()
	if b.closed {
		errStr := fmt.Sprintf("attempt to subscribe to closed broadcaster: %v", b.eventName)
		log.InfoP(errStr)
		sub <- errors.New(errStr)
		close(sub)
		return
	}
	b.subscribers = append(b.subscribers, sub)
}

// Send 广播一个消息给所有订阅者
func (b *Broadcaster) Send(msg any) {
	if err := nilObjReturn(b); err != nil {
		return
	}
	b.mux.Lock()
	subCount := len(b.subscribers)
	closed := b.closed
	b.mux.Unlock()

	if closed {
		log.InfoP(fmt.Sprintf("broadcast to closed event: %v", b.eventName)) // nolint:gosimple
		return
	}

	if subCount == 0 {
		log.InfoP(fmt.Sprintf("no subscribers for event: %v", b.eventName)) // nolint:gosimple
		return
	}

	select {
	case b.broadcast <- msg:
		// 消息发送成功
	default:
		// 通道已满，记录警告
		log.InfoP(fmt.Sprintf("broadcast channel is full for event %v", b.eventName)) // nolint:gosimple
	}
}

// nilObjReturn 检查 Broadcaster 是否为 nil
func nilObjReturn(b *Broadcaster) (err error) {
	if b == nil {
		errStr := "called on nil broadcaster"
		log.InfoP(errStr)
		return errors.New(errStr)
	}
	return nil
}
