enhance(backend): RedisへのTLの構築をListで行うように

#11404

5 files changed, 112 insertions, 105 deletions

diff --git a/packages/backend/src/core/AntennaService.ts b/packages/backend/src/core/AntennaService.ts
index b64120772c..ca7624b1d4 100644
--- a/packages/backend/src/core/AntennaService.ts
+++ b/packages/backend/src/core/AntennaService.ts
@@ -16,6 +16,7 @@ import type { AntennasRepository, UserListMembershipsRepository } from '@/models
 import { UtilityService } from '@/core/UtilityService.js';
 import { bindThis } from '@/decorators.js';
 import type { GlobalEvents } from '@/core/GlobalEventService.js';
+import { RedisTimelineService } from '@/core/RedisTimelineService.js';
 import type { OnApplicationShutdown } from '@nestjs/common';
 
 @Injectable()
@@ -38,6 +39,7 @@ export class AntennaService implements OnApplicationShutdown {
 
 		private utilityService: UtilityService,
 		private globalEventService: GlobalEventService,
+		private redisTimelineService: RedisTimelineService,
 	) {
 		this.antennasFetched = false;
 		this.antennas = [];
@@ -77,9 +79,6 @@ export class AntennaService implements OnApplicationShutdown {
 
 	@bindThis
 	public async addNoteToAntennas(note: MiNote, noteUser: { id: MiUser['id']; username: string; host: string | null; }): Promise<void> {
-		// リモートから遅れて届いた(もしくは後から追加された)投稿日時が古い投稿が追加されるとページネーション時に問題を引き起こすため、3分以内に投稿されたもののみを追加する
-		if (Date.now() - note.createdAt.getTime() > 1000 * 60 * 3) return;
-
 		const antennas = await this.getAntennas();
 		const antennasWithMatchResult = await Promise.all(antennas.map(antenna => this.checkHitAntenna(antenna, note, noteUser).then(hit => [antenna, hit] as const)));
 		const matchedAntennas = antennasWithMatchResult.filter(([, hit]) => hit).map(([antenna]) => antenna);
@@ -87,12 +86,7 @@ export class AntennaService implements OnApplicationShutdown {
 		const redisPipeline = this.redisForTimelines.pipeline();
 
 		for (const antenna of matchedAntennas) {
-			redisPipeline.xadd(
-				`antennaTimeline:${antenna.id}`,
-				'MAXLEN', '~', '200',
-				'*',
-				'note', note.id);
-
+			this.redisTimelineService.push(`antennaTimeline:${antenna.id}`, note.id, 200, redisPipeline);
 			this.globalEventService.publishAntennaStream(antenna.id, 'note', note);
 		}
 
diff --git a/packages/backend/src/core/CoreModule.ts b/packages/backend/src/core/CoreModule.ts
index 1984d9e6c2..0dc025d998 100644
--- a/packages/backend/src/core/CoreModule.ts
+++ b/packages/backend/src/core/CoreModule.ts
@@ -61,6 +61,7 @@ import { FileInfoService } from './FileInfoService.js';
 import { SearchService } from './SearchService.js';
 import { ClipService } from './ClipService.js';
 import { FeaturedService } from './FeaturedService.js';
+import { RedisTimelineService } from './RedisTimelineService.js';
 import { ChartLoggerService } from './chart/ChartLoggerService.js';
 import FederationChart from './chart/charts/federation.js';
 import NotesChart from './chart/charts/notes.js';
@@ -189,6 +190,7 @@ const $FileInfoService: Provider = { provide: 'FileInfoService', useExisting: Fi
 const $SearchService: Provider = { provide: 'SearchService', useExisting: SearchService };
 const $ClipService: Provider = { provide: 'ClipService', useExisting: ClipService };
 const $FeaturedService: Provider = { provide: 'FeaturedService', useExisting: FeaturedService };
+const $RedisTimelineService: Provider = { provide: 'RedisTimelineService', useExisting: RedisTimelineService };
 
 const $ChartLoggerService: Provider = { provide: 'ChartLoggerService', useExisting: ChartLoggerService };
 const $FederationChart: Provider = { provide: 'FederationChart', useExisting: FederationChart };
@@ -321,6 +323,7 @@ const $ApQuestionService: Provider = { provide: 'ApQuestionService', useExisting
 		SearchService,
 		ClipService,
 		FeaturedService,
+		RedisTimelineService,
 		ChartLoggerService,
 		FederationChart,
 		NotesChart,
@@ -446,6 +449,7 @@ const $ApQuestionService: Provider = { provide: 'ApQuestionService', useExisting
 		$SearchService,
 		$ClipService,
 		$FeaturedService,
+		$RedisTimelineService,
 		$ChartLoggerService,
 		$FederationChart,
 		$NotesChart,
@@ -572,6 +576,7 @@ const $ApQuestionService: Provider = { provide: 'ApQuestionService', useExisting
 		SearchService,
 		ClipService,
 		FeaturedService,
+		RedisTimelineService,
 		FederationChart,
 		NotesChart,
 		UsersChart,
@@ -696,6 +701,7 @@ const $ApQuestionService: Provider = { provide: 'ApQuestionService', useExisting
 		$SearchService,
 		$ClipService,
 		$FeaturedService,
+		$RedisTimelineService,
 		$FederationChart,
 		$NotesChart,
 		$UsersChart,
diff --git a/packages/backend/src/core/NoteCreateService.ts b/packages/backend/src/core/NoteCreateService.ts
index 65beb9f970..2a73467122 100644
--- a/packages/backend/src/core/NoteCreateService.ts
+++ b/packages/backend/src/core/NoteCreateService.ts
@@ -54,6 +54,7 @@ import { RoleService } from '@/core/RoleService.js';
 import { MetaService } from '@/core/MetaService.js';
 import { SearchService } from '@/core/SearchService.js';
 import { FeaturedService } from '@/core/FeaturedService.js';
+import { RedisTimelineService } from '@/core/RedisTimelineService.js';
 
 type NotificationType = 'reply' | 'renote' | 'quote' | 'mention';
 
@@ -194,6 +195,7 @@ export class NoteCreateService implements OnApplicationShutdown {
 		private idService: IdService,
 		private globalEventService: GlobalEventService,
 		private queueService: QueueService,
+		private redisTimelineService: RedisTimelineService,
 		private noteReadService: NoteReadService,
 		private notificationService: NotificationService,
 		private relayService: RelayService,
@@ -347,14 +349,6 @@ export class NoteCreateService implements OnApplicationShutdown {
 
 		const note = await this.insertNote(user, data, tags, emojis, mentionedUsers);
 
-		if (data.channel) {
-			this.redisForTimelines.xadd(
-				`channelTimeline:${data.channel.id}`,
-				'MAXLEN', '~', this.config.perChannelMaxNoteCacheCount.toString(),
-				'*',
-				'note', note.id);
-		}
-
 		setImmediate('post created', { signal: this.#shutdownController.signal }).then(
 			() => this.postNoteCreated(note, user, data, silent, tags!, mentionedUsers!),
 			() => { /* aborted, ignore this */ },
@@ -822,20 +816,14 @@ export class NoteCreateService implements OnApplicationShutdown {
 
 	@bindThis
 	private async pushToTl(note: MiNote, user: { id: MiUser['id']; host: MiUser['host']; }) {
-		// リモートから遅れて届いた(もしくは後から追加された)投稿日時が古い投稿が追加されるとページネーション時に問題を引き起こすため、3分以内に投稿されたもののみを追加する
-		// TODO: https://github.com/misskey-dev/misskey/issues/11404#issuecomment-1752480890 をやる
-		if (note.userHost != null && (Date.now() - note.createdAt.getTime()) > 1000 * 60 * 3) return;
-
 		const meta = await this.metaService.fetch();
 
-		const redisPipeline = this.redisForTimelines.pipeline();
+		const r = this.redisForTimelines.pipeline();
 
 		if (note.channelId) {
-			redisPipeline.xadd(
-				`userTimelineWithChannel:${user.id}`,
-				'MAXLEN', '~', note.userHost == null ? meta.perLocalUserUserTimelineCacheMax.toString() : meta.perRemoteUserUserTimelineCacheMax.toString(),
-				'*',
-				'note', note.id);
+			this.redisTimelineService.push(`channelTimeline:${note.channelId}`, note.id, this.config.perChannelMaxNoteCacheCount, r);
+
+			this.redisTimelineService.push(`userTimelineWithChannel:${user.id}`, note.id, note.userHost == null ? meta.perLocalUserUserTimelineCacheMax : meta.perRemoteUserUserTimelineCacheMax, r);
 
 			const channelFollowings = await this.channelFollowingsRepository.find({
 				where: {
@@ -845,18 +833,9 @@ export class NoteCreateService implements OnApplicationShutdown {
 			});
 
 			for (const channelFollowing of channelFollowings) {
-				redisPipeline.xadd(
-					`homeTimeline:${channelFollowing.followerId}`,
-					'MAXLEN', '~', meta.perUserHomeTimelineCacheMax.toString(),
-					'*',
-					'note', note.id);
-
+				this.redisTimelineService.push(`homeTimeline:${channelFollowing.followerId}`, note.id, meta.perUserHomeTimelineCacheMax, r);
 				if (note.fileIds.length > 0) {
-					redisPipeline.xadd(
-						`homeTimelineWithFiles:${channelFollowing.followerId}`,
-						'MAXLEN', '~', (meta.perUserHomeTimelineCacheMax / 2).toString(),
-						'*',
-						'note', note.id);
+					this.redisTimelineService.push(`homeTimelineWithFiles:${channelFollowing.followerId}`, note.id, meta.perUserHomeTimelineCacheMax / 2, r);
 				}
 			}
 		} else {
@@ -894,18 +873,9 @@ export class NoteCreateService implements OnApplicationShutdown {
 					if (!following.withReplies) continue;
 				}
 
-				redisPipeline.xadd(
-					`homeTimeline:${following.followerId}`,
-					'MAXLEN', '~', meta.perUserHomeTimelineCacheMax.toString(),
-					'*',
-					'note', note.id);
-
+				this.redisTimelineService.push(`homeTimeline:${following.followerId}`, note.id, meta.perUserHomeTimelineCacheMax, r);
 				if (note.fileIds.length > 0) {
-					redisPipeline.xadd(
-						`homeTimelineWithFiles:${following.followerId}`,
-						'MAXLEN', '~', (meta.perUserHomeTimelineCacheMax / 2).toString(),
-						'*',
-						'note', note.id);
+					this.redisTimelineService.push(`homeTimelineWithFiles:${following.followerId}`, note.id, meta.perUserHomeTimelineCacheMax / 2, r);
 				}
 			}
 
@@ -921,72 +891,32 @@ export class NoteCreateService implements OnApplicationShutdown {
 					if (!userListMembership.withReplies) continue;
 				}
 
-				redisPipeline.xadd(
-					`userListTimeline:${userListMembership.userListId}`,
-					'MAXLEN', '~', meta.perUserListTimelineCacheMax.toString(),
-					'*',
-					'note', note.id);
-
+				this.redisTimelineService.push(`userListTimeline:${userListMembership.userListId}`, note.id, meta.perUserListTimelineCacheMax, r);
 				if (note.fileIds.length > 0) {
-					redisPipeline.xadd(
-						`userListTimelineWithFiles:${userListMembership.userListId}`,
-						'MAXLEN', '~', (meta.perUserListTimelineCacheMax / 2).toString(),
-						'*',
-						'note', note.id);
+					this.redisTimelineService.push(`userListTimelineWithFiles:${userListMembership.userListId}`, note.id, meta.perUserListTimelineCacheMax / 2, r);
 				}
 			}
 
 			if (note.visibility !== 'specified' || !note.visibleUserIds.some(v => v === user.id)) { // 自分自身のHTL
-				redisPipeline.xadd(
-					`homeTimeline:${user.id}`,
-					'MAXLEN', '~', meta.perUserHomeTimelineCacheMax.toString(),
-					'*',
-					'note', note.id);
-
+				this.redisTimelineService.push(`homeTimeline:${user.id}`, note.id, meta.perUserHomeTimelineCacheMax, r);
 				if (note.fileIds.length > 0) {
-					redisPipeline.xadd(
-						`homeTimelineWithFiles:${user.id}`,
-						'MAXLEN', '~', (meta.perUserHomeTimelineCacheMax / 2).toString(),
-						'*',
-						'note', note.id);
+					this.redisTimelineService.push(`homeTimelineWithFiles:${user.id}`, note.id, meta.perUserHomeTimelineCacheMax / 2, r);
 				}
 			}
 
 			// 自分自身以外への返信
 			if (note.replyId && note.replyUserId !== note.userId) {
-				redisPipeline.xadd(
-					`userTimelineWithReplies:${user.id}`,
-					'MAXLEN', '~', note.userHost == null ? meta.perLocalUserUserTimelineCacheMax.toString() : meta.perRemoteUserUserTimelineCacheMax.toString(),
-					'*',
-					'note', note.id);
+				this.redisTimelineService.push(`userTimelineWithReplies:${user.id}`, note.id, note.userHost == null ? meta.perLocalUserUserTimelineCacheMax : meta.perRemoteUserUserTimelineCacheMax, r);
 			} else {
-				redisPipeline.xadd(
-					`userTimeline:${user.id}`,
-					'MAXLEN', '~', note.userHost == null ? meta.perLocalUserUserTimelineCacheMax.toString() : meta.perRemoteUserUserTimelineCacheMax.toString(),
-					'*',
-					'note', note.id);
-
+				this.redisTimelineService.push(`userTimeline:${user.id}`, note.id, note.userHost == null ? meta.perLocalUserUserTimelineCacheMax : meta.perRemoteUserUserTimelineCacheMax, r);
 				if (note.fileIds.length > 0) {
-					redisPipeline.xadd(
-						`userTimelineWithFiles:${user.id}`,
-						'MAXLEN', '~', note.userHost == null ? (meta.perLocalUserUserTimelineCacheMax / 2).toString() : (meta.perRemoteUserUserTimelineCacheMax / 2).toString(),
-						'*',
-						'note', note.id);
+					this.redisTimelineService.push(`userTimelineWithFiles:${user.id}`, note.id, note.userHost == null ? meta.perLocalUserUserTimelineCacheMax / 2 : meta.perRemoteUserUserTimelineCacheMax / 2, r);
 				}
 
 				if (note.visibility === 'public' && note.userHost == null) {
-					redisPipeline.xadd(
-						'localTimeline',
-						'MAXLEN', '~', '1000',
-						'*',
-						'note', note.id);
-
+					this.redisTimelineService.push('localTimeline', note.id, 1000, r);
 					if (note.fileIds.length > 0) {
-						redisPipeline.xadd(
-							'localTimelineWithFiles',
-							'MAXLEN', '~', '500',
-							'*',
-							'note', note.id);
+						this.redisTimelineService.push('localTimelineWithFiles', note.id, 500, r);
 					}
 				}
 			}
@@ -998,7 +928,7 @@ export class NoteCreateService implements OnApplicationShutdown {
 			}
 		}
 
-		redisPipeline.exec();
+		r.exec();
 	}
 
 	@bindThis
diff --git a/packages/backend/src/core/RedisTimelineService.ts b/packages/backend/src/core/RedisTimelineService.ts
new file mode 100644
index 0000000000..f0ca2726d6
--- /dev/null
+++ b/packages/backend/src/core/RedisTimelineService.ts
@@ -0,0 +1,80 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and other misskey contributors
+ * 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 { bindThis } from '@/decorators.js';
+import { IdService } from '@/core/IdService.js';
+
+@Injectable()
+export class RedisTimelineService {
+	constructor(
+		@Inject(DI.redisForTimelines)
+		private redisForTimelines: Redis.Redis,
+
+		private idService: IdService,
+	) {
+	}
+
+	@bindThis
+	public push(tl: string, id: string, maxlen: number, pipeline: Redis.ChainableCommander) {
+		// リモートから遅れて届いた(もしくは後から追加された)投稿日時が古い投稿が追加されるとページネーション時に問題を引き起こすため、
+		// 3分以内に投稿されたものでない場合、Redisにある最古のIDより新しい場合のみ追加する
+		if (this.idService.parse(id).date.getTime() > Date.now() - 1000 * 60 * 3) {
+			pipeline.lpush('list:' + tl, id);
+			if (Math.random() < 0.1) { // 10%の確率でトリム
+				pipeline.ltrim('list:' + tl, 0, maxlen - 1);
+			}
+		} else {
+			// 末尾のIDを取得
+			this.redisForTimelines.lindex('list:' + tl, -1).then(lastId => {
+				if (lastId == null || (this.idService.parse(id).date.getTime() > this.idService.parse(lastId).date.getTime())) {
+					this.redisForTimelines.lpush('list:' + tl, id);
+				} else {
+					Promise.resolve();
+				}
+			});
+		}
+	}
+
+	@bindThis
+	public get(name: string, untilId?: string | null, sinceId?: string | null) {
+		if (untilId && sinceId) {
+			return this.redisForTimelines.lrange('list:' + name, 0, -1)
+				.then(ids => ids.filter(id => id > untilId && id < sinceId).sort((a, b) => a > b ? -1 : 1));
+		} else if (untilId) {
+			return this.redisForTimelines.lrange('list:' + name, 0, -1)
+				.then(ids => ids.filter(id => id > untilId).sort((a, b) => a > b ? -1 : 1));
+		} else if (sinceId) {
+			return this.redisForTimelines.lrange('list:' + name, 0, -1)
+				.then(ids => ids.filter(id => id < sinceId).sort((a, b) => a < b ? -1 : 1));
+		} else {
+			return this.redisForTimelines.lrange('list:' + name, 0, -1)
+				.then(ids => ids.sort((a, b) => a > b ? -1 : 1));
+		}
+	}
+
+	@bindThis
+	public getMulti(name: string[], untilId?: string | null, sinceId?: string | null): Promise<string[][]> {
+		const pipeline = this.redisForTimelines.pipeline();
+		for (const n of name) {
+			pipeline.lrange('list:' + n, 0, -1);
+		}
+		return pipeline.exec().then(res => {
+			if (res == null) return [];
+			const tls = res.map(r => r[1] as string[]);
+			return tls.map(ids =>
+				(untilId && sinceId)
+					? ids.filter(id => id > untilId && id < sinceId).sort((a, b) => a > b ? -1 : 1)
+					: untilId
+						? ids.filter(id => id > untilId).sort((a, b) => a > b ? -1 : 1)
+						: sinceId
+							? ids.filter(id => id < sinceId).sort((a, b) => a < b ? -1 : 1)
+							: ids.sort((a, b) => a > b ? -1 : 1),
+			);
+		});
+	}
+}
diff --git a/packages/backend/src/core/RoleService.ts b/packages/backend/src/core/RoleService.ts
index f2bd9de5ee..2c3547e4ac 100644
--- a/packages/backend/src/core/RoleService.ts
+++ b/packages/backend/src/core/RoleService.ts
@@ -20,6 +20,7 @@ import { IdService } from '@/core/IdService.js';
 import { GlobalEventService } from '@/core/GlobalEventService.js';
 import { ModerationLogService } from '@/core/ModerationLogService.js';
 import type { Packed } from '@/misc/json-schema.js';
+import { RedisTimelineService } from '@/core/RedisTimelineService.js';
 import type { OnApplicationShutdown } from '@nestjs/common';
 
 export type RolePolicies = {
@@ -102,6 +103,7 @@ export class RoleService implements OnApplicationShutdown {
 		private globalEventService: GlobalEventService,
 		private idService: IdService,
 		private moderationLogService: ModerationLogService,
+		private redisTimelineService: RedisTimelineService,
 	) {
 		//this.onMessage = this.onMessage.bind(this);
 
@@ -472,12 +474,7 @@ export class RoleService implements OnApplicationShutdown {
 		const redisPipeline = this.redisClient.pipeline();
 
 		for (const role of roles) {
-			redisPipeline.xadd(
-				`roleTimeline:${role.id}`,
-				'MAXLEN', '~', '1000',
-				'*',
-				'note', note.id);
-
+			this.redisTimelineService.push(`roleTimeline:${role.id}`, note.id, 1000, redisPipeline);
 			this.globalEventService.publishRoleTimelineStream(role.id, 'note', note);
 		}