Pick redis push tl

.config/docker_example.yml

@@ -95,6 +95,14 @@ redis:
 #  #prefix: example-prefix
 #  #db: 1
 
+#redisForTimelines:
+#  host: redis
+#  port: 6379
+#  #family: 0  # 0=Both, 4=IPv4, 6=IPv6
+#  #pass: example-pass
+#  #prefix: example-prefix
+#  #db: 1
+
 #   ┌───────────────────────────┐
 #───┘ MeiliSearch configuration └─────────────────────────────
 

.config/example.yml

@@ -105,6 +105,16 @@ redis:
 #  # You can specify more ioredis options...
 #  #username: example-username
 
+#redisForTimelines:
+#  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 └─────────────────────────────
 

.devcontainer/devcontainer.yml

@@ -95,6 +95,14 @@ redis:
 #  #prefix: example-prefix
 #  #db: 1
 
+#redisForTimelines:
+#  host: redis
+#  port: 6379
+#  #family: 0  # 0=Both, 4=IPv4, 6=IPv6
+#  #pass: example-pass
+#  #prefix: example-prefix
+#  #db: 1
+
 #   ┌───────────────────────────┐
 #───┘ MeiliSearch configuration └─────────────────────────────
 

chart/files/default.yml

@@ -116,6 +116,14 @@ redis:
 #  #prefix: example-prefix
 #  #db: 1
 
+#redisForTimelines:
+#  host: redis
+#  port: 6379
+#  #family: 0  # 0=Both, 4=IPv4, 6=IPv6
+#  #pass: example-pass
+#  #prefix: example-prefix
+#  #db: 1
+
 #   ┌───────────────────────────┐
 #───┘ MeiliSearch configuration └─────────────────────────────
 

packages/backend/jest.config.cjs

@@ -216,4 +216,6 @@ module.exports = {
 	maxWorkers: 1, // Make it use worker (that can be killed and restarted)
 	logHeapUsage: true, // To debug when out-of-memory happens on CI
 	workerIdleMemoryLimit: '1GiB', // Limit the worker to 1GB (GitHub Workflows dies at 2GB)
+
+	maxConcurrency: 32,
 };

packages/backend/migration/1696222183852-withReplies.js

@@ -0,0 +1,20 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and other misskey contributors
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+export class WithReplies1696222183852 {
+    name = 'WithReplies1696222183852'
+
+    async up(queryRunner) {
+        await queryRunner.query(`ALTER TABLE "following" ADD "withReplies" boolean NOT NULL DEFAULT false`);
+        await queryRunner.query(`ALTER TABLE "user_list_joining" ADD "withReplies" boolean NOT NULL DEFAULT true`);
+        await queryRunner.query(`CREATE INDEX "IDX_d74d8ab5efa7e3bb82825c0fa2" ON "following" ("followeeId", "followerHost") `);
+    }
+
+    async down(queryRunner) {
+        await queryRunner.query(`DROP INDEX "public"."IDX_d74d8ab5efa7e3bb82825c0fa2"`);
+        await queryRunner.query(`ALTER TABLE "user_list_joining" DROP COLUMN "withReplies"`);
+        await queryRunner.query(`ALTER TABLE "following" DROP COLUMN "withReplies"`);
+    }
+}

packages/backend/migration/1696323464251-user-list-membership.js

@@ -0,0 +1,11 @@
+export class UserListMembership1696323464251 {
+    name = 'UserListMembership1696323464251'
+
+    async up(queryRunner) {
+        await queryRunner.query(`ALTER TABLE "user_list_joining" RENAME TO "user_list_membership"`);
+    }
+
+    async down(queryRunner) {
+			await queryRunner.query(`ALTER TABLE "user_list_membership" RENAME TO "user_list_joining"`);
+    }
+}

packages/backend/migration/1696331570827-hibernation.js

@@ -0,0 +1,17 @@
+export class Hibernation1696331570827 {
+    name = 'Hibernation1696331570827'
+
+    async up(queryRunner) {
+				await queryRunner.query(`DROP INDEX "public"."IDX_d74d8ab5efa7e3bb82825c0fa2"`);
+        await queryRunner.query(`ALTER TABLE "user" ADD "isHibernated" boolean NOT NULL DEFAULT false`);
+        await queryRunner.query(`ALTER TABLE "following" ADD "isFollowerHibernated" boolean NOT NULL DEFAULT false`);
+        await queryRunner.query(`CREATE INDEX "IDX_ce62b50d882d4e9dee10ad0d2f" ON "following" ("followeeId", "followerHost", "isFollowerHibernated") `);
+    }
+
+    async down(queryRunner) {
+        await queryRunner.query(`DROP INDEX "public"."IDX_ce62b50d882d4e9dee10ad0d2f"`);
+        await queryRunner.query(`ALTER TABLE "following" DROP COLUMN "isFollowerHibernated"`);
+        await queryRunner.query(`ALTER TABLE "user" DROP COLUMN "isHibernated"`);
+        await queryRunner.query(`CREATE INDEX "IDX_d74d8ab5efa7e3bb82825c0fa2" ON "following" ("followeeId", "followerHost") `);
+    }
+}

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/GlobalModule.ts

@@ -70,18 +70,27 @@ const $redisForSub: Provider = {
 	inject: [DI.config],
 };
 
+const $redisForTimelines: Provider = {
+	provide: DI.redisForTimelines,
+	useFactory: (config: Config) => {
+		return new Redis.Redis(config.redisForTimelines);
+	},
+	inject: [DI.config],
+};
+
 @Global()
 @Module({
 	imports: [RepositoryModule],
-	providers: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub],
-	exports: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, RepositoryModule],
+	providers: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines],
+	exports: [$config, $db, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines, RepositoryModule],
 })
 export class GlobalModule implements OnApplicationShutdown {
 	constructor(
 		@Inject(DI.db) private db: DataSource,
 		@Inject(DI.redis) private redisClient: Redis.Redis,
 		@Inject(DI.redisForPub) private redisForPub: Redis.Redis,
 		@Inject(DI.redisForSub) private redisForSub: Redis.Redis,
+		@Inject(DI.redisForTimelines) private redisForTimelines: Redis.Redis,
 	) {}
 
 	public async dispose(): Promise<void> {
@@ -98,6 +107,7 @@ export class GlobalModule implements OnApplicationShutdown {
 			this.redisClient.disconnect(),
 			this.redisForPub.disconnect(),
 			this.redisForSub.disconnect(),
+			this.redisForTimelines.disconnect(),
 		]);
 	}
 

packages/backend/src/config.ts

@@ -47,6 +47,7 @@ type Source = {
 	redis: RedisOptionsSource;
 	redisForPubsub?: RedisOptionsSource;
 	redisForJobQueue?: RedisOptionsSource;
+	redisForTimelines?: RedisOptionsSource;
 	meilisearch?: {
 		host: string;
 		port: string;
@@ -167,6 +168,7 @@ export type Config = {
 	redis: RedisOptions & RedisOptionsSource;
 	redisForPubsub: RedisOptions & RedisOptionsSource;
 	redisForJobQueue: RedisOptions & RedisOptionsSource;
+	redisForTimelines: RedisOptions & RedisOptionsSource;
 	perChannelMaxNoteCacheCount: number;
 	perUserNotificationsMaxCount: number;
 	deactivateAntennaThreshold: number;
@@ -241,6 +243,7 @@ export function loadConfig(): Config {
 		redis,
 		redisForPubsub: config.redisForPubsub ? convertRedisOptions(config.redisForPubsub, host) : redis,
 		redisForJobQueue: config.redisForJobQueue ? convertRedisOptions(config.redisForJobQueue, host) : redis,
+		redisForTimelines: config.redisForTimelines ? convertRedisOptions(config.redisForTimelines, host) : redis,
 		id: config.id,
 		proxy: config.proxy,
 		proxySmtp: config.proxySmtp,

packages/backend/src/core/AccountMoveService.ts

@@ -9,7 +9,7 @@ import { IsNull, In, MoreThan, Not } from 'typeorm';
 import { bindThis } from '@/decorators.js';
 import { DI } from '@/di-symbols.js';
 import type { MiLocalUser, MiRemoteUser, MiUser } from '@/models/User.js';
-import type { BlockingsRepository, FollowingsRepository, InstancesRepository, MutingsRepository, UserListJoiningsRepository, UsersRepository } from '@/models/_.js';
+import type { BlockingsRepository, FollowingsRepository, InstancesRepository, MutingsRepository, UserListMembershipsRepository, UsersRepository } from '@/models/_.js';
 import type { RelationshipJobData, ThinUser } from '@/queue/types.js';
 
 import { IdService } from '@/core/IdService.js';
@@ -42,8 +42,8 @@ export class AccountMoveService {
 		@Inject(DI.mutingsRepository)
 		private mutingsRepository: MutingsRepository,
 
-		@Inject(DI.userListJoiningsRepository)
-		private userListJoiningsRepository: UserListJoiningsRepository,
+		@Inject(DI.userListMembershipsRepository)
+		private userListMembershipsRepository: UserListMembershipsRepository,
 
 		@Inject(DI.instancesRepository)
 		private instancesRepository: InstancesRepository,
@@ -215,40 +215,41 @@ export class AccountMoveService {
 	@bindThis
 	public async updateLists(src: ThinUser, dst: MiUser): Promise<void> {
 		// Return if there is no list to be updated.
-		const oldJoinings = await this.userListJoiningsRepository.find({
+		const oldMemberships = await this.userListMembershipsRepository.find({
 			where: {
 				userId: src.id,
 			},
 		});
-		if (oldJoinings.length === 0) return;
+		if (oldMemberships.length === 0) return;
 
-		const existingUserListIds = await this.userListJoiningsRepository.find({
+		const existingUserListIds = await this.userListMembershipsRepository.find({
 			where: {
 				userId: dst.id,
 			},
-		}).then(joinings => joinings.map(joining => joining.userListId));
+		}).then(memberships => memberships.map(membership => membership.userListId));
 
-		const newJoinings: Map<string, { createdAt: Date; userId: string; userListId: string; }> = new Map();
+		const newMemberships: Map<string, { createdAt: Date; userId: string; userListId: string; userListUserId: string; }> = new Map();
 
 		// 重複しないようにIDを生成
 		const genId = (): string => {
 			let id: string;
 			do {
 				id = this.idService.genId();
-			} while (newJoinings.has(id));
+			} while (newMemberships.has(id));
 			return id;
 		};
-		for (const joining of oldJoinings) {
-			if (existingUserListIds.includes(joining.userListId)) continue; // skip if dst exists in this user's list
-			newJoinings.set(genId(), {
+		for (const membership of oldMemberships) {
+			if (existingUserListIds.includes(membership.userListId)) continue; // skip if dst exists in this user's list
+			newMemberships.set(genId(), {
 				createdAt: new Date(),
 				userId: dst.id,
-				userListId: joining.userListId,
+				userListId: membership.userListId,
+				userListUserId: membership.userListUserId,
 			});
 		}
 
-		const arrayToInsert = Array.from(newJoinings.entries()).map(entry => ({ ...entry[1], id: entry[0] }));
-		await this.userListJoiningsRepository.insert(arrayToInsert);
+		const arrayToInsert = Array.from(newMemberships.entries()).map(entry => ({ ...entry[1], id: entry[0] }));
+		await this.userListMembershipsRepository.insert(arrayToInsert);
 
 		// Have the proxy account follow the new account in the same way as UserListService.push
 		if (this.userEntityService.isRemoteUser(dst)) {

packages/backend/src/core/AntennaService.ts

@@ -12,10 +12,11 @@ import { GlobalEventService } from '@/core/GlobalEventService.js';
 import * as Acct from '@/misc/acct.js';
 import type { Packed } from '@/misc/json-schema.js';
 import { DI } from '@/di-symbols.js';
-import type { AntennasRepository, UserListJoiningsRepository } from '@/models/_.js';
+import type { AntennasRepository, UserListMembershipsRepository } from '@/models/_.js';
 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()
@@ -24,20 +25,21 @@ export class AntennaService implements OnApplicationShutdown {
 	private antennas: MiAntenna[];
 
 	constructor(
-		@Inject(DI.redis)
-		private redisClient: Redis.Redis,
+		@Inject(DI.redisForTimelines)
+		private redisForTimelines: Redis.Redis,
 
 		@Inject(DI.redisForSub)
 		private redisForSub: Redis.Redis,
 
 		@Inject(DI.antennasRepository)
 		private antennasRepository: AntennasRepository,
 
-		@Inject(DI.userListJoiningsRepository)
-		private userListJoiningsRepository: UserListJoiningsRepository,
+		@Inject(DI.userListMembershipsRepository)
+		private userListMembershipsRepository: UserListMembershipsRepository,
 
 		private utilityService: UtilityService,
 		private globalEventService: GlobalEventService,
+		private redisTimelineService: RedisTimelineService,
 	) {
 		this.antennasFetched = false;
 		this.antennas = [];
@@ -81,15 +83,15 @@ export class AntennaService implements OnApplicationShutdown {
 		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);
 
-		const redisPipeline = this.redisClient.pipeline();
+		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);
 		}
 
@@ -108,7 +110,7 @@ export class AntennaService implements OnApplicationShutdown {
 		if (antenna.src === 'home') {
 			// TODO
 		} else if (antenna.src === 'list') {
-			const listUsers = (await this.userListJoiningsRepository.findBy({
+			const listUsers = (await this.userListMembershipsRepository.findBy({
 				userListId: antenna.userListId!,
 			})).map(x => x.userId);
 

packages/backend/src/core/CacheService.ts

@@ -5,7 +5,7 @@
 
 import { Inject, Injectable } from '@nestjs/common';
 import * as Redis from 'ioredis';
-import type { BlockingsRepository, ChannelFollowingsRepository, FollowingsRepository, MutingsRepository, RenoteMutingsRepository, MiUserProfile, UserProfilesRepository, UsersRepository } from '@/models/_.js';
+import type { BlockingsRepository, ChannelFollowingsRepository, FollowingsRepository, MutingsRepository, RenoteMutingsRepository, MiUserProfile, UserProfilesRepository, UsersRepository, MiFollowing } from '@/models/_.js';
 import { MemoryKVCache, RedisKVCache } from '@/misc/cache.js';
 import type { MiLocalUser, MiUser } from '@/models/User.js';
 import { DI } from '@/di-symbols.js';