Function apply

packages/hydrooj/src/model/task.ts:142–190

(ctx: Context)

const id = process.env.exec_mode === 'cluster_mode' ? hostname() : nanoid();

export async function apply(ctx: Context) {
    ctx.on('domain/delete', (domainId) => coll.deleteMany({ domainId }));
    ctx.on('bus/broadcast', (event, payload, trace) => {
        collEvent.insertOne({
            ack: [id],
            event,
            payload: BSON.EJSON.stringify(payload),
            expire: new Date(Date.now() + 10000),
            trace,
        });
    });

    if (process.env.NODE_APP_INSTANCE !== '0') return;
    const stream = collEvent.watch();
    const handleEvent = async (doc: EventDoc) => {
        process.send?.({ type: 'hydro:broadcast', data: doc });
        const payload = BSON.EJSON.parse(doc.payload);
        await (bus.parallel as any)(doc.event, ...payload);
    };
    stream.on('change', async (change) => {
        if (change.operationType !== 'insert') return;
        if (change.fullDocument.ack.includes(id)) return;
        await handleEvent(change.fullDocument);
    });
    stream.on('error', async () => {
        // The $changeStream stage is only supported on replica sets
        logger.info('No replica set found.');
        while (true) {
            let res;
            try {
                // eslint-disable-next-line no-await-in-loop
                res = await collEvent.findOneAndUpdate(
                    { expire: { $gt: new Date() }, ack: { $nin: [id] } },
                    { $push: { ack: id } },
                );
            } catch (e) {
                logger.error(e);
                // eslint-disable-next-line no-await-in-loop
                await sleep(50); // This allows exiting when shutting down
                continue;
            }
            if (argv.options.showEvent) logger.info('Event: %o', res);
            // eslint-disable-next-line no-await-in-loop
            await (res ? handleEvent(res) : sleep(500));
        }
    });
    await db.ensureIndexes(collEvent, { name: 'expire', key: { expire: 1 }, expireAfterSeconds: 0 });
    await db.ensureIndexes(coll, { name: 'task', key: { type: 1, subType: 1, priority: -1 } });
}

export default TaskModel;
global.Hydro.model.task = TaskModel;