enhance(server): Use job queue for account delete (#525)

* enhance(server): Use job queue for account delete (#7668)

* enhance(server): Use job queue for account delete

Fix #5336

* ジョブをひとつに

* remove done call

* clean up

* add User.isDeleted

* コミット忘れ

* Update 1629512953000-user-is-deleted.ts

* show dialog

* lint

* Update 1629512953000-user-is-deleted.ts

* chore: tsconfigの整理

* fix: error

* chore: 調整

* fix: インポート

* chore: tune

* chore: tune

* chore: tune

* feat: delete actor

* refactor: unused

* docs: update

CHANGELOG.md

@@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - デフォルトは`262144000B`
 - オブジェクトストレージの`S3ForcePathStyle`を指定できるように
 - 藍ちゃんウィジェット
+- アカウント削除の連合
 
 ### Changed
 
@@ -26,6 +27,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - 一部の条件下(DMなど)で既読がつかない問題を修正
 - アンテナに既読がつくようになりました
 - 空のプッシュ通知が発生しないように
+- アカウント削除の安定性を向上
 
 ## [11.37.1-rei0784-5.16.0] 2021-09-12
 

migration/1629512953000-user-is-deleted.ts

@@ -0,0 +1,15 @@
+import {MigrationInterface, QueryRunner} from "typeorm";
+
+export class isUserDeleted1629512953000 implements MigrationInterface {
+    name = 'isUserDeleted1629512953000'
+
+    public async up(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`ALTER TABLE "user" ADD "isDeleted" boolean NOT NULL DEFAULT false`);
+        await queryRunner.query(`COMMENT ON COLUMN "user"."isDeleted" IS 'Whether the User is deleted.'`);
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`ALTER TABLE "user" DROP COLUMN "isDeleted"`);
+    }
+
+}

src/client/app/account.ts

@@ -10,6 +10,7 @@ type Account = {
 	token: string;
 	isModerator: boolean;
 	isAdmin: boolean;
+	isDeleted: boolean;
 };
 
 const data = localStorage.getItem('account');

src/models/entities/user.ts

@@ -191,6 +191,13 @@ export class User {
 	})
 	public isExplorable: boolean;
 
+	// アカウントが削除されたかどうかのフラグだが、完全に削除される際は物理削除なので実質削除されるまでの「削除が進行しているかどうか」のフラグ
+	@Column('boolean', {
+		default: false,
+		comment: 'Whether the User is deleted.'
+	})
+	public isDeleted: boolean;
+
 	@Column('varchar', {
 		length: 128, array: true, default: '{}'
 	})

src/models/repositories/user.ts

@@ -332,6 +332,7 @@ export class UserRepository extends Repository<User> {
 				}).then(count => count > 0),
 				noCrawle: profile!.noCrawle,
 				isExplorable: user.isExplorable,
+				isDeleted: user.isDeleted,
 				hideOnlineStatus: user.hideOnlineStatus,
 				hasUnreadAnnouncement: this.getHasUnreadAnnouncement(user.id),
 				hasUnreadAntenna: this.getHasUnreadAntenna(user.id),

src/queue/index.ts

@@ -1,7 +1,7 @@
 import * as httpSignature from 'http-signature';
 
 import config from '../config';
-import { ILocalUser } from '../models/entities/user';
+import { ILocalUser, User } from '../models/entities/user';
 import { program } from '../argv';
 
 import processDeliver from './processors/deliver';
@@ -180,6 +180,15 @@ export function createImportUserListsJob(user: ILocalUser, fileId: DriveFile['id
 	});
 }
 
+export function createDeleteAccountJob(user: User) {
+	return dbQueue.add('deleteAccount', {
+		user: user
+	}, {
+		removeOnComplete: true,
+		removeOnFail: true
+	});
+}
+
 export function createDeleteObjectStorageFileJob(key: string) {
 	return objectStorageQueue.add('deleteFile', {
 		key: key

src/queue/processors/db/delete-account.ts

@@ -0,0 +1,79 @@
+import * as Bull from 'bull';
+import { queueLogger } from '../../logger';
+import { DriveFiles, Notes, Users } from '../../../models/index';
+import { DbUserJobData } from '../../type';
+import { Note } from '../../../models/entities/note';
+import { DriveFile } from '../../../models/entities/drive-file';
+import { MoreThan } from 'typeorm';
+import { deleteFileSync } from '../../../services/drive/delete-file';
+
+const logger = queueLogger.createSubLogger('delete-account');
+
+export async function deleteAccount(job: Bull.Job<DbUserJobData>): Promise<string | void> {
+	logger.info(`Deleting account of ${job.data.user.id} ...`);
+
+	const user = await Users.findOne(job.data.user.id);
+	if (user == null) {
+		return;
+	}
+
+	{ // Delete notes
+		let cursor: Note['id'] | null = null;
+
+		while (true) {
+			const notes = await Notes.find({
+				where: {
+					userId: user.id,
+					...(cursor ? { id: MoreThan(cursor) } : {})
+				},
+				take: 100,
+				order: {
+					id: 1
+				}
+			});
+
+			if (notes.length === 0) {
+				break;
+			}
+
+			cursor = notes[notes.length - 1].id;
+
+			await Notes.delete(notes.map(note => note.id));
+		}
+
+		logger.succ(`All of notes deleted`);
+	}
+
+	{ // Delete files
+		let cursor: DriveFile['id'] | null = null;
+
+		while (true) {
+			const files = await DriveFiles.find({
+				where: {
+					userId: user.id,
+					...(cursor ? { id: MoreThan(cursor) } : {})
+				},
+				take: 10,
+				order: {
+					id: 1
+				}
+			});
+
+			if (files.length === 0) {
+				break;
+			}
+
+			cursor = files[files.length - 1].id;
+
+			for (const file of files) {
+				await deleteFileSync(file);
+			}
+		}
+
+		logger.succ(`All of files deleted`);
+	}
+
+	await Users.delete(job.data.user.id);
+
+	return 'Account deleted';
+}

src/queue/processors/db/index.ts

@@ -8,7 +8,8 @@ import { exportUserLists } from './export-user-lists';
 import { importFollowing } from './import-following';
 import { importBlocking } from './import-blocking';
 import { importUserLists } from './import-user-lists';
-import { DbJobData } from '../../types';
+import { deleteAccount } from './delete-account';
+import { DbJobData } from '../../type';
 
 const jobs = {
 	deleteDriveFiles,
@@ -19,7 +20,8 @@ const jobs = {
 	exportUserLists,
 	importFollowing,
 	importBlocking,
-	importUserLists
+	importUserLists,
+	deleteAccount,
 } as Record<string, Bull.ProcessCallbackFunction<DbJobData> | Bull.ProcessPromiseFunction<DbJobData>>;
 
 export default function(dbQueue: Bull.Queue<DbJobData>) {

src/queue/type.ts

@@ -1,11 +1,11 @@
 //import { ObjectID } from 'mongodb';
 import * as httpSignature from 'http-signature';
-import { ILocalUser } from '../models/entities/user';
+import { ILocalUser, User } from '../models/entities/user';
 import { IActivity } from '../remote/activitypub/type';
 
 export type DeliverJobData = {
 	/** Actor */
-	user: ILocalUser;
+	user: User;
 	/** Activity */
 	content: any;
 	/** inbox URL to deliver */
@@ -35,3 +35,7 @@ export type DeleteObjectStorageFileJobData = {
 };
 
 export type CleanRemoteFilesJobData = {};
+
+export type ThinUser = {
+	id: User['id'];
+};

src/remote/activitypub/kernel/delete/actor.ts

@@ -0,0 +1,26 @@
+import { apLogger } from '../../logger';
+import { createDeleteAccountJob } from '../../../../queue';
+import { IRemoteUser } from '../../../../models/entities/user';
+import { Users } from '../../../../models';
+
+const logger = apLogger;
+
+export async function deleteActor(actor: IRemoteUser, uri: string): Promise<string> {
+	logger.info(`Deleting the Actor: ${uri}`);
+
+	if (actor.uri !== uri) {
+		return `skip: delete actor ${actor.uri} !== ${uri}`;
+	}
+
+	if (actor.isDeleted) {
+		logger.info(`skip: already deleted`);
+	}
+
+	const job = await createDeleteAccountJob(actor);
+
+	await Users.update(actor.id, {
+		isDeleted: true,
+	});
+
+	return `ok: queued ${job.name} ${job.id}`;
+}

src/remote/activitypub/kernel/delete/index.ts

@@ -1,7 +1,8 @@
 import deleteNote from './note';
-import { IRemoteUser } from '../../../../models/entities/user';
-import { IDelete, getApId, isTombstone, IObject, validPost, validActor } from '../../type';
+import { IDelete, getApId, IObject, validPost, validActor, isTombstone } from '../../type';
 import { toSingle } from '../../../../prelude/array';
+import { IRemoteUser } from '../../../../models/entities/user';
+import { deleteActor } from './actor';
 
 /**
  * 削除アクティビティを捌きます
@@ -41,7 +42,7 @@ export default async (actor: IRemoteUser, activity: IDelete): Promise<string> =>
 	if (validPost.includes(formarType)) {
 		return await deleteNote(actor, uri);
 	} else if (validActor.includes(formarType)) {
-		return `Delete Actor is not implanted`;
+		return await deleteActor(actor, uri);
 	} else {
 		return `Unknown type ${formarType}`;
 	}

src/server/api/endpoints/i/delete-account.ts

@@ -1,9 +1,10 @@
 import $ from 'cafy';
 import * as bcrypt from 'bcryptjs';
 import define from '../../define';
-import { Users, UserProfiles } from '../../../../models';
-import { ensure } from '../../../../prelude/ensure';
+import { UserProfiles, Users } from '../../../../models';
 import { doPostSuspend } from '../../../../services/suspend-user';
+//import { publishUserEvent } from '@/services/stream';
+import { createDeleteAccountJob } from '../../../../queue';
 
 export const meta = {
 	requireCredential: true,
@@ -18,7 +19,11 @@ export const meta = {
 };
 
 export default define(meta, async (ps, user) => {
-	const profile = await UserProfiles.findOne(user.id).then(ensure);
+	const profile = await UserProfiles.findOneOrFail(user.id);
+	const userDetailed = await Users.findOneOrFail(user.id);
+	if (userDetailed.isDeleted) {
+		return;
+	}
 
 	// Compare password
 	const same = await bcrypt.compare(ps.password, profile.password!);
@@ -30,5 +35,12 @@ export default define(meta, async (ps, user) => {
 	// 物理削除する前にDelete activityを送信する
 	await doPostSuspend(user).catch(e => {});
 
-	await Users.delete(user.id);
+	createDeleteAccountJob(user);
+
+	await Users.update(user.id, {
+		isDeleted: true,
+	});
+
+	// Terminate streaming
+	//publishUserEvent(user.id, 'terminate', {});
 });

src/services/drive/delete-file.ts

@@ -60,10 +60,10 @@ export async function deleteFileSync(file: DriveFile, isExpired = false) {
 		await Promise.all(promises);
 	}
 
-	postProcess(file, isExpired);
+	await postProcess(file, isExpired);
 }
 
-function postProcess(file: DriveFile, isExpired = false) {
+async function postProcess(file: DriveFile, isExpired = false) {
 	// リモートファイル期限切れ削除後は直リンクにする
 	if (isExpired && file.userHost !== null && file.uri != null) {
 		DriveFiles.update(file.id, {
@@ -78,7 +78,7 @@ function postProcess(file: DriveFile, isExpired = false) {
 			webpublicAccessKey: 'webpublic-' + uuid(),
 		});
 	} else {
-		DriveFiles.delete(file.id);
+		await DriveFiles.delete(file.id);
 	}
 
 	// 統計を更新

webpack.config.ts

@@ -213,7 +213,7 @@ module.exports = {
 			'.js', '.ts', '.json'
 		],
 		alias: {
-			'@client': __dirname + '/src/client',
+			'@client': __dirname + '/src/client/app',
 			'@': __dirname + '/src',
 			'const.styl': __dirname + '/src/client/const.styl'
 		}