// Package eventbus is a teaching event bus that makes the core
// event-driven patterns concrete: topic-based publish/subscribe with
// fan-out, at-least-once delivery with bounded retries, idempotent
// consumers (dedup by event id), and a dead-letter queue for poison
// messages. Stdlib-only and deterministic (synchronous delivery).
package eventbus

import (
	"sync"
	"sync/atomic"
)

// Event is a published message. ID is the dedup key that makes
// at-least-once delivery safe for idempotent consumers.
type Event struct {
	ID      string
	Topic   string
	Payload []byte
}

// Handler processes an event. Returning an error triggers a retry (and,
// after MaxRetries, a dead-letter).
type Handler func(Event) error

type subscription struct {
	name       string
	topic      string
	handler    Handler
	maxRetries int

	mu   sync.Mutex
	seen map[string]bool // idempotent consumer: event ids already processed
}

// DeadLetter records an event that exhausted its retries on a subscriber.
type DeadLetter struct {
	Event      Event
	Subscriber string
	Err        string
	Attempts   int
}

// Bus is a synchronous, topic-based pub/sub with retries + DLQ.
type Bus struct {
	mu  sync.RWMutex
	subs map[string][]*subscription
	dlq  []DeadLetter

	Delivered    atomic.Int64
	Retried      atomic.Int64
	DeadLettered atomic.Int64
	Deduped      atomic.Int64
}

func NewBus() *Bus { return &Bus{subs: map[string][]*subscription{}} }

// Subscribe registers a handler for a topic. maxRetries is the number of
// RETRIES after the first attempt (so total attempts = maxRetries+1).
func (b *Bus) Subscribe(name, topic string, maxRetries int, h Handler) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.subs[topic] = append(b.subs[topic], &subscription{
		name: name, topic: topic, handler: h, maxRetries: maxRetries,
		seen: map[string]bool{},
	})
}

// Publish fans the event out to every subscriber on its topic. Each
// subscriber gets at-least-once delivery with retries; a duplicate id is
// deduped (idempotent consumer); a handler that keeps failing is
// dead-lettered.
func (b *Bus) Publish(e Event) {
	b.mu.RLock()
	subs := append([]*subscription(nil), b.subs[e.Topic]...)
	b.mu.RUnlock()
	for _, sub := range subs {
		b.deliver(sub, e)
	}
}

func (b *Bus) deliver(sub *subscription, e Event) {
	sub.mu.Lock()
	if sub.seen[e.ID] {
		sub.mu.Unlock()
		b.Deduped.Add(1) // at-least-once duplicate; idempotent consumer skips it
		return
	}
	sub.mu.Unlock()

	var lastErr error
	for attempt := 0; attempt <= sub.maxRetries; attempt++ {
		err := sub.handler(e)
		if err == nil {
			sub.mu.Lock()
			sub.seen[e.ID] = true
			sub.mu.Unlock()
			b.Delivered.Add(1)
			return
		}
		lastErr = err
		if attempt < sub.maxRetries {
			b.Retried.Add(1)
		}
	}
	b.DeadLettered.Add(1)
	b.mu.Lock()
	b.dlq = append(b.dlq, DeadLetter{
		Event: e, Subscriber: sub.name, Err: lastErr.Error(), Attempts: sub.maxRetries + 1,
	})
	b.mu.Unlock()
}

// DLQ returns a snapshot of dead-lettered events (operators inspect and
// replay these after fixing the consumer).
func (b *Bus) DLQ() []DeadLetter {
	b.mu.RLock()
	defer b.mu.RUnlock()
	return append([]DeadLetter(nil), b.dlq...)
}