<commit_before>var redis = require('./redis');
var queue = require('./queue');
var utility = require('./utility');
var config = require('./config');
var getReplayUrl = require('./getReplayUrl');
var db = require('./db');
var insertMatch = require('./queries').insertMatch;
var buildSets = require('./buildSets');
var bodyParser = require('body-parser');
var express = require('express');
var moment = require('moment');
var app = express();
var port = config.PORT || config.WORK_PORT;
var active_jobs = {};
var pooled_jobs = {};
var startedAt = moment();
var memwatch = require('memwatch-next');
var hd = new memwatch.HeapDiff();
memwatch.on('leak', function(info) {
    console.error(info);
});
memwatch.on('stats', function(stats) {
    var diff = hd.end();
    console.error(JSON.stringify(diff));
    hd = new memwatch.HeapDiff();
    console.error(stats);
});
buildSets(db, redis, function(err) {
    if (err) {
        throw err;
    }
    var pool_size = 100;
    queue.parse.process(pool_size, function(job, cb) {
        //save the callback for this job
        job.cb = cb;
        //put it in the pool
        pooled_jobs[job.jobId] = job;
        console.log('loaded job %s into pool, %s jobs in pool, %s jobs active', job.jobId, Object.keys(pooled_jobs).length, Object.keys(active_jobs).length);
    });
    start();
});

function start() {
    app.use(bodyParser.json({
        limit: '1mb'
    }));
    app.get("/", function(req, res) {
        res.json({
            started_at: startedAt.format(),
            started_ago: startedAt.fromNow()
        });
    });
    app.get('/parse', function(req, res) {
        if (config.RETRIEVER_SECRET && req.query.key !== config.RETRIEVER_SECRET) {
            return res.status(500).json({
                error: "invalid key"
            });
        }
        var ip = req.headers['x-forwarded-for'] || req.connection.remoteAddress;
        console.log('client %s requested work', ip);
        var job = pooled_jobs[Object.keys(pooled_jobs)[0]];
        if (!job) {
            console.log('no work available');
            return res.status(500).json({
                error: "no work available"
            });
        }
        delete pooled_jobs[job.jobId];
        active_jobs[job.jobId] = job;
        var match = job.data.payload;
        job.submitWork = function submitWork(parsed_data) {
            if (parsed_data.error) {
                return job.exit(parsed_data.error);
            }
            delete parsed_data.key;
            delete parsed_data.jobId;
            //extend match object with parsed data, keep existing data if key conflict (match_id)
            //match.players was deleted earlier during insertion of api data
            for (var key in parsed_data) {
                match[key] = match[key] || parsed_data[key];
            }
            match.parse_status = 2;
            return insertMatch(db, redis, queue, match, {
                type: "parsed"
            }, job.exit);
        };
        job.exit = function exit(err) {
            clearTimeout(job.expire);
            delete active_jobs[job.jobId];
            return job.cb(err);
        };
        //get the replay url and save it
        return getReplayUrl(db, redis, match, function(err) {
            if (err) {
                //server won't send back a job in this case, let client timeout
                return job.exit(err);
            }
            job.expire = setTimeout(function() {
                console.log('job %s expired', job.jobId);
                return job.exit("timeout");
            }, 180 * 1000);
            console.log('server sent jobid %s', job.jobId);
            return res.json({
                jobId: job.jobId,
                data: job.data
            });
        });
    });
    app.post('/parse', function(req, res) {
        //validate request
        if (config.RETRIEVER_SECRET && req.body.key !== config.RETRIEVER_SECRET) {
            return res.status(500).json({
                error: "invalid key"
            });
        }
        console.log('received submitted work');
        if (active_jobs[req.body.jobId]) {
            var job = active_jobs[req.body.jobId];
            job.submitWork(req.body);
            return res.json({
                error: null
            });
        }
        else {
            return res.status(500).json({
                error: "no active job with this jobid"
            });
        }
    });
    var server = app.listen(port, function() {
        var host = server.address().address;
        console.log('[WORKSERVER] listening at http://%s:%s', host, port);
    });<commit_msg>send back failure to get replay url<commit_after>var redis = require('./redis');
var queue = require('./queue');
var utility = require('./utility');
var config = require('./config');
var getReplayUrl = require('./getReplayUrl');
var db = require('./db');
var insertMatch = require('./queries').insertMatch;
var buildSets = require('./buildSets');
var bodyParser = require('body-parser');
var express = require('express');
var moment = require('moment');
var app = express();
var port = config.PORT || config.WORK_PORT;
var active_jobs = {};
var pooled_jobs = {};
var startedAt = moment();
var memwatch = require('memwatch-next');
var hd = new memwatch.HeapDiff();
memwatch.on('leak', function(info) {
    console.error(info);
});
memwatch.on('stats', function(stats) {
    var diff = hd.end();
    console.error(JSON.stringify(diff));
    hd = new memwatch.HeapDiff();
    console.error(stats);
});
buildSets(db, redis, function(err) {
    if (err) {
        throw err;
    }
    var pool_size = 100;
    queue.parse.process(pool_size, function(job, cb) {
        //save the callback for this job
        job.cb = cb;
        //put it in the pool
        pooled_jobs[job.jobId] = job;
        console.log('loaded job %s into pool, %s jobs in pool, %s jobs active', job.jobId, Object.keys(pooled_jobs).length, Object.keys(active_jobs).length);
    });
    start();
});

function start() {
    app.use(bodyParser.json({
        limit: '1mb'
    }));
    app.get("/", function(req, res) {
        res.json({
            started_at: startedAt.format(),
            started_ago: startedAt.fromNow()
        });
    });
    app.get('/parse', function(req, res) {
        if (config.RETRIEVER_SECRET && req.query.key !== config.RETRIEVER_SECRET) {
            return res.status(500).json({
                error: "invalid key"
            });
        }
        var ip = req.headers['x-forwarded-for'] || req.connection.remoteAddress;
        console.log('client %s requested work', ip);
        var job = pooled_jobs[Object.keys(pooled_jobs)[0]];
        if (!job) {
            console.log('no work available');
            return res.status(500).json({
                error: "no work available"
            });
        }
        delete pooled_jobs[job.jobId];
        active_jobs[job.jobId] = job;
        var match = job.data.payload;
        job.submitWork = function submitWork(parsed_data) {
            if (parsed_data.error) {
                return job.exit(parsed_data.error);
            }
            delete parsed_data.key;
            delete parsed_data.jobId;
            //extend match object with parsed data, keep existing data if key conflict (match_id)
            //match.players was deleted earlier during insertion of api data
            for (var key in parsed_data) {
                match[key] = match[key] || parsed_data[key];
            }
            match.parse_status = 2;
            return insertMatch(db, redis, queue, match, {
                type: "parsed"
            }, job.exit);
        };
        job.exit = function exit(err) {
            clearTimeout(job.expire);
            delete active_jobs[job.jobId];
            return job.cb(err);
        };
        //get the replay url and save it
        return getReplayUrl(db, redis, match, function(err) {
            if (err) {
                res.status(500).json({
                    error: "failed to get replay url"
                });
                return job.exit(err);
            }
            else {
                job.expire = setTimeout(function() {
                    console.log('job %s expired', job.jobId);
                    return job.exit("timeout");
                }, 180 * 1000);
                console.log('server sent jobid %s', job.jobId);
                return res.json({
                    jobId: job.jobId,
                    data: job.data
                });
            }
        });
    });
    app.post('/parse', function(req, res) {
        //validate request
        if (config.RETRIEVER_SECRET && req.body.key !== config.RETRIEVER_SECRET) {
            return res.status(500).json({
                error: "invalid key"
            });
        }
        console.log('received submitted work');
        if (active_jobs[req.body.jobId]) {
            var job = active_jobs[req.body.jobId];
            job.submitWork(req.body);
            return res.json({
                error: null
            });
        }
        else {
            return res.status(500).json({
                error: "no active job with this jobid"
            });
        }
    });
    var server = app.listen(port, function() {
        var host = server.address().address;
        console.log('[WORKSERVER] listening at http://%s:%s', host, port);
    });
{'fraction_non_alphanumeric': 0.13875292615450097, 'fraction_numerical': 0.0056394977654820174, 'mean_word_length': 1.881361128142244, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7339995', 'n_tokens_mistral': 2783, 'n_tokens_neox': 2682, 'n_words': 650}