upgrade to cordis v3 (#733)

Co-authored-by: panda <panda_dtdyy@outlook.com>

packages/hydrooj/package.json

@@ -12,23 +12,23 @@
     },
     "preferUnplugged": true,
     "dependencies": {
-        "@aws-sdk/client-s3": "3.515.0",
-        "@aws-sdk/lib-storage": "3.515.0",
-        "@aws-sdk/s3-presigned-post": "3.515.0",
-        "@aws-sdk/s3-request-presigner": "3.515.0",
-        "@graphql-tools/schema": "^10.0.2",
+        "@aws-sdk/client-s3": "3.525.0",
+        "@aws-sdk/lib-storage": "3.525.1",
+        "@aws-sdk/s3-presigned-post": "3.525.0",
+        "@aws-sdk/s3-request-presigner": "3.525.0",
+        "@graphql-tools/schema": "^10.0.3",
         "@hydrooj/utils": "workspace:*",
         "@simplewebauthn/server": "7.4.0",
         "@undefined-moe/require-resolve-hook": "^1.1.0",
         "adm-zip": "0.5.5",
         "cac": "^6.7.14",
-        "cordis": "2.8.8",
+        "cordis": "3.13.3",
         "detect-browser": "^5.3.0",
         "emoji-regex": "^10.3.0",
         "emojis-list": "2.1.0",
         "fs-extra": "^11.2.0",
         "graphql": "^16.8.1",
-        "graphql-scalars": "^1.22.4",
+        "graphql-scalars": "^1.22.5",
         "js-yaml": "^4.1.0",
         "koa": "^2.15.0",
         "koa-body": "^6.0.1",
@@ -42,7 +42,7 @@
         "moment-timezone": "^0.5.45",
         "mongodb": "^5.9.2",
         "nanoid": "^5.0.6",
-        "nodemailer": "^6.9.10",
+        "nodemailer": "^6.9.11",
         "notp": "^2.0.3",
         "p-queue": "^8.0.1",
         "path-to-regexp": "^6.2.1",
@@ -69,9 +69,10 @@
         "@types/nodemailer": "^6.4.14",
         "@types/notp": "^2.0.5",
         "@types/saslprep": "^1.0.3",
-        "@types/semver": "^7.5.7",
-        "@types/superagent": "^8.1.3",
+        "@types/semver": "^7.5.8",
+        "@types/superagent": "^8.1.4",
         "@types/tar": "^6.1.11",
+        "chokidar": "^3.6.0",
         "moment": "^2.30.1"
     }
 }

packages/hydrooj/src/context.ts

@@ -6,8 +6,7 @@ import { Loader } from './loader';
 import { EventMap } from './service/bus';
 import type { CheckService } from './service/check';
 
-export interface Events<C extends Context = Context> extends cordis.Events<C>, EventMap {
-}
+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.`);
@@ -21,6 +20,12 @@ function provideModule<T extends keyof ModuleInterfaces>(type: T, id: string, mo
     return () => delete global.Hydro.module[type][id];
 }
 
+export type EffectScope = cordis.EffectScope<Context>;
+export type ForkScope = cordis.ForkScope<Context>;
+export type MainScope = cordis.MainScope<Context>;
+
+export type { Disposable, ScopeStatus, Plugin } from 'cordis';
+
 export interface Context {
     [Context.events]: Events<Context>;
     loader: Loader;
@@ -28,55 +33,62 @@ export interface Context {
     Connection: typeof import('./service/server').Connection;
     withHandlerClass: import('./service/server').RouteService['withHandlerClass'];
     check: CheckService;
-    setTimeout: typeof setTimeout;
-    setInterval: typeof setInterval;
     setImmediate: typeof setImmediate;
     addScript: typeof addScript;
     provideModule: typeof provideModule;
-    /** @deprecated use injectUI instead */
-    inject: typeof inject;
     injectUI: typeof inject;
-    api: ApiMixin;
     broadcast: Context['emit'];
     geoip?: GeoIP;
 }
 
-export class Context extends cordis.Context {
-    domain?: DomainDoc;
-}
-
-export namespace Context {
-    export interface Config extends cordis.Context.Config { }
-}
-
-export type MainScope = cordis.MainScope<Context>;
-export type EffectScope = cordis.EffectScope<Context>;
-export type ForkScope = cordis.ForkScope<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>;
+export abstract class Service<T = any, C extends Context = Context> extends cordis.Service<T, C> {
+    [cordis.Service.setup]() {
+        this.ctx = new Context() as C;
+    }
 }
 
 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()));
+        this[Context.current]?.on('dispose', () => (disposeFunc ? disposeFunc(res) : res()));
     };
+
 export class ApiMixin extends Service {
-    static readonly methods = ['addScript', 'setInterval', 'setTimeout', 'setImmediate', 'provideModule', 'inject', 'injectUI', 'broadcast'];
     addScript = T(addScript);
-    setInterval = T(setInterval, clearInterval);
-    setTimeout = T(setTimeout, clearTimeout);
     setImmediate = T(setImmediate, clearImmediate);
     provideModule = T(provideModule);
-    inject = T(inject);
     injectUI = T(inject);
     broadcast = (event: keyof EventMap, ...payload) => this.ctx.emit('bus/broadcast', event, payload);
     constructor(ctx) {
-        super(ctx, 'api', true);
+        super(ctx, '$api', true);
+        ctx.mixin('$api', ['addScript', 'setImmediate', 'provideModule', 'injectUI', 'broadcast']);
     }
 }
-Context.service('api', ApiMixin);
+
+export class Context extends cordis.Context {
+    domain?: DomainDoc;
+
+    constructor(config: {} = {}) {
+        super(config);
+        this.plugin(ApiMixin);
+    }
+
+    /** @deprecated use `ctx.root` instead */
+    get app() {
+        return this.root;
+    }
+}
+
+export namespace Context {
+    export type Associate<P extends string, C extends Context = Context> = cordis.Context.Associate<P, C>;
+}
+
+const old = cordis.Registry.prototype.inject;
+// cordis.Registry.prototype.using = old;
+cordis.Registry.prototype.inject = function wrapper(...args) {
+    if (typeof args[0] === 'string') {
+        console.warn('old functionality of ctx.inject is deprecated. please use ctx.injectUI instead.');
+        return T(inject).call(this, ...args as any) as any;
+    }
+    return old.call(this, ...args);
+};

packages/hydrooj/src/entry/cli.ts

@@ -107,6 +107,7 @@ export async function load(ctx: Context) {
         lib(pending, fail, ctx),
         service(pending, fail, ctx),
     ]);
+    ctx.plugin(require('../service/worker'));
     await builtinModel(ctx);
     await model(pending, fail, ctx);
     await setting(pending, fail, require('../model/setting'));

packages/hydrooj/src/entry/worker.ts

@@ -38,6 +38,7 @@ export async function apply(ctx: Context) {
     await modelSystem.runConfig();
     const storage = require('../service/storage');
     await storage.loadStorageService();
+    await require('../service/worker').apply(ctx);
     await require('../service/server').apply(ctx);
     // Make sure everything is ready and then start main entry
     if (argv.options.watch) ctx.plugin(require('../service/watcher').default, {});

packages/hydrooj/src/handler/contest.ts

@@ -57,20 +57,6 @@ registerResolver(
     'Get a contest by ID',
 );
 
-ScheduleModel.Worker.addHandler('contest', async (doc) => {
-    const tdoc = await contest.get(doc.domainId, doc.tid);
-    if (!tdoc) return;
-    const tasks = [];
-    for (const op of doc.operation) {
-        if (op === 'unhide') {
-            for (const pid of tdoc.pids) {
-                tasks.push(problem.edit(doc.domainId, pid, { hidden: false }));
-            }
-        }
-    }
-    await Promise.all(tasks);
-});
-
 export class ContestListHandler extends Handler {
     @param('rule', Types.Range(contest.RULES), true)
     @param('group', Types.Name, true)
@@ -776,4 +762,17 @@ export async function apply(ctx: Context) {
     ctx.Route('contest_file_download', '/contest/:tid/file/:filename', ContestFileDownloadHandler, PERM.PERM_VIEW_CONTEST);
     ctx.Route('contest_user', '/contest/:tid/user', ContestUserHandler, PERM.PERM_VIEW_CONTEST);
     ctx.Route('contest_balloon', '/contest/:tid/balloon', ContestBalloonHandler, PERM.PERM_VIEW_CONTEST);
+    ctx.worker.addHandler('contest', async (doc) => {
+        const tdoc = await contest.get(doc.domainId, doc.tid);
+        if (!tdoc) return;
+        const tasks = [];
+        for (const op of doc.operation) {
+            if (op === 'unhide') {
+                for (const pid of tdoc.pids) {
+                    tasks.push(problem.edit(doc.domainId, pid, { hidden: false }));
+                }
+            }
+        }
+        await Promise.all(tasks);
+    });
 }

packages/hydrooj/src/handler/judge.ts

@@ -300,6 +300,10 @@ export class JudgeConnectionHandler extends ConnectionHandler {
                 this.query.lang = { $in: msg.lang };
                 this.consumer?.setQuery(this.query);
             }
+            if (msg.type instanceof Array && msg.type.every((i) => typeof i === 'string')) {
+                this.query.type = { $in: msg.type };
+                this.consumer?.setQuery(this.query);
+            }
         } else if (msg.key === 'start') {
             clearTimeout(this.startTimeout);
             this.consumer ||= task.consume(this.query, this.newTask.bind(this), true, this.concurrency);

packages/hydrooj/src/lib/i18n.ts

@@ -21,7 +21,7 @@ class I18nService extends Service {
     load(lang: string, content: Record<string, string>) {
         translations[lang] ||= [];
         translations[lang].unshift(content);
-        this.caller?.on('dispose', () => {
+        this[Context.current]?.on('dispose', () => {
             translations[lang] = translations[lang].filter((i) => i !== content);
         });
     }
@@ -35,7 +35,7 @@ class I18nService extends Service {
     }
 }
 
-Context.service('i18n');
+app.provide('i18n', undefined, true);
 app.i18n = new I18nService(app);
 
 String.prototype.translate = function translate(...languages: string[]) {

packages/hydrooj/src/loader.ts

@@ -34,7 +34,6 @@ export function resolveConfig(plugin: any, config: any) {
     if (schema && plugin['schema'] !== false) config = schema(config);
     return config;
 }
-Context.service('loader');
 
 const timeout = Symbol.for('loader.timeout');
 const showLoadTime = argv.options.showLoadTime;
@@ -43,7 +42,7 @@ export class Loader {
     static readonly Record = Symbol.for('loader.record');
 
     public app: Context;
-    public config: Context.Config;
+    public config: {};
     public suspend = false;
     public cache: Record<string, string> = Object.create(null);
 
@@ -147,8 +146,8 @@ export function addScript(name: string, description: string) {
     };
 }
 
-Context.service('loader');
 const loader = new Loader();
+app.provide('loader');
 app.loader = loader;
 loader.app = app;
 app.state[Loader.Record] = Object.create(null);

packages/hydrooj/src/model/schedule.ts

@@ -1,11 +1,10 @@
 import moment from 'moment-timezone';
 import { Filter, ObjectId } from 'mongodb';
-import { sleep } from '@hydrooj/utils/lib/utils';
-import { Context, Service } from '../context';
+import { Context } from '../context';
 import { Schedule } from '../interface';
 import { Logger } from '../logger';
-import * as bus from '../service/bus';
 import db from '../service/db';
+import type { WorkerService } from '../service/worker';
 import RecordModel from './record';
 
 const logger = new Logger('model/schedule');
@@ -27,73 +26,6 @@ async function getFirst(query: Filter<Schedule>) {
     return null;
 }
 
-class Consumer {
-    consuming: boolean;
-    running?: any;
-
-    constructor(public filter: any, public func: Function, public destroyOnError = true) {
-        this.consuming = true;
-        this.consume();
-        bus.on('app/exit', this.destroy);
-    }
-
-    async consume() {
-        while (this.consuming) {
-            try {
-                // eslint-disable-next-line no-await-in-loop
-                const res = await getFirst(this.filter);
-                if (res) {
-                    this.running = res;
-                    // eslint-disable-next-line no-await-in-loop
-                    await this.func(res);
-                    this.running = null;
-                    // eslint-disable-next-line no-await-in-loop
-                } else await sleep(1000);
-            } catch (err) {
-                logger.error(err);
-                if (this.destroyOnError) this.destroy();
-            }
-        }
-    }
-
-    async destroy() {
-        this.consuming = false;
-    }
-}
-
-class WorkerService extends Service {
-    private handlers: Record<string, Function> = {};
-    public consumer = new Consumer(
-        { type: 'schedule', subType: { $in: Object.keys(this.handlers) } },
-        async (doc) => {
-            try {
-                logger.debug('Worker task: %o', doc);
-                const start = Date.now();
-                await Promise.race([
-                    this.handlers[doc.subType](doc),
-                    sleep(1200000),
-                ]);
-                const spent = Date.now() - start;
-                if (spent > 500) logger.warn('Slow worker task (%d ms): %o', spent, doc);
-            } catch (e) {
-                logger.error('Worker task fail: ', e);
-                logger.error('%o', doc);
-            }
-        },
-        false,
-    );
-
-    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(app, 'worker', false);
-
 class ScheduleModel {
     static coll = coll;
 
@@ -123,28 +55,24 @@ class ScheduleModel {
     }
 
     static getFirst = getFirst;
-    static Worker = Worker;
-}
-
-declare module '../context' {
-    interface Context {
-        worker: WorkerService;
-    }
+    /** @deprecated use ctx.inject(['worker'], cb) instead */
+    static Worker: WorkerService;
 }
 
 export async function apply(ctx: Context) {
-    Context.service('worker', WorkerService);
-    ctx.worker = Worker;
-
-    Worker.addHandler('task.daily', async () => {
-        await RecordModel.coll.deleteMany({ contest: { $in: [RecordModel.RECORD_PRETEST, RecordModel.RECORD_GENERATE] } });
-        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.inject(['worker'], (c) => {
+        ScheduleModel.Worker = c.worker;
+        c.worker.addHandler('task.daily', async () => {
+            await RecordModel.coll.deleteMany({ contest: { $in: [RecordModel.RECORD_PRETEST, RecordModel.RECORD_GENERATE] } });
+            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 }));
 
     if (process.env.NODE_APP_INSTANCE !== '0') return;

packages/hydrooj/src/model/setting.ts

@@ -1,6 +1,6 @@
 /* eslint-disable max-len */
 /* eslint-disable no-await-in-loop */
-import cordis from 'cordis';
+import * as cordis from 'cordis';
 import yaml from 'js-yaml';
 import { Dictionary } from 'lodash';
 import moment from 'moment-timezone';
@@ -300,11 +300,10 @@ declare module '../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()));
+        this[Context.current]?.on('dispose', () => (disposeFunc ? disposeFunc(res) : res()));
     };
 
 export class SettingService extends Service {
-    static readonly methods = ['PreferenceSetting', 'AccountSetting', 'DomainSetting', 'DomainUserSetting', 'SystemSetting'];
     PreferenceSetting = T(PreferenceSetting);
     AccountSetting = T(AccountSetting);
     DomainSetting = T(DomainSetting);
@@ -312,16 +311,16 @@ export class SettingService extends Service {
     SystemSetting = T(SystemSetting);
     constructor(ctx: Context) {
         super(ctx, 'setting', true);
+        ctx.mixin('setting', ['PreferenceSetting', 'AccountSetting', 'DomainSetting', 'DomainUserSetting', 'SystemSetting']);
     }
 
     get(key: string) {
-        return (this.caller ? this.caller.domain?.config?.[key.replace(/\./g, '$')] : null) ?? global.Hydro.model.system.get(key);
+        return (this[Context.current] ? this[Context.current].domain?.config?.[key.replace(/\./g, '$')] : null) ?? global.Hydro.model.system.get(key);
     }
 }
 
 export async function apply(ctx: Context) {
-    Context.service('setting', SettingService);
-    ctx.setting = new SettingService(ctx);
+    ctx.plugin(SettingService);
     logger.info('Ensuring settings');
     const system = global.Hydro.model.system;
     for (const setting of SYSTEM_SETTINGS) {

packages/hydrooj/src/model/storage.ts

@@ -3,8 +3,8 @@ import { escapeRegExp } from 'lodash';
 import moment from 'moment-timezone';
 import { nanoid } from 'nanoid';
 import type { Readable } from 'stream';
+import { Context } from '../context';
 import mime from '../lib/mime';
-import * as bus from '../service/bus';
 import db from '../service/db';
 import storage from '../service/storage';
 import ScheduleModel from './schedule';
@@ -135,26 +135,35 @@ async function cleanFiles() {
         res = await StorageModel.coll.findOneAndDelete({ autoDelete: { $lte: new Date() } });
     }
 }
-ScheduleModel.Worker.addHandler('storage.prune', cleanFiles);
-bus.on('ready', async () => {
-    if (process.env.NODE_APP_INSTANCE !== '0') return;
-    await db.ensureIndexes(
-        StorageModel.coll,
-        { key: { path: 1, autoDelete: 1 }, sparse: true, name: 'autoDelete' },
-    );
-    if (!await ScheduleModel.count({ type: 'schedule', subType: 'storage.prune' })) {
-        await ScheduleModel.add({
-            type: 'schedule',
-            subType: 'storage.prune',
-            executeAfter: moment().startOf('hour').toDate(),
-            interval: [1, 'hour'],
-        });
-    }
-});
-bus.on('domain/delete', async (domainId) => {
-    const files = await StorageModel.list(`problem/${domainId}`);
-    await StorageModel.del(files.map((i) => i.path));
-});
+
+export function apply(ctx: Context) {
+    ctx.inject(['worker'], (c) => {
+        c.worker.addHandler('storage.prune', cleanFiles);
+    });
+    ctx.on('domain/delete', async (domainId) => {
+        const [problemFiles, contestFiles, trainingFiles] = await Promise.all([
+            StorageModel.list(`problem/${domainId}`),
+            StorageModel.list(`contest/${domainId}`),
+            StorageModel.list(`training/${domainId}`),
+        ]);
+        await StorageModel.del(problemFiles.concat(contestFiles).concat(trainingFiles).map((i) => i.path));
+    });
+    ctx.on('ready', async () => {
+        if (process.env.NODE_APP_INSTANCE !== '0') return;
+        await db.ensureIndexes(
+            StorageModel.coll,
+            { key: { path: 1, autoDelete: 1 }, sparse: true, name: 'autoDelete' },
+        );
+        if (!await ScheduleModel.count({ type: 'schedule', subType: 'storage.prune' })) {
+            await ScheduleModel.add({
+                type: 'schedule',
+                subType: 'storage.prune',
+                executeAfter: moment().startOf('hour').toDate(),
+                interval: [1, 'hour'],
+            });
+        }
+    });
+}
 
 global.Hydro.model.storage = StorageModel;
 export default StorageModel;

packages/hydrooj/src/service/check.ts

@@ -14,7 +14,7 @@ export class CheckService extends Service {
     c = {};
     addChecker(type: string, checkFunc: CheckItem) {
         this.checkers[`check${type}`] = checkFunc;
-        this.caller?.on('dispose', () => {
+        this[Context.current]?.on('dispose', () => {
             delete this.checkers[`check${type}`];
         });
     }
@@ -73,6 +73,6 @@ check.addChecker('Setting', async (ctx, log, warn) => {
 });
 
 export async function apply(ctx: Context) {
-    Context.service('check', CheckService);
+    ctx.provide('check', CheckService, true);
     ctx.check = check;
 }

packages/hydrooj/src/service/server.ts

@@ -479,12 +479,12 @@ class NotFoundHandler extends Handler {
 }
 
 export class RouteService extends Service {
-    static readonly methods = ['Route', 'Connection', 'withHandlerClass'];
     private registry = {};
     private registrationCount = Counter();
 
-    constructor(ctx) {
+    constructor(ctx: Context) {
         super(ctx, 'server', true);
+        ctx.mixin('server', ['Route', 'Connection', 'withHandlerClass']);
     }
 
     private register(func: typeof Route | typeof Connection, ...args: Parameters<typeof Route>) {
@@ -498,7 +498,7 @@ export class RouteService extends Service {
         this.registrationCount[name]++;
         const res = func(...args);
         this.ctx.parallel(`handler/register/${name}`, HandlerClass);
-        this.caller?.on('dispose', () => {
+        this[Context.current]?.on('dispose', () => {
             this.registrationCount[name]--;
             if (!this.registrationCount[name]) delete this.registry[name];
             res();
@@ -508,7 +508,7 @@ export class RouteService extends Service {
     public withHandlerClass(name: string, callback: (HandlerClass: typeof HandlerCommon) => any) {
         if (this.registry[name]) callback(this.registry[name]);
         this.ctx.on(`handler/register/${name}`, callback);
-        this.caller?.on('dispose', () => {
+        this[Context.current]?.on('dispose', () => {
             this.ctx.off(`handler/register/${name}`, callback);
         });
     }
@@ -531,7 +531,7 @@ declare module '../context' {
 }
 
 export async function apply(pluginContext: Context) {
-    Context.service('server', RouteService);
+    pluginContext.provide('server', undefined, true);
     pluginContext.server = new RouteService(pluginContext);
     app.keys = system.get('session.keys') as unknown as string[];
     if (process.env.HYDRO_CLI) return;

packages/hydrooj/src/service/watcher.ts

@@ -38,10 +38,9 @@ export default class Watcher extends Service {
     private declined: Set<string>;
     private stashed = new Set<string>();
 
-    constructor(public ctx: Context, private config: any) {
+    constructor(public ctx: Context) {
         super(ctx, 'watcher', true);
         this.externals = new Set(Object.keys(require.cache));
-        Context.service('watcher');
     }
 
     start() {

packages/hydrooj/src/service/worker.ts

@@ -0,0 +1,90 @@
+import moment from 'moment-timezone';
+import { sleep } from '@hydrooj/utils';
+import { Context, Service } from '../context';
+import db from './db';
+
+declare module '../context' {
+    interface Context {
+        worker: WorkerService;
+    }
+}
+
+export class WorkerService extends Service {
+    private handlers: Record<string, Function> = {};
+    consuming = true;
+    running = null;
+    promise: Promise<any> = null;
+    coll = db.collection('schedule');
+
+    constructor(ctx: Context) {
+        super(ctx, 'worker', true);
+        this.consume();
+    }
+
+    async getFirst() {
+        if (process.env.CI) return null;
+        const q = {
+            executeAfter: { $lt: new Date() },
+            type: 'schedule',
+            subType: { $in: Object.keys(this.handlers) },
+        };
+        const res = await this.coll.findOneAndDelete(q);
+        if (res.value) {
+            this.ctx.logger.debug('%o', res.value);
+            if (res.value.interval) {
+                const executeAfter = moment(res.value.executeAfter).add(...res.value.interval).toDate();
+                await this.coll.insertOne({ ...res.value, executeAfter });
+            }
+            return res.value;
+        }
+        return null;
+    }
+
+    async consume() {
+        while (this.consuming) {
+            try {
+                // eslint-disable-next-line no-await-in-loop
+                const doc = await this.getFirst();
+                if (!doc) {
+                    // eslint-disable-next-line no-await-in-loop
+                    await sleep(1000);
+                    continue;
+                }
+                this.running = doc;
+                this.ctx.logger.debug('Worker task: %o', doc);
+                const start = Date.now();
+                this.promise = Promise.race([
+                    this.handlers[doc.subType](doc).catch((e) => {
+                        this.ctx.logger.error('Worker task fail: ', e);
+                        this.ctx.logger.error('%o', doc);
+                    }),
+                    sleep(1200000),
+                ]);
+                // eslint-disable-next-line no-await-in-loop
+                await this.promise;
+                const spent = Date.now() - start;
+                if (spent > 500) this.ctx.logger.warn('Slow worker task (%d ms): %o', spent, doc);
+                this.running = null;
+            } catch (err) {
+                this.ctx.logger.error(err);
+            }
+        }
+    }
+
+    public addHandler(type: string, handler: Function) {
+        this.handlers[type] = handler;
+        this[Context.current]?.on('dispose', () => {
+            delete this.handlers[type];
+        });
+    }
+
+    public async stop() {
+        this.consuming = false;
+        await this.promise;
+    }
+}
+
+export async function apply(ctx: Context) {
+    ctx.provide('worker', undefined, true);
+    ctx.worker = new WorkerService(ctx);
+}