init

src/jobs/index.js

@@ -0,0 +1,19 @@
+'use strict';
+const { inject } = require('../common/Provider');
+const Agenda = require('agenda');
+const mongoose = require('mongoose');
+const mountJobs = require('./jobs');
+
+module.exports = () => inject(async function startJob(SysConfig) {
+    const { DB_USER_NAME, DB_PASSWD, DB_HOST, DB_PORT, JOB_DB_NAME } = SysConfig;
+    mylog.info('连接到任务队列数据库...');
+    const jobConn = await mongoose.createConnection(`mongodb://${DB_USER_NAME}:${DB_PASSWD}@${DB_HOST}:${DB_PORT}/${JOB_DB_NAME}`, {
+        useNewUrlParser: true,
+        autoReconnect: true,
+    });
+    mylog.info('连接到任务队列数据库成功');
+    const Job = new Agenda({ mongo: jobConn });
+    mountJobs(Job);
+    await Job.start();
+    mylog.info('任务队列启动...');
+});

src/jobs/jobs.js

@@ -0,0 +1,126 @@
+'use strict';
+/**
+ * 为何要把分属于不同业务模块的job集中到此处？
+ * 如果后续要把任务执行与web服务器分离，就会很方便，就算不分离，也没多大坏处。
+
+ * 任务包括：
+ * transaction : {
+    * 1. 每小时去清理指定startTime的地产，将
+    * {
+    *      status: -> 1,
+    *      price: -> price * (1 + profit),
+    *      currentOwner： -> ''
+    * }
+    * 2. 将currentOwner对应的用户的asset.log加上a
+ * }
+ */
+const mongoose = require('mongoose');
+const User = require('../modules/user/user.model');
+const Estate = require('../modules/estate/estate.model');
+const EstateOrder = require('../modules/estateOrder/estateOrder.model');
+const { createAction } = require('../modules/action/action.handler');
+const jobs = [
+    {
+        name: 'refreshEstateAndPayBack',
+        processor: async (job, done) => {
+            //todo: 程序内检查是否可以执行，保证不会启动时错误的自动执行一次
+            const now = new Date().getMinutes();
+            if (now > 0 && now < 58) {
+                mylog.error('[Schedule Job] 不在合理时间启动...程序结束');
+                return 0;
+            }
+            mylog.info('开始定时计算作业...');
+            const query = { startTime: +(new Date().getUTCHours()) + 8 + 1 };
+            const elist = await Estate.find(query).exec();
+            if (!elist || elist.length <= 0) return 0;
+            const [ estateSession, userSession, orderSession ] = await Promise.all([
+                mongoose.startSession(),
+                mongoose.startSession(),
+                mongoose.startSession()
+            ]);
+            await Promise.all([
+                estateSession.startTransaction(),
+                userSession.startTransaction(),
+                orderSession.startTransaction()
+            ]);
+            job.touch();
+            for (let i = 0, length = elist.length; i < length; i++) {
+                const estate = elist[i];
+                const estateId = estate.id;
+                const userId = estate.currentOwner;
+                // todo: 更精确判读是否要回收地产
+                if (!userId) continue;
+                const [ user, order ] = await Promise.all([
+                    User.findById(userId).exec(),
+                    EstateOrder.findOne({
+                        userId,
+                        estateId,
+                        status: 'holding'
+                    }).exec()
+                ]);
+                const profit = +estate.price * +estate.profit;
+                const payback = profit + estate.price;
+                const action = createAction(3, { actor: userId, estate, amount: profit, remain: user.asset });
+                if (!Number.isFinite(payback)) throw new Error('非法的收益数值！');
+                const userOpts = { returnOriginal: false, session: userSession };
+                const orderOpts = { returnOriginal: false, session: orderSession };
+                const estateOpts = { returnOriginal: false, session: estateSession };
+                try {
+                    await Promise.all([
+                        user.increase('profit', payback, userOpts),
+                        estate.sell(estateOpts),
+                        order.finish(orderOpts),
+                        action.save()
+                    ]);
+                    await Promise.all([
+                        userSession.commitTransaction(),
+                        orderSession.commitTransaction(),
+                        estateSession.commitTransaction()
+                    ]);
+                } catch (error) {
+                    mylog.error(error);
+                    await Promise.all([
+                        action.remove(),
+                        userSession.abortTransaction(),
+                        orderSession.abortTransaction(),
+                        estateSession.abortTransaction()
+                    ]);
+                } finally {
+                    await Promise.all([
+                        userSession.endSession(),
+                        orderSession.endSession(),
+                        estateSession.endSession()
+                    ]);
+                }
+                job.touch();
+            }
+            mylog.info('收益计算分配完成');
+            done();
+        },
+        type: 'every',
+        // 写成 '*/59 * * * *' 会出现59分执行一次，下一小时的0分再执行一次
+        // 写成 '*/59 * * * *' 是每59
+        // 最差的解决方案是 在58分执行
+        slot: '59 */1 * * *',   // 59th minute per hour
+        priority: 'highest',
+        unique: {
+            id: 'payback001'
+        },
+        skipImmediate: true,
+        timezone: "ETC/GMT+8"
+    }
+];
+function mountJobs (Job) {
+    jobs.map(jobConfig => {
+        const { name, processor, type, slot, priority, unique, skipImmediate, timezone } = jobConfig;
+        Job.define(name, {
+            timezone,
+            skipImmediate,
+            priority,
+            unique: { 'data.type': 'active', 'data.userId': unique.id }
+        }, processor);
+        Job[type](slot, name);
+    });
+}
+
+module.exports = Job => mountJobs(Job);