Misskey® Reactions Buffering Technology™ (#14579)
* wip * wip * Update ReactionsBufferingService.ts * Update ReactionsBufferingService.ts * wip * wip * wip * Update ReactionsBufferingService.ts * wip * wip * wip * Update NoteEntityService.ts * wip * wip * wip * wip * Update CHANGELOG.md
This commit is contained in:
		| @@ -103,6 +103,14 @@ redis: | ||||
| #  #prefix: example-prefix | ||||
| #  #db: 1 | ||||
|  | ||||
| #redisForReactions: | ||||
| #  host: redis | ||||
| #  port: 6379 | ||||
| #  #family: 0  # 0=Both, 4=IPv4, 6=IPv6 | ||||
| #  #pass: example-pass | ||||
| #  #prefix: example-prefix | ||||
| #  #db: 1 | ||||
|  | ||||
| #   ┌───────────────────────────┐ | ||||
| #───┘ MeiliSearch configuration └───────────────────────────── | ||||
|  | ||||
|   | ||||
| @@ -106,6 +106,14 @@ redis: | ||||
| #  #prefix: example-prefix | ||||
| #  #db: 1 | ||||
|  | ||||
| #redisForReactions: | ||||
| #  host: redis | ||||
| #  port: 6379 | ||||
| #  #family: 0  # 0=Both, 4=IPv4, 6=IPv6 | ||||
| #  #pass: example-pass | ||||
| #  #prefix: example-prefix | ||||
| #  #db: 1 | ||||
|  | ||||
| #   ┌───────────────────────────┐ | ||||
| #───┘ MeiliSearch configuration └───────────────────────────── | ||||
|  | ||||
|   | ||||
| @@ -172,6 +172,16 @@ redis: | ||||
| #  # You can specify more ioredis options... | ||||
| #  #username: example-username | ||||
|  | ||||
| #redisForReactions: | ||||
| #  host: localhost | ||||
| #  port: 6379 | ||||
| #  #family: 0  # 0=Both, 4=IPv4, 6=IPv6 | ||||
| #  #pass: example-pass | ||||
| #  #prefix: example-prefix | ||||
| #  #db: 1 | ||||
| #  # You can specify more ioredis options... | ||||
| #  #username: example-username | ||||
|  | ||||
| #   ┌───────────────────────────┐ | ||||
| #───┘ MeiliSearch configuration └───────────────────────────── | ||||
|  | ||||
|   | ||||
| @@ -103,6 +103,14 @@ redis: | ||||
| #  #prefix: example-prefix | ||||
| #  #db: 1 | ||||
|  | ||||
| #redisForReactions: | ||||
| #  host: redis | ||||
| #  port: 6379 | ||||
| #  #family: 0  # 0=Both, 4=IPv4, 6=IPv6 | ||||
| #  #pass: example-pass | ||||
| #  #prefix: example-prefix | ||||
| #  #db: 1 | ||||
|  | ||||
| #   ┌───────────────────────────┐ | ||||
| #───┘ MeiliSearch configuration └───────────────────────────── | ||||
|  | ||||
|   | ||||
| @@ -17,6 +17,7 @@ | ||||
| - Fix: 設定変更時のリロード確認ダイアログが複数個表示されることがある問題を修正 | ||||
|  | ||||
| ### Server | ||||
| - Feat: Misskey® Reactions Buffering Technology™ (RBT)により、リアクションの作成負荷を低減することが可能に | ||||
| - Fix: アンテナの書き込み時にキーワードが与えられなかった場合のエラーをApiErrorとして投げるように | ||||
|   - この変更により、公式フロントエンドでは入力の不備が内部エラーとして報告される代わりに一般的なエラーダイアログで報告されます | ||||
| - Fix: ファイルがサイズの制限を超えてアップロードされた際にエラーを返さなかった問題を修正 | ||||
|   | ||||
| @@ -124,6 +124,14 @@ redis: | ||||
| #  #prefix: example-prefix | ||||
| #  #db: 1 | ||||
|  | ||||
| #redisForReactions: | ||||
| #  host: redis | ||||
| #  port: 6379 | ||||
| #  #family: 0  # 0=Both, 4=IPv4, 6=IPv6 | ||||
| #  #pass: example-pass | ||||
| #  #prefix: example-prefix | ||||
| #  #db: 1 | ||||
|  | ||||
| #   ┌───────────────────────────┐ | ||||
| #───┘ MeiliSearch configuration └───────────────────────────── | ||||
|  | ||||
|   | ||||
							
								
								
									
										4
									
								
								locales/index.d.ts
									
									
									
									
										vendored
									
									
								
							
							
						
						
									
										4
									
								
								locales/index.d.ts
									
									
									
									
										vendored
									
									
								
							| @@ -5583,6 +5583,10 @@ export interface Locale extends ILocale { | ||||
|          * 有効にすると、タイムラインがキャッシュされていない場合にDBへ追加で問い合わせを行うフォールバック処理を行います。無効にすると、フォールバック処理を行わないことでさらにサーバーの負荷を軽減することができますが、タイムラインが取得できる範囲に制限が生じます。 | ||||
|          */ | ||||
|         "fanoutTimelineDbFallbackDescription": string; | ||||
|         /** | ||||
|          * 有効にすると、リアクション作成時のパフォーマンスが大幅に向上し、データベースへの負荷を軽減することが可能です。ただし、Redisのメモリ使用量は増加します。 | ||||
|          */ | ||||
|         "reactionsBufferingDescription": string; | ||||
|         /** | ||||
|          * 問い合わせ先URL | ||||
|          */ | ||||
|   | ||||
| @@ -1411,6 +1411,7 @@ _serverSettings: | ||||
|   fanoutTimelineDescription: "有効にすると、各種タイムラインを取得する際のパフォーマンスが大幅に向上し、データベースへの負荷を軽減することが可能です。ただし、Redisのメモリ使用量は増加します。サーバーのメモリ容量が少ない場合、または動作が不安定な場合は無効にすることができます。" | ||||
|   fanoutTimelineDbFallback: "データベースへのフォールバック" | ||||
|   fanoutTimelineDbFallbackDescription: "有効にすると、タイムラインがキャッシュされていない場合にDBへ追加で問い合わせを行うフォールバック処理を行います。無効にすると、フォールバック処理を行わないことでさらにサーバーの負荷を軽減することができますが、タイムラインが取得できる範囲に制限が生じます。" | ||||
|   reactionsBufferingDescription: "有効にすると、リアクション作成時のパフォーマンスが大幅に向上し、データベースへの負荷を軽減することが可能です。ただし、Redisのメモリ使用量は増加します。" | ||||
|   inquiryUrl: "問い合わせ先URL" | ||||
|   inquiryUrlDescription: "サーバー運営者へのお問い合わせフォームのURLや、運営者の連絡先等が記載されたWebページのURLを指定します。" | ||||
|  | ||||
|   | ||||
| @@ -0,0 +1,16 @@ | ||||
| /* | ||||
|  * SPDX-FileCopyrightText: syuilo and misskey-project | ||||
|  * SPDX-License-Identifier: AGPL-3.0-only | ||||
|  */ | ||||
|  | ||||
| export class ReactionsBuffering1726804538569 { | ||||
|     name = 'ReactionsBuffering1726804538569' | ||||
|  | ||||
|     async up(queryRunner) { | ||||
|         await queryRunner.query(`ALTER TABLE "meta" ADD "enableReactionsBuffering" boolean NOT NULL DEFAULT false`); | ||||
|     } | ||||
|  | ||||
|     async down(queryRunner) { | ||||
|         await queryRunner.query(`ALTER TABLE "meta" DROP COLUMN "enableReactionsBuffering"`); | ||||
|     } | ||||
| } | ||||
| @@ -78,11 +78,19 @@ const $redisForTimelines: Provider = { | ||||
| 	inject: [DI.config], | ||||
| }; | ||||
|  | ||||
| const $redisForReactions: Provider = { | ||||
| 	provide: DI.redisForReactions, | ||||
| 	useFactory: (config: Config) => { | ||||
| 		return new Redis.Redis(config.redisForReactions); | ||||
| 	}, | ||||
| 	inject: [DI.config], | ||||
| }; | ||||
|  | ||||
| @Global() | ||||
| @Module({ | ||||
| 	imports: [RepositoryModule], | ||||
| 	providers: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines], | ||||
| 	exports: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines, RepositoryModule], | ||||
| 	providers: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines, $redisForReactions], | ||||
| 	exports: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines, $redisForReactions, RepositoryModule], | ||||
| }) | ||||
| export class GlobalModule implements OnApplicationShutdown { | ||||
| 	constructor( | ||||
| @@ -91,6 +99,7 @@ export class GlobalModule implements OnApplicationShutdown { | ||||
| 		@Inject(DI.redisForPub) private redisForPub: Redis.Redis, | ||||
| 		@Inject(DI.redisForSub) private redisForSub: Redis.Redis, | ||||
| 		@Inject(DI.redisForTimelines) private redisForTimelines: Redis.Redis, | ||||
| 		@Inject(DI.redisForReactions) private redisForReactions: Redis.Redis, | ||||
| 	) { } | ||||
|  | ||||
| 	public async dispose(): Promise<void> { | ||||
| @@ -103,6 +112,7 @@ export class GlobalModule implements OnApplicationShutdown { | ||||
| 			this.redisForPub.disconnect(), | ||||
| 			this.redisForSub.disconnect(), | ||||
| 			this.redisForTimelines.disconnect(), | ||||
| 			this.redisForReactions.disconnect(), | ||||
| 		]); | ||||
| 	} | ||||
|  | ||||
|   | ||||
| @@ -49,6 +49,7 @@ type Source = { | ||||
| 	redisForPubsub?: RedisOptionsSource; | ||||
| 	redisForJobQueue?: RedisOptionsSource; | ||||
| 	redisForTimelines?: RedisOptionsSource; | ||||
| 	redisForReactions?: RedisOptionsSource; | ||||
| 	meilisearch?: { | ||||
| 		host: string; | ||||
| 		port: string; | ||||
| @@ -171,6 +172,7 @@ export type Config = { | ||||
| 	redisForPubsub: RedisOptions & RedisOptionsSource; | ||||
| 	redisForJobQueue: RedisOptions & RedisOptionsSource; | ||||
| 	redisForTimelines: RedisOptions & RedisOptionsSource; | ||||
| 	redisForReactions: RedisOptions & RedisOptionsSource; | ||||
| 	sentryForBackend: { options: Partial<Sentry.NodeOptions>; enableNodeProfiling: boolean; } | undefined; | ||||
| 	sentryForFrontend: { options: Partial<Sentry.NodeOptions> } | undefined; | ||||
| 	perChannelMaxNoteCacheCount: number; | ||||
| @@ -251,6 +253,7 @@ export function loadConfig(): Config { | ||||
| 		redisForPubsub: config.redisForPubsub ? convertRedisOptions(config.redisForPubsub, host) : redis, | ||||
| 		redisForJobQueue: config.redisForJobQueue ? convertRedisOptions(config.redisForJobQueue, host) : redis, | ||||
| 		redisForTimelines: config.redisForTimelines ? convertRedisOptions(config.redisForTimelines, host) : redis, | ||||
| 		redisForReactions: config.redisForReactions ? convertRedisOptions(config.redisForReactions, host) : redis, | ||||
| 		sentryForBackend: config.sentryForBackend, | ||||
| 		sentryForFrontend: config.sentryForFrontend, | ||||
| 		id: config.id, | ||||
|   | ||||
| @@ -8,6 +8,8 @@ export const MAX_NOTE_TEXT_LENGTH = 3000; | ||||
| export const USER_ONLINE_THRESHOLD = 1000 * 60 * 10; // 10min | ||||
| export const USER_ACTIVE_THRESHOLD = 1000 * 60 * 60 * 24 * 3; // 3days | ||||
|  | ||||
| export const PER_NOTE_REACTION_USER_PAIR_CACHE_MAX = 16; | ||||
|  | ||||
| //#region hard limits | ||||
| // If you change DB_* values, you must also change the DB schema. | ||||
|  | ||||
|   | ||||
| @@ -50,6 +50,7 @@ import { PollService } from './PollService.js'; | ||||
| import { PushNotificationService } from './PushNotificationService.js'; | ||||
| import { QueryService } from './QueryService.js'; | ||||
| import { ReactionService } from './ReactionService.js'; | ||||
| import { ReactionsBufferingService } from './ReactionsBufferingService.js'; | ||||
| import { RelayService } from './RelayService.js'; | ||||
| import { RoleService } from './RoleService.js'; | ||||
| import { S3Service } from './S3Service.js'; | ||||
| @@ -193,6 +194,7 @@ const $ProxyAccountService: Provider = { provide: 'ProxyAccountService', useExis | ||||
| const $PushNotificationService: Provider = { provide: 'PushNotificationService', useExisting: PushNotificationService }; | ||||
| const $QueryService: Provider = { provide: 'QueryService', useExisting: QueryService }; | ||||
| const $ReactionService: Provider = { provide: 'ReactionService', useExisting: ReactionService }; | ||||
| const $ReactionsBufferingService: Provider = { provide: 'ReactionsBufferingService', useExisting: ReactionsBufferingService }; | ||||
| const $RelayService: Provider = { provide: 'RelayService', useExisting: RelayService }; | ||||
| const $RoleService: Provider = { provide: 'RoleService', useExisting: RoleService }; | ||||
| const $S3Service: Provider = { provide: 'S3Service', useExisting: S3Service }; | ||||
| @@ -342,6 +344,7 @@ const $ApQuestionService: Provider = { provide: 'ApQuestionService', useExisting | ||||
| 		PushNotificationService, | ||||
| 		QueryService, | ||||
| 		ReactionService, | ||||
| 		ReactionsBufferingService, | ||||
| 		RelayService, | ||||
| 		RoleService, | ||||
| 		S3Service, | ||||
| @@ -487,6 +490,7 @@ const $ApQuestionService: Provider = { provide: 'ApQuestionService', useExisting | ||||
| 		$PushNotificationService, | ||||
| 		$QueryService, | ||||
| 		$ReactionService, | ||||
| 		$ReactionsBufferingService, | ||||
| 		$RelayService, | ||||
| 		$RoleService, | ||||
| 		$S3Service, | ||||
| @@ -633,6 +637,7 @@ const $ApQuestionService: Provider = { provide: 'ApQuestionService', useExisting | ||||
| 		PushNotificationService, | ||||
| 		QueryService, | ||||
| 		ReactionService, | ||||
| 		ReactionsBufferingService, | ||||
| 		RelayService, | ||||
| 		RoleService, | ||||
| 		S3Service, | ||||
| @@ -777,6 +782,7 @@ const $ApQuestionService: Provider = { provide: 'ApQuestionService', useExisting | ||||
| 		$PushNotificationService, | ||||
| 		$QueryService, | ||||
| 		$ReactionService, | ||||
| 		$ReactionsBufferingService, | ||||
| 		$RelayService, | ||||
| 		$RoleService, | ||||
| 		$S3Service, | ||||
|   | ||||
| @@ -87,6 +87,12 @@ export class QueueService { | ||||
| 			repeat: { pattern: '*/5 * * * *' }, | ||||
| 			removeOnComplete: true, | ||||
| 		}); | ||||
|  | ||||
| 		this.systemQueue.add('bakeBufferedReactions', { | ||||
| 		}, { | ||||
| 			repeat: { pattern: '0 0 * * *' }, | ||||
| 			removeOnComplete: true, | ||||
| 		}); | ||||
| 	} | ||||
|  | ||||
| 	@bindThis | ||||
|   | ||||
| @@ -4,7 +4,6 @@ | ||||
|  */ | ||||
|  | ||||
| import { Inject, Injectable } from '@nestjs/common'; | ||||
| import * as Redis from 'ioredis'; | ||||
| import { DI } from '@/di-symbols.js'; | ||||
| import type { EmojisRepository, NoteReactionsRepository, UsersRepository, NotesRepository } from '@/models/_.js'; | ||||
| import { IdentifiableError } from '@/misc/identifiable-error.js'; | ||||
| @@ -30,9 +29,10 @@ import { RoleService } from '@/core/RoleService.js'; | ||||
| import { FeaturedService } from '@/core/FeaturedService.js'; | ||||
| import { trackPromise } from '@/misc/promise-tracker.js'; | ||||
| import { isQuote, isRenote } from '@/misc/is-renote.js'; | ||||
| import { ReactionsBufferingService } from '@/core/ReactionsBufferingService.js'; | ||||
| import { PER_NOTE_REACTION_USER_PAIR_CACHE_MAX } from '@/const.js'; | ||||
|  | ||||
| const FALLBACK = '\u2764'; | ||||
| const PER_NOTE_REACTION_USER_PAIR_CACHE_MAX = 16; | ||||
|  | ||||
| const legacies: Record<string, string> = { | ||||
| 	'like': '👍', | ||||
| @@ -71,9 +71,6 @@ const decodeCustomEmojiRegexp = /^:([\w+-]+)(?:@([\w.-]+))?:$/; | ||||
| @Injectable() | ||||
| export class ReactionService { | ||||
| 	constructor( | ||||
| 		@Inject(DI.redis) | ||||
| 		private redisClient: Redis.Redis, | ||||
|  | ||||
| 		@Inject(DI.usersRepository) | ||||
| 		private usersRepository: UsersRepository, | ||||
|  | ||||
| @@ -93,6 +90,7 @@ export class ReactionService { | ||||
| 		private userEntityService: UserEntityService, | ||||
| 		private noteEntityService: NoteEntityService, | ||||
| 		private userBlockingService: UserBlockingService, | ||||
| 		private reactionsBufferingService: ReactionsBufferingService, | ||||
| 		private idService: IdService, | ||||
| 		private featuredService: FeaturedService, | ||||
| 		private globalEventService: GlobalEventService, | ||||
| @@ -174,7 +172,6 @@ export class ReactionService { | ||||
| 			reaction, | ||||
| 		}; | ||||
|  | ||||
| 		// Create reaction | ||||
| 		try { | ||||
| 			await this.noteReactionsRepository.insert(record); | ||||
| 		} catch (e) { | ||||
| @@ -198,16 +195,25 @@ export class ReactionService { | ||||
| 		} | ||||
|  | ||||
| 		// Increment reactions count | ||||
| 		const sql = `jsonb_set("reactions", '{${reaction}}', (COALESCE("reactions"->>'${reaction}', '0')::int + 1)::text::jsonb)`; | ||||
| 		await this.notesRepository.createQueryBuilder().update() | ||||
| 			.set({ | ||||
| 				reactions: () => sql, | ||||
| 				...(note.reactionAndUserPairCache.length < PER_NOTE_REACTION_USER_PAIR_CACHE_MAX ? { | ||||
| 					reactionAndUserPairCache: () => `array_append("reactionAndUserPairCache", '${user.id}/${reaction}')`, | ||||
| 				} : {}), | ||||
| 			}) | ||||
| 			.where('id = :id', { id: note.id }) | ||||
| 			.execute(); | ||||
| 		if (meta.enableReactionsBuffering) { | ||||
| 			await this.reactionsBufferingService.create(note.id, user.id, reaction, note.reactionAndUserPairCache); | ||||
|  | ||||
| 			// for debugging | ||||
| 			if (reaction === ':angry_ai:') { | ||||
| 				this.reactionsBufferingService.bake(); | ||||
| 			} | ||||
| 		} else { | ||||
| 			const sql = `jsonb_set("reactions", '{${reaction}}', (COALESCE("reactions"->>'${reaction}', '0')::int + 1)::text::jsonb)`; | ||||
| 			await this.notesRepository.createQueryBuilder().update() | ||||
| 				.set({ | ||||
| 					reactions: () => sql, | ||||
| 					...(note.reactionAndUserPairCache.length < PER_NOTE_REACTION_USER_PAIR_CACHE_MAX ? { | ||||
| 						reactionAndUserPairCache: () => `array_append("reactionAndUserPairCache", '${user.id}/${reaction}')`, | ||||
| 					} : {}), | ||||
| 				}) | ||||
| 				.where('id = :id', { id: note.id }) | ||||
| 				.execute(); | ||||
| 		} | ||||
|  | ||||
| 		// 30%の確率、セルフではない、3日以内に投稿されたノートの場合ハイライト用ランキング更新 | ||||
| 		if ( | ||||
| @@ -304,15 +310,21 @@ export class ReactionService { | ||||
| 			throw new IdentifiableError('60527ec9-b4cb-4a88-a6bd-32d3ad26817d', 'not reacted'); | ||||
| 		} | ||||
|  | ||||
| 		const meta = await this.metaService.fetch(); | ||||
|  | ||||
| 		// Decrement reactions count | ||||
| 		const sql = `jsonb_set("reactions", '{${exist.reaction}}', (COALESCE("reactions"->>'${exist.reaction}', '0')::int - 1)::text::jsonb)`; | ||||
| 		await this.notesRepository.createQueryBuilder().update() | ||||
| 			.set({ | ||||
| 				reactions: () => sql, | ||||
| 				reactionAndUserPairCache: () => `array_remove("reactionAndUserPairCache", '${user.id}/${exist.reaction}')`, | ||||
| 			}) | ||||
| 			.where('id = :id', { id: note.id }) | ||||
| 			.execute(); | ||||
| 		if (meta.enableReactionsBuffering) { | ||||
| 			await this.reactionsBufferingService.delete(note.id, user.id, exist.reaction); | ||||
| 		} else { | ||||
| 			const sql = `jsonb_set("reactions", '{${exist.reaction}}', (COALESCE("reactions"->>'${exist.reaction}', '0')::int - 1)::text::jsonb)`; | ||||
| 			await this.notesRepository.createQueryBuilder().update() | ||||
| 				.set({ | ||||
| 					reactions: () => sql, | ||||
| 					reactionAndUserPairCache: () => `array_remove("reactionAndUserPairCache", '${user.id}/${exist.reaction}')`, | ||||
| 				}) | ||||
| 				.where('id = :id', { id: note.id }) | ||||
| 				.execute(); | ||||
| 		} | ||||
|  | ||||
| 		this.globalEventService.publishNoteStream(note.id, 'unreacted', { | ||||
| 			reaction: this.decodeReaction(exist.reaction).reaction, | ||||
|   | ||||
							
								
								
									
										162
									
								
								packages/backend/src/core/ReactionsBufferingService.ts
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										162
									
								
								packages/backend/src/core/ReactionsBufferingService.ts
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,162 @@ | ||||
| /* | ||||
|  * SPDX-FileCopyrightText: syuilo and misskey-project | ||||
|  * SPDX-License-Identifier: AGPL-3.0-only | ||||
|  */ | ||||
|  | ||||
| import { Inject, Injectable } from '@nestjs/common'; | ||||
| import * as Redis from 'ioredis'; | ||||
| import { DI } from '@/di-symbols.js'; | ||||
| import type { MiNote } from '@/models/Note.js'; | ||||
| import { bindThis } from '@/decorators.js'; | ||||
| import type { MiUser, NotesRepository } from '@/models/_.js'; | ||||
| import type { Config } from '@/config.js'; | ||||
| import { PER_NOTE_REACTION_USER_PAIR_CACHE_MAX } from '@/const.js'; | ||||
|  | ||||
| const REDIS_DELTA_PREFIX = 'reactionsBufferDeltas'; | ||||
| const REDIS_PAIR_PREFIX = 'reactionsBufferPairs'; | ||||
|  | ||||
| @Injectable() | ||||
| export class ReactionsBufferingService { | ||||
| 	constructor( | ||||
| 		@Inject(DI.config) | ||||
| 		private config: Config, | ||||
|  | ||||
| 		@Inject(DI.redisForReactions) | ||||
| 		private redisForReactions: Redis.Redis, // TODO: 専用のRedisインスタンスにする | ||||
|  | ||||
| 		@Inject(DI.notesRepository) | ||||
| 		private notesRepository: NotesRepository, | ||||
| 	) { | ||||
| 	} | ||||
|  | ||||
| 	@bindThis | ||||
| 	public async create(noteId: MiNote['id'], userId: MiUser['id'], reaction: string, currentPairs: string[]): Promise<void> { | ||||
| 		const pipeline = this.redisForReactions.pipeline(); | ||||
| 		pipeline.hincrby(`${REDIS_DELTA_PREFIX}:${noteId}`, reaction, 1); | ||||
| 		for (let i = 0; i < currentPairs.length; i++) { | ||||
| 			pipeline.zadd(`${REDIS_PAIR_PREFIX}:${noteId}`, i, currentPairs[i]); | ||||
| 		} | ||||
| 		pipeline.zadd(`${REDIS_PAIR_PREFIX}:${noteId}`, Date.now(), `${userId}/${reaction}`); | ||||
| 		pipeline.zremrangebyrank(`${REDIS_PAIR_PREFIX}:${noteId}`, 0, -(PER_NOTE_REACTION_USER_PAIR_CACHE_MAX + 1)); | ||||
| 		await pipeline.exec(); | ||||
| 	} | ||||
|  | ||||
| 	@bindThis | ||||
| 	public async delete(noteId: MiNote['id'], userId: MiUser['id'], reaction: string): Promise<void> { | ||||
| 		const pipeline = this.redisForReactions.pipeline(); | ||||
| 		pipeline.hincrby(`${REDIS_DELTA_PREFIX}:${noteId}`, reaction, -1); | ||||
| 		pipeline.zrem(`${REDIS_PAIR_PREFIX}:${noteId}`, `${userId}/${reaction}`); | ||||
| 		// TODO: 「消した要素一覧」も持っておかないとcreateされた時に上書きされて復活する | ||||
| 		await pipeline.exec(); | ||||
| 	} | ||||
|  | ||||
| 	@bindThis | ||||
| 	public async get(noteId: MiNote['id']): Promise<{ | ||||
| 		deltas: Record<string, number>; | ||||
| 		pairs: ([MiUser['id'], string])[]; | ||||
| 	}> { | ||||
| 		const pipeline = this.redisForReactions.pipeline(); | ||||
| 		pipeline.hgetall(`${REDIS_DELTA_PREFIX}:${noteId}`); | ||||
| 		pipeline.zrange(`${REDIS_PAIR_PREFIX}:${noteId}`, 0, -1); | ||||
| 		const results = await pipeline.exec(); | ||||
|  | ||||
| 		const resultDeltas = results![0][1] as Record<string, string>; | ||||
| 		const resultPairs = results![1][1] as string[]; | ||||
|  | ||||
| 		const deltas = {} as Record<string, number>; | ||||
| 		for (const [name, count] of Object.entries(resultDeltas)) { | ||||
| 			deltas[name] = parseInt(count); | ||||
| 		} | ||||
|  | ||||
| 		const pairs = resultPairs.map(x => x.split('/') as [MiUser['id'], string]); | ||||
|  | ||||
| 		return { | ||||
| 			deltas, | ||||
| 			pairs, | ||||
| 		}; | ||||
| 	} | ||||
|  | ||||
| 	@bindThis | ||||
| 	public async getMany(noteIds: MiNote['id'][]): Promise<Map<MiNote['id'], { | ||||
| 		deltas: Record<string, number>; | ||||
| 		pairs: ([MiUser['id'], string])[]; | ||||
| 	}>> { | ||||
| 		const map = new Map<MiNote['id'], { | ||||
| 			deltas: Record<string, number>; | ||||
| 			pairs: ([MiUser['id'], string])[]; | ||||
| 		}>(); | ||||
|  | ||||
| 		const pipeline = this.redisForReactions.pipeline(); | ||||
| 		for (const noteId of noteIds) { | ||||
| 			pipeline.hgetall(`${REDIS_DELTA_PREFIX}:${noteId}`); | ||||
| 			pipeline.zrange(`${REDIS_PAIR_PREFIX}:${noteId}`, 0, -1); | ||||
| 		} | ||||
| 		const results = await pipeline.exec(); | ||||
|  | ||||
| 		const opsForEachNotes = 2; | ||||
| 		for (let i = 0; i < noteIds.length; i++) { | ||||
| 			const noteId = noteIds[i]; | ||||
| 			const resultDeltas = results![i * opsForEachNotes][1] as Record<string, string>; | ||||
| 			const resultPairs = results![i * opsForEachNotes + 1][1] as string[]; | ||||
|  | ||||
| 			const deltas = {} as Record<string, number>; | ||||
| 			for (const [name, count] of Object.entries(resultDeltas)) { | ||||
| 				deltas[name] = parseInt(count); | ||||
| 			} | ||||
|  | ||||
| 			const pairs = resultPairs.map(x => x.split('/') as [MiUser['id'], string]); | ||||
|  | ||||
| 			map.set(noteId, { | ||||
| 				deltas, | ||||
| 				pairs, | ||||
| 			}); | ||||
| 		} | ||||
|  | ||||
| 		return map; | ||||
| 	} | ||||
|  | ||||
| 	// TODO: scanは重い可能性があるので、別途 bufferedNoteIds を直接Redis上に持っておいてもいいかもしれない | ||||
| 	@bindThis | ||||
| 	public async bake(): Promise<void> { | ||||
| 		const bufferedNoteIds = []; | ||||
| 		let cursor = '0'; | ||||
| 		do { | ||||
| 			// https://github.com/redis/ioredis#transparent-key-prefixing | ||||
| 			const result = await this.redisForReactions.scan( | ||||
| 				cursor, | ||||
| 				'MATCH', | ||||
| 				`${this.config.redis.prefix}:${REDIS_DELTA_PREFIX}:*`, | ||||
| 				'COUNT', | ||||
| 				'1000'); | ||||
|  | ||||
| 			cursor = result[0]; | ||||
| 			bufferedNoteIds.push(...result[1].map(x => x.replace(`${this.config.redis.prefix}:${REDIS_DELTA_PREFIX}:`, ''))); | ||||
| 		} while (cursor !== '0'); | ||||
|  | ||||
| 		const bufferedMap = await this.getMany(bufferedNoteIds); | ||||
|  | ||||
| 		// clear | ||||
| 		const pipeline = this.redisForReactions.pipeline(); | ||||
| 		for (const noteId of bufferedNoteIds) { | ||||
| 			pipeline.del(`${REDIS_DELTA_PREFIX}:${noteId}`); | ||||
| 			pipeline.del(`${REDIS_PAIR_PREFIX}:${noteId}`); | ||||
| 		} | ||||
| 		await pipeline.exec(); | ||||
|  | ||||
| 		// TODO: SQL一個にまとめたい | ||||
| 		for (const [noteId, buffered] of bufferedMap) { | ||||
| 			const sql = Object.entries(buffered.deltas) | ||||
| 				.map(([reaction, count]) => | ||||
| 					`jsonb_set("reactions", '{${reaction}}', (COALESCE("reactions"->>'${reaction}', '0')::int + ${count})::text::jsonb)`) | ||||
| 				.join(' || '); | ||||
|  | ||||
| 			this.notesRepository.createQueryBuilder().update() | ||||
| 				.set({ | ||||
| 					reactions: () => sql, | ||||
| 					reactionAndUserPairCache: buffered.pairs.map(x => x.join('/')), | ||||
| 				}) | ||||
| 				.where('id = :id', { id: noteId }) | ||||
| 				.execute(); | ||||
| 		} | ||||
| 	} | ||||
| } | ||||
| @@ -11,24 +11,39 @@ import type { Packed } from '@/misc/json-schema.js'; | ||||
| import { awaitAll } from '@/misc/prelude/await-all.js'; | ||||
| import type { MiUser } from '@/models/User.js'; | ||||
| import type { MiNote } from '@/models/Note.js'; | ||||
| import type { MiNoteReaction } from '@/models/NoteReaction.js'; | ||||
| import type { UsersRepository, NotesRepository, FollowingsRepository, PollsRepository, PollVotesRepository, NoteReactionsRepository, ChannelsRepository } from '@/models/_.js'; | ||||
| import { bindThis } from '@/decorators.js'; | ||||
| import { DebounceLoader } from '@/misc/loader.js'; | ||||
| import { IdService } from '@/core/IdService.js'; | ||||
| import { ReactionsBufferingService } from '@/core/ReactionsBufferingService.js'; | ||||
| import { MetaService } from '@/core/MetaService.js'; | ||||
| import type { OnModuleInit } from '@nestjs/common'; | ||||
| import type { CustomEmojiService } from '../CustomEmojiService.js'; | ||||
| import type { ReactionService } from '../ReactionService.js'; | ||||
| import type { UserEntityService } from './UserEntityService.js'; | ||||
| import type { DriveFileEntityService } from './DriveFileEntityService.js'; | ||||
|  | ||||
| function mergeReactions(src: Record<string, number>, delta: Record<string, number>) { | ||||
| 	const reactions = { ...src }; | ||||
| 	for (const [name, count] of Object.entries(delta)) { | ||||
| 		if (reactions[name] != null) { | ||||
| 			reactions[name] += count; | ||||
| 		} else { | ||||
| 			reactions[name] = count; | ||||
| 		} | ||||
| 	} | ||||
| 	return reactions; | ||||
| } | ||||
|  | ||||
| @Injectable() | ||||
| export class NoteEntityService implements OnModuleInit { | ||||
| 	private userEntityService: UserEntityService; | ||||
| 	private driveFileEntityService: DriveFileEntityService; | ||||
| 	private customEmojiService: CustomEmojiService; | ||||
| 	private reactionService: ReactionService; | ||||
| 	private reactionsBufferingService: ReactionsBufferingService; | ||||
| 	private idService: IdService; | ||||
| 	private metaService: MetaService; | ||||
| 	private noteLoader = new DebounceLoader(this.findNoteOrFail); | ||||
|  | ||||
| 	constructor( | ||||
| @@ -59,6 +74,9 @@ export class NoteEntityService implements OnModuleInit { | ||||
| 		//private driveFileEntityService: DriveFileEntityService, | ||||
| 		//private customEmojiService: CustomEmojiService, | ||||
| 		//private reactionService: ReactionService, | ||||
| 		//private reactionsBufferingService: ReactionsBufferingService, | ||||
| 		//private idService: IdService, | ||||
| 		//private metaService: MetaService, | ||||
| 	) { | ||||
| 	} | ||||
|  | ||||
| @@ -67,7 +85,9 @@ export class NoteEntityService implements OnModuleInit { | ||||
| 		this.driveFileEntityService = this.moduleRef.get('DriveFileEntityService'); | ||||
| 		this.customEmojiService = this.moduleRef.get('CustomEmojiService'); | ||||
| 		this.reactionService = this.moduleRef.get('ReactionService'); | ||||
| 		this.reactionsBufferingService = this.moduleRef.get('ReactionsBufferingService'); | ||||
| 		this.idService = this.moduleRef.get('IdService'); | ||||
| 		this.metaService = this.moduleRef.get('MetaService'); | ||||
| 	} | ||||
|  | ||||
| 	@bindThis | ||||
| @@ -287,6 +307,7 @@ export class NoteEntityService implements OnModuleInit { | ||||
| 			skipHide?: boolean; | ||||
| 			withReactionAndUserPairCache?: boolean; | ||||
| 			_hint_?: { | ||||
| 				bufferdReactions: Map<MiNote['id'], { deltas: Record<string, number>; pairs: ([MiUser['id'], string])[] }> | null; | ||||
| 				myReactions: Map<MiNote['id'], string | null>; | ||||
| 				packedFiles: Map<MiNote['fileIds'][number], Packed<'DriveFile'> | null>; | ||||
| 				packedUsers: Map<MiUser['id'], Packed<'UserLite'>> | ||||
| @@ -303,6 +324,16 @@ export class NoteEntityService implements OnModuleInit { | ||||
| 		const note = typeof src === 'object' ? src : await this.noteLoader.load(src); | ||||
| 		const host = note.userHost; | ||||
|  | ||||
| 		const bufferdReactions = opts._hint_?.bufferdReactions != null ? (opts._hint_.bufferdReactions.get(note.id) ?? { deltas: {}, pairs: [] }) : await this.reactionsBufferingService.get(note.id); | ||||
| 		const reactions = mergeReactions(note.reactions, bufferdReactions.deltas ?? {}); | ||||
| 		for (const [name, count] of Object.entries(reactions)) { | ||||
| 			if (count <= 0) { | ||||
| 				delete reactions[name]; | ||||
| 			} | ||||
| 		} | ||||
|  | ||||
| 		const reactionAndUserPairCache = note.reactionAndUserPairCache.concat(bufferdReactions.pairs.map(x => x.join('/'))); | ||||
|  | ||||
| 		let text = note.text; | ||||
|  | ||||
| 		if (note.name && (note.url ?? note.uri)) { | ||||
| @@ -315,7 +346,7 @@ export class NoteEntityService implements OnModuleInit { | ||||
| 				: await this.channelsRepository.findOneBy({ id: note.channelId }) | ||||
| 			: null; | ||||
|  | ||||
| 		const reactionEmojiNames = Object.keys(note.reactions) | ||||
| 		const reactionEmojiNames = Object.keys(reactions) | ||||
| 			.filter(x => x.startsWith(':') && x.includes('@') && !x.includes('@.')) // リモートカスタム絵文字のみ | ||||
| 			.map(x => this.reactionService.decodeReaction(x).reaction.replaceAll(':', '')); | ||||
| 		const packedFiles = options?._hint_?.packedFiles; | ||||
| @@ -334,10 +365,10 @@ export class NoteEntityService implements OnModuleInit { | ||||
| 			visibleUserIds: note.visibility === 'specified' ? note.visibleUserIds : undefined, | ||||
| 			renoteCount: note.renoteCount, | ||||
| 			repliesCount: note.repliesCount, | ||||
| 			reactionCount: Object.values(note.reactions).reduce((a, b) => a + b, 0), | ||||
| 			reactions: this.reactionService.convertLegacyReactions(note.reactions), | ||||
| 			reactionCount: Object.values(reactions).reduce((a, b) => a + b, 0), | ||||
| 			reactions: reactions, | ||||
| 			reactionEmojis: this.customEmojiService.populateEmojis(reactionEmojiNames, host), | ||||
| 			reactionAndUserPairCache: opts.withReactionAndUserPairCache ? note.reactionAndUserPairCache : undefined, | ||||
| 			reactionAndUserPairCache: opts.withReactionAndUserPairCache ? reactionAndUserPairCache : undefined, | ||||
| 			emojis: host != null ? this.customEmojiService.populateEmojis(note.emojis, host) : undefined, | ||||
| 			tags: note.tags.length > 0 ? note.tags : undefined, | ||||
| 			fileIds: note.fileIds, | ||||
| @@ -376,8 +407,12 @@ export class NoteEntityService implements OnModuleInit { | ||||
|  | ||||
| 				poll: note.hasPoll ? this.populatePoll(note, meId) : undefined, | ||||
|  | ||||
| 				...(meId && Object.keys(note.reactions).length > 0 ? { | ||||
| 					myReaction: this.populateMyReaction(note, meId, options?._hint_), | ||||
| 				...(meId && Object.keys(reactions).length > 0 ? { | ||||
| 					myReaction: this.populateMyReaction({ | ||||
| 						id: note.id, | ||||
| 						reactions: reactions, | ||||
| 						reactionAndUserPairCache: reactionAndUserPairCache, | ||||
| 					}, meId, options?._hint_), | ||||
| 				} : {}), | ||||
| 			} : {}), | ||||
| 		}); | ||||
| @@ -400,6 +435,10 @@ export class NoteEntityService implements OnModuleInit { | ||||
| 	) { | ||||
| 		if (notes.length === 0) return []; | ||||
|  | ||||
| 		const meta = await this.metaService.fetch(); | ||||
|  | ||||
| 		const bufferdReactions = meta.enableReactionsBuffering ? await this.reactionsBufferingService.getMany(notes.map(x => x.id)) : null; | ||||
|  | ||||
| 		const meId = me ? me.id : null; | ||||
| 		const myReactionsMap = new Map<MiNote['id'], string | null>(); | ||||
| 		if (meId) { | ||||
| @@ -410,23 +449,33 @@ export class NoteEntityService implements OnModuleInit { | ||||
|  | ||||
| 			for (const note of notes) { | ||||
| 				if (note.renote && (note.text == null && note.fileIds.length === 0)) { // pure renote | ||||
| 					const reactionsCount = Object.values(note.renote.reactions).reduce((a, b) => a + b, 0); | ||||
| 					const reactionsCount = Object.values(mergeReactions(note.renote.reactions, bufferdReactions?.get(note.renote.id)?.deltas ?? {})).reduce((a, b) => a + b, 0); | ||||
| 					if (reactionsCount === 0) { | ||||
| 						myReactionsMap.set(note.renote.id, null); | ||||
| 					} else if (reactionsCount <= note.renote.reactionAndUserPairCache.length) { | ||||
| 						const pair = note.renote.reactionAndUserPairCache.find(p => p.startsWith(meId)); | ||||
| 						myReactionsMap.set(note.renote.id, pair ? pair.split('/')[1] : null); | ||||
| 					} else if (reactionsCount <= note.renote.reactionAndUserPairCache.length + (bufferdReactions?.get(note.renote.id)?.pairs.length ?? 0)) { | ||||
| 						const pairInBuffer = bufferdReactions?.get(note.renote.id)?.pairs.find(p => p[0] === meId); | ||||
| 						if (pairInBuffer) { | ||||
| 							myReactionsMap.set(note.renote.id, pairInBuffer[1]); | ||||
| 						} else { | ||||
| 							const pair = note.renote.reactionAndUserPairCache.find(p => p.startsWith(meId)); | ||||
| 							myReactionsMap.set(note.renote.id, pair ? pair.split('/')[1] : null); | ||||
| 						} | ||||
| 					} else { | ||||
| 						idsNeedFetchMyReaction.add(note.renote.id); | ||||
| 					} | ||||
| 				} else { | ||||
| 					if (note.id < oldId) { | ||||
| 						const reactionsCount = Object.values(note.reactions).reduce((a, b) => a + b, 0); | ||||
| 						const reactionsCount = Object.values(mergeReactions(note.reactions, bufferdReactions?.get(note.id)?.deltas ?? {})).reduce((a, b) => a + b, 0); | ||||
| 						if (reactionsCount === 0) { | ||||
| 							myReactionsMap.set(note.id, null); | ||||
| 						} else if (reactionsCount <= note.reactionAndUserPairCache.length) { | ||||
| 							const pair = note.reactionAndUserPairCache.find(p => p.startsWith(meId)); | ||||
| 							myReactionsMap.set(note.id, pair ? pair.split('/')[1] : null); | ||||
| 						} else if (reactionsCount <= note.reactionAndUserPairCache.length + (bufferdReactions?.get(note.id)?.pairs.length ?? 0)) { | ||||
| 							const pairInBuffer = bufferdReactions?.get(note.id)?.pairs.find(p => p[0] === meId); | ||||
| 							if (pairInBuffer) { | ||||
| 								myReactionsMap.set(note.id, pairInBuffer[1]); | ||||
| 							} else { | ||||
| 								const pair = note.reactionAndUserPairCache.find(p => p.startsWith(meId)); | ||||
| 								myReactionsMap.set(note.id, pair ? pair.split('/')[1] : null); | ||||
| 							} | ||||
| 						} else { | ||||
| 							idsNeedFetchMyReaction.add(note.id); | ||||
| 						} | ||||
| @@ -461,6 +510,7 @@ export class NoteEntityService implements OnModuleInit { | ||||
| 		return await Promise.all(notes.map(n => this.pack(n, me, { | ||||
| 			...options, | ||||
| 			_hint_: { | ||||
| 				bufferdReactions, | ||||
| 				myReactions: myReactionsMap, | ||||
| 				packedFiles, | ||||
| 				packedUsers, | ||||
|   | ||||
| @@ -11,6 +11,7 @@ export const DI = { | ||||
| 	redisForPub: Symbol('redisForPub'), | ||||
| 	redisForSub: Symbol('redisForSub'), | ||||
| 	redisForTimelines: Symbol('redisForTimelines'), | ||||
| 	redisForReactions: Symbol('redisForReactions'), | ||||
|  | ||||
| 	//#region Repositories | ||||
| 	usersRepository: Symbol('usersRepository'), | ||||
|   | ||||
| @@ -589,6 +589,11 @@ export class MiMeta { | ||||
| 	}) | ||||
| 	public perUserListTimelineCacheMax: number; | ||||
|  | ||||
| 	@Column('boolean', { | ||||
| 		default: false, | ||||
| 	}) | ||||
| 	public enableReactionsBuffering: boolean; | ||||
|  | ||||
| 	@Column('integer', { | ||||
| 		default: 0, | ||||
| 	}) | ||||
|   | ||||
| @@ -14,6 +14,7 @@ import { InboxProcessorService } from './processors/InboxProcessorService.js'; | ||||
| import { UserWebhookDeliverProcessorService } from './processors/UserWebhookDeliverProcessorService.js'; | ||||
| import { SystemWebhookDeliverProcessorService } from './processors/SystemWebhookDeliverProcessorService.js'; | ||||
| import { CheckExpiredMutingsProcessorService } from './processors/CheckExpiredMutingsProcessorService.js'; | ||||
| import { BakeBufferedReactionsProcessorService } from './processors/BakeBufferedReactionsProcessorService.js'; | ||||
| import { CleanChartsProcessorService } from './processors/CleanChartsProcessorService.js'; | ||||
| import { CleanProcessorService } from './processors/CleanProcessorService.js'; | ||||
| import { CleanRemoteFilesProcessorService } from './processors/CleanRemoteFilesProcessorService.js'; | ||||
| @@ -51,6 +52,7 @@ import { RelationshipProcessorService } from './processors/RelationshipProcessor | ||||
| 		ResyncChartsProcessorService, | ||||
| 		CleanChartsProcessorService, | ||||
| 		CheckExpiredMutingsProcessorService, | ||||
| 		BakeBufferedReactionsProcessorService, | ||||
| 		CleanProcessorService, | ||||
| 		DeleteDriveFilesProcessorService, | ||||
| 		ExportCustomEmojisProcessorService, | ||||
|   | ||||
| @@ -39,6 +39,7 @@ import { TickChartsProcessorService } from './processors/TickChartsProcessorServ | ||||
| import { ResyncChartsProcessorService } from './processors/ResyncChartsProcessorService.js'; | ||||
| import { CleanChartsProcessorService } from './processors/CleanChartsProcessorService.js'; | ||||
| import { CheckExpiredMutingsProcessorService } from './processors/CheckExpiredMutingsProcessorService.js'; | ||||
| import { BakeBufferedReactionsProcessorService } from './processors/BakeBufferedReactionsProcessorService.js'; | ||||
| import { CleanProcessorService } from './processors/CleanProcessorService.js'; | ||||
| import { AggregateRetentionProcessorService } from './processors/AggregateRetentionProcessorService.js'; | ||||
| import { QueueLoggerService } from './QueueLoggerService.js'; | ||||
| @@ -118,6 +119,7 @@ export class QueueProcessorService implements OnApplicationShutdown { | ||||
| 		private cleanChartsProcessorService: CleanChartsProcessorService, | ||||
| 		private aggregateRetentionProcessorService: AggregateRetentionProcessorService, | ||||
| 		private checkExpiredMutingsProcessorService: CheckExpiredMutingsProcessorService, | ||||
| 		private bakeBufferedReactionsProcessorService: BakeBufferedReactionsProcessorService, | ||||
| 		private cleanProcessorService: CleanProcessorService, | ||||
| 	) { | ||||
| 		this.logger = this.queueLoggerService.logger; | ||||
| @@ -147,6 +149,7 @@ export class QueueProcessorService implements OnApplicationShutdown { | ||||
| 					case 'cleanCharts': return this.cleanChartsProcessorService.process(); | ||||
| 					case 'aggregateRetention': return this.aggregateRetentionProcessorService.process(); | ||||
| 					case 'checkExpiredMutings': return this.checkExpiredMutingsProcessorService.process(); | ||||
| 					case 'bakeBufferedReactions': return this.bakeBufferedReactionsProcessorService.process(); | ||||
| 					case 'clean': return this.cleanProcessorService.process(); | ||||
| 					default: throw new Error(`unrecognized job type ${job.name} for system`); | ||||
| 				} | ||||
|   | ||||
| @@ -0,0 +1,40 @@ | ||||
| /* | ||||
|  * SPDX-FileCopyrightText: syuilo and misskey-project | ||||
|  * SPDX-License-Identifier: AGPL-3.0-only | ||||
|  */ | ||||
|  | ||||
| import { Inject, Injectable } from '@nestjs/common'; | ||||
| import type Logger from '@/logger.js'; | ||||
| import { bindThis } from '@/decorators.js'; | ||||
| import { ReactionsBufferingService } from '@/core/ReactionsBufferingService.js'; | ||||
| import { MetaService } from '@/core/MetaService.js'; | ||||
| import { QueueLoggerService } from '../QueueLoggerService.js'; | ||||
| import type * as Bull from 'bullmq'; | ||||
|  | ||||
| @Injectable() | ||||
| export class BakeBufferedReactionsProcessorService { | ||||
| 	private logger: Logger; | ||||
|  | ||||
| 	constructor( | ||||
| 		private reactionsBufferingService: ReactionsBufferingService, | ||||
| 		private metaService: MetaService, | ||||
| 		private queueLoggerService: QueueLoggerService, | ||||
| 	) { | ||||
| 		this.logger = this.queueLoggerService.logger.createSubLogger('bake-buffered-reactions'); | ||||
| 	} | ||||
|  | ||||
| 	@bindThis | ||||
| 	public async process(): Promise<void> { | ||||
| 		const meta = await this.metaService.fetch(); | ||||
| 		if (!meta.enableReactionsBuffering) { | ||||
| 			this.logger.info('Reactions buffering is disabled. Skipping...'); | ||||
| 			return; | ||||
| 		} | ||||
|  | ||||
| 		this.logger.info('Baking buffered reactions...'); | ||||
|  | ||||
| 		await this.reactionsBufferingService.bake(); | ||||
|  | ||||
| 		this.logger.succ('All buffered reactions baked.'); | ||||
| 	} | ||||
| } | ||||
| @@ -27,6 +27,9 @@ export class HealthServerService { | ||||
| 		@Inject(DI.redisForTimelines) | ||||
| 		private redisForTimelines: Redis.Redis, | ||||
|  | ||||
| 		@Inject(DI.redisForReactions) | ||||
| 		private redisForReactions: Redis.Redis, | ||||
|  | ||||
| 		@Inject(DI.db) | ||||
| 		private db: DataSource, | ||||
|  | ||||
| @@ -43,6 +46,7 @@ export class HealthServerService { | ||||
| 				this.redisForPub.ping(), | ||||
| 				this.redisForSub.ping(), | ||||
| 				this.redisForTimelines.ping(), | ||||
| 				this.redisForReactions.ping(), | ||||
| 				this.db.query('SELECT 1'), | ||||
| 				...(this.meilisearch ? [this.meilisearch.health()] : []), | ||||
| 			]).then(() => 200, () => 503)); | ||||
|   | ||||
| @@ -377,6 +377,10 @@ export const meta = { | ||||
| 				type: 'number', | ||||
| 				optional: false, nullable: false, | ||||
| 			}, | ||||
| 			enableReactionsBuffering: { | ||||
| 				type: 'boolean', | ||||
| 				optional: false, nullable: false, | ||||
| 			}, | ||||
| 			notesPerOneAd: { | ||||
| 				type: 'number', | ||||
| 				optional: false, nullable: false, | ||||
| @@ -617,6 +621,7 @@ export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint- | ||||
| 				perRemoteUserUserTimelineCacheMax: instance.perRemoteUserUserTimelineCacheMax, | ||||
| 				perUserHomeTimelineCacheMax: instance.perUserHomeTimelineCacheMax, | ||||
| 				perUserListTimelineCacheMax: instance.perUserListTimelineCacheMax, | ||||
| 				enableReactionsBuffering: instance.enableReactionsBuffering, | ||||
| 				notesPerOneAd: instance.notesPerOneAd, | ||||
| 				summalyProxy: instance.urlPreviewSummaryProxyUrl, | ||||
| 				urlPreviewEnabled: instance.urlPreviewEnabled, | ||||
|   | ||||
| @@ -142,6 +142,7 @@ export const paramDef = { | ||||
| 		perRemoteUserUserTimelineCacheMax: { type: 'integer' }, | ||||
| 		perUserHomeTimelineCacheMax: { type: 'integer' }, | ||||
| 		perUserListTimelineCacheMax: { type: 'integer' }, | ||||
| 		enableReactionsBuffering: { type: 'boolean' }, | ||||
| 		notesPerOneAd: { type: 'integer' }, | ||||
| 		silencedHosts: { | ||||
| 			type: 'array', | ||||
| @@ -598,6 +599,10 @@ export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint- | ||||
| 				set.perUserListTimelineCacheMax = ps.perUserListTimelineCacheMax; | ||||
| 			} | ||||
|  | ||||
| 			if (ps.enableReactionsBuffering !== undefined) { | ||||
| 				set.enableReactionsBuffering = ps.enableReactionsBuffering; | ||||
| 			} | ||||
|  | ||||
| 			if (ps.notesPerOneAd !== undefined) { | ||||
| 				set.notesPerOneAd = ps.notesPerOneAd; | ||||
| 			} | ||||
|   | ||||
| @@ -4,10 +4,10 @@ | ||||
|  */ | ||||
|  | ||||
| import { Test, TestingModule } from '@nestjs/testing'; | ||||
| import type { MiUser } from '@/models/User.js'; | ||||
| import { UserEntityService } from '@/core/entities/UserEntityService.js'; | ||||
| import { GlobalModule } from '@/GlobalModule.js'; | ||||
| import { CoreModule } from '@/core/CoreModule.js'; | ||||
| import type { MiUser } from '@/models/User.js'; | ||||
| import { secureRndstr } from '@/misc/secure-rndstr.js'; | ||||
| import { genAidx } from '@/misc/id/aidx.js'; | ||||
| import { | ||||
| @@ -49,6 +49,7 @@ import { ApLoggerService } from '@/core/activitypub/ApLoggerService.js'; | ||||
| import { AccountMoveService } from '@/core/AccountMoveService.js'; | ||||
| import { ReactionService } from '@/core/ReactionService.js'; | ||||
| import { NotificationService } from '@/core/NotificationService.js'; | ||||
| import { ReactionsBufferingService } from '@/core/ReactionsBufferingService.js'; | ||||
|  | ||||
| process.env.NODE_ENV = 'test'; | ||||
|  | ||||
| @@ -169,6 +170,7 @@ describe('UserEntityService', () => { | ||||
| 				ApLoggerService, | ||||
| 				AccountMoveService, | ||||
| 				ReactionService, | ||||
| 				ReactionsBufferingService, | ||||
| 				NotificationService, | ||||
| 			]; | ||||
|  | ||||
|   | ||||
| @@ -36,6 +36,55 @@ SPDX-License-Identifier: AGPL-3.0-only | ||||
| 						<template #caption>{{ i18n.ts.turnOffToImprovePerformance }}</template> | ||||
| 					</MkSwitch> | ||||
| 				</div> | ||||
|  | ||||
| 				<MkFolder :defaultOpen="true"> | ||||
| 					<template #icon><i class="ti ti-bolt"></i></template> | ||||
| 					<template #label>Misskey® Fan-out Timeline Technology™ (FTT)</template> | ||||
| 					<template v-if="enableFanoutTimeline" #suffix>Enabled</template> | ||||
| 					<template v-else #suffix>Disabled</template> | ||||
|  | ||||
| 					<div class="_gaps_m"> | ||||
| 						<MkSwitch v-model="enableFanoutTimeline"> | ||||
| 							<template #label>{{ i18n.ts.enable }}</template> | ||||
| 							<template #caption>{{ i18n.ts._serverSettings.fanoutTimelineDescription }}</template> | ||||
| 						</MkSwitch> | ||||
|  | ||||
| 						<MkSwitch v-model="enableFanoutTimelineDbFallback"> | ||||
| 							<template #label>{{ i18n.ts._serverSettings.fanoutTimelineDbFallback }}</template> | ||||
| 							<template #caption>{{ i18n.ts._serverSettings.fanoutTimelineDbFallbackDescription }}</template> | ||||
| 						</MkSwitch> | ||||
|  | ||||
| 						<MkInput v-model="perLocalUserUserTimelineCacheMax" type="number"> | ||||
| 							<template #label>perLocalUserUserTimelineCacheMax</template> | ||||
| 						</MkInput> | ||||
|  | ||||
| 						<MkInput v-model="perRemoteUserUserTimelineCacheMax" type="number"> | ||||
| 							<template #label>perRemoteUserUserTimelineCacheMax</template> | ||||
| 						</MkInput> | ||||
|  | ||||
| 						<MkInput v-model="perUserHomeTimelineCacheMax" type="number"> | ||||
| 							<template #label>perUserHomeTimelineCacheMax</template> | ||||
| 						</MkInput> | ||||
|  | ||||
| 						<MkInput v-model="perUserListTimelineCacheMax" type="number"> | ||||
| 							<template #label>perUserListTimelineCacheMax</template> | ||||
| 						</MkInput> | ||||
| 					</div> | ||||
| 				</MkFolder> | ||||
|  | ||||
| 				<MkFolder :defaultOpen="true"> | ||||
| 					<template #icon><i class="ti ti-bolt"></i></template> | ||||
| 					<template #label>Misskey® Reactions Buffering Technology™ (RBT)<span class="_beta">{{ i18n.ts.beta }}</span></template> | ||||
| 					<template v-if="enableReactionsBuffering" #suffix>Enabled</template> | ||||
| 					<template v-else #suffix>Disabled</template> | ||||
|  | ||||
| 					<div class="_gaps_m"> | ||||
| 						<MkSwitch v-model="enableReactionsBuffering"> | ||||
| 							<template #label>{{ i18n.ts.enable }}</template> | ||||
| 							<template #caption>{{ i18n.ts._serverSettings.reactionsBufferingDescription }}</template> | ||||
| 						</MkSwitch> | ||||
| 					</div> | ||||
| 				</MkFolder> | ||||
| 			</div> | ||||
| 		</FormSuspense> | ||||
| 	</MkSpacer> | ||||
| @@ -52,11 +101,20 @@ import { fetchInstance } from '@/instance.js'; | ||||
| import { i18n } from '@/i18n.js'; | ||||
| import { definePageMetadata } from '@/scripts/page-metadata.js'; | ||||
| import MkSwitch from '@/components/MkSwitch.vue'; | ||||
| import MkFolder from '@/components/MkFolder.vue'; | ||||
| import MkInput from '@/components/MkInput.vue'; | ||||
|  | ||||
| const enableServerMachineStats = ref<boolean>(false); | ||||
| const enableIdenticonGeneration = ref<boolean>(false); | ||||
| const enableChartsForRemoteUser = ref<boolean>(false); | ||||
| const enableChartsForFederatedInstances = ref<boolean>(false); | ||||
| const enableFanoutTimeline = ref<boolean>(false); | ||||
| const enableFanoutTimelineDbFallback = ref<boolean>(false); | ||||
| const perLocalUserUserTimelineCacheMax = ref<number>(0); | ||||
| const perRemoteUserUserTimelineCacheMax = ref<number>(0); | ||||
| const perUserHomeTimelineCacheMax = ref<number>(0); | ||||
| const perUserListTimelineCacheMax = ref<number>(0); | ||||
| const enableReactionsBuffering = ref<boolean>(false); | ||||
|  | ||||
| async function init() { | ||||
| 	const meta = await misskeyApi('admin/meta'); | ||||
| @@ -64,6 +122,13 @@ async function init() { | ||||
| 	enableIdenticonGeneration.value = meta.enableIdenticonGeneration; | ||||
| 	enableChartsForRemoteUser.value = meta.enableChartsForRemoteUser; | ||||
| 	enableChartsForFederatedInstances.value = meta.enableChartsForFederatedInstances; | ||||
| 	enableFanoutTimeline.value = meta.enableFanoutTimeline; | ||||
| 	enableFanoutTimelineDbFallback.value = meta.enableFanoutTimelineDbFallback; | ||||
| 	perLocalUserUserTimelineCacheMax.value = meta.perLocalUserUserTimelineCacheMax; | ||||
| 	perRemoteUserUserTimelineCacheMax.value = meta.perRemoteUserUserTimelineCacheMax; | ||||
| 	perUserHomeTimelineCacheMax.value = meta.perUserHomeTimelineCacheMax; | ||||
| 	perUserListTimelineCacheMax.value = meta.perUserListTimelineCacheMax; | ||||
| 	enableReactionsBuffering.value = meta.enableReactionsBuffering; | ||||
| } | ||||
|  | ||||
| function save() { | ||||
| @@ -72,6 +137,13 @@ function save() { | ||||
| 		enableIdenticonGeneration: enableIdenticonGeneration.value, | ||||
| 		enableChartsForRemoteUser: enableChartsForRemoteUser.value, | ||||
| 		enableChartsForFederatedInstances: enableChartsForFederatedInstances.value, | ||||
| 		enableFanoutTimeline: enableFanoutTimeline.value, | ||||
| 		enableFanoutTimelineDbFallback: enableFanoutTimelineDbFallback.value, | ||||
| 		perLocalUserUserTimelineCacheMax: perLocalUserUserTimelineCacheMax.value, | ||||
| 		perRemoteUserUserTimelineCacheMax: perRemoteUserUserTimelineCacheMax.value, | ||||
| 		perUserHomeTimelineCacheMax: perUserHomeTimelineCacheMax.value, | ||||
| 		perUserListTimelineCacheMax: perUserListTimelineCacheMax.value, | ||||
| 		enableReactionsBuffering: enableReactionsBuffering.value, | ||||
| 	}).then(() => { | ||||
| 		fetchInstance(true); | ||||
| 	}); | ||||
|   | ||||
| @@ -96,38 +96,6 @@ SPDX-License-Identifier: AGPL-3.0-only | ||||
| 						</div> | ||||
| 					</FormSection> | ||||
|  | ||||
| 					<FormSection> | ||||
| 						<template #label>Misskey® Fan-out Timeline Technology™ (FTT)</template> | ||||
|  | ||||
| 						<div class="_gaps_m"> | ||||
| 							<MkSwitch v-model="enableFanoutTimeline"> | ||||
| 								<template #label>{{ i18n.ts.enable }}</template> | ||||
| 								<template #caption>{{ i18n.ts._serverSettings.fanoutTimelineDescription }}</template> | ||||
| 							</MkSwitch> | ||||
|  | ||||
| 							<MkSwitch v-model="enableFanoutTimelineDbFallback"> | ||||
| 								<template #label>{{ i18n.ts._serverSettings.fanoutTimelineDbFallback }}</template> | ||||
| 								<template #caption>{{ i18n.ts._serverSettings.fanoutTimelineDbFallbackDescription }}</template> | ||||
| 							</MkSwitch> | ||||
|  | ||||
| 							<MkInput v-model="perLocalUserUserTimelineCacheMax" type="number"> | ||||
| 								<template #label>perLocalUserUserTimelineCacheMax</template> | ||||
| 							</MkInput> | ||||
|  | ||||
| 							<MkInput v-model="perRemoteUserUserTimelineCacheMax" type="number"> | ||||
| 								<template #label>perRemoteUserUserTimelineCacheMax</template> | ||||
| 							</MkInput> | ||||
|  | ||||
| 							<MkInput v-model="perUserHomeTimelineCacheMax" type="number"> | ||||
| 								<template #label>perUserHomeTimelineCacheMax</template> | ||||
| 							</MkInput> | ||||
|  | ||||
| 							<MkInput v-model="perUserListTimelineCacheMax" type="number"> | ||||
| 								<template #label>perUserListTimelineCacheMax</template> | ||||
| 							</MkInput> | ||||
| 						</div> | ||||
| 					</FormSection> | ||||
|  | ||||
| 					<FormSection> | ||||
| 						<template #label>{{ i18n.ts._ad.adsSettings }}</template> | ||||
|  | ||||
| @@ -236,12 +204,6 @@ const cacheRemoteSensitiveFiles = ref<boolean>(false); | ||||
| const enableServiceWorker = ref<boolean>(false); | ||||
| const swPublicKey = ref<string | null>(null); | ||||
| const swPrivateKey = ref<string | null>(null); | ||||
| const enableFanoutTimeline = ref<boolean>(false); | ||||
| const enableFanoutTimelineDbFallback = ref<boolean>(false); | ||||
| const perLocalUserUserTimelineCacheMax = ref<number>(0); | ||||
| const perRemoteUserUserTimelineCacheMax = ref<number>(0); | ||||
| const perUserHomeTimelineCacheMax = ref<number>(0); | ||||
| const perUserListTimelineCacheMax = ref<number>(0); | ||||
| const notesPerOneAd = ref<number>(0); | ||||
| const urlPreviewEnabled = ref<boolean>(true); | ||||
| const urlPreviewTimeout = ref<number>(10000); | ||||
| @@ -265,12 +227,6 @@ async function init(): Promise<void> { | ||||
| 	enableServiceWorker.value = meta.enableServiceWorker; | ||||
| 	swPublicKey.value = meta.swPublickey; | ||||
| 	swPrivateKey.value = meta.swPrivateKey; | ||||
| 	enableFanoutTimeline.value = meta.enableFanoutTimeline; | ||||
| 	enableFanoutTimelineDbFallback.value = meta.enableFanoutTimelineDbFallback; | ||||
| 	perLocalUserUserTimelineCacheMax.value = meta.perLocalUserUserTimelineCacheMax; | ||||
| 	perRemoteUserUserTimelineCacheMax.value = meta.perRemoteUserUserTimelineCacheMax; | ||||
| 	perUserHomeTimelineCacheMax.value = meta.perUserHomeTimelineCacheMax; | ||||
| 	perUserListTimelineCacheMax.value = meta.perUserListTimelineCacheMax; | ||||
| 	notesPerOneAd.value = meta.notesPerOneAd; | ||||
| 	urlPreviewEnabled.value = meta.urlPreviewEnabled; | ||||
| 	urlPreviewTimeout.value = meta.urlPreviewTimeout; | ||||
| @@ -295,12 +251,6 @@ async function save() { | ||||
| 		enableServiceWorker: enableServiceWorker.value, | ||||
| 		swPublicKey: swPublicKey.value, | ||||
| 		swPrivateKey: swPrivateKey.value, | ||||
| 		enableFanoutTimeline: enableFanoutTimeline.value, | ||||
| 		enableFanoutTimelineDbFallback: enableFanoutTimelineDbFallback.value, | ||||
| 		perLocalUserUserTimelineCacheMax: perLocalUserUserTimelineCacheMax.value, | ||||
| 		perRemoteUserUserTimelineCacheMax: perRemoteUserUserTimelineCacheMax.value, | ||||
| 		perUserHomeTimelineCacheMax: perUserHomeTimelineCacheMax.value, | ||||
| 		perUserListTimelineCacheMax: perUserListTimelineCacheMax.value, | ||||
| 		notesPerOneAd: notesPerOneAd.value, | ||||
| 		urlPreviewEnabled: urlPreviewEnabled.value, | ||||
| 		urlPreviewTimeout: urlPreviewTimeout.value, | ||||
|   | ||||
| @@ -5125,6 +5125,7 @@ export type operations = { | ||||
|             perRemoteUserUserTimelineCacheMax: number; | ||||
|             perUserHomeTimelineCacheMax: number; | ||||
|             perUserListTimelineCacheMax: number; | ||||
|             enableReactionsBuffering: boolean; | ||||
|             notesPerOneAd: number; | ||||
|             backgroundImageUrl: string | null; | ||||
|             deeplAuthKey: string | null; | ||||
| @@ -9395,6 +9396,7 @@ export type operations = { | ||||
|           perRemoteUserUserTimelineCacheMax?: number; | ||||
|           perUserHomeTimelineCacheMax?: number; | ||||
|           perUserListTimelineCacheMax?: number; | ||||
|           enableReactionsBuffering?: boolean; | ||||
|           notesPerOneAd?: number; | ||||
|           silencedHosts?: string[] | null; | ||||
|           mediaSilencedHosts?: string[] | null; | ||||
|   | ||||
		Reference in New Issue
	
	Block a user
	 syuilo
					syuilo