Refactored the scheduler and source interfaces to accommondate both polling (e.g., HTTP) sources and streaming (e.g., message queue) sources.

sources/registry.go

@@ -13,13 +13,18 @@ import (
 // domain-specific source drivers (Open-Meteo, NWS, RSS, etc.) while feedkit
 // remains domain-agnostic.
 type Factory func(cfg config.SourceConfig) (Source, error)
+type StreamFactory func(cfg config.SourceConfig) (StreamSource, error)
 
 type Registry struct {
-	byDriver map[string]Factory
+	byDriver       map[string]Factory
+	byStreamDriver map[string]StreamFactory
 }
 
 func NewRegistry() *Registry {
-	return &Registry{byDriver: map[string]Factory{}}
+	return &Registry{
+		byDriver:       map[string]Factory{},
+		byStreamDriver: map[string]StreamFactory{},
+	}
 }
 
 // Register associates a driver name (e.g. "openmeteo_observation") with a factory.
@@ -35,10 +40,27 @@ func (r *Registry) Register(driver string, f Factory) {
 	if f == nil {
 		panic(fmt.Sprintf("sources.Registry.Register: factory cannot be nil (driver=%q)", driver))
 	}
-
+	if _, exists := r.byStreamDriver[driver]; exists {
+		panic(fmt.Sprintf("sources.Registry.Register: driver %q already registered as a stream source", driver))
+	}
 	r.byDriver[driver] = f
 }
 
+// RegisterStream is the StreamSource equivalent of Register.
+func (r *Registry) RegisterStream(driver string, f StreamFactory) {
+	driver = strings.TrimSpace(driver)
+	if driver == "" {
+		panic("sources.Registry.RegisterStream: driver cannot be empty")
+	}
+	if f == nil {
+		panic(fmt.Sprintf("sources.Registry.RegisterStream: factory cannot be nil (driver=%q)", driver))
+	}
+	if _, exists := r.byDriver[driver]; exists {
+		panic(fmt.Sprintf("sources.Registry.RegisterStream: driver %q already registered as a polling source", driver))
+	}
+	r.byStreamDriver[driver] = f
+}
+
 // Build constructs a Source from a SourceConfig by looking up cfg.Driver.
 func (r *Registry) Build(cfg config.SourceConfig) (Source, error) {
 	f, ok := r.byDriver[cfg.Driver]
@@ -47,3 +69,14 @@ func (r *Registry) Build(cfg config.SourceConfig) (Source, error) {
 	}
 	return f(cfg)
 }
+
+// BuildInput can return either a polling Source or a StreamSource.
+func (r *Registry) BuildInput(cfg config.SourceConfig) (Input, error) {
+	if f, ok := r.byStreamDriver[cfg.Driver]; ok {
+		return f(cfg)
+	}
+	if f, ok := r.byDriver[cfg.Driver]; ok {
+		return f(cfg)
+	}
+	return nil, fmt.Errorf("unknown source driver: %q", cfg.Driver)
+}

sources/source.go

@@ -6,6 +6,12 @@ import (
 	"gitea.maximumdirect.net/ejr/feedkit/event"
 )
 
+// Input is the common surface shared by all source types.
+type Input interface {
+	Name() string
+	Kind() event.Kind
+}
+
 // Source is a configured polling job that emits 0..N events per poll.
 //
 // Source implementations live in domain modules (weatherfeeder/newsfeeder/...)
@@ -28,3 +34,12 @@ type Source interface {
 	// Implementations should honor ctx.Done() for network calls and other I/O.
 	Poll(ctx context.Context) ([]event.Event, error)
 }
+
+// StreamSource is an event-driven source (NATS/RabbitMQ/MQTT/etc).
+//
+// Run should block, producing events into `out` until ctx is cancelled or a fatal error occurs.
+// It MUST NOT close out (the scheduler/daemon owns the bus).
+type StreamSource interface {
+	Input
+	Run(ctx context.Context, out chan<- event.Event) error
+}