Merge branch 'feature/xiaofeijingsai' into dev

src/api_s2c/event/payForDiamond/ApiCanReceive.ts

@@ -26,10 +26,10 @@ export async function playerCanReceive(call: ApiCall, callError : boolean = true
     } 
     activityId = activityInfo.hdid;
     const hasReceivedKey = hasGotKeyPrefix + activityId;
-    const hasReceivedStr = await G.iorediscross.get(hasReceivedKey);
+    const hasReceivedStr = await G.crossioredis.get(hasReceivedKey);
     const hasReceived = hasReceivedStr? parseInt(hasReceivedStr) : 0;
     const remaining = activityInfo.data['totalmoney'] - hasReceived;
-    const showOffResult = await G.iorediscross.lrange(showOffListKeyPrefix + activityId, 0, -1);
+    const showOffResult = await G.crossioredis.lrange(showOffListKeyPrefix + activityId, 0, -1);
     const showOffList = showOffResult.map(result => JSON.parse(result));
     const zeroTime = PublicShared.getToDayZeroTime();
     const vipScore = await ActionLog.getDayLog(call.uid, 'pay');

src/api_s2c/event/payForDiamond/ApiReceive.ts

@@ -71,12 +71,12 @@ export default async function (call: ApiCall<ReqReceive, ResReceive>) {
         }
         const activityData = canReceiveResult.activityInfo.data;
         // 更新 redis 领取记录之前先加锁, 防止多领
-        const lockResult = await G.iorediscross.setnx(hasGotLockKey, 1);
+        const lockResult = await G.crossioredis.setnx(hasGotLockKey, 1);
         if (lockResult) {
-            await G.iorediscross.expire(hasGotLockKey, 1); // 设置 ttl 避免死锁
+            await G.crossioredis.expire(hasGotLockKey, 1); // 设置 ttl 避免死锁
             const activityId = call.req.activityId;
             const hasReceivedKey = hasGotKeyPrefix + activityId;
-            const hasReceivedStr = await G.iorediscross.get(hasReceivedKey);
+            const hasReceivedStr = await G.crossioredis.get(hasReceivedKey);
             const hasReceived = hasReceivedStr? parseInt(hasReceivedStr) : 0;
             const remaining = activityData['totalmoney'] - hasReceived;
             if (remaining <= 0) {
@@ -86,8 +86,8 @@ export default async function (call: ApiCall<ReqReceive, ResReceive>) {
             } else {
                 const { group, maxAmount } = randomWithWeight(activityData['groupConf']['base']['arr']);
                 const gotAmount = calcDiamondGot(remaining, group, maxAmount);
-                await G.iorediscross.incrby(hasReceivedKey, Math.abs(gotAmount)); // 添加已领取的额度
+                await G.crossioredis.incrby(hasReceivedKey, Math.abs(gotAmount)); // 添加已领取的额度
-                await G.iorediscross.del(hasGotLockKey); // 移除锁
+                await G.crossioredis.del(hasGotLockKey); // 移除锁
                 await PlayerFun.sendPrize(call,  [{ 'a': 'attr', 't': 'rmbmoney', 'n': gotAmount }]);
                 const showOff = gotAmount >= activityData['groupConf']['base']['loglimit'];
                 call.succ({ // 领取核心逻辑完成, 请求可以返回了
@@ -108,8 +108,8 @@ export default async function (call: ApiCall<ReqReceive, ResReceive>) {
                         name: call.conn.gud.name, gotAmount, serverID: call.conn.gud.sid
                     });
                     const showOffListKey = showOffListKeyPrefix + activityId;
-                    await G.iorediscross.lpush(showOffListKey, msg);
+                    await G.crossioredis.lpush(showOffListKey, msg);
-                    await G.iorediscross.ltrim(showOffListKey, 0, 49); // 限制列表保存 50 条消息, 避免无限增长
+                    await G.crossioredis.ltrim(showOffListKey, 0, 49); // 限制列表保存 50 条消息, 避免无限增长
                 }
             }
         } else {

src/api_s2c/event/xiaofeijingsai/ApiOpen.ts

@@ -4,13 +4,13 @@ import {HuoDongFun} from "../../../public/huodongfun";
 
 /**
  * 消费竞赛
- * redis缓存120秒
+ * redis缓存60秒
  * 返回活动日期内的
  * @param call
  */
 export default async function (call: ApiCall<ReqOpen, ResOpen>) {
 
-    let ioList = await G.ioredis.get(`rank:xiaofeijingsai`);
+    let ioList = await G.crossioredis.get(`rank:xiaofeijingsai`);
     if (ioList) {
         let myData = await getMyData(call, JSON.parse(ioList))
         return call.succ({list: JSON.parse(ioList), myData})
@@ -21,28 +21,27 @@ export default async function (call: ApiCall<ReqOpen, ResOpen>) {
 
     let limit = _hd.data?.rank?.slice(-1)?.[0]?.rank?.slice(-1)?.[0] || call.req.limit || 100
 
-    let rmbuse = await G.mongodb.collection('rmbuse').aggregate([
+    let rmbuse = await G.crossmongodb.collection('rmbuse').find({
-        {$match: {isAdd: false, cTime: {$gte: _hd.stime, $lte: _hd.etime}}},
+        time: {
-        {$group: {_id: "$uid", total: {$sum: "$change"}}},
+            $gte: _hd.stime,
-        {$sort: {total: 1}},
+            $lte: _hd.etime + 10
-        {$limit: limit}
+        }
-    ]).toArray()
+    }).sort({change: 1}).limit(limit).toArray()
-
-    let list: any = rmbuse.map(i => ({...i, total: R.negate(i.total)}))
 
+    let list: any = rmbuse.map(i => ({...i, total: R.negate(i.change)}))
 
     let rankList = sortRankList(_hd.data.rank, list)
 
-    let users = await G.mongodb.collection('user').find({uid: {$in: rankList.map(i => i._id).filter(i => i._id != 'system')}}).toArray()
+    let users = await G.crossmongodb.collection('xfjs_user').find({uid: {$in: rankList.map(i => i._id).filter(i => i._id != 'system')}}).toArray()
 
     rankList = rankList.map(i => ({...i, player: users.find(v => v.uid == i._id) || {}}))
 
     // 活动结束前半小时，缓存过期时间改为10秒
-    let exTime = (G.time + 1800) > _hd.etime ? 10 : 120
+    let exTime = (G.time + 1800) > _hd.etime ? 10 : 60
 
-    G.ioredis.setex(`rank:xiaofeijingsai`, exTime, JSON.stringify(rankList));
+    G.crossioredis.setex(`rank:xiaofeijingsai`, exTime, JSON.stringify(rankList));
 
-    let myData = await getMyData(call, rankList, _hd)
+    let myData = await getMyData(call, rankList)
 
     call.succ({list: rankList, myData})
 }
@@ -69,19 +68,14 @@ export function sortRankList(rank, list) {
 }
 
 // 获取自己的信息
-async function getMyData(call, rankList, _hd?) {
+async function getMyData(call, rankList) {
     let myData = rankList.find(i => i._id == call.uid)
     if (myData) return myData
 
-    if (!_hd) {
+    let myCut: any = await G.crossmongodb.collection('rmbuse').findOne({uid: call.uid})
-        _hd = (await HuoDongFun.gethdList(call, 11))[0]
-    }
-    let myCut = (await G.mongodb.collection('rmbuse').aggregate([
-        {$match: {uid: call.uid, isAdd: false, cTime: {$gte: _hd.stime, $lte: _hd.etime}}},
-        {$group: {_id: "$uid", total: {$sum: "$change"}}}
-    ]).toArray())[0]
 
     let myUser = await G.mongodb.collection('user').findOne({uid: call.uid})
+    G.crossmongodb.collection('xfjs_user').updateOne({uid: call.uid}, myUser, {upsert: true})
 
     if (!myCut) {
         myCut = {_id: myUser.uid, total: 0}

src/api_s2c/eventlist/ApihdGetList.ts

@@ -8,4 +8,11 @@ export default async function (call: ApiCall<ReqhdGetList, ReshdGetList>) {
     let _hdList = await HuoDongFun.gethdList(call)
 
     call.succ({hdlist: _hdList});
+
+    // 消费竞赛是跨服活动，活动开启时，同步当前用户信息到跨服数据库
+    G.huodong.xfjs = !!_hdList.find(i => i.htype == 11);
+    if (G.huodong.xfjs){
+        let myUser = await G.mongodb.collection('user').findOne({uid: call.uid})
+        G.crossmongodb.collection('xfjs_user').updateOne({uid: call.uid}, myUser, {upsert: true})
+    }
 }

src/global.ts

@@ -1,22 +1,22 @@
 import EventEmitter from 'events';
-import { existsSync, readdirSync, readFileSync, writeFileSync } from 'fs';
+import {existsSync, readdirSync, readFileSync, writeFileSync} from 'fs';
-import { parse } from 'json5';
+import {parse} from 'json5';
 import * as mathjs from 'mathjs';
-import { join, resolve } from 'path';
+import {join, resolve} from 'path';
-import { argv, env } from 'process';
+import {argv, env} from 'process';
-import { HttpServer, WsClient, WsServer } from 'tsrpc';
+import {HttpServer, WsClient, WsServer} from 'tsrpc';
-import { ServiceType as ServiceTypeCross } from './cross/protocols/serviceProto';
+import {ServiceType as ServiceTypeCross} from './cross/protocols/serviceProto';
-import { MyEvent } from './event';
+import {MyEvent} from './event';
-import { addListener, gEventType } from './globalListener';
+import {addListener, gEventType} from './globalListener';
 import localConfig from './localConfig';
-import { ServiceType as ServiceTypeHttp } from './monopoly/protocols/serviceProto';
+import {ServiceType as ServiceTypeHttp} from './monopoly/protocols/serviceProto';
-import { SchedulerManage } from './public/scheduler/scheduler';
+import {SchedulerManage} from './public/scheduler/scheduler';
-import { _mongodb } from './setMongodb';
+import {_mongodb} from './setMongodb';
-import { redisJsonFun } from './setRedis';
+import {redisJsonFun} from './setRedis';
-import { ResGetList } from './shared/protocols/pay/PtlGetList';
+import {ResGetList} from './shared/protocols/pay/PtlGetList';
-import { ServiceType as ServiceTypeWs } from './shared/protocols/serviceProto';
+import {ServiceType as ServiceTypeWs} from './shared/protocols/serviceProto';
-import { PublicShared } from './shared/public/public';
+import {PublicShared} from './shared/public/public';
-import { clusterRunOnce } from './clusterUtils';
+import {clusterRunOnce} from './clusterUtils';
 import * as ramda from 'ramda'
 import Redis from 'ioredis';
 
@@ -33,22 +33,24 @@ declare global {
     type atn = { a: string, t: string | number | any, n: number; colour?: number; shiwuBuff?: any; };
 
     /**类型过滤 */
-    type FilterConditionally<Source, Condition> = Pick<
+    type FilterConditionally<Source, Condition> = Pick<Source,
-        Source,
         {
             [K in keyof Source]: Source[K] extends Condition ? K : never
-        }[keyof Source]
+        }[keyof Source]>;
-    >;
 
     interface Array<T> {
         /**数组随机取值 */
         random(): T;
+
         /**取一个数组在当前数组中的交集 */
         intersection(other: T[]): T[];
+
         /**取一个数组在当前数组中的差集 */
         difference(other: T[]): T[];
+
         /**数组是否存在重复元素 */
         isDuplication(): boolean;
+
         /**打乱数组 */
         shuffle(): this;
     }
@@ -69,9 +71,9 @@ class _G {
         /**服务器日志模式 error | debug*/
         logModel: string;
     } = {
-            serverType: 'msg',
+        serverType: 'msg',
-            logModel: 'error'
+        logModel: 'error'
-        };
+    };
     /**当前时间对象 */
     date: Date;
     /**当前时间戳 */
@@ -93,7 +95,7 @@ class _G {
     /**ioredis连接对象 */
     ioredis: Redis;
     /** 跨服 ioredis 连接对象 */
-    iorediscross: Redis;
+    crossioredis: Redis;
     /**mongodb连接对象 */
     mongodb: _mongodb;
     /**crossmongodb连接对象 */
@@ -102,6 +104,11 @@ class _G {
     /**所有玩家的充值记录 */
     allPlayerPayLog: k_v<ResGetList['list']> = {};
 
+    /**跨服活动——消费竞赛的开启状态 */
+    huodong = {
+        xfjs: false
+    };
+
     private event = new EventEmitter();
 
     /**映射开服时间 */
@@ -115,7 +122,7 @@ class _G {
         off(type: any, callback: any, caller?: any): void;
         emit(type: any, ...args: any[]): void;
         debug(): any;
-        removeAllListeners(type?:any):void;
+        removeAllListeners(type?: any): void;
     } {
         return MyEvent as any;
     }
@@ -142,10 +149,10 @@ class _G {
     }
 
     on<T extends keyof gEventType>(event: T, callback: gEventType[T]) {
-        return this.event.on(event, (...args)=>{
+        return this.event.on(event, (...args) => {
-            try{
+            try {
                 callback.call(this, ...args);
-            }catch(e){
+            } catch (e) {
                 console.error(e)
             }
         });
@@ -166,7 +173,7 @@ class _G {
             if (file.endsWith('.json5')) {
                 let json = parse(readFileSync(join(jsonPath, file), 'utf-8'));
                 this.gc[file.split('.')[0]] = json;
-            } else if(file.endsWith('.json')) {
+            } else if (file.endsWith('.json')) {
                 let json = JSON.parse(readFileSync(join(jsonPath, file), 'utf-8'));
                 this.gc[file.split('.')[0]] = json;
             }

src/ioredis.ts

@@ -16,7 +16,7 @@ export async function initIORedis() {
     G.ioredis = new Redis(G.argv.serverType == 'cross' ? G.config.crossRedisUrl : G.config.redisUrl,{
         keyPrefix: preKey,
     });
-    G.iorediscross = new Redis(G.config.crossRedisUrl,{
+    G.crossioredis = new Redis(G.config.crossRedisUrl,{
         keyPrefix: "cross_",
     });
 }

src/module/mongodb.ts

@@ -133,4 +133,6 @@ export type MongodbCollections = {
     fightLog: CollectionFightLog
     shop: CollectionShop
     pushgift:CollectionPushGift
+
+    xfjs_user: CollectionUser;
 };

src/public/player.ts

@@ -214,7 +214,6 @@ export class PlayerFun {
     }
 
     static async changeAttrLog(uid: string, change, atn, before) {
-
         let data = {
             uid,
             before,
@@ -225,6 +224,13 @@ export class PlayerFun {
             atn
         }
         G.mongodb.collection('rmbuse').insertOne(data);
+        // 消费竞赛开启时写入跨服数据库
+        if (G.huodong.xfjs && !data.isAdd) {
+            G.crossmongodb.collection('rmbuse').updateOne({uid: data.uid}, {
+                time: G.time,
+                $inc: {change: data.change}
+            }, {upsert: true});
+        }
     }
 
     static async changeAttr(uid: string, change: Partial<player>) {

src/public/scheduler/scheduler_xiaofeijingsai.ts

@@ -38,14 +38,15 @@ export class Scheduler_xfjs_Local_Ctor extends Scheduler {
         if (!_hd) return
 
         let limit = _hd.data?.rank?.slice(-1)?.[0]?.rank?.slice(-1)?.[0] || 100
-        let rmbuse = await G.mongodb.collection('rmbuse').aggregate([
-            {$match: {isAdd: false, cTime: {$gte: _hd.stime, $lte: _hd.etime}}},
-            {$group: {_id: "$uid", total: {$sum: "$change"}}},
-            {$sort: {total: 1}},
-            {$limit: limit}
-        ]).toArray()
 
-        let list: any = rmbuse.map(i => ({...i, total: R.negate(i.total)}))
+        let rmbuse = await G.crossmongodb.collection('rmbuse').find({
+            time: {
+                $gte: _hd.stime,
+                $lte: _hd.etime + 10
+            }
+        }).sort({change: 1}).limit(limit).toArray()
+
+        let list: any = rmbuse.map(i => ({...i, total: R.negate(i.change)}))
 
         let ranklist = sortRankList(_hd.data.rank, list)
 
@@ -53,6 +54,7 @@ export class Scheduler_xfjs_Local_Ctor extends Scheduler {
             let users = R.slice(i.rank[0] - 1, i.rank[1])(ranklist)
             users.map(v => {
                 if (v._id == 'system') return
+                if (G.config.serverId != users.sid) return;
                 EmailFun.addEmail({
                     uid: v._id,
                     type: 'system',