package main

import "log/slog"

// PushEvent is broadcast to all SSE subscribers when a webhook is received.
// It contains just enough information for converge clients to decide whether
// to pull.
type PushEvent struct {
	FullName string `json:"full_name"` // e.g. "myorg/myrepo"
	Branch   string `json:"branch"`    // e.g. "main"
	Ref      string `json:"ref"`       // e.g. "refs/heads/main"
}

// Broker implements a pub/sub fan-out for PushEvents.
// Subscribe to get a channel that receives events; unsubscribe when done.
// Slow subscribers (channel full) are silently dropped for that event.
type Broker struct {
	subscribe   chan chan PushEvent
	unsubscribe chan chan PushEvent
	publish     chan PushEvent
	stop        chan struct{}
	subs        map[chan PushEvent]struct{}
}

// NewBroker creates a ready-to-run Broker.
func NewBroker() *Broker {
	return &Broker{
		subscribe:   make(chan chan PushEvent),
		unsubscribe: make(chan chan PushEvent),
		publish:     make(chan PushEvent, 64),
		stop:        make(chan struct{}),
		subs:        make(map[chan PushEvent]struct{}),
	}
}

// Run starts the broker event loop. Must be called in its own goroutine.
func (b *Broker) Run() {
	for {
		select {
		case ch := <-b.subscribe:
			b.subs[ch] = struct{}{}
			slog.Debug("subscriber connected", "total", len(b.subs))

		case ch := <-b.unsubscribe:
			delete(b.subs, ch)
			// Drain and close the channel so readers exit cleanly
			close(ch)
			for range ch {
			}
			slog.Debug("subscriber disconnected", "total", len(b.subs))

		case evt := <-b.publish:
			slog.Info("publishing event", "full_name", evt.FullName, "branch", evt.Branch)
			for ch := range b.subs {
				select {
				case ch <- evt:
				default:
					// Subscriber too slow; drop the event for this subscriber
					slog.Warn("dropping event for slow subscriber")
				}
			}

		case <-b.stop:
			for ch := range b.subs {
				close(ch)
			}
			return
		}
	}
}

// Subscribe returns a buffered channel that receives PushEvents.
func (b *Broker) Subscribe() chan PushEvent {
	// Small buffer to smooth out delivery; slow consumers are dropped by
	// the broker's non-blocking publish anyway.
	ch := make(chan PushEvent, 16)
	b.subscribe <- ch
	return ch
}

// Unsubscribe removes a subscriber channel.
func (b *Broker) Unsubscribe(ch chan PushEvent) {
	b.unsubscribe <- ch
}

// Publish sends an event to all current subscribers. Never blocks.
func (b *Broker) Publish(evt PushEvent) {
	b.publish <- evt
}

// Stop shuts down the broker.
func (b *Broker) Stop() {
	close(b.stop)
}