fix(invoicesync): use post-commit publish with error propagation

Use transaction.Run to return a postTxEvent from the transaction body,
then publish after commit using the original (non-tx) context. This
defers event publishing until DB state is visible while keeping publish
errors in the normal return flow. Also fix large payload test to use
JSONEq for Postgres jsonb normalization.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: hekike

openmeter/app/stripe/invoicesync/adapter/adapter_test.go

@@ -506,7 +506,9 @@ func TestCreateSyncPlan_LargePayload(t *testing.T) {
 	fetched, err := setup.adapter.GetSyncPlan(ctx, plan.ID)
 	require.NoError(t, err)
 	require.Len(t, fetched.Operations, 1)
-	assert.Equal(t, rawPayload, []byte(fetched.Operations[0].Payload))
+	assert.JSONEq(t, string(rawPayload), string(fetched.Operations[0].Payload))
+	// Verify no truncation: fetched payload should be at least as large as the input.
+	assert.GreaterOrEqual(t, len(fetched.Operations[0].Payload), len(rawPayload))
 }
 
 // TestCreateSyncPlan_DuplicateIdempotencyKey verifies that creating two plans that happen to share

openmeter/app/stripe/invoicesync/handler.go

@@ -14,10 +14,17 @@ import (
 	"github.com/openmeterio/openmeter/openmeter/secret"
 	secretentity "github.com/openmeterio/openmeter/openmeter/secret/entity"
 	"github.com/openmeterio/openmeter/openmeter/watermill/eventbus"
+	"github.com/openmeterio/openmeter/openmeter/watermill/marshaler"
 	"github.com/openmeterio/openmeter/pkg/clock"
 	"github.com/openmeterio/openmeter/pkg/framework/transaction"
 )
 
+// postTxEvent holds an event that must be published after the transaction commits.
+// Using a concrete type instead of closures avoids capturing the transaction context.
+type postTxEvent struct {
+	event marshaler.Event // nil means nothing to publish
+}
+
 // LockFunc acquires an advisory lock for the given invoice within the current transaction context.
 // This ensures only one sync plan executes at a time per invoice.
 // It should block until the lock is available, or return ErrSyncPlanLocked if the lock is
@@ -132,44 +139,45 @@ func (h *Handler) Handle(ctx context.Context, event *ExecuteSyncPlanEvent) error
 
 	// Execute within a transaction with an advisory lock to prevent parallel execution
 	// of multiple plans for the same invoice (e.g., draft and issuing plans).
-	return transaction.RunWithNoValue(ctx, h.adapter, func(ctx context.Context) error {
+	// Returns a postTxEvent carrying any event that must be published after commit.
+	result, err := transaction.Run(ctx, h.adapter, func(ctx context.Context) (postTxEvent, error) {
 		// Acquire advisory lock scoped to this invoice
 		if err := h.lockFunc(ctx, event.Namespace, event.InvoiceID); err != nil {
 			if errors.Is(err, ErrSyncPlanLocked) {
 				logger.InfoContext(ctx, "invoice sync locked by another plan, skipping")
-				return nil
+				return postTxEvent{}, nil
 			}
-			return fmt.Errorf("acquiring sync plan lock: %w", err)
+			return postTxEvent{}, fmt.Errorf("acquiring sync plan lock: %w", err)
 		}
 
 		// Re-fetch plan inside lock — another worker may have advanced it
 		plan, err = h.adapter.GetSyncPlan(ctx, event.PlanID)
 		if err != nil {
-			return fmt.Errorf("getting sync plan under lock: %w", err)
+			return postTxEvent{}, fmt.Errorf("getting sync plan under lock: %w", err)
 		}
 		if plan == nil || plan.Status == PlanStatusCompleted || plan.Status == PlanStatusFailed {
-			return nil
+			return postTxEvent{}, nil
 		}
 
 		// Check if a newer plan exists for this invoice — if so, this plan has been
 		// superseded and should not continue executing (even if cancelAllActivePlans
 		// already marked it as failed, the events may already be in-flight).
 		superseded, err := h.isSuperseded(ctx, plan)
 		if err != nil {
-			return fmt.Errorf("checking if plan is superseded: %w", err)
+			return postTxEvent{}, fmt.Errorf("checking if plan is superseded: %w", err)
 		}
 		if superseded {
 			logger.InfoContext(ctx, "plan superseded by a newer plan, canceling")
 			if err := h.adapter.FailPlan(ctx, plan.ID, "superseded by newer plan"); err != nil {
 				logger.ErrorContext(ctx, "failed to cancel superseded plan", "error", err)
 			}
-			return nil
+			return postTxEvent{}, nil
 		}
 
 		// Create Stripe client
 		stripeClient, err := h.createStripeClient(ctx, event.Namespace, plan)
 		if err != nil {
-			return fmt.Errorf("creating stripe client: %w", err)
+			return postTxEvent{}, fmt.Errorf("creating stripe client: %w", err)
 		}
 
 		// Execute next operation
@@ -178,9 +186,9 @@ func (h *Handler) Handle(ctx context.Context, event *ExecuteSyncPlanEvent) error
 			Logger:  logger,
 		}
 
-		result, err := executor.ExecuteNextOperation(ctx, stripeClient, plan)
+		execResult, err := executor.ExecuteNextOperation(ctx, stripeClient, plan)
 		if err != nil {
-			return err
+			return postTxEvent{}, err
 		}
 
 		// Write back external IDs immediately so the invoice always reflects Stripe state.
@@ -189,55 +197,57 @@ func (h *Handler) Handle(ctx context.Context, event *ExecuteSyncPlanEvent) error
 		//
 		// On failure this rolls back the transaction (including CompleteOperation), causing
 		// Kafka to redeliver. The idempotency key ensures no duplicate Stripe API calls.
-		if result.InvoicingExternalID != nil || len(result.LineExternalIDs) > 0 || len(result.LineDiscountExternalIDs) > 0 {
+		if execResult.InvoicingExternalID != nil || len(execResult.LineExternalIDs) > 0 || len(execResult.LineDiscountExternalIDs) > 0 {
 			if err := h.billingService.SyncExternalIDs(ctx, billing.SyncExternalIDsInput{
 				Invoice: billing.InvoiceID{
 					Namespace: event.Namespace,
 					ID:        event.InvoiceID,
 				},
-				InvoicingExternalID:     result.InvoicingExternalID,
-				LineExternalIDs:         result.LineExternalIDs,
-				LineDiscountExternalIDs: result.LineDiscountExternalIDs,
+				InvoicingExternalID:     execResult.InvoicingExternalID,
+				LineExternalIDs:         execResult.LineExternalIDs,
+				LineDiscountExternalIDs: execResult.LineDiscountExternalIDs,
 			}); err != nil {
-				return fmt.Errorf("syncing external IDs: %w", err)
+				return postTxEvent{}, fmt.Errorf("syncing external IDs: %w", err)
 			}
 		}
 
-		if !result.Done {
-			transaction.OnCommit(ctx, func(ctx context.Context) {
-				if err := h.publisher.Publish(ctx, ExecuteSyncPlanEvent{
-					PlanID:     event.PlanID,
-					InvoiceID:  event.InvoiceID,
-					Namespace:  event.Namespace,
-					CustomerID: event.CustomerID,
-				}); err != nil {
-					h.logger.ErrorContext(ctx, "failed to publish next sync plan event",
-						"plan_id", event.PlanID,
-						"invoice_id", event.InvoiceID,
-						"error", err,
-					)
-				}
-			})
-			return nil
+		if !execResult.Done {
+			return postTxEvent{event: ExecuteSyncPlanEvent{
+				PlanID:     event.PlanID,
+				InvoiceID:  event.InvoiceID,
+				Namespace:  event.Namespace,
+				CustomerID: event.CustomerID,
+			}}, nil
 		}
 
-		if result.Failed {
-			logger.ErrorContext(ctx, "sync plan failed", "error", result.FailError)
-			return h.handlePlanFailure(ctx, event, plan.Phase, result.FailError)
+		if execResult.Failed {
+			logger.ErrorContext(ctx, "sync plan failed", "error", execResult.FailError)
+			return postTxEvent{}, h.handlePlanFailure(ctx, event, plan.Phase, execResult.FailError)
 		}
 
 		// Refresh plan to get completed operation responses
 		plan, err = h.adapter.GetSyncPlan(ctx, event.PlanID)
 		if err != nil {
-			return fmt.Errorf("refreshing sync plan: %w", err)
+			return postTxEvent{}, fmt.Errorf("refreshing sync plan: %w", err)
 		}
 
 		return h.handlePlanCompletion(ctx, event, plan)
 	})
+	if err != nil {
+		return err
+	}
+
+	// Publish after commit so DB state is visible, using the original (non-tx) context.
+	if result.event != nil {
+		return h.publisher.Publish(ctx, result.event)
+	}
+
+	return nil
 }
 
 // handlePlanCompletion writes results back to the invoice and triggers advancement.
-func (h *Handler) handlePlanCompletion(ctx context.Context, event *ExecuteSyncPlanEvent, plan *SyncPlan) error {
+// Returns a postTxEvent if an event must be published after the transaction commits.
+func (h *Handler) handlePlanCompletion(ctx context.Context, event *ExecuteSyncPlanEvent, plan *SyncPlan) (postTxEvent, error) {
 	invoiceID := billing.InvoiceID{
 		Namespace: event.Namespace,
 		ID:        event.InvoiceID,
@@ -248,7 +258,7 @@ func (h *Handler) handlePlanCompletion(ctx context.Context, event *ExecuteSyncPl
 	case SyncPlanPhaseDraft:
 		upsertResult, err := BuildUpsertResultFromPlan(plan)
 		if err != nil {
-			return fmt.Errorf("building upsert result: %w", err)
+			return postTxEvent{}, fmt.Errorf("building upsert result: %w", err)
 		}
 
 		_, err = h.billingService.SyncDraftInvoice(ctx, billing.SyncDraftStandardInvoiceInput{
@@ -268,15 +278,15 @@ func (h *Handler) handlePlanCompletion(ctx context.Context, event *ExecuteSyncPl
 					"plan_id", plan.ID,
 					"error", err,
 				)
-				return nil
+				return postTxEvent{}, nil
 			}
-			return fmt.Errorf("syncing draft invoice: %w", err)
+			return postTxEvent{}, fmt.Errorf("syncing draft invoice: %w", err)
 		}
 
 	case SyncPlanPhaseIssuing:
 		finalizeResult, err := BuildFinalizeResultFromPlan(plan)
 		if err != nil {
-			return fmt.Errorf("building finalize result: %w", err)
+			return postTxEvent{}, fmt.Errorf("building finalize result: %w", err)
 		}
 
 		_, err = h.billingService.SyncIssuingInvoice(ctx, billing.SyncIssuingStandardInvoiceInput{
@@ -293,31 +303,23 @@ func (h *Handler) handlePlanCompletion(ctx context.Context, event *ExecuteSyncPl
 					"plan_id", plan.ID,
 					"error", err,
 				)
-				return nil
+				return postTxEvent{}, nil
 			}
-			return fmt.Errorf("syncing issuing invoice: %w", err)
+			return postTxEvent{}, fmt.Errorf("syncing issuing invoice: %w", err)
 		}
 
 	case SyncPlanPhaseDelete:
-		// For delete, we just need to advance the invoice
-		transaction.OnCommit(ctx, func(ctx context.Context) {
-			if err := h.publisher.Publish(ctx, billing.AdvanceStandardInvoiceEvent{
-				Invoice:    invoiceID,
-				CustomerID: event.CustomerID,
-			}); err != nil {
-				h.logger.ErrorContext(ctx, "failed to publish advance event after delete sync",
-					"invoice_id", invoiceID.ID,
-					"error", err,
-				)
-			}
-		})
-		return nil
+		// For delete, we just need to advance the invoice after the transaction commits.
+		return postTxEvent{event: billing.AdvanceStandardInvoiceEvent{
+			Invoice:    invoiceID,
+			CustomerID: event.CustomerID,
+		}}, nil
 
 	default:
-		return fmt.Errorf("unknown sync plan phase %q for plan %s", plan.Phase, plan.ID)
+		return postTxEvent{}, fmt.Errorf("unknown sync plan phase %q for plan %s", plan.Phase, plan.ID)
 	}
 
-	return nil
+	return postTxEvent{}, nil
 }
 
 // handlePlanFailure triggers the invoice into a sync-failed state, surfacing the Stripe error