core: 优化钩子逻辑

.eslintrc.js

@@ -29,6 +29,7 @@ module.exports = {
         'max-len': ['warn', 150],
         'no-bitwise': 'off',
         'no-console': 'off',
+        'no-continue': 'off',
         'no-dupe-class-members': 'off',
         'no-extend-native': 'off',
         'no-inner-declarations': 'off',

packages/hydrooj/TODO

@@ -4,4 +4,5 @@ SETTINGS_PRIVACY.allowRegisteredUsers
 model.setting.ui.name
 pdoc.secretConfig?
 message.markAsRead
-基于document的model难以lint，建议拆分
+基于document的model难以lint，建议拆分
+检查多余的索引

packages/hydrooj/package.json

@@ -1,6 +1,6 @@
 {
     "name": "hydrooj",
-    "version": "2.13.0",
+    "version": "2.13.1",
     "bin": "bin/hydrooj.js",
     "main": "dist/loader.js",
     "typings": "dist/loader.d.ts",

packages/hydrooj/src/entry/cli.ts

@@ -5,6 +5,7 @@ import {
     lib, service, model,
     builtinLib, builtinModel,
 } from './common';
+import * as bus from '../service/bus';
 
 const COMMENTS = /((\/\/.*$)|(\/\*[\s\S]*?\*\/))/mg;
 const ARR = /=>.*$/mg;
@@ -51,13 +52,8 @@ export async function load() {
     require('../utils');
     require('../error');
     require('../options');
-    const bus = require('../service/bus');
     await new Promise((resolve) => {
-        const h = () => {
-            bus.unsubscribe(['system_database_connected'], h);
-            resolve();
-        };
-        bus.subscribe(['system_database_connected'], h);
+        bus.once('database/connect', resolve);
         require('../service/db');
     });
     for (const i of builtinLib) require(`../lib/${i}`);
@@ -66,6 +62,6 @@ export async function load() {
     await service(pending, fail);
     for (const i of builtinModel) require(`../model/${i}`);
     await model(pending, fail);
-    for (const postInit of global.Hydro.postInit) await postInit();
+    await bus.parallel('app/started');
     await cli();
 }

packages/hydrooj/src/entry/master.ts

@@ -7,6 +7,7 @@ import pslist from 'ps-list';
 import { argv } from 'yargs';
 import { builtinModel } from './common';
 import { Entry } from '../loader';
+import * as bus from '../service/bus';
 
 const tmpdir = path.resolve(os.tmpdir(), 'hydro');
 const lockfile = path.resolve(tmpdir, 'lock.json');
@@ -31,9 +32,6 @@ export async function load(call: Entry) {
         ppid: process.ppid,
     };
     await fs.writeFile(lockfile, JSON.stringify(context));
-    global.onDestory.push(() => {
-        fs.removeSync(lockfile);
-    });
     require('../lib/i18n');
     require('../utils');
     require('../error');
@@ -45,14 +43,15 @@ export async function load(call: Entry) {
             process.exit(1);
         });
     }
-    const bus = require('../service/bus');
+    bus.once('app/exit', () => {
+        fs.removeSync(lockfile);
+    });
     await new Promise((resolve) => {
         const h = () => {
             console.log('Database connected');
-            bus.unsubscribe(['system_database_connected'], h);
             resolve();
         };
-        bus.subscribe(['system_database_connected'], h);
+        bus.once('database/connect', h);
         require('../service/db');
     });
     require('../service/monitor');
@@ -63,11 +62,6 @@ export async function load(call: Entry) {
         const ins = require('../script/upgrade0_1');
         await ins.run({ username: 'Root', password: 'rootroot' });
     }
-    for (const postInit of global.Hydro.postInit) {
-        await postInit().catch((e) => {
-            console.error('Error running postInit:', postInit.toString());
-            console.error(e);
-        });
-    }
+    bus.serial('app/started');
     return await modelSystem.get('server.worker');
 }

packages/hydrooj/src/entry/worker.ts

@@ -4,6 +4,7 @@ import {
     locale, template, lib, service, model, handler, script, setting, uistatic,
     builtinLib, builtinScript, builtinHandler, builtinModel,
 } from './common';
+import * as bus from '../service/bus';
 
 export async function load() {
     const pending = global.addons;
@@ -18,14 +19,12 @@ export async function load() {
         template(pending, fail),
         uistatic(pending, fail),
     ]);
-    const bus = require('../service/bus');
     await new Promise((resolve) => {
         const h = () => {
             console.log('Database connected');
-            bus.unsubscribe(['system_database_connected'], h);
             resolve();
         };
-        bus.subscribe(['system_database_connected'], h);
+        bus.once('database/connect', h);
         require('../service/db');
     });
     for (const i of builtinLib) require(`../lib/${i}`);
@@ -48,7 +47,7 @@ export async function load() {
     await notfound.apply();
     for (const i of builtinScript) require(`../script/${i}`);
     await script(pending, fail, active);
-    for (const postInit of global.Hydro.postInit) await postInit();
+    await bus.serial('app/started');
     await server.start();
     setInterval(() => {
         process.send({ event: 'stat', count: global.Hydro.stat.reqCount });

packages/hydrooj/src/handler/contest.ts

@@ -310,7 +310,7 @@ class ContestProblemSubmitHandler extends ContestProblemHandler {
             record.get(domainId, rid),
             contest.updateStatus(domainId, this.tdoc.docId, this.user._id, rid, this.pdoc.docId),
         ]);
-        bus.publish('record_change', rdoc);
+        bus.boardcast('record/change', rdoc);
         if (!this.canShowRecord(this.tdoc)) {
             this.response.body = { tid };
             this.response.redirect = this.url('contest_detail', { tid });

packages/hydrooj/src/handler/home.ts

@@ -25,6 +25,7 @@ import { PERM, PRIV, CONSTANT } from '../model/builtin';
 import {
     isContent, isPassword, isEmail, isTitle,
 } from '../lib/validator';
+import { Mdoc } from '../interface';
 
 const { geoip, useragent } = global.Hydro.lib;
 
@@ -358,18 +359,19 @@ class HomeMessagesHandler extends Handler {
 }
 
 class HomeMessagesConnectionHandler extends ConnectionHandler {
-    id: string;
+    dispose: bus.Disposable;
 
     async prepare() {
-        bus.subscribe([`user_message-${this.user._id}`], this.onMessageReceived.bind(this));
+        this.dispose = bus.on('user/message', this.onMessageReceived.bind(this));
     }
 
-    async onMessageReceived(e: any) {
-        this.send(e.value);
+    async onMessageReceived(uid: number, e: any) {
+        if (uid !== this.user._id) return;
+        this.send(e);
     }
 
     async cleanup() {
-        bus.unsubscribe([`user_message-${this.user._id}`], this.id);
+        if (this.dispose) this.dispose();
     }
 }
 

packages/hydrooj/src/handler/homework.ts

@@ -189,7 +189,7 @@ class HomeworkDetailProblemSubmitHandler extends HomeworkDetailProblemHandler {
             contest.updateStatus(domainId, tid, this.user._id,
                 rid, pid, false, 0, document.TYPE_HOMEWORK),
         ]);
-        bus.publish('record_change', rdoc);
+        bus.boardcast('record/change', rdoc);
         this.response.body = { tid, rid };
         if (this.canShowRecord(this.tdoc)) this.response.redirect = this.url('record_detail', { rid });
         else this.response.redirect = this.url('homework_detail', { tid });

packages/hydrooj/src/handler/judge.ts

@@ -62,7 +62,7 @@ export async function next(body: JudgeResultBody) {
     if (body.memory) $set.memory = body.memory;
     if (body.progress) $set.progress = body.progress;
     rdoc = await record.update(body.domainId, body.rid, $set, $push);
-    bus.publish('record_change', { rdoc, $set, $push });
+    bus.boardcast('record/change', rdoc, $set, $push);
 }
 
 export async function end(body: JudgeResultBody) {
@@ -87,7 +87,7 @@ export async function end(body: JudgeResultBody) {
     $set.judger = body.judger;
     rdoc = await record.update(body.domainId, body.rid, $set, $push, $unset);
     await _postJudge(rdoc);
-    bus.publish('record_change', { rdoc }); // trigger a full update
+    bus.boardcast('record/change', rdoc); // trigger a full update
 }
 
 class JudgeHandler extends Handler {

packages/hydrooj/src/handler/problem.ts

@@ -1,6 +1,6 @@
 import { isSafeInteger, flatten } from 'lodash';
 import yaml from 'js-yaml';
-import { ObjectID } from 'mongodb';
+import { FilterQuery, ObjectID } from 'mongodb';
 import {
     NoProblemError, ProblemDataNotFoundError, BadRequestError,
     SolutionNotFoundError, ProblemNotFoundError, ValidationError,
@@ -61,16 +61,14 @@ class ProblemMainHandler extends ProblemHandler {
     @param('q', Types.String, true)
     async get(domainId: string, page = 1, q = '') {
         this.response.template = 'problem_main.html';
-        const query: any = {};
+        const query: FilterQuery<Pdoc> = {};
         let psdict = {};
         const path: PathComponent[] = [
             ['Hydro', 'homepage'],
             ['problem_main', null],
         ];
         if (q) {
-            q = q.toLowerCase();
-            const $regex = new RegExp(`\\A\\Q${q}\\E`, 'gmi');
-            query.title = { $regex };
+            query.$text = { $search: q };
             path.push([q, null, null, true]);
         }
         if (!this.user.hasPerm(PERM.PERM_VIEW_PROBLEM_HIDDEN)) query.hidden = false;
@@ -252,7 +250,7 @@ class ProblemSubmitHandler extends ProblemDetailHandler {
             record.get(domainId, rid),
             domain.incUserInDomain(domainId, this.user._id, 'nSubmit'),
         ]);
-        bus.publish('record_change', { rdoc });
+        bus.boardcast('record/change', rdoc);
         this.response.body = { rid };
         this.response.redirect = this.url('record_detail', { rid });
     }
@@ -275,7 +273,7 @@ class ProblemPretestHandler extends ProblemDetailHandler {
             },
         );
         const rdoc = await record.get(domainId, rid);
-        bus.publish('record_change', { rdoc });
+        bus.boardcast('record/change', rdoc);
         this.response.body = { rid };
     }
 }
@@ -285,7 +283,7 @@ class ProblemPretestConnectionHandler extends ConnectionHandler {
 
     domainId: string;
 
-    id: string;
+    dispose: bus.Disposable;
 
     @param('pid', Types.String)
     async prepare(domainId: string, pid: string) {
@@ -294,11 +292,10 @@ class ProblemPretestConnectionHandler extends ConnectionHandler {
         if (!pdoc) throw new ProblemNotFoundError(domainId, pid);
         this.pid = pdoc.docId.toString();
         this.domainId = domainId;
-        this.id = bus.subscribe(['record_change'], this.onRecordChange.bind(this));
+        this.dispose = bus.on('record/change', this.onRecordChange.bind(this));
     }
 
-    async onRecordChange(data) {
-        const rdoc: Rdoc = data.value.rdoc;
+    async onRecordChange(rdoc: Rdoc) {
         if (
             rdoc.uid !== this.user._id
             || rdoc.pid.toString() !== this.pid
@@ -316,7 +313,7 @@ class ProblemPretestConnectionHandler extends ConnectionHandler {
     }
 
     async cleanup() {
-        bus.unsubscribe(['record_change'], this.id);
+        if (this.dispose) this.dispose();
     }
 }
 

packages/hydrooj/src/handler/record.ts

@@ -102,6 +102,8 @@ class RecordRejudgeHandler extends Handler {
 const RecordConnectionHandler = contest.ContestHandlerMixin(ConnectionHandler);
 
 class RecordMainConnectionHandler extends RecordConnectionHandler {
+    dispose: bus.Disposable;
+
     @param('tid', Types.ObjectID, true)
     async prepare(domainId: string, tid: ObjectID) {
         this.domainId = domainId;
@@ -113,7 +115,7 @@ class RecordMainConnectionHandler extends RecordConnectionHandler {
                 return;
             }
         }
-        this.id = bus.subscribe(['record_change'], this.onRecordChange.bind(this));
+        this.dispose = bus.on('record/change', this.onRecordChange.bind(this));
     }
 
     async message(msg) {
@@ -123,17 +125,16 @@ class RecordMainConnectionHandler extends RecordConnectionHandler {
                 this.domainId, { _id: { $in: rids } },
             ).toArray();
             for (const rdoc of rdocs) {
-                this.onRecordChange({ value: { rdoc } });
+                this.onRecordChange(rdoc);
             }
         }
     }
 
     async cleanup() {
-        bus.unsubscribe(['record_change'], this.id);
+        if (this.dispose) this.dispose();
     }
 
-    async onRecordChange(data) {
-        const rdoc: Rdoc = data.value.rdoc;
+    async onRecordChange(rdoc: Rdoc) {
         if (rdoc.contest && rdoc.contest.tid.toString() !== this.tid) return;
         // eslint-disable-next-line prefer-const
         let [udoc, pdoc] = await Promise.all([
@@ -146,6 +147,8 @@ class RecordMainConnectionHandler extends RecordConnectionHandler {
 }
 
 class RecordDetailConnectionHandler extends contest.ContestHandlerMixin(ConnectionHandler) {
+    dispose: bus.Disposable;
+
     @param('rid', Types.ObjectID)
     async prepare(domainId: string, rid: ObjectID) {
         const rdoc = await record.get(domainId, rid);
@@ -157,13 +160,11 @@ class RecordDetailConnectionHandler extends contest.ContestHandlerMixin(Connecti
             }
         }
         this.rid = rid.toString();
-        this.id = bus.subscribe(['record_change'], this.onRecordChange.bind(this));
-        this.onRecordChange({ value: { rdoc } });
+        this.dispose = bus.on('record/change', this.onRecordChange.bind(this));
+        this.onRecordChange(rdoc);
     }
 
-    async onRecordChange(data) {
-        const rdoc: Rdoc = data.value.rdoc;
-        const { $set, $push } = data.value;
+    async onRecordChange(rdoc: Rdoc, $set?: any, $push?: any) {
         if (rdoc._id.toString() !== this.rid) return;
         if ($set) this.send({ $set, $push });
         else {
@@ -175,7 +176,7 @@ class RecordDetailConnectionHandler extends contest.ContestHandlerMixin(Connecti
     }
 
     async cleanup() {
-        bus.unsubscribe(['record_change'], this.id);
+        if (this.dispose) this.dispose();
     }
 }
 

packages/hydrooj/src/interface.ts

@@ -455,9 +455,7 @@ declare global {
                 ui: UI,
                 error: typeof import('./error'),
                 locales: Dict<Dict<string>>,
-                postInit: Array<() => Promise<any>>,
             },
-            onDestory: Array<() => void | Promise<void>>,
             addons: string[],
         }
     }

packages/hydrooj/src/loader.ts

@@ -36,9 +36,7 @@ if (!global.Hydro) {
         // @ts-ignore
         error: {},
         locales: {},
-        postInit: [],
     };
-    global.onDestory = [];
     global.addons = [];
 }
 
@@ -47,9 +45,12 @@ if (argv.debug) {
     console.log(argv);
 }
 
+// eslint-disable-next-line import/first
+import * as bus from './service/bus';
+
 async function terminate() {
     try {
-        for (const task of global.onDestory) await task();
+        await bus.parallel('app/exit');
     } catch (e) {
         process.exit(1);
     }
@@ -127,7 +128,7 @@ async function messageHandler(worker: cluster.Worker, msg: any) {
                     cluster.workers[i].send(msg);
                 }
             } else {
-                global.Hydro.service.bus.publish(msg.eventName, msg.payload, false);
+                global.Hydro.service.bus.emit(msg.eventName, ...msg.payload);
             }
         } else if (msg.event === 'stat') {
             global.Hydro.stat.reqCount += msg.count;

packages/hydrooj/src/model/blacklist.ts

@@ -1,5 +1,6 @@
 import { Collection } from 'mongodb';
 import * as db from '../service/db';
+import * as bus from '../service/bus';
 import { Bdoc } from '../interface';
 
 const coll: Collection<Bdoc> = db.collection('blacklist');
@@ -36,5 +37,5 @@ function ensureIndexes() {
     return coll.createIndex('expireAt', { expireAfterSeconds: 0 });
 }
 
-global.Hydro.postInit.push(ensureIndexes);
+bus.once('app/started', ensureIndexes);
 global.Hydro.model.blacklist = { add, get, del };

packages/hydrooj/src/model/document.ts

@@ -1,11 +1,12 @@
 import assert from 'assert';
 import { ObjectID } from 'mongodb';
 import * as db from '../service/db';
+import * as bus from '../service/bus';
 
 type DocID = ObjectID | string | number;
 
-const coll = db.collection('document');
-const collStatus = db.collection('document.status');
+export const coll = db.collection('document');
+export const collStatus = db.collection('document.status');
 
 export const TYPE_PROBLEM = 10;
 export const TYPE_PROBLEM_SOLUTION = 11;
@@ -51,6 +52,7 @@ export async function add(
         doc.parentType = parentType;
         doc.parentId = parentId;
     }
+    await bus.serial('document/add', doc);
     const res = await coll.insertOne(doc);
     return docId || res.insertedId;
 }
@@ -337,6 +339,7 @@ async function ensureIndexes() {
     await ic({ domainId: 1, docType: 1, docId: 1 }, u);
     await ic({ domainId: 1, docType: 1, owner: 1, docId: -1 });
     // For problem
+    await ic({ domainId: 1, docType: 1, title: 'text' }, s);
     await ic({ domainId: 1, docType: 1, category: 1, docId: 1 }, s);
     await ic({ domainId: 1, docType: 1, hidden: 1, category: 1, docId: 1 }, s);
     await ic({ domainId: 1, docType: 1, tag: 1, docId: 1 }, s);
@@ -364,8 +367,11 @@ async function ensureIndexes() {
     await is({ domainId: 1, docType: 1, uid: 1, enroll: 1, docId: 1 }, s);
 }
 
-global.Hydro.postInit.push(ensureIndexes);
+bus.once('app/started', ensureIndexes);
 global.Hydro.model.document = {
+    coll,
+    collStatus,
+
     add,
     addToSet,
     cappedIncStatus,

packages/hydrooj/src/model/file.ts

@@ -2,10 +2,11 @@ import { ObjectID, Collection } from 'mongodb';
 import * as fs from 'fs';
 import { Dictionary } from 'lodash';
 import { ForbiddenError, NotFoundError } from '../error';
+import { Ufdoc } from '../interface';
 import * as db from '../service/db';
+import * as bus from '../service/bus';
 import gridfs from '../service/gridfs';
 import hash from '../lib/hash.hydro';
-import { Ufdoc } from '../interface';
 
 const coll: Collection<Ufdoc> = db.collection('file');
 const collFile = db.collection('fs.files');
@@ -121,7 +122,7 @@ function ensureIndexes() {
     ]);
 }
 
-global.Hydro.postInit.push(ensureIndexes);
+bus.once('app/started', ensureIndexes);
 global.Hydro.model.file = {
     add, get, del, getMeta, getWithSecret, getMulti, getMetaDict, url,
 };

packages/hydrooj/src/model/message.ts

@@ -22,7 +22,7 @@ export async function send(
     if (from !== to) {
         // ENHANCE domainId?
         const udoc = await user.getById('system', to);
-        bus.publish(`user_message-${to}`, { mdoc, udoc });
+        bus.boardcast('user/message', to, mdoc, udoc);
     }
     return mdoc;
 }
@@ -69,7 +69,7 @@ function ensureIndexes() {
     ]);
 }
 
-global.Hydro.postInit.push(ensureIndexes);
+bus.once('app/started', ensureIndexes);
 global.Hydro.model.message = {
     FLAG_UNREAD,
     FLAG_ALERT,

packages/hydrooj/src/model/opcount.ts

@@ -1,4 +1,5 @@
 import { OpcountExceededError } from '../error';
+import * as bus from '../service/bus';
 import * as db from '../service/db';
 
 const coll = db.collection('opcount');
@@ -23,5 +24,5 @@ function ensureIndexes() {
     return coll.createIndex('expireAt', { expireAfterSeconds: 0 });
 }
 
-global.Hydro.postInit.push(ensureIndexes);
+bus.once('app/started', ensureIndexes);
 global.Hydro.model.opcount = { inc };

packages/hydrooj/src/model/setting.ts

@@ -3,6 +3,7 @@ import moment from 'moment-timezone';
 import { Dictionary } from 'lodash';
 import * as builtin from './builtin';
 import { Setting as _Setting } from '../interface';
+import * as bus from '../service/bus';
 
 type SettingDict = Dictionary<_Setting>;
 
@@ -139,18 +140,16 @@ SystemSetting(
     Setting('setting_storage', 'user', null, 1, 'number', 'User Counter', null, FLAG_DISABLED | FLAG_HIDDEN),
 );
 
-global.Hydro.postInit.push(
-    async () => {
-        for (const setting of SYSTEM_SETTINGS) {
-            if (setting.value) {
-                const current = await global.Hydro.model.system.get(setting.key);
-                if (current === null || current === '') {
-                    await global.Hydro.model.system.set(setting.key, setting.value);
-                }
+bus.once('app/started', async () => {
+    for (const setting of SYSTEM_SETTINGS) {
+        if (setting.value) {
+            const current = await global.Hydro.model.system.get(setting.key);
+            if (current === null || current === '') {
+                await global.Hydro.model.system.set(setting.key, setting.value);
             }
         }
-    },
-);
+    }
+});
 
 global.Hydro.model.setting = {
     langRange,

packages/hydrooj/src/model/token.ts

@@ -1,6 +1,7 @@
 import { Collection } from 'mongodb';
 import { TokenDoc } from '../interface';
 import * as db from '../service/db';
+import * as bus from '../service/bus';
 
 export const coll: Collection<TokenDoc> = db.collection('token');
 
@@ -87,7 +88,7 @@ export function delByUid(uid: number) {
     return coll.deleteMany({ uid });
 }
 
-global.Hydro.postInit.push(ensureIndexes);
+bus.once('app/started', ensureIndexes);
 global.Hydro.model.token = {
     coll,
 

packages/hydrooj/src/model/user.ts

@@ -5,9 +5,10 @@ import * as setting from './setting';
 import * as domain from './domain';
 import { BUILTIN_USERS, PRIV } from './builtin';
 import { UserNotFoundError, UserAlreadyExistError, LoginError } from '../error';
+import { User as _User, Udoc, Udict } from '../interface';
 import pwhash from '../lib/hash.hydro';
 import * as db from '../service/db';
-import { User as _User, Udoc, Udict } from '../interface';
+import * as bus from '../service/bus';
 
 const coll: Collection<Udoc> = db.collection('user');
 
@@ -248,7 +249,7 @@ function ensureIndexes() {
     ]);
 }
 
-global.Hydro.postInit.push(ensureIndexes);
+bus.once('app/started', ensureIndexes);
 global.Hydro.model.user = {
     User,
     create,

packages/hydrooj/src/service/bus.ts

@@ -1,41 +1,122 @@
+/* eslint-disable no-await-in-loop */
 import cluster from 'cluster';
+import { Db } from 'mongodb';
+import { Mdoc, Rdoc, User } from '../interface';
 
-const bus = {};
+const _hooks: Record<keyof any, Array<(...args: any[]) => any>> = {};
+const _disposables = [];
 
-export function subscribe(events: string[], handler: any) {
-    const id = String.random(16);
-    for (const event of events) {
-        if (!bus[event]) bus[event] = {};
-        bus[event][id] = handler;
-    }
-    return id;
+function isBailed(value: any) {
+    return value !== null && value !== false && value !== undefined;
 }
 
-export function unsubscribe(events: string[], id: string) {
-    for (const event of events) {
-        if (!bus[event]) bus[event] = {};
-        delete bus[event][id];
-    }
+export type Disposable = () => void
+
+interface EventMap {
+    'app/started': () => void
+    'app/exit': () => Promise<void> | void
+    'dispose': () => void
+
+    'database/connect': (db: Db) => void
+
+    'user/message': (uid: number, mdoc: Mdoc, udoc: User) => void
+
+    'document/add': (doc: any) => string | void
+
+    'record/change': (rdoc: Rdoc, $set?: any, $push?: any) => void
 }
 
-export function publish(event: string, payload: any, isMaster = true) {
+function getHooks<K extends keyof EventMap>(name: K) {
+    const hooks = _hooks[name] || (_hooks[name] = []);
+    if (hooks.length >= 128) {
+        console.warn(
+            'max listener count (128) for event "%s" exceeded, which may be caused by a memory leak',
+            name,
+        );
+    }
+    return hooks;
+}
+
+export function removeListener<K extends keyof EventMap>(name: K, listener: EventMap[K]) {
+    const index = (_hooks[name] || []).findIndex((callback) => callback === listener);
+    if (index >= 0) {
+        _hooks[name].splice(index, 1);
+        return true;
+    }
+    return false;
+}
+
+export function addListener<K extends keyof EventMap>(name: K, listener: EventMap[K]) {
+    getHooks(name).push(listener);
+    const dispose = () => removeListener(name, listener);
+    _disposables.push(name === 'dispose' ? listener as Disposable : dispose);
+    return dispose;
+}
+
+export function prependListener<K extends keyof EventMap>(name: K, listener: EventMap[K]) {
+    getHooks(name).unshift(listener);
+    const dispose = () => removeListener(name, listener);
+    _disposables.push(name === 'dispose' ? listener as Disposable : dispose);
+    return dispose;
+}
+
+export function once<K extends keyof EventMap>(name: K, listener: EventMap[K]) {
+    // @ts-ignore
+    const dispose = addListener(name, (...args: Parameters<EventMap[K]>) => {
+        dispose();
+        // @ts-ignore
+        return listener.apply(this, args);
+    });
+    return dispose;
+}
+
+export function on<K extends keyof EventMap>(name: K, listener: EventMap[K]) {
+    return addListener(name, listener);
+}
+
+export function off<K extends keyof EventMap>(name: K, listener: EventMap[K]) {
+    return removeListener(name, listener);
+}
+
+export async function parallel<K extends keyof EventMap>(name: K, ...args: Parameters<EventMap[K]>): Promise<void> {
+    const tasks: Promise<any>[] = [];
+    for (const callback of _hooks[name] || []) {
+        tasks.push(callback.apply(this, args));
+    }
+    await Promise.all(tasks);
+}
+
+export function emit<K extends keyof EventMap>(name: K, ...args: Parameters<EventMap[K]>) {
+    return parallel(name, ...args);
+}
+
+export async function serial<K extends keyof EventMap>(name: K, ...args: Parameters<EventMap[K]>): Promise<ReturnType<EventMap[K]>> {
+    for (const callback of _hooks[name] || []) {
+        const result = await callback.apply(this, args);
+        if (isBailed(result)) return result;
+    }
+    return null;
+}
+
+export function bail<K extends keyof EventMap>(name: K, ...args: Parameters<EventMap[K]>): ReturnType<EventMap[K]> {
+    for (const callback of _hooks[name] || []) {
+        const result = callback.apply(this, args);
+        if (isBailed(result)) return result;
+    }
+    return null;
+}
+
+export function boardcast<K extends keyof EventMap>(event: K, ...payload: Parameters<EventMap[K]>) {
     // Process forked by pm2 would also have process.send
-    if (isMaster && process.send && !cluster.isMaster) {
+    if (process.send && !cluster.isMaster) {
         process.send({
             event: 'bus',
             eventName: event,
             payload,
         });
-    } else {
-        if (!bus[event]) bus[event] = {};
-        const funcs = Object.keys(bus[event]);
-        Promise.all(funcs.map((func) => bus[event][func]({
-            event,
-            value: payload,
-        })));
-    }
+    } else parallel(event, ...payload);
 }
 
 global.Hydro.service.bus = {
-    subscribe, unsubscribe, publish,
+    addListener, bail, boardcast, emit, on, off, once, parallel, prependListener, removeListener, serial,
 };

packages/hydrooj/src/service/db.ts

@@ -21,7 +21,7 @@ mongodb.MongoClient.connect(mongourl, { useNewUrlParser: true, useUnifiedTopolog
             .then((Client1) => {
                 db2 = Client1.db(opts.name);
                 global.Hydro.service.db.db2 = db2;
-                bus.publish('system_database_connected', null);
+                bus.parallel('database/connect', db);
             });
     });
 

packages/hydrooj/src/service/monitor.ts

@@ -1,5 +1,6 @@
 import cluster from 'cluster';
 import * as db from './db';
+import * as bus from './bus';
 import * as sysinfo from '../lib/sysinfo';
 
 const coll = db.collection('status');
@@ -24,17 +25,15 @@ export function updateJudger(args) {
 }
 
 if (cluster.isMaster) {
-    global.Hydro.postInit.push(
-        async () => {
-            const info = await sysinfo.get();
-            await coll.updateOne(
-                { mid: info.mid, type: 'server' },
-                { $set: { ...info, updateAt: new Date(), type: 'server' } },
-                { upsert: true },
-            );
-            setInterval(update, 60 * 1000);
-        },
-    );
+    bus.on('app/started', async () => {
+        const info = await sysinfo.get();
+        await coll.updateOne(
+            { mid: info.mid, type: 'server' },
+            { $set: { ...info, updateAt: new Date(), type: 'server' } },
+            { upsert: true },
+        );
+        setInterval(update, 60 * 1000);
+    });
 }
 
 global.Hydro.service.monitor = { update, updateJudger };

packages/login-with-qq/package.json

@@ -1,6 +1,6 @@
 {
     "name": "@hydrooj/login-with-qq",
-    "version": "0.0.3",
+    "version": "0.0.4",
     "main": "package.json",
     "repository": "git@github.com:hydro-dev/Hydro.git",
     "author": "undefined <masnn0@outlook.com>",
@@ -10,7 +10,7 @@
         "build": "tsc"
     },
     "peerDependencies": {
-        "hydrooj": "^2.12.1||^2.13.0"
+        "hydrooj": "^2.13.1"
     },
     "dependencies": {
         "ws": "^7.3.1"

packages/login-with-qq/service.ts

@@ -1,4 +1,5 @@
 import { } from 'hydrooj';
+import * as bus from 'hydrooj/dist/service/bus';
 import WebSocket from 'ws';
 
 export async function postInit() {
@@ -32,4 +33,4 @@ export async function postInit() {
     });
 }
 
-global.Hydro.postInit.push(postInit);
+bus.subscribe(['app/started'], postInit);