* Functional event scheduler - still some to do, but works

* WIP on message area cleanup via scheduler
* Some const cleanup
This commit is contained in:
Bryan Ashby 2016-06-19 21:09:45 -06:00
parent 3aa87432a1
commit 36ce2354e3
7 changed files with 427 additions and 67 deletions

View File

@ -13,7 +13,7 @@ exports.getModuleAsset = getModuleAsset;
exports.resolveConfigAsset = resolveConfigAsset; exports.resolveConfigAsset = resolveConfigAsset;
exports.getViewPropertyAsset = getViewPropertyAsset; exports.getViewPropertyAsset = getViewPropertyAsset;
var ALL_ASSETS = [ const ALL_ASSETS = [
'art', 'art',
'menu', 'menu',
'method', 'method',

View File

@ -169,6 +169,10 @@ function initialize(cb) {
}, },
function readyMessageNetworkSupport(callback) { function readyMessageNetworkSupport(callback) {
require('./msg_network.js').startup(callback); require('./msg_network.js').startup(callback);
},
function readyEventScheduler(callback) {
const EventSchedulerModule = require('./event_scheduler.js').EventSchedulerModule;
EventSchedulerModule.loadAndStart(callback);
} }
], ],
function onComplete(err) { function onComplete(err) {

View File

@ -211,16 +211,25 @@ function getDefaultConfig() {
archivers : { archivers : {
zip : { zip : {
sig : "504b0304", sig : '504b0304',
offset : 0, offset : 0,
compressCmd : "7z", compressCmd : '7z',
compressArgs : [ "a", "-tzip", "{archivePath}", "{fileList}" ], compressArgs : [ 'a', '-tzip', '{archivePath}', '{fileList}' ],
decompressCmd : "7z", decompressCmd : '7z',
decompressArgs : [ "e", "-o{extractPath}", "{archivePath}" ] decompressArgs : [ 'e', '-o{extractPath}', '{archivePath}' ]
} }
}, },
messageAreaDefaults : {
//
// The following can be override per-area as well
//
maxMessages : 1024, // 0 = unlimited
maxAgeDays : 0, // 0 = unlimited
},
messageConferences : { messageConferences : {
system_internal : { system_internal : {
name : 'System Internal', name : 'System Internal',
desc : 'Built in conference for private messages, bulletins, etc.', desc : 'Built in conference for private messages, bulletins, etc.',
@ -256,6 +265,26 @@ function getDefaultConfig() {
bundleTargetByteSize : 2048000, // 2M, before creating another archive bundleTargetByteSize : 2048000, // 2M, before creating another archive
} }
}, },
eventScheduler : {
events : {
trimMessageAreas : {
// may optionally use [or ]@watch:/path/to/file
//schedule : 'every 24 hours after 3:30 am',
schedule : 'every 1 minutes',
// action:
// - @method:path/to/module.js:theMethodName
// (path is relative to engima base dir)
//
// - @execute:/path/to/something/executable.sh
//
action : '@method:core/message_area.js:trimMessageAreasScheduledEvent', // see method for avail args
}
}
},
misc : { misc : {
idleLogoutSeconds : 60 * 6, // 6m idleLogoutSeconds : 60 * 6, // 6m

202
core/event_scheduler.js Normal file
View File

@ -0,0 +1,202 @@
/* jslint node: true */
'use strict';
// ENiGMA½
const PluginModule = require('./plugin_module.js').PluginModule;
const Config = require('./config.js').config;
const Log = require('./logger.js').log;
const _ = require('lodash');
const later = require('later');
const path = require('path');
exports.getModule = EventSchedulerModule;
exports.EventSchedulerModule = EventSchedulerModule; // allow for loadAndStart
exports.moduleInfo = {
name : 'Event Scheduler',
desc : 'Support for scheduling arbritary events',
author : 'NuSkooler',
};
const SCHEDULE_REGEXP = /(?:^|or )?(@watch\:)([^\0]+)?$/;
const ACTION_REGEXP = /\@(method|execute)\:([^\0]+)?$/;
class ScheduledEvent {
constructor(events, name) {
this.name = name;
this.schedule = this.parseScheduleString(events[name].schedule);
this.action = this.parseActionSpec(events[name].action);
if(this.action) {
this.action.args = events[name].args;
}
}
get isValid() {
if((!this.schedule || (!this.schedule.sched && !this.schedule.watchFile)) || !this.action) {
return false;
}
if('method' === this.action.type && !this.action.location) {
return false;
}
return true;
}
parseScheduleString(schedStr) {
if(!schedStr) {
return false;
}
let schedule = {};
const m = SCHEDULE_REGEXP.exec(schedStr);
if(m) {
schedStr = schedStr.substr(0, m.index).trim();
if('@watch:' === m[1]) {
schedule.watchFile = m[2];
}
}
if(schedStr.length > 0) {
const sched = later.parse.text(schedStr);
if(-1 === sched.error) {
schedule.sched = sched;
}
}
// return undefined if we couldn't parse out anything useful
if(!_.isEmpty(schedule)) {
return schedule;
}
}
parseActionSpec(actionSpec) {
if(actionSpec) {
if('@' === actionSpec[0]) {
const m = ACTION_REGEXP.exec(actionSpec);
if(m) {
if(m[2].indexOf(':') > -1) {
const parts = m[2].split(':');
return {
type : m[1],
location : parts[0],
what : parts[1],
};
} else {
return {
type : m[1],
what : m[2],
};
}
}
} else {
return {
type : 'execute',
what : actionSpec,
};
}
}
}
}
function EventSchedulerModule(options) {
PluginModule.call(this, options);
if(_.has(Config, 'eventScheduler')) {
this.moduleConfig = Config.eventScheduler;
}
const self = this;
this.runningActions = new Set();
this.performAction = function(schedEvent) {
if(self.runningActions.has(schedEvent.name)) {
return; // already running
}
self.runningActions.add(schedEvent.name);
if('method' === schedEvent.action.type) {
const modulePath = path.join(__dirname, '../', schedEvent.action.location); // enigma-bbs base + supplied location (path/file.js')
try {
const methodModule = require(modulePath);
methodModule[schedEvent.action.what](schedEvent.action.args, err => {
if(err) {
Log.debug(
{ error : err.toString(), eventName : schedEvent.name, action : schedEvent.action },
'Error while performing scheduled event action');
}
self.runningActions.delete(schedEvent.name);
});
} catch(e) {
Log.warn(
{ error : e.toString(), eventName : schedEvent.name, action : schedEvent.action },
'Failed to perform scheduled event action');
self.runningActions.delete(schedEvent.name);
}
}
};
}
// convienence static method for direct load + start
EventSchedulerModule.loadAndStart = function(cb) {
const loadModuleEx = require('./module_util.js').loadModuleEx;
const loadOpts = {
name : path.basename(__filename, '.js'),
path : __dirname,
};
loadModuleEx(loadOpts, (err, mod) => {
if(err) {
return cb(err);
}
const modInst = new mod.getModule();
modInst.startup( err => {
return cb(err);
});
});
};
EventSchedulerModule.prototype.startup = function(cb) {
this.eventTimers = [];
const self = this;
if(this.moduleConfig && _.has(this.moduleConfig, 'events')) {
const events = Object.keys(this.moduleConfig.events).map( name => {
return new ScheduledEvent(this.moduleConfig.events, name);
});
events.forEach( schedEvent => {
if(!schedEvent.isValid) {
Log.warn( { eventName : schedEvent.name }, 'Invalid scheduled event entry');
return;
}
if(schedEvent.schedule.sched) {
this.eventTimers.push(later.setInterval( () => {
self.performAction(schedEvent);
}, schedEvent.schedule.sched));
}
// :TODO: handle watchfile -> performAction
});
}
cb(null);
};
EventSchedulerModule.prototype.shutdown = function(cb) {
if(this.eventTimers) {
this.eventTimers.forEach( et => et.clear() );
}
cb(null);
};

View File

@ -27,6 +27,7 @@ exports.getNewMessagesInAreaForUser = getNewMessagesInAreaForUser;
exports.getMessageAreaLastReadId = getMessageAreaLastReadId; exports.getMessageAreaLastReadId = getMessageAreaLastReadId;
exports.updateMessageAreaLastReadId = updateMessageAreaLastReadId; exports.updateMessageAreaLastReadId = updateMessageAreaLastReadId;
exports.persistMessage = persistMessage; exports.persistMessage = persistMessage;
exports.trimMessageAreasScheduledEvent = trimMessageAreasScheduledEvent;
const CONF_AREA_RW_ACS_DEFAULT = 'GM[users]'; const CONF_AREA_RW_ACS_DEFAULT = 'GM[users]';
const AREA_MANAGE_ACS_DEFAULT = 'GM[sysops]'; const AREA_MANAGE_ACS_DEFAULT = 'GM[sysops]';
@ -120,50 +121,50 @@ function getDefaultMessageConferenceTag(client, disableAcsCheck) {
// //
// Note that built in 'system_internal' is always ommited here // Note that built in 'system_internal' is always ommited here
// //
let defaultConf = _.findKey(Config.messageConferences, o => o.default); let defaultConf = _.findKey(Config.messageConferences, o => o.default);
if(defaultConf) { if(defaultConf) {
const acs = Config.messageConferences[defaultConf].acs || CONF_AREA_RW_ACS_DEFAULT; const acs = Config.messageConferences[defaultConf].acs || CONF_AREA_RW_ACS_DEFAULT;
if(true === disableAcsCheck || checkAcs(client, acs)) { if(true === disableAcsCheck || checkAcs(client, acs)) {
return defaultConf; return defaultConf;
} }
} }
// just use anything we can
defaultConf = _.findKey(Config.messageConferences, (o, k) => {
const acs = o.acs || CONF_AREA_RW_ACS_DEFAULT;
return 'system_internal' !== k && (true === disableAcsCheck || checkAcs(client, acs));
});
// just use anything we can return defaultConf;
defaultConf = _.findKey(Config.messageConferences, (o, k) => {
const acs = o.acs || CONF_AREA_RW_ACS_DEFAULT;
return 'system_internal' !== k && (true === disableAcsCheck || checkAcs(client, acs));
});
return defaultConf;
} }
function getDefaultMessageAreaTagByConfTag(client, confTag, disableAcsCheck) { function getDefaultMessageAreaTagByConfTag(client, confTag, disableAcsCheck) {
// //
// Similar to finding the default conference: // Similar to finding the default conference:
// Find the first entry marked 'default', if any. If found, check | client| against // Find the first entry marked 'default', if any. If found, check | client| against
// *read* ACS. If this fails, just find the first one we can that passes checks. // *read* ACS. If this fails, just find the first one we can that passes checks.
// //
// It's possible that we end up with nothing! // It's possible that we end up with nothing!
// //
confTag = confTag || getDefaultMessageConferenceTag(client); confTag = confTag || getDefaultMessageConferenceTag(client);
if(confTag && _.has(Config.messageConferences, [ confTag, 'areas' ])) { if(confTag && _.has(Config.messageConferences, [ confTag, 'areas' ])) {
const areaPool = Config.messageConferences[confTag].areas; const areaPool = Config.messageConferences[confTag].areas;
let defaultArea = _.findKey(areaPool, o => o.default); let defaultArea = _.findKey(areaPool, o => o.default);
if(defaultArea) { if(defaultArea) {
const readAcs = _.has(areaPool, [ defaultArea, 'acs', 'read' ]) ? areaPool[defaultArea].acs.read : AREA_ACS_DEFAULT.read; const readAcs = _.has(areaPool, [ defaultArea, 'acs', 'read' ]) ? areaPool[defaultArea].acs.read : AREA_ACS_DEFAULT.read;
if(true === disableAcsCheck || checkAcs(client, readAcs)) { if(true === disableAcsCheck || checkAcs(client, readAcs)) {
return defaultArea; return defaultArea;
} }
} }
defaultArea = _.findKey(areaPool, (o, k) => { defaultArea = _.findKey(areaPool, (o, k) => {
const readAcs = _.has(areaPool, [ defaultArea, 'acs', 'read' ]) ? areaPool[defaultArea].acs.read : AREA_ACS_DEFAULT.read; const readAcs = _.has(areaPool, [ defaultArea, 'acs', 'read' ]) ? areaPool[defaultArea].acs.read : AREA_ACS_DEFAULT.read;
return (true === disableAcsCheck || checkAcs(client, readAcs)); return (true === disableAcsCheck || checkAcs(client, readAcs));
}); });
return defaultArea; return defaultArea;
} }
} }
function getMessageConferenceByTag(confTag) { function getMessageConferenceByTag(confTag) {
@ -171,26 +172,26 @@ function getMessageConferenceByTag(confTag) {
} }
function getMessageAreaByTag(areaTag, optionalConfTag) { function getMessageAreaByTag(areaTag, optionalConfTag) {
const confs = Config.messageConferences; const confs = Config.messageConferences;
if(_.isString(optionalConfTag)) { if(_.isString(optionalConfTag)) {
if(_.has(confs, [ optionalConfTag, 'areas', areaTag ])) { if(_.has(confs, [ optionalConfTag, 'areas', areaTag ])) {
return confs[optionalConfTag].areas[areaTag]; return confs[optionalConfTag].areas[areaTag];
} }
} else { } else {
// //
// No confTag to work with - we'll have to search through them all // No confTag to work with - we'll have to search through them all
// //
var area; var area;
_.forEach(confs, (v, k) => { _.forEach(confs, (v) => {
if(_.has(v, [ 'areas', areaTag ])) { if(_.has(v, [ 'areas', areaTag ])) {
area = v.areas[areaTag]; area = v.areas[areaTag];
return false; // stop iteration return false; // stop iteration
} }
}); });
return area; return area;
} }
} }
function changeMessageConference(client, confTag, cb) { function changeMessageConference(client, confTag, cb) {
@ -463,4 +464,112 @@ function persistMessage(message, cb) {
], ],
cb cb
); );
}
function trimMessagesToMax(areaTag, maxMessages, archivePath, cb) {
async.waterfall(
[
function getRemoteCount(callback) {
let removeCount = 0;
msgDb.get(
`SELECT COUNT(area_tag) AS msgCount
FROM message
WHERE area_tag = ?`,
[ areaTag ],
(err, row) => {
if(!err) {
if(row.msgCount >= maxMessages) {
removeCount = row.msgCount - maxMessages;
}
}
return callback(err, removeCount);
}
);
},
function trimMessages(removeCount, callback) {
if(0 === removeCount) {
return callback(null);
}
if(archivePath) {
} else {
// just delete 'em
}
}
],
err => {
return cb(err);
}
);
}
// method exposed for event scheduler
function trimMessageAreasScheduledEvent(args, cb) {
//
// Available args:
// - archive:/path/to/archive/dir/
//
let archivePath;
if(args) {
args.forEach(a => {
if(a.startsWith('archive:')) {
archivePath = a.split(':')[1];
}
});
}
//
// Find all area_tag's in message. We don't rely on user configurations
// in case one is no longer available. From there we can trim messages
// that meet the criteria (too old, too many, ...) and optionally archive
// them via moving them to a new DB with the same layout
//
async.waterfall(
[
function getAreaTags(callback) {
let areaTags = [];
msgDb.each(
`SELECT DISTINCT area_tag
FROM message;`,
(err, row) => {
if(err) {
return callback(err);
}
areaTags.push(row.area_tag);
},
err => {
return callback(err, areaTags);
}
);
},
function trimAreas(areaTags, callback) {
areaTags.forEach(areaTag => {
let maxMessages = Config.messageAreaDefaults.maxMessages;
let maxAgeDays = Config.messageAreaDefaults.maxAgeDays;
const area = getMessageAreaByTag(areaTag); // note: we don't know the conf
if(area) {
if(area.maxMessages) {
maxMessages = area.maxMessages;
}
if(area.maxAgeDays) {
maxAgeDays = area.maxAgeDays;
}
}
if(maxMessages) {
trimMessagesToMax(areaTag, maxMessages, archivePath, err => {
});
}
});
}
]
);
console.log('trimming messages from scheduled event') // :TODO: remove me!!!
} }

View File

@ -1140,6 +1140,8 @@ function FTNMessageScanTossModule() {
require('util').inherits(FTNMessageScanTossModule, MessageScanTossModule); require('util').inherits(FTNMessageScanTossModule, MessageScanTossModule);
// :TODO: *scheduled* portion of this stuff should probably use event_scheduler - @immediate would still use record().
FTNMessageScanTossModule.prototype.startup = function(cb) { FTNMessageScanTossModule.prototype.startup = function(cb) {
Log.info(`${exports.moduleInfo.name} Scanner/Tosser starting up`); Log.info(`${exports.moduleInfo.name} Scanner/Tosser starting up`);

View File

@ -939,6 +939,10 @@
value: { command: "DL" } value: { command: "DL" }
action: @menu:doorDarkLands action: @menu:doorDarkLands
} }
{
value: { command: "DP" }
action: @menu:doorParty
}
] ]
} }
@ -1006,6 +1010,16 @@
door: tw door: tw
} }
} }
doorParty: {
desc: Using DoorParty!
module: @systemModule:door_party
config: {
username: XXXXXXXX
password: XXXXXXXX
bbsTag: XX
}
}
/////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////
// Message Area Menu // Message Area Menu
/////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////