// Package eventbus is the in-process typed pub/sub primitive every reactive
// subsystem hangs off. Phase 3 ships the foundational shape; Phase 4 grows
// it with the autobrr/qBittorrent/cross-seed webhook topics and the SSE
// dispatcher.
//
// Delivery is best-effort and at-most-once. Slow subscribers get
// unsubscribed with a logged event rather than blocking publishers. This
// matches DECISIONS.md D18.
package eventbus

import (
	"sync"
	"sync/atomic"

	"github.com/rs/zerolog"
)

// Topic is the string identifier of a stream. Define new topics in topics.go
// as typed constants so callers cannot mistype names.
type Topic string

// Event is the generic envelope carrying a typed payload around the bus.
// The payload is an `any`; subscribers type-assert based on the topic they
// subscribed to. Misuse (subscribing to topic "torrents" then asserting an
// "autobrr.grab" payload type) is a programmer error.
type Event struct {
	Topic   Topic
	Payload any
}

// Subscription is the receive end of a topic stream. Read via Chan(); call
// Close to unsubscribe. Closing more than once is safe.
type Subscription struct {
	id     uint64
	topic  Topic
	ch     chan Event
	closed atomic.Bool
	parent *Bus
}

// Chan returns the receive channel.
func (s *Subscription) Chan() <-chan Event { return s.ch }

// Close removes the subscription. Drains the channel to unblock the publisher
// if it happened to be mid-send. Idempotent.
func (s *Subscription) Close() {
	if s.closed.Swap(true) {
		return
	}
	s.parent.unsubscribe(s.topic, s.id)
}

// Bus is a typed pub/sub fanout. One bus per process; the constructor is
// cheap so tests get their own.
type Bus struct {
	logger zerolog.Logger

	mu          sync.RWMutex
	subscribers map[Topic]map[uint64]*Subscription

	nextID atomic.Uint64

	// BufferSize is the per-subscription channel buffer. When a subscriber
	// can't keep up beyond this depth, Publish drops the event for that
	// subscriber and logs a structured `dropped` event. Default 64.
	BufferSize int
}

// New constructs a fresh bus.
func New(logger zerolog.Logger) *Bus {
	return &Bus{
		logger:      logger.With().Str("component", "eventbus").Logger(),
		subscribers: map[Topic]map[uint64]*Subscription{},
		BufferSize:  64,
	}
}

// Subscribe registers a new listener on topic. The returned Subscription is
// usable immediately. The caller MUST Close() it when done.
func (b *Bus) Subscribe(topic Topic) *Subscription {
	id := b.nextID.Add(1)
	sub := &Subscription{
		id:     id,
		topic:  topic,
		ch:     make(chan Event, b.bufferSize()),
		parent: b,
	}
	b.mu.Lock()
	if b.subscribers[topic] == nil {
		b.subscribers[topic] = map[uint64]*Subscription{}
	}
	b.subscribers[topic][id] = sub
	b.mu.Unlock()
	return sub
}

// Publish fans the event out to every subscription on its topic. Returns the
// number of subscribers that received the event (i.e., did NOT drop it).
func (b *Bus) Publish(ev Event) int {
	b.mu.RLock()
	subs := b.subscribers[ev.Topic]
	// Snapshot the slice so we don't hold the lock during sends.
	sl := make([]*Subscription, 0, len(subs))
	for _, s := range subs {
		sl = append(sl, s)
	}
	b.mu.RUnlock()

	delivered := 0
	for _, s := range sl {
		if s.closed.Load() {
			continue
		}
		select {
		case s.ch <- ev:
			delivered++
		default:
			// Slow subscriber. Drop and emit a structured warning.
			b.logger.Warn().
				Str("topic", string(ev.Topic)).
				Uint64("subscription_id", s.id).
				Msg("dropped event: subscriber buffer full")
		}
	}
	return delivered
}

// SubscriberCount returns the number of active subscriptions on topic.
// Useful for tests and for the system-events readout.
func (b *Bus) SubscriberCount(topic Topic) int {
	b.mu.RLock()
	defer b.mu.RUnlock()
	return len(b.subscribers[topic])
}

func (b *Bus) bufferSize() int {
	if b.BufferSize <= 0 {
		return 64
	}
	return b.BufferSize
}

func (b *Bus) unsubscribe(topic Topic, id uint64) {
	b.mu.Lock()
	if subs, ok := b.subscribers[topic]; ok {
		if _, ok := subs[id]; ok {
			delete(subs, id)
			// Intentionally do NOT close(s.ch) here. Closing would race
			// with the concurrent send path in Publish: even with the
			// closed.Load() guard, a close-then-send window remains. The
			// channel is GC'd once both the (now-detached) subscriber
			// and the bus drop their references; consumers reading from
			// Chan() simply stop receiving, which is what they want
			// after Close().
			if len(subs) == 0 {
				delete(b.subscribers, topic)
			}
		}
	}
	b.mu.Unlock()
}