fix(framework/event): add panic recovery and HandlePublish to MQTT broker

A panicking handler in route() propagated through paho's internal
goroutine and crashed the entire MQTT connection. Add defer/recover
via deliverToHandler.

NewMQTTBrokerFrom never wired OnPublishReceived, making handlers
unreachable. Add public HandlePublish method and SetClient for
two-step wiring pattern.

Adds 4 MQTT-specific tests for panic recovery and HandlePublish.

Author: ConsoleTVs

framework/event/mqtt.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"log/slog"
 	"net/url"
 	"strconv"
 	"strings"
@@ -196,11 +197,7 @@ func NewMQTTBrokerWith(options *MQTTBrokerOptions) (*MQTTBroker, error) {
 		ClientConfig: paho.ClientConfig{
 			ClientID: "",
 			OnPublishReceived: []func(paho.PublishReceived) (bool, error){
-				func(pr paho.PublishReceived) (bool, error) {
-					broker.route(pr.Packet)
-
-					return true, nil
-				},
+				broker.HandlePublish,
 			},
 		},
 	}
@@ -225,6 +222,19 @@ func NewMQTTBrokerWith(options *MQTTBrokerOptions) (*MQTTBroker, error) {
 // autopaho ConnectionManager and QoS level. This constructor is
 // useful for advanced scenarios where the user needs full control
 // over the MQTT connection configuration.
+//
+// Because autopaho.ConnectionManager does not allow post-creation
+// configuration changes, the caller must wire up message routing
+// when building the paho config. Use [MQTTBroker.HandlePublish] in
+// the ClientConfig.OnPublishReceived slice:
+//
+//	broker := event.NewMQTTBrokerFrom(nil, 1)
+//	cfg.ClientConfig.OnPublishReceived = []func(paho.PublishReceived) (bool, error){
+//	    broker.HandlePublish,
+//	}
+//	cm, err := autopaho.NewConnection(ctx, cfg)
+//	// then set the client so Publish/Subscribe/Close work:
+//	broker.SetClient(cm)
 func NewMQTTBrokerFrom(
 	client *autopaho.ConnectionManager,
 	qos byte,
@@ -237,25 +247,69 @@ func NewMQTTBrokerFrom(
 	}
 }
 
+// SetClient sets the underlying autopaho ConnectionManager. This is
+// intended for use with [NewMQTTBrokerFrom] when the broker must be
+// created before the ConnectionManager so that [MQTTBroker.HandlePublish]
+// can be wired into the paho config.
+func (broker *MQTTBroker) SetClient(client *autopaho.ConnectionManager) {
+	broker.client = client
+}
+
+// HandlePublish is a paho OnPublishReceived callback that routes
+// incoming MQTT messages to registered handlers. Callers using
+// [NewMQTTBrokerFrom] must include this method in the paho
+// ClientConfig.OnPublishReceived slice so that subscribed handlers
+// receive messages.
+func (broker *MQTTBroker) HandlePublish(pr paho.PublishReceived) (bool, error) {
+	broker.route(pr.Packet)
+
+	return true, nil
+}
+
 // route delivers an incoming MQTT message to all matching
 // handlers based on topic pattern matching. This implements
 // fan-out behavior where multiple handlers can receive the
 // same message if they subscribed to matching patterns.
+//
+// Each handler is called with panic recovery so that a
+// panicking handler does not crash the paho client goroutine
+// and tear down the entire MQTT connection.
 func (broker *MQTTBroker) route(pb *paho.Publish) {
 	broker.mu.RLock()
 	defer broker.mu.RUnlock()
 
 	for pattern, handlers := range broker.handlers {
 		if matchTopic(pattern, pb.Topic) {
 			for _, handler := range handlers {
-				handler(func(dest any) error {
-					return json.Unmarshal(pb.Payload, dest)
-				})
+				broker.deliverToHandler(handler, pb.Topic, pb.Payload)
 			}
 		}
 	}
 }
 
+// deliverToHandler invokes a single handler with panic recovery.
+// Recovered panics are logged via slog so they remain visible for
+// debugging without propagating to the caller.
+func (broker *MQTTBroker) deliverToHandler(
+	handler contract.EventHandler,
+	topic string,
+	payload []byte,
+) {
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			slog.Error(
+				"mqtt event handler panicked",
+				"topic", topic,
+				"error", recovered,
+			)
+		}
+	}()
+
+	handler(func(dest any) error {
+		return json.Unmarshal(payload, dest)
+	})
+}
+
 // Publish sends an event with the given name and payload to all
 // subscribers listening for that event. The payload is serialized
 // to JSON and the event name is converted to MQTT topic format.

framework/event/mqtt_internal_test.go

@@ -296,3 +296,103 @@ func TestMQTTBrokerRouteUnmarshalsJSONPayload(t *testing.T) {
 
 	require.Equal(t, "hello world", received)
 }
+
+func TestMQTTBrokerRoutePanicDoesNotPropagate(t *testing.T) {
+	t.Parallel()
+
+	broker := &MQTTBroker{
+		handlers: map[string]map[string]contract.EventHandler{
+			"user/created": {
+				"1": func(payload contract.EventPayload) {
+					panic("handler panic")
+				},
+			},
+		},
+	}
+
+	require.NotPanics(t, func() {
+		broker.route(&paho.Publish{
+			Topic:   "user/created",
+			Payload: []byte(`"data"`),
+		})
+	})
+}
+
+func TestMQTTBrokerRoutePanicDoesNotAffectOtherHandlers(t *testing.T) {
+	t.Parallel()
+
+	var called atomic.Bool
+
+	broker := &MQTTBroker{
+		handlers: map[string]map[string]contract.EventHandler{
+			"user/created": {
+				"1": func(payload contract.EventPayload) {
+					panic("handler panic")
+				},
+				"2": func(payload contract.EventPayload) {
+					called.Store(true)
+				},
+			},
+		},
+	}
+
+	broker.route(&paho.Publish{
+		Topic:   "user/created",
+		Payload: []byte(`"data"`),
+	})
+
+	require.True(t, called.Load())
+}
+
+func TestMQTTBrokerHandlePublishRoutesMessage(t *testing.T) {
+	t.Parallel()
+
+	var called atomic.Bool
+
+	broker := &MQTTBroker{
+		handlers: map[string]map[string]contract.EventHandler{
+			"user/created": {
+				"1": func(payload contract.EventPayload) {
+					called.Store(true)
+				},
+			},
+		},
+	}
+
+	handled, err := broker.HandlePublish(paho.PublishReceived{
+		Packet: &paho.Publish{
+			Topic:   "user/created",
+			Payload: []byte(`"hello"`),
+		},
+	})
+
+	require.NoError(t, err)
+	require.True(t, handled)
+	require.True(t, called.Load())
+}
+
+func TestMQTTBrokerHandlePublishRecoversPanic(t *testing.T) {
+	t.Parallel()
+
+	broker := &MQTTBroker{
+		handlers: map[string]map[string]contract.EventHandler{
+			"user/created": {
+				"1": func(payload contract.EventPayload) {
+					panic("handler panic")
+				},
+			},
+		},
+	}
+
+	require.NotPanics(t, func() {
+		handled, err := broker.HandlePublish(paho.PublishReceived{
+			Packet: &paho.Publish{
+				Topic:   "user/created",
+				Payload: []byte(`"data"`),
+			},
+		})
+
+		require.NoError(t, err)
+		require.True(t, handled)
+	})
+}