From 3adde783a3d394f3dd14ec45f7c1dda9c652c3c3 Mon Sep 17 00:00:00 2001
From: undefined <i@undefined.moe>
Date: Tue, 21 Sep 2021 15:24:26 +0800
Subject: [PATCH] core: pm2 event bus

---
 packages/hydrooj/package.json          |  8 ++++++--
 packages/hydrooj/src/handler/home.ts   |  1 -
 packages/hydrooj/src/handler/manage.ts |  1 -
 packages/hydrooj/src/model/task.ts     |  5 +++--
 packages/hydrooj/src/service/bus.ts    | 18 ++++++++++++++++++
 packages/hydrooj/src/service/server.ts | 12 ++++++++----
 6 files changed, 35 insertions(+), 10 deletions(-)

diff --git a/packages/hydrooj/package.json b/packages/hydrooj/package.json
index 0f3a750cf..505ac2826 100644
--- a/packages/hydrooj/package.json
+++ b/packages/hydrooj/package.json
@@ -1,6 +1,6 @@
 {
     "name": "hydrooj",
-    "version": "2.36.15",
+    "version": "2.37.0",
     "bin": "bin/hydrooj.js",
     "main": "src/loader",
     "module": "src/loader",
@@ -64,6 +64,10 @@
         "@types/source-map-support": "^0",
         "@types/superagent": "^4.1.12",
         "formidable": "^1.2.2",
-        "moment": "^2.29.1"
+        "moment": "^2.29.1",
+        "pm2": "^5.1.1"
+    },
+    "peerDependencies": {
+        "pm2": "*"
     }
 }
diff --git a/packages/hydrooj/src/handler/home.ts b/packages/hydrooj/src/handler/home.ts
index 608fa65bd..47516a57e 100644
--- a/packages/hydrooj/src/handler/home.ts
+++ b/packages/hydrooj/src/handler/home.ts
@@ -261,7 +261,6 @@ class HomeSettingsHandler extends Handler {
         }
         for (const key in booleanKeys) if (!args[key]) $set[key] = false;
         await setter($set);
-        await sleep(100);
         this.back();
     }
 }
diff --git a/packages/hydrooj/src/handler/manage.ts b/packages/hydrooj/src/handler/manage.ts
index b62d4602a..bca9d294b 100644
--- a/packages/hydrooj/src/handler/manage.ts
+++ b/packages/hydrooj/src/handler/manage.ts
@@ -179,7 +179,6 @@ class SystemSettingHandler extends SystemHandler {
         }
         tasks.push(bus.parallel('system/setting', args));
         await Promise.all(tasks);
-        await sleep(100);
         this.back();
     }
 }
diff --git a/packages/hydrooj/src/model/task.ts b/packages/hydrooj/src/model/task.ts
index 13482db3c..5537788bf 100644
--- a/packages/hydrooj/src/model/task.ts
+++ b/packages/hydrooj/src/model/task.ts
@@ -135,12 +135,14 @@ class TaskModel {
     static Worker = Worker;
 }
 
+const id = hostname();
 Worker.addHandler('task.daily', async () => {
     await global.Hydro.script.rp?.run({}, new Logger('task/rp').debug);
     await global.Hydro.script.problemStat?.run({}, new Logger('task/problem').debug);
 });
 bus.on('domain/delete', (domainId) => coll.deleteMany({ domainId }));
 bus.once('app/started', async () => {
+    if (process.env.NODE_APP_INSTANCE !== '0') return;
     if (!await TaskModel.count({ type: 'schedule', subType: 'task.daily' })) {
         await TaskModel.add({
             type: 'schedule',
@@ -150,7 +152,6 @@ bus.once('app/started', async () => {
         });
     }
     await collEvent.createIndex({ expire: 1 }, { expireAfterSeconds: 0 });
-    const id = hostname() + process.env.NODE_APP_INSTANCE;
     (async () => {
         // eslint-disable-next-line no-constant-condition
         while (true) {
@@ -167,7 +168,7 @@ bus.once('app/started', async () => {
 });
 bus.on('bus/broadcast', (event, payload) => {
     collEvent.insertOne({
-        ack: [],
+        ack: [id],
         event,
         payload: JSON.stringify(payload),
         expire: new Date(Date.now() + 10000),
diff --git a/packages/hydrooj/src/service/bus.ts b/packages/hydrooj/src/service/bus.ts
index ab480f4c0..3bdeb1d97 100644
--- a/packages/hydrooj/src/service/bus.ts
+++ b/packages/hydrooj/src/service/bus.ts
@@ -183,6 +183,24 @@ export function broadcast<K extends keyof EventMap>(event: K, ...payload: Parame
     return parallel('bus/broadcast', event, payload);
 }
 
+try {
+    if (!process.send) throw new Error();
+    const pm2: typeof import('pm2') = require('pm2');
+    pm2.launchBus((err, bus) => {
+        if (err) throw new Error();
+        bus.on('hydro:broadcast', (packet) => {
+            parallel(packet.data.event, ...packet.data.payload);
+        });
+        on('bus/broadcast', (event, payload) => {
+            process.send({ type: 'hydro:broadcast', data: { event, payload } });
+        });
+        console.debug('Using pm2 event bus');
+    });
+} catch (e) {
+    on('bus/broadcast', (event, payload) => parallel(event, ...payload));
+    console.debug('Using mongodb external event bus');
+}
+
 global.Hydro.service.bus = {
     addListener, bail, broadcast, emit, on, off, once, parallel, prependListener, removeListener, serial,
 };
diff --git a/packages/hydrooj/src/service/server.ts b/packages/hydrooj/src/service/server.ts
index a75b3e2da..ac19b7119 100644
--- a/packages/hydrooj/src/service/server.ts
+++ b/packages/hydrooj/src/service/server.ts
@@ -887,13 +887,17 @@ export function Connection(
 let started = false;
 
 // TODO use postInit?
-export function start() {
+export async function start() {
     if (started) return;
     const port = system.get('server.port');
     app.use(router.routes()).use(router.allowedMethods());
-    server.listen(argv.options.port || port);
-    logger.success('Server listening at: %d', argv.options.port || port);
-    started = true;
+    await new Promise((resolve) => {
+        server.listen(argv.options.port || port, () => {
+            logger.success('Server listening at: %d', argv.options.port || port);
+            started = true;
+            resolve(true);
+        });
+    });
 }
 
 global.Hydro.service.server = {