From 429bacfaf25241ebd6932466c59e308256329405 Mon Sep 17 00:00:00 2001
From: undefined <i@undefined.moe>
Date: Wed, 6 Apr 2022 05:24:15 +0800
Subject: [PATCH] core: use coll.watch to stream events

---
 packages/hydrooj/package.json      |  2 +-
 packages/hydrooj/src/interface.ts  |  2 +-
 packages/hydrooj/src/model/task.ts | 27 +++++++++------------------
 3 files changed, 11 insertions(+), 20 deletions(-)

diff --git a/packages/hydrooj/package.json b/packages/hydrooj/package.json
index 33c486a50..8d1d5df8d 100644
--- a/packages/hydrooj/package.json
+++ b/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",
diff --git a/packages/hydrooj/src/interface.ts b/packages/hydrooj/src/interface.ts
index b2492784e..5fe7b3637 100644
--- a/packages/hydrooj/src/interface.ts
+++ b/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;
diff --git a/packages/hydrooj/src/model/task.ts b/packages/hydrooj/src/model/task.ts
index 03069d6d4..6185e45e5 100644
--- a/packages/hydrooj/src/model/task.ts
+++ b/packages/hydrooj/src/model/task.ts
@@ -161,27 +161,18 @@ bus.once('app/started', async () => {
         });
     }
     await collEvent.createIndex({ expire: 1 }, { expireAfterSeconds: 0 });
-    (async () => {
-        // eslint-disable-next-line no-constant-condition
-        while (true) {
-            // eslint-disable-next-line no-await-in-loop
-            const res = await collEvent.findOneAndUpdate(
-                { ack: { $nin: [id] } },
-                { $push: { ack: id } },
-            );
-            // 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);
-            }
-        }
-    })();
+    const stream = collEvent.watch([{ $match: { sender: { $ne: id } } }]);
+    stream.on('change', async (change) => {
+        if (change.operationType !== 'insert') return;
+        const doc = change.fullDocument;
+        const payload = JSON.parse(doc.payload);
+        if (process.send) process.send({ type: 'hydro:broadcast', data: { event: doc.event, payload } });
+        if (doc) bus.parallel(doc.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),