fix(triggers): make event processing idempotent under redelivery and retries (EN-1221)

internal/triggers/activities.go

@@ -81,8 +81,13 @@ func (a Activities) EvalTriggerVariables(ctx context.Context, trigger Trigger, r
 }
 
 func (a Activities) InsertTriggerOccurrence(ctx context.Context, occurrence Occurrence) error {
+	// Idempotent: a Temporal retry after a lost ack (row committed but the
+	// activity result never reached the server) must not fail on the duplicate
+	// primary key. The occurrence id is deterministic (see
+	// NewTriggerOccurrence), so DO NOTHING is safe.
 	_, err := a.db.NewInsert().
 		Model(pointer.For(occurrence)).
+		On("CONFLICT DO NOTHING").
 		Exec(ctx)
 	return err
 }

internal/triggers/listener.go

@@ -148,14 +148,24 @@ func handleMessage(
 			searchAttributes[workflow.SearchAttributeTriggerID] = trigger.ID
 		}
 
-		options := client.StartWorkflowOptions{
-			TaskQueue:        taskQueue,
-			SearchAttributes: searchAttributes,
-		}
+		// Derive a deterministic workflow ID so an at-least-once redelivery of
+		// the same event does not start a second trigger execution (which would
+		// replay side-effecting stages such as money movements). For
+		// SAVED_PAYMENT/SAVED_ACCOUNT we key on the object id (dedup across
+		// distinct deliveries of the same object); for every other event type
+		// we fall back to the message UUID, which is preserved across
+		// redeliveries.
+		dedupKey := msg.UUID
 		if objectID != nil {
-			options.ID = taskIDPrefix + "-" + trigger.ID + "-" + *objectID
-			options.WorkflowIDReusePolicy = enums.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE
-			options.WorkflowExecutionErrorWhenAlreadyStarted = true
+			dedupKey = *objectID
+		}
+
+		options := client.StartWorkflowOptions{
+			TaskQueue:                                taskQueue,
+			SearchAttributes:                         searchAttributes,
+			ID:                                       taskIDPrefix + "-" + trigger.ID + "-" + dedupKey,
+			WorkflowIDReusePolicy:                    enums.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE,
+			WorkflowExecutionErrorWhenAlreadyStarted: true,
 		}
 
 		_, execErr := temporalClient.ExecuteWorkflow(ctx, options, ExecuteTrigger, ProcessEventRequest{

internal/triggers/listener_test.go

@@ -379,4 +379,42 @@ func TestHandleMessage(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, 1, count)
 	})
+
+	t.Run("redelivery of a non-payment event is skipped", func(t *testing.T) {
+		t.Parallel()
+
+		db := setupTestDB(t)
+		taskQueue := setupWorker(t, db)
+
+		w := insertNoOpWorkflow(t, db)
+		insertTrigger(t, db, w.ID, "NEW_TRANSACTION", nil, nil)
+
+		event := makeMessage("NEW_TRANSACTION", "v1", map[string]any{})
+		evaluator := NewDefaultExpressionEvaluator()
+
+		// Simulate an at-least-once redelivery: the broker re-delivers the same
+		// message, preserving its UUID.
+		msg1 := publish.NewMessage(logging.TestingContext(), *event)
+		msg2 := publish.NewMessage(logging.TestingContext(), *event)
+		msg2.UUID = msg1.UUID
+
+		require.NoError(t, handleMessage(devServer.Client(), db, evaluator, "test", "test", taskQueue, false, msg1))
+		require.Eventually(t, func() bool {
+			count, err := db.NewSelect().
+				Model((*Occurrence)(nil)).
+				Count(logging.TestingContext())
+			return err == nil && count == 1
+		}, 10*time.Second, 200*time.Millisecond)
+
+		require.NoError(t, handleMessage(devServer.Client(), db, evaluator, "test", "test", taskQueue, false, msg2))
+
+		// The deterministic workflow id (keyed on the message UUID) must reject
+		// the duplicate, so no second trigger execution / occurrence is created.
+		time.Sleep(2 * time.Second)
+		count, err := db.NewSelect().
+			Model((*Occurrence)(nil)).
+			Count(logging.TestingContext())
+		require.NoError(t, err)
+		require.Equal(t, 1, count)
+	})
 }

internal/triggers/trigger.go

@@ -112,9 +112,13 @@ type Occurrence struct {
 	Error              *string              `json:"error,omitempty" bun:"error"`
 }
 
-func NewTriggerOccurrence(triggerID string, event publish.EventMessage, at time.Time) Occurrence {
+// NewTriggerOccurrence builds an occurrence with a caller-provided id. The id
+// must be deterministic when called from workflow code (e.g. the workflow
+// execution id) so that it stays stable across Temporal replays — generating a
+// random uuid here would yield a different occurrence id on every replay.
+func NewTriggerOccurrence(id, triggerID string, event publish.EventMessage, at time.Time) Occurrence {
 	return Occurrence{
-		ID:        uuid.NewString(),
+		ID:        id,
 		TriggerID: triggerID,
 		Date:      at,
 		Event:     event,

internal/triggers/workflow_trigger.go

@@ -68,7 +68,11 @@ func (w triggerWorkflow) RunTrigger(ctx temporalworkflow.Context, req ProcessEve
 func (w triggerWorkflow) ExecuteTrigger(ctx temporalworkflow.Context, req ProcessEventRequest, trigger Trigger) error {
 
 	vars := make(map[string]string)
-	occurrence := NewTriggerOccurrence(trigger.ID, req.Event, temporalworkflow.Now(ctx))
+	// Use the (deterministic, replay-stable) workflow execution id as the
+	// occurrence id rather than a random uuid generated in workflow code.
+	occurrence := NewTriggerOccurrence(
+		temporalworkflow.GetInfo(ctx).WorkflowExecution.ID,
+		trigger.ID, req.Event, temporalworkflow.Now(ctx))
 	err := temporalworkflow.ExecuteActivity(
 		temporalworkflow.WithActivityOptions(ctx, temporalworkflow.ActivityOptions{
 			StartToCloseTimeout: 10 * time.Second,

internal/workflow/activities.go

@@ -70,9 +70,14 @@ func (a Activities) SendWorkflowStageTerminationEvent(ctx context.Context, insta
 
 func (a Activities) InsertNewInstance(ctx context.Context, workflowID string) (*Instance, error) {
 	instance := NewInstance(activity.GetInfo(ctx).WorkflowExecution.ID, workflowID)
+	// Idempotent: the primary key is the (deterministic) workflow execution id,
+	// so a Temporal retry after a lost ack must not fail on the duplicate key.
+	// The returned instance is rebuilt deterministically, so it is correct even
+	// when the row already exists.
 	if _, err := a.db.
 		NewInsert().
 		Model(&instance).
+		On("CONFLICT DO NOTHING").
 		Exec(ctx); err != nil {
 		return nil, err
 	}
@@ -90,8 +95,12 @@ func (a Activities) UpdateInstance(ctx context.Context, instance *Instance) erro
 
 func (a Activities) InsertNewStage(ctx context.Context, instance Instance, ind int) (*Stage, error) {
 	stage := NewStage(instance.ID, activity.GetInfo(ctx).WorkflowExecution.RunID, ind)
+	// Idempotent: the primary key is deterministic (instance id + run id +
+	// index), so a Temporal retry after a lost ack must not fail on the
+	// duplicate key.
 	if _, err := a.db.NewInsert().
 		Model(&stage).
+		On("CONFLICT DO NOTHING").
 		Exec(ctx); err != nil {
 		return nil, err
 	}