core: mark model as plugin

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');

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']);

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;

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);

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);

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;

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;

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,

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<T extends keyof DocStatusType>(
     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,

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;

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;

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 };

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' },

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,

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<string, Function> = {};
-    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<Task> & { 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;

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 },

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 },

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<K extends keyof Collections>(c: K): Collection<Collections[K]> {

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' },

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';

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 {