core: better discussion sort method

packages/hydrooj/src/model/discussion.ts

@@ -1,9 +1,11 @@
 import { FilterQuery, ObjectID } from 'mongodb';
 import { omit } from 'lodash';
+import moment from 'moment';
 import problem from './problem';
 import * as contest from './contest';
 import * as training from './training';
 import * as document from './document';
+import TaskModel from './task';
 import { DiscussionNodeNotFoundError, DocumentNotFoundError } from '../error';
 import { DiscussionReplyDoc, DiscussionTailReplyDoc, Document } from '../interface';
 import { buildProjection } from '../utils';
@@ -82,6 +84,7 @@ export async function add(
         pin,
         updateAt: new Date(),
         views: 0,
+        sort: 100,
     };
     await bus.serial('discussion/before-add', payload);
     const res = await document.add(
@@ -125,7 +128,7 @@ export function count(domainId: string, query: FilterQuery<DiscussionDoc>) {
 
 export function getMulti(domainId: string, query: FilterQuery<DiscussionDoc> = {}, projection = PROJECTION_LIST) {
     return document.getMulti(domainId, document.TYPE_DISCUSSION, query)
-        .sort({ pin: -1, updateAt: -1 })
+        .sort({ pin: -1, sort: -1 })
         .project(buildProjection(projection));
 }
 
@@ -277,6 +280,33 @@ bus.on('problem/delete', async (domainId, docId) => {
     ]);
 });
 
+const t = Math.exp(-0.15);
+
+async function updateSort() {
+    const cursor = document.coll.find({ docType: document.TYPE_DISCUSSION });
+    while (cursor.hasNext()) {
+        // eslint-disable-next-line no-await-in-loop
+        const data = await cursor.next();
+        // eslint-disable-next-line no-await-in-loop
+        const rCount = await getMultiReply(data.domainId, data.docId).count();
+        const sort = ((data.sort || 100) + Math.max(rCount - (data.lastRCount || 0), 0) * 10) * t;
+        // eslint-disable-next-line no-await-in-loop
+        await document.coll.updateOne({ _id: data._id }, { $set: { sort, lastRCount: rCount } });
+    }
+}
+TaskModel.Worker.addHandler('discussion.sort', updateSort);
+
+bus.once('app/started', 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'],
+        });
+    }
+});
+
 global.Hydro.model.discussion = {
     typeDisplay,
     DiscussionDoc,

packages/hydrooj/src/model/task.ts

@@ -25,14 +25,10 @@ async function getFirst(query: FilterQuery<Task>) {
 class Consumer {
     consuming: boolean;
     running?: any;
-    filter: any;
-    func: Function;
     timeout: NodeJS.Timeout;
 
-    constructor(filter: any, func: Function) {
+    constructor(public filter: any, public func: Function, public destoryOnError = true) {
         this.consuming = true;
-        this.filter = filter;
-        this.func = func;
         this.get = this.get.bind(this);
         this.get();
     }
@@ -48,7 +44,7 @@ class Consumer {
             }
         } catch (e) {
             logger.error(e);
-            this.destory();
+            if (this.destoryOnError) this.destory();
         }
         this.timeout = setTimeout(this.get, 100);
     }
@@ -71,6 +67,7 @@ class WorkerService implements BaseService {
                 logger.debug('Worker task: %o', doc);
                 return this.handlers[doc.subType](doc);
             },
+            false,
         );
         this.started = true;
     }
@@ -113,7 +110,7 @@ class TaskModel {
 
     static async getDelay(query?: FilterQuery<Task>) {
         const now = new Date();
-        const [res] = await coll.find(query).sort({ executeAfter: 1 }).limit(1).toArray();
+        const res = await coll.findOne(query, { sort: { executeAfter: 1 } });
         if (res) return [Math.max(0, now.getTime() - res.executeAfter.getTime()), res.executeAfter];
         return [0, now];
     }