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

Replace transaction.OnCommit fire-and-forget publish with a postCommit
closure pattern that defers event publishing until after transaction
commits while still propagating publish errors to the caller. 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

@@ -130,9 +130,14 @@ func (h *Handler) Handle(ctx context.Context, event *ExecuteSyncPlanEvent) error
 		return fmt.Errorf("sync plan %s has no app ID", plan.ID)
 	}
 
+	// postCommit holds a publish call that must run after the transaction commits.
+	// By capturing it here and executing after RunWithNoValue returns, we ensure
+	// the DB state is visible before the event fires, while still propagating errors.
+	var postCommit func() 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 {
+	if err := transaction.RunWithNoValue(ctx, h.adapter, func(ctx context.Context) error {
 		// Acquire advisory lock scoped to this invoice
 		if err := h.lockFunc(ctx, event.Namespace, event.InvoiceID); err != nil {
 			if errors.Is(err, ErrSyncPlanLocked) {
@@ -204,20 +209,14 @@ func (h *Handler) Handle(ctx context.Context, event *ExecuteSyncPlanEvent) error
 		}
 
 		if !result.Done {
-			transaction.OnCommit(ctx, func(ctx context.Context) {
-				if err := h.publisher.Publish(ctx, ExecuteSyncPlanEvent{
+			postCommit = func() error {
+				return 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
 		}
 
@@ -232,12 +231,21 @@ func (h *Handler) Handle(ctx context.Context, event *ExecuteSyncPlanEvent) error
 			return fmt.Errorf("refreshing sync plan: %w", err)
 		}
 
-		return h.handlePlanCompletion(ctx, event, plan)
-	})
+		return h.handlePlanCompletion(ctx, event, plan, &postCommit)
+	}); err != nil {
+		return err
+	}
+
+	if postCommit != nil {
+		return postCommit()
+	}
+
+	return nil
 }
 
 // handlePlanCompletion writes results back to the invoice and triggers advancement.
-func (h *Handler) handlePlanCompletion(ctx context.Context, event *ExecuteSyncPlanEvent, plan *SyncPlan) error {
+// The postCommit parameter captures any publish that must happen after the transaction commits.
+func (h *Handler) handlePlanCompletion(ctx context.Context, event *ExecuteSyncPlanEvent, plan *SyncPlan, postCommit *func() error) error {
 	invoiceID := billing.InvoiceID{
 		Namespace: event.Namespace,
 		ID:        event.InvoiceID,
@@ -299,18 +307,13 @@ func (h *Handler) handlePlanCompletion(ctx context.Context, event *ExecuteSyncPl
 		}
 
 	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{
+		// For delete, we just need to advance the invoice after the transaction commits.
+		*postCommit = func() error {
+			return 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
 
 	default: