From 4f8ef63111ed015e0328409f3d7a939f82544427 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=AC=B4=EB=9D=BC=EC=BF=A0=EB=AA=A8?= Date: Fri, 6 Sep 2024 17:02:06 +0900 Subject: [PATCH] feat: note reindexing for search service --- packages/backend/src/core/QueueService.ts | 8 ++ packages/backend/src/core/SearchService.ts | 50 +++++++----- .../src/queue/QueueProcessorService.ts | 3 + .../ReindexNotesProcessorService.ts | 77 +++++++++++++++++++ .../backend/src/server/api/EndpointsModule.ts | 4 + packages/backend/src/server/api/endpoints.ts | 2 + .../api/endpoints/admin/reindex-search.ts | 34 ++++++++ 7 files changed, 157 insertions(+), 21 deletions(-) create mode 100644 packages/backend/src/queue/processors/ReindexNotesProcessorService.ts create mode 100644 packages/backend/src/server/api/endpoints/admin/reindex-search.ts diff --git a/packages/backend/src/core/QueueService.ts b/packages/backend/src/core/QueueService.ts index 7c92dd402..3a4724a25 100644 --- a/packages/backend/src/core/QueueService.ts +++ b/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', { diff --git a/packages/backend/src/core/SearchService.ts b/packages/backend/src/core/SearchService.ts index 318b8fea1..1a92ace57 100644 --- a/packages/backend/src/core/SearchService.ts +++ b/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 { + 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 { 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(), new Set()]; + ]) : [new Set(), new Set(), new Set()]; 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 [ - 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(), new Set(), new Set()]; - const notes = (await this.notesRepository.findBy({ + 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; }); - - return notes.sort((a, b) => a.id > b.id ? -1 : 1); + const promises = notes.map(async note => ({ note: note, result: (await this.filter(me, note)) })); + 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); diff --git a/packages/backend/src/queue/QueueProcessorService.ts b/packages/backend/src/queue/QueueProcessorService.ts index 540a58b8b..b69ea10bb 100644 --- a/packages/backend/src/queue/QueueProcessorService.ts +++ b/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); diff --git a/packages/backend/src/queue/processors/ReindexNotesProcessorService.ts b/packages/backend/src/queue/processors/ReindexNotesProcessorService.ts new file mode 100644 index 000000000..2af3e1294 --- /dev/null +++ b/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>): Promise { + 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.'); + } +} diff --git a/packages/backend/src/server/api/EndpointsModule.ts b/packages/backend/src/server/api/EndpointsModule.ts index a09665eb0..ce80f06c0 100644 --- a/packages/backend/src/server/api/EndpointsModule.ts +++ b/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, diff --git a/packages/backend/src/server/api/endpoints.ts b/packages/backend/src/server/api/endpoints.ts index 4cb398094..bac986e9c 100644 --- a/packages/backend/src/server/api/endpoints.ts +++ b/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], diff --git a/packages/backend/src/server/api/endpoints/admin/reindex-search.ts b/packages/backend/src/server/api/endpoints/admin/reindex-search.ts new file mode 100644 index 000000000..0ebc16679 --- /dev/null +++ b/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 { // eslint-disable-line import/no-default-export + constructor( + private queueService: QueueService, + ) { + super(meta, paramDef, async (ps, me) => { + this.queueService.createNoteReindexJob(); + }); + } +}