nostreams.js

var config = require('../config');
var db = require('./db')(config);

function readUsers(callback) {
	db.users.find({unsubscribed: {$exists: false}, firstTimeUser: {$exists: false}}).forEach(function (err, user) {
		if (err) {
			throw callback(err);
		}

		if (!user) {
			return callback();
		}

		console.log(toCsv(user));
	});

	function toCsv(user) {
		return user.email + ',' + user.name + ',' + (user.displayName || '');
	}
}

readUsers(function (err) {
	if (err) {
		console.error(err);
	}

	db.close();
});

streams.js

var through = require('through');

var config = require('../config');
var db = require('./db')(config);

function readUsers(callback) {
	var csv = through(function (user) {
		var line = user.email + ',' + user.name + ',' + (user.displayName || '') + '\n';
		this.queue(line);
	});

	var stream = db.users
		.find({unsubscribed: {$exists: false}, firstTimeUser: {$exists: false}})
		.sort({_id: -1});

	stream
		.pipe(csv)
		.pipe(process.stdout);

	stream.on('error', callback);
	stream.on('end', callback);
}

readUsers(function (err) {
	if (err) {
		console.error(err);
	}

	db.close();
});