Use row streaming for pg so that all users aren't loaded into memory at once (#3205)

And only transfer the columns from users table that we care about

Author: hughns

tools/syn2mas/package-lock.json

@@ -57,6 +57,7 @@
     "knex": "^3.0.1",
     "log4js": "^6.9.1",
     "pg": "^8.11.3",
+    "pg-query-stream": "^4.6.0",
     "sqlite3": "^5.1.6",
     "ts-command-line-args": "^2.5.1",
     "yaml": "^2.3.3",

tools/syn2mas/package.json

@@ -192,13 +192,7 @@ export async function migrate(): Promise<void> {
     );
   }
 
-  // Get all Synapse users, except appservice owned users who don't need to be migrated
-  const synapseUsers = await synapse
-    .select("*")
-    .from<SUser>("users")
-    .whereNull("appservice_id");
-  log.info(`Found ${synapseUsers.length} users in Synapse`);
-  for (const user of synapseUsers) {
+  async function migrateUser(user: SUser): Promise<void> {
     const localpart = user.name.split(":")[0].substring(1);
     log.info(`Processing user ${user.name} as ${localpart}`);
 
@@ -430,10 +424,43 @@ export async function migrate(): Promise<void> {
       }
     }
   }
+
+  // this is a workaround to get the list of columns that we care about from the SUser type
+  const SUserColumns: Record<keyof SUser, undefined> = {
+    name: undefined,
+    password_hash: undefined,
+    admin: undefined,
+    is_guest: undefined,
+    deactivated: undefined,
+    creation_ts: undefined,
+    appservice_id: undefined,
+  };
+
+  // Get all Synapse users, except appservice owned users who don't need to be migrated
+  const synapseUserQuery = synapse
+    .select(Object.keys(SUserColumns) as (keyof SUser)[])
+    .from<SUser>("users")
+    .whereNull("appservice_id");
+
+  let synapseUsers = 0;
+  if (synapseConfig.database.name === "sqlite3") {
+    // SQLite doesn't support streaming
+    const synapseUserRows = (await synapseUserQuery) as unknown as SUser[];
+    for (const user of synapseUserRows) {
+      synapseUsers += 1;
+      await migrateUser(user);
+    }
+  } else {
+    // Stream users from the database
+    const synapseUserStream = synapseUserQuery.stream();
+    for await (const user of synapseUserStream) {
+      synapseUsers += 1;
+      await migrateUser(user as unknown as SUser);
+    }
+  }
+
   log.info(
-    `Completed migration ${args.dryRun ? "dry-run " : ""}of ${
-      synapseUsers.length
-    } users with ${fatals} fatals and ${warnings.length} warnings:`,
+    `Completed migration ${args.dryRun ? "dry-run " : ""}of ${synapseUsers} users with ${fatals} fatals and ${warnings.length} warnings:`,
   );
   warnings.forEach((w) => log.warn(w));
   if (fatals > 0) {