core: update hook usage

packages/hydrooj/src/entry/common.ts

@@ -46,7 +46,7 @@ const getLoader = (type: string, filename: string) => async function loader(pend
             }
         }
     }
-    await bus.serial(`app/load/${type}`);
+    await bus.parallel(`app/load/${type}`);
 };
 
 export const handler = getLoader('handler', 'handler');
@@ -76,7 +76,7 @@ export async function locale(pending: string[], fail: string[]) {
             }
         }
     }));
-    await bus.serial('app/load/locale');
+    await bus.parallel('app/load/locale');
 }
 
 export async function setting(pending: string[], fail: string[], modelSetting: typeof import('../model/setting')) {
@@ -117,7 +117,7 @@ export async function setting(pending: string[], fail: string[], modelSetting: t
             }
         }
     }
-    await bus.serial('app/load/setting');
+    await bus.parallel('app/load/setting');
 }
 
 export async function template(pending: string[], fail: string[]) {
@@ -138,5 +138,5 @@ export async function template(pending: string[], fail: string[]) {
             }
         }
     }
-    await bus.serial('app/load/template');
+    await bus.parallel('app/load/template');
 }

packages/hydrooj/src/entry/worker.ts

@@ -6,7 +6,6 @@ import cac from 'cac';
 import fs from 'fs-extra';
 import { Context } from '../context';
 import { Logger } from '../logger';
-import * as bus from '../service/bus';
 import db from '../service/db';
 import {
     addon, handler, lib, locale, model,
@@ -61,7 +60,7 @@ export async function apply(ctx: Context) {
     for (const i in global.Hydro.handler) await global.Hydro.handler[i]();
     require('../script/index');
     await script(pending, fail, ctx);
-    await bus.serial('app/started');
+    await ctx.parallel('app/started');
     if (process.env.NODE_APP_INSTANCE === '0') {
         const scripts = require('../upgrade').default;
         let dbVer = (await modelSystem.get('db.ver')) ?? 0;
@@ -82,6 +81,6 @@ export async function apply(ctx: Context) {
     }
     logger.success('Server started');
     process.send?.('ready');
-    await bus.serial('app/ready');
+    await ctx.parallel('app/ready');
     return { fail };
 }

packages/hydrooj/src/handler/api.ts

@@ -1,6 +1,8 @@
 import { makeExecutableSchema } from '@graphql-tools/schema';
 import { graphql, GraphQLSchema } from 'graphql';
 import { resolvers, typeDefs } from 'graphql-scalars';
+import { debounce } from 'lodash';
+import { Context as PluginContext } from '../context';
 import * as bus from '../service/bus';
 import { Handler } from '../service/server';
 
@@ -90,10 +92,6 @@ export function rebuild() {
         console.error(e);
     }
 }
-bus.on('app/started', () => {
-    rebuild();
-    bus.on('api/update', rebuild);
-});
 
 class ApiHandler extends Handler {
     query(q: string, variables: any) {
@@ -124,6 +122,8 @@ class ApiHandler extends Handler {
 }
 
 export const sideEffect = true;
-export function apply(ctx) {
+export function apply(ctx: PluginContext) {
     ctx.Route('api', '/api', ApiHandler);
+    ctx.on('ready', rebuild);
+    ctx.on('api/update', debounce(rebuild));
 }

packages/hydrooj/src/handler/judge.ts

@@ -112,7 +112,7 @@ export async function postJudge(rdoc: RecordDoc) {
             }
         }
     }
-    await bus.serial('record/judge', rdoc, updated);
+    await bus.parallel('record/judge', rdoc, updated);
 }
 
 export async function end(body: Partial<JudgeResultBody>) {

packages/hydrooj/src/handler/problem.ts

@@ -171,7 +171,7 @@ export class ProblemMainHandler extends Handler {
             });
             sort = result.hits;
         }
-        await bus.serial('problem/list', query, this);
+        await bus.parallel('problem/list', query, this);
         // eslint-disable-next-line prefer-const
         let [pdocs, ppcount, pcount] = fail
             ? [[], 0, 0]
@@ -304,7 +304,7 @@ export class ProblemRandomHandler extends Handler {
             .map((i) => i.split('category:')[1]?.split(',')));
         const q = buildQuery(this.user);
         if (category.length) q.$and = category.map((tag) => ({ tag }));
-        await bus.serial('problem/list', q, this);
+        await bus.parallel('problem/list', q, this);
         const pid = await problem.random(domainId, q);
         if (!pid) throw new NoProblemError();
         this.response.body = { pid };
@@ -360,7 +360,7 @@ export class ProblemDetailHandler extends ContestDetailBaseHandler {
             if (this.domain.langs) t.push(this.domain.langs.split(',').map((i) => i.trim()).filter((i) => i));
             this.pdoc.config.langs = intersection(baseLangs, ...t);
         }
-        await bus.serial('problem/get', this.pdoc, this);
+        await bus.parallel('problem/get', this.pdoc, this);
         [this.psdoc, this.udoc] = await Promise.all([
             problem.getStatus(domainId, this.pdoc.docId, this.user._id),
             user.getById(domainId, this.pdoc.owner),
@@ -846,7 +846,7 @@ export class ProblemSolutionHandler extends ProblemDetailHandler {
         this.response.body = {
             psdocs, page, pcount, pscount, udict, pssdict, pdoc: this.pdoc, sid,
         };
-        await bus.serial('handler/solution/get', this);
+        await bus.parallel('handler/solution/get', this);
     }
 
     @param('content', Types.Content)

packages/hydrooj/src/handler/status.ts

@@ -70,7 +70,7 @@ class StatusUpdateHandler extends Handler {
     }
 }
 
-bus.once('app/started', () => coll.createIndex('updateAt', { expireAfterSeconds: 24 * 3600 }));
+bus.on('app/started', () => coll.createIndex('updateAt', { expireAfterSeconds: 24 * 3600 }));
 
 export async function apply(ctx) {
     ctx.Route('status', '/status', StatusHandler);

packages/hydrooj/src/handler/training.ts

@@ -56,7 +56,7 @@ class TrainingMainHandler extends Handler {
     @param('page', Types.PositiveInt, true)
     async get(domainId: string, page = 1) {
         const query: FilterQuery<TrainingDoc> = {};
-        await bus.serial('training/list', query, this);
+        await bus.parallel('training/list', query, this);
         const [tdocs, tpcount] = await paginate(
             training.getMulti(domainId),
             page,
@@ -93,7 +93,7 @@ class TrainingDetailHandler extends Handler {
     @param('tid', Types.ObjectID)
     async get(domainId: string, tid: ObjectID) {
         const tdoc = await training.get(domainId, tid);
-        await bus.serial('training/get', tdoc, this);
+        await bus.parallel('training/get', tdoc, this);
         const pids = training.getPids(tdoc.dag);
         const canViewHidden = this.user.hasPerm(PERM.PERM_VIEW_PROBLEM_HIDDEN) || this.user._id;
         const [owner, pdict] = await Promise.all([

packages/hydrooj/src/model/blacklist.ts

@@ -32,7 +32,7 @@ class BlackListModel {
     }
 }
 
-bus.once('app/started', () => db.ensureIndexes(
+bus.on('app/started', () => db.ensureIndexes(
     coll,
     { key: { expireAt: -1 }, name: 'expire', expireAfterSeconds: 0 },
 ));

packages/hydrooj/src/model/blog.ts

@@ -18,7 +18,7 @@ export async function add(
         updateAt: new Date(),
         views: 0,
     };
-    await bus.serial('blog/before-add', payload);
+    await bus.parallel('blog/before-add', payload);
     const res = await document.add(
         'system', payload.content!, payload.owner!, document.TYPE_BLOG,
         null, null, null, omit(payload, ['domainId', 'content', 'owner']),

packages/hydrooj/src/model/contest.ts

@@ -490,11 +490,11 @@ export async function add(
         content, owner, title, rule, beginAt, endAt, pids, attend: 0,
     });
     RULES[rule].check(data);
-    await bus.serial('contest/before-add', data);
+    await bus.parallel('contest/before-add', data);
     const res = await document.add(domainId, content, owner, document.TYPE_CONTEST, null, null, null, {
         ...data, title, rule, beginAt, endAt, pids, attend: 0, rated,
     });
-    await bus.serial('contest/add', data, res);
+    await bus.parallel('contest/add', data, res);
     return res;
 }
 

packages/hydrooj/src/model/discussion.ts

@@ -53,13 +53,13 @@ export async function add(
         views: 0,
         sort: 100,
     };
-    await bus.serial('discussion/before-add', payload);
+    await bus.parallel('discussion/before-add', payload);
     const res = await document.add(
         payload.domainId!, payload.content!, payload.owner!, document.TYPE_DISCUSSION,
         null, payload.parentType, payload.parentId, omit(payload, ['domainId', 'content', 'owner', 'parentType', 'parentId']),
     );
     payload.docId = res;
-    await bus.emit('discussion/add', payload);
+    await bus.parallel('discussion/add', payload);
     return payload.docId;
 }
 
@@ -291,7 +291,7 @@ async function updateSort() {
 }
 TaskModel.Worker.addHandler('discussion.sort', updateSort);
 
-bus.once('app/started', async () => {
+bus.on('app/ready', async () => {
     if (!await TaskModel.count({ type: 'schedule', subType: 'discussion.sort' })) {
         await TaskModel.add({
             type: 'schedule',

packages/hydrooj/src/model/document.ts

@@ -80,7 +80,7 @@ export async function add(
         doc.parentType = parentType;
         doc.parentId = parentId;
     }
-    await bus.serial('document/add', doc);
+    await bus.parallel('document/add', doc);
     const res = await coll.insertOne(doc);
     return docId || res.insertedId;
 }
@@ -405,7 +405,7 @@ export async function revSetStatus<T extends keyof DocStatusType>(
     return res.value;
 }
 
-bus.once('app/started', async () => {
+bus.on('app/started', async () => {
     await db.ensureIndexes(
         coll,
         { key: { domainId: 1, docType: 1, docId: 1 }, name: 'basic', unique: true },

packages/hydrooj/src/model/domain.ts

@@ -52,7 +52,7 @@ class DomainModel {
             roles: {},
             avatar: '',
         };
-        await bus.serial('domain/create', ddoc);
+        await bus.parallel('domain/create', ddoc);
         await coll.insertOne(ddoc);
         await DomainModel.setUserRole(domainId, owner, 'root');
         return domainId;
@@ -61,18 +61,18 @@ class DomainModel {
     @ArgMethod
     static async get(domainId: string): Promise<DomainDoc | null> {
         const query: FilterQuery<DomainDoc> = { lower: domainId.toLowerCase() };
-        await bus.serial('domain/before-get', query);
+        await bus.parallel('domain/before-get', query);
         const result = await coll.findOne(query);
-        if (result) await bus.serial('domain/get', result);
+        if (result) await bus.parallel('domain/get', result);
         return result;
     }
 
     @ArgMethod
     static async getByHost(host: string): Promise<DomainDoc | null> {
         const query: FilterQuery<DomainDoc> = { host };
-        await bus.serial('domain/before-get', query);
+        await bus.parallel('domain/before-get', query);
         const result = await coll.findOne(query);
-        if (result) await bus.serial('domain/get', result);
+        if (result) await bus.parallel('domain/get', result);
         return result;
     }
 
@@ -81,9 +81,9 @@ class DomainModel {
     }
 
     static async edit(domainId: string, $set: Partial<DomainDoc>) {
-        await bus.serial('domain/before-update', domainId, $set);
+        await bus.parallel('domain/before-update', domainId, $set);
         const result = await coll.findOneAndUpdate({ _id: domainId }, { $set }, { returnDocument: 'after' });
-        if (result.value) await bus.serial('domain/update', domainId, $set, result.value);
+        if (result.value) await bus.parallel('domain/update', domainId, $set, result.value);
         return result.value;
     }
 
@@ -274,7 +274,7 @@ class DomainModel {
     }
 }
 
-bus.once('app/started', async () => {
+bus.on('app/started', async () => {
     await db.ensureIndexes(
         coll,
         { key: { lower: 1 }, name: 'lower', unique: true },

packages/hydrooj/src/model/message.ts

@@ -77,7 +77,7 @@ class MessageModel {
     }
 }
 
-bus.once('app/started', () => db.ensureIndexes(
+bus.on('app/started', () => db.ensureIndexes(
     coll,
     { key: { to: 1, _id: -1 }, name: 'to' },
     { key: { from: 1, _id: -1 }, name: 'from' },

packages/hydrooj/src/model/opcount.ts

@@ -21,7 +21,7 @@ export async function inc(op: string, ident: string, periodSecs: number, maxOper
     }
 }
 
-bus.once('app/started', () => db.ensureIndexes(
+bus.on('app/started', () => db.ensureIndexes(
     coll,
     { key: { expireAt: -1 }, name: 'expire', expireAfterSeconds: 0 },
     { key: { op: 1, ident: 1, expireAt: 1 }, name: 'unique', unique: true },

packages/hydrooj/src/model/oplog.ts

@@ -19,7 +19,7 @@ export async function add(data: Partial<OplogDoc> & { type: string }): Promise<O
 export async function log<T extends Handler>(handler: T, type: string, data: any) {
     const args = { ...handler.args };
     delete args.password;
-    await bus.serial('oplog/log', type, handler, args, data);
+    await bus.parallel('oplog/log', type, handler, args, data);
     const res = await coll.insertOne({
         ...data,
         _id: new ObjectID(),

packages/hydrooj/src/model/problem.ts

@@ -115,7 +115,7 @@ export class ProblemModel {
             title, tag, hidden, assign, nSubmit: 0, nAccept: 0, sort: sortable(pid || `P${docId}`),
         };
         if (pid) args.pid = pid;
-        await bus.serial('problem/before-add', domainId, content, owner, docId, args);
+        await bus.parallel('problem/before-add', domainId, content, owner, docId, args);
         const result = await document.add(domainId, content, owner, document.TYPE_PROBLEM, docId, null, null, args);
         args.content = content;
         args.owner = owner;
@@ -193,7 +193,7 @@ export class ProblemModel {
         } else if ($set.pid) {
             $set.sort = sortable($set.pid);
         }
-        await bus.serial('problem/before-edit', $set);
+        await bus.parallel('problem/before-edit', $set);
         const result = await document.set(domainId, document.TYPE_PROBLEM, _id, $set, delpid ? { pid: '' } : undefined);
         await bus.emit('problem/edit', result);
         return result;
@@ -230,7 +230,7 @@ export class ProblemModel {
     }
 
     static async del(domainId: string, docId: number) {
-        await bus.serial('problem/before-del', domainId, docId);
+        await bus.parallel('problem/before-del', domainId, docId);
         const res = await Promise.all([
             document.deleteOne(domainId, document.TYPE_PROBLEM, docId),
             document.deleteMultiStatus(domainId, document.TYPE_PROBLEM, { docId }),

packages/hydrooj/src/model/record.ts

@@ -234,7 +234,7 @@ class RecordModel {
 bus.on('problem/delete', (domainId, docId) => RecordModel.coll.updateMany({ domainId, pid: docId }, { $set: { pid: -1 } }));
 bus.on('domain/delete', (domainId) => RecordModel.coll.deleteMany({ domainId }));
 
-bus.once('app/started', () => db.ensureIndexes(
+bus.on('app/started', () => db.ensureIndexes(
     RecordModel.coll,
     { key: { domainId: 1, contest: 1, _id: -1 }, name: 'basic' },
     { key: { domainId: 1, contest: 1, uid: 1, _id: -1 }, name: 'withUser' },

packages/hydrooj/src/model/setting.ts

@@ -240,7 +240,7 @@ SystemSetting(
 // eslint-disable-next-line import/no-mutable-exports
 export let langs = {};
 
-bus.once('app/started', async () => {
+bus.on('app/started', async () => {
     logger.debug('Ensuring settings');
     const system = global.Hydro.model.system;
     for (const setting of SYSTEM_SETTINGS) {

packages/hydrooj/src/model/storage.ts

@@ -136,7 +136,7 @@ async function cleanFiles() {
     }
 }
 TaskModel.Worker.addHandler('storage.prune', cleanFiles);
-bus.once('app/started', async () => {
+bus.on('app/started', async () => {
     if (process.env.NODE_APP_INSTANCE !== '0') return;
     await db.ensureIndexes(
         StorageModel.coll,

packages/hydrooj/src/model/task.ts

@@ -148,10 +148,10 @@ Worker.addHandler('task.daily', async () => {
     if (global.Hydro.model.system.get('server.checkUpdate') && !(new Date().getDay() % 3)) {
         await global.Hydro.script.checkUpdate?.run({}, new Logger('task/checkUpdate').debug);
     }
-    await bus.serial('task/daily');
+    await bus.parallel('task/daily');
 });
 bus.on('domain/delete', (domainId) => coll.deleteMany({ domainId }));
-bus.once('app/started', async () => {
+bus.on('app/started', async () => {
     if (process.env.NODE_APP_INSTANCE !== '0') return;
     if (!await TaskModel.count({ type: 'schedule', subType: 'task.daily' })) {
         await TaskModel.add({

packages/hydrooj/src/model/token.ts

@@ -95,7 +95,7 @@ class TokenModel {
     }
 }
 
-bus.once('app/started', () => db.ensureIndexes(
+bus.on('app/started', () => db.ensureIndexes(
     TokenModel.coll,
     { key: { uid: 1, tokenType: 1, updateAt: -1 }, name: 'basic', sparse: true },
     { key: { expireAt: -1 }, name: 'expire', expireAfterSeconds: 0 },

packages/hydrooj/src/model/user.ts

@@ -102,7 +102,7 @@ export class User {
     }
 
     async init() {
-        await bus.serial('user/get', this);
+        await bus.parallel('user/get', this);
         return this;
     }
 
@@ -403,7 +403,7 @@ class UserModel {
     }
 }
 
-bus.once('app/started', () => Promise.all([
+bus.on('app/started', () => Promise.all([
     db.ensureIndexes(
         coll,
         { key: { unameLower: 1 }, name: 'uname', unique: true },

packages/hydrooj/src/service/bus.ts

@@ -119,9 +119,10 @@ export const on = (a, b, c?) => app.on(a, b, c);
 export const off = (a, b) => app.off(a, b);
 export const once = (a, b, c?) => app.once(a, b, c);
 export const parallel = (a, ...b) => app.parallel(a, ...b);
-export const emit = (a, ...b) => app.emit(a, ...b);
+export const emit = (a, ...b) => app.parallel(a, ...b);
 export const bail = (a, ...b) => app.bail(a, ...b);
-export const serial: any = (a, ...b) => app.serial(a, ...b);
+// For backward compatibility
+export const serial: any = (a, ...b) => app.parallel(a, ...b);
 export const broadcast = (a, ...b) => app.broadcast(a, ...b);
 
 global.Hydro.service.bus = app as any;

packages/hydrooj/src/service/monitor.ts

@@ -72,7 +72,7 @@ export async function update() {
         updateAt: new Date(),
         reqCount: 0,
     };
-    await bus.serial('monitor/update', 'server', $set);
+    await bus.parallel('monitor/update', 'server', $set);
     await coll.updateOne(
         { mid, type: 'server' },
         { $set },
@@ -82,7 +82,7 @@ export async function update() {
 
 export async function updateJudge(args) {
     const $set = { ...args, updateAt: new Date() };
-    await bus.serial('monitor/update', 'judge', $set);
+    await bus.parallel('monitor/update', 'judge', $set);
     return await coll.updateOne(
         { mid: args.mid, type: 'judge' },
         { $set },

packages/hydrooj/src/service/server.ts

@@ -228,7 +228,7 @@ async function handle(ctx: KoaContext, HandlerClass, checker) {
             ? `_${ctx.request.body.operation}`.replace(/_([a-z])/gm, (s) => s[1].toUpperCase())
             : '';
 
-        await bus.serial('handler/create', h);
+        await bus.parallel('handler/create', h);
 
         if (checker) checker.call(h);
         if (method === 'post') {
@@ -357,7 +357,7 @@ export function Connection(
             args, request, response, user, domain, UiContext,
         } = ctx.HydroContext;
         const h = new RouteConnHandler(ctx, args, request, response, user, domain, UiContext);
-        await bus.emit('connection/create', h);
+        await bus.parallel('connection/create', h);
         ctx.handler = h;
         h.conn = conn;
         try {
@@ -373,7 +373,7 @@ export function Connection(
                 bus.emit('connection/close', h);
                 h.cleanup?.(args);
             };
-            bus.emit('connection/active', h);
+            await bus.parallel('connection/active', h);
         } catch (e) {
             await h.onerror(e);
         }
@@ -468,7 +468,7 @@ ${ctx.response.status} ${endTime - startTime}ms ${ctx.response.length}`);
         socket.close();
     });
     const port = system.get('server.port');
-    pluginContext.on('ready', async () => {
+    pluginContext.on('app/ready', async () => {
         await new Promise((r) => {
             httpServer.listen(argv.options.port || port, () => {
                 logger.success('Server listening at: %d', argv.options.port || port);

packages/hydrooj/src/service/watcher.ts

@@ -7,7 +7,6 @@ import { debounce } from 'lodash';
 import { Context, Runtime, Service } from '../context';
 import { Logger } from '../logger';
 import { unwrapExports } from '../utils';
-import * as bus from './bus';
 
 function loadDependencies(filename: string, ignored: Set<string>) {
     const dependencies = new Set<string>();
@@ -60,20 +59,20 @@ export default class Watcher extends Service {
 
         this.watcher.on('change', (path) => {
             logger.debug('change detected:', relative(this.root, path));
-            bus.emit('app/watch/change', path);
+            this.ctx.emit('app/watch/change', path);
 
             if (this.externals.has(path)) {
                 logger.warn('Require full reload');
             } else if (require.cache[path]) {
                 this.stashed.add(path);
-                bus.emit('app/before-reload', this.stashed);
+                this.ctx.emit('app/before-reload', this.stashed);
                 triggerLocalReload();
-                bus.emit('app/reload', this.stashed);
+                this.ctx.emit('app/reload', this.stashed);
             }
         });
         this.watcher.on('unlink', (path) => {
             logger.debug('change detected:', `-${relative(this.root, path)}`);
-            bus.emit('app/watch/unlink', path);
+            this.ctx.emit('app/watch/unlink', path);
         });
     }