Function processCSVRowsParallel

internal/files/files.go:252–317

Parallel CSV processing with worker pool

(ctx context.Context, uploadID, source string, csvReader *csv.Reader, columnMap map[string]int, store StoreFunc)

// Parallel CSV processing with worker pool
func processCSVRowsParallel(ctx context.Context, uploadID, source string, csvReader *csv.Reader, columnMap map[string]int, store StoreFunc) (int, error) {
	workerCount := runtime.NumCPU()
	if workerCount > DefaultWorkerCount {
		workerCount = DefaultWorkerCount
	}

	rowChan := make(chan []string, DefaultChannelSize)
	errChan := make(chan error, workerCount)
	var wg sync.WaitGroup
	totalCount := int64(0)
	var countMu sync.Mutex

	// Start workers
	for i := 0; i < workerCount; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			processWorker(ctx, uploadID, source, rowChan, columnMap, store, &totalCount, &countMu, errChan)
		}()
	}

	// Read CSV rows and send to workers
	go func() {
		rowNum := 1
		for {
			record, err := csvReader.Read()
			if err == io.EOF {
				break
			}
			if err != nil {
				errChan <- fmt.Errorf("error reading row %d: %w", rowNum, err)
				continue
			}

			rowNum++

			// Make a copy since csvReader.ReuseRecord is true
			recordCopy := make([]string, len(record))
			copy(recordCopy, record)

			select {
			case <-ctx.Done():
				close(rowChan)
				return
			case rowChan <- recordCopy:
			}
		}
		close(rowChan)
	}()

	// Wait for workers to finish
	wg.Wait()
	close(errChan)

	// Collect errors
	var errs []error
	for err := range errChan {
		errs = append(errs, err)