package runtimeevents

import (
	"context"
	"encoding/json"
	"sync"
	"sync/atomic"
)

// DropPolicy controls what happens when a subscriber's buffer is full.
type DropPolicy int

const (
	// DropNewest silently discards the incoming event when the buffer is full.
	DropNewest DropPolicy = iota
	// DropOldest removes the oldest event to make room for the new one.
	DropOldest
)

// Filter is a predicate that decides whether an event should be delivered to
// a subscriber. A nil filter matches all events.
type Filter func(Event) bool

// Subscription represents an active subscription to the Bus.
type Subscription interface {
	// Name returns the subscriber name (for logging and diagnostics).
	Name() string
	// Events returns a read-only channel of matched events.
	Events() <-chan Event
	// Unsubscribe removes this subscription from the Bus so future Publish calls
	// stop delivering to it. Callers should stop their receive agentteam via context
	// cancellation; the events channel is intentionally not closed here.
	Unsubscribe()
	// Dropped returns the cumulative number of events dropped for this
	// subscriber due to a full buffer.
	Dropped() uint64
}

// Bus is the daemon-internal event bus. Publish never blocks the caller for
// significant time; slow subscribers have events dropped according to their
// configured DropPolicy.
type Bus interface {
	// Publish sends an event to all matching subscribers.
	Publish(ctx context.Context, ev Event)
	// Subscribe registers a named subscriber with optional filter, buffer
	// size, and drop policy.
	Subscribe(name string, filter Filter, buffer int, policy DropPolicy) Subscription
	// Recent returns up to limit recent events matching filter from the
	// internal ring buffer. If filter is nil, all recent events are returned.
	Recent(filter Filter, limit int) []Event
	// Close shuts down the Bus, closing all subscriber channels.
	Close()
}

// bus is the default implementation of Bus.
type bus struct {
	mu            sync.RWMutex
	subs          map[string]*subscription
	ring          []Event
	ringBytes     []int
	ringSize      int
	ringByteLimit int
	ringBytesUsed int
	ringHead      int
	ringCount     int
	ringMu        sync.Mutex
	closed        atomic.Bool
}

// Option configures Bus behaviour.
type Option func(*bus)

// WithRingSize sets the in-memory ring buffer size for Recent(). Default 4096.
func WithRingSize(n int) Option {
	return func(b *bus) { b.ringSize = n }
}

// WithRingByteLimit sets an approximate byte budget for the Recent() ring.
func WithRingByteLimit(n int) Option {
	return func(b *bus) { b.ringByteLimit = n }
}

// NewBus creates a new Bus. The default ring size is 4096 and the ring byte
// budget is 8 MB.
func NewBus(opts ...Option) Bus {
	b := &bus{
		subs:          make(map[string]*subscription),
		ringSize:      4096,
		ringByteLimit: 8 << 20,
	}
	for _, opt := range opts {
		opt(b)
	}
	if b.ringSize <= 0 {
		b.ringSize = 1
	}
	if b.ringByteLimit <= 0 {
		b.ringByteLimit = 1
	}
	b.ring = make([]Event, b.ringSize)
	b.ringBytes = make([]int, b.ringSize)
	return b
}

// Publish sends ev to all matching subscribers and appends a scrubbed copy to
// the Recent() ring. Subscribers receive the original event so internal metrics
// sinks can still consume fields such as raw API keys; retained diagnostic
// history is scrubbed and byte-bounded.
func (b *bus) Publish(_ context.Context, ev Event) {
	if b.closed.Load() {
		return
	}

	ringEvent := ScrubEvent(ev)

	// Append to ring.
	b.ringMu.Lock()
	b.ringBytesUsed -= b.ringBytes[b.ringHead]
	b.ring[b.ringHead] = ringEvent
	b.ringBytes[b.ringHead] = estimateEventBytes(ringEvent)
	b.ringBytesUsed += b.ringBytes[b.ringHead]
	b.ringHead = (b.ringHead + 1) % b.ringSize
	if b.ringCount < b.ringSize {
		b.ringCount++
	}
	b.pruneRingBytesLocked()
	b.ringMu.Unlock()

	// Fan-out to subscribers while holding the read lock. Delivery is
	// non-blocking, and this prevents Close from closing subscriber channels
	// while a publish is sending to them.
	b.mu.RLock()
	for _, s := range b.subs {
		if s.filter != nil && !s.filter(ev) {
			continue
		}
		s.deliver(ev)
	}
	b.mu.RUnlock()
}

func (b *bus) pruneRingBytesLocked() {
	for b.ringCount > 0 && b.ringBytesUsed > b.ringByteLimit {
		oldest := (b.ringHead - b.ringCount + b.ringSize) % b.ringSize
		b.ringBytesUsed -= b.ringBytes[oldest]
		b.ring[oldest] = Event{}
		b.ringBytes[oldest] = 0
		b.ringCount--
	}
	if b.ringBytesUsed < 0 {
		b.ringBytesUsed = 0
	}
}

func estimateEventBytes(ev Event) int {
	data, err := json.Marshal(ev)
	if err != nil {
		return 512
	}
	return len(data)
}

// Subscribe registers or replaces a named subscriber.
func (b *bus) Subscribe(name string, filter Filter, buffer int, policy DropPolicy) Subscription {
	if buffer <= 0 {
		buffer = 256
	}
	s := &subscription{
		name:   name,
		filter: filter,
		ch:     make(chan Event, buffer),
		policy: policy,
		bus:    b,
	}
	b.mu.Lock()
	if b.closed.Load() {
		close(s.ch)
		b.mu.Unlock()
		return s
	}
	b.subs[name] = s
	b.mu.Unlock()
	return s
}

// Recent returns recent events from the ring buffer matching the filter.
func (b *bus) Recent(filter Filter, limit int) []Event {
	b.ringMu.Lock()
	defer b.ringMu.Unlock()

	if b.ringCount == 0 {
		return nil
	}

	result := make([]Event, 0, limit)
	// Read from oldest to newest.
	start := (b.ringHead - b.ringCount + b.ringSize) % b.ringSize
	for i := 0; i < b.ringCount && len(result) < limit; i++ {
		idx := (start + i) % b.ringSize
		ev := b.ring[idx]
		if filter == nil || filter(ev) {
			result = append(result, ev)
		}
	}
	return result
}

// Close shuts down the bus and closes all subscriber channels.
func (b *bus) Close() {
	b.closed.Store(true)
	b.mu.Lock()
	defer b.mu.Unlock()
	for name, s := range b.subs {
		close(s.ch)
		delete(b.subs, name)
	}
}

func (b *bus) removeSubscription(name string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	delete(b.subs, name)
}

// subscription implements Subscription.
type subscription struct {
	name    string
	filter  Filter
	ch      chan Event
	policy  DropPolicy
	dropped atomic.Uint64
	bus     *bus
}

// Name returns the subscriber name registered with Bus.Subscribe.
func (s *subscription) Name() string { return s.name }

// Events returns the matched event channel for this subscription.
func (s *subscription) Events() <-chan Event { return s.ch }

// Dropped returns the cumulative number of dropped events for this subscription.
func (s *subscription) Dropped() uint64 { return s.dropped.Load() }

// Unsubscribe unregisters the subscriber. It does not close Events; callers
// should use their own context cancellation to stop receive loops.
func (s *subscription) Unsubscribe() {
	s.bus.removeSubscription(s.name)
}

// deliver sends the event to the subscriber, respecting the DropPolicy.
func (s *subscription) deliver(ev Event) {
	switch s.policy {
	case DropOldest:
		select {
		case s.ch <- ev:
		default:
			// Buffer full — drain oldest then send.
			select {
			case <-s.ch:
			default:
			}
			s.dropped.Add(1)
			select {
			case s.ch <- ev:
			default:
			}
		}
	default: // DropNewest
		select {
		case s.ch <- ev:
		default:
			s.dropped.Add(1)
		}
	}
}