logfile-spec.js

var bucket = "a-bucket";
var filename = "logs/abc";
var event = {
  "Records": [
    {
      "s3": {
        "object": {
          "key": filename,
        },
        "bucket": {
          "name": bucket,
        }
      }
    }
  ]
};

var aws = require('aws-sdk');
var sinon = require('sinon');
var sandbox;

describe('logfile', () => {
    beforeEach(() => {
        sandbox = sinon.sandbox.create();

        this.s3object = { createReadStream: ()=>{} };
        this.dynamodb = createStubObject(sandbox, ["batchGetItem"]);
        this.s3 = createStubObject(sandbox, ["getObject"])
        this.s3object = createStubObject(sandbox, ["createReadStream"]);

        sandbox.stub(aws, 'S3').returns(this.s3);
        sandbox.stub(aws, 'DynamoDB').returns(this.dynamodb);

        this.s3.getObject.returns(this.s3object);

        delete require.cache[require.resolve('../logfile.js')]
        this.logfile = require('../logfile');
    });

    afterEach(()=>{
        sandbox.restore();
    });

    it('loads from S3', (done) => {
        var context = createStubObject(sandbox, ["succeed", done, "done", done, "fail", done]);
        // assemble
        var log = '91005e2a7c02dcf52e55ab04d8db7d7fa0e3c1e279880f7e705e90040df16d4d xxx-json [02/Dec/2015:02:44:04 +0000] 107.15.84.222 - 42A5D95192F2F0D7 REST.GET.OBJECT C4U6Z2.json "GET /xxx-json/C4U6Z2.json HTTP/1.1" 200 - 34 34 14 13 "-" "HTTPie/0.9.2" -';
        var batchGetItemResponse = { Responses: { slugs: [{slug:{S:'C4U6Z2'},user:{S:'y'}}]}};
        var updateItemResponse = {};

        this.s3object.createReadStream.returns(stringStream("y"));
        this.logfile.handler(event,context);
    });

    it('exits without speaking to db if it finds no slugs', (done) => {
        // assemble
        var context = createStubObject(sandbox, ["succeed", done, "done", done, "fail", done]);
        this.s3object.createReadStream.returns(stringStream('GET /xxx-json/xxxx.logs'));

        // act
        this.logfile.handler(event,context);

        // assert
        expect(this.s3object.createReadStream.called).toBe(true);
        expect(this.dynamodb.batchGetItem.called).toBe(false);
    });
});

logfile.js

'use strict';
var aws = require('aws-sdk'),
    dynamodb = new aws.DynamoDB(),
    s3 = new aws.S3({ apiVersion: '2006-03-01' }),
    StringDecoder = new require('string_decoder').StringDecoder,
    d = new StringDecoder('utf8');

var contextToCallBack = function(context) {
    return function(err,data) {
        debug("contextToCallback, data=", data);
        if (err) { context.fail(err); }
        else { context.succeed(data); }
    };
};
var updateScores = function(slugs, cb) {
    debug("Update Scores:" , slugs);

    var updates = Object.keys(slugs).map(function(slug) {
        return {
            Key: {
                slug: {
                    S: slug
                },
                type: {
                    S: 'static'
                }
            },
            TableName : 'scores',
            ExpressionAttributeNames: {
                '#c': 'score',
                '#u': 'user'
            },
            ExpressionAttributeValues: {
                ':count': { N: slugs[slug].count.toString() },
                ':uzer' : { S: slugs[slug].user }
            },
            UpdateExpression: 'ADD #c :count SET #u = :uzer'
        };
    });

    var errorToReturn = null;
    var results = [];
    var frec = function() {
        debug("> FREC enter");
        if (updates.length > 0) {
            var head = updates.splice(0,1)[0];
            debug("updating item", head);
            dynamodb.updateItem(head, function(err,data) {
                debug("updateItem returned err", err);
                debug("updateItem returned data", data);
                if (err) { errorToReturn = err;}
                results.push(head.Key.slug.S);
                frec();
            });
        } else {
            // i'm done.
            cb(errorToReturn, results);
        }
    };
    frec();
};

/* update 'slugs' with the authors of the slug
 { abc: {count : 7 , user: 'xxx'} }
 */
var loadUsersForSlugs = function(slugs, cb) {
    var params = {
        RequestItems: {
            slugs : {
                Keys : [],
                ProjectionExpression:"slug,#u",
                ExpressionAttributeNames: { "#u": "user" }
            }
        }
    };
    for (var slug in slugs) {
        params.RequestItems.slugs.Keys.push({ "slug" : {"S": slug }});
    }
    dynamodb.batchGetItem(params, function(err, data) {
        if (err) { return cb(err,null); }
        data.Responses.slugs.map(function(result) {
            var slug = result.slug.S;
            var user = result.user.S;
            slugs[slug].user = user;
        });
        cb(err, null);
    });
}

var onSlug = function(slugs, slug) {
    if (typeof(slugs[slug]) === 'undefined') {
        slugs[slug] = { count: 1 };
    } else {
        slugs[slug].count = slugs[slug].count + 1;
    }
};

var onSlugsEnd = (slugs, cb) => {
    if (Object.keys(slugs).length === 0) {
        debug("calling back from onSlugsEnd immediately");
        return cb(null,null);
    }
    debug("Update slugs ", slugs);
    loadUsersForSlugs(slugs, function(err, data) {
        if (err) { return cb(err,null); }
        updateScores(slugs, cb);
    });
};

var quiet = true;
var debug = (msg, obj) => {
    if (!quiet) {
        console.log("DEBUG " + msg, obj);
    }
};

var error = (msg, obj) => {
    console.log("ERROR " + msg, obj);
};
exports.quiet = (b) => { quiet = b; }
exports.handler = function(event, context) {
    debug('Received event:', JSON.stringify(event, null, 2));

    // Get the object from the event and show its content type
    var bucket = event.Records[0].s3.bucket.name;
    var key = decodeURIComponent(event.Records[0].s3.object.key.replace(/\+/g, ' '));
    var params = {
        Bucket: bucket,
        Key: key
    };
    var slugs = {};
    var stream = s3.getObject(params).createReadStream();

    stream.on('data', (data) => {
        debug("DATA", data);
        var chunk = d.write(data);
        var myRe=/GET (?:.+)?\/([^\/]+)\.json/g;
        var tmp;
        while (tmp = myRe.exec(chunk)) {
            onSlug(slugs, tmp[1]);
        }
    })
    .on('error', (err) => {
        error("ERROR", err);
        context.fail(err);
    })
    .on('end', () => {
        debug("END");
        onSlugsEnd(slugs, contextToCallBack(context));
    });
};

spec_helper.js

'use strict';
var sinon = require('sinon'),
    Readable = require('stream').Readable;

beforeAll(function() {
  global.stringStream = (contents) => {
    var s = new Readable();
    s.push(contents);
    s.push(null);
    return s;
  };
});


beforeAll(() => {
    /**
     * creates a stub object like jasmine.createSpyObj
     * but creates it in a sandbox
     */
    var createStubObject = function(sandbox, methods) {
        var x = {};
        for (var i=0;i < methods.length;i++) {
            var m = methods[i];
            var f = sandbox.stub();
            if (methods.length > i+1 && typeof(methods[i+1]) === 'function') {
                var oldf = methods[i+1];
                f = () => { /*console.log("mocked " + m + " called");*/ oldf(); }
                i++;
            }
            x[m] = f;
        };
        return x;
    };

    global.createStubObject = createStubObject;
});