From 90a9fa95f4b3d36152c14a815b9345e2612f8381 Mon Sep 17 00:00:00 2001 From: undefined Date: Fri, 30 Sep 2022 23:34:54 +0800 Subject: [PATCH] core: mark model as plugin --- packages/hydrooj/bin/hydrooj.js | 2 +- packages/hydrooj/src/entry/common.ts | 8 ++++ packages/hydrooj/src/entry/worker.ts | 14 ++++-- packages/hydrooj/src/handler/domain.ts | 3 +- packages/hydrooj/src/handler/status.ts | 2 +- packages/hydrooj/src/interface.ts | 7 --- packages/hydrooj/src/model/blacklist.ts | 15 +++--- packages/hydrooj/src/model/discussion.ts | 26 +++++----- packages/hydrooj/src/model/document.ts | 13 ++--- packages/hydrooj/src/model/domain.ts | 12 ++--- packages/hydrooj/src/model/message.ts | 12 +++-- packages/hydrooj/src/model/opcount.ts | 6 +-- packages/hydrooj/src/model/record.ts | 3 +- packages/hydrooj/src/model/storage.ts | 2 +- packages/hydrooj/src/model/task.ts | 61 ++++++++++++++---------- packages/hydrooj/src/model/token.ts | 2 +- packages/hydrooj/src/model/user.ts | 2 +- packages/hydrooj/src/service/db.ts | 6 +-- packages/hydrooj/src/service/monitor.ts | 2 +- packages/hydrooj/src/utils.ts | 4 +- packages/vjudge/src/model.ts | 4 +- 21 files changed, 117 insertions(+), 89 deletions(-) diff --git a/packages/hydrooj/bin/hydrooj.js b/packages/hydrooj/bin/hydrooj.js index 34cf3bcc3..79c7966ab 100755 --- a/packages/hydrooj/bin/hydrooj.js +++ b/packages/hydrooj/bin/hydrooj.js @@ -2,6 +2,6 @@ require('@hydrooj/utils/lib/register'); const { default: hook } = require('require-resolve-hook'); -const { bypass } = hook(/^(hydrooj|@hydrooj\/utils)/, (id) => bypass(() => require.resolve(id))); +const { bypass } = hook(/^(hydrooj|@hydrooj\/utils|cordis|schemastery|lodash|js-yaml)/, (id) => bypass(() => require.resolve(id))); require('./commands'); diff --git a/packages/hydrooj/src/entry/common.ts b/packages/hydrooj/src/entry/common.ts index 37344cb9a..446cee94c 100644 --- a/packages/hydrooj/src/entry/common.ts +++ b/packages/hydrooj/src/entry/common.ts @@ -56,6 +56,14 @@ export const lib = getLoader('lib', 'lib'); export const script = getLoader('script', 'script'); export const service = getLoader('service', 'service'); +export async function builtinModel(ctx: Context) { + const modelDir = path.resolve(__dirname, '..', 'model'); + const models = ['task', 'blacklist', 'discussion', 'document']; + for (const t of models) { + ctx.loader.reloadPlugin(ctx, path.resolve(modelDir, t), {}, `hydrooj/model/${t}`); + } +} + export async function locale(pending: string[], fail: string[]) { await Promise.all(pending.map(async (i) => { const p = locateFile(i, ['locale', 'locales']); diff --git a/packages/hydrooj/src/entry/worker.ts b/packages/hydrooj/src/entry/worker.ts index 4a23aa463..4a57ed837 100644 --- a/packages/hydrooj/src/entry/worker.ts +++ b/packages/hydrooj/src/entry/worker.ts @@ -8,7 +8,7 @@ import { Context } from '../context'; import { Logger } from '../logger'; import db from '../service/db'; import { - addon, handler, lib, locale, model, + addon, builtinModel, handler, lib, locale, model, script, service, setting, template, } from './common'; @@ -46,15 +46,17 @@ export async function apply(ctx: Context) { await lib(pending, fail, ctx); require('../service/monitor'); await service(pending, fail, ctx); - require('../model/index'); + await builtinModel(ctx); + await model(pending, fail, ctx); + await ctx.lifecycle.flush(); + + const modelSetting = require('../model/setting'); + await setting(pending, fail, modelSetting); const handlerDir = path.resolve(__dirname, '..', 'handler'); const handlers = await fs.readdir(handlerDir); for (const h of handlers) { ctx.loader.reloadPlugin(ctx, path.resolve(handlerDir, h), {}, `hydrooj/handler/${h.split('.')[0]}`); } - await model(pending, fail, ctx); - const modelSetting = require('../model/setting'); - await setting(pending, fail, modelSetting); await handler(pending, fail, ctx); await addon(pending, fail, ctx); for (const i in global.Hydro.handler) await global.Hydro.handler[i](); @@ -62,7 +64,9 @@ export async function apply(ctx: Context) { for (const h of await fs.readdir(scriptDir)) { ctx.loader.reloadPlugin(ctx, path.resolve(scriptDir, h), {}, `hydrooj/script/${h.split('.')[0]}`); } + await ctx.lifecycle.flush(); await script(pending, fail, ctx); + await ctx.lifecycle.flush(); await ctx.parallel('app/started'); if (process.env.NODE_APP_INSTANCE === '0') { const scripts = require('../upgrade').default; diff --git a/packages/hydrooj/src/handler/domain.ts b/packages/hydrooj/src/handler/domain.ts index 374a0f2a8..228eb5bfd 100644 --- a/packages/hydrooj/src/handler/domain.ts +++ b/packages/hydrooj/src/handler/domain.ts @@ -1,6 +1,7 @@ import { load } from 'js-yaml'; import { Dictionary } from 'lodash'; import moment from 'moment-timezone'; +import { Context } from '../context'; import { DomainJoinAlreadyMemberError, DomainJoinForbiddenError, ForbiddenError, InvalidJoinInvitationCodeError, PermissionError, RoleAlreadyExistError, ValidationError, @@ -334,7 +335,7 @@ class DomainSearchHandler extends Handler { } } -export async function apply(ctx) { +export async function apply(ctx: Context) { ctx.Route('ranking', '/ranking', DomainRankHandler, PERM.PERM_VIEW_RANKING); ctx.Route('domain_dashboard', '/domain/dashboard', DomainDashboardHandler); ctx.Route('domain_edit', '/domain/edit', DomainEditHandler); diff --git a/packages/hydrooj/src/handler/status.ts b/packages/hydrooj/src/handler/status.ts index 7b644f89d..06ae5e441 100644 --- a/packages/hydrooj/src/handler/status.ts +++ b/packages/hydrooj/src/handler/status.ts @@ -70,7 +70,7 @@ class StatusUpdateHandler extends Handler { } } -bus.on('app/started', () => coll.createIndex('updateAt', { expireAfterSeconds: 24 * 3600 })); +bus.on('ready', () => coll.createIndex('updateAt', { expireAfterSeconds: 24 * 3600 })); export async function apply(ctx) { ctx.Route('status', '/status', StatusHandler); diff --git a/packages/hydrooj/src/interface.ts b/packages/hydrooj/src/interface.ts index 8c717770c..b161c0072 100644 --- a/packages/hydrooj/src/interface.ts +++ b/packages/hydrooj/src/interface.ts @@ -561,13 +561,6 @@ export interface Task { [key: string]: any; } -export interface BaseService { - started: boolean; - error?: Error | string; - start: Function; - stop?: Function; -} - export interface FileNode { /** File Path In MinIO */ _id: string; diff --git a/packages/hydrooj/src/model/blacklist.ts b/packages/hydrooj/src/model/blacklist.ts index 186b4594c..e02502cca 100644 --- a/packages/hydrooj/src/model/blacklist.ts +++ b/packages/hydrooj/src/model/blacklist.ts @@ -1,9 +1,9 @@ import moment from 'moment-timezone'; -import * as bus from '../service/bus'; +import { Context } from '../context'; import db from '../service/db'; import { ArgMethod } from '../utils'; -const coll = db.collection('blacklist'); +let coll = db.collection('blacklist'); class BlackListModel { @ArgMethod @@ -32,9 +32,12 @@ class BlackListModel { } } -bus.on('app/started', () => db.ensureIndexes( - coll, - { key: { expireAt: -1 }, name: 'expire', expireAfterSeconds: 0 }, -)); +export function apply(ctx: Context) { + coll = db.collection('blacklist'); + ctx.on('ready', () => db.ensureIndexes( + coll, + { key: { expireAt: -1 }, name: 'expire', expireAfterSeconds: 0 }, + )); +} export default BlackListModel; global.Hydro.model.blacklist = BlackListModel; diff --git a/packages/hydrooj/src/model/discussion.ts b/packages/hydrooj/src/model/discussion.ts index 49836c0b6..a23600c7c 100644 --- a/packages/hydrooj/src/model/discussion.ts +++ b/packages/hydrooj/src/model/discussion.ts @@ -1,6 +1,7 @@ import { omit } from 'lodash'; import moment from 'moment'; import { FilterQuery, ObjectID } from 'mongodb'; +import { Context } from '../context'; import { DiscussionNodeNotFoundError, DocumentNotFoundError } from '../error'; import { DiscussionReplyDoc, DiscussionTailReplyDoc, Document } from '../interface'; import * as bus from '../service/bus'; @@ -289,24 +290,27 @@ async function updateSort() { await document.coll.updateOne({ _id: data._id }, { $set: { sort, lastRCount: rCount } }); } } -TaskModel.Worker.addHandler('discussion.sort', updateSort); -bus.on('app/ready', async () => { - if (!await TaskModel.count({ type: 'schedule', subType: 'discussion.sort' })) { - await TaskModel.add({ - type: 'schedule', - subType: 'discussion.sort', - executeAfter: moment().minute(0).second(0).millisecond(0).toDate(), - interval: [1, 'hour'], - }); - } -}); +export async function apply(ctx: Context) { + ctx.using(['worker'], async ({ worker }) => { + worker.addHandler('discussion.sort', updateSort); + if (!await TaskModel.count({ type: 'schedule', subType: 'discussion.sort' })) { + await TaskModel.add({ + type: 'schedule', + subType: 'discussion.sort', + executeAfter: moment().minute(0).second(0).millisecond(0).toDate(), + interval: [1, 'hour'], + }); + } + }); +} global.Hydro.model.DiscussionModel = { typeDisplay, PROJECTION_LIST, PROJECTION_PUBLIC, + apply, add, get, inc, diff --git a/packages/hydrooj/src/model/document.ts b/packages/hydrooj/src/model/document.ts index 0de018e4f..23ca953dd 100644 --- a/packages/hydrooj/src/model/document.ts +++ b/packages/hydrooj/src/model/document.ts @@ -3,6 +3,7 @@ import assert from 'assert'; import { Cursor, FilterQuery, ObjectID, OnlyFieldsOfType, UpdateQuery, } from 'mongodb'; +import { Context } from '../context'; import { BlogDoc, Content, DiscussionDoc, DiscussionReplyDoc, ProblemDoc, ProblemStatusDoc, @@ -405,7 +406,11 @@ export async function revSetStatus( return res.value; } -bus.on('app/started', async () => { +export async function apply(ctx: Context) { + ctx.on('domain/delete', (domainId) => Promise.all([ + coll.deleteMany({ domainId }), + collStatus.deleteMany({ domainId }), + ])); await db.ensureIndexes( coll, { key: { domainId: 1, docType: 1, docId: 1 }, name: 'basic', unique: true }, @@ -436,11 +441,7 @@ bus.on('app/started', async () => { { key: { domainId: 1, docType: 1, docId: 1, accept: -1, time: 1 }, name: 'contestRuleACM', sparse: true }, { key: { domainId: 1, docType: 1, uid: 1, enroll: 1, docId: 1 }, name: 'training', sparse: true }, ); -}); -bus.on('domain/delete', (domainId) => Promise.all([ - coll.deleteMany({ domainId }), - collStatus.deleteMany({ domainId }), -])); +} global.Hydro.model.document = { coll, diff --git a/packages/hydrooj/src/model/domain.ts b/packages/hydrooj/src/model/domain.ts index 3409f6c31..fd9e9da4d 100644 --- a/packages/hydrooj/src/model/domain.ts +++ b/packages/hydrooj/src/model/domain.ts @@ -274,16 +274,16 @@ class DomainModel { } } -bus.on('app/started', async () => { - await db.ensureIndexes( +bus.on('ready', () => Promise.all([ + db.ensureIndexes( coll, { key: { lower: 1 }, name: 'lower', unique: true }, - ); - await db.ensureIndexes( + ), + db.ensureIndexes( collUser, { key: { domainId: 1, uid: 1 }, name: 'uid', unique: true }, { key: { domainId: 1, rp: -1, uid: 1 }, name: 'rp', sparse: true }, - ); -}); + ), +])); export default DomainModel; global.Hydro.model.domain = DomainModel; diff --git a/packages/hydrooj/src/model/message.ts b/packages/hydrooj/src/model/message.ts index 8acb55585..44965d0c6 100644 --- a/packages/hydrooj/src/model/message.ts +++ b/packages/hydrooj/src/model/message.ts @@ -77,10 +77,12 @@ class MessageModel { } } -bus.on('app/started', () => db.ensureIndexes( - coll, - { key: { to: 1, _id: -1 }, name: 'to' }, - { key: { from: 1, _id: -1 }, name: 'from' }, -)); +export async function apply() { + return db.ensureIndexes( + coll, + { key: { to: 1, _id: -1 }, name: 'to' }, + { key: { from: 1, _id: -1 }, name: 'from' }, + ); +} export default MessageModel; global.Hydro.model.message = MessageModel; diff --git a/packages/hydrooj/src/model/opcount.ts b/packages/hydrooj/src/model/opcount.ts index fde36813c..b8ad4dd4e 100644 --- a/packages/hydrooj/src/model/opcount.ts +++ b/packages/hydrooj/src/model/opcount.ts @@ -21,9 +21,9 @@ export async function inc(op: string, ident: string, periodSecs: number, maxOper } } -bus.on('app/started', () => db.ensureIndexes( +export const apply = () => db.ensureIndexes( coll, { key: { expireAt: -1 }, name: 'expire', expireAfterSeconds: 0 }, { key: { op: 1, ident: 1, expireAt: 1 }, name: 'unique', unique: true }, -)); -global.Hydro.model.opcount = { inc }; +); +global.Hydro.model.opcount = { inc, apply }; diff --git a/packages/hydrooj/src/model/record.ts b/packages/hydrooj/src/model/record.ts index 7382f5013..118d7922e 100644 --- a/packages/hydrooj/src/model/record.ts +++ b/packages/hydrooj/src/model/record.ts @@ -233,8 +233,7 @@ class RecordModel { // Mark problem as deleted 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.on('app/started', () => db.ensureIndexes( +bus.on('ready', () => db.ensureIndexes( RecordModel.coll, { key: { domainId: 1, contest: 1, _id: -1 }, name: 'basic' }, { key: { domainId: 1, contest: 1, uid: 1, _id: -1 }, name: 'withUser' }, diff --git a/packages/hydrooj/src/model/storage.ts b/packages/hydrooj/src/model/storage.ts index 1e34946d2..35d9c7576 100644 --- a/packages/hydrooj/src/model/storage.ts +++ b/packages/hydrooj/src/model/storage.ts @@ -136,7 +136,7 @@ async function cleanFiles() { } } TaskModel.Worker.addHandler('storage.prune', cleanFiles); -bus.on('app/started', async () => { +bus.on('ready', async () => { if (process.env.NODE_APP_INSTANCE !== '0') return; await db.ensureIndexes( StorageModel.coll, diff --git a/packages/hydrooj/src/model/task.ts b/packages/hydrooj/src/model/task.ts index fcb40958a..df71a18bb 100644 --- a/packages/hydrooj/src/model/task.ts +++ b/packages/hydrooj/src/model/task.ts @@ -3,7 +3,8 @@ import moment from 'moment-timezone'; import { FilterQuery, ObjectID } from 'mongodb'; import { nanoid } from 'nanoid'; import { sleep } from '@hydrooj/utils/lib/utils'; -import { BaseService, EventDoc, Task } from '../interface'; +import { Context, Service } from '../context'; +import { EventDoc, Task } from '../interface'; import { Logger } from '../logger'; import * as bus from '../service/bus'; import db from '../service/db'; @@ -62,10 +63,8 @@ class Consumer { } } -class WorkerService implements BaseService { +class WorkerService extends Service { private handlers: Record = {}; - public started = true; - public start = () => { }; public consumer = new Consumer( { type: 'schedule', subType: { $in: Object.keys(this.handlers) } }, async (doc) => { @@ -89,10 +88,13 @@ class WorkerService implements BaseService { public addHandler(type: string, handler: Function) { this.handlers[type] = handler; this.consumer.filter = { type: 'schedule', subType: { $in: Object.keys(this.handlers) } }; + this.caller?.on('dispose', () => { + delete this.handlers[type]; + }); } } -const Worker = new WorkerService(); +const Worker = new WorkerService(app, 'worker', false); class TaskModel { static async add(task: Partial & { type: string }) { @@ -141,17 +143,36 @@ class TaskModel { } const id = process.env.exec_mode === 'cluster_mode' ? hostname() : nanoid(); -Worker.addHandler('task.daily', async () => { - await global.Hydro.model.record.coll.deleteMany({ contest: new ObjectID('000000000000000000000000') }); - await global.Hydro.script.rp?.run({}, new Logger('task/rp').debug); - await global.Hydro.script.problemStat?.run({}, new Logger('task/problem').debug); - if (global.Hydro.model.system.get('server.checkUpdate') && !(new Date().getDay() % 3)) { - await global.Hydro.script.checkUpdate?.run({}, new Logger('task/checkUpdate').debug); + +declare module '../context' { + interface Context { + worker: WorkerService; } - await bus.parallel('task/daily'); -}); -bus.on('domain/delete', (domainId) => coll.deleteMany({ domainId })); -bus.on('app/started', async () => { +} + +export async function apply(ctx: Context) { + Context.service('worker', WorkerService); + ctx.worker = Worker; + + Worker.addHandler('task.daily', async () => { + await global.Hydro.model.record.coll.deleteMany({ contest: new ObjectID('000000000000000000000000') }); + await global.Hydro.script.rp?.run({}, new Logger('task/rp').debug); + await global.Hydro.script.problemStat?.run({}, new Logger('task/problem').debug); + if (global.Hydro.model.system.get('server.checkUpdate') && !(new Date().getDay() % 3)) { + await global.Hydro.script.checkUpdate?.run({}, new Logger('task/checkUpdate').debug); + } + await ctx.parallel('task/daily'); + }); + ctx.on('domain/delete', (domainId) => coll.deleteMany({ domainId })); + ctx.on('bus/broadcast', (event, payload) => { + collEvent.insertOne({ + ack: [id], + event, + payload: JSON.stringify(payload), + expire: new Date(Date.now() + 10000), + }); + }); + if (process.env.NODE_APP_INSTANCE !== '0') return; if (!await TaskModel.count({ type: 'schedule', subType: 'task.daily' })) { await TaskModel.add({ @@ -187,15 +208,7 @@ bus.on('app/started', async () => { await (res.value ? handleEvent(res.value) : sleep(500)); } }); -}); -bus.on('bus/broadcast', (event, payload) => { - collEvent.insertOne({ - ack: [id], - event, - payload: JSON.stringify(payload), - expire: new Date(Date.now() + 10000), - }); -}); +} export default TaskModel; global.Hydro.model.task = TaskModel; diff --git a/packages/hydrooj/src/model/token.ts b/packages/hydrooj/src/model/token.ts index 85dc1b06a..aa349aeef 100644 --- a/packages/hydrooj/src/model/token.ts +++ b/packages/hydrooj/src/model/token.ts @@ -95,7 +95,7 @@ class TokenModel { } } -bus.on('app/started', () => db.ensureIndexes( +bus.on('ready', () => db.ensureIndexes( TokenModel.coll, { key: { uid: 1, tokenType: 1, updateAt: -1 }, name: 'basic', sparse: true }, { key: { expireAt: -1 }, name: 'expire', expireAfterSeconds: 0 }, diff --git a/packages/hydrooj/src/model/user.ts b/packages/hydrooj/src/model/user.ts index f2a9484b4..b65f642a4 100644 --- a/packages/hydrooj/src/model/user.ts +++ b/packages/hydrooj/src/model/user.ts @@ -403,7 +403,7 @@ class UserModel { } } -bus.on('app/started', () => Promise.all([ +bus.on('ready', () => Promise.all([ db.ensureIndexes( coll, { key: { unameLower: 1 }, name: 'uname', unique: true }, diff --git a/packages/hydrooj/src/service/db.ts b/packages/hydrooj/src/service/db.ts index 92a92c658..884a4f20e 100644 --- a/packages/hydrooj/src/service/db.ts +++ b/packages/hydrooj/src/service/db.ts @@ -2,7 +2,7 @@ import { Collection, Db, IndexSpecification, MongoClient, } from 'mongodb'; -import { BaseService, Collections } from '../interface'; +import { Collections } from '../interface'; import { Logger } from '../logger'; import options from '../options'; import * as bus from './bus'; @@ -20,10 +20,9 @@ interface MongoConfig { prefix?: string, } -class MongoService implements BaseService { +class MongoService { public client: MongoClient; public db: Db; - public started = false; private opts: MongoConfig; static buildUrl(opts: MongoConfig) { @@ -46,7 +45,6 @@ class MongoService implements BaseService { this.client = await MongoClient.connect(mongourl, { useNewUrlParser: true, useUnifiedTopology: true }); this.db = this.client.db(opts.name || 'hydro'); await bus.parallel('database/connect', this.db); - this.started = true; } public collection(c: K): Collection { diff --git a/packages/hydrooj/src/service/monitor.ts b/packages/hydrooj/src/service/monitor.ts index 9c971abd6..f2bee6dd8 100644 --- a/packages/hydrooj/src/service/monitor.ts +++ b/packages/hydrooj/src/service/monitor.ts @@ -91,7 +91,7 @@ export async function updateJudge(args) { } if (process.env.NODE_APP_INSTANCE === '0') { - bus.on('app/started', async () => { + bus.on('ready', async () => { sysinfo.get().then((info) => { coll.updateOne( { mid: info.mid, type: 'server' }, diff --git a/packages/hydrooj/src/utils.ts b/packages/hydrooj/src/utils.ts index f5430f56e..96655b91c 100644 --- a/packages/hydrooj/src/utils.ts +++ b/packages/hydrooj/src/utils.ts @@ -17,7 +17,9 @@ export function ArgMethod(target: any, funcName: string, obj: any) { } export function unwrapExports(module: any) { - return module?.default || module; + return (!module || typeof module !== 'object') ? module + : 'apply' in module && typeof module.apply === 'function' ? module + : 'default' in module ? module.default : module; } export * from '@hydrooj/utils/lib/utils'; diff --git a/packages/vjudge/src/model.ts b/packages/vjudge/src/model.ts index 7701afd44..9f42ab827 100644 --- a/packages/vjudge/src/model.ts +++ b/packages/vjudge/src/model.ts @@ -2,7 +2,7 @@ import os from 'os'; import { Context, db, DomainModel, JudgeHandler, Logger, - ProblemModel, sleep, STATUS, TaskModel, + ProblemModel, sleep, STATUS, TaskModel, Time, } from 'hydrooj'; import { BasicProvider, IBasicProvider, RemoteAccount } from './interface'; import { getDifficulty } from './providers/codeforces'; @@ -99,7 +99,7 @@ class Service { async main() { const res = await this.login(); if (!res) return; - setInterval(() => this.login(), 1 * 3600 * 1000); + setInterval(() => this.login(), Time.hour); TaskModel.consume({ type: 'remotejudge', subType: this.account.type }, this.judge.bind(this), false); const ddocs = await DomainModel.getMulti({ mount: this.account.type }).toArray(); do {