core: use cordis to manage app lifecycle

packages/hydrooj/package.json

@@ -17,6 +17,7 @@
         "adm-zip": "0.5.5",
         "cac": "^6.7.14",
         "cookies": "^0.8.0",
+        "cordis": "^2.2.0",
         "detect-browser": "^5.3.0",
         "emoji-regex": "^10.1.0",
         "emojis-list": "2.1.0",

packages/hydrooj/src/context.ts

@@ -0,0 +1,75 @@
+import * as cordis from 'cordis';
+import Schema from 'schemastery';
+import { inject } from './lib/ui';
+import { Loader, ModuleInterfaces } from './loader';
+import { EventMap } from './service/bus';
+
+export interface Events<C extends Context = Context> extends cordis.Events<C>, EventMap {
+}
+
+function addScript<K>(name: string, description: string, validate: Schema<K>, run: (args: K, report: any) => boolean | Promise<boolean>) {
+    if (global.Hydro.script[name]) throw new Error(`duplicate script ${name} registered.`);
+    global.Hydro.script[name] = { description, validate, run };
+    return () => delete global.Hydro.script[name];
+}
+
+function provideModule<T extends keyof ModuleInterfaces>(type: T, id: string, module: ModuleInterfaces[T]) {
+    if (global.Hydro.module[type][id]) throw new Error(`duplicate script ${type}/${id} registered.`);
+    global.Hydro.module[type as any][id] = module;
+    return () => delete global.Hydro.module[type][id];
+}
+
+export interface Context {
+    [Context.events]: Events<this>;
+    options: Context.Config;
+    loader: Loader;
+    Route: typeof import('./service/server').Route;
+    Connection: typeof import('./service/server').Connection;
+    setTimeout: typeof setTimeout;
+    setInterval: typeof setInterval;
+    setImmediate: typeof setImmediate;
+    addScript: typeof addScript;
+    provideModule: typeof provideModule;
+    inject: typeof inject;
+    api: ApiMixin;
+    broadcast: Context['emit'];
+}
+
+export class Context extends cordis.Context {
+    static readonly session = Symbol('session');
+}
+
+export namespace Context {
+    export interface Config extends cordis.Context.Config { }
+}
+
+export type Runtime = cordis.Runtime<Context>;
+export type State = cordis.State<Context>;
+export type Fork = cordis.Fork<Context>;
+export type Plugin = cordis.Plugin<Context>;
+export const Service = cordis.Service<Context>;
+export namespace Plugin {
+    export type Function<T = any> = cordis.Plugin.Function<T, Context>;
+    export type Constructor<T = any> = cordis.Plugin.Constructor<T, Context>;
+    export type Object<S = any, T = any> = cordis.Plugin.Object<S, T, Context>;
+}
+
+const T = <F extends (...args: any[]) => any>(origFunc: F, disposeFunc?) =>
+    function method(this: cordis.Service, ...args: Parameters<F>) {
+        const res = origFunc(...args);
+        this.caller?.on('dispose', () => (disposeFunc ? disposeFunc(res) : res()));
+    };
+export class ApiMixin extends Service {
+    static readonly methods = ['addScript', 'setInterval', 'setTimeout', 'setImmediate', 'provideModule', 'inject', 'broadcast'];
+    addScript = T(addScript);
+    setInterval = T(setInterval, clearInterval);
+    setTimeout = T(setTimeout, clearTimeout);
+    setImmediate = T(setImmediate, clearImmediate);
+    provideModule = T(provideModule);
+    inject = T(inject);
+    broadcast = (event: string, ...payload) => this.ctx.emit('bus/broadcast', event, payload);
+    constructor(ctx) {
+        super(ctx, 'api', true);
+    }
+}
+Context.service('api', ApiMixin);

packages/hydrooj/src/entry/cli.ts

@@ -2,8 +2,6 @@
 /* eslint-disable no-await-in-loop */
 import cac from 'cac';
 import { ObjectID } from 'mongodb';
-import { validate } from '../lib/validator';
-import options from '../options';
 import db from '../service/db';
 import {
     lib, model, script, service,
@@ -24,12 +22,7 @@ function parseParameters(fn: Function) {
 async function runScript(name: string, arg: any) {
     const s = global.Hydro.script[name];
     if (!s) return console.error('Script %s not found.', name);
-    if (typeof s.validate === 'function') {
+    if (typeof s.validate === 'function') arg = s.validate(arg);
-        arg = s.validate(arg);
-    } else {
-        console.warn('You are using the legacy script validation API, which will be dropped in the future.');
-        validate(s.validate, arg);
-    }
     return await s.run(arg, console.info);
 }
 
@@ -84,26 +77,24 @@ async function cli() {
     return console.log(result);
 }
 
-export async function load() {
+export async function load(ctx) {
     const pending = global.addons;
     const fail = [];
     require('../lib/i18n');
     require('../utils');
     require('../error');
-    require('../options');
+    await db.start();
-    const opts = options();
-    await db.start(opts);
     await require('../settings').loadConfig();
     const storage = require('../service/storage');
     await storage.loadStorageService();
     require('../lib/index');
-    await lib(pending, fail);
+    await lib(pending, fail, ctx);
     const systemModel = require('../model/system');
     await systemModel.runConfig();
-    await service(pending, fail);
+    await service(pending, fail, ctx);
     require('../model/index');
-    await model(pending, fail);
+    await model(pending, fail, ctx);
     require('../script/index');
-    await script(pending, fail);
+    await script(pending, fail, ctx);
     await cli();
 }

packages/hydrooj/src/entry/common.ts

@@ -5,10 +5,10 @@ import os from 'os';
 import path from 'path';
 import fs from 'fs-extra';
 import yaml from 'js-yaml';
+import { Context } from '../context';
 import i18n from '../lib/i18n';
 import { Logger } from '../logger';
 import * as bus from '../service/bus';
-import { Runtime } from '../service/module';
 
 const logger = new Logger('common');
 
@@ -31,17 +31,14 @@ function locateFile(basePath: string, filenames: string[]) {
     return null;
 }
 
-const getLoader = (type: string, filename: string) => async function loader(pending: string[], fail: string[]) {
+const getLoader = (type: string, filename: string) => async function loader(pending: string[], fail: string[], ctx: Context) {
     for (const i of pending) {
         const p = locateFile(i, [`${filename}.ts`, `${filename}.js`]);
         if (p && !fail.includes(i)) {
             try {
-                logger.info(`${type.replace(/^(.)/, (t) => t.toUpperCase())} init: %s`, i);
+                const m = require(p);
-                const res = require(p);
+                if (m.apply) ctx.loader.reloadPlugin(ctx, p, {});
-                if (res.apply) {
+                else logger.info(`${type.replace(/^(.)/, (t) => t.toUpperCase())} init: %s`, i);
-                    const runtime = new Runtime(p);
-                    runtime.load(res);
-                }
             } catch (e) {
                 fail.push(i);
                 logger.info(`${type.replace(/^(.)/, (t) => t.toUpperCase())} load fail: %s`, i);
@@ -57,6 +54,7 @@ export const addon = getLoader('addon', 'index');
 export const model = getLoader('model', 'model');
 export const lib = getLoader('lib', 'lib');
 export const script = getLoader('script', 'script');
+export const service = getLoader('service', 'service');
 
 export async function locale(pending: string[], fail: string[]) {
     await Promise.all(pending.map(async (i) => {
@@ -142,25 +140,3 @@ export async function template(pending: string[], fail: string[]) {
     }
     await bus.serial('app/load/template');
 }
-
-export async function service(pending: string[], fail: string[]) {
-    for (const i of pending) {
-        const p = locateFile(i, ['service.ts', 'service.js']);
-        if (p && !fail.includes(i)) {
-            try {
-                logger.info('Service init: %s', i);
-                require(p);
-            } catch (e) {
-                fail.push(i);
-                logger.error('Service Load Fail: %s', i);
-                logger.error(e);
-            }
-        }
-    }
-    for (const key in global.Hydro.service) {
-        if (key === 'server') continue;
-        const srv = global.Hydro.service[key];
-        if (!srv.started && srv.start) await srv.start();
-    }
-    await bus.serial('app/load/service');
-}

packages/hydrooj/src/entry/setup.ts

@@ -10,6 +10,7 @@ import { Logger } from '../logger';
 
 const logger = new Logger('setup');
 const listenPort = cac().parse().options.port || 8888;
+let resolve;
 
 async function get(ctx: Context) {
     ctx.body = `<!DOCTYPE html>
@@ -105,7 +106,8 @@ async function post(ctx: Context) {
         fs.writeFileSync(path.resolve(os.homedir(), '.hydro', 'config.json'), JSON.stringify({
             host, port, name, username, password,
         }));
-        ctx.body = '<h1>Done! Please restart Hydro.</h1>';
+        ctx.body = '<h1>Done! Hydro is now starting.</h1>';
+        resolve?.();
     } catch (e) {
         ctx.body = `Error connecting to database: ${e.message}\n${e.stack}`;
     }
@@ -123,4 +125,7 @@ export function load() {
         });
     server.listen(listenPort);
     logger.success('Server listening at: %d', listenPort);
+    return new Promise((r) => {
+        resolve = r;
+    });
 }

packages/hydrooj/src/entry/worker.ts

@@ -4,21 +4,20 @@ import os from 'os';
 import path from 'path';
 import cac from 'cac';
 import fs from 'fs-extra';
+import { Context } from '../context';
 import { Logger } from '../logger';
-import options from '../options';
 import * as bus from '../service/bus';
 import db from '../service/db';
-import { Runtime } from '../service/module';
 import {
     addon, handler, lib, locale, model,
     script, service, setting, template,
 } from './common';
 
+const argv = cac().parse();
 const logger = new Logger('worker');
-const { loaderDetail: detail, watch } = cac().parse().options;
 const tmpdir = path.resolve(os.tmpdir(), 'hydro');
 
-export async function load() {
+export async function apply(ctx: Context) {
     fs.ensureDirSync(tmpdir);
     require('../lib/i18n');
     require('../utils');
@@ -26,66 +25,43 @@ export async function load() {
     const config = require('../options')();
     if (!process.env.CI && !config) {
         logger.info('Starting setup');
-        return require('./setup').load();
+        await require('./setup').load();
     }
     const pending = global.addons;
     const fail = [];
-    if (detail) logger.info('start');
     await Promise.all([
         locale(pending, fail),
         template(pending, fail),
     ]);
-    if (detail) logger.info('finish: locale/template/static');
+    await db.start();
-    const opts = options();
-    await db.start(opts);
     await require('../settings').loadConfig();
-    if (detail) logger.info('finish: db.connect');
     const modelSystem = require('../model/system');
     await modelSystem.runConfig();
-    if (detail) logger.info('finish: config');
     const storage = require('../service/storage');
     await storage.loadStorageService();
-    if (detail) logger.info('finish: storage.connect');
+    await require('hydrooj/src/service/server').apply(ctx);
+    // Make sure everything is ready and then start main entry
+    if (argv.options.watch) await ctx.loader.reloadPlugin(ctx, 'hydrooj/src/service/watcher', {});
+    await ctx.root.start();
     require('../lib/index');
-    if (detail) logger.info('finish: lib.builtin');
+    await lib(pending, fail, ctx);
-    await lib(pending, fail);
-    if (detail) logger.info('finish: lib.extra');
     require('../service/monitor');
-    if (detail) logger.info('finish: monitor');
+    await service(pending, fail, ctx);
-    const server = require('../service/server');
-    await server.prepare();
-    if (detail) logger.info('finish: server');
-    if (watch) require('../service/watcher');
-    await service(pending, fail);
-    if (detail) logger.info('finish: service.extra');
     require('../model/index');
-    if (detail) logger.info('finish: model.builtin');
     const handlerDir = path.resolve(__dirname, '..', 'handler');
     const handlers = await fs.readdir(handlerDir);
     for (const h of handlers) {
-        const f = path.resolve(handlerDir, h);
+        ctx.loader.reloadPlugin(ctx, path.resolve(handlerDir, h), {}, `hydrooj/handler/${h.split('.')[0]}`);
-        const m = require(f);
-        if (m.apply) new Runtime(f).load(m);
     }
-    if (detail) logger.info('finish: handler.builtin');
+    await model(pending, fail, ctx);
-    await model(pending, fail);
-    if (detail) logger.info('finish: model.extra');
     const modelSetting = require('../model/setting');
     await setting(pending, fail, modelSetting);
-    if (detail) logger.info('finish: setting');
+    await handler(pending, fail, ctx);
-    await handler(pending, fail);
+    await addon(pending, fail, ctx);
-    await addon(pending, fail);
-    if (detail) logger.info('finish: handler.extra');
     for (const i in global.Hydro.handler) await global.Hydro.handler[i]();
-    if (detail) logger.info('finish: handler.apply');
     require('../script/index');
-    if (detail) logger.info('finish: script.builtin');
+    await script(pending, fail, ctx);
-    await script(pending, fail);
-    if (detail) logger.info('finish: script.extra');
     await bus.serial('app/started');
-    if (detail) logger.info('finish: bus.serial(start)');
-    await server.start();
-    if (detail) logger.info('finish: server.start');
     if (process.env.NODE_APP_INSTANCE === '0') {
         const scripts = require('../upgrade').default;
         let dbVer = (await modelSystem.get('db.ver')) ?? 0;

packages/hydrooj/src/handler/manage.ts

@@ -4,9 +4,7 @@ import * as yaml from 'js-yaml';
 import Schema from 'schemastery';
 import * as check from '../check';
 import { BadRequestError, ValidationError } from '../error';
-import {
+import { isEmail, isPassword, isUname } from '../lib/validator';
-    isEmail, isPassword, isUname, validate,
-} from '../lib/validator';
 import { Logger } from '../logger';
 import { PRIV, STATUS } from '../model/builtin';
 import domain from '../model/domain';
@@ -106,9 +104,6 @@ class SystemScriptHandler extends SystemHandler {
         let args = JSON.parse(raw);
         if (typeof global.Hydro.script[id].validate === 'function') {
             args = global.Hydro.script[id].validate(args);
-        } else {
-            logger.warn('You are using the legacy script validation API, which will be dropped in the future.');
-            validate(global.Hydro.script[id].validate, args);
         }
         const rid = await record.add(domainId, -1, this.user._id, '-', id, false, { input: raw, type: 'pretest' });
         const report = (data) => judge.next({ domainId, rid, ...data });

packages/hydrooj/src/init.ts

@@ -35,3 +35,4 @@ if (!global.Hydro) {
     };
     global.addons = [];
 }
+global.app = new (require('./context').Context)();

packages/hydrooj/src/interface.ts

@@ -2,6 +2,7 @@ import type fs from 'fs';
 import type { Dictionary, NumericDictionary } from 'lodash';
 import type { ItemBucketMetadata } from 'minio';
 import type { Cursor, ObjectID } from 'mongodb';
+import { Context } from './context';
 import type { ProblemDoc } from './model/problem';
 import type { Handler } from './service/server';
 
@@ -663,7 +664,8 @@ export interface Model {
 }
 
 export interface Service {
-    bus: typeof import('./service/bus'),
+    /** @deprecated */
+    bus: Context,
     db: typeof import('./service/db'),
     monitor: typeof import('./service/monitor'),
     server: typeof import('./service/server'),
@@ -687,7 +689,7 @@ export interface ProblemSearchOptions {
 
 export type ProblemSearch = (domainId: string, q: string, options?: ProblemSearchOptions) => Promise<ProblemSearchResponse>;
 
-export { Context as PluginContext } from './service/module';
+export { Context as PluginContext } from './context';
 
 export interface Lib extends Record<string, any> {
     difficulty: typeof import('./lib/difficulty'),
@@ -746,6 +748,7 @@ declare global {
             addons: string[],
         }
     }
+    var app: Context; // eslint-disable-line
     var Hydro: HydroGlobal; // eslint-disable-line
     var addons: string[]; // eslint-disable-line
 }

packages/hydrooj/src/lib/validator.ts

@@ -7,58 +7,6 @@ const RE_UNAME = /^.{3,31}$/i;
 const RE_ROLE = /^[_0-9A-Za-z]{1,31}$/i;
 const RE_MAIL = /^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+(?:\.[a-zA-Z0-9-]+)+$/i;
 
-// eslint-disable-next-line consistent-return
-function _validate(scheme: any, arg: any, base: string): void {
-    if (!(scheme instanceof Object)) return _validate({ $type: scheme }, arg, base);
-    if (scheme.$validate) {
-        const result = scheme.$validate(arg);
-        if (result === false) throw new ValidationError(base);
-    }
-    if (scheme.$group) _validate(scheme.$group, arg, base);
-    if (scheme.$or) {
-        for (const s of scheme.$or) {
-            let success = false;
-            try {
-                _validate(s, arg, base);
-                success = true;
-            } catch { }
-            // TODO beautify output
-            if (!success) throw new ValidationError(scheme.$or);
-        }
-    }
-    if (scheme.$and) for (const s of scheme.$and) _validate(s, arg, base);
-    if (scheme.$type) {
-        if (typeof scheme.$type === 'string') {
-            if (!scheme.$type.endsWith('?')) {
-                if (!typeof arg === scheme.$type) {
-                    throw new ValidationError(base);
-                }
-            } else if (arg && !scheme.$type.includes(typeof arg)) {
-                throw new ValidationError(base);
-            }
-        } else if (!(arg instanceof scheme.$type)) {
-            throw new ValidationError(base);
-        }
-    }
-    if (scheme.$eq && scheme.$eq !== arg) throw new ValidationError(base);
-    if (scheme.$not && scheme.$not === arg) throw new ValidationError(base);
-    if (scheme.$in && !(scheme.$in.includes(arg))) throw new ValidationError(base);
-    if (scheme.$nin && scheme.$nin.includes(arg)) throw new ValidationError(base);
-    for (const key in scheme) {
-        if (key[0] !== '$') {
-            if (arg instanceof Object) {
-                _validate(scheme[key], arg[key], `${base}.${key}`);
-            } else {
-                throw new ValidationError(base);
-            }
-        }
-    }
-}
-
-export function validate(scheme: any, arg: any) {
-    return _validate(scheme, arg, '');
-}
-
 export const isTitle = (s) => typeof s === 'string' && s.trim().length && s.trim().length < 64;
 export const checkTitle = (s) => { if (!(s?.trim()?.length && s.trim().length < 64)) throw new ValidationError('title'); else return s; };
 export const isDomainId = (s) => RE_DOMAINID.test(s);
@@ -83,7 +31,6 @@ export const checkIntro = (s) => { if (!isIntro()) throw new ValidationError('in
 export const parsePid = (s) => (Number.isNaN(+s) ? s : `P${s}`);
 
 global.Hydro.lib.validator = {
-    validate,
     isTitle,
     checkTitle,
     isDomainId,

packages/hydrooj/src/loader.ts

@@ -1,3 +1,5 @@
+/* eslint-disable import/no-dynamic-require */
+/* eslint-disable consistent-return */
 /* eslint-disable simple-import-sort/imports */
 /* eslint-disable no-await-in-loop */
 /* eslint-disable no-eval */
@@ -5,15 +7,18 @@ import './init';
 import './interface';
 import Schema from 'schemastery';
 import path from 'path';
+// eslint-disable-next-line import/no-duplicates
 import './utils';
 import cac from 'cac';
-import { Logger } from './logger';
 import './ui';
 
+import { Logger } from './logger';
+import { ApiMixin, Context } from './context';
+// eslint-disable-next-line import/no-duplicates
+import { unwrapExports } from './utils';
 // This is the main entry. So let's re-export some modules.
 export * from './interface';
 export { Schema, Logger };
-export { requestConfig } from './settings';
 
 const argv = cac().parse();
 const logger = new Logger('loader');
@@ -22,6 +27,66 @@ logger.debug('%o', argv);
 process.on('unhandledRejection', logger.error);
 process.on('uncaughtException', logger.error);
 
+export function resolveConfig(plugin: any, config: any) {
+    if (config === false) return;
+    if (config === true) config = undefined;
+    config ??= {};
+    const schema = plugin['Config'] || plugin['schema'];
+    if (schema && plugin['schema'] !== false) config = schema(config);
+    return config;
+}
+Context.service('loader');
+
+export class Loader {
+    static readonly kUpdate = Symbol.for('koishi.loader.update');
+    static readonly kRecord = Symbol.for('koishi.loader.record');
+
+    public app: Context;
+    public config: Context.Config;
+    public suspend = false;
+    public cache: Record<string, string> = Object.create(null);
+
+    unloadPlugin(ctx: Context, key: string) {
+        const fork = ctx.state[Loader.kRecord][key];
+        if (fork) {
+            fork.dispose();
+            delete ctx.state[Loader.kRecord][key];
+            logger.info('unload plugin %c', key);
+        }
+    }
+
+    async reloadPlugin(parent: Context, key: string, config: any, asName = '') {
+        let fork = parent.state[Loader.kRecord]?.[key];
+        if (fork) {
+            logger.info('reload plugin %c', key);
+            fork[Loader.kUpdate] = true;
+            fork.update(config);
+        } else {
+            logger.info('apply plugin %c', key);
+            const plugin = await this.resolvePlugin(key);
+            if (!plugin) return;
+            resolveConfig(plugin, config);
+            if (asName) plugin.name = asName;
+            // fork = parent.plugin(plugin, this.interpolate(config));
+            fork = parent.plugin(plugin, config);
+            if (!fork) return;
+            parent.state[Loader.kRecord] ||= Object.create(null);
+            parent.state[Loader.kRecord][key] = fork;
+        }
+        return fork;
+    }
+
+    async resolvePlugin(name: string) {
+        try {
+            this.cache[name] ||= require.resolve(name);
+        } catch (err) {
+            logger.error(err.message);
+            return;
+        }
+        return unwrapExports(require(this.cache[name]));
+    }
+}
+
 export function addon(addonPath: string, prepend = false) {
     try {
         // Is a npm package
@@ -52,15 +117,22 @@ export function addScript(name: string, description: string) {
     };
 }
 
+Context.service('loader');
+app.api = new ApiMixin(app);
+const loader = new Loader();
+app.loader = loader;
+loader.app = app;
+app.state[Loader.kRecord] = Object.create(null);
+
 export async function load() {
     addon(path.resolve(__dirname, '..'), true);
     Error.stackTraceLimit = 50;
-    require('./entry/worker').load();
+    await require('./entry/worker').apply(app);
     global.gc?.();
 }
 
 export async function loadCli() {
     process.env.HYDRO_CLI = 'true';
-    await require('./entry/cli').load();
+    await require('./entry/cli').load(app);
     setTimeout(() => process.exit(0), 300);
 }

packages/hydrooj/src/service/bus.ts

@@ -1,9 +1,9 @@
 /* eslint-disable no-await-in-loop */
-import cac from 'cac';
 import type {
     Db, FilterQuery, ObjectID, OnlyFieldsOfType,
 } from 'mongodb';
 import pm2 from '@hydrooj/utils/lib/locate-pm2';
+import { Context } from '../context';
 import type { ProblemSolutionHandler } from '../handler/problem';
 import type { UserRegisterHandler } from '../handler/user';
 import type {
@@ -11,18 +11,9 @@ import type {
     MessageDoc, ProblemDoc, RecordDoc,
     Tdoc, TrainingDoc, User,
 } from '../interface';
-import { Logger } from '../logger';
 import type { DocType } from '../model/document';
 import type { Handler } from './server';
 
-const _hooks: Record<keyof any, Array<(...args: any[]) => any>> = {};
-const logger = new Logger('bus');
-const argv = cac().parse();
-
-function isBailed(value: any) {
-    return value !== null && value !== false && value !== undefined;
-}
-
 export type Disposable = () => void;
 export type VoidReturn = Promise<any> | any;
 
@@ -104,110 +95,33 @@ export interface EventMap extends Record<string, any> {
 }
 /* eslint-enable @typescript-eslint/naming-convention */
 
-function getHooks<K extends keyof EventMap>(name: K) {
+export function apply(ctx: Context) {
-    const hooks = _hooks[name] || (_hooks[name] = []);
+    try {
-    if (hooks.length >= 2048) {
+        if (!process.send || !pm2 || process.env.exec_mode !== 'cluster_mode') throw new Error();
-        logger.warn(
+        pm2.launchBus((err, bus) => {
-            'max listener count (2048) for event "%s" exceeded, which may be caused by a memory leak',
+            if (err) throw new Error();
-            name,
+            bus.on('hydro:broadcast', (packet) => {
-        );
+                ctx.parallel(packet.data.event, ...packet.data.payload);
-    }
+            });
-    return hooks;
+            ctx.on('bus/broadcast', (event, payload) => {
-}
+                process.send({ type: 'hydro:broadcast', data: { event, payload } });
-
+            });
-export function removeListener<K extends keyof EventMap>(name: K, listener: EventMap[K]) {
+            console.debug('Using pm2 event bus');
-    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);
-    return () => removeListener(name, listener);
-}
-
-export function prependListener<K extends keyof EventMap>(name: K, listener: EventMap[K]) {
-    getHooks(name).unshift(listener);
-    return () => removeListener(name, listener);
-}
-
-export function once<K extends keyof EventMap>(name: K, listener: EventMap[K]) {
-    let dispose;
-    function _listener(...args: any[]) {
-        dispose();
-        return listener.apply(this, args);
-    }
-    _listener.toString = () => `// Once \n${listener.toString()}`;
-    dispose = addListener(name, _listener);
-    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>[] = [];
-    if (argv.options.showBus) logger.debug('parallel: %s %o', name, args);
-    for (const callback of _hooks[name] || []) {
-        if (argv.options.busDetail) logger.debug(callback.toString());
-        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<void> {
-    if (argv.options.showBus) logger.debug('serial: %s %o', name, args);
-    const hooks = Array.from(_hooks[name] || []);
-    for (const callback of hooks) {
-        if (argv.options.busDetail) logger.debug(callback.toString());
-        await callback.apply(this, args);
-    }
-}
-
-export async function bail<K extends keyof EventMap>(name: K, ...args: Parameters<EventMap[K]>): Promise<ReturnType<EventMap[K]>> {
-    if (argv.options.showBus) logger.debug('bail: %s %o', name, args);
-    const hooks = Array.from(_hooks[name] || []);
-    for (const callback of hooks) {
-        let result = callback.apply(this, args);
-        if (result instanceof Promise) result = await result;
-        if (isBailed(result)) return result;
-    }
-    return null;
-}
-
-export function broadcast<K extends keyof EventMap>(event: K, ...payload: Parameters<EventMap[K]>) {
-    return parallel('bus/broadcast', event, payload);
-}
-
-try {
-    if (!process.send || !pm2 || process.env.exec_mode !== 'cluster_mode') throw new Error();
-    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) => {
+    } catch (e) {
-            process.send({ type: 'hydro:broadcast', data: { event, payload } });
+        ctx.on('bus/broadcast', (event, payload) => ctx.parallel(event, ...payload));
-        });
+        console.debug('Using mongodb external event bus');
-        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 = {
+export default app;
-    addListener, bail, broadcast, emit, on, off, once, parallel, prependListener, removeListener, serial,
+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 bail = (a, ...b) => app.bail(a, ...b);
+export const serial: any = (a, ...b) => app.serial(a, ...b);
+export const broadcast = (a, ...b) => app.broadcast(a, ...b);
+
+global.Hydro.service.bus = app as any;

packages/hydrooj/src/service/db.ts

@@ -4,6 +4,7 @@ import {
 } from 'mongodb';
 import { BaseService, Collections } from '../interface';
 import { Logger } from '../logger';
+import options from '../options';
 import * as bus from './bus';
 
 const logger = new Logger('mongo');
@@ -33,8 +34,8 @@ class MongoService implements BaseService {
         return mongourl;
     }
 
-    async start(opts: MongoConfig) {
+    async start() {
-        opts ||= {};
+        const opts = options() || {};
         let mongourl = MongoService.buildUrl(opts);
         if (process.env.CI) {
             const { MongoMemoryServer } = require('mongodb-memory-server');
@@ -83,6 +84,11 @@ class MongoService implements BaseService {
             }
         }
     }
+
+    public async apply(ctx) {
+        await this.start();
+        ctx.on('dispose', () => this.client.close());
+    }
 }
 
 const service = new MongoService();

packages/hydrooj/src/service/module.ts

@@ -1,71 +0,0 @@
-import Schema from 'schemastery';
-import { inject } from '../lib/ui';
-import { ModuleInterfaces } from '../loader';
-import * as bus from './bus';
-
-export const runtimes: Record<string, Runtime> = Object.create(null);
-
-function addScript<K>(name: string, description: string, validate: Schema<K>, run: (args: K, report: any) => boolean | Promise<boolean>) {
-    if (global.Hydro.script[name]) throw new Error(`duplicate script ${name} registered.`);
-    global.Hydro.script[name] = { description, validate, run };
-    return () => delete global.Hydro.script[name];
-}
-
-function provideModule<T extends keyof ModuleInterfaces>(type: T, id: string, module: ModuleInterfaces[T]) {
-    if (global.Hydro.module[type][id]) throw new Error(`duplicate script ${type}/${id} registered.`);
-    global.Hydro.module[type as any][id] = module;
-    return () => delete global.Hydro.module[type][id];
-}
-
-export interface Context {
-    Route: typeof import('./server').Route;
-    Connection: typeof import('./server').Connection;
-    on: typeof bus['on'];
-    once: typeof bus['once'];
-    off: typeof bus['off'];
-    addScript: typeof addScript;
-    provideModule: typeof provideModule;
-    inject: typeof inject;
-}
-
-export class Runtime {
-    private disposables = [];
-    public sideEffect = false;
-    public loaded = false;
-
-    constructor(public filename: string) {
-        runtimes[filename] = this;
-    }
-
-    load(module: any) {
-        if (this.loaded) throw new Error(`Module ${this.filename} was already loaded`);
-        this.sideEffect = false;
-        if (typeof module.dispose === 'function') this.disposables.push(module.dispose);
-        const T = (origFunc, disposeFunc?) => (origFunc ? (...args) => {
-            const res = origFunc(...args);
-            this.disposables.push(disposeFunc ? () => disposeFunc(res) : res);
-            return res;
-        } : null);
-        module.apply({
-            Route: T(global.Hydro.service.server?.Route),
-            Connection: T(global.Hydro.service.server?.Connection),
-            on: T(bus.on),
-            once: T(bus.once),
-            off: bus.off,
-            addScript: T(addScript),
-            setInterval: T(setInterval, clearInterval),
-            setTimeout: T(setTimeout, clearTimeout),
-            setImmediate: T(setImmediate, clearImmediate),
-            provideModule: T(provideModule),
-            inject: T(inject),
-        } as Context);
-        if (module.sideEffect) this.sideEffect = true;
-        this.loaded = true;
-    }
-
-    dispose() {
-        this.disposables.map((i) => i());
-        this.disposables = [];
-        this.loaded = false;
-    }
-}

packages/hydrooj/src/service/server.ts

@@ -2,13 +2,14 @@ import http from 'http';
 import { resolve } from 'path';
 import cac from 'cac';
 import fs from 'fs-extra';
-import Koa, { Context } from 'koa';
+import Koa from 'koa';
 import Body from 'koa-body';
 import Compress from 'koa-compress';
 import proxy from 'koa-proxies';
 import cache from 'koa-static-cache';
 import WebSocket from 'ws';
 import { parseMemoryMB } from '@hydrooj/utils/lib/utils';
+import { Context, Service } from '../context';
 import {
     HydroError, InvalidOperationError, MethodNotAllowedError,
     NotFoundError, PermissionError, PrivilegeError,
@@ -377,16 +378,36 @@ export function Connection(
             await h.onerror(e);
         }
     });
+    return router.disposeLastOp;
 }
 
-let started = false;
-
 class NotFoundHandler extends Handler {
     prepare() { throw new NotFoundError(this.request.path); }
     all() { }
 }
 
-export async function prepare() {
+class RouteService extends Service {
+    static readonly methods = ['Route', 'Connection'];
+    constructor(ctx) {
+        super(ctx, 'server', true);
+    }
+
+    // eslint-disable-next-line @typescript-eslint/naming-convention
+    Route(...args: Parameters<typeof Route>) {
+        const res = Route(...args);
+        this.caller?.on('dispose', () => res());
+    }
+
+    // eslint-disable-next-line @typescript-eslint/naming-convention
+    Connection(...args: Parameters<typeof Connection>) {
+        const res = Connection(...args);
+        this.caller?.on('dispose', () => res());
+    }
+}
+
+export async function apply(pluginContext) {
+    Context.service('server', RouteService);
+    pluginContext.server = new RouteService(pluginContext);
     app.keys = system.get('session.keys') as unknown as string[];
     const proxyMiddleware = proxy('/fs', {
         target: builtinConfig.file.endPoint,
@@ -414,7 +435,7 @@ export async function prepare() {
         }));
     }
     if (process.env.DEV) {
-        app.use(async (ctx: Context, next: Function) => {
+        app.use(async (ctx: Koa.Context, next: Function) => {
             const startTime = Date.now();
             await next();
             const endTime = Date.now();
@@ -446,17 +467,17 @@ ${ctx.response.status} ${endTime - startTime}ms ${ctx.response.length}`);
         }
         socket.close();
     });
-}
-
-// TODO use postInit?
-export async function start() {
-    if (started) return;
     const port = system.get('server.port');
-    await new Promise((r) => {
+    pluginContext.on('ready', async () => {
-        httpServer.listen(argv.options.port || port, () => {
+        await new Promise((r) => {
-            logger.success('Server listening at: %d', argv.options.port || port);
+            httpServer.listen(argv.options.port || port, () => {
-            started = true;
+                logger.success('Server listening at: %d', argv.options.port || port);
-            r(true);
+                r(true);
+            });
+        });
+        pluginContext.on('dispose', () => {
+            httpServer.close();
+            wsServer.close();
         });
     });
 }
@@ -473,6 +494,5 @@ global.Hydro.service.server = {
     ConnectionHandler,
     Route,
     Connection,
-    prepare,
+    apply,
-    start,
 };

packages/hydrooj/src/service/watcher.ts

@@ -4,9 +4,10 @@
 import { relative, resolve } from 'path';
 import { FSWatcher, watch } from 'chokidar';
 import { debounce } from 'lodash';
+import { Context, Runtime, Service } from '../context';
 import { Logger } from '../logger';
+import { unwrapExports } from '../utils';
 import * as bus from './bus';
-import { Runtime, runtimes } from './module';
 
 function loadDependencies(filename: string, ignored: Set<string>) {
     const dependencies = new Set<string>();
@@ -19,13 +20,9 @@ function loadDependencies(filename: string, ignored: Set<string>) {
     return dependencies;
 }
 
-export function unwrapExports(module: any) {
-    return module?.default || module;
-}
-
 const logger = new Logger('watch');
 
-export function coerce(val: any) {
+function coerce(val: any) {
     // resolve error when stack is undefined, e.g. axios error with status code 401
     const { message, stack } = val instanceof Error && val.stack ? val : new Error(val as any);
     const lines = stack.split('\n');
@@ -33,7 +30,7 @@ export function coerce(val: any) {
     return lines.slice(index).join('\n');
 }
 
-class Watcher {
+export default class Watcher extends Service {
     private root: string;
     private watcher: FSWatcher;
     private externals: Set<string>;
@@ -41,9 +38,10 @@ class Watcher {
     private declined: Set<string>;
     private stashed = new Set<string>();
 
-    constructor(private config: any) {
+    constructor(public ctx, private config: any) {
+        super(ctx, 'watcher', true);
         this.externals = new Set(Object.keys(require.cache));
-        bus.on('app/ready', () => this.start());
+        Context.service('watcher');
     }
 
     start() {
@@ -155,16 +153,17 @@ class Watcher {
         for (const filename in require.cache) {
             const module = require.cache[filename];
             const plugin = unwrapExports(module.exports);
-            const runtime = runtimes[filename];
+            const runtime = this.ctx.registry.get(plugin);
             if (!runtime || this.declined.has(filename)) continue;
             pending.set(filename, runtime);
-            if (!runtimes[filename]?.sideEffect && !plugin?.['sideEffect']) this.declined.add(filename);
+            if (!plugin?.['sideEffect']) this.declined.add(filename);
         }
 
         for (const [filename, runtime] of pending) {
             // check if it is a dependent of the changed file
             this.declined.delete(filename);
             const dependencies = [...loadDependencies(filename, this.declined)];
+            if (!runtime.plugin['sideEffect']) this.declined.add(filename);
 
             // we only detect reloads at plugin level
             // a plugin will be reloaded if any of its dependencies are accepted
@@ -183,6 +182,9 @@ class Watcher {
                     isMarked = true;
                     break;
                 }
+                for (const state of runtime.children) {
+                    queued.push(state.runtime);
+                }
             }
             if (!isMarked) reloads.set(runtime, filename);
         }
@@ -213,16 +215,19 @@ class Watcher {
         try {
             for (const [runtime, filename] of reloads) {
                 const path = relative(this.root, filename);
+                const states = runtime.children.slice();
 
                 try {
-                    runtime.dispose();
+                    this.ctx.registry.delete(runtime.plugin);
                 } catch (err) {
                     logger.warn(`failed to dispose plugin at %c\n${coerce(err)}`, path);
                 }
 
                 try {
-                    runtime.load(attempts[filename]);
+                    const plugin = attempts[filename];
-                    logger.info('reload plugin at %c', path);
+                    for (const state of states) {
+                        state.parent.plugin(plugin, state.config);
+                    }
                 } catch (err) {
                     logger.warn(`failed to reload plugin at %c\n${coerce(err)}`, path);
                     throw err;
@@ -231,6 +236,14 @@ class Watcher {
         } catch {
             // rollback require.cache and plugin states
             rollback();
+            for (const [runtime, filename] of reloads) {
+                try {
+                    this.ctx.registry.delete(attempts[filename]);
+                    runtime.parent.plugin(runtime.plugin, runtime.config);
+                } catch (err) {
+                    logger.warn(err);
+                }
+            }
             logger.warn('Rolling back changes');
             return;
         }
@@ -240,5 +253,3 @@ class Watcher {
         logger.success('Reload done in %d ms', Date.now() - start);
     }
 }
-
-export default new Watcher({});

packages/hydrooj/src/settings.ts

@@ -1,8 +1,8 @@
 import yaml from 'js-yaml';
 import { nanoid } from 'nanoid';
 import Schema from 'schemastery';
-import * as bus from 'hydrooj/src/service/bus';
 import { Logger } from './logger';
+import * as bus from './service/bus';
 import { NestKeys } from './typeutils';
 
 const defaultPath = process.env.CI ? '/tmp/file'

packages/hydrooj/src/utils.ts

@@ -16,4 +16,8 @@ export function ArgMethod(target: any, funcName: string, obj: any) {
     return obj;
 }
 
+export function unwrapExports(module: any) {
+    return module?.default || module;
+}
+
 export * from '@hydrooj/utils/lib/utils';

packages/ui-default/handler.ts

@@ -237,10 +237,16 @@ export function apply(ctx) {
   ctx.Route('lang', '/l/:lang', LanguageHandler);
   ctx.Route('media', '/media', RichMediaHandler);
   ctx.on('app/started', buildUI);
-  ctx.on('app/watch/change', buildUI);
-  ctx.on('app/watch/unlink', buildUI);
   ctx.on('app/started', updateLogo);
   ctx.on('system/setting', updateLogo);
+  ctx.on('app/watch/change', (path) => {
+    if (!path.includes('/ui-default/') && !path.includes('/public/')) return;
+    buildUI();
+  });
+  ctx.on('app/watch/unlink', (path) => {
+    if (!path.includes('/ui-default/') && !path.includes('/public/')) return;
+    buildUI();
+  });
   buildUI();
   updateLogo();
 }