core: use coll.watch to stream events

packages/hydrooj/package.json

@@ -1,6 +1,6 @@
 {
     "name": "hydrooj",
-    "version": "3.11.9",
+    "version": "3.11.10",
     "bin": "bin/hydrooj.js",
     "main": "src/loader",
     "module": "src/loader",

packages/hydrooj/src/interface.ts

@@ -562,7 +562,7 @@ export interface FileNode {
 }
 
 export interface EventDoc {
-    ack: string[];
+    sender: string;
     event: number | string;
     payload: string;
     expire: Date;

packages/hydrooj/src/model/task.ts

@@ -161,27 +161,18 @@ bus.once('app/started', async () => {
         });
     }
     await collEvent.createIndex({ expire: 1 }, { expireAfterSeconds: 0 });
-    (async () => {
+    const stream = collEvent.watch([{ $match: { sender: { $ne: id } } }]);
-        // eslint-disable-next-line no-constant-condition
+    stream.on('change', async (change) => {
-        while (true) {
+        if (change.operationType !== 'insert') return;
-            // eslint-disable-next-line no-await-in-loop
+        const doc = change.fullDocument;
-            const res = await collEvent.findOneAndUpdate(
+        const payload = JSON.parse(doc.payload);
-                { ack: { $nin: [id] } },
+        if (process.send) process.send({ type: 'hydro:broadcast', data: { event: doc.event, payload } });
-                { $push: { ack: id } },
+        if (doc) bus.parallel(doc.event, ...payload);
-            );
+    });
-            // eslint-disable-next-line no-await-in-loop
-            if (!res.value) await sleep(100);
-            else {
-                const payload = JSON.parse(res.value.payload);
-                if (process.send) process.send({ type: 'hydro:broadcast', data: { event: res.value.event, payload } });
-                if (res.value) bus.parallel(res.value.event, ...payload);
-            }
-        }
-    })();
 });
 bus.on('bus/broadcast', (event, payload) => {
     collEvent.insertOne({
-        ack: [id],
+        sender: id,
         event,
         payload: JSON.stringify(payload),
         expire: new Date(Date.now() + 10000),