多worker共享事件（不稳定

hydro/handler/domain.js

@@ -73,7 +73,7 @@ class DomainEditHandler extends ManageHandler {
 
     async post(args) {
         const $set = {};
-        for (const key of args) {
+        for (const key in args) {
             if (DOMAIN_SETTINGS_BY_KEY[key]) $set[key] = args[key];
         }
         await domain.edit(args.domainId, $set);

hydro/handler/home.js

@@ -272,7 +272,7 @@ class HomeMessagesHandler extends Handler {
 
 class HomeMessagesConnectionHandler extends ConnectionHandler {
     async prepare() {
-        bus.subscribe([`user_message-${this.user._id}`], (...args) => this.onMessageReceived(...args));
+        bus.subscribe([`user_message-${this.user._id}`], this, 'onMessageReceived');
     }
 
     async onMessageReceived(e) {
@@ -280,7 +280,7 @@ class HomeMessagesConnectionHandler extends ConnectionHandler {
     }
 
     async cleanup() {
-        bus.unsubscribe(this.onMessageReceived);
+        bus.unsubscribe([`user_message-${this.user._id}`], this, 'onMessageReceived');
     }
 }
 

hydro/handler/manage.js

@@ -106,7 +106,7 @@ class SystemScriptHandler extends SystemHandler {
                 });
         }, 500);
         this.response.body = { rid };
-        this.response.redirect = this.url('record_main', { rid });
+        this.response.redirect = this.url('record_detail', { rid });
     }
 }
 

hydro/handler/record.js

@@ -103,7 +103,7 @@ class RecordMainConnectionHandler extends RecordConnectionHandler {
                 return;
             }
         }
-        bus.subscribe(['record_change'], this.onRecordChange);
+        bus.subscribe(['record_change'], this, 'onRecordChange');
     }
 
     async message(msg) {
@@ -118,7 +118,7 @@ class RecordMainConnectionHandler extends RecordConnectionHandler {
     }
 
     async cleanup() {
-        bus.unsubscribe(['record_change'], this.onRecordChange);
+        bus.unsubscribe(['record_change'], this, 'onRecordChange');
     }
 
     async onRecordChange(data) {
@@ -145,7 +145,7 @@ class RecordDetailConnectionHandler extends contest.ContestHandlerMixin(Connecti
             }
         }
         this.rid = rid;
-        bus.subscribe(['record_change'], this.onRecordChange);
+        bus.subscribe(['record_change'], this, 'onRecordChange');
         this.onRecordChange({ value: rdoc });
     }
 
@@ -159,7 +159,7 @@ class RecordDetailConnectionHandler extends contest.ContestHandlerMixin(Connecti
     }
 
     async cleanup() {
-        bus.unsubscribe(['record_change'], this.onRecordChange);
+        bus.unsubscribe(['record_change'], this, 'onRecordChange');
     }
 }
 

hydro/model/task.js

@@ -1,3 +1,4 @@
+const cluster = require('cluster');
 const moment = require('moment-timezone');
 const db = require('../service/db');
 
@@ -6,6 +7,8 @@ const coll = db.collection('task');
 async function add(task) {
     const t = { ...task };
     if (typeof t.executeAfter === 'object') t.executeAfter = t.executeAfter.getTime();
+    t.count = t.count || 1;
+    t.wait = t.wait || Object.keys(cluster.workers);
     t.executeAfter = t.executeAfter || new Date().getTime();
     const res = await coll.insertOne(t);
     return res.insertedId;
@@ -26,9 +29,16 @@ function del(_id) {
 async function getFirst(query) {
     const q = { ...query };
     q.executeAfter = q.executeAfter || { $lt: new Date().getTime() };
+    q.wait = { $elemMatch: { $eq: cluster.worker.id } };
     const res = await coll.find(q).sort('_id', 1).limit(1).toArray();
     if (res.length) {
-        await coll.deleteOne({ _id: res[0]._id });
+        if (res[0].count === 1) await coll.deleteOne({ _id: res[0]._id });
+        else {
+            await coll.updateOne(
+                { _id: res[0]._id },
+                { $inc: { count: -1 }, $pull: { wait: cluster.worker.id } },
+            );
+        }
         if (res[0].interval) {
             await coll.insertOne({
                 ...res[0], executeAfter: moment().add(...res[0].interval).toDate(),

hydro/model/user.js

@@ -109,8 +109,11 @@ function setPassword(uid, password) {
     );
 }
 
-function setById(uid, args) {
-    return coll.findOneAndUpdate({ _id: uid }, { $set: args });
+function setById(uid, $set, $unset) {
+    const op = {};
+    if ($set && Object.keys($set).length) op.$set = $set;
+    if ($unset && Object.keys($unset).length) op.$unset = $unset;
+    return coll.findOneAndUpdate({ _id: uid }, op);
 }
 
 function setEmail(uid, mail) {

hydro/service/bus.js

@@ -1,15 +1,52 @@
 const { EventEmitter } = require('events');
+const cluster = require('cluster');
 
 const bus = new EventEmitter();
 
-global.Hydro.service.bus = module.exports = {
-    subscribe(events, handler) {
+function subscribe(events, handler, funcName) {
+    if (!funcName) {
         for (const event of events) bus.on(event, handler);
-    },
-    unsubscribe(events, handler) {
+    } else {
+        handler.__bus = (...args) => {
+            handler[funcName].call(handler, ...args);
+        };
+        for (const event of events) bus.on(event, handler.__bus);
+    }
+}
+
+function unsubscribe(events, handler, funcName) {
+    if (!funcName) {
         for (const event of events) bus.off(event, handler);
-    },
-    publish(event, data) {
-        bus.emit(event, { value: data, event });
-    },
+    } else {
+        handler.__bus = (...args) => {
+            handler[funcName].call(handler, ...args);
+        };
+        for (const event of events) bus.off(event, handler.__bus);
+        delete handler.__bus;
+    }
+}
+
+function publish(event, data, isMaster = true) {
+    bus.emit(event, { value: data, event });
+    if (isMaster && global.Hydro.model.task) {
+        global.Hydro.model.task.add({
+            type: 'bus',
+            count: Object.keys(cluster.workers).length,
+            sender: cluster.worker.id,
+            event,
+            data,
+        });
+    }
+}
+
+function postInit() {
+    const { task } = global.Hydro.model;
+    task.consume({ type: 'bus' }, (data) => {
+        if (data.sender === cluster.worker.id) return;
+        publish(data.event, data.value, false);
+    });
+}
+
+global.Hydro.service.bus = module.exports = {
+    subscribe, unsubscribe, publish, postInit,
 };

hydro/utils.js

@@ -1,10 +1,15 @@
 const fs = require('fs');
+const cluster = require('cluster');
 const path = require('path');
 const superagent = require('superagent');
 const proxy = require('superagent-proxy');
 
 proxy(superagent);
 
+if (!cluster.worker) {
+    cluster.worker = { id: 0 };
+}
+
 const dict = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890';
 
 String.random = function random(digit = 32) {

module/migrate-vijos/script.js

@@ -2,7 +2,7 @@
 const fs = require('fs');
 const os = require('os');
 
-const { mongodb } = global.nodeModules;
+const { mongodb, bson } = global.nodeModules;
 const dst = global.Hydro.service.db;
 const { file, problem, discussion } = global.Hydro.model;
 const { readConfig } = global.Hydro.lib;
@@ -378,6 +378,42 @@ async function fixProblem(report) {
     }
 }
 
+/**
+ * @param {Date} ts
+ */
+function objid(ts) {
+    const p = Math.floor(ts.getTime() / 1000).toString(16);
+    const id = new bson.ObjectID();
+    return p + id.toString().slice(8, 8 + 6 + 4 + 6);
+}
+
+/**
+ * @param {import('mongodb').Db} src
+ * @param {*} report
+ */
+async function message(src, report) {
+    const count = await src.collection('message').find().count();
+    await report({ progress: 1, message: `Messages: ${count}` });
+    const total = Math.floor(count / 50);
+    for (let i = 0; i <= total; i++) {
+        const docs = await dst.collection('message')
+            .find().skip(i * 50).limit(50)
+            .toArray();
+        for (const doc of docs) {
+            for (const msg of doc.reply) {
+                await dst.collection('message').insertOne({
+                    _id: objid(msg.at),
+                    from: msg.sender_uid,
+                    to: msg.sender_uid === doc.sender_uid ? doc.sendee_uid : doc.sender_uid,
+                    content: msg.content,
+                    unread: msg.status !== 0,
+                });
+            }
+        }
+        await report({ progress: Math.round(100 * ((i + 1) / (total + 1))) });
+    }
+}
+
 async function task(name, src, report) {
     const count = await cursor[name](src).count();
     await report({ progress: 1, message: `${name}: ${count}` });
@@ -482,11 +518,13 @@ async function migrateVijos({
         await dst.collection('system').insertOne({ _id: 'migrateVijosFs', value: true });
     }
     await dst.collection('user').deleteMany({ _id: { $nin: [0, 1] } });
+    await dst.collection('message').deleteMany();
     const d = ['user', 'document', 'document.status', 'record', 'file'];
     for (const i of d) await task(i, src, report);
     await fixProblem(report);
     await discussionNode(src, report);
     await domainUser(src, report);
+    await message(src, report);
     return true;
 }