Implement ordered concurrency

package.json

@@ -49,6 +49,7 @@
     "typescript": "^5.5.4"
   },
   "dependencies": {
+    "@supercharge/promise-pool": "^3.2.0",
     "@types/showdown": "^2.0.6",
     "axios": "^1.7.2",
     "dotenv": "^16.4.5",

src/app.ts

@@ -3,14 +3,15 @@ dotenv.config()
 import { AxiosError } from 'axios'
 import lineByLine from 'n-readlines'
 import { exit } from 'node:process'
+import { PromisePool } from '@supercharge/promise-pool'
 import 'reflect-metadata'
 import { Entity, entities } from './Entities'
 import { handleDirectChats } from './handlers/directChats'
 import { handleRoomMemberships } from './handlers/handleRoomMemberships'
-import { handle as handleMessage } from './handlers/messages'
+import { handle as handleMessage, RcMessage } from './handlers/messages'
 import { handlePinnedMessages } from './handlers/pinnedMessages'
-import { handle as handleRoom } from './handlers/rooms'
-import { handle as handleUser } from './handlers/users'
+import { handle as handleRoom, RcRoom } from './handlers/rooms'
+import { handle as handleUser, RcUser } from './handlers/users'
 import log from './helpers/logger'
 import { initStorage } from './helpers/storage'
 import { whoami } from './helpers/synapse'
@@ -22,28 +23,53 @@ log.info('rocketchat2matrix starts.')
  * @param entity The Entity with it's file name and type definitions
  */
 async function loadRcExport(entity: Entity) {
+  const concurrency = parseInt(process.env.CONCURRENCY_LIMIT || '50')
+  const user_queue: RcUser[] = []
+  const room_queue: RcRoom[] = []
+  const messages_per_room: Map<string, RcMessage[]> = new Map()
+
   const rl = new lineByLine(`./inputs/${entities[entity].filename}`)
 
   let line: false | Buffer
   while ((line = rl.next())) {
     const item = JSON.parse(line.toString())
     switch (entity) {
       case Entity.Users:
-        await handleUser(item)
+        user_queue.push(item)
         break
 
       case Entity.Rooms:
-        await handleRoom(item)
+        room_queue.push(item)
         break
 
       case Entity.Messages:
-        await handleMessage(item)
+        if (messages_per_room.has(item.rid)) {
+          messages_per_room.get(item.rid)?.push(item)
+        } else {
+          messages_per_room.set(item.rid, [item])
+        }
         break
 
       default:
         throw new Error(`Unhandled Entity: ${entity}`)
     }
   }
+
+  await PromisePool.withConcurrency(concurrency)
+    .for(user_queue)
+    .process((item) => handleUser(item))
+
+  await PromisePool.withConcurrency(concurrency)
+    .for(room_queue)
+    .process((item) => handleRoom(item))
+
+  await PromisePool.withConcurrency(concurrency)
+    .for(messages_per_room.values())
+    .process(async (room) => {
+      for (const item of room) {
+        await handleMessage(item)
+      }
+    })
 }
 
 async function main() {