wip: resolvers

packages/backend/native-utils/scylla-migration/cql/1689400417034_timeline/up.cql

@@ -244,6 +244,7 @@ CREATE MATERIALIZED VIEW reaction_by_id AS
 CREATE TABLE poll_vote (
 	"noteId" ascii,
 	"userId" ascii,
+	"userHost" text,
 	"choice" set<int>,
 	"createdAt" timestamp,
 	PRIMARY KEY ("noteId", "userId")

packages/backend/src/db/cql.ts

@@ -160,7 +160,7 @@ export const scyllaQueries = {
 	},
 	poll: {
 		select: `SELECT * FROM poll_vote WHERE "noteId" = ?`,
-		insert: `INSERT INTO poll_vote ("noteId", "userId", "choice", "createdAt") VALUES (?, ?, ?, ?)`,
+		insert: `INSERT INTO poll_vote ("noteId", "userId", "userHost", "choice", "createdAt") VALUES (?, ?, ?, ?, ?)`,
 	},
 	notification: {
 		insert: `INSERT INTO notification

packages/backend/src/db/scylla.ts

@@ -155,6 +155,7 @@ export interface ScyllaPoll {
 export interface ScyllaPollVote {
 	noteId: string;
 	userId: string;
+	userHost: string | null;
 	choice: Set<number>;
 	createdAt: Date;
 }
@@ -163,6 +164,7 @@ export function parseScyllaPollVote(row: types.Row): ScyllaPollVote {
 	return {
 		noteId: row.get("noteId"),
 		userId: row.get("userId"),
+		userHost: row.get("userHost") ?? null,
 		choice: new Set(row.get("choice") ?? []),
 		createdAt: row.get("createdAt"),
 	};

packages/backend/src/queue/processors/ended-poll-notification.ts

@@ -4,6 +4,14 @@ import { queueLogger } from "../logger.js";
 import type { EndedPollNotificationJobData } from "@/queue/types.js";
 import { createNotification } from "@/services/create-notification.js";
 import { deliverQuestionUpdate } from "@/services/note/polls/update.js";
+import {
+	parseScyllaNote,
+	parseScyllaPollVote,
+	prepared,
+	scyllaClient,
+	type ScyllaNote,
+} from "@/db/scylla.js";
+import type { Note } from "@/models/entities/note.js";
 
 const logger = queueLogger.createSubLogger("ended-poll-notification");
 
@@ -11,22 +19,45 @@ export async function endedPollNotification(
 	job: Bull.Job<EndedPollNotificationJobData>,
 	done: any,
 ): Promise<void> {
-	const note = await Notes.findOneBy({ id: job.data.noteId });
-	if (note == null || !note.hasPoll) {
+	let note: Note | ScyllaNote | null = null;
+	if (scyllaClient) {
+		const result = await scyllaClient.execute(
+			prepared.note.select.byId,
+			[job.data.noteId],
+			{ prepare: true },
+		);
+		if (result.rowLength > 0) {
+			note = parseScyllaNote(result.first());
+		}
+	} else {
+		note = await Notes.findOneBy({ id: job.data.noteId });
+	}
+
+	if (!note?.hasPoll) {
 		done();
 		return;
 	}
 
-	const votes = await PollVotes.createQueryBuilder("vote")
-		.select("vote.userId")
-		.where("vote.noteId = :noteId", { noteId: note.id })
-		.innerJoinAndSelect("vote.user", "user")
-		.andWhere("user.host IS NULL")
-		.getMany();
+	const userIds = [note.userId];
 
-	const userIds = [...new Set([note.userId, ...votes.map((v) => v.userId)])];
+	if (scyllaClient) {
+		const votes = await scyllaClient
+			.execute(prepared.poll.select, [note.id], { prepare: true })
+			.then((result) => result.rows.map(parseScyllaPollVote));
+		const localVotes = votes.filter((vote) => !vote.userHost);
+		userIds.push(...localVotes.map(({ userId }) => userId));
+	} else {
+		const votes = await PollVotes.createQueryBuilder("vote")
+			.select("vote.userId")
+			.where("vote.noteId = :noteId", { noteId: note.id })
+			.innerJoinAndSelect("vote.user", "user")
+			.andWhere("user.host IS NULL")
+			.getMany();
 
-	for (const userId of userIds) {
+		userIds.push(...votes.map((v) => v.userId));
+	}
+
+	for (const userId of new Set(userIds)) {
 		createNotification(userId, "pollEnded", {
 			noteId: note.id,
 		});

packages/backend/src/remote/activitypub/db-resolver.ts

@@ -18,6 +18,7 @@ import { uriPersonCache, userByIdCache } from "@/services/user-cache.js";
 import type { IObject } from "./type.js";
 import { getApId } from "./type.js";
 import { resolvePerson } from "./models/person.js";
+import { parseScyllaNote, prepared, scyllaClient } from "@/db/scylla.js";
 
 const publicKeyCache = new Cache<UserPublickey | null>("publicKey", 60 * 30);
 const publicKeyByUserIdCache = new Cache<UserPublickey | null>(
@@ -78,10 +79,40 @@ export default class DbResolver {
 		if (parsed.local) {
 			if (parsed.type !== "notes") return null;
 
+			if (scyllaClient) {
+				const result = await scyllaClient.execute(
+					prepared.note.select.byId,
+					[parsed.id],
+					{ prepare: true },
+				);
+				if (result.rowLength > 0) {
+					return parseScyllaNote(result.first());
+				}
+				return null;
+			}
+
 			return await Notes.findOneBy({
 				id: parsed.id,
 			});
 		} else {
+			if (scyllaClient) {
+				let result = await scyllaClient.execute(
+					prepared.note.select.byUri,
+					[parsed.uri],
+					{ prepare: true },
+				);
+				if (result.rowLength === 0) {
+					result = await scyllaClient.execute(
+						prepared.note.select.byUrl,
+						[parsed.uri],
+						{ prepare: true },
+					);
+				}
+				if (result.rowLength > 0) {
+					return parseScyllaNote(result.first());
+				}
+				return null;
+			}
 			return await Notes.findOne({
 				where: [
 					{

packages/backend/src/remote/activitypub/renderer/like.ts

@@ -6,7 +6,10 @@ import { Emojis } from "@/models/index.js";
 import renderEmoji from "./emoji.js";
 import { fetchMeta } from "@/misc/fetch-meta.js";
 
-export const renderLike = async (noteReaction: NoteReaction, note: Note) => {
+export const renderLike = async (
+	noteReaction: NoteReaction,
+	note: { uri: string | null },
+) => {
 	const reaction = noteReaction.reaction;
 	const meta = await fetchMeta();
 

packages/backend/src/remote/activitypub/renderer/question.ts

@@ -2,25 +2,49 @@ import config from "@/config/index.js";
 import type { User } from "@/models/entities/user.js";
 import type { Note } from "@/models/entities/note.js";
 import type { Poll } from "@/models/entities/poll.js";
+import { type ScyllaPoll, parseScyllaPollVote, prepared, scyllaClient } from "@/db/scylla.js";
 
 export default async function renderQuestion(
 	user: { id: User["id"] },
 	note: Note,
-	poll: Poll,
+	poll: Poll | ScyllaPoll,
 ) {
+	let choices: {
+		type: "Note";
+		name: string;
+		replies: { type: "Collection"; totalItems: number };
+	}[] = [];
+	if (scyllaClient) {
+		const votes = await scyllaClient
+			.execute(prepared.poll.select, [note.id], { prepare: true })
+			.then((result) => result.rows.map(parseScyllaPollVote));
+		choices = Object.entries((poll as ScyllaPoll).choices).map(
+			([index, text]) => ({
+				type: "Note",
+				name: text,
+				replies: {
+					type: "Collection",
+					totalItems: votes.filter((vote) => vote.choice.has(parseInt(index)))
+						.length,
+				},
+			}),
+		);
+	} else {
+		choices = (poll as Poll).choices.map((text, i) => ({
+			type: "Note",
+			name: text,
+			replies: {
+				type: "Collection",
+				totalItems: (poll as Poll).votes[i],
+			},
+		}));
+	}
 	const question = {
 		type: "Question",
 		id: `${config.url}/questions/${note.id}`,
 		actor: `${config.url}/users/${user.id}`,
 		content: note.text || "",
-		[poll.multiple ? "anyOf" : "oneOf"]: poll.choices.map((text, i) => ({
-			name: text,
-			_misskey_votes: poll.votes[i],
-			replies: {
-				type: "Collection",
-				totalItems: poll.votes[i],
-			},
-		})),
+		[poll.multiple ? "anyOf" : "oneOf"]: choices,
 	};
 
 	return question;

packages/backend/src/remote/activitypub/resolver.ts

@@ -24,6 +24,12 @@ import { renderActivity } from "@/remote/activitypub/renderer/index.js";
 import renderFollow from "@/remote/activitypub/renderer/follow.js";
 import { shouldBlockInstance } from "@/misc/should-block-instance.js";
 import { apLogger } from "@/remote/activitypub/logger.js";
+import {
+	parseScyllaNote,
+	parseScyllaReaction,
+	prepared,
+	scyllaClient,
+} from "@/db/scylla.js";
 
 export default class Resolver {
 	private history: Set<string>;
@@ -146,10 +152,26 @@ export default class Resolver {
 
 		switch (parsed.type) {
 			case "notes":
+				if (scyllaClient) {
+					return scyllaClient
+						.execute(prepared.note.select.byId, [parsed.id], { prepare: true })
+						.then((result) => {
+							if (result.rowLength > 0) {
+								const note = parseScyllaNote(result.first());
+								if (parsed.rest === "activity") {
+									// this refers to the create activity and not the note itself
+									return renderActivity(renderCreate(renderNote(note), note));
+								} else {
+									return renderNote(note);
+								}
+							}
+							throw new Error("Note not found");
+						});
+				}
 				return Notes.findOneByOrFail({ id: parsed.id }).then((note) => {
 					if (parsed.rest === "activity") {
 						// this refers to the create activity and not the note itself
-						return renderActivity(renderCreate(renderNote(note)));
+						return renderActivity(renderCreate(renderNote(note), note));
 					} else {
 						return renderNote(note);
 					}
@@ -159,6 +181,19 @@ export default class Resolver {
 					renderPerson(user as ILocalUser),
 				);
 			case "questions":
+				if (scyllaClient) {
+					return scyllaClient
+						.execute(prepared.note.select.byId, [parsed.id], { prepare: true })
+						.then((result) => {
+							if (result.rowLength > 0) {
+								const note = parseScyllaNote(result.first());
+								if (note.hasPoll && note.poll) {
+									return renderQuestion({ id: note.userId }, note, note.poll);
+								}
+							}
+							throw new Error("Question not found");
+						});
+				}
 				// Polls are indexed by the note they are attached to.
 				return Promise.all([
 					Notes.findOneByOrFail({ id: parsed.id }),
@@ -167,6 +202,19 @@ export default class Resolver {
 					renderQuestion({ id: note.userId }, note, poll),
 				);
 			case "likes":
+				if (scyllaClient) {
+					return scyllaClient
+						.execute(prepared.reaction.select.byId, [parsed.id], {
+							prepare: true,
+						})
+						.then((result) => {
+							if (result.rowLength > 0) {
+								const reaction = parseScyllaReaction(result.first());
+								return renderActivity(renderLike(reaction, { uri: null }));
+							}
+							throw new Error("Reaction not found");
+						});
+				}
 				return NoteReactions.findOneByOrFail({ id: parsed.id }).then(
 					(reaction) => renderActivity(renderLike(reaction, { uri: null })),
 				);

packages/backend/src/services/note/polls/vote.ts

@@ -57,7 +57,7 @@ export default async function (
 
 		await scyllaClient.execute(
 			prepared.poll.insert,
-			[scyllaNote.id, user.id, Array.from(newChoice), new Date()],
+			[scyllaNote.id, user.id, user.host, Array.from(newChoice), new Date()],
 			{ prepare: true },
 		);
 	} else {