feat(calls): YouTubeを同期視聴するWatch Togetherを追加

.github/misskey/test.yml

@@ -17,3 +17,9 @@ redis:
 id: aidx
 
 proxyRemoteFiles: true
+
+# Calls API tests do not establish real audio connections.
+cloudflareRealtime:
+  enabled: true
+  appId: test-app
+  appSecret: test-secret

locales/ja-JP.yml

@@ -3996,3 +3996,29 @@ _events:
   tagsCaption: "カンマ区切りで入力"
   rateLimitReached: "本日のイベント作成上限に達しました"
   unknownError: "エラーが発生しました"
+
+_watchTogether:
+  title: "Watch Together"
+  description: "YouTubeの動画を一緒に見る"
+  videoUrl: "YouTubeの動画URL"
+  shareVideo: "この動画を見る"
+  invalidUrl: "YouTubeの動画URLを入力してください"
+  empty: "ホスト・モデレーターが動画を選ぶと、ここに表示されます"
+  hostControls: "ホスト・モデレーターの操作に同期します"
+  sharingControls: "YouTubeの操作を共有中"
+  changeVideo: "動画を変更"
+  playNow: "今すぐ再生"
+  addToQueue: "キューに追加"
+  removeFromQueue: "キューから削除"
+  queue: "次に再生"
+  emptyQueue: "キューは空です。動画を追加すると、再生終了後に順番に再生します。"
+  queueFull: "キューに追加できるのは50本までです"
+  startWatching: "視聴を開始"
+  startWatchingDescription: "YouTubeのプレーヤーに接続します。再生位置はルームに合わせます。"
+  clear: "動画を閉じる"
+  playerError: "動画を再生できません。動画が非公開・削除済み、または埋め込みが許可されていない可能性があります。"
+  apiError: "視聴状態を取得・更新できませんでした"
+  autoplayBlocked: "ブラウザーが自動再生を止めました。「再生を許可」を押してください。"
+  allowPlayback: "再生を許可"
+  openYouTube: "YouTubeで開く"
+  ended: "ルームが終了しました"

packages/backend/src/core/CoreModule.ts

@@ -90,6 +90,7 @@ import { LoginBonusService } from './LoginBonusService.js';
 import { EventService } from './EventService.js';
 import { CallsLiveConnectionService } from './calls/CallsLiveConnectionService.js';
 import { CallsRoomService } from './calls/CallsRoomService.js';
+import { CallsWatchTogetherService } from './calls/CallsWatchTogetherService.js';
 import { CloudflareRealtimeClient } from './calls/CloudflareRealtimeClient.js';
 import { CallsMediaBindingService } from './calls/CallsMediaBindingService.js';
 import { CallsMediaService } from './calls/CallsMediaService.js';
@@ -427,6 +428,7 @@ const $ApQuestionService: Provider = { provide: 'ApQuestionService', useExisting
 		LoginBonusService,
 		EventService,
 		CallsRoomService,
+		CallsWatchTogetherService,
 		CallsLiveConnectionService,
 		CloudflareRealtimeClient,
 		CallsMediaBindingService,
@@ -762,6 +764,7 @@ const $ApQuestionService: Provider = { provide: 'ApQuestionService', useExisting
 
 		EventService,
 		CallsRoomService,
+		CallsWatchTogetherService,
 		CallsLiveConnectionService,
 		CloudflareRealtimeClient,
 		CallsMediaBindingService,

packages/backend/src/core/GlobalEventService.ts

@@ -21,6 +21,7 @@ import type { MiWebhook } from '@/models/Webhook.js';
 import type { MiSystemWebhook } from '@/models/SystemWebhook.js';
 import type { MiMeta } from '@/models/Meta.js';
 import { MiAvatarDecoration, MiChatMessage, MiChatRoom, MiReversiGame, MiRole, MiRoleAssignment } from '@/models/_.js';
+import type { WatchTogetherSnapshot } from '@/core/calls/CallsWatchTogetherService.js';
 import type { MiCallsRoom } from '@/models/CallsRoom.js';
 import type { Packed } from '@/misc/json-schema.js';
 import { DI } from '@/di-symbols.js';
@@ -36,6 +37,7 @@ type CallsParticipantWithUser = Packed<'CallsParticipant'> & {
 	user: (Packed<'UserLite'> & { isFollowing: boolean; isFollowed: boolean }) | null;
 };
 export interface CallsRoomEventTypes {
+	watchTogether: CallsRoomEventBase & { state: WatchTogetherSnapshot };
 	title: CallsRoomEventBase & { title: string };
 	lifecycle: CallsRoomEventBase & { state: 'scheduled' | 'open' | 'ended' | 'cancelled'; reason?: 'host-timeout' };
 	participant: CallsRoomEventBase & { participantId: string; action: 'joined' | 'left' | 'removed' | 'updated'; participant?: CallsParticipantWithUser; moderatorUserIds?: string[] };

packages/backend/src/core/calls/CallsRoomService.ts

@@ -764,7 +764,7 @@ export class CallsRoomService implements OnModuleInit, OnApplicationShutdown {
 		this.callsTelemetryService.lifecycle({ action: 'participant-removed', roomId, participantId: participant.id, reason: 'moderation' });
 	}
 
-	private async assertCanModerateParticipants(actor: MiUser, room: MiCallsRoom): Promise<void> {
+	public async assertCanModerateParticipants(actor: MiUser, room: MiCallsRoom): Promise<void> {
 		if (room.state !== 'open') throw new CallsRoomError('invalid-state');
 		if (room.ownerUserId === actor.id || await this.roleService.isModerator(actor)) return;
 		if (!room.moderatorUserIds.includes(actor.id) || !(await this.callsParticipantsRepository.existsBy({ roomId: room.id, userId: actor.id, state: 'active' }))) throw new CallsRoomError('access-denied');

packages/backend/src/core/calls/CallsWatchTogetherService.ts

@@ -0,0 +1,82 @@
+/*
+ * 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 { CallsParticipantsRepository, MiUser } from '@/models/_.js';
+import { CallsRoomError, CallsRoomService } from './CallsRoomService.js';
+import { CallsLiveConnectionService } from './CallsLiveConnectionService.js';
+import { CallsEventService } from './CallsEventService.js';
+
+export type WatchTogetherState = {
+	queue: string[];
+	videoId: string | null;
+	playing: boolean;
+	position: number;
+	updatedAt: number;
+	revision: number;
+};
+export type WatchTogetherSnapshot = WatchTogetherState & { serverTime: number };
+
+@Injectable()
+export class CallsWatchTogetherService {
+	constructor(
+		@Inject(DI.redis) private redis: Redis.Redis,
+		@Inject(DI.callsParticipantsRepository) private participants: CallsParticipantsRepository,
+		private rooms: CallsRoomService,
+		private connections: CallsLiveConnectionService,
+		private events: CallsEventService,
+	) {}
+
+	public async assertAccess(user: MiUser, roomId: string) {
+		const room = await this.rooms.getRoom(roomId);
+		await this.rooms.assertCanAccess(user, room);
+		if (await this.participants.existsBy({ roomId, userId: user.id, state: 'removed' })) throw new CallsRoomError('access-denied');
+		return room;
+	}
+
+	private async read(roomId: string): Promise<WatchTogetherState> {
+		const raw = await this.redis.get(`calls:watch-together:${roomId}`);
+		return { queue: [], ...(raw == null ? { videoId: null, playing: false, position: 0, updatedAt: Date.now(), revision: 0 } : JSON.parse(raw) as WatchTogetherState) };
+	}
+
+	public async show(user: MiUser, roomId: string): Promise<WatchTogetherSnapshot> {
+		const room = await this.assertAccess(user, roomId);
+		const state = await this.read(roomId);
+		if (room.state === 'open' && room.ownerUserId === user.id) await this.connections.touchHost(roomId);
+		if (room.state !== 'open' && state.playing) {
+			state.position += Math.max(0, (room.endedAt?.getTime() ?? Date.now()) - state.updatedAt) / 1000;
+			state.playing = false;
+		}
+		return { ...state, serverTime: Date.now() };
+	}
+
+	public async update(user: MiUser, roomId: string, params: { expectedRevision: number; videoId?: string | null; playing?: boolean; position?: number; queue?: string[] }): Promise<WatchTogetherSnapshot> {
+		return this.connections.withRoomLock(roomId, async assertHeld => {
+			const room = await this.assertAccess(user, roomId);
+			await this.rooms.assertCanModerateParticipants(user, room);
+			const previous = await this.read(roomId);
+			if (previous.revision !== params.expectedRevision) throw new CallsRoomError('stale-revision');
+			const now = Date.now();
+			const changingVideo = params.videoId !== undefined;
+			const videoId = changingVideo ? params.videoId! : previous.videoId;
+			const state: WatchTogetherState = {
+				queue: params.queue ?? previous.queue,
+				videoId,
+				playing: videoId != null && (params.playing ?? (changingVideo ? false : previous.playing)),
+				position: videoId == null ? 0 : params.position ?? (changingVideo ? 0 : previous.position + (previous.playing ? (now - previous.updatedAt) / 1000 : 0)),
+				updatedAt: now,
+				revision: previous.revision + 1,
+			};
+			await assertHeld();
+			await this.redis.set(`calls:watch-together:${roomId}`, JSON.stringify(state), 'EX', 7 * 24 * 60 * 60);
+			if (room.ownerUserId === user.id) await this.connections.touchHost(roomId);
+			const snapshot = { ...state, serverTime: Date.now() };
+			await this.events.publish(roomId, room.revision, 'watchTogether', { state: snapshot });
+			return snapshot;
+		});
+	}
+}

packages/backend/src/server/api/endpoint-list.ts

@@ -428,6 +428,8 @@ export * as 'calls/media/tracks/close' from './endpoints/calls/media/close.js';
 export * as 'calls/media/reconcile' from './endpoints/calls/media/reconcile.js';
 export * as 'calls/media/turn-credentials' from './endpoints/calls/media/turn-credentials.js';
 export * as 'calls/media/credential/refresh' from './endpoints/calls/media/credential-refresh.js';
+export * as 'calls/watch-together/show' from './endpoints/calls/watch-together/show.js';
+export * as 'calls/watch-together/update' from './endpoints/calls/watch-together/update.js';
 export * as 'point/send' from './endpoints/point/send.js';
 export * as 'pinned-users' from './endpoints/pinned-users.js';
 export * as 'promo/read' from './endpoints/promo/read.js';

packages/backend/src/server/api/endpoints/calls/watch-together/_shared.ts

@@ -0,0 +1,17 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+export const watchTogetherSchema = {
+	type: 'object', optional: false, nullable: false,
+	properties: {
+		queue: { type: 'array', optional: false, nullable: false, items: { type: 'string', optional: false, nullable: false } },
+		videoId: { type: 'string', optional: false, nullable: true },
+		playing: { type: 'boolean', optional: false, nullable: false },
+		position: { type: 'number', optional: false, nullable: false },
+		updatedAt: { type: 'number', optional: false, nullable: false },
+		revision: { type: 'integer', optional: false, nullable: false },
+		serverTime: { type: 'number', optional: false, nullable: false },
+	},
+} as const;

packages/backend/src/server/api/endpoints/calls/watch-together/show.ts

@@ -0,0 +1,29 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { Injectable } from '@nestjs/common';
+import { CallsWatchTogetherService } from '@/core/calls/CallsWatchTogetherService.js';
+import { Endpoint } from '@/server/api/endpoint-base.js';
+import { callsApiError, callsErrors } from '../_shared.js';
+import { watchTogetherSchema } from './_shared.js';
+
+export const meta = {
+	tags: ['calls'], stability: 'experimental', requireCredential: true, kind: 'read:calls',
+
+	errors: callsErrors, res: watchTogetherSchema,
+} as const;
+export const paramDef = {
+	type: 'object', properties: { roomId: { type: 'string', format: 'misskey:id' } },
+	required: ['roomId'],
+} as const;
+
+@Injectable()
+export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint-disable-line import/no-default-export
+	constructor(service: CallsWatchTogetherService) {
+		super(meta, paramDef, async (ps, me) => {
+			try { return await service.show(me, ps.roomId); } catch (error) { callsApiError(error); }
+		});
+	}
+}

packages/backend/src/server/api/endpoints/calls/watch-together/update.ts

@@ -0,0 +1,36 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { Injectable } from '@nestjs/common';
+import { CallsWatchTogetherService } from '@/core/calls/CallsWatchTogetherService.js';
+import { Endpoint } from '@/server/api/endpoint-base.js';
+import { callsApiError, callsErrors } from '../_shared.js';
+import { watchTogetherSchema } from './_shared.js';
+
+export const meta = {
+	tags: ['calls'], stability: 'experimental', requireCredential: true, kind: 'write:calls',
+	prohibitMoved: true, limit: { duration: 60000, max: 60 },
+	errors: callsErrors, res: watchTogetherSchema,
+} as const;
+export const paramDef = {
+	type: 'object', properties: {
+		roomId: { type: 'string', format: 'misskey:id' },
+		queue: { type: 'array', maxItems: 50, items: { type: 'string', pattern: '^[A-Za-z0-9_-]{11}$' } },
+		expectedRevision: { type: 'integer', minimum: 0 },
+		videoId: { type: 'string', nullable: true, pattern: '^[A-Za-z0-9_-]{11}$' },
+		playing: { type: 'boolean' },
+		position: { type: 'number', minimum: 0 },
+	},
+	required: ['roomId', 'expectedRevision'],
+} as const;
+
+@Injectable()
+export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint-disable-line import/no-default-export
+	constructor(service: CallsWatchTogetherService) {
+		super(meta, paramDef, async (ps, me) => {
+			try { return await service.update(me, ps.roomId, ps); } catch (error) { callsApiError(error); }
+		});
+	}
+}

packages/backend/src/server/api/stream/channels/calls-room.ts

@@ -8,6 +8,7 @@ import { REQUEST } from '@nestjs/core';
 import { bindThis } from '@/decorators.js';
 import type { GlobalEvents } from '@/core/GlobalEventService.js';
 import { CallsFeatureDisabledError, CallsRoomError, CallsRoomService } from '@/core/calls/CallsRoomService.js';
+import { CallsWatchTogetherService } from '@/core/calls/CallsWatchTogetherService.js';
 import { CallsMediaService } from '@/core/calls/CallsMediaService.js';
 import { CallsLiveConnectionService } from '@/core/calls/CallsLiveConnectionService.js';
 import { CallsEntityService } from '@/core/entities/CallsEntityService.js';
@@ -27,6 +28,7 @@ export class CallsRoomChannel extends Channel {
 	constructor(
 		@Inject(REQUEST) request: ChannelRequest,
 		private callsRoomService: CallsRoomService,
+		private watchTogetherService: CallsWatchTogetherService,
 		private callsMediaService: CallsMediaService,
 		@Inject(DI.callsParticipantsRepository)
 		private callsParticipantsRepository: CallsParticipantsRepository,
@@ -53,6 +55,7 @@ export class CallsRoomChannel extends Channel {
 		try {
 			room = await this.callsRoomService.getRoom(this.roomId);
 			await this.callsRoomService.assertCanAccess(this.user, room);
+			if (data.type === 'watchTogether') await this.watchTogetherService.assertAccess(this.user, this.roomId);
 		} catch (error) {
 			this.dispose();
 			if (error instanceof CallsFeatureDisabledError || (error instanceof CallsRoomError && error.code === 'access-denied')) {

packages/backend/test/e2e/calls.ts

@@ -6,7 +6,7 @@
 process.env.NODE_ENV = 'test';
 
 import { beforeAll, describe, expect, test } from 'vitest';
-import { api, castAsError, signup } from '../utils.js';
+import { api, castAsError, createAppToken, signup } from '../utils.js';
 import type * as misskey from 'misskey-js';
 
 describe('Calls', () => {
@@ -60,4 +60,44 @@ describe('Calls', () => {
 			expect(castAsError(response.body ?? {}).error.code).toBe('CALLS_INVALID_STATE');
 		});
 	});
+	describe('Watch Together', () => {
+		test('共有状態の取得と操作権限、入力、終了後の制限', async () => {
+			const viewer = await signup({ username: 'watch_viewer' });
+			const created = await api('calls/rooms/create', { attachmentType: 'personal', title: 'Watch Together', visibility: 'public' }, alice);
+			expect(created.status).toBe(200);
+			const roomId = created.body.id;
+			await api('calls/rooms/open', { roomId, expectedRevision: created.body.revision }, alice);
+			const invalid = await api('calls/watch-together/update', { roomId, expectedRevision: 0, videoId: 'https://example.com/video' }, alice);
+			expect(invalid.status).toBe(400);
+			const selected = await api('calls/watch-together/update', { roomId, expectedRevision: 0, videoId: 'M7lc1UVf-VE' }, alice);
+			expect(selected.status).toBe(200);
+			const show = await api('calls/watch-together/show', { roomId }, viewer);
+			expect(show.status).toBe(200);
+			expect(show.body).toMatchObject({ videoId: 'M7lc1UVf-VE', playing: false, revision: 1 });
+			const denied = await api('calls/watch-together/update', { roomId, expectedRevision: 1, playing: true }, viewer);
+			expect(castAsError(denied.body).error.code).toBe('CALLS_ACCESS_DENIED');
+			const readOnlyToken = await createAppToken(alice, ['read:calls']);
+			expect((await api('calls/watch-together/update', { roomId, expectedRevision: 1, playing: true }, { token: readOnlyToken })).status).toBe(403);
+			const playing = await api('calls/watch-together/update', { roomId, expectedRevision: 1, playing: true, position: 30 }, alice);
+			expect(playing.body).toMatchObject({ playing: true, position: 30, revision: 2 });
+			const queued = await api('calls/watch-together/update', { roomId, expectedRevision: 2, queue: ['dQw4w9WgXcQ'] }, alice);
+			expect(queued.status).toBe(200);
+			expect(queued.body).toMatchObject({ videoId: 'M7lc1UVf-VE', playing: true, queue: ['dQw4w9WgXcQ'], revision: 3 });
+			expect((await api('calls/watch-together/show', { roomId }, viewer)).body.queue).toEqual(['dQw4w9WgXcQ']);
+			expect(castAsError((await api('calls/watch-together/update', { roomId, expectedRevision: 3, queue: [] }, viewer)).body).error.code).toBe('CALLS_ACCESS_DENIED');
+			expect((await api('calls/watch-together/update', { roomId, expectedRevision: 3, queue: ['invalid'] }, alice)).status).toBe(400);
+			const snapshot = await api('calls/rooms/show', { roomId }, alice);
+			await api('calls/rooms/end', { roomId, expectedRevision: snapshot.body.room.revision }, alice);
+			expect((await api('calls/watch-together/show', { roomId }, viewer)).body.playing).toBe(false);
+			expect(castAsError((await api('calls/watch-together/update', { roomId, expectedRevision: 2, playing: true }, alice)).body).error.code).toBe('CALLS_INVALID_STATE');
+		});
+
+		test('非公開ルームの視聴状態を外部ユーザーに公開しない', async () => {
+			const outsider = await signup({ username: 'watch_outsider' });
+			const room = await api('calls/rooms/create', { attachmentType: 'personal', title: 'Private Watch Together', visibility: 'specified', visibleUserIds: [] }, alice);
+			expect(room.status).toBe(200);
+			expect(castAsError((await api('calls/watch-together/show', { roomId: room.body.id }, outsider)).body).error.code).toBe('CALLS_ACCESS_DENIED');
+		});
+	});
+
 });