Pick redis push tl

packages/backend/migration/1696807733453-userListUserId.js

@@ -0,0 +1,22 @@
+
+/*
+ * SPDX-FileCopyrightText: syuilo and other misskey contributors
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+export class UserListUserId1696807733453 {
+	name = 'UserListUserId1696807733453'
+
+	async up(queryRunner) {
+		await queryRunner.query(`ALTER TABLE "user_list_membership" ADD "userListUserId" character varying(32) NOT NULL DEFAULT ''`);
+		const memberships = await queryRunner.query(`SELECT "id", "userListId" FROM "user_list_membership"`);
+		for(let i = 0; i < memberships.length; i++) {
+			const userList = await queryRunner.query(`SELECT "userId" FROM "user_list" WHERE "id" = $1`, [memberships[i].userListId]);
+			await queryRunner.query(`UPDATE "user_list_membership" SET "userListUserId" = $1 WHERE "id" = $2`, [userList[0].userId, memberships[i].id]);
+		}
+	}
+
+	async down(queryRunner) {
+		await queryRunner.query(`ALTER TABLE "user_list_membership" DROP COLUMN "userListUserId"`);
+	}
+}

packages/backend/migration/1696808725134-userListUserId-2.js

@@ -0,0 +1,16 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and other misskey contributors
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+export class UserListUserId21696808725134 {
+	name = 'UserListUserId21696808725134'
+
+	async up(queryRunner) {
+		await queryRunner.query(`ALTER TABLE "user_list_membership" ALTER COLUMN "userListUserId" DROP DEFAULT`);
+	}
+
+	async down(queryRunner) {
+		await queryRunner.query(`ALTER TABLE "user_list_membership" ALTER COLUMN "userListUserId" SET DEFAULT ''`);
+	}
+}

packages/backend/src/config.ts

@@ -95,6 +95,7 @@ type Source = {
 		abuseDiscordHook?: string;
 		disableAbuseRepository?: boolean;
 		maxWebImageSize?: number;
+		withRepliesInHomeTL?: boolean;
 	}
 };
 
@@ -176,6 +177,7 @@ export type Config = {
 		abuseDiscordHook?: string;
 		disableAbuseRepository?: boolean;
 		maxWebImageSize?: number;
+		withRepliesInHomeTL?: boolean,
 	}
 };
 

packages/backend/src/core/AccountMoveService.ts

@@ -228,7 +228,7 @@ export class AccountMoveService {
 			},
 		}).then(joinings => joinings.map(joining => joining.userListId));
 
-		const newJoinings: Map<string, { createdAt: Date; userId: string; userListId: string; }> = new Map();
+		const newJoinings: Map<string, { createdAt: Date; userId: string; userListId: string; userListUserId: string; }> = new Map();
 
 		// 重複しないようにIDを生成
 		const genId = (): string => {
@@ -244,6 +244,7 @@ export class AccountMoveService {
 				createdAt: new Date(),
 				userId: dst.id,
 				userListId: joining.userListId,
+				userListUserId: joining.userListUserId,
 			});
 		}
 

packages/backend/src/core/AntennaService.ts

@@ -16,6 +16,7 @@ import type { AntennasRepository, UserListJoiningsRepository } 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 = [];
@@ -89,7 +91,7 @@ export class AntennaService implements OnApplicationShutdown {
 				'MAXLEN', '~', '200',
 				'*',
 				'note', note.id);
-
+			this.redisTimelineService.push(`antennaTimeline:${antenna.id}`, note.id, 200, redisPipeline);
 			this.globalEventService.publishAntennaStream(antenna.id, 'note', note);
 		}
 

packages/backend/src/core/CoreModule.ts

@@ -59,6 +59,7 @@ import { UtilityService } from './UtilityService.js';
 import { FileInfoService } from './FileInfoService.js';
 import { SearchService } from './SearchService.js';
 import { ClipService } from './ClipService.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';
@@ -186,6 +187,7 @@ const $UtilityService: Provider = { provide: 'UtilityService', useExisting: Util
 const $FileInfoService: Provider = { provide: 'FileInfoService', useExisting: FileInfoService };
 const $SearchService: Provider = { provide: 'SearchService', useExisting: SearchService };
 const $ClipService: Provider = { provide: 'ClipService', useExisting: ClipService };
+const $RedisTimelineService: Provider = { provide: 'RedisTimelineService', useExisting: RedisTimelineService };
 
 const $ChartLoggerService: Provider = { provide: 'ChartLoggerService', useExisting: ChartLoggerService };
 const $FederationChart: Provider = { provide: 'FederationChart', useExisting: FederationChart };
@@ -316,6 +318,7 @@ const $ApQuestionService: Provider = { provide: 'ApQuestionService', useExisting
 		FileInfoService,
 		SearchService,
 		ClipService,
+		RedisTimelineService,
 		ChartLoggerService,
 		FederationChart,
 		NotesChart,
@@ -440,6 +443,7 @@ const $ApQuestionService: Provider = { provide: 'ApQuestionService', useExisting
 		$FileInfoService,
 		$SearchService,
 		$ClipService,
+		$RedisTimelineService,
 		$ChartLoggerService,
 		$FederationChart,
 		$NotesChart,
@@ -564,6 +568,7 @@ const $ApQuestionService: Provider = { provide: 'ApQuestionService', useExisting
 		FileInfoService,
 		SearchService,
 		ClipService,
+		RedisTimelineService,
 		FederationChart,
 		NotesChart,
 		UsersChart,
@@ -687,6 +692,7 @@ const $ApQuestionService: Provider = { provide: 'ApQuestionService', useExisting
 		$FileInfoService,
 		$SearchService,
 		$ClipService,
+		$RedisTimelineService,
 		$FederationChart,
 		$NotesChart,
 		$UsersChart,

packages/backend/src/core/NoteCreateService.ts

@@ -5,7 +5,7 @@
 
 import { setImmediate } from 'node:timers/promises';
 import * as mfm from 'mfm-js';
-import { In, DataSource } from 'typeorm';
+import { In, DataSource, IsNull, LessThan } from 'typeorm';
 import * as Redis from 'ioredis';
 import { Inject, Injectable, OnApplicationShutdown } from '@nestjs/common';
 import RE2 from 're2';
@@ -14,7 +14,7 @@ import { extractCustomEmojisFromMfm } from '@/misc/extract-custom-emojis-from-mf
 import { extractHashtags } from '@/misc/extract-hashtags.js';
 import type { IMentionedRemoteUsers } from '@/models/Note.js';
 import { MiNote } from '@/models/Note.js';
-import type { ChannelsRepository, FollowingsRepository, InstancesRepository, MutedNotesRepository, MutingsRepository, NotesRepository, NoteThreadMutingsRepository, UserProfilesRepository, UsersRepository } from '@/models/_.js';
+import type { ChannelFollowingsRepository, ChannelsRepository, FollowingsRepository, InstancesRepository, MutedNotesRepository, MiFollowing, MutingsRepository, NotesRepository, NoteThreadMutingsRepository, UserListJoiningsRepository, UserProfilesRepository, UsersRepository } from '@/models/_.js';
 import type { MiDriveFile } from '@/models/DriveFile.js';
 import type { MiApp } from '@/models/App.js';
 import { concat } from '@/misc/prelude/array.js';
@@ -53,6 +53,7 @@ import { DB_MAX_NOTE_TEXT_LENGTH } from '@/const.js';
 import { RoleService } from '@/core/RoleService.js';
 import { MetaService } from '@/core/MetaService.js';
 import { SearchService } from '@/core/SearchService.js';
+import { RedisTimelineService } from '@/core/RedisTimelineService.js';
 
 const mutedWordsCache = new MemorySingleCache<{ userId: MiUserProfile['userId']; mutedWords: MiUserProfile['mutedWords']; }[]>(1000 * 60 * 5);
 
@@ -175,6 +176,9 @@ export class NoteCreateService implements OnApplicationShutdown {
 		@Inject(DI.userProfilesRepository)
 		private userProfilesRepository: UserProfilesRepository,
 
+		@Inject(DI.userListJoiningsRepository)
+		private userListJoiningsRepository: UserListJoiningsRepository,
+
 		@Inject(DI.mutedNotesRepository)
 		private mutedNotesRepository: MutedNotesRepository,
 
@@ -187,11 +191,15 @@ export class NoteCreateService implements OnApplicationShutdown {
 		@Inject(DI.followingsRepository)
 		private followingsRepository: FollowingsRepository,
 
+		@Inject(DI.channelFollowingsRepository)
+		private channelFollowingsRepository: ChannelFollowingsRepository,
+
 		private userEntityService: UserEntityService,
 		private noteEntityService: NoteEntityService,
 		private idService: IdService,
 		private globalEventService: GlobalEventService,
 		private queueService: QueueService,
+		private redisTimelineService: RedisTimelineService,
 		private noteReadService: NoteReadService,
 		private notificationService: NotificationService,
 		private relayService: RelayService,
@@ -251,19 +259,30 @@ export class NoteCreateService implements OnApplicationShutdown {
 			}
 		}
 
-		// Renote対象が「ホームまたは全体」以外の公開範囲ならreject
-		if (data.renote && data.renote.visibility !== 'public' && data.renote.visibility !== 'home' && data.renote.userId !== user.id) {
-			throw new Error('Renote target is not public or home');
-		}
-
-		// Renote対象がpublicではないならhomeにする
-		if (data.renote && data.renote.visibility !== 'public' && data.visibility === 'public') {
-			data.visibility = 'home';
-		}
+		if (data.renote) {
+			switch (data.renote.visibility) {
+				case 'public':
+					// public noteは無条件にrenote可能
+					break;
+				case 'home':
+					// home noteはhome以下にrenote可能
+					if (data.visibility === 'public') {
+						data.visibility = 'home';
+					}
+					break;
+				case 'followers':
+					// 他人のfollowers noteはreject
+					if (data.renote.userId !== user.id) {
+						throw new Error('Renote target is not public or home');
+					}
 
-		// Renote対象がfollowersならfollowersにする
-		if (data.renote && data.renote.visibility === 'followers') {
-			data.visibility = 'followers';
+					// Renote対象がfollowersならfollowersにする
+					data.visibility = 'followers';
+					break;
+				case 'specified':
+					// specified / direct noteはreject
+					throw new Error('Renote target is not public or home');
+			}
 		}
 
 		// 返信対象がpublicではないならhomeにする
@@ -480,6 +499,8 @@ export class NoteCreateService implements OnApplicationShutdown {
 		// Increment notes count (user)
 		this.incNotesCountOfUser(user);
 
+		this.pushToTl(note, user);
+
 		// Word mute
 		mutedWordsCache.fetch(() => this.userProfilesRepository.find({
 			where: {
@@ -520,9 +541,8 @@ export class NoteCreateService implements OnApplicationShutdown {
 			});
 		}
 
-		// この投稿を除く指定したユーザーによる指定したノートのリノートが存在しないとき
-		if (data.renote && (await this.noteEntityService.countSameRenotes(user.id, data.renote.id, note.id) === 0)) {
-			if (!user.isBot) this.incRenoteCount(data.renote);
+		if (data.renote && data.renote.userId !== user.id && !user.isBot) {
+			this.incRenoteCount(data.renote);
 		}
 
 		if (data.poll && data.poll.expiresAt) {
@@ -812,6 +832,123 @@ export class NoteCreateService implements OnApplicationShutdown {
 		return mentionedUsers;
 	}
 
+	@bindThis
+	private async pushToTl(note: MiNote, user: { id: MiUser['id']; host: MiUser['host']; }) {
+		const meta = await this.metaService.fetch();
+
+		const r = this.redisClient.pipeline();
+
+		if (note.channelId) {
+			this.redisTimelineService.push(`channelTimeline:${note.channelId}`, note.id, this.config.perChannelMaxNoteCacheCount, r);
+
+			this.redisTimelineService.push(`userTimelineWithChannel:${user.id}`, note.id, 300, r);
+
+			const channelFollowings = await this.channelFollowingsRepository.find({
+				where: {
+					followeeId: note.channelId,
+				},
+				select: ['followerId'],
+			});
+
+			for (const channelFollowing of channelFollowings) {
+				this.redisTimelineService.push(`homeTimeline:${channelFollowing.followerId}`, note.id, 300, r);
+				if (note.fileIds.length > 0) {
+					this.redisTimelineService.push(`homeTimelineWithFiles:${channelFollowing.followerId}`, note.id, 300 / 2, r);
+				}
+			}
+		} else {
+			// TODO: キャッシュ？
+			// eslint-disable-next-line prefer-const
+			let [followings, userListMemberships] = await Promise.all([
+				this.followingsRepository.find({
+					where: {
+						followeeId: user.id,
+						followerHost: IsNull(),
+					},
+					select: ['followerId'],
+				}),
+				this.userListJoiningsRepository.find({
+					where: {
+						userId: user.id,
+					},
+					select: ['userListId', 'userListUserId', 'withReplies'],
+				}),
+			]);
+
+			if (note.visibility === 'followers') {
+				// TODO: 重そうだから何とかしたい Set 使う？
+				userListMemberships = userListMemberships.filter(x => followings.some(f => f.followerId === x.userListUserId));
+			}
+
+			// TODO: あまりにも数が多いと redisPipeline.exec に失敗する(理由は不明)ため、3万件程度を目安に分割して実行するようにする
+			for (const following of followings) {
+				// 基本的にvisibleUserIdsには自身のidが含まれている前提であること
+				if (note.visibility === 'specified' && !note.visibleUserIds.some(v => v === following.followerId)) continue;
+
+				// 自分自身以外への返信
+				if (note.replyId && note.replyUserId !== note.userId) {
+					if (!this.config.nirila.withRepliesInHomeTL) continue;
+				}
+
+				this.redisTimelineService.push(`homeTimeline:${following.followerId}`, note.id, 300, r);
+				if (note.fileIds.length > 0) {
+					this.redisTimelineService.push(`homeTimelineWithFiles:${following.followerId}`, note.id, 300 / 2, r);
+				}
+			}
+
+			for (const userListMembership of userListMemberships) {
+				// ダイレクトのとき、そのリストが対象外のユーザーの場合
+				if (
+					note.visibility === 'specified' &&
+					!note.visibleUserIds.some(v => v === userListMembership.userListUserId)
+				) continue;
+
+				// 自分自身以外への返信
+				if (note.replyId && note.replyUserId !== note.userId) {
+					if (!userListMembership.withReplies) continue;
+				}
+
+				this.redisTimelineService.push(`userListTimeline:${userListMembership.userListId}`, note.id, 300, r);
+				if (note.fileIds.length > 0) {
+					this.redisTimelineService.push(`userListTimelineWithFiles:${userListMembership.userListId}`, note.id, 300 / 2, r);
+				}
+			}
+
+			if (note.visibility !== 'specified' || !note.visibleUserIds.some(v => v === user.id)) { // 自分自身のHTL
+				this.redisTimelineService.push(`homeTimeline:${user.id}`, note.id, 300, r);
+				if (note.fileIds.length > 0) {
+					this.redisTimelineService.push(`homeTimelineWithFiles:${user.id}`, note.id, 300 / 2, r);
+				}
+			}
+
+			// 自分自身以外への返信
+			if (note.replyId && note.replyUserId !== note.userId) {
+				this.redisTimelineService.push(`userTimelineWithReplies:${user.id}`, note.id, 300, r);
+
+				if (note.visibility === 'public' && note.userHost == null) {
+					this.redisTimelineService.push('localTimelineWithReplies', note.id, 300, r);
+				}
+			} else {
+				this.redisTimelineService.push(`userTimeline:${user.id}`, note.id, 300, r);
+				if (note.fileIds.length > 0) {
+					this.redisTimelineService.push(`userTimelineWithFiles:${user.id}`, note.id, 300 / 2, r);
+				}
+
+				if (note.visibility === 'public' && note.userHost == null) {
+					this.redisTimelineService.push('localTimeline', note.id, 1000, r);
+					if (note.fileIds.length > 0) {
+						this.redisTimelineService.push('localTimelineWithFiles', note.id, 500, r);
+					}
+				}
+			}
+
+			if (Math.random() < 0.1) {
+			}
+		}
+
+		r.exec();
+	}
+
 	@bindThis
 	public dispose(): void {
 		this.#shutdownController.abort();