WiP on mailers

obsolete/lib/db-old.js

@@ -0,0 +1,117 @@
+'use strict';
+
+let config = require('config');
+let mysql = require('mysql2');
+let redis = require('redis');
+let Lock = require('redfour');
+let stringifyDate = require('json-stringify-date');
+let senders = require('./senders');
+
+const bluebird = require('bluebird');
+
+module.exports = mysql.createPool(config.mysql);
+if (config.redis && config.redis.enabled) {
+
+    module.exports.redis = redis.createClient(config.redis);
+
+    let queueLock = new Lock({
+        redis: config.redis,
+        namespace: 'mailtrain:lock'
+    });
+
+    module.exports.getLock = bluebird.promisify((id, callback) => {
+        queueLock.waitAcquireLock(id, 60 * 1000 /* Lock expires after 60sec */ , 10 * 1000 /* Wait for lock for up to 10sec */ , (err, lock) => {
+            if (err) {
+                return callback(err);
+            }
+            if (!lock) {
+                return callback(null, false);
+            }
+            return callback(null, {
+                lock,
+                release(done) {
+                    queueLock.releaseLock(lock, done);
+                }
+            });
+        });
+    });
+
+    module.exports.clearCache = bluebird.promisify((key, callback) => {
+        module.exports.redis.del('mailtrain:cache:' + key, err => callback(err));
+    });
+
+    module.exports.addToCache = bluebird.promisify((key, value, callback) => {
+        if (!value) {
+            return setImmediate(() => callback());
+        }
+        module.exports.redis.multi().
+            lpush('mailtrain:cache:' + key, stringifyDate.stringify(value)).
+            expire('mailtrain:cache:' + key, 24 * 3600).
+            exec(err => callback(err));
+    });
+
+    module.exports.getFromCache = bluebird.promisify((key, callback) => {
+        module.exports.redis.rpop('mailtrain:cache:' + key, (err, value) => {
+            if (err) {
+                return callback(err);
+            }
+            try {
+                value = stringifyDate.parse(value);
+            } catch (E) {
+                return callback(E);
+            }
+
+            return callback(null, value);
+        });
+    });
+
+} else {
+    // fakelock. does not lock anything
+    module.exports.getLock = bluebird.promisify((id, callback) => {
+        setImmediate(() => callback(null, {
+            lock: false,
+            release(done) {
+                setImmediate(done);
+            }
+        }));
+    });
+
+    let caches = new Map();
+
+    module.exports.clearCache = bluebird.promisify((key, callback) => {
+        caches.delete(key);
+        // This notifies the callback below - i.e. process.on(...) - which is installed in the sender process.
+        senders.workers.forEach(child => {
+            child.send({
+                cmd: 'db.clearCache',
+                key
+            });
+        });
+        setImmediate(() => callback());
+    });
+
+    process.on('message', m => {
+        if (m && m.cmd === 'db.clearCache' && m.key) {
+            caches.delete(m.key);
+        }
+    });
+
+    module.exports.addToCache = bluebird.promisify((key, value, callback) => {
+        if (!caches.has(key)) {
+            caches.set(key, []);
+        }
+        caches.get(key).push(value);
+        setImmediate(() => callback());
+    });
+
+    module.exports.getFromCache = (key, callback) => {
+        let value;
+        if (caches.has(key)) {
+            value = caches.get(key).shift();
+            if (!caches.get(key).length) {
+                caches.delete(key);
+            }
+        }
+        setImmediate(() => callback(null, value));
+    };
+}