2016-06-21 05:57:54 +08:00
|
|
|
const DatabaseConnectionFactory = require(`${__base}/core/database-connection-factory`)
|
2016-06-22 05:58:20 +08:00
|
|
|
const {processors} = require('./processors')
|
2016-06-21 05:57:54 +08:00
|
|
|
|
2016-06-23 01:59:22 +08:00
|
|
|
// List of the attributes of Message that the processor should be allowed to change.
|
2016-06-22 05:58:20 +08:00
|
|
|
// The message may move between folders, get starred, etc. while it's being
|
|
|
|
// processed, and it shouldn't overwrite changes to those fields.
|
|
|
|
const MessageAttributes = ['body', 'processed']
|
|
|
|
const MessageProcessorVersion = 1;
|
2016-06-21 05:57:54 +08:00
|
|
|
|
2016-06-23 01:59:22 +08:00
|
|
|
|
|
|
|
function runPipeline(accountId, message) {
|
|
|
|
return processors.reduce((prevPromise, processor) => (
|
|
|
|
prevPromise.then((msg) => processor({message: msg, accountId}))
|
|
|
|
), Promise.resolve(message))
|
|
|
|
}
|
|
|
|
|
|
|
|
function saveMessage(message) {
|
|
|
|
message.processed = MessageProcessorVersion;
|
|
|
|
return message.save({
|
|
|
|
fields: MessageAttributes,
|
|
|
|
});
|
2016-06-21 05:57:54 +08:00
|
|
|
}
|
|
|
|
|
2016-06-22 05:58:20 +08:00
|
|
|
function processMessage({messageId, accountId}) {
|
2016-06-23 01:59:22 +08:00
|
|
|
DatabaseConnectionFactory.forAccount(accountId)
|
|
|
|
.then(({Message}) =>
|
|
|
|
Message.find({where: {id: messageId}}).then((message) =>
|
|
|
|
runPipeline(accountId, message)
|
|
|
|
.then((processedMessage) => saveMessage(processedMessage))
|
2016-06-22 05:58:20 +08:00
|
|
|
.catch((err) =>
|
|
|
|
console.error(`MessageProcessor Failed: ${err}`)
|
|
|
|
)
|
|
|
|
)
|
|
|
|
.catch((err) =>
|
|
|
|
console.error(`MessageProcessor: Couldn't find message id ${messageId} in accountId: ${accountId}: ${err}`)
|
|
|
|
)
|
|
|
|
)
|
2016-06-21 05:57:54 +08:00
|
|
|
}
|
|
|
|
|
|
|
|
module.exports = {
|
|
|
|
processMessage,
|
|
|
|
}
|