feat: bnote-workerを追加

package.json

@@ -28,6 +28,7 @@
 		"build-gulp": "gulp build",
 		"export": "pnpm build-ts && node ./built/migration/migrateAyuskeyNext.js",
 		"worker": "node ./built/migration/worker.js",
+		"bnote-worker": "node ./built/migration/bNoteWorker.js",
 		"note-worker": "node ./built/migration/noteWorker.js",
 		"user-worker": "node ./built/migration/userAfterHookWorker.js",
 		"bull-board": "node ./built/migration/bullDashboard.js",

src/migration/bNoteWorker.ts

@@ -0,0 +1,31 @@
+import { logger } from "./common";
+import { noteQueue } from "./jobqueue";
+
+
+const cluster = require('cluster');
+
+async function main() {
+	const numWorkers = 64;
+
+	if (cluster.isPrimary) {
+		for (let i = 0; i < numWorkers; i++) {
+			cluster.fork();
+		}
+		cluster.on("exit", (worker: any, code :any, signal:any) => {
+			console.log(`worker ${worker.process.pid} died`);
+		});
+	} else {
+
+
+		noteQueue.process(__dirname + "/processor/note.processor.js");
+		noteQueue.on("completed", (job) => {
+			logger.succ(`Note: ${job.data.note.id} の処理が完了しました`);
+		});
+	}
+}
+
+
+main().catch((e) => {
+	console.warn(e);
+	process.exit(1);
+});

src/migration/jobqueue.ts

@@ -10,7 +10,7 @@ const queueRedisConf: Queue.QueueOptions = {
 	prefix: "ayuskey_next",
 	limiter: {
 		max: 1000,
-		duration: 1000,
+		duration: 500,
 	},
 };
 export const hashtagQueue = new Queue("hashtag", queueRedisConf);

src/migration/noteWorker.ts

@@ -1,15 +1,11 @@
-import { initDb } from "@/db/postgre";
 import { logger } from "./common";
 import { noteQueue } from "./jobqueue";
 import noteProcessor from "./processor/note.processor";
-import { createConnection } from "typeorm";
-import config from "@/config";
-import { AyuskeyNextEntities } from "@/v13/models";
 
 const cluster = require('cluster');
 
 async function main() {
-	const numWorkers = 20;
+	const numWorkers = 32;
 
 	if (cluster.isPrimary) {
 		for (let i = 0; i < numWorkers; i++) {
@@ -19,17 +15,7 @@ async function main() {
 			console.log(`worker ${worker.process.pid} died`);
 		});
 	} else {
-		await initDb();
-		await createConnection({
-			name: "nextDb",
-			type: "postgres",
-			host: config.db.nextDb.host,
-			port: config.db.nextDb.port,
-			username: config.db.nextDb.user,
-			password: config.db.nextDb.pass,
-			database: config.db.nextDb.db,
-			entities: AyuskeyNextEntities,
-		});
+
 
 		noteQueue.process(noteProcessor);
 		noteQueue.on("completed", (job) => {

src/migration/processor/note.processor.ts

@@ -1,9 +1,26 @@
 import { Job } from "bull";
 import { migrateNote } from "../note";
 import { Note } from "@/models/entities/note";
+import { initDb } from "@/db/postgre";
+import { createConnection } from "typeorm";
+import config from "@/config";
+import { AyuskeyNextEntities } from "@/v13/models";
 
 export default async (job: Job<{noteId: string, note: Note}>, done: any) => {
+	await initDb();
+	const con = await createConnection({
+		name: "nextDb",
+		type: "postgres",
+		host: config.db.nextDb.host,
+		port: config.db.nextDb.port,
+		username: config.db.nextDb.user,
+		password: config.db.nextDb.pass,
+		database: config.db.nextDb.db,
+		entities: AyuskeyNextEntities,
+	});
     job.progress(job.data.noteId);
     await migrateNote(job.data.noteId, job.data.note);
     done();
-}
+		await con.close();
+
+}