events.go

// SPDX-License-Identifier: AGPL-3.0-or-later

package runtime

import (
	"github.com/pilot-protocol/common/coreapi"
	"github.com/pilot-protocol/common/daemonapi"
)

// daemonEventBus adapts daemon's in-process bus to coreapi.EventBus
// for plugin Deps. Publish forwards through Daemon.PublishEvent;
// Subscribe wraps the bus channel with type conversion daemonapi.Event
// → coreapi.Event.

type daemonEventBus struct{ d daemonapi.Daemon }

func (b daemonEventBus) Publish(topic string, payload map[string]any) {
	if b.d == nil {
		return
	}
	b.d.PublishEvent(topic, payload)
}

func (b daemonEventBus) Subscribe(pattern string) (<-chan coreapi.Event, func()) {
	if b.d == nil || b.d.Bus() == nil {
		ch := make(chan coreapi.Event)
		close(ch)
		return ch, func() {}
	}
	src, cancel := b.d.Bus().Subscribe(pattern)
	out := make(chan coreapi.Event, cap(src))
	go func() {
		defer close(out)
		for ev := range src {
			// Non-blocking, drop-on-full. The underlying bus deliberately
			// drops rather than blocking its publishers; forwarding with a
			// blocking send silently converted those semantics and turned
			// this adapter into a goroutine leak.
			//
			// If a plugin consumer stopped draining `out` — its handler
			// loop exited, panicked past a recover, or simply ran slower
			// than the publisher for cap(src) events — this goroutine
			// parked on the send forever, retaining itself, the buffered
			// contents of src, and every payload map they referenced.
			// cancel() does not rescue it: cancel closes src, which does
			// nothing for a goroutine already blocked sending to out.
			//
			// Dropping matches what the bus would have done anyway once
			// the buffer filled, so a slow consumer now loses events
			// instead of leaking a goroutine for the process lifetime.
			select {
			case out <- coreapi.Event{
				Topic:   ev.Topic,
				NodeID:  ev.NodeID,
				Time:    ev.Time,
				Payload: ev.Payload,
			}:
			default:
			}
		}
	}()
	return out, cancel
}

var _ coreapi.EventBus = daemonEventBus{}