// stream.go 实现通知 inbox 的实时推送（spec §11 步骤8）。
//
// StreamHub 是 per-user 的订阅扇出中心：每个浏览器标签页 Subscribe 一个有界缓冲
// channel；Dispatcher 落库成功后 Publish 到该用户的全部订阅者。慢订阅者（缓冲满）
// 直接丢弃该消息（drop-on-slow），绝不阻塞 Publish / Dispatch，避免 goroutine /
// 内存泄漏（镜像 chat StreamerHub 的丢弃纪律）。
package notify

import (
	"sync"

	"github.com/google/uuid"
)

// StreamHub 是 Dispatcher 推送实时通知所需的窄接口。
type StreamHub interface {
	Subscribe(userID uuid.UUID) (<-chan Message, func())
	Publish(userID uuid.UUID, m Message)
}

// streamSubBuffer 是单个订阅者的缓冲深度。满时新消息被丢弃（客户端可用 inbox
// 列表接口补偿，与 chat WS onopen 主动拉取同理）。
const streamSubBuffer = 16

// memHub 是 StreamHub 的进程内实现。
type memHub struct {
	mu   sync.Mutex
	subs map[uuid.UUID]map[int]chan Message
	next int
}

// NewStreamHub 构造进程内的 per-user 通知推送中心。
func NewStreamHub() StreamHub {
	return &memHub{subs: map[uuid.UUID]map[int]chan Message{}}
}

// Subscribe 为某用户注册一个订阅者，返回只读 channel 与取消函数。取消函数幂等：
// 关闭 channel 并从订阅表移除；调用方应在请求结束时 defer 调用。
func (h *memHub) Subscribe(userID uuid.UUID) (<-chan Message, func()) {
	ch := make(chan Message, streamSubBuffer)
	h.mu.Lock()
	if h.subs[userID] == nil {
		h.subs[userID] = map[int]chan Message{}
	}
	id := h.next
	h.next++
	h.subs[userID][id] = ch
	h.mu.Unlock()

	var once sync.Once
	cancel := func() {
		once.Do(func() {
			h.mu.Lock()
			if m, ok := h.subs[userID]; ok {
				if c, ok := m[id]; ok {
					delete(m, id)
					close(c)
				}
				if len(m) == 0 {
					delete(h.subs, userID)
				}
			}
			h.mu.Unlock()
		})
	}
	return ch, cancel
}

// Publish 向某用户的全部订阅者非阻塞投递一条消息。缓冲满的订阅者被跳过。
//
// 关键：整个投递循环持有 h.mu。因为发送是非阻塞的（select default），持锁开销极小，
// 但这让 Publish 与 cancel()（同样持 h.mu 后 close channel）互斥——否则快照 channel
// 释放锁后、cancel 并发 close，会 panic "send on closed channel"。
func (h *memHub) Publish(userID uuid.UUID, m Message) {
	h.mu.Lock()
	defer h.mu.Unlock()
	for _, c := range h.subs[userID] {
		select {
		case c <- m:
		default:
			// drop-on-slow：缓冲满，丢弃，避免阻塞 Dispatch。
		}
	}
}