fix(triggers): surface listener panics as errors instead of dropping events (EN-1222)

internal/triggers/listener.go

@@ -28,25 +28,25 @@ import (
 )
 
 // Quick hack to filter already processed events
-func getWorkflowIDFromEvent(event publish.EventMessage) *string {
+func getWorkflowIDFromEvent(event publish.EventMessage) (*string, error) {
 	switch event.Type {
 	case "SAVED_PAYMENT", "SAVED_ACCOUNT":
 		data, err := json.Marshal(event.Payload)
 		if err != nil {
-			panic(err)
+			return nil, errors.Wrap(err, "marshalling event payload")
 		}
 
 		type object struct {
 			ID string `json:"id"`
 		}
 		o := &object{}
 		if err := json.Unmarshal(data, o); err != nil {
-			panic(err)
+			return nil, errors.Wrap(err, "unmarshalling event payload")
 		}
 
-		return pointer.For(o.ID)
+		return pointer.For(o.ID), nil
 	default:
-		return nil
+		return nil, nil
 	}
 }
 
@@ -96,11 +96,14 @@ func handleMessage(
 	stack, taskIDPrefix, taskQueue string,
 	includeSearchAttributes bool,
 	msg *message.Message,
-) error {
+) (err error) {
 	defer func() {
 		if e := recover(); e != nil {
-			fmt.Println(e)
-			debug.PrintStack()
+			// Convert a panic into a returned error so the message is NACKed
+			// (and redelivered / dead-lettered) instead of being silently
+			// acked and lost.
+			err = fmt.Errorf("panic while handling event: %v", e)
+			logging.FromContext(msg.Context()).Errorf("%s\n%s", err, debug.Stack())
 		}
 	}()
 
@@ -138,7 +141,10 @@ func handleMessage(
 		return nil
 	}
 
-	objectID := getWorkflowIDFromEvent(*event)
+	objectID, err := getWorkflowIDFromEvent(*event)
+	if err != nil {
+		return errors.Wrap(err, "extracting workflow id from event")
+	}
 
 	for _, trigger := range matched {
 		searchAttributes := map[string]interface{}{

internal/triggers/listener_test.go

@@ -379,4 +379,24 @@ func TestHandleMessage(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, 1, count)
 	})
+
+	t.Run("malformed payment payload returns an error instead of being dropped", func(t *testing.T) {
+		t.Parallel()
+
+		db := setupTestDB(t)
+		taskQueue := setupWorker(t, db)
+
+		w := insertNoOpWorkflow(t, db)
+		insertTrigger(t, db, w.ID, "SAVED_PAYMENT", nil, nil)
+
+		// "id" is a number, so extracting the dedup id fails. This used to
+		// panic and be silently acked; it must now surface as an error so the
+		// message is NACKed.
+		event := makeMessage("SAVED_PAYMENT", "v1", map[string]any{"id": 123})
+		msg := publish.NewMessage(logging.TestingContext(), *event)
+
+		evaluator := NewDefaultExpressionEvaluator()
+		err := handleMessage(devServer.Client(), db, evaluator, "test", "test", taskQueue, false, msg)
+		require.Error(t, err)
+	})
 }