feat: note reindexing for search service

packages/backend/src/core/QueueService.ts

@@ -164,6 +164,14 @@ export class QueueService {
 		});
 	}
 
+	@bindThis
+	public createNoteReindexJob() {
+		return this.dbQueue.add('reindexNotes', {}, {
+			removeOnComplete: true,
+			removeOnFail: true,
+		});
+	}
+
 	@bindThis
 	public createExportCustomEmojisJob(user: ThinUser) {
 		return this.dbQueue.add('exportCustomEmojis', {

packages/backend/src/core/SearchService.ts

@@ -225,7 +225,7 @@ export class SearchService {
 			await this.elasticsearch.index({
 				index: `${this.elasticsearchNoteIndex}-${createdAt.toISOString().slice(0, 7).replace(/-/g, '')}`,
 				id: note.id,
-				body: body,
+				document: body,
 			}).catch((error: any) => {
 				this.logger.error(error);
 			});
@@ -237,7 +237,7 @@ export class SearchService {
 		// if (!['home', 'public'].includes(note.visibility)) return;
 
 		if (this.meilisearch) {
-			this.meilisearchNoteIndex?.deleteDocument(note.id);
+			await this.meilisearchNoteIndex?.deleteDocument(note.id);
 		} else if (this.elasticsearch) {
 			await this.elasticsearch.delete({
 				index: `${this.elasticsearchNoteIndex}-${this.idService.parse(note.id).date.toISOString().slice(0, 7).replace(/-/g, '')}`,
@@ -248,19 +248,38 @@ export class SearchService {
 		}
 	}
 
+	@bindThis
+	public async unindexAllNotes(): Promise<void> {
+		if (this.meilisearch) {
+			await this.meilisearchNoteIndex?.deleteAllDocuments();
+		} else if (this.elasticsearch) {
+			await this.elasticsearch.deleteByQuery({
+				index: this.elasticsearchNoteIndex + '*' as string,
+				query: {
+					match_all: {},
+				},
+			}).catch((error) => {
+				this.logger.error(error);
+			});
+		}
+	}
+
 	@bindThis
 	private async filter(me: MiUser | null, note: MiNote): Promise<boolean> {
 		const [
 			userIdsWhoMeMuting,
+			userIdsWhoMeBlocking,
 			userIdsWhoBlockingMe,
 		] = me ? await Promise.all([
 			this.cacheService.userMutingsCache.fetch(me.id),
+			this.cacheService.userBlockingCache.fetch(me.id),
 			this.cacheService.userBlockedCache.fetch(me.id),
-		]) : [new Set<string>(), new Set<string>()];
+		]) : [new Set<string>(), new Set<string>(), new Set<string>()];
 		if (me && isUserRelated(note, userIdsWhoBlockingMe)) return false;
+		if (me && isUserRelated(note, userIdsWhoMeBlocking)) return false;
 		if (me && isUserRelated(note, userIdsWhoMeMuting)) return false;
 		if (['followers', 'specified'].includes(note.visibility)) {
-			if (!me) return false;
+			if (me == null) return false;
 			if (note.visibility === 'followers') {
 				const relationship = await this.userEntityService.getRelation(me.id, note.userId);
 				if (relationship.isFollowing) return true;
@@ -358,25 +377,14 @@ export class SearchService {
 
 			const noteIds = res.hits.hits.map((hit: any) => hit._id);
 			if (noteIds.length === 0) return [];
-			const [
+			const notes = await this.notesRepository.findBy({
-				userIdsWhoMeMuting,
-				userIdsWhoMeBlocking,
-				userIdsWhoBlockingMe,
-			] = me ? await Promise.all([
-				this.cacheService.userMutingsCache.fetch(me.id),
-				this.cacheService.userBlockingCache.fetch(me.id),
-				this.cacheService.userBlockedCache.fetch(me.id),
-			]) : [new Set<string>(), new Set<string>(), new Set<string>()];
-			const notes = (await this.notesRepository.findBy({
 				id: In(noteIds),
-			})).filter(note => {
-				if (me && isUserRelated(note, userIdsWhoBlockingMe)) return false;
-				if (me && isUserRelated(note, userIdsWhoMeBlocking)) return false;
-				if (me && isUserRelated(note, userIdsWhoMeMuting)) return false;
-				return true;
 			});
-
+			const promises = notes.map(async note => ({ note: note, result: (await this.filter(me, note)) }));
-			return notes.sort((a, b) => a.id > b.id ? -1 : 1);
+			const data = await Promise.all(promises);
+			const dataFilter = data.filter(d => d.result);
+			const filteredNotes = dataFilter.map(d => d.note);
+			return filteredNotes.sort((a, b) => a.id > b.id ? -1 : 1);
 		} else {
 			const query = this.queryService.makePaginationQuery(this.notesRepository.createQueryBuilder('note'), pagination.sinceId, pagination.untilId);
 

packages/backend/src/queue/QueueProcessorService.ts

@@ -9,6 +9,7 @@ import type { Config } from '@/config.js';
 import { DI } from '@/di-symbols.js';
 import type Logger from '@/logger.js';
 import { bindThis } from '@/decorators.js';
+import { ReindexNotesProcessorService } from "@/queue/processors/ReindexNotesProcessorService.js";
 import { WebhookDeliverProcessorService } from './processors/WebhookDeliverProcessorService.js';
 import { EndedPollNotificationProcessorService } from './processors/EndedPollNotificationProcessorService.js';
 import { DeliverProcessorService } from './processors/DeliverProcessorService.js';
@@ -91,6 +92,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 		private deliverProcessorService: DeliverProcessorService,
 		private inboxProcessorService: InboxProcessorService,
 		private deleteDriveFilesProcessorService: DeleteDriveFilesProcessorService,
+		private reindexNotesProcessorService: ReindexNotesProcessorService,
 		private exportCustomEmojisProcessorService: ExportCustomEmojisProcessorService,
 		private exportNotesProcessorService: ExportNotesProcessorService,
 		private exportClipsProcessorService: ExportClipsProcessorService,
@@ -166,6 +168,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 		this.dbQueueWorker = new Bull.Worker(QUEUE.DB, (job) => {
 			switch (job.name) {
 				case 'deleteDriveFiles': return this.deleteDriveFilesProcessorService.process(job);
+				case 'reindexNotes': return this.reindexNotesProcessorService.process(job);
 				case 'exportCustomEmojis': return this.exportCustomEmojisProcessorService.process(job);
 				case 'exportNotes': return this.exportNotesProcessorService.process(job);
 				case 'exportClips': return this.exportClipsProcessorService.process(job);

packages/backend/src/queue/processors/ReindexNotesProcessorService.ts

@@ -0,0 +1,77 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { Inject, Injectable } from '@nestjs/common';
+import { LessThanOrEqual, MoreThan } from 'typeorm';
+import { DI } from '@/di-symbols.js';
+import type { MiNote, NotesRepository } from '@/models/_.js';
+import type Logger from '@/logger.js';
+import { bindThis } from '@/decorators.js';
+import { SearchService } from '@/core/SearchService.js';
+import { QueueLoggerService } from '../QueueLoggerService.js';
+import type * as Bull from 'bullmq';
+
+@Injectable()
+export class ReindexNotesProcessorService {
+	private logger: Logger;
+
+	constructor(
+		@Inject(DI.notesRepository)
+		private notesRepository: NotesRepository,
+
+		private searchService: SearchService,
+		private queueLoggerService: QueueLoggerService,
+	) {
+		this.logger = this.queueLoggerService.logger.createSubLogger('clean-remote-files');
+	}
+
+	@bindThis
+	public async process(job: Bull.Job<Record<string, unknown>>): Promise<void> {
+		this.logger.info('Removing all indexes from search engine...');
+
+		await this.searchService.unindexAllNotes();
+
+		this.logger.info('Removed all indexes from search engine.');
+		this.logger.info('Indexing all notes to search engine...');
+
+		const lastNote = await this.notesRepository.findOneOrFail({
+			order: { id: 'DESC' },
+		});
+		let indexedCount = 0;
+		let cursor: MiNote['id'] | null = null;
+
+		while (true) {
+			const notes = await this.notesRepository.find({
+				where: {
+					id: LessThanOrEqual(lastNote.id),
+					...(cursor ? { id: MoreThan(cursor) } : {}),
+				},
+				take: 20,
+				order: {
+					id: 'ASC',
+				},
+			});
+
+			if (notes.length === 0) {
+				await job.updateProgress(100);
+				break;
+			}
+
+			cursor = notes[notes.length - 1].id;
+
+			await Promise.all(notes.map(note => this.searchService.indexNote(note)));
+
+			indexedCount += 20;
+
+			const total = await this.notesRepository.countBy({
+				id: LessThanOrEqual(lastNote.id),
+			});
+
+			job.updateProgress(100 / total * indexedCount);
+		}
+
+		this.logger.succ('Successfully re-indexed all notes to search engine.');
+	}
+}

packages/backend/src/server/api/EndpointsModule.ts

@@ -70,6 +70,7 @@ import * as ep___admin_queue_deliverDelayed from './endpoints/admin/queue/delive
 import * as ep___admin_queue_inboxDelayed from './endpoints/admin/queue/inbox-delayed.js';
 import * as ep___admin_queue_promote from './endpoints/admin/queue/promote.js';
 import * as ep___admin_queue_stats from './endpoints/admin/queue/stats.js';
+import * as ep___admin_reindexSearch from './endpoints/admin/reindex-search.js';
 import * as ep___admin_relays_add from './endpoints/admin/relays/add.js';
 import * as ep___admin_relays_list from './endpoints/admin/relays/list.js';
 import * as ep___admin_relays_remove from './endpoints/admin/relays/remove.js';
@@ -462,6 +463,7 @@ const $admin_queue_deliverDelayed: Provider = { provide: 'ep:admin/queue/deliver
 const $admin_queue_inboxDelayed: Provider = { provide: 'ep:admin/queue/inbox-delayed', useClass: ep___admin_queue_inboxDelayed.default };
 const $admin_queue_promote: Provider = { provide: 'ep:admin/queue/promote', useClass: ep___admin_queue_promote.default };
 const $admin_queue_stats: Provider = { provide: 'ep:admin/queue/stats', useClass: ep___admin_queue_stats.default };
+const $admin_reindexSearch: Provider = { provide: 'ep:admin/reindex-search', useClass: ep___admin_reindexSearch.default };
 const $admin_relays_add: Provider = { provide: 'ep:admin/relays/add', useClass: ep___admin_relays_add.default };
 const $admin_relays_list: Provider = { provide: 'ep:admin/relays/list', useClass: ep___admin_relays_list.default };
 const $admin_relays_remove: Provider = { provide: 'ep:admin/relays/remove', useClass: ep___admin_relays_remove.default };
@@ -858,6 +860,7 @@ const $reversi_verify: Provider = { provide: 'ep:reversi/verify', useClass: ep__
 		$admin_queue_inboxDelayed,
 		$admin_queue_promote,
 		$admin_queue_stats,
+		$admin_reindexSearch,
 		$admin_relays_add,
 		$admin_relays_list,
 		$admin_relays_remove,
@@ -1248,6 +1251,7 @@ const $reversi_verify: Provider = { provide: 'ep:reversi/verify', useClass: ep__
 		$admin_queue_inboxDelayed,
 		$admin_queue_promote,
 		$admin_queue_stats,
+		$admin_reindexSearch,
 		$admin_relays_add,
 		$admin_relays_list,
 		$admin_relays_remove,

packages/backend/src/server/api/endpoints.ts

@@ -70,6 +70,7 @@ import * as ep___admin_queue_deliverDelayed from './endpoints/admin/queue/delive
 import * as ep___admin_queue_inboxDelayed from './endpoints/admin/queue/inbox-delayed.js';
 import * as ep___admin_queue_promote from './endpoints/admin/queue/promote.js';
 import * as ep___admin_queue_stats from './endpoints/admin/queue/stats.js';
+import * as ep___admin_reindexSearch from './endpoints/admin/reindex-search.js';
 import * as ep___admin_relays_add from './endpoints/admin/relays/add.js';
 import * as ep___admin_relays_list from './endpoints/admin/relays/list.js';
 import * as ep___admin_relays_remove from './endpoints/admin/relays/remove.js';
@@ -460,6 +461,7 @@ const eps = [
 	['admin/queue/inbox-delayed', ep___admin_queue_inboxDelayed],
 	['admin/queue/promote', ep___admin_queue_promote],
 	['admin/queue/stats', ep___admin_queue_stats],
+	['admin/reindex-search', ep___admin_reindexSearch],
 	['admin/relays/add', ep___admin_relays_add],
 	['admin/relays/list', ep___admin_relays_list],
 	['admin/relays/remove', ep___admin_relays_remove],

packages/backend/src/server/api/endpoints/admin/reindex-search.ts

@@ -0,0 +1,34 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { Injectable } from '@nestjs/common';
+import { Endpoint } from '@/server/api/endpoint-base.js';
+import { QueueService } from '@/core/QueueService.js';
+
+export const meta = {
+	tags: ['admin'],
+
+	requireCredential: true,
+	requireAdmin: true,
+	secure: true,
+	kind: 'write:admin:search',
+} as const;
+
+export const paramDef = {
+	type: 'object',
+	properties: {},
+	required: [],
+} as const;
+
+@Injectable()
+export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint-disable-line import/no-default-export
+	constructor(
+		private queueService: QueueService,
+	) {
+		super(meta, paramDef, async (ps, me) => {
+			this.queueService.createNoteReindexJob();
+		});
+	}
+}