1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
#!/usr/bin/env node
/* jshint esnext: true */
"use strict";
const Promise = require("bluebird"),
Service = require("./service_skeleton.js");
const logger = global.logger,
CRUNCH_QUEUE = process.env.CRUNCH_QUEUE || "crunch",
CRUNCH_TOURNAMENT_QUEUE = process.env.CRUNCH_TOURNAMENT_QUEUE || "crunch_tournament",
SHOVEL_SIZE = parseInt(process.env.SHOVEL_SIZE) || 1000;
module.exports = class Cruncher extends Service {
constructor() {
super();
this.setTargets({
"regular": CRUNCH_QUEUE,
"tournament": CRUNCH_TOURNAMENT_QUEUE
});
this.setRoutes({
// crunch global meta
"/api/crunch/:category*?": async (req, res) => {
this.crunchGlobal(req.params.category || "regular");
res.sendStatus(204);
},
// crunch a player
"/api/player/:name/crunch/:category*?": async (req, res) => {
const category = req.params.category || "regular",
db = this.getDatabase(req.params.category || "regular"),
players = await db.Player.findAll({ where: { name: req.params.name } });
if (players == undefined) {
logger.error("player not found in db, won't crunch",
{ name: req.params.name });
res.sendStatus(404);
return;
}
logger.info("player in db, crunching", { name: req.params.name });
players.forEach((player) =>
this.crunchPlayer(category, player.api_id)); // fire away
res.sendStatus(204);
},
// crunch a team
"/api/team/:id/crunch": async (req, res) => {
const db = this.getDatabase("regular"),
team = await db.Team.findOne({ where: { id: req.params.id } });
if (team == undefined) {
logger.error("team not found in db, won't crunch",
{ name: req.params.id });
res.sendStatus(404);
return;
}
logger.info("team in db, crunching", { name: team.id });
crunchTeam(team.id); // fire away
res.sendStatus(204);
}
});
}
// upcrunch player's stats
async crunchPlayer(category, api_id) {
const db = this.getDatabase(category),
where = { player_api_id: api_id },
last_crunch = await db.PlayerPoint.findOne({
attributes: ["updated_at"],
where,
order: [ ["updated_at", "DESC"] ]
});
if (last_crunch) where.created_at = { $gt: last_crunch.updated_at };
// get all participants for this player
const participations = await db.Participant.findAll({
attributes: ["api_id"],
where
});
// send everything to cruncher
logger.info("sending participations to cruncher",
{ length: participations.length });
await Promise.map(participations, async (p) =>
await this.forward(this.getQueue(category),
p.api_id, { persistent: true, type: "player" }));
// jobs with the type "player" won't be taken into account for global stats
// global stats would increase on every player refresh otherwise
}
// reset fame and crunch
// TODO: incremental crunch possible?
async crunchTeam(team_id) {
await this.forward(CRUNCH_QUEUE, team_id,
{ persistent: true, type: "team" });
}
// crunch global stats
async crunchGlobal(category) {
const db = this.getDatabase(category);
// get lcpid from keys table
let last_crunch_participant_id = await this.getKey(category,
"global_last_crunch_participant_id", 0);
// don't load the whole Participant table at once into memory
let participations;
logger.info("loading all participations into cruncher",
{ last_crunch_participant_id: last_crunch_participant_id });
do {
participations = await db.Participant.findAll({
attributes: ["api_id", "id"],
where: {
id: { $gt: last_crunch_participant_id }
},
limit: SHOVEL_SIZE,
order: [ ["id", "ASC"] ]
});
await Promise.map(participations, async (p) =>
await this.forward(this.getQueue(category), p.api_id,
{ persistent: true, type: "global" }));
// update lpcid & refetch
if (participations.length > 0) {
last_crunch_participant_id = participations[participations.length-1].id;
await this.setKey(category, "global_last_crunch_participant_id",
last_crunch_participant_id);
}
logger.info("loading more participations into cruncher", {
limit: SHOVEL_SIZE,
size: participations.length,
last_crunch_participant_id: last_crunch_participant_id
});
} while (participations.length == SHOVEL_SIZE);
logger.info("done loading participations into cruncher");
}
}
|
