Bugfixes in sending campaigns

app-builder.js

@@ -1,7 +1,7 @@
 'use strict';
 
 const config = require('config');
-const log = require('npmlog');
+const log = require('./lib/log');
 
 const express = require('express');
 const bodyParser = require('body-parser');

index.js

@@ -1,7 +1,7 @@
 'use strict';
 
 const config = require('config');
-const log = require('npmlog');
+const log = require('./lib/log');
 const appBuilder = require('./app-builder');
 const http = require('http');
 const triggers = require('./services/triggers');
@@ -29,8 +29,6 @@ if (config.title) {
     process.title = config.title;
 }
 
-log.level = config.log.level;
-
 
 function startHTTPServer(appType, appName, port, callback) {
     const app = appBuilder.createApp(appType);
@@ -82,6 +80,15 @@ dbcheck(err => { // Check if database needs upgrading before starting the server
     .then(() => shares.regenerateRoleNamesTable())
     .then(() => shares.rebuildPermissions())
 
+/*
+    .then(() =>
+        testServer(() => {
+            senders.spawn(() => {
+            });
+        })
+    );
+*/
+
     .then(() =>
         executor.spawn(() => {
             testServer(() => {

lib/campaign-sender.js

@@ -148,7 +148,7 @@ class CampaignSender {
             return;
         }
 
-        const list = this.listsById.get(list.id);
+        const list = this.listsById.get(listId);
         const subscriptionGrouped = await subscriptions.getByEmail(contextHelpers.getAdminContext(), list.id, email);
         const flds = this.listsFieldsGrouped.get(listId);
         const campaign = this.campaign;
@@ -171,7 +171,7 @@ class CampaignSender {
         if (!list.listunsubscribe_disabled) {
             listUnsubscribe = campaign.unsubscribe_url
                 ? tools.formatMessage(campaign, list, subscriptionGrouped, mergeTags, campaign.unsubscribe_url)
-                : getPublicUrl('/subscription/' + list.cid + '/unsubscribe/' + subscriptionGrouped.subscription.cid);
+                : getPublicUrl('/subscription/' + list.cid + '/unsubscribe/' + subscriptionGrouped.cid);
         }
 
         const mailer = await mailers.getOrCreateMailer(sendConfiguration.id);
@@ -257,7 +257,7 @@ class CampaignSender {
         await knex('campaign_messages').insert({
             campaign: this.campaign.id,
             list: listId,
-            subscriptions: subscriptionGrouped.id,
+            subscription: subscriptionGrouped.id,
             send_configuration: sendConfiguration.id,
             status,
             response,

lib/dbcheck.js

@@ -4,20 +4,20 @@
 This module handles Mailtrain database initialization and upgrades
 */
 
-let config = require('config');
-let mysql = require('mysql2');
-let log = require('npmlog');
-let fs = require('fs');
-let pathlib = require('path');
-let Handlebars = require('handlebars');
+const config = require('config');
+const mysql = require('mysql2');
+const log = require('./log');
+const fs = require('fs');
+const pathlib = require('path');
+const Handlebars = require('handlebars');
 
 const highestLegacySchemaVersion = 29;
 
-let mysqlConfig = {
+const mysqlConfig = {
     multipleStatements: true
 };
 Object.keys(config.mysql).forEach(key => mysqlConfig[key] = config.mysql[key]);
-let db = mysql.createPool(mysqlConfig);
+const db = mysql.createPool(mysqlConfig);
 
 function listTables(callback) {
     db.getConnection((err, connection) => {

lib/executor.js

@@ -1,7 +1,7 @@
 'use strict';
 
 const fork = require('child_process').fork;
-const log = require('npmlog');
+const log = require('./log');
 const path = require('path');
 
 const requestCallbacks = {};

lib/feedcheck.js

@@ -1,7 +1,7 @@
 'use strict';
 
 const fork = require('child_process').fork;
-const log = require('npmlog');
+const log = require('./log');
 const path = require('path');
 
 let feedcheckProcess;

lib/importer.js

@@ -2,7 +2,7 @@
 
 const knex = require('./knex');
 const fork = require('child_process').fork;
-const log = require('npmlog');
+const log = require('./log');
 const path = require('path');
 const {ImportStatus, RunStatus} = require('../shared/imports');
 

lib/log.js

@@ -0,0 +1,8 @@
+'use strict';
+
+const config = require('config');
+const log = require('npmlog');
+
+log.level = config.log.level;
+
+module.exports = log;

lib/mailers.js

@@ -1,6 +1,6 @@
 'use strict';
 
-const log = require('npmlog');
+const log = require('./log');
 const config = require('config');
 
 const Handlebars = require('handlebars');

lib/passport.js

@@ -1,7 +1,7 @@
 'use strict';
 
 const config = require('config');
-const log = require('npmlog');
+const log = require('./log');
 const _ = require('./translate')._;
 const util = require('util');
 

lib/privilege-helpers.js

@@ -1,6 +1,6 @@
 'use strict';
 
-const log = require('npmlog');
+const log = require('./log');
 const config = require('config');
 
 const fs = require('fs');

lib/report-processor.js

@@ -1,6 +1,6 @@
 'use strict';
 
-const log = require('npmlog');
+const log = require('./log');
 const reports = require('../models/reports');
 const executor = require('./executor');
 const contextHelpers = require('../lib/context-helpers');

lib/senders.js

@@ -1,7 +1,7 @@
 'use strict';
 
 const fork = require('child_process').fork;
-const log = require('npmlog');
+const log = require('./log');
 const path = require('path');
 const knex = require('../lib/knex');
 const {CampaignStatus} = require('../shared/campaigns');
@@ -12,10 +12,8 @@ let senderProcess;
 function spawn(callback) {
     log.verbose('Senders', 'Spawning master sender process');
 
-    knex.transaction(async tx => {
-        await tx('campaigns').where('status', CampaignStatus.SENDING).update({status: CampaignStatus.SCHEDULED});
-
-    }).then(() => {
+    knex('campaigns').where('status', CampaignStatus.SENDING).update({status: CampaignStatus.SCHEDULED})
+    .then(() => {
         senderProcess = fork(path.join(__dirname, '..', 'services', 'sender-master.js'), [], {
             cwd: path.join(__dirname, '..'),
             env: {NODE_ENV: process.env.NODE_ENV}

lib/subscription-mail-helpers.js

@@ -1,6 +1,6 @@
 'use strict';
 
-const log = require('npmlog');
+const log = require('./log');
 const fields = require('../models/fields');
 const settings = require('../models/settings');
 const {getTrustedUrl} = require('./urls');

lib/translate.js

@@ -6,7 +6,7 @@ const Gettext = require('node-gettext');
 const gt = new Gettext();
 const fs = require('fs');
 const path = require('path');
-const log = require('npmlog');
+const log = require('./log');
 const gettextParser = require('gettext-parser');
 const fakelang = require('./fakelang');
 

models/campaigns.js

@@ -599,7 +599,7 @@ async function updateMessageResponse(context, message, response, responseId) {
     });
 }
 
-async function getSubscribersQueryGeneratorTx(tx, campaignId, onlyUnsent, batchSize) {
+async function getSubscribersQueryGeneratorTx(tx, campaignId, onlyUnsent) {
     /*
     This is supposed to produce queries like this:
 

models/fields.js

@@ -130,7 +130,7 @@ fieldTypes['checkbox-grouped'] = {
     cardinality: Cardinality.MULTIPLE,
     getHbsType: field => 'typeCheckboxGrouped',
     render: (field, value) => {
-        const subItems = value.map(col => field.groupedOptions[col].name);
+        const subItems = (value || []).map(col => field.groupedOptions[col].name);
 
         if (field.settings.groupTemplate) {
             return render(field.settings.groupTemplate, {
@@ -149,7 +149,10 @@ fieldTypes['radio-grouped'] = {
     enumerated: false,
     cardinality: Cardinality.SINGLE,
     getHbsType: field => 'typeRadioGrouped',
-    render: (field, value) => field.groupedOptions[value].name
+    render: (field, value) => {
+        const fld = field.groupedOptions[value];
+        return fld ? fld.name : '';
+    }
 };
 
 fieldTypes['dropdown-grouped'] = {
@@ -159,7 +162,10 @@ fieldTypes['dropdown-grouped'] = {
     enumerated: false,
     cardinality: Cardinality.SINGLE,
     getHbsType: field => 'typeDropdownGrouped',
-    render: (field, value) => field.groupedOptions[value].name
+    render: (field, value) => {
+        const fld = field.groupedOptions[value];
+        return fld ? fld.name : '';
+    }
 };
 
 fieldTypes['radio-enum'] = {
@@ -173,7 +179,10 @@ fieldTypes['radio-enum'] = {
     enumerated: true,
     cardinality: Cardinality.SINGLE,
     getHbsType: field => 'typeRadioEnum',
-    render: (field, value) => field.groupedOptions[value].name
+    render: (field, value) => {
+        const fld = field.groupedOptions[value];
+        return fld ? fld.name : '';
+    }
 };
 
 fieldTypes['dropdown-enum'] = {
@@ -187,7 +196,10 @@ fieldTypes['dropdown-enum'] = {
     enumerated: true,
     cardinality: Cardinality.SINGLE,
     getHbsType: field => 'typeDropdownEnum',
-    render: (field, value) => field.groupedOptions[value].name
+    render: (field, value) => {
+        const fld = field.groupedOptions[value];
+        return fld ? fld.name : '';
+    }
 };
 
 fieldTypes.option = {

models/links.js

@@ -1,6 +1,6 @@
 'use strict';
 
-const log = require('npmlog');
+const log = require('../lib/log');
 const knex = require('../lib/knex');
 const dtHelpers = require('../lib/dt-helpers');
 const shares = require('./shares');

models/shares.js

@@ -6,7 +6,7 @@ const { enforce } = require('../lib/helpers');
 const dtHelpers = require('../lib/dt-helpers');
 const entitySettings = require('../lib/entity-settings');
 const interoperableErrors = require('../shared/interoperable-errors');
-const log = require('npmlog');
+const log = require('../lib/log');
 const {getGlobalNamespaceId} = require('../shared/namespaces');
 const {getAdminId} = require('../shared/users');
 

routes/api.js

@@ -7,7 +7,7 @@ const fields = require('../models/fields');
 const { SubscriptionStatus, SubscriptionSource } = require('../shared/lists');
 const subscriptions = require('../models/subscriptions');
 const confirmations = require('../models/confirmations');
-const log = require('npmlog');
+const log = require('../lib/log');
 const router = require('../lib/router-async').create();
 const mailHelpers = require('../lib/subscription-mail-helpers');
 const interoperableErrors = require('../shared/interoperable-errors');

routes/links.js

@@ -1,6 +1,6 @@
 'use strict';
 
-const log = require('npmlog');
+const log = require('../lib/log');
 const config = require('config');
 const router = require('../lib/router-async').create();
 const links = require('../models/links');

routes/subscription.js

@@ -1,6 +1,6 @@
 'use strict';
 
-const log = require('npmlog');
+const log = require('../lib/log');
 const config = require('config');
 const router = require('../lib/router-async').create();
 const confirmations = require('../models/confirmations');

routes/webhooks.js

@@ -7,7 +7,7 @@ const sendConfigurations = require('../models/send-configurations');
 const contextHelpers = require('../lib/context-helpers');
 const {SubscriptionStatus} = require('../shared/lists');
 const {MailerType} = require('../shared/send-configurations');
-const log = require('npmlog');
+const log = require('../lib/log');
 const multer = require('multer');
 const uploads = multer();
 

services/executor.js

@@ -7,7 +7,7 @@
 const reportHelpers = require('../lib/report-helpers');
 const fork = require('child_process').fork;
 const path = require('path');
-const log = require('npmlog');
+const log = require('../lib/log');
 const fs = require('fs');
 const privilegeHelpers = require('../lib/privilege-helpers');
 

services/feedcheck.js

@@ -1,6 +1,6 @@
 'use strict';
 
-const log = require('npmlog');
+const log = require('../lib/log');
 const knex = require('../lib/knex');
 const feedparser = require('feedparser-promised');
 const { CampaignType, CampaignStatus, CampaignSource } = require('../shared/campaigns');

services/importer.js

@@ -2,7 +2,7 @@
 
 const knex = require('../lib/knex');
 const path = require('path');
-const log = require('npmlog');
+const log = require('../lib/log');
 const fsExtra = require('fs-extra-promise');
 const {ImportSource, MappingType, ImportStatus, RunStatus} = require('../shared/imports');
 const imports = require('../models/imports');

services/postfix-bounce-server.js

@@ -1,6 +1,6 @@
 'use strict';
 
-const log = require('npmlog');
+const log = require('../lib/log');
 const config = require('config');
 const net = require('net');
 const campaigns = require('../models/campaigns');

services/sender-master.js

@@ -2,7 +2,7 @@
 
 const config = require('config');
 const fork = require('child_process').fork;
-const log = require('npmlog');
+const log = require('../lib/log');
 const path = require('path');
 const knex = require('../lib/knex');
 const {CampaignStatus, CampaignType} = require('../shared/campaigns');
@@ -24,10 +24,13 @@ const workerBatchSize = 100;
 const messageQueue = new Map(); // campaignId -> [{listId, email}]
 const messageQueueCont = new Map(); // campaignId -> next batch callback
 
+const workAssignment = new Map(); // workerId -> { campaignId, subscribers: [{listId, email}] }
+
 let workerSchedulerCont = null;
 
 
 function messagesProcessed(workerId) {
+    workAssignment.delete(workerId);
     idleWorkers.push(workerId);
 
     if (workerSchedulerCont) {
@@ -70,6 +73,7 @@ async function scheduleWorkers() {
 
             if (queue.length > 0) {
                 const subscribers = queue.splice(0, workerBatchSize);
+                workAssignment.set(workerId, {campaignId, subscribers});
 
                 if (queue.length === 0 && messageQueueCont.has(campaignId)) {
                     const scheduleMessages = messageQueueCont.get(campaignId);
@@ -113,11 +117,21 @@ async function processCampaign(campaignId) {
 
         let qryGen;
         await knex.transaction(async tx => {
-            qryGen = await campaigns.getSubscribersQueryGeneratorTx(tx, campaignId, true, retrieveBatchSize);
+            qryGen = await campaigns.getSubscribersQueryGeneratorTx(tx, campaignId, true);
         });
 
         if (qryGen) {
-            const qry = qryGen(knex).select(['pending_subscriptions.email', 'campaign_lists.list']);
+            let subscribersInProcessing = [...msgQueue];
+            for (const wa of workAssignment.values()) {
+                if (wa.campaignId === campaignId) {
+                    subscribersInProcessing = subscribersInProcessing.concat(wa.subscribers);
+                }
+            }
+
+            const qry = qryGen(knex)
+                .whereNotIn('pending_subscriptions.email', subscribersInProcessing.map(x => x.email))
+                .select(['pending_subscriptions.email', 'campaign_lists.list'])
+                .limit(retrieveBatchSize);
             const subs = await qry;
 
             if (subs.length === 0) {
@@ -261,7 +275,7 @@ async function init() {
     });
 
     process.send({
-        type: 'sender-started'
+        type: 'master-sender-started'
     });
 
     periodicCampaignsCheck();

services/sender-worker.js

@@ -1,7 +1,7 @@
 'use strict';
 
 const config = require('config');
-const log = require('npmlog');
+const log = require('../lib/log');
 const mailers = require('../lib/mailers');
 const CampaignSender = require('../lib/campaign-sender');
 
@@ -25,6 +25,7 @@ async function processMessages(campaignId, subscribers) {
             log.verbose('Senders', 'Message sent and status updated for %s:%s', subData.listId, subData.email);
         } catch (err) {
             log.error('Senders', `Sending message to ${subData.listId}:${subData.email} failed with error: ${err.message}`)
+            log.verbose(err);
         }
     }
 

services/test-server.js

@@ -1,6 +1,6 @@
 'use strict';
 
-const log = require('npmlog');
+const log = require('../lib/log');
 const config = require('config');
 const crypto = require('crypto');
 const humanize = require('humanize');

services/triggers.js

@@ -1,6 +1,6 @@
 'use strict';
 
-const log = require('npmlog');
+const log = require('../lib/log');
 const knex = require('../lib/knex');
 const triggers = require('../models/triggers');
 const campaigns = require('../models/campaigns');

services/tzupdate.js

@@ -10,7 +10,7 @@
 
 const moment = require('moment-timezone');
 const knex = require('../lib/knex');
-const log = require('npmlog');
+const log = require('../lib/log');
 let lastCheck = false;
 
 const timezone_timeout = 60 * 60 * 1000;

services/verp-server.js

@@ -1,7 +1,7 @@
 'use strict';
 
 const { nodeifyFunction, nodeifyPromise } = require('../lib/nodeify');
-const log = require('npmlog');
+const log = require('../lib/log');
 const config = require('config');
 const {MailerError} = require('../lib/mailers');
 const campaigns = require('../models/campaigns');

workers/reports/report-processor.js

@@ -10,7 +10,7 @@ const handlebarsHelpers = require('../../lib/handlebars-helpers');
 const _ = require('../../lib/translate')._;
 const hbs = require('hbs');
 const vm = require('vm');
-const log = require('npmlog');
+const log = require('../../lib/log');
 const fs = require('fs');
 const knex = require('../../lib/knex');
 const contextHelpers = require('../../lib/context-helpers');