Streaming CSV import with batched inserts

                  package importers

import (
  "context"
  "encoding/csv"
  "io"

  "github.com/jackc/pgx/v5/pgxpool"
)

type Row struct {
  Email string
  Name  string
}

func ImportUsers(ctx context.Context, db *pgxpool.Pool, r io.Reader) error {
  cr := csv.NewReader(r)
  batch := make([]Row, 0, 500)

  flush := func() error {
    if len(batch) == 0 {
      return nil
    }
    b := &pgxpool.Batch{}
    for _, row := range batch {
      b.Queue(`INSERT INTO users(email, name) VALUES($1, $2) ON CONFLICT (email) DO UPDATE SET name=EXCLUDED.name`, row.Email, row.Name)
    }
    br := db.SendBatch(ctx, b)
    defer br.Close()
    if _, err := br.Exec(); err != nil {
      return err
    }
    batch = batch[:0]
    return nil
  }

  for {
    rec, err := cr.Read()
    if err == io.EOF {
      return flush()
    }
    if err != nil {
      return err
    }
    if len(rec) < 2 {
      continue
    }
    batch = append(batch, Row{Email: rec[0], Name: rec[1]})
    if len(batch) >= 500 {
      if err := flush(); err != nil {
        return err
      }
    }
  }
}