/*

Copyright 2024 Blnk Finance Authors.

Licensed under the Apache License, Version 2.0 (the "License");

you may not use this file except in compliance with the License.

You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software

distributed under the License is distributed on an "AS IS" BASIS,

WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

See the License for the specific language governing permissions and

limitations under the License.

*/

package blnk

import (

"context"

"encoding/json"

"fmt"

"math/big"

"regexp"

"sort"

"strings"

"sync"

"testing"

"time"

"github.com/blnkfinance/blnk/config"

"github.com/blnkfinance/blnk/database"

"github.com/blnkfinance/blnk/database/mocks"

redlock "github.com/blnkfinance/blnk/internal/lock"

redis_db "github.com/blnkfinance/blnk/internal/redis-db"

"github.com/blnkfinance/blnk/model"

"github.com/go-redis/redismock/v9"

"github.com/hibiken/asynq"

"github.com/brianvoe/gofakeit/v6"

"github.com/DATA-DOG/go-sqlmock"

"github.com/stretchr/testify/assert"

"github.com/stretchr/testify/mock"

"github.com/stretchr/testify/require"

)

func TestAcquireTransactionLocker(t *testing.T) {

t.Run("fail fast when wait timeout disabled", func(t *testing.T) {

db, mock := redismock.NewClientMock()

blnk := &Blnk{

redis: db,

config: &config.Configuration{

Transaction: config.TransactionConfig{

LockDuration: 5 * time.Second,

},

},

}

locker := redlock.NewMultiLocker(db, []string{"key-b", "key-a"}, "test-value")

mock.ExpectSetNX("key-a", "test-value", 5*time.Second).SetVal(true)

mock.ExpectSetNX("key-b", "test-value", 5*time.Second).SetVal(true)

err := blnk.acquireTransactionLocker(context.Background(), locker)

require.NoError(t, err)

require.NoError(t, mock.ExpectationsWereMet())

})

t.Run("waits when configured", func(t *testing.T) {

db, mock := redismock.NewClientMock()

blnk := &Blnk{

redis: db,

config: &config.Configuration{

Transaction: config.TransactionConfig{

LockDuration: 5 * time.Second,

LockWaitTimeout: time.Second,

},

},

}

locker := redlock.NewMultiLocker(db, []string{"key-b", "key-a"}, "test-value")

mock.ExpectSetNX("key-a", "test-value", 5*time.Second).SetVal(false)

mock.ExpectSetNX("key-a", "test-value", 5*time.Second).SetVal(true)

mock.ExpectSetNX("key-b", "test-value", 5*time.Second).SetVal(true)

err := blnk.acquireTransactionLocker(context.Background(), locker)

require.NoError(t, err)

require.NoError(t, mock.ExpectationsWereMet())

})

}

// pollForTransactionStatus polls for a transaction until it reaches the expected status or a timeout occurs.

func pollForTransactionStatus(ctx context.Context, ds database.IDataSource, transactionRef string, expectedStatus string, pollInterval time.Duration, timeoutDuration time.Duration) (*model.Transaction, error) {

timeoutCtx, cancel := context.WithTimeout(ctx, timeoutDuration)

defer cancel()

ticker := time.NewTicker(pollInterval)

defer ticker.Stop()

for {

select {

case <-timeoutCtx.Done():

return nil, fmt.Errorf("timed out waiting for transaction ref %s to reach status %s: %w", transactionRef, expectedStatus, timeoutCtx.Err())

case <-ticker.C:

txn, err := ds.GetTransactionByRef(timeoutCtx, transactionRef) // Use timeoutCtx for the DB call as well

if err != nil {

// If the error is that the transaction is not found, we should continue polling.

// For other errors, the outer timeout will eventually catch persistent issues.

continue

}

// If err is nil, txn is a valid model.Transaction struct.

if txn.Status == expectedStatus {

return &txn, nil // Return a pointer to the transaction

}

}

}

}

func TestRecordTransaction(t *testing.T) {

cnf := &config.Configuration{

Redis: config.RedisConfig{

Dns: "localhost:6379",

},

Queue: config.QueueConfig{

WebhookQueue: "webhook_queue",

TransactionQueue: "transaction_queue",

IndexQueue: "index_queue",

NumberOfQueues: 1,

},

Server: config.ServerConfig{SecretKey: "some-secret"},

AccountNumberGeneration: config.AccountNumberGenerationConfig{

HttpService: config.AccountGenerationHttpService{

Url: "http://example.com/generateAccount",

},

},

}

config.ConfigStore.Store(cnf)

datasource, mock, err := newTestDataSource()

assert.NoError(t, err)

// Important: Set ExpectationsWereMet to ensure execution occurs in order of appearance

mock.MatchExpectationsInOrder(false)

d, err := NewBlnk(datasource)

assert.NoError(t, err)

// Use fixed UUIDs for better predictability

source := "source-balance-id-123"

destination := "destination-balance-id-456"

reference := "transaction-reference-789"

txn := &model.Transaction{

Reference: reference,

Source: source,

Destination: destination,

Amount: 10,

AllowOverdraft: false,

Precision: 100,

Currency: "NGN",

}

// First, check if transaction exists

mock.ExpectQuery(regexp.QuoteMeta(`

SELECT EXISTS(SELECT 1 FROM blnk.transactions WHERE reference = $1)

`)).WithArgs(txn.Reference).WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(false))

// Set up source balance response

sourceBalanceRows := sqlmock.NewRows([]string{"balance_id", "indicator", "currency", "ledger_id", "balance", "credit_balance", "debit_balance", "inflight_balance", "inflight_credit_balance", "inflight_debit_balance", "created_at", "version", "track_fund_lineage", "allocation_strategy", "identity_id"}).

AddRow(source, "NGN", "", "ledger-id-source", int64(10000), int64(10000), 0, 0, 0, 0, time.Now(), 0, false, "FIFO", "")

// Set up destination balance response

destinationBalanceRows := sqlmock.NewRows([]string{"balance_id", "indicator", "currency", "ledger_id", "balance", "credit_balance", "debit_balance", "inflight_balance", "inflight_credit_balance", "inflight_debit_balance", "created_at", "version", "track_fund_lineage", "allocation_strategy", "identity_id"}).

AddRow(destination, "", "NGN", "ledger-id-destination", 0, 0, 0, 0, 0, 0, time.Now(), 0, false, "FIFO", "")

balanceQuery := `SELECT balance_id, indicator, currency, ledger_id, balance, credit_balance, debit_balance, inflight_balance, inflight_credit_balance, inflight_debit_balance, created_at, version, track_fund_lineage, COALESCE\(allocation_strategy, 'FIFO'\) as allocation_strategy, COALESCE\(identity_id, ''\) as identity_id FROM blnk.balances WHERE balance_id = \$1`

balanceQueryPattern := regexp.MustCompile(`\s+`).ReplaceAllString(balanceQuery, `\s*`)

// Expect balance queries with the source/destination IDs (order can vary)

mock.ExpectQuery(balanceQueryPattern).WithArgs(source).WillReturnRows(sourceBalanceRows)

mock.ExpectQuery(balanceQueryPattern).WithArgs(destination).WillReturnRows(destinationBalanceRows)

// Start atomic transaction for balance updates AND transaction insert

mock.ExpectBegin()

// Update source and destination balances (order doesn't matter since MatchExpectationsInOrder is false)

mock.ExpectExec(regexp.QuoteMeta(`

UPDATE blnk.balances

SET balance = $2, credit_balance = $3, debit_balance = $4, inflight_balance = $5, inflight_credit_balance = $6, inflight_debit_balance = $7, currency = $8, ledger_id = $9, created_at = $10, version = version + 1

WHERE balance_id = $1 AND version = $11

`)).WithArgs(

source, // balance_id

sqlmock.AnyArg(), // balance

sqlmock.AnyArg(), // credit_balance

sqlmock.AnyArg(), // debit_balance

sqlmock.AnyArg(), // inflight_balance

sqlmock.AnyArg(), // inflight_credit_balance

sqlmock.AnyArg(), // inflight_debit_balance

sqlmock.AnyArg(), // currency

sqlmock.AnyArg(), // ledger_id

sqlmock.AnyArg(), // created_at

sqlmock.AnyArg(), // version

).WillReturnResult(sqlmock.NewResult(1, 1))

mock.ExpectExec(regexp.QuoteMeta(`

UPDATE blnk.balances

SET balance = $2, credit_balance = $3, debit_balance = $4, inflight_balance = $5, inflight_credit_balance = $6, inflight_debit_balance = $7, currency = $8, ledger_id = $9, created_at = $10, version = version + 1

WHERE balance_id = $1 AND version = $11

`)).WithArgs(

destination, // balance_id

sqlmock.AnyArg(), // balance

sqlmock.AnyArg(), // credit_balance

sqlmock.AnyArg(), // debit_balance

sqlmock.AnyArg(), // inflight_balance

sqlmock.AnyArg(), // inflight_credit_balance

sqlmock.AnyArg(), // inflight_debit_balance

sqlmock.AnyArg(), // currency

sqlmock.AnyArg(), // ledger_id

sqlmock.AnyArg(), // created_at

sqlmock.AnyArg(), // version

).WillReturnResult(sqlmock.NewResult(1, 1))

// Expect transaction insertion within the same atomic transaction

expectedSQL := `INSERT INTO blnk.transactions(transaction_id, parent_transaction, source, reference, amount, precise_amount, precision, currency, destination, description, status, created_at, meta_data, scheduled_for, hash, effective_date) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)`

mock.ExpectExec(regexp.QuoteMeta(expectedSQL)).WithArgs(

sqlmock.AnyArg(), // transaction_id

sqlmock.AnyArg(), // parent_transaction

source, // source

reference, // reference

txn.AmountString, // amount

"1000", // precise amount

txn.Precision, // precision

txn.Currency, // currency

destination, // destination

sqlmock.AnyArg(), // description

sqlmock.AnyArg(), // status

sqlmock.AnyArg(), // created_at

sqlmock.AnyArg(), // meta_data

sqlmock.AnyArg(), // scheduled_for

sqlmock.AnyArg(), // hash

sqlmock.AnyArg(), // effective_date

).WillReturnResult(sqlmock.NewResult(1, 1))

// Commit atomic transaction (balance updates + transaction insert)

mock.ExpectCommit()

// Execute the function being tested

result, err := d.RecordTransaction(context.Background(), txn)

// Assert no errors

assert.NoError(t, err)

// Check if all expectations were met

if err := mock.ExpectationsWereMet(); err != nil {

t.Errorf("there were unfulfilled expectations: %s", err)

}

// Optionally check the result if needed

assert.NotNil(t, result)

}

func TestRecordTransaction_AtomicRollback_InsertFails(t *testing.T) {

cnf := &config.Configuration{

Redis: config.RedisConfig{

Dns: "localhost:6379",

},

Queue: config.QueueConfig{

WebhookQueue: "webhook_queue",

TransactionQueue: "transaction_queue",

IndexQueue: "index_queue",

NumberOfQueues: 1,

},

Server: config.ServerConfig{SecretKey: "some-secret"},

AccountNumberGeneration: config.AccountNumberGenerationConfig{

HttpService: config.AccountGenerationHttpService{

Url: "http://example.com/generateAccount",

},

},

}

config.ConfigStore.Store(cnf)

datasource, mock, err := newTestDataSource()

assert.NoError(t, err)

mock.MatchExpectationsInOrder(false)

d, err := NewBlnk(datasource)

assert.NoError(t, err)

source := "source-balance-id-atomicity-test"

destination := "destination-balance-id-atomicity-test"

reference := "transaction-reference-atomicity-test"

txn := &model.Transaction{

Reference: reference,

Source: source,

Destination: destination,

Amount: 100,

AllowOverdraft: true,

Precision: 100,

Currency: "NGN",

}

mock.ExpectQuery(regexp.QuoteMeta(`

SELECT EXISTS(SELECT 1 FROM blnk.transactions WHERE reference = $1)

`)).WithArgs(txn.Reference).WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(false))

sourceBalanceRows := sqlmock.NewRows([]string{"balance_id", "indicator", "currency", "ledger_id", "balance", "credit_balance", "debit_balance", "inflight_balance", "inflight_credit_balance", "inflight_debit_balance", "created_at", "version", "track_fund_lineage", "allocation_strategy", "identity_id"}).

AddRow(source, "", "NGN", "ledger-id-source", int64(100000), int64(100000), 0, 0, 0, 0, time.Now(), 0, false, "FIFO", "")

destinationBalanceRows := sqlmock.NewRows([]string{"balance_id", "indicator", "currency", "ledger_id", "balance", "credit_balance", "debit_balance", "inflight_balance", "inflight_credit_balance", "inflight_debit_balance", "created_at", "version", "track_fund_lineage", "allocation_strategy", "identity_id"}).

AddRow(destination, "", "NGN", "ledger-id-destination", 0, 0, 0, 0, 0, 0, time.Now(), 0, false, "FIFO", "")

balanceQuery := `SELECT balance_id, indicator, currency, ledger_id, balance, credit_balance, debit_balance, inflight_balance, inflight_credit_balance, inflight_debit_balance, created_at, version, track_fund_lineage, COALESCE\(allocation_strategy, 'FIFO'\) as allocation_strategy, COALESCE\(identity_id, ''\) as identity_id FROM blnk.balances WHERE balance_id = \$1`

balanceQueryPattern := regexp.MustCompile(`\s+`).ReplaceAllString(balanceQuery, `\s*`)

mock.ExpectQuery(balanceQueryPattern).WithArgs(source).WillReturnRows(sourceBalanceRows)

mock.ExpectQuery(balanceQueryPattern).WithArgs(destination).WillReturnRows(destinationBalanceRows)

mock.ExpectBegin()

mock.ExpectExec(regexp.QuoteMeta(`

UPDATE blnk.balances

SET balance = $2, credit_balance = $3, debit_balance = $4, inflight_balance = $5, inflight_credit_balance = $6, inflight_debit_balance = $7, currency = $8, ledger_id = $9, created_at = $10, version = version + 1

WHERE balance_id = $1 AND version = $11

`)).WithArgs(

source,

sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),

sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),

sqlmock.AnyArg(), sqlmock.AnyArg(),

).WillReturnResult(sqlmock.NewResult(1, 1))

mock.ExpectExec(regexp.QuoteMeta(`

UPDATE blnk.balances

SET balance = $2, credit_balance = $3, debit_balance = $4, inflight_balance = $5, inflight_credit_balance = $6, inflight_debit_balance = $7, currency = $8, ledger_id = $9, created_at = $10, version = version + 1

WHERE balance_id = $1 AND version = $11

`)).WithArgs(

destination,

sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),

sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),

sqlmock.AnyArg(), sqlmock.AnyArg(),

).WillReturnResult(sqlmock.NewResult(1, 1))

expectedSQL := `INSERT INTO blnk.transactions(transaction_id, parent_transaction, source, reference, amount, precise_amount, precision, currency, destination, description, status, created_at, meta_data, scheduled_for, hash, effective_date) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)`

mock.ExpectExec(regexp.QuoteMeta(expectedSQL)).WithArgs(

sqlmock.AnyArg(), sqlmock.AnyArg(), source, reference,

sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),

destination, sqlmock.AnyArg(), sqlmock.AnyArg(),

sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),

).WillReturnError(fmt.Errorf("database connection lost - simulated failure"))

mock.ExpectRollback()

_, err = d.RecordTransaction(context.Background(), txn)

assert.Error(t, err)

assert.Contains(t, err.Error(), "failed to persist transaction with balances")

if err := mock.ExpectationsWereMet(); err != nil {

t.Errorf("there were unfulfilled expectations: %s", err)

}

}

func TestRecordTransaction_AtomicRollback_SecondBalanceUpdateFails(t *testing.T) {

cnf := &config.Configuration{

Redis: config.RedisConfig{

Dns: "localhost:6379",

},

Queue: config.QueueConfig{

WebhookQueue: "webhook_queue",

TransactionQueue: "transaction_queue",

IndexQueue: "index_queue",

NumberOfQueues: 1,

},

Server: config.ServerConfig{SecretKey: "some-secret"},

AccountNumberGeneration: config.AccountNumberGenerationConfig{

HttpService: config.AccountGenerationHttpService{

Url: "http://example.com/generateAccount",

},

},

}

config.ConfigStore.Store(cnf)

datasource, mock, err := newTestDataSource()

assert.NoError(t, err)

mock.MatchExpectationsInOrder(true)

d, err := NewBlnk(datasource)

assert.NoError(t, err)

source := "source-balance-id-atomicity-test-2"

destination := "destination-balance-id-atomicity-test-2"

reference := "transaction-reference-atomicity-test-2"

txn := &model.Transaction{

Reference: reference,

Source: source,

Destination: destination,

Amount: 100,

AllowOverdraft: true,

Precision: 100,

Currency: "NGN",

}

mock.ExpectQuery(regexp.QuoteMeta(`

SELECT EXISTS(SELECT 1 FROM blnk.transactions WHERE reference = $1)

`)).WithArgs(txn.Reference).WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(false))

sourceBalanceRows := sqlmock.NewRows([]string{"balance_id", "indicator", "currency", "ledger_id", "balance", "credit_balance", "debit_balance", "inflight_balance", "inflight_credit_balance", "inflight_debit_balance", "created_at", "version", "track_fund_lineage", "allocation_strategy", "identity_id"}).

AddRow(source, "", "NGN", "ledger-id-source", int64(100000), int64(100000), 0, 0, 0, 0, time.Now(), 0, false, "FIFO", "")

destinationBalanceRows := sqlmock.NewRows([]string{"balance_id", "indicator", "currency", "ledger_id", "balance", "credit_balance", "debit_balance", "inflight_balance", "inflight_credit_balance", "inflight_debit_balance", "created_at", "version", "track_fund_lineage", "allocation_strategy", "identity_id"}).

AddRow(destination, "", "NGN", "ledger-id-destination", 0, 0, 0, 0, 0, 0, time.Now(), 0, false, "FIFO", "")

balanceQuery := `SELECT balance_id, indicator, currency, ledger_id, balance, credit_balance, debit_balance, inflight_balance, inflight_credit_balance, inflight_debit_balance, created_at, version, track_fund_lineage, COALESCE\(allocation_strategy, 'FIFO'\) as allocation_strategy, COALESCE\(identity_id, ''\) as identity_id FROM blnk.balances WHERE balance_id = \$1`

balanceQueryPattern := regexp.MustCompile(`\s+`).ReplaceAllString(balanceQuery, `\s*`)

mock.ExpectQuery(balanceQueryPattern).WithArgs(source).WillReturnRows(sourceBalanceRows)

mock.ExpectQuery(balanceQueryPattern).WithArgs(destination).WillReturnRows(destinationBalanceRows)

mock.ExpectBegin()

mock.ExpectExec(regexp.QuoteMeta(`

UPDATE blnk.balances

SET balance = $2, credit_balance = $3, debit_balance = $4, inflight_balance = $5, inflight_credit_balance = $6, inflight_debit_balance = $7, currency = $8, ledger_id = $9, created_at = $10, version = version + 1

WHERE balance_id = $1 AND version = $11

`)).WithArgs(

source,

sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),

sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),

sqlmock.AnyArg(), sqlmock.AnyArg(),

).WillReturnResult(sqlmock.NewResult(1, 1))

mock.ExpectExec(regexp.QuoteMeta(`

UPDATE blnk.balances

SET balance = $2, credit_balance = $3, debit_balance = $4, inflight_balance = $5, inflight_credit_balance = $6, inflight_debit_balance = $7, currency = $8, ledger_id = $9, created_at = $10, version = version + 1

WHERE balance_id = $1 AND version = $11

`)).WithArgs(

destination,

sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),

sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),

sqlmock.AnyArg(), sqlmock.AnyArg(),

).WillReturnError(fmt.Errorf("optimistic lock failure - balance was modified by another transaction"))

mock.ExpectRollback()

_, err = d.RecordTransaction(context.Background(), txn)

assert.Error(t, err)

assert.Contains(t, err.Error(), "failed to persist transaction with balances")

if err := mock.ExpectationsWereMet(); err != nil {

t.Errorf("there were unfulfilled expectations: %s", err)

}

}

func TestRecordTransaction_AtomicRollback_FirstBalanceUpdateFails(t *testing.T) {

cnf := &config.Configuration{

Redis: config.RedisConfig{

Dns: "localhost:6379",

},

Queue: config.QueueConfig{

WebhookQueue: "webhook_queue",

TransactionQueue: "transaction_queue",

IndexQueue: "index_queue",

NumberOfQueues: 1,

},

Server: config.ServerConfig{SecretKey: "some-secret"},

AccountNumberGeneration: config.AccountNumberGenerationConfig{

HttpService: config.AccountGenerationHttpService{

Url: "http://example.com/generateAccount",

},

},

}

config.ConfigStore.Store(cnf)

datasource, mock, err := newTestDataSource()

assert.NoError(t, err)

mock.MatchExpectationsInOrder(true)

d, err := NewBlnk(datasource)

assert.NoError(t, err)

source := "source-balance-id-atomicity-test-3"

destination := "destination-balance-id-atomicity-test-3"

reference := "transaction-reference-atomicity-test-3"

txn := &model.Transaction{

Reference: reference,

Source: source,

Destination: destination,

Amount: 100,

AllowOverdraft: true,

Precision: 100,

Currency: "NGN",

}

mock.ExpectQuery(regexp.QuoteMeta(`

SELECT EXISTS(SELECT 1 FROM blnk.transactions WHERE reference = $1)

`)).WithArgs(txn.Reference).WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(false))

sourceBalanceRows := sqlmock.NewRows([]string{"balance_id", "indicator", "currency", "ledger_id", "balance", "credit_balance", "debit_balance", "inflight_balance", "inflight_credit_balance", "inflight_debit_balance", "created_at", "version", "track_fund_lineage", "allocation_strategy", "identity_id"}).

AddRow(source, "", "NGN", "ledger-id-source", int64(100000), int64(100000), 0, 0, 0, 0, time.Now(), 0, false, "FIFO", "")

destinationBalanceRows := sqlmock.NewRows([]string{"balance_id", "indicator", "currency", "ledger_id", "balance", "credit_balance", "debit_balance", "inflight_balance", "inflight_credit_balance", "inflight_debit_balance", "created_at", "version", "track_fund_lineage", "allocation_strategy", "identity_id"}).

AddRow(destination, "", "NGN", "ledger-id-destination", 0, 0, 0, 0, 0, 0, time.Now(), 0, false, "FIFO", "")

balanceQuery := `SELECT balance_id, indicator, currency, ledger_id, balance, credit_balance, debit_balance, inflight_balance, inflight_credit_balance, inflight_debit_balance, created_at, version, track_fund_lineage, COALESCE\(allocation_strategy, 'FIFO'\) as allocation_strategy, COALESCE\(identity_id, ''\) as identity_id FROM blnk.balances WHERE balance_id = \$1`

balanceQueryPattern := regexp.MustCompile(`\s+`).ReplaceAllString(balanceQuery, `\s*`)

mock.ExpectQuery(balanceQueryPattern).WithArgs(source).WillReturnRows(sourceBalanceRows)

mock.ExpectQuery(balanceQueryPattern).WithArgs(destination).WillReturnRows(destinationBalanceRows)

mock.ExpectBegin()

mock.ExpectExec(regexp.QuoteMeta(`

UPDATE blnk.balances

SET balance = $2, credit_balance = $3, debit_balance = $4, inflight_balance = $5, inflight_credit_balance = $6, inflight_debit_balance = $7, currency = $8, ledger_id = $9, created_at = $10, version = version + 1

WHERE balance_id = $1 AND version = $11

`)).WithArgs(

source,

sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),

sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),

sqlmock.AnyArg(), sqlmock.AnyArg(),

).WillReturnError(fmt.Errorf("database write error - disk full"))

mock.ExpectRollback()

_, err = d.RecordTransaction(context.Background(), txn)

assert.Error(t, err)

assert.Contains(t, err.Error(), "failed to persist transaction with balances")

if err := mock.ExpectationsWereMet(); err != nil {

t.Errorf("there were unfulfilled expectations: %s", err)

}

}

func TestVoidInflightTransaction_Negative(t *testing.T) {

datasource, mock, err := newTestDataSource()

assert.NoError(t, err)

d, err := NewBlnk(datasource)

assert.NoError(t, err)

source := gofakeit.UUID()

destination := gofakeit.UUID()

metaDataJSON, _ := json.Marshal(map[string]interface{}{"key": "value"})

t.Run("Transaction not in INFLIGHT status", func(t *testing.T) {

transactionID := gofakeit.UUID()

mock.ExpectQuery(regexp.QuoteMeta(`SELECT transaction_id, source, reference, amount, precise_amount, precision, currency, destination, description, status, created_at, meta_data, parent_transaction, hash FROM blnk.transactions WHERE transaction_id = $1`)).

WithArgs(transactionID).

WillReturnRows(sqlmock.NewRows([]string{"transaction_id", "source", "reference", "amount", "precise_amount", "precision", "currency", "destination", "description", "status", "created_at", "meta_data", "parent_transaction", "hash"}).

AddRow(transactionID, source, gofakeit.UUID(), 100.0, 10000, 100, "USD", destination, gofakeit.UUID(), "APPLIED", time.Now(), metaDataJSON, "", ""))

_, err := d.VoidInflightTransaction(context.Background(), transactionID)

assert.Error(t, err)

assert.Contains(t, err.Error(), "transaction is not in inflight status")

})

t.Run("Transaction already voided", func(t *testing.T) {

transactionID := gofakeit.UUID()

mock.ExpectQuery(regexp.QuoteMeta(`SELECT transaction_id, source, reference, amount, precise_amount, precision, currency, destination, description, status, created_at, meta_data, parent_transaction, hash FROM blnk.transactions WHERE transaction_id = $1`)).

WithArgs(transactionID).

WillReturnRows(sqlmock.NewRows([]string{"transaction_id", "source", "reference", "amount", "precise_amount", "precision", "currency", "destination", "description", "status", "created_at", "meta_data", "parent_transaction", "hash"}).

AddRow(transactionID, source, gofakeit.UUID(), 100.0, 10000, 100, "USD", destination, gofakeit.UUID(), "INFLIGHT", time.Now(), metaDataJSON, "", ""))

// Mock IsParentTransactionVoid

mock.ExpectQuery(regexp.QuoteMeta(`SELECT EXISTS ( SELECT 1 FROM blnk.transactions WHERE parent_transaction = $1 AND status = 'VOID' )`)).

WithArgs(transactionID).

WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(true))

_, err := d.VoidInflightTransaction(context.Background(), transactionID)

assert.Error(t, err)

assert.Contains(t, err.Error(), "transaction has already been voided")

})

// Verify all expectations were met

if err := mock.ExpectationsWereMet(); err != nil {

t.Errorf("there were unfulfilled expectations: %s", err)

}

}

func TestQueueTransactionFlow(t *testing.T) {

// Skip in short mode as this is a long-running test

if testing.Short() {

t.Skip("Skipping queue flow test in short mode")

}

// Initialize test context

ctx := context.Background()

// Setup real test configuration

cnf := &config.Configuration{

Redis: config.RedisConfig{

Dns: "localhost:6379",

},

DataSource: config.DataSourceConfig{

Dns: "postgres://postgres:password@localhost:5432/blnk?sslmode=disable",

},

Queue: config.QueueConfig{

WebhookQueue: "webhook_queue_test",

IndexQueue: "index_queue_test",

TransactionQueue: "transaction_queue_test",

NumberOfQueues: 1,

},

Server: config.ServerConfig{

SecretKey: "test-secret",

},

Transaction: config.TransactionConfig{

BatchSize: 100,

MaxQueueSize: 1000,

LockDuration: time.Second * 30,

IndexQueuePrefix: "test_index",

},

}

config.ConfigStore.Store(cnf)

// Create real datasource for integration test

ds, err := database.NewDataSource(cnf)

require.NoError(t, err, "Failed to create datasource")

// Create real Blnk instance

blnk, err := NewBlnk(ds)

require.NoError(t, err, "Failed to create Blnk instance")

txnRef := "txn_" + model.GenerateUUIDWithSuffix("test")

// Create source balance

sourceBalance := &model.Balance{

Currency: "USD",

LedgerID: "general_ledger_id",

}

// Create destination balance

destBalance := &model.Balance{

Currency: "USD",

LedgerID: "general_ledger_id",

}

// Create balances in database

source, err := ds.CreateBalance(*sourceBalance)

require.NoError(t, err, "Failed to create source balance")

dest, err := ds.CreateBalance(*destBalance)

require.NoError(t, err, "Failed to create destination balance")

sourceID := source.BalanceID

destID := dest.BalanceID

// Create transaction

txn := &model.Transaction{

Reference: txnRef,

Source: sourceID,

Destination: destID,

Amount: 500, // $5.00

Currency: "USD",

AllowOverdraft: true,

Precision: 100,

MetaData: map[string]interface{}{"test": true},

}

// Queue the transaction

queuedTxn, err := blnk.QueueTransaction(ctx, txn)

require.NoError(t, err, "Failed to queue transaction")

// Verify initial transaction state

require.Equal(t, StatusQueued, queuedTxn.Status, "Initial transaction should be QUEUED")

// Store the original transaction ID

originalTxnID := queuedTxn.TransactionID

queueCopy := createQueueCopy(queuedTxn, queuedTxn.Reference)

_, err = blnk.RecordTransaction(ctx, queueCopy)

assert.NoError(t, err)

ref := fmt.Sprintf("%s_%s", txnRef, "q")

// Verify the processed transaction

appliedTxn, err := ds.GetTransactionByRef(ctx, ref)

require.NoError(t, err, "Failed to get processed transactions")

require.NotEmpty(t, appliedTxn, "Should have processed transactions")

// Verify the applied transaction

require.NotNil(t, appliedTxn, "Should have an APPLIED transaction")

require.Equal(t, originalTxnID, appliedTxn.ParentTransaction, "Applied transaction should reference original transaction")

require.Equal(t, txn.Amount, appliedTxn.Amount, "Amount should match")

// Verify final balance states

updatedSource, err := ds.GetBalanceByIDLite(sourceID)

require.NoError(t, err, "Failed to get updated source balance")

updatedDest, err := ds.GetBalanceByIDLite(destID)

require.NoError(t, err, "Failed to get updated destination balance")

fmt.Println("Updated Source Balance: ", updatedSource.Balance)

fmt.Println("Updated Destination Balance: ", updatedDest.Balance)

require.Equal(t, 0, updatedSource.Balance.Cmp(updatedSource.Balance),

"Source balance should be reduced by transaction amount")

require.Equal(t, 0, updatedDest.Balance.Cmp(updatedDest.Balance),

"Destination balance should be increased by transaction amount")

}

func TestQueueTransactionFlowWithSkipQueue(t *testing.T) {

// Skip in short mode

if testing.Short() {

t.Skip("Skipping queue flow test in short mode")

}

ctx := context.Background()

cnf := &config.Configuration{

Redis: config.RedisConfig{

Dns: "localhost:6379",

},

DataSource: config.DataSourceConfig{

Dns: "postgres://postgres:password@localhost:5432/blnk?sslmode=disable",

},

Queue: config.QueueConfig{

WebhookQueue: "webhook_queue_test",

IndexQueue: "index_queue_test",

TransactionQueue: "transaction_queue_test",

NumberOfQueues: 1,

},

Server: config.ServerConfig{

SecretKey: "test-secret",

},

Transaction: config.TransactionConfig{

BatchSize: 100,

MaxQueueSize: 1000,

LockDuration: time.Second * 30,

IndexQueuePrefix: "test_index",

},

}

config.ConfigStore.Store(cnf)

ds, err := database.NewDataSource(cnf)

require.NoError(t, err, "Failed to create datasource")

blnk, err := NewBlnk(ds)

require.NoError(t, err, "Failed to create Blnk instance")

txnRef := "txn_" + model.GenerateUUIDWithSuffix("test")

// Create test balances

sourceBalance := &model.Balance{

Currency: "USD",

LedgerID: "general_ledger_id",

}

destBalance := &model.Balance{

Currency: "USD",

LedgerID: "general_ledger_id",

}

source, err := ds.CreateBalance(*sourceBalance)

require.NoError(t, err, "Failed to create source balance")

dest, err := ds.CreateBalance(*destBalance)

require.NoError(t, err, "Failed to create destination balance")

// Create transaction with skip_queue set to true

txn := &model.Transaction{

Reference: txnRef,

Source: source.BalanceID,

Destination: dest.BalanceID,

Amount: 500,

Currency: "USD",

AllowOverdraft: true,

Precision: 100,

MetaData: map[string]interface{}{"test": true},

SkipQueue: true, // Enable skip queue

}

// Queue the transaction

queuedTxn, err := blnk.QueueTransaction(ctx, txn)

require.NoError(t, err, "Failed to queue transaction")

// Verify that the transaction was processed immediately

require.Equal(t, StatusApplied, queuedTxn.Status, "Transaction should be APPLIED immediately when skip_queue is true")

// Verify balances were updated immediately

updatedSource, err := ds.GetBalanceByIDLite(source.BalanceID)

require.NoError(t, err, "Failed to get updated source balance")

updatedDest, err := ds.GetBalanceByIDLite(dest.BalanceID)

require.NoError(t, err, "Failed to get updated destination balance")

// Calculate expected balance changes

expectedDebit := big.NewInt(int64(-500) * 100) // Amount * precision

expectedCredit := big.NewInt(int64(500) * 100)

// Verify balance changes

require.Equal(t, 0, updatedSource.Balance.Cmp(expectedDebit),

"Source balance should be immediately reduced by transaction amount")

require.Equal(t, 0, updatedDest.Balance.Cmp(expectedCredit),

"Destination balance should be immediately increased by transaction amount")

// Verify no queued entry exists

queuedEntry, err := ds.GetTransactionByRef(ctx, txnRef)

require.NoError(t, err, "Failed to get queued transaction entry")

require.Equal(t, StatusApplied, queuedEntry.Status, "Should have an APPLIED transaction entry")

}

func TestInflightTransactionFlowWithSkipQueue(t *testing.T) {

// Skip in short mode

if testing.Short() {

t.Skip("Skipping queue flow test in short mode")

}

ctx := context.Background()

cnf := &config.Configuration{

Redis: config.RedisConfig{

Dns: "localhost:6379",

},

DataSource: config.DataSourceConfig{

Dns: "postgres://postgres:password@localhost:5432/blnk?sslmode=disable",

},

Queue: config.QueueConfig{

WebhookQueue: "webhook_queue_test",

IndexQueue: "index_queue_test",

TransactionQueue: "transaction_queue_test",

NumberOfQueues: 1,

},

Server: config.ServerConfig{

SecretKey: "test-secret",

},

Transaction: config.TransactionConfig{

BatchSize: 100,

MaxQueueSize: 1000,

LockDuration: time.Second * 30,

IndexQueuePrefix: "test_index",

},

}

config.ConfigStore.Store(cnf)

ds, err := database.NewDataSource(cnf)

require.NoError(t, err, "Failed to create datasource")

blnk, err := NewBlnk(ds)

require.NoError(t, err, "Failed to create Blnk instance")

txnRef := "txn_" + model.GenerateUUIDWithSuffix("test")

// Create test balances

sourceBalance := &model.Balance{

Currency: "USD",

LedgerID: "general_ledger_id",

}

destBalance := &model.Balance{

Currency: "USD",

LedgerID: "general_ledger_id",

}

source, err := ds.CreateBalance(*sourceBalance)

require.NoError(t, err, "Failed to create source balance")

dest, err := ds.CreateBalance(*destBalance)

require.NoError(t, err, "Failed to create destination balance")

// Create transaction with skip_queue set to true

txn := &model.Transaction{

Reference: txnRef,

Source: source.BalanceID,

Destination: dest.BalanceID,

Amount: 500,

Inflight: true,

Currency: "USD",

AllowOverdraft: true,

Precision: 100,

MetaData: map[string]interface{}{"test": true},

SkipQueue: true, // Enable skip queue

}

// Queue the transaction

queuedTxn, err := blnk.QueueTransaction(ctx, txn)

require.NoError(t, err, "Failed to queue transaction")

// Verify that the transaction was processed immediately

require.Equal(t, StatusInflight, queuedTxn.Status, "Transaction should be INFLIGHT immediately when skip_queue is true")

// Verify balances were updated immediately

updatedSource, err := ds.GetBalanceByIDLite(source.BalanceID)

require.NoError(t, err, "Failed to get updated source balance")

updatedDest, err := ds.GetBalanceByIDLite(dest.BalanceID)

require.NoError(t, err, "Failed to get updated destination balance")

// Calculate expected balance changes

expectedDebit := big.NewInt(int64(-500) * 100) // Amount * precision

expectedCredit := big.NewInt(int64(500) * 100)

// Verify balance changes

require.Equal(t, 0, updatedSource.InflightBalance.Cmp(expectedDebit),

"Source balance should be immediately reduced by transaction amount")

require.Equal(t, 0, updatedDest.InflightBalance.Cmp(expectedCredit),

"Destination balance should be immediately increased by transaction amount")

// Verify no queued entry exists

queuedEntry, err := ds.GetTransactionByRef(ctx, txnRef)

require.NoError(t, err, "Failed to get queued transaction entry")

require.Equal(t, StatusInflight, queuedEntry.Status, "Should have an INFLIGHT transaction entry")

}

func TestInflightTransactionFlowWithSkipQueueThenCommit(t *testing.T) {

// Skip in short mode

if testing.Short() {

t.Skip("Skipping queue flow test in short mode")

}

ctx := context.Background()

cnf := &config.Configuration{

Redis: config.RedisConfig{

Dns: "localhost:6379",

},

DataSource: config.DataSourceConfig{

Dns: "postgres://postgres:password@localhost:5432/blnk?sslmode=disable",

},

Queue: config.QueueConfig{

WebhookQueue: "webhook_queue_test",

IndexQueue: "index_queue_test",

TransactionQueue: "transaction_queue_test",

NumberOfQueues: 1,

},

Server: config.ServerConfig{

SecretKey: "test-secret",

},

Transaction: config.TransactionConfig{

BatchSize: 100,

MaxQueueSize: 1000,

LockDuration: time.Second * 30,

IndexQueuePrefix: "test_index",

},

}

config.ConfigStore.Store(cnf)

ds, err := database.NewDataSource(cnf)

require.NoError(t, err, "Failed to create datasource")

blnk, err := NewBlnk(ds)

require.NoError(t, err, "Failed to create Blnk instance")

txnRef := "txn_" + model.GenerateUUIDWithSuffix("test")

// Create test balances

sourceBalance := &model.Balance{

Currency: "USD",

LedgerID: "general_ledger_id",

}

destBalance := &model.Balance{

Currency: "USD",

LedgerID: "general_ledger_id",

}

source, err := ds.CreateBalance(*sourceBalance)

require.NoError(t, err, "Failed to create source balance")

dest, err := ds.CreateBalance(*destBalance)

require.NoError(t, err, "Failed to create destination balance")

// Create transaction with skip_queue set to true

txn := &model.Transaction{

Reference: txnRef,

Source: source.BalanceID,

Destination: dest.BalanceID,

Amount: 500,

Inflight: true,

Currency: "USD",

AllowOverdraft: true,

Precision: 100,

MetaData: map[string]interface{}{"test": true},

SkipQueue: true, // Enable skip queue

}

// Queue the transaction

queuedTxn, err := blnk.QueueTransaction(ctx, txn)

require.NoError(t, err, "Failed to queue transaction")

// Verify that the transaction was processed immediately

require.Equal(t, StatusInflight, queuedTxn.Status, "Transaction should be INFLIGHT immediately when skip_queue is true")

// Verify balances were updated immediately

updatedSource, err := ds.GetBalanceByIDLite(source.BalanceID)

require.NoError(t, err, "Failed to get updated source balance")

updatedDest, err := ds.GetBalanceByIDLite(dest.BalanceID)

require.NoError(t, err, "Failed to get updated destination balance")

// Calculate expected balance changes

expectedDebit := big.NewInt(int64(-500) * 100) // Amount * precision

expectedCredit := big.NewInt(int64(500) * 100)

// Verify balance changes

require.Equal(t, 0, updatedSource.InflightBalance.Cmp(expectedDebit),

"Source balance should be immediately reduced by transaction amount")

require.Equal(t, 0, updatedDest.InflightBalance.Cmp(expectedCredit),

"Destination balance should be immediately increased by transaction amount")