Update / Remove users in Customer.io from CSV files using Node.js

about.md

Updating Customer.io using CSVs

Ever wanted to update a Customer.io environment using a CSV that you have, maybe acquired from an event, or exported?

Now you can.

Usage

You can obtain your Customer.io credentials on the integration page.

export SITE_KEY=YOUR_CUSTOMERIO_SITE_KEY
export API_KEY=YOUR_CUSTOMERIO_API_KEY
export CSV_FILE=customers.csv

node process.js

Note: By default, the script will delete users that are inside of the csv file, to change this, continue reading!

Variables

name	description
SITE_KEY	Customer.io API Integration Site Key
API_Key	Customer.io API Integration API Key
CSV_FILE	Relative or Absolute path to CSV file you want to process
ITEMS_PER_RUN	Number of items to process per iteration (default 100)

Methods

• removeUser(id) - Delete user from Customer.io
• updateUser(details) - Create/Update user on Customer.io, you can remove attributes by using an empty string.
  • See update.queue-method.js

Custom Methods

You can define a custom method in the section labelled Customer.io Queue Methods.

Once you have defined your method you can set the script to call it from inside of the onItem method by passing it as the second argument to the addToQueue method.

Useful Resources:

• Needle Documentation
• Customer.io API

License

MIT - 2016 - Nijiko Yonskai

process.js

const csv = require('csv-parser')
const fs = require('fs')
const n = require('needle')

// Customer.io Integration / Authentication settings
let options = {
  username: process.env.SITE_KEY || 'INSERT_CUSTOMERIO_SITE_KEY_HERE',
  password: process.env.API_KEY || 'INSERT_CUSTOMERIO_API_KEY_HERE'
}

// File to be parsed and iterated over
const CSV_FILE = process.env.CSV_FILE || 'customers.csv'

// Number of items to process per queue iteration
const ITEMS_PER_RUN = process.env.ITEMS_PER_RUN || 100

/*****************************************************************************************************************************************
 * CSV Item Iterator
 *****************************************************************************************************************************************
 * Each entry in the CSV is passed to this method as an Object named `item` each field is accessible like so:
 *
 *     item.<field_name>
 *
 * The `method` to be ran when the queue item is processed is passed as the second variable, the default is `removeUser`
 *****************************************************************************************************************************************/

function onItem (item) {
  let method = removeUser
  let data = null

  data = item.customer_id ? item.customer_id : item.id

  addToQueue(data, method)
}

/*****************************************************************************************************************************************
 * Customer.io Queue Methods
 *****************************************************************************************************************************************
 * Add functions here to be processed for each item added to the queue.
 *
 * To modifiy the queue entry data, you must modify the `onItem` method above
 *****************************************************************************************************************************************/

function removeUser (id) {
  return new Promise((resolve, reject) => {
    n.delete(`https://track.customer.io/api/v1/customers/${id}`, null, options, (err, res) => resolve(err || res.statusCode))
  })
}

/*****************************************************************************************************************************************
 * Queue Handler / Processing
 *****************************************************************************************************************************************
 * The two functions here handle processing of queue data, modify at your own risk
 *****************************************************************************************************************************************/

let queue = []
let running = false

function addToQueue (id, method) {
  queue.push([id, method])
  processQueue()
}

function processQueue () {
  if (running) return
  if (!queue.length) return console.log('empty queue')

  let ids = queue.splice(0, queue.length > ITEMS_PER_RUN ? ITEMS_PER_RUN : queue.length)
  running = true

  Promise.all(ids.map(d => d[1](d[0]))).then(results => {
    running = false
    console.log(results)
    processQueue()
  })
}

/*****************************************************************************************************************************************
 * CSV Processing
 *****************************************************************************************************************************************
 * Start processing CSV file
 *****************************************************************************************************************************************/

fs.createReadStream(CSV_FILE).pipe(csv()).on('data', onItem)

update.queue-method.js

// updateUser({ id: String, data: Object })
//
// Example for `addToQueue`
//
//     addToQueue({ id: String, data: Object }, updateUser)
//
function updateUser (details) {
  return new Promise((resolve, reject) => {
    n.put(`https://track.customer.io/api/v1/customers/${details.id}`, details.data, options, (err, res) => {
      resolve(err || res.statusCode)
    })
  })
}