Merge pull request #883 from syuilo/run-migrations-parallely-with-given-dop

マイグレーションを並列で走らせる

package.json

@@ -22,6 +22,7 @@
 		"format": "gulp format"
 	},
 	"devDependencies": {
+		"@prezzemolo/zip": "0.0.3",
 		"@types/bcryptjs": "2.4.1",
 		"@types/body-parser": "1.16.7",
 		"@types/chai": "4.0.4",

tools/migration/change-gridfs-metadata-name-to-filename.js

@@ -19,12 +19,32 @@ async function applyNewChange (doc) {
 }
 
 async function main () {
-	const oldTypeDocs = await DriveFile.find({
+	const query = {
 		'metadata.name': {
 			$exists: true
 		}
+	}
+
+	const count = await DriveFile.count(query)
+
+	const dop = Number.parseInt(process.argv[2]) || 5
+	const idop = ((count - (count % dop)) / dop) + 1
+
+	return zip(
+		1,
+		async (time) => {
+			console.log(`${time} / ${idop}`)
+			const doc = await db.get('drive_files').find(query, {
+				limit: dop, skip: time * dop
+			})
+			return Promise.all(doc.map(applyNewChange))
+		},
+		idop
+	).then(a => {
+		const rv = []
+		a.forEach(e => rv.push(...e))
+		return rv
 	})
-	return await Promise.all(oldTypeDocs.map(applyNewChange))
 }
 
 main().then(console.dir).catch(console.error)

tools/migration/issue_882.js

@@ -1,6 +1,7 @@
 // for Node.js interpret
 
 const { default: DriveFile } = require('../../built/api/models/drive-file')
+const { default: zip } = require('@prezzemolo/zip')
 
 const migrate = async (doc) => {
 	const result = await DriveFile.update(doc._id, {
@@ -15,30 +16,32 @@ const migrate = async (doc) => {
 }
 
 async function main() {
-	let i = 0;
-
-	const count = await DriveFile.count({});
-
-	const iterate = async () => {
-		if (i == count) return true;
-		console.log(`${i} / ${count}`);
-		const doc = (await DriveFile.find({}, { limit: 1, skip: i }))[0]
-		const res = await migrate(doc);
-		if (!res) {
-			return false;
-		} else {
-			i++
-			return await iterate();
+	const query = {
+		'metadata.type': {
+			$exists: true
 		}
 	}
 
-	const res = await iterate();
+	const count = await DriveFile.count(query);
 
-	if (res) {
-		return 'ok';
-	} else {
-		throw 'something happened';
-	}
+	const dop = Number.parseInt(process.argv[2]) || 5
+	const idop = ((count - (count % dop)) / dop) + 1
+
+	return zip(
+		1,
+		async (time) => {
+			console.log(`${time} / ${idop}`)
+			const doc = await db.get('drive_files').find(query, {
+				limit: dop, skip: time * dop
+			})
+			return Promise.all(doc.map(migrate))
+		},
+		idop
+	).then(a => {
+		const rv = []
+		a.forEach(e => rv.push(...e))
+		return rv
+	})
 }
 
 main().then(console.dir).catch(console.error)

tools/migration/use-gridfs.js

@@ -3,6 +3,7 @@
 const { default: db } = require('../../built/db/mongodb')
 const { default: DriveFile, getGridFSBucket } = require('../../built/api/models/drive-file')
 const { Duplex } = require('stream')
+const { default: zip } = require('@prezzemolo/zip')
 
 const writeToGridFS = (bucket, buffer, ...rest) => new Promise((resolve, reject) => {
 	const writeStream = bucket.openUploadStreamWithId(...rest)
@@ -45,30 +46,26 @@ const migrateToGridFS = async (doc) => {
 }
 
 async function main() {
-	let i = 0;
-
 	const count = await db.get('drive_files').count({});
 
-	const iterate = async () => {
-		if (i == count) return true;
-		console.log(`${i} / ${count}`);
-		const doc = (await db.get('drive_files').find({}, { limit: 1, skip: i }))[0]
-		const res = await migrateToGridFS(doc);
-		if (!res) {
-			return false;
-		} else {
-			i++
-			return await iterate();
-		}
-	}
+	console.log(`there are ${count} files.`)
 
-	const res = await iterate();
+	const dop = Number.parseInt(process.argv[2]) || 5
+	const idop = ((count - (count % dop)) / dop) + 1
 
-	if (res) {
-		return 'ok';
-	} else {
-		throw 'something happened';
-	}
+	return zip(
+		1,
+		async (time) => {
+			console.log(`${time} / ${idop}`)
+			const doc = await db.get('drive_files').find({}, { limit: dop, skip: time * dop })
+			return Promise.all(doc.map(migrateToGridFS))
+		},
+		idop
+	).then(a => {
+		const rv = []
+		a.forEach(e => rv.push(...e))
+		return rv
+	})
 }
 
 main().then(console.dir).catch(console.error)