_usage.md

Usage

$ mongoexport -d test -c activities | node event-convert-stream.js | mongoimport -d test -c events

event-convert-stream.js

var Transform = require('stream').Transform
  , util = require('util')
  , split = require('split')

util.inherits(ActivityToEventStream, Transform)

function ActivityToEventStream() {
  Transform.call(this, {
    objectMode: true
  })
}

ActivityToEventStream.prototype._write = function(obj, encoding, done) {
  var e

  // Skip non 'EPISODE WATCHED' events
  if (obj.activityType !== 'EPISODE WATCHED')
    return done()
  
  e = {
    _id: obj._id,
    _type: 'ViewedEpisode',
    app: obj.a,
    showId: obj.sourceId,
    timestamp: obj.timestamp || {$date: Date.now()}
  }

  // Set the userId or anonId
  if (obj.a === '468')
    e.anonId = obj.userId
  else if (obj.userId.$oid)
    e.userId = obj.userId.$oid
  else
    e.userId = obj.userId

  // Set the episodeId if provided
  if (obj.secondarySourceId) e.episodeId = obj.secondarySourceId
  
  this.push(JSON.stringify(e) + '\n')
  done()
}

process.stdin
  .pipe(split(JSON.parse))
  .pipe(new ActivityToEventStream())
  .pipe(process.stdout)