refactor: statement: optimise for huge statements (#118)

Author: npv12

server/internal/api/controller/statement_controller_test.go

@@ -42,15 +42,15 @@ func createAccount(testHelper *TestHelper, name string, balance float64) float64
 func waitForStatementDone(testHelper *TestHelper, statementId float64) map[string]any {
 	var status string
 	var data map[string]any
-	for i := 0; i < 20; i++ {
+	for i := 0; i < 8; i++ {
 		resp, response := testHelper.MakeRequest(http.MethodGet, "/statement/"+strconv.FormatFloat(statementId, 'f', 0, 64), nil)
 		Expect(resp.StatusCode).To(Equal(http.StatusOK))
 		data = response["data"].(map[string]any)
 		status = data["status"].(string)
-		if status == "done" {
+		if status != "processing" {
 			break
 		}
-		time.Sleep(1 * time.Second)
+		time.Sleep(5 * time.Second)
 	}
 	Expect(status).To(Equal("done"))
 	return data
@@ -835,15 +835,15 @@ var _ = Describe("StatementController", func() {
 			Expect(transactions).To(HaveLen(3))
 		})
 
-		It("should parse statement with 10000 transactions", func() {
+		It("should parse statement with 100K transactions", func() {
 			testHelper := createUniqueUser(baseURL)
 			accountId := createAccount(testHelper, "Account 1", 1000.0)
 
-			// Build XLSX data (header + 10000 rows)
+			// Build XLSX data (header + 100K rows)
 			data := [][]string{
 				{"Txn Date", "Details", "Ref No.", "Debit", "Credit", "Balance"},
 			}
-			for i := 1; i <= 1000; i++ {
+			for i := 1; i <= 100000; i++ {
 				row := []string{
 					"22 Aug 2022",
 					fmt.Sprintf("Desc%d", i),
@@ -877,7 +877,7 @@ var _ = Describe("StatementController", func() {
 			transactions := txData["transactions"].([]any)
 			Expect(len(transactions)).To(Equal(10))
 			Expect(txData).To(HaveKey("total"))
-			Expect(txData["total"]).To(Equal(1000.0))
+			Expect(txData["total"]).To(Equal(100000.0))
 		})
 
 		It("should error when accountId doesn't exist", func() {

server/internal/mock/repository/statement_repository.go

@@ -61,6 +61,19 @@ func (m *MockStatementRepository) CreateStatementTxn(ctx context.Context, statem
 	return nil
 }
 
+// CreateStatementTxns adds multiple mappings between statement and transactions for testing
+func (m *MockStatementRepository) CreateStatementTxns(ctx context.Context, statementId int64, transactionIds []int64) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for _, txId := range transactionIds {
+		m.statementTxnMappings = append(m.statementTxnMappings, statementTxnMapping{
+			StatementId:   statementId,
+			TransactionId: txId,
+		})
+	}
+	return nil
+}
+
 func (m *MockStatementRepository) UpdateStatementStatus(ctx context.Context, statementId int64, input models.UpdateStatementStatusInput) (models.StatementResponse, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()

server/internal/mock/repository/transaction_repository.go

@@ -76,6 +76,62 @@ func (m *MockTransactionRepository) CreateTransaction(ctx context.Context, input
 	return tx, nil
 }
 
+func (m *MockTransactionRepository) CreateTransactions(ctx context.Context, inputs []models.CreateBaseTransactionInput, categoryIds [][]int64) ([]models.TransactionResponse, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if len(inputs) != len(categoryIds) {
+		return nil, customErrors.NewTransactionAlreadyExistsError(nil)
+	}
+
+	results := make([]models.TransactionResponse, 0, len(inputs))
+
+	for i, input := range inputs {
+		// Check for duplicate transaction
+		for _, tx := range m.transactions {
+			if tx.CreatedBy == input.CreatedBy &&
+				tx.Date.Format("2006-01-02") == input.Date.Format("2006-01-02") &&
+				tx.Name == input.Name &&
+				tx.Amount == *input.Amount {
+				existingDesc := ""
+				if tx.Description != nil {
+					existingDesc = *tx.Description
+				}
+				inputDesc := input.Description
+				if existingDesc == inputDesc {
+					return nil, customErrors.NewTransactionAlreadyExistsError(nil)
+				}
+			}
+		}
+
+		// Create new transaction
+		newId := m.nextId
+		m.nextId++
+
+		baseTx := models.TransactionBaseResponse{
+			Id:          newId,
+			Name:        input.Name,
+			Description: &input.Description,
+			Amount:      *input.Amount,
+			Date:        input.Date,
+			CreatedBy:   input.CreatedBy,
+			AccountId:   input.AccountId,
+		}
+
+		tx := models.TransactionResponse{
+			TransactionBaseResponse: baseTx,
+			CategoryIds:             categoryIds[i],
+		}
+
+		m.transactions[newId] = tx
+		m.categoryMap[newId] = categoryIds[i]
+
+		results = append(results, tx)
+	}
+
+	return results, nil
+}
+
 func (m *MockTransactionRepository) UpdateCategoryMapping(ctx context.Context, transactionId int64, userId int64, categoryIds []int64) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -99,6 +155,20 @@ func (m *MockTransactionRepository) GetTransactionById(ctx context.Context, tran
 	return tx, nil
 }
 
+func (m *MockTransactionRepository) GetTransactionsByIds(ctx context.Context, transactionIds []int64, userId int64) ([]models.TransactionResponse, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	var transactions []models.TransactionResponse
+	for _, id := range transactionIds {
+		tx, ok := m.transactions[id]
+		if ok && tx.CreatedBy == userId {
+			transactions = append(transactions, tx)
+		}
+	}
+	return transactions, nil
+}
+
 func (m *MockTransactionRepository) UpdateTransaction(ctx context.Context, transactionId int64, userId int64, input models.UpdateBaseTransactionInput) error {
 	tx, ok := m.transactions[transactionId]
 	if !ok || tx.CreatedBy != userId {

server/internal/repository/statement_repository.go

@@ -18,6 +18,7 @@ import (
 type StatementRepositoryInterface interface {
 	CreateStatement(ctx context.Context, input models.CreateStatementInput) (models.StatementResponse, error)
 	CreateStatementTxn(ctx context.Context, statementId int64, transactionId int64) error
+	CreateStatementTxns(ctx context.Context, statementId int64, transactionIds []int64) error
 	UpdateStatementStatus(ctx context.Context, statementId int64, input models.UpdateStatementStatusInput) (models.StatementResponse, error)
 	GetStatementByID(ctx context.Context, statementId int64, userId int64) (models.StatementResponse, error)
 	ListStatementByUserId(ctx context.Context, userId int64, limit, offset int) ([]models.StatementResponse, error)
@@ -65,6 +66,45 @@ func (r *StatementRepository) CreateStatementTxn(ctx context.Context, statementI
 	return nil
 }
 
+func (r *StatementRepository) CreateStatementTxns(ctx context.Context, statementId int64, transactionIds []int64) error {
+	if len(transactionIds) == 0 {
+		return nil
+	}
+
+	const batchSize = 1000
+
+	// Process in batches of 1000
+	for batchStart := 0; batchStart < len(transactionIds); batchStart += batchSize {
+		batchEnd := batchStart + batchSize
+		if batchEnd > len(transactionIds) {
+			batchEnd = len(transactionIds)
+		}
+
+		batchTxIds := transactionIds[batchStart:batchEnd]
+
+		// Build bulk insert using VALUES for this batch
+		placeholders := make([]string, 0, len(batchTxIds))
+		args := make([]interface{}, 0, len(batchTxIds)*2)
+		argIndex := 1
+
+		for _, txID := range batchTxIds {
+			placeholders = append(placeholders, fmt.Sprintf("($%d, $%d)", argIndex, argIndex+1))
+			args = append(args, statementId, txID)
+			argIndex += 2
+		}
+
+		query := fmt.Sprintf(`INSERT INTO %s.%s (statement_id, transaction_id) VALUES %s`,
+			r.schema, r.mappingTableName, strings.Join(placeholders, ", "))
+
+		_, err := r.db.ExecuteQuery(ctx, query, args...)
+		if err != nil {
+			return statementErrors.NewStatementCreateError(err)
+		}
+	}
+
+	return nil
+}
+
 func (r *StatementRepository) UpdateStatementStatus(ctx context.Context, statementId int64, input models.UpdateStatementStatusInput) (models.StatementResponse, error) {
 	logger.Debugf("Updating statement %d with status %s", statementId, input.Status)
 	fieldsClause, argValues, argIndex, err := helper.CreateUpdateParams(&input)