Read input files synchronously line by line

package-lock.json

@@ -11,12 +11,14 @@
       "dependencies": {
         "axios": "^1.4.0",
         "dotenv": "^16.0.3",
+        "n-readlines": "^1.0.1",
         "reflect-metadata": "^0.1.13",
         "sqlite3": "^5.1.6",
         "typeorm": "^0.3.16",
         "winston": "^3.8.2"
       },
       "devDependencies": {
+        "@types/n-readlines": "^1.0.3",
         "@types/node": "^20.2.1",
         "@typescript-eslint/eslint-plugin": "^5.59.6",
         "@typescript-eslint/parser": "^5.59.6",
@@ -273,6 +275,15 @@
       "integrity": "sha512-dRLjCWHYg4oaA77cxO64oO+7JwCwnIzkZPdrrC71jQmQtlhM556pwKo5bUzqvZndkVbeFLIIi+9TC40JNF5hNQ==",
       "dev": true
     },
+    "node_modules/@types/n-readlines": {
+      "version": "1.0.3",
+      "resolved": "https://registry.npmjs.org/@types/n-readlines/-/n-readlines-1.0.3.tgz",
+      "integrity": "sha512-6zMvlOZp56sEEndnkEX+A05Q+LrPB7yDUX1d5oE0aOF9rIE10ndhZHEy0mBFqzBUIw5sSFMUypwgD3FehplPvg==",
+      "dev": true,
+      "dependencies": {
+        "@types/node": "*"
+      }
+    },
     "node_modules/@types/node": {
       "version": "20.2.1",
       "resolved": "https://registry.npmjs.org/@types/node/-/node-20.2.1.tgz",
@@ -3487,6 +3498,14 @@
         "thenify-all": "^1.0.0"
       }
     },
+    "node_modules/n-readlines": {
+      "version": "1.0.1",
+      "resolved": "https://registry.npmjs.org/n-readlines/-/n-readlines-1.0.1.tgz",
+      "integrity": "sha512-z4SyAIVgMy7CkgsoNw7YVz40v0g4+WWvvqy8+ZdHrCtgevcEO758WQyrYcw3XPxcLxF+//RszTz/rO48nzD0wQ==",
+      "engines": {
+        "node": ">=6.x.x"
+      }
+    },
     "node_modules/natural-compare": {
       "version": "1.4.0",
       "resolved": "https://registry.npmjs.org/natural-compare/-/natural-compare-1.4.0.tgz",
@@ -5486,6 +5505,15 @@
       "integrity": "sha512-dRLjCWHYg4oaA77cxO64oO+7JwCwnIzkZPdrrC71jQmQtlhM556pwKo5bUzqvZndkVbeFLIIi+9TC40JNF5hNQ==",
       "dev": true
     },
+    "@types/n-readlines": {
+      "version": "1.0.3",
+      "resolved": "https://registry.npmjs.org/@types/n-readlines/-/n-readlines-1.0.3.tgz",
+      "integrity": "sha512-6zMvlOZp56sEEndnkEX+A05Q+LrPB7yDUX1d5oE0aOF9rIE10ndhZHEy0mBFqzBUIw5sSFMUypwgD3FehplPvg==",
+      "dev": true,
+      "requires": {
+        "@types/node": "*"
+      }
+    },
     "@types/node": {
       "version": "20.2.1",
       "resolved": "https://registry.npmjs.org/@types/node/-/node-20.2.1.tgz",
@@ -7784,6 +7812,11 @@
         "thenify-all": "^1.0.0"
       }
     },
+    "n-readlines": {
+      "version": "1.0.1",
+      "resolved": "https://registry.npmjs.org/n-readlines/-/n-readlines-1.0.1.tgz",
+      "integrity": "sha512-z4SyAIVgMy7CkgsoNw7YVz40v0g4+WWvvqy8+ZdHrCtgevcEO758WQyrYcw3XPxcLxF+//RszTz/rO48nzD0wQ=="
+    },
     "natural-compare": {
       "version": "1.4.0",
       "resolved": "https://registry.npmjs.org/natural-compare/-/natural-compare-1.4.0.tgz",

package.json

@@ -28,6 +28,7 @@
   },
   "version": "0.1.0",
   "devDependencies": {
+    "@types/n-readlines": "^1.0.3",
     "@types/node": "^20.2.1",
     "@typescript-eslint/eslint-plugin": "^5.59.6",
     "@typescript-eslint/parser": "^5.59.6",
@@ -46,6 +47,7 @@
   "dependencies": {
     "axios": "^1.4.0",
     "dotenv": "^16.0.3",
+    "n-readlines": "^1.0.1",
     "reflect-metadata": "^0.1.13",
     "sqlite3": "^5.1.6",
     "typeorm": "^0.3.16",

src/app.ts

@@ -1,8 +1,7 @@
 import dotenv from 'dotenv'
 dotenv.config()
-import fs from 'node:fs'
+import lineByLine from 'n-readlines'
 import log from './logger'
-import readline from 'node:readline'
 import { RcUser, createUser } from './users'
 import { whoami } from './synapse'
 import 'reflect-metadata'
@@ -26,20 +25,16 @@ const enum Entities {
   Messages = 'rocketchat_message.json',
 }
 
-function loadRcExport(entity: Entities): Promise<void> {
+async function loadRcExport(entity: Entities) {
-  const rl = readline.createInterface({
+  const rl = new lineByLine(`./inputs/${entity}`)
-    input: fs.createReadStream(`./inputs/${entity}`, {
-      encoding: 'utf-8',
-    }),
-    crlfDelay: Infinity,
-  })
 
-  rl.on('line', async (line) => {
+  let line: false | Buffer
-    const item = JSON.parse(line)
+  while ((line = rl.next())) {
+    const item = JSON.parse(line.toString())
     switch (entity) {
       case Entities.Users:
         const rcUser: RcUser = item
-        log.info(`User: ${rcUser.name}: ${rcUser._id}`)
+        log.debug(`Parsing user: ${rcUser.name}: ${rcUser._id}`)
 
         // Check for exclusion
         if (
@@ -90,12 +85,7 @@ function loadRcExport(entity: Entities): Promise<void> {
       default:
         throw new Error(`Unhandled Entity: ${entity}`)
     }
-  })
+  }
-  return new Promise((resolve) => {
-    rl.on('close', () => {
-      resolve()
-    })
-  })
 }
 
 async function main() {