From 0c7ef38dcef7a15d629ccb32b441d58c43901348 Mon Sep 17 00:00:00 2001
From: Maxence Maireaux <maxence@formance.com>
Date: Thu, 11 Jun 2026 10:15:54 +0200
Subject: [PATCH] fix(triggers): surface listener panics as errors instead of
 dropping events

handleMessage had an unnamed error return, so its recover() returned nil
and watermill ACKed the message. getWorkflowIDFromEvent deliberately
panicked on (un)marshal failures (e.g. a SAVED_PAYMENT payload whose id is
not a string), so a malformed event was permanently dropped with only a
stdout print.

- getWorkflowIDFromEvent now returns an error instead of panicking.
- handleMessage uses a named return so a recovered panic is converted into
  a returned error (NACK + redelivery/DLQ) and logged with its stack via
  the structured logger.

Adds a test asserting a malformed payment payload returns an error.

Note: overlaps internal/triggers/listener.go with #184 (idempotence); the
two are independent and may need a trivial merge.
---
 internal/triggers/listener.go      | 24 +++++++++++++++---------
 internal/triggers/listener_test.go | 20 ++++++++++++++++++++
 2 files changed, 35 insertions(+), 9 deletions(-)

diff --git a/internal/triggers/listener.go b/internal/triggers/listener.go
index de0cc17..23dc32f 100644
--- a/internal/triggers/listener.go
+++ b/internal/triggers/listener.go
@@ -28,12 +28,12 @@ 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 {
@@ -41,12 +41,12 @@ func getWorkflowIDFromEvent(event publish.EventMessage) *string {
 		}
 		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{}{
diff --git a/internal/triggers/listener_test.go b/internal/triggers/listener_test.go
index fb25dfd..c745f40 100644
--- a/internal/triggers/listener_test.go
+++ b/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)
+	})
 }