// Package events is the Controller's in-memory pub-sub for live agent
// events (instance state, app state, player, log). The Panel service's
// StreamEvents RPC subscribes; agent message handlers publish. Subscribers
// have bounded buffers and are dropped rather than back-pressuring the
// whole pipeline if they can't keep up.
package events

import (
	"sync"
	"sync/atomic"

	panelv1 "github.com/dbledeez/panel/proto/panel/v1"
)

// Filter narrows which events a subscriber receives. Empty string on a
// field means "no filter on that field".
type Filter struct {
	InstanceID string
	AgentID    string
	// ExcludeLogs drops Event_Log payloads for this subscriber. The global
	// dashboard stream sets this: log lines are by far the bulk of event
	// volume (a 7DTD boot is thousands of lines in seconds), the dashboard
	// has no DOM target for them, and shipping them anyway is what
	// overflowed the subscriber buffer (dropped events → stale status) and
	// kept the browser main thread busy. Consoles + readiness scanners use
	// per-instance subscriptions, which keep their logs.
	ExcludeLogs bool
}

// Subscription is the subscriber's handle. Call Close when done.
type Subscription struct {
	bus     *Bus
	id      int64
	filter  Filter
	ch      chan *panelv1.Event
	// sch is non-nil only for SubscribeStamped subscribers; exactly one of
	// ch/sch carries events for a given subscription.
	sch     chan Stamped
	dropped atomic.Uint64
	closed  atomic.Bool
}

// Events returns the receive channel. Consumer should range over it.
func (s *Subscription) Events() <-chan *panelv1.Event { return s.ch }

// Stamped returns the id-stamped receive channel. Only valid on
// subscriptions created via SubscribeStamped (nil otherwise).
func (s *Subscription) Stamped() <-chan Stamped { return s.sch }

// Dropped returns the number of events that were dropped because the
// subscriber's buffer was full.
func (s *Subscription) Dropped() uint64 { return s.dropped.Load() }

// Close unsubscribes and closes the channel. Safe to call multiple times.
func (s *Subscription) Close() {
	if !s.closed.CompareAndSwap(false, true) {
		return
	}
	s.bus.unsubscribe(s.id)
	if s.sch != nil {
		close(s.sch)
	} else {
		close(s.ch)
	}
}

// Bus is a concurrency-safe event fan-out with per-subscriber bounded
// buffers. Subscribers register with optional filters; publishers deliver
// each matching event non-blocking (drop on full buffer).
type Bus struct {
	mu     sync.RWMutex
	next   int64
	subs   map[int64]*Subscription
	bufCap int
	// ring retains the most recent events (with monotonic ids) so SSE
	// clients can resume via Last-Event-ID. See ring.go.
	ring *replayRing
	// droppedTotal counts every event dropped on a full subscriber buffer,
	// across all subscribers (including since-closed ones). Metrics surface.
	droppedTotal atomic.Uint64
}

// New creates a Bus. bufCap is the per-subscriber channel buffer.
//
// The default is intentionally generous: a busy install/download (SteamCMD
// preallocation, large Minecraft world seed, etc.) can emit 40+ log lines
// per second per instance. At the old default of 256, a browser subscriber
// that paused for half a second could lose dozens of intermediate events —
// the user saw "0 → 88% → 7.11%" jumps because the progress-bar frames
// between them hit the default: branch in Publish.
func New(bufCap int) *Bus {
	if bufCap <= 0 {
		bufCap = 4096
	}
	return &Bus{subs: map[int64]*Subscription{}, bufCap: bufCap, ring: newReplayRing(DefaultRingCap)}
}

// Subscribe returns a new Subscription. Call Sub.Close when done — typically
// via defer from the stream handler.
func (b *Bus) Subscribe(filter Filter) *Subscription {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.next++
	s := &Subscription{
		bus:    b,
		id:     b.next,
		filter: filter,
		ch:     make(chan *panelv1.Event, b.bufCap),
	}
	b.subs[s.id] = s
	return s
}

// SubscribeStamped is Subscribe, but the subscription's channel carries the
// event's monotonic replay id alongside the event (Stamped()). Used by the
// SSE handler to emit `id:` lines for Last-Event-ID resume.
func (b *Bus) SubscribeStamped(filter Filter) *Subscription {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.next++
	s := &Subscription{
		bus:    b,
		id:     b.next,
		filter: filter,
		sch:    make(chan Stamped, b.bufCap),
	}
	b.subs[s.id] = s
	return s
}

// Replay returns the retained events with ID > afterID that match the
// filter, oldest first. ok=false signals the resume point was overwritten
// (a gap precedes the returned events) — callers should resync via snapshot.
func (b *Bus) Replay(afterID uint64, f Filter) ([]Stamped, bool) {
	all, ok := b.ring.since(afterID)
	out := all[:0]
	for _, s := range all {
		if match(f, s.Event) {
			out = append(out, s)
		}
	}
	return out, ok
}

// LastID returns the id of the most recently published event (0 if none).
func (b *Bus) LastID() uint64 { return b.ring.lastID() }

// DroppedTotal returns the process-lifetime count of events dropped on
// full subscriber buffers.
func (b *Bus) DroppedTotal() uint64 { return b.droppedTotal.Load() }

func (b *Bus) unsubscribe(id int64) {
	b.mu.Lock()
	delete(b.subs, id)
	b.mu.Unlock()
}

// Publish fans the event to every matching subscriber. Non-blocking —
// slow subscribers lose events rather than back-pressuring publishers.
func (b *Bus) Publish(e *panelv1.Event) {
	// Stamp + retain first so replay ids are assigned in publish order.
	id := b.ring.append(e)
	b.mu.RLock()
	subs := make([]*Subscription, 0, len(b.subs))
	for _, s := range b.subs {
		if match(s.filter, e) {
			subs = append(subs, s)
		}
	}
	b.mu.RUnlock()
	for _, s := range subs {
		if s.sch != nil {
			select {
			case s.sch <- Stamped{ID: id, Event: e}:
			default:
				s.dropped.Add(1)
				b.droppedTotal.Add(1)
			}
			continue
		}
		select {
		case s.ch <- e:
		default:
			s.dropped.Add(1)
			b.droppedTotal.Add(1)
		}
	}
}

// SubscriberCount is useful for metrics / debug.
func (b *Bus) SubscriberCount() int {
	b.mu.RLock()
	defer b.mu.RUnlock()
	return len(b.subs)
}

func match(f Filter, e *panelv1.Event) bool {
	if f.AgentID != "" && f.AgentID != e.AgentId {
		return false
	}
	if f.InstanceID != "" {
		if eventInstanceID(e) != f.InstanceID {
			return false
		}
	}
	if f.ExcludeLogs {
		if _, isLog := e.Payload.(*panelv1.Event_Log); isLog {
			return false
		}
	}
	return true
}

func eventInstanceID(e *panelv1.Event) string {
	switch p := e.Payload.(type) {
	case *panelv1.Event_InstanceState:
		return p.InstanceState.InstanceId
	case *panelv1.Event_AppState:
		return p.AppState.InstanceId
	case *panelv1.Event_Player:
		return p.Player.InstanceId
	case *panelv1.Event_Log:
		return p.Log.InstanceId
	case *panelv1.Event_InstanceStats:
		return p.InstanceStats.InstanceId
	}
	return ""
}