// Package push is the gateway's live-event fan-out. The gateway holds one
// backend gRPC subscription that feeds Publish; each connected client opens a
// Subscribe stream and receives only the events addressed to its user id. A slow
// client never blocks the backend feed — its bounded queue drops on overflow.
package push

import "sync"

// Event is one live event addressed to a user. Payload is the FlatBuffers body
// the gateway forwards verbatim to the client.
type Event struct {
	UserID  string
	Kind    string
	Payload []byte
	EventID string
}

// defaultBuffer is the per-client queue depth used when NewHub is given a
// non-positive size.
const defaultBuffer = 64

// Hub fans backend events out to per-user client subscriptions.
type Hub struct {
	bufSize int

	mu     sync.Mutex
	nextID int
	subs   map[int]*subscription
}

type subscription struct {
	userID string
	ch     chan Event
}

// NewHub constructs a Hub whose per-client queue holds bufSize events.
func NewHub(bufSize int) *Hub {
	if bufSize <= 0 {
		bufSize = defaultBuffer
	}
	return &Hub{bufSize: bufSize, subs: make(map[int]*subscription)}
}

// Publish delivers e to every subscription for e.UserID, dropping it for any
// whose queue is full.
func (h *Hub) Publish(e Event) {
	h.mu.Lock()
	defer h.mu.Unlock()
	for _, s := range h.subs {
		if s.userID != e.UserID {
			continue
		}
		select {
		case s.ch <- e:
		default:
		}
	}
}

// Subscribe registers a client stream for userID and returns its event channel
// and an unsubscribe func that closes the channel.
func (h *Hub) Subscribe(userID string) (<-chan Event, func()) {
	h.mu.Lock()
	defer h.mu.Unlock()
	id := h.nextID
	h.nextID++
	s := &subscription{userID: userID, ch: make(chan Event, h.bufSize)}
	h.subs[id] = s
	return s.ch, func() { h.unsubscribe(id) }
}

// unsubscribe removes and closes a subscription. It holds the same lock as
// Publish, so it never closes a channel mid-send.
func (h *Hub) unsubscribe(id int) {
	h.mu.Lock()
	defer h.mu.Unlock()
	if s, ok := h.subs[id]; ok {
		delete(h.subs, id)
		close(s.ch)
	}
}

// SubscriberCount returns the number of active subscriptions (for tests/metrics).
func (h *Hub) SubscriberCount() int {
	h.mu.Lock()
	defer h.mu.Unlock()
	return len(h.subs)
}