Implement Update Question (#4435)

* Update remote votes count

* save updatedAt

* deliver Update

* use renderNote

* use id

* fix typeof

src/models/note.ts

@@ -35,6 +35,7 @@ export type INote = {
 	_id: mongo.ObjectID;
 	createdAt: Date;
 	deletedAt: Date;
+	updatedAt?: Date;
 	fileIds: mongo.ObjectID[];
 	replyId: mongo.ObjectID;
 	renoteId: mongo.ObjectID;

src/queue/processors/http/process-inbox.ts

@@ -82,7 +82,7 @@ export default async (job: bq.Job, done: any): Promise<void> => {
 		}) as IRemoteUser;
 	}
 
-	// Update activityの場合は、ここで署名検証/更新処理まで実施して終了
+	// Update Person activityの場合は、ここで署名検証/更新処理まで実施して終了
 	if (activity.type === 'Update') {
 		if (activity.object && activity.object.type === 'Person') {
 			if (user == null) {
@@ -92,10 +92,10 @@ export default async (job: bq.Job, done: any): Promise<void> => {
 			} else {
 				updatePerson(activity.actor, null, activity.object);
 			}
-		}
 			done();
 			return;
 		}
+	}
 
 	// アクティビティを送信してきたユーザーがまだMisskeyサーバーに登録されていなかったら登録する
 	if (user === null) {

src/remote/activitypub/kernel/index.ts

@@ -2,6 +2,7 @@ import { Object } from '../type';
 import { IRemoteUser } from '../../../models/user';
 import create from './create';
 import performDeleteActivity from './delete';
+import performUpdateActivity from './update';
 import follow from './follow';
 import undo from './undo';
 import like from './like';
@@ -23,6 +24,10 @@ const self = async (actor: IRemoteUser, activity: Object): Promise<void> => {
 		await performDeleteActivity(actor, activity);
 		break;
 
+	case 'Update':
+		await performUpdateActivity(actor, activity);
+		break;
+
 	case 'Follow':
 		await follow(actor, activity);
 		break;

src/remote/activitypub/kernel/update/index.ts

@@ -0,0 +1,28 @@
+import { IRemoteUser } from '../../../../models/user';
+import { IUpdate, IObject } from '../../type';
+import { apLogger } from '../../logger';
+import { updateQuestion } from '../../models/question';
+
+/**
+ * Updateアクティビティを捌きます
+ */
+export default async (actor: IRemoteUser, activity: IUpdate): Promise<void> => {
+	if ('actor' in activity && actor.uri !== activity.actor) {
+		throw new Error('invalid actor');
+	}
+
+	apLogger.debug('Update');
+
+	const object = activity.object as IObject;
+
+	switch (object.type) {
+		case 'Question':
+			apLogger.debug('Question');
+			await updateQuestion(object).catch(e => console.log(e));
+			break;
+
+		default:
+			apLogger.warn(`Unknown type: ${object.type}`);
+			break;
+	}
+};

src/remote/activitypub/models/note.ts

@@ -18,6 +18,7 @@ import { extractPollFromQuestion } from './question';
 import vote from '../../../services/note/polls/vote';
 import { apLogger } from '../logger';
 import { IDriveFile } from '../../../models/drive-file';
+import { deliverQuestionUpdate } from '../../../services/note/polls/update';
 
 const logger = apLogger;
 
@@ -136,6 +137,9 @@ export async function createNote(value: any, resolver?: Resolver, silent = false
 			} else if (index >= 0) {
 				logger.info(`vote from AP: actor=${actor.username}@${actor.host}, note=${note.id}, choice=${name}`);
 				await vote(actor, reply, index);
+
+				// リモートフォロワーにUpdate配信
+				deliverQuestionUpdate(reply._id);
 			}
 			return null;
 		};

src/remote/activitypub/models/question.ts

@@ -1,18 +1,8 @@
-import { IChoice, IPoll } from '../../../models/note';
+import config from '../../../config';
+import Note, { IChoice, IPoll } from '../../../models/note';
 import Resolver from '../resolver';
-import { ICollection } from '../type';
-
-interface IQuestionChoice {
-	name?: string;
-	replies?: ICollection;
-	_misskey_votes?: number;
-}
-
-interface IQuestion {
-	oneOf?: IQuestionChoice[];
-	anyOf?: IQuestionChoice[];
-	endTime?: Date;
-}
+import { IQuestion } from '../type';
+import { apLogger } from '../logger';
 
 export async function extractPollFromQuestion(source: string | IQuestion): Promise<IPoll> {
 	const question = typeof source === 'string' ? await new Resolver().resolve(source) as IQuestion : source;
@@ -36,3 +26,54 @@ export async function extractPollFromQuestion(source: string | IQuestion): Promi
 		expiresAt
 	};
 }
+
+/**
+ * Update votes of Question
+ * @param uri URI of AP Question object
+ * @returns true if updated
+ */
+export async function updateQuestion(value: any) {
+	const uri = typeof value == 'string' ? value : value.id;
+
+	// URIがこのサーバーを指しているならスキップ
+	if (uri.startsWith(config.url + '/')) throw 'uri points local';
+
+	//#region このサーバーに既に登録されているか
+	const note = await Note.findOne({ uri });
+
+	if (note == null) throw 'Question is not registed';
+	//#endregion
+
+	// resolve new Question object
+	const resolver = new Resolver();
+	const question = await resolver.resolve(value) as IQuestion;
+	apLogger.debug(`fetched question: ${JSON.stringify(question, null, 2)}`);
+
+	if (question.type !== 'Question') throw 'object is not a Question';
+
+	const apChoices = question.oneOf || question.anyOf;
+	const dbChoices = note.poll.choices;
+
+	let changed = false;
+
+	for (const db of dbChoices) {
+		const oldCount = db.votes;
+		const newCount = apChoices.filter(ap => ap.name === db.text)[0].replies.totalItems;
+
+		if (oldCount != newCount) {
+			changed = true;
+			db.votes = newCount;
+		}
+	}
+
+	await Note.update({
+		_id: note._id
+	}, {
+		$set: {
+			'poll.choices': dbChoices,
+			updatedAt: new Date(),
+		}
+	});
+
+	return changed;
+}

src/remote/activitypub/type.ts

@@ -43,12 +43,28 @@ export interface IOrderedCollection extends IObject {
 }
 
 export interface INote extends IObject {
-	type: 'Note';
+	type: 'Note' | 'Question';
 	_misskey_content: string;
 	_misskey_quote: string;
 	_misskey_question: string;
 }
 
+export interface IQuestion extends IObject {
+	type: 'Note' | 'Question';
+	_misskey_content: string;
+	_misskey_quote: string;
+	_misskey_question: string;
+	oneOf?: IQuestionChoice[];
+	anyOf?: IQuestionChoice[];
+	endTime?: Date;
+}
+
+interface IQuestionChoice {
+	name?: string;
+	replies?: ICollection;
+	_misskey_votes?: number;
+}
+
 export interface IPerson extends IObject {
 	type: 'Person';
 	name: string;
@@ -81,6 +97,10 @@ export interface IDelete extends IActivity {
 	type: 'Delete';
 }
 
+export interface IUpdate extends IActivity {
+	type: 'Update';
+}
+
 export interface IUndo extends IActivity {
 	type: 'Undo';
 }
@@ -123,6 +143,7 @@ export type Object =
 	IOrderedCollection |
 	ICreate |
 	IDelete |
+	IUpdate |
 	IUndo |
 	IFollow |
 	IAccept |

src/server/api/endpoints/notes/polls/vote.ts

@@ -13,6 +13,7 @@ import { getNote } from '../../../common/getters';
 import { deliver } from '../../../../../queue';
 import { renderActivity } from '../../../../../remote/activitypub/renderer';
 import renderVote from '../../../../../remote/activitypub/renderer/vote';
+import { deliverQuestionUpdate } from '../../../../../services/note/polls/update';
 
 export const meta = {
 	desc: {
@@ -172,5 +173,8 @@ export default define(meta, async (ps, user) => {
 		deliver(user, renderActivity(await renderVote(user, vote, note, pollOwner)), pollOwner.inbox);
 	}
 
+	// リモートフォロワーにUpdate配信
+	deliverQuestionUpdate(note._id);
+
 	return;
 });

src/services/note/polls/update.ts

@@ -0,0 +1,61 @@
+import * as mongo from 'mongodb';
+import Note, { INote } from '../../../models/note';
+import { updateQuestion } from '../../../remote/activitypub/models/question';
+import ms = require('ms');
+import Logger from '../../logger';
+import User, { isLocalUser, isRemoteUser } from '../../../models/user';
+import Following from '../../../models/following';
+import renderUpdate from '../../../remote/activitypub/renderer/update';
+import { renderActivity } from '../../../remote/activitypub/renderer';
+import { deliver } from '../../../queue';
+import renderNote from '../../../remote/activitypub/renderer/note';
+
+const logger = new Logger('pollsUpdate');
+
+export async function triggerUpdate(note: INote) {
+	if (!note.updatedAt || Date.now() - new Date(note.updatedAt).getTime() > ms('1min')) {
+		logger.info(`Updating ${note._id}`);
+
+		try {
+			const updated = await updateQuestion(note.uri);
+			logger.info(`Updated ${note._id} ${updated ? 'changed' : 'nochange'}`);
+		} catch (e) {
+			logger.error(e);
+		}
+	}
+}
+
+export async function deliverQuestionUpdate(noteId: mongo.ObjectID) {
+	const note = await Note.findOne({
+		_id: noteId,
+	});
+
+	const user = await User.findOne({
+		_id: note.userId
+	});
+
+	const followers = await Following.find({
+		followeeId: user._id
+	});
+
+	const queue: string[] = [];
+
+	// フォロワーがリモートユーザーかつ投稿者がローカルユーザーならUpdateを配信
+	if (isLocalUser(user)) {
+		for (const following of followers) {
+			const follower = following._follower;
+
+			if (isRemoteUser(follower)) {
+				const inbox = follower.sharedInbox || follower.inbox;
+				if (!queue.includes(inbox)) queue.push(inbox);
+			}
+		}
+
+		if (queue.length > 0) {
+			const content = renderActivity(renderUpdate(await renderNote(note, false), user));
+			for (const inbox of queue) {
+				deliver(user, content, inbox);
+			}
+		}
+	}
+}

src/tools/refresh-question.ts

@@ -0,0 +1,14 @@
+import { updateQuestion } from '../remote/activitypub/models/question';
+
+async function main(uri: string): Promise<any> {
+	return await updateQuestion(uri);
+}
+
+const args = process.argv.slice(2);
+const uri = args[0];
+
+main(uri).then(result => {
+	console.log(`Done: ${result}`);
+}).catch(e => {
+	console.warn(e);
+});