diff --git a/src/upgrades/3.3.0/chat_room_refactor.js b/src/upgrades/3.3.0/chat_room_refactor.js index 7c008a9764..0d9242b10c 100644 --- a/src/upgrades/3.3.0/chat_room_refactor.js +++ b/src/upgrades/3.3.0/chat_room_refactor.js @@ -1,8 +1,6 @@ 'use strict'; -const _ = require('lodash'); - const db = require('../../database'); const batch = require('../../batch'); @@ -18,46 +16,69 @@ module.exports = { for (let i = 1; i <= nextChatRoomId; i++) { allRoomIds.push(i); } - progress.total = allRoomIds.length; + progress.total = 0; + + // calculate user count and set progress.total + await batch.processArray(allRoomIds, async (roomIds) => { + await Promise.all(roomIds.map(async (roomId) => { + const userCount = await db.sortedSetCard(`chat:room:${roomId}:uids`); + await db.setObjectField(`chat:room:${roomId}`, 'userCount', userCount); + progress.total += userCount; + })); + }, { + batch: 500, + }); + await batch.processArray(allRoomIds, async (roomIds) => { - progress.incr(roomIds.length); - const [arrayOfUids, arrayOfRoomData] = await Promise.all([ - db.getSortedSetsMembers(roomIds.map(roomId => `chat:room:${roomId}:uids`)), - db.getObjects(roomIds.map(roomId => `chat:room:${roomId}`)), - ]); + const arrayOfRoomData = await db.getObjects(roomIds.map(roomId => `chat:room:${roomId}`)); await Promise.all(roomIds.map(async (roomId, index) => { - const uids = arrayOfUids[index]; const roomData = arrayOfRoomData[index]; - if (!uids.length && !roomData) { + if (!roomData) { return; } - if (roomData && roomData.owner && !uids.includes(String(roomData.owner))) { - uids.push(roomData.owner); - } - const userKeys = uids.map(uid => `uid:${uid}:chat:room:${roomId}:mids`); - const mids = await db.getSortedSetsMembers(userKeys); - const uniqMids = _.uniq(_.flatten(mids)); - let messageData = await db.getObjects(uniqMids.map(mid => `message:${mid}`)); - messageData.forEach((m, idx) => { - if (m) { - m.mid = parseInt(uniqMids[idx], 10); - } - }); - messageData = messageData.filter(Boolean); - const bulkSet = messageData.map( - msg => [`message:${msg.mid}`, { roomId: roomId }] - ); + const midsSeen = Object.create(null); + await batch.processSortedSet(`chat:room:${roomId}:uids`, async (uids) => { + for (const uid of uids) { + // eslint-disable-next-line no-await-in-loop + await batch.processSortedSet(`uid:${uid}:chat:room:${roomId}:mids`, async (mids) => { + const uniqMids = mids.filter(mid => !midsSeen.hasOwnProperty(mid)); + if (!uniqMids.length) { + return; + } + + let messageData = await db.getObjects(uniqMids.map(mid => `message:${mid}`)); + messageData.forEach((m, idx) => { + if (m) { + m.mid = parseInt(uniqMids[idx], 10); + } + }); + messageData = messageData.filter(Boolean); - await db.setObjectBulk(bulkSet); - await db.setObjectField(`chat:room:${roomId}`, 'userCount', uids.length); - await db.sortedSetAdd( - `chat:room:${roomId}:mids`, - messageData.map(m => m.timestamp), - messageData.map(m => m.mid), - ); - await db.deleteAll(userKeys); + const bulkSet = messageData.map( + msg => [`message:${msg.mid}`, { roomId: roomId }] + ); + + await db.setObjectBulk(bulkSet); + await db.sortedSetAdd( + `chat:room:${roomId}:mids`, + messageData.map(m => m.timestamp), + messageData.map(m => m.mid), + ); + uniqMids.forEach((mid) => { + midsSeen[mid] = 1; + }); + }, { + batch: 500, + }); + // eslint-disable-next-line no-await-in-loop + await db.deleteAll(`uid:${uid}:chat:room:${roomId}:mids`); + } + progress.incr(uids.length); + }, { + batch: 500, + }); })); }, { batch: 500,