Merge pull request #1901 from LerianStudio/feature/mdz-1880-4

feat: add balance scheduling tests and async persistence support - balance update

Author: Ygohr

components/transaction/internal/adapters/redis/balance_schedule_test.go

@@ -0,0 +1,45 @@
+// Copyright (c) 2026 Lerian Studio. All rights reserved.
+// Use of this source code is governed by the Elastic License 2.0
+// that can be found in the LICENSE file.
+
+package redis
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/LerianStudio/midaz/v3/pkg/utils"
+	"github.com/stretchr/testify/assert"
+)
+
+// TestLuaScript_ContainsScheduleLogic is a structural verification test that checks
+// if the Lua script contains the expected scheduling tokens (ZADD, scheduleSync, etc.).
+//
+// IMPORTANT: This test only asserts string presence, not runtime behavior.
+// It protects against accidental removal of scheduling logic during refactors.
+// Actual scheduling behavior is tested in integration tests (balance.worker_integration_test.go).
+func TestLuaScript_ContainsScheduleLogic(t *testing.T) {
+	// Verify the script has ZADD for scheduling
+	assert.True(t, strings.Contains(balanceAtomicOperationLua, "ZADD"),
+		"Lua script should contain ZADD for balance scheduling")
+
+	// Verify the script checks scheduleSync flag
+	assert.True(t, strings.Contains(balanceAtomicOperationLua, "scheduleSync"),
+		"Lua script should check scheduleSync flag")
+
+	// Verify the script uses scheduleKey from KEYS[3]
+	assert.True(t, strings.Contains(balanceAtomicOperationLua, "KEYS[3]"),
+		"Lua script should reference schedule key from KEYS[3]")
+
+	// Verify the script calculates dueAt for pre-expiry warning
+	assert.True(t, strings.Contains(balanceAtomicOperationLua, "dueAt"),
+		"Lua script should calculate dueAt for scheduling")
+}
+
+// TestScheduleKeyConstant verifies the schedule key constant matches
+// what the BalanceSyncWorker expects.
+func TestScheduleKeyConstant(t *testing.T) {
+	expectedKey := "schedule:{transactions}:balance-sync"
+	assert.Equal(t, expectedKey, utils.BalanceSyncScheduleKey,
+		"Schedule key constant should match expected format")
+}

components/transaction/internal/adapters/redis/consumer.redis.go

@@ -356,6 +356,20 @@ func balanceRedisToBalance(b mmodel.BalanceRedis, mapBalances map[string]*mmodel
 	}
 }
 
+// ProcessBalanceAtomicOperation executes the balance_atomic_operation.lua script.
+//
+// BALANCE SCHEDULING FLOW:
+//  1. This method calls the Lua script with scheduleSync=1 (when balanceSyncEnabled is true)
+//  2. Lua script updates hot balance in Redis atomically
+//  3. Lua script calculates dueAt = now + ttl - 600 (10 min before expiry)
+//  4. Lua script calls ZADD to schedule:{transactions}:balance-sync with score=dueAt
+//  5. BalanceSyncWorker polls the sorted set for due balances
+//  6. Worker calls SyncBalancesBatch to persist to database
+//
+// This enables decoupled balance persistence:
+//   - Transaction handler is append-only (no blocking on balance DB write)
+//   - Balance DB writes are aggregated and batched by the worker
+//   - Only the latest version per balance is persisted (aggregation)
 func (rr *RedisConsumerRepository) ProcessBalanceAtomicOperation(ctx context.Context, organizationID, ledgerID, transactionID uuid.UUID, transactionStatus string, pending bool, balancesOperation []mmodel.BalanceOperation) (*mmodel.BalanceAtomicResult, error) {
 	logger, tracer, _, _ := libCommons.NewTrackingFromContext(ctx)
 

components/transaction/internal/services/command/create-balance-transaction-operations-async.go

@@ -28,7 +28,16 @@ import (
 	"go.opentelemetry.io/otel/trace"
 )
 
-// CreateBalanceTransactionOperationsAsync func that is responsible to create all transactions at the same async.
+// CreateBalanceTransactionOperationsAsync processes transaction asynchronously.
+// This is an append-only handler for transactions and operations:
+// - Hot balance already updated atomically by Lua script during validation
+// - Cold balance scheduled for async sync via sorted set (Lua script does ZADD)
+// - Transaction and operations persisted to database
+// - Events sent asynchronously
+//
+// Balance persistence is fully async via BalanceSyncWorker.
+// The Lua script (balance_atomic_operation.lua) does ZADD to schedule:balance-sync
+// when scheduleSync=1, which is the default for all balance-affecting transactions.
 func (uc *UseCase) CreateBalanceTransactionOperationsAsync(ctx context.Context, data mmodel.Queue) error {
 	logger, tracer, _, _ := libCommons.NewTrackingFromContext(ctx)
 
@@ -45,22 +54,6 @@ func (uc *UseCase) CreateBalanceTransactionOperationsAsync(ctx context.Context,
 		}
 	}
 
-	if t.Transaction.Status.Code != constant.NOTED {
-		ctxProcessBalances, spanUpdateBalances := tracer.Start(ctx, "command.create_balance_transaction_operations.update_balances")
-		defer spanUpdateBalances.End()
-
-		logger.Infof("Trying to update balances")
-
-		err := uc.UpdateBalances(ctxProcessBalances, data.OrganizationID, data.LedgerID, *t.Validate, t.Balances, t.BalancesAfter)
-		if err != nil {
-			libOpentelemetry.HandleSpanBusinessErrorEvent(&spanUpdateBalances, "Failed to update balances", err)
-
-			logger.Errorf("Failed to update balances: %v", err.Error())
-
-			return err
-		}
-	}
-
 	ctxProcessTransaction, spanUpdateTransaction := tracer.Start(ctx, "command.create_balance_transaction_operations.create_transaction")
 	defer spanUpdateTransaction.End()