Hydro Plugin Development: Event System & Hooks

This skill covers Hydro's event system: how to listen for system events, broadcast across processes, manage plugin lifecycle resources, and provide replaceable modules.

1. Event API Overview

All event methods are available on the Context object (ctx):

Cluster safety

`ctx.emit()` / `ctx.parallel()` / `ctx.serial()` — in-process only. They do NOT cross process boundaries. If you have PM2 with 4 instances, calling ctx.parallel('problem/add', ...) only triggers listeners in the current process.

`ctx.broadcast()` — cross-process AND cross-server (cluster safe). The implementation in packages/hydrooj/src/service/bus.ts and packages/hydrooj/src/model/task.ts:

1. PM2 cluster mode (multiple processes on same server): Uses PM2 launchBus to relay events across all processes.
2. Multi-server / standalone mode: Uses MongoDB event collection:

• broadcast inserts a document into the event collection
• Each server watches via MongoDB Change Stream (collEvent.watch())
• If Change Stream unavailable (standalone MongoDB), falls back to polling (findOneAndUpdate every 500ms)
• Events have ack array (server IDs that processed it) and expire (TTL auto-delete)
• Each server processes unacknowledged events and calls app.parallel() locally

ctx.broadcast('record/judge', rdoc, updated, pdoc)
  → ctx.emit('bus/broadcast', 'record/judge', [rdoc, updated, pdoc])

  PM2 mode:
    → process.send({ type: 'hydro:broadcast', ... }) via PM2 bus
    → All processes on same server receive

  Multi-server / standalone:
    → Insert into MongoDB 'event' collection
    → Each server's watcher/poller detects new event
    → Each server calls app.parallel('record/judge', rdoc, updated, pdoc)

Rule of thumb for plugin developers:

• If the event affects shared state (e.g., "record judged", "problem deleted"), use ctx.broadcast() so all processes and servers react.
• If the event is local (e.g., handler lifecycle), ctx.parallel() is fine.
• When listening: ctx.on() registers in the current process only. For broadcast events, every process/server has its own listeners, and broadcast triggers all of them.

2. Complete Event Catalog

Defined in packages/hydrooj/src/service/bus.ts (EventMap interface).

Application lifecycle

Database

User

Domain

Problem

Contest

Record (submission)

Discussion

Training

Document

System & monitoring

Handler hooks

WebSocket / Subscription

File watching (dev mode)

3. Real-world Examples

Search index sync (ElasticSearch plugin)

this.ctx.on('problem/add', async (doc, docId) => {
    await this.client.index({
        index: 'problem',
        id: `${doc.domainId}/${docId}`,
        document: processDocument(doc),
    });
});

this.ctx.on('problem/edit', async (pdoc) => {
    await this.client.index({
        index: 'problem',
        id: `${pdoc.domainId}/${pdoc.docId}`,
        document: processDocument(pdoc),
    });
});

this.ctx.on('problem/del', async (domainId, docId) => {
    await this.client.delete({
        index: 'problem',
        id: `${domainId}/${docId}`,
    });
});

Post-handler hook (UI Default plugin)

// Modify response after a specific handler method runs
ctx.on('handler/after/DiscussionRaw', async (that) => {
    if (that.args.render && that.response.type === 'text/markdown') {
        that.response.type = 'text/html';
        that.response.body = await markdown.render(that.response.body);
    }
});

// Run after ALL handlers
ctx.on('handler/after', async (that) => {
    that.UiContext.SWConfig = {
        preload: SystemModel.get('ui-default.preload'),
        // ...
    };
});

One-time setup on user registration (A11Y plugin)

ctx.on('handler/after/UserRegisterWithCode#post', async (that) => {
    if (that.session.uid === 2) await UserModel.setSuperAdmin(2);
});

Modify domain query before execution

ctx.on('domain/before-get', (query) => {
    // Modify the query before it's executed
    query.someField = 'value';
});

Monitor contest balloon events

ctx.on('contest/balloon', (domainId, tid, bdoc) => {
    // Send notification, update scoreboard, etc.
});

4. ctx.effect() — Automatic Resource Cleanup

ctx.effect() registers a cleanup function that runs when the plugin is unloaded or the context is disposed.

// Inside a Service class
constructor(ctx: Context) {
    super(ctx, 'myService');

    ctx.effect(() => {
        const conn = createConnection();
        const timer = setInterval(() => { /* ... */ }, 5000);
        return () => {
            conn.close();
            clearInterval(timer);
        };
    });
}

Real example (VJudge plugin)

this.ctx.effect(() => {
    this.providers[type] = provider;
    const services = [];
    for (const account of this.accounts.filter((a) => a.type === type)) {
        if (account.enableOn && !account.enableOn.includes(os.hostname())) continue;
        const service = new AccountService(provider, account, this.ctx);
        services.push(service);
        this.pool[`${account.type}/${account.handle}`] = service;
    }
    return () => {
        // Cleanup: stop all services for this provider
        for (const service of services) service.stop();
        delete this.providers[type];
    };
});

ctx.on() already returns a dispose function

const dispose = ctx.on('problem/add', handler);
// dispose is called automatically when plugin unloads
// You can also call dispose() manually to unregister early

5. ctx.interval() — Scheduled Tasks

Registers a recurring task that auto-cancels on plugin unload.

ctx.interval(async () => {
    // This runs every 5 seconds (adjustable via config)
    const metrics = await collectMetrics();
    ctx.broadcast('metrics', hostname(), metrics);
}, 5000);

Real example (Prometheus client)

ctx.interval(async () => {
    try {
        const [gateway, name, pass] = SystemModel.getMany([
            'prom-client.gateway', 'prom-client.name', 'prom-client.password',
        ]);
        if (gateway) {
            const prefix = gateway.endsWith('/') ? gateway : `${gateway}/`;
            const endpoint = `${prefix}metrics/job/hydro-web/instance/${encodeURIComponent(hostname())}:${process.env.NODE_APP_INSTANCE}`;
            let req = superagent.post(endpoint);
            if (name) req = req.auth(name, pass, { type: 'basic' });
            await req.send(await registry.metrics());
        } else {
            ctx.broadcast('metrics', `${hostname()}/${process.env.NODE_APP_INSTANCE}`, await registry.getMetricsAsJSON());
        }
    } catch (e) {
        pushError = e.message;
    }
}, 5000 * (+SystemModel.get('prom-client.collect_rate') || 1));

Real example (VJudge — weekly sync)

this.ctx.interval(this.sync.bind(this), Time.week);

6. ctx.provideModule() — Replaceable Modules

Registers a named implementation for a module type. Multiple modules can coexist; the system or user selects which one to use.

Currently supported module types

Registration

// Register a search backend
ctx.provideModule('problemSearch', 'elastic', async (domainId, q, opts) => {
    const limit = opts?.limit || 20;
    const result = await client.search({
        index: 'problem',
        body: { query: { multi_match: { query: q, fields: ['title', 'content'] } } },
        size: limit,
    });
    return {
        hits: result.hits.hits.map((h) => h._id),
        total: result.hits.total.value,
        countRelation: result.hits.total.relation,
    };
});

Cleanup (auto via ctx.effect)

// provideModule returns a dispose function
const dispose = ctx.provideModule('problemSearch', 'mybackend', mySearchFn);
// Auto-cleaned on plugin unload, or call dispose() manually

7. Generator-based event registration (using yield)

In Service constructors, you can use yield this.ctx.on(...) for cleaner lifecycle management:

// Inside a service that uses generators
* [Context.init]() {
    yield this.ctx.on('problem/add', async (doc, docId) => { /* ... */ });
    yield this.ctx.on('problem/edit', async (pdoc) => { /* ... */ });
    yield this.ctx.provideModule('problemSearch', 'mysearch', this.search.bind(this));
}

The yield pattern ensures the returned dispose function is tracked by the Cordis framework and called automatically on disposal.

8. Event Flow Diagram

                          ┌─────────────┐
                          │   Plugin A  │
                          │  ctx.on()   │
                          └──────┬──────┘
                                 │
┌──────────────┐    emit     ┌───┴───────────────────┐
│  Core System │───────────► │  Cordis Event System  │
│  (models,    │             │                       │
│   handlers)  │    ◄─────── │  ctx.parallel()       │
└──────┬───────┘  broadcast  │  ctx.serial()         │
       │                      └───┬───────────────────┘
       │                          │
       │              ┌───────────┴───────────┐
       │              │                       │
       │        ┌─────┴─────┐          ┌──────┴──────┐
       │        │ Plugin B  │          │ Plugin C    │
       │        │ listener  │          │ listener    │
       │        └───────────┘          └─────────────┘
       │
       │   ctx.broadcast() for cross-process
       │          │
       │    ┌─────┴─────────────────────┐
       │    │  PM2 Bus / MongoDB Events │
       │    │  (cross-process relay)    │
       │    └───────────────────────────┘