Runs a lambda function (gateway) on every resource in a S3 bucket.

lambdaBucket.js

var AWS = require("aws-sdk");
var async = require("async");
var request = require("request");
var s3 = new AWS.S3();

// max number of concurrent requests
var CONCURRENT = 90;
var first = true;
// optional start key
var last = process.env.START;

function lambda(bucket, content, callback) {
	request.post({
		url: process.env.LAMBDA_URL,
		headers: {
			// protect your lambda gateway with an api key
			"x-api-key": process.env.API_KEY
		},
		json: true,
		body: {
			Records: [{
				s3: {
					bucket: { name: bucket },
					object: { key: content.Key }
				}
			}]
		}
	}, callback);
}

function list(bucket, continuationToken, callback) {
	var params = {
		Bucket: bucket,
		EncodingType: "url"
	};
	if (continuationToken) {
		params.ContinuationToken = continuationToken;
	}
	if (last && first) {
		first = false;
		params.StartAfter = last;
	}
	s3.listObjectsV2(params, function (err, data) {
		if (err) {
			return callback(err);
		}
		async.parallelLimit(
			data.Contents.map(function (content) {
				return lambda.bind(null, bucket, content);
			}),
			CONCURRENT,
			function (err, res) {
				if (err) {
					return callback(err);
				}
				if (data.Contents.length) {
					last = data.Contents[data.Contents.length - 1].Key;
					console.log(last);
				}
				if (data.NextContinuationToken) {
					return list(bucket, data.NextContinuationToken, callback);
				}
				return callback(null);
			}
		);
	});
}

function work() {
	list(process.env.BUCKET, null, function (err, res) {
		if (err) {
			// Frequent ENOTFOUND errors while resolving DNS for lambda url, ignore and continue
			if (err.code === "ENOTFOUND") {
				console.log("ENOTFOUND");
				first = true;
				return work();
			}
			return console.error(err);
		}
		console.log("DONE");
	});
}

work();