Function processWorker

internal/files/files.go:319–370

(ctx context.Context, uploadID, source string, rowChan <-chan []string, columnMap map[string]int, store StoreFunc, totalCount *int64, countMu *sync.Mutex, errChan chan<- error)

}

func processWorker(ctx context.Context, uploadID, source string, rowChan <-chan []string, columnMap map[string]int, store StoreFunc, totalCount *int64, countMu *sync.Mutex, errChan chan<- error) {
	batch := make([]model.ExternalTransaction, 0, DefaultBatchSize)
	processCount := 0

	for record := range rowChan {
		select {
		case <-ctx.Done():
			return
		default:
		}

		externalTxn, err := parseExternalTransaction(record, columnMap, source)
		if err != nil {
			errChan <- err
			continue
		}

		batch = append(batch, externalTxn)

		// Process batch when full
		if len(batch) >= DefaultBatchSize {
			if err := storeBatch(ctx, uploadID, batch, store); err != nil {
				errChan <- err
			} else {
				countMu.Lock()
				*totalCount += int64(len(batch))
				countMu.Unlock()
			}
			batch = batch[:0] // Reset batch
		}

		processCount++
		if processCount%ContextCheckInterval == 0 {
			select {
			case <-ctx.Done():
				return
			default:
			}
		}
	}

	// Process remaining items in batch
	if len(batch) > 0 {
		if err := storeBatch(ctx, uploadID, batch, store); err != nil {
			errChan <- err
		} else {
			countMu.Lock()
			*totalCount += int64(len(batch))
			countMu.Unlock()
		}
	}
}

func storeBatch(ctx context.Context, uploadID string, batch []model.ExternalTransaction, store StoreFunc) error {
	for _, txn := range batch {