Function startTestAsynqWorker

transaction_test.go:4760–4824

startTestAsynqWorker sets up and starts an Asynq server for testing purposes. It takes the testing object, configuration, LedgerForge instance, and the specific transaction queue name. It returns a cleanup function that should be deferred by the caller to shut down the server.

(t *testing.T, cnf *config.Configuration, ledgerforgeInstance *LedgerForge, transactionQueueName string)

// It takes the testing object, configuration, LedgerForge instance, and the specific transaction queue name.
// It returns a cleanup function that should be deferred by the caller to shut down the server.
func startTestAsynqWorker(t *testing.T, cnf *config.Configuration, ledgerforgeInstance *LedgerForge, transactionQueueName string) func() {
	redisOption, err := redis_db.ParseRedisURL(cnf.Redis.Dns, false)
	require.NoError(t, err, "Failed to parse Redis URL for Asynq")

	queues := make(map[string]int)
	queues[transactionQueueName] = 1 // Concurrency for the transaction queue

	srv := asynq.NewServer(
		asynq.RedisClientOpt{
			Addr:      redisOption.Addr,
			Password:  redisOption.Password,
			DB:        redisOption.DB,
			TLSConfig: redisOption.TLSConfig,
		},
		asynq.Config{
			Concurrency: 1, // Overall server concurrency
			Queues:      queues,
			Logger:      newTestLogger(t),
		},
	)

	mux := asynq.NewServeMux()

	// Define transaction processing handler
	processTransactionHandler := func(ctx context.Context, task *asynq.Task) error {
		var txn model.Transaction
		if err := json.Unmarshal(task.Payload(), &txn); err != nil {
			t.Logf("TEST_WORKER: Error unmarshalling transaction: %v", err)
			return fmt.Errorf("failed to unmarshal transaction: %w", err)
		}

		t.Logf("TEST_WORKER: Picked up transaction %s (Ref: %s) for processing.", txn.TransactionID, txn.Reference)
		processedTxn, err := ledgerforgeInstance.RecordTransaction(ctx, &txn) // Use ledgerforgeInstance from the outer scope
		if err != nil {
			t.Logf("TEST_WORKER: Error recording transaction %s (Ref: %s): %v", txn.TransactionID, txn.Reference, err)
			if strings.Contains(strings.ToLower(err.Error()), "insufficient funds") || strings.Contains(strings.ToLower(err.Error()), "transaction exceeds overdraft limit") {
				_, rejectErr := ledgerforgeInstance.RejectTransaction(ctx, &txn, err.Error())
				if rejectErr != nil {
					t.Logf("TEST_WORKER: Error rejecting transaction %s after processing error: %v", txn.TransactionID, rejectErr)
					return fmt.Errorf("processing error: %v, rejection error: %w", err, rejectErr)
				}
				t.Logf("TEST_WORKER: Rejected transaction %s (Ref: %s) due to: %v", txn.TransactionID, txn.Reference, err)
				return nil // Assuming rejection is a final state for this test handler.
			}
			return err // Allow Asynq to retry for other errors
		}
		t.Logf("TEST_WORKER: Successfully processed transaction %s (Ref: %s), new ID: %s, new Ref: %s, Status: %s", txn.TransactionID, txn.Reference, processedTxn.TransactionID, processedTxn.Reference, processedTxn.Status)
		return nil
	}

	mux.HandleFunc(transactionQueueName, processTransactionHandler)

	go func() {
		t.Logf("TEST_WORKER: Starting Asynq server, listening on queue: %s", transactionQueueName)
		if err := srv.Run(mux); err != nil {
			t.Errorf("TEST_WORKER: Asynq server Run() error: %v", err)
		}
	}()