diff --git a/packages/hydrooj/package.json b/packages/hydrooj/package.json index 96f47f066..63804ea3c 100644 --- a/packages/hydrooj/package.json +++ b/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", diff --git a/packages/hydrooj/src/context.ts b/packages/hydrooj/src/context.ts new file mode 100644 index 000000000..2f67cd6cc --- /dev/null +++ b/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 extends cordis.Events, EventMap { +} + +function addScript(name: string, description: string, validate: Schema, run: (args: K, report: any) => boolean | Promise) { + 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(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; + 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; +export type State = cordis.State; +export type Fork = cordis.Fork; +export type Plugin = cordis.Plugin; +export const Service = cordis.Service; +export namespace Plugin { + export type Function = cordis.Plugin.Function; + export type Constructor = cordis.Plugin.Constructor; + export type Object = cordis.Plugin.Object; +} + +const T = any>(origFunc: F, disposeFunc?) => + function method(this: cordis.Service, ...args: Parameters) { + 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); diff --git a/packages/hydrooj/src/entry/cli.ts b/packages/hydrooj/src/entry/cli.ts index e37d8996e..edd119fd4 100644 --- a/packages/hydrooj/src/entry/cli.ts +++ b/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') { - 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); - } + if (typeof s.validate === 'function') arg = 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'); - const opts = options(); - await db.start(opts); + await db.start(); 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(); } diff --git a/packages/hydrooj/src/entry/common.ts b/packages/hydrooj/src/entry/common.ts index 40be86b18..db23368c8 100644 --- a/packages/hydrooj/src/entry/common.ts +++ b/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 res = require(p); - if (res.apply) { - const runtime = new Runtime(p); - runtime.load(res); - } + const m = require(p); + if (m.apply) ctx.loader.reloadPlugin(ctx, p, {}); + else logger.info(`${type.replace(/^(.)/, (t) => t.toUpperCase())} init: %s`, i); } 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'); -} diff --git a/packages/hydrooj/src/entry/setup.ts b/packages/hydrooj/src/entry/setup.ts index e8f052933..3a541c3ff 100644 --- a/packages/hydrooj/src/entry/setup.ts +++ b/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 = ` @@ -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 = '

Done! Please restart Hydro.

'; + ctx.body = '

Done! Hydro is now starting.

'; + 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; + }); } diff --git a/packages/hydrooj/src/entry/worker.ts b/packages/hydrooj/src/entry/worker.ts index a06d60ea9..f5b0d32d3 100644 --- a/packages/hydrooj/src/entry/worker.ts +++ b/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'); - const opts = options(); - await db.start(opts); + await db.start(); 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); - if (detail) logger.info('finish: lib.extra'); + await lib(pending, fail, ctx); require('../service/monitor'); - if (detail) logger.info('finish: monitor'); - 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'); + await service(pending, fail, ctx); 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); - const m = require(f); - if (m.apply) new Runtime(f).load(m); + ctx.loader.reloadPlugin(ctx, path.resolve(handlerDir, h), {}, `hydrooj/handler/${h.split('.')[0]}`); } - if (detail) logger.info('finish: handler.builtin'); - await model(pending, fail); - if (detail) logger.info('finish: model.extra'); + await model(pending, fail, ctx); const modelSetting = require('../model/setting'); await setting(pending, fail, modelSetting); - if (detail) logger.info('finish: setting'); - await handler(pending, fail); - await addon(pending, fail); - if (detail) logger.info('finish: handler.extra'); + await handler(pending, fail, ctx); + await addon(pending, fail, ctx); 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); - if (detail) logger.info('finish: script.extra'); + await script(pending, fail, ctx); 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; diff --git a/packages/hydrooj/src/handler/manage.ts b/packages/hydrooj/src/handler/manage.ts index 9ce422508..b11cafa93 100644 --- a/packages/hydrooj/src/handler/manage.ts +++ b/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 { - isEmail, isPassword, isUname, validate, -} from '../lib/validator'; +import { isEmail, isPassword, isUname } 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 }); diff --git a/packages/hydrooj/src/init.ts b/packages/hydrooj/src/init.ts index 0f4750b5a..62e8edd30 100644 --- a/packages/hydrooj/src/init.ts +++ b/packages/hydrooj/src/init.ts @@ -35,3 +35,4 @@ if (!global.Hydro) { }; global.addons = []; } +global.app = new (require('./context').Context)(); diff --git a/packages/hydrooj/src/interface.ts b/packages/hydrooj/src/interface.ts index 537e11646..7ed11efd1 100644 --- a/packages/hydrooj/src/interface.ts +++ b/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; -export { Context as PluginContext } from './service/module'; +export { Context as PluginContext } from './context'; export interface Lib extends Record { 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 } diff --git a/packages/hydrooj/src/lib/validator.ts b/packages/hydrooj/src/lib/validator.ts index 432c38df2..fd85c7cd5 100644 --- a/packages/hydrooj/src/lib/validator.ts +++ b/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, diff --git a/packages/hydrooj/src/loader.ts b/packages/hydrooj/src/loader.ts index bfb66317b..542e2a19c 100644 --- a/packages/hydrooj/src/loader.ts +++ b/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 = 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); } diff --git a/packages/hydrooj/src/service/bus.ts b/packages/hydrooj/src/service/bus.ts index e8e59ad7a..0cb8efda7 100644 --- a/packages/hydrooj/src/service/bus.ts +++ b/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 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; @@ -104,110 +95,33 @@ export interface EventMap extends Record { } /* eslint-enable @typescript-eslint/naming-convention */ -function getHooks(name: K) { - const hooks = _hooks[name] || (_hooks[name] = []); - if (hooks.length >= 2048) { - logger.warn( - 'max listener count (2048) for event "%s" exceeded, which may be caused by a memory leak', - name, - ); - } - return hooks; -} - -export function removeListener(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(name: K, listener: EventMap[K]) { - getHooks(name).push(listener); - return () => removeListener(name, listener); -} - -export function prependListener(name: K, listener: EventMap[K]) { - getHooks(name).unshift(listener); - return () => removeListener(name, listener); -} - -export function once(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(name: K, listener: EventMap[K]) { - return addListener(name, listener); -} - -export function off(name: K, listener: EventMap[K]) { - return removeListener(name, listener); -} - -export async function parallel(name: K, ...args: Parameters): Promise { - const tasks: Promise[] = []; - 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(name: K, ...args: Parameters) { - return parallel(name, ...args); -} - -export async function serial(name: K, ...args: Parameters): Promise { - 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(name: K, ...args: Parameters): Promise> { - 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(event: K, ...payload: Parameters) { - 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); +export function apply(ctx: Context) { + 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) => { + ctx.parallel(packet.data.event, ...packet.data.payload); + }); + ctx.on('bus/broadcast', (event, payload) => { + process.send({ type: 'hydro:broadcast', data: { event, payload } }); + }); + console.debug('Using pm2 event bus'); }); - on('bus/broadcast', (event, payload) => { - process.send({ type: 'hydro:broadcast', data: { event, payload } }); - }); - console.debug('Using pm2 event bus'); - }); -} catch (e) { - on('bus/broadcast', (event, payload) => parallel(event, ...payload)); - console.debug('Using mongodb external event bus'); + } catch (e) { + ctx.on('bus/broadcast', (event, payload) => ctx.parallel(event, ...payload)); + console.debug('Using mongodb external event bus'); + } } -global.Hydro.service.bus = { - addListener, bail, broadcast, emit, on, off, once, parallel, prependListener, removeListener, serial, -}; +export default app; +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; diff --git a/packages/hydrooj/src/service/db.ts b/packages/hydrooj/src/service/db.ts index 44bf07f3c..92a92c658 100644 --- a/packages/hydrooj/src/service/db.ts +++ b/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) { - opts ||= {}; + async start() { + 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(); diff --git a/packages/hydrooj/src/service/module.ts b/packages/hydrooj/src/service/module.ts deleted file mode 100644 index 60ac661fa..000000000 --- a/packages/hydrooj/src/service/module.ts +++ /dev/null @@ -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 = Object.create(null); - -function addScript(name: string, description: string, validate: Schema, run: (args: K, report: any) => boolean | Promise) { - 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(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; - } -} diff --git a/packages/hydrooj/src/service/server.ts b/packages/hydrooj/src/service/server.ts index 75e7fb638..63af143b6 100644 --- a/packages/hydrooj/src/service/server.ts +++ b/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) { + const res = Route(...args); + this.caller?.on('dispose', () => res()); + } + + // eslint-disable-next-line @typescript-eslint/naming-convention + Connection(...args: Parameters) { + 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) => { - httpServer.listen(argv.options.port || port, () => { - logger.success('Server listening at: %d', argv.options.port || port); - started = true; - r(true); + pluginContext.on('ready', async () => { + await new Promise((r) => { + httpServer.listen(argv.options.port || port, () => { + logger.success('Server listening at: %d', argv.options.port || port); + r(true); + }); + }); + pluginContext.on('dispose', () => { + httpServer.close(); + wsServer.close(); }); }); } @@ -473,6 +494,5 @@ global.Hydro.service.server = { ConnectionHandler, Route, Connection, - prepare, - start, + apply, }; diff --git a/packages/hydrooj/src/service/watcher.ts b/packages/hydrooj/src/service/watcher.ts index 5d6a9c304..f362784c2 100644 --- a/packages/hydrooj/src/service/watcher.ts +++ b/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) { const dependencies = new Set(); @@ -19,13 +20,9 @@ function loadDependencies(filename: string, ignored: Set) { 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; @@ -41,9 +38,10 @@ class Watcher { private declined: Set; private stashed = new Set(); - 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]); - logger.info('reload plugin at %c', path); + const plugin = attempts[filename]; + 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({}); diff --git a/packages/hydrooj/src/settings.ts b/packages/hydrooj/src/settings.ts index 48bf1ac4e..eec9b9d7e 100644 --- a/packages/hydrooj/src/settings.ts +++ b/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' diff --git a/packages/hydrooj/src/utils.ts b/packages/hydrooj/src/utils.ts index c708bf006..f5430f56e 100644 --- a/packages/hydrooj/src/utils.ts +++ b/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'; diff --git a/packages/ui-default/handler.ts b/packages/ui-default/handler.ts index 93ee85f22..046bcb27a 100644 --- a/packages/ui-default/handler.ts +++ b/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(); }