add poll type and table

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

@@ -1,3 +1,4 @@
+DROP TABLE IF EXISTS poll_vote;
 DROP MATERIALIZED VIEW IF EXISTS reaction_by_id;
 DROP MATERIALIZED VIEW IF EXISTS reaction_by_user_id;
 DROP INDEX IF EXISTS reaction_by_id;
@@ -15,6 +16,7 @@ DROP INDEX IF EXISTS note_by_reply_id;
 DROP INDEX IF EXISTS note_by_uri;
 DROP INDEX IF EXISTS note_by_url;
 DROP TABLE IF EXISTS note;
+DROP TYPE IF EXISTS poll;
 DROP TYPE IF EXISTS emoji;
 DROP TYPE IF EXISTS note_edit_history;
 DROP TYPE IF EXISTS drive_file;

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

@@ -29,6 +29,12 @@ CREATE TYPE IF NOT EXISTS emoji (
 	"height" int,
 );
 
+CREATE TYPE IF NOT EXISTS poll (
+	"expiresAt" timestamp,
+	"multiple" boolean,
+	"choices" map<int, text>,
+);
+
 CREATE TABLE IF NOT EXISTS note ( -- Store all posts
 	"createdAtDate" date, -- For partitioning
 	"createdAt" timestamp,
@@ -50,6 +56,7 @@ CREATE TABLE IF NOT EXISTS note ( -- Store all posts
 	"emojis" set<text>,
 	"tags" set<text>,
 	"hasPoll" boolean,
+	"poll" poll,
 	"threadId" ascii,
 	"channelId" ascii, -- Channel
 	"userId" ascii, -- User
@@ -172,6 +179,7 @@ CREATE TABLE IF NOT EXISTS home_timeline (
 	"emojis" set<text>,
 	"tags" set<text>,
 	"hasPoll" boolean,
+	"poll" poll,
 	"threadId" ascii,
 	"channelId" ascii, -- Channel
 	"userId" ascii, -- User
@@ -220,3 +228,11 @@ CREATE MATERIALIZED VIEW IF NOT EXISTS reaction_by_id AS
 	AND "reaction" IS NOT NULL
 	AND "userId" IS NOT NULL
 	PRIMARY KEY ("noteId", "reaction", "userId");
+
+CREATE TABLE IF NOT EXISTS poll_vote (
+	"noteId" ascii,
+	"userId" ascii,
+	"choice" set<int>,
+	"createdAt" timestamp,
+	PRIMARY KEY ("noteId", "userId")
+);

packages/backend/src/db/cql.ts

@@ -21,6 +21,7 @@ export const scyllaQueries = {
 				"emojis",
 				"tags",
 				"hasPoll",
+				"poll",
 				"threadId",
 				"channelId",
 				"userId",
@@ -40,7 +41,7 @@ export const scyllaQueries = {
 				"reactions",
 				"noteEdit",
 				"updatedAt"
-			) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+			) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
 		select: {
 			byDate: `SELECT * FROM note WHERE "createdAtDate" = ?`,
 			byUri: `SELECT * FROM note WHERE "uri" = ?`,
@@ -91,6 +92,7 @@ export const scyllaQueries = {
 				"emojis",
 				"tags",
 				"hasPoll",
+				"poll",
 				"threadId",
 				"channelId",
 				"userId",
@@ -110,7 +112,7 @@ export const scyllaQueries = {
 				"reactions",
 				"noteEdit",
 				"updatedAt"
-			) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+			) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
 		select: {
 			byUserAndDate: `SELECT * FROM home_timeline WHERE "feedUserId" = ? AND "createdAtDate" = ?`,
 			byId: `SELECT * FROM home_timeline WHERE "id" = ?`,
@@ -153,4 +155,8 @@ export const scyllaQueries = {
 		},
 		delete: `DELETE FROM reaction WHERE "noteId" = ? AND "userId" = ?`,
 	},
+	poll: {
+		select: `SELECT * FROM poll_vote WHERE "noteId" = ?`,
+		insert: `INSERT INTO poll_vote ("noteId", "userId", "choice", "createdAt") VALUES (?, ?, ?, ?)`,
+	},
 };

packages/backend/src/db/scylla.ts

@@ -111,6 +111,28 @@ export interface ScyllaNoteEditHistory {
 	updatedAt: Date;
 }
 
+export interface ScyllaPoll {
+	expiresAt: Date | null;
+	multiple: boolean;
+	choices: Map<number, string>,
+}
+
+export interface ScyllaPollVote {
+	noteId: string,
+	userId: string,
+	choice: Set<number>,
+	createdAt: Date,
+}
+
+export function parseScyllaPollVote(row: types.Row): ScyllaPollVote {
+	return {
+		noteId: row.get("noteId"),
+		userId: row.get("userId"),
+		choice: row.get("choice"),
+		createdAt: row.get("createdAt"),
+	}
+}
+
 export type ScyllaNote = Note & {
 	createdAtDate: Date;
 	files: ScyllaDriveFile[];
@@ -121,6 +143,7 @@ export type ScyllaNote = Note & {
 	renoteText: string | null;
 	renoteCw: string | null;
 	renoteFiles: ScyllaDriveFile[];
+	poll: ScyllaPoll | null;
 };
 
 export function parseScyllaNote(row: types.Row): ScyllaNote {
@@ -149,6 +172,7 @@ export function parseScyllaNote(row: types.Row): ScyllaNote {
 		emojis: row.get("emojis") ?? [],
 		tags: row.get("tags") ?? [],
 		hasPoll: row.get("hasPoll") ?? false,
+		poll: row.get("poll") ?? null,
 		threadId: row.get("threadId") ?? null,
 		channelId: row.get("channelId") ?? null,
 		userId: row.get("userId"),

packages/backend/src/remote/activitypub/models/note.ts

@@ -53,6 +53,7 @@ import { DB_MAX_IMAGE_COMMENT_LENGTH } from "@/misc/hard-limits.js";
 import { truncate } from "@/misc/truncate.js";
 import { type Size, getEmojiSize } from "@/misc/emoji-meta.js";
 import { fetchMeta } from "@/misc/fetch-meta.js";
+import { parseScyllaNote, prepared, scyllaClient } from "@/db/scylla.js";
 
 const logger = apLogger;
 
@@ -317,7 +318,37 @@ export async function createNote(
 	}
 
 	// vote
-	if (reply?.hasPoll) {
+	if (reply?.hasPoll && note.name) {
+		if (scyllaClient) {
+			const result = await scyllaClient.execute(
+				prepared.note.select.byId,
+				[reply.id],
+				{ prepare: true },
+			);
+			if (result.rowLength === 0) {
+				throw new Error("reply target note not found");
+			}
+			const scyllaNote = parseScyllaNote(result.first());
+			if (!scyllaNote.hasPoll || !scyllaNote.poll) {
+				throw new Error("reply target does not have poll");
+			}
+			if (scyllaNote.poll.expiresAt && scyllaNote.poll.expiresAt < new Date()) {
+				logger.warn(
+					`vote to expired poll from AP: actor=${actor.username}@${actor.host}, note=${scyllaNote.id}, choice=${note.name}`,
+				);
+				return null;
+			}
+
+			const entry = Array.from(scyllaNote.poll.choices.entries()).find(
+				([_, v]) => v === note.name,
+			);
+			if (entry) {
+				await vote(actor, scyllaNote, entry[0]);
+			}
+
+			return null;
+		}
+
 		const poll = await Polls.findOneByOrFail({ noteId: reply.id });
 
 		const tryCreateVote = async (
@@ -337,13 +368,11 @@ export async function createNote(
 			return null;
 		};
 
-		if (note.name) {
 		return await tryCreateVote(
 			note.name,
 			poll.choices.findIndex((x) => x === note.name),
 		);
 	}
-	}
 
 	const emojis = await extractEmojis(note.tag || [], actor.host).catch((e) => {
 		logger.info(`extractEmojis: ${e}`);

packages/backend/src/services/note/create.ts

@@ -73,6 +73,7 @@ import {
 	parseScyllaNote,
 	prepared,
 	scyllaClient,
+	ScyllaPoll,
 } from "@/db/scylla.js";
 
 export const mutedWordsCache = new Cache<
@@ -703,13 +704,13 @@ async function incRenoteCount(renote: Note) {
 			],
 			{ prepare: true },
 		);
-		const homeTimelines = await scyllaClient
+		scyllaClient.eachRow(
-			.execute(prepared.homeTimeline.select.byId, [renote.id], {
+			prepared.homeTimeline.select.byId,
-				prepare: true,
+			[renote.id],
-			})
+			{ prepare: true },
-			.then((result) => result.rows.map(parseHomeTimeline));
+			(_, row) => {
-		// Do not issue BATCH because different home timelines involve different partitions
+				if (scyllaClient) {
-		for (const timeline of homeTimelines) {
+					const timeline = parseHomeTimeline(row);
 					scyllaClient.execute(
 						prepared.homeTimeline.update.renoteCount,
 						[
@@ -723,6 +724,8 @@ async function incRenoteCount(renote: Note) {
 						{ prepare: true },
 					);
 				}
+			},
+		);
 	} else {
 		Notes.createQueryBuilder()
 			.update()
@@ -806,7 +809,7 @@ async function insertNote(
 		);
 	}
 
-	// 投稿を作成
+	// Insert post to DB
 	try {
 		if (scyllaClient) {
 			const fileMapper = (file: DriveFile) => ({
@@ -830,6 +833,24 @@ async function insertNote(
 				  )
 				: null;
 
+			let poll: ScyllaPoll | null = null;
+
+			if (data.poll) {
+				insert.hasPoll = true;
+				let expiresAt: Date | null;
+				if (!data.poll.expiresAt || isNaN(data.poll.expiresAt.getTime())) {
+					expiresAt = null;
+				} else {
+					expiresAt = data.poll.expiresAt;
+				}
+
+				poll = {
+					expiresAt,
+					choices: new Map(data.poll.choices.map((v, i) => [i, v] as [number, string])),
+					multiple: data.poll.multiple,
+				};
+			}
+
 			const params = [
 				insert.createdAt,
 				insert.createdAt,
@@ -851,6 +872,7 @@ async function insertNote(
 				insert.emojis,
 				insert.tags,
 				insert.hasPoll,
+				poll,
 				insert.threadId,
 				insert.channelId,
 				insert.userId,
@@ -895,15 +917,13 @@ async function insertNote(
 					},
 				);
 			}
-		}
+		} else {
 			if (insert.hasPoll) {
 				// Start transaction
 				await db.transaction(async (transactionalEntityManager) => {
 					if (!data.poll) throw new Error("Empty poll data");
 
-				if (!scyllaClient) {
 					await transactionalEntityManager.insert(Note, insert);
-				}
 
 					let expiresAt: Date | null;
 					if (!data.poll.expiresAt || isNaN(data.poll.expiresAt.getTime())) {
@@ -925,9 +945,10 @@ async function insertNote(
 
 					await transactionalEntityManager.insert(Poll, poll);
 				});
-		} else if (!scyllaClient) {
+			} else {
 				await Notes.insert(insert);
 			}
+		}
 
 		if (scyllaClient) {
 			const result = await scyllaClient.execute(
@@ -1074,13 +1095,13 @@ async function saveReply(reply: Note) {
 			],
 			{ prepare: true },
 		);
-		const homeTimelines = await scyllaClient
+		scyllaClient.eachRow(
-			.execute(prepared.homeTimeline.select.byId, [reply.id], {
+			prepared.homeTimeline.select.byId,
-				prepare: true,
+			[reply.id],
-			})
+			{ prepare: true },
-			.then((result) => result.rows.map(parseHomeTimeline));
+			(_, row) => {
-		// Do not issue BATCH because different home timelines involve different partitions
+				if (scyllaClient) {
-		for (const timeline of homeTimelines) {
+					const timeline = parseHomeTimeline(row);
 					scyllaClient.execute(
 						prepared.homeTimeline.update.repliesCount,
 						[
@@ -1093,6 +1114,8 @@ async function saveReply(reply: Note) {
 						{ prepare: true },
 					);
 				}
+			},
+		);
 	} else {
 		await Notes.increment({ id: reply.id }, "repliesCount", 1);
 	}

packages/backend/src/services/note/delete.ts

@@ -70,13 +70,15 @@ export default async function (
 					],
 					{ prepare: true },
 				);
-				const homeTimelines = await scyllaClient
+				scyllaClient.eachRow(
-					.execute(prepared.homeTimeline.select.byId, [renote.id], {
+					prepared.homeTimeline.select.byId,
+					[renote.id],
+					{
 						prepare: true,
-					})
+					},
-					.then((result) => result.rows.map(parseHomeTimeline));
+					(_, row) => {
-				// Do not issue BATCH because different home timelines involve different partitions
+						if (scyllaClient) {
-				for (const timeline of homeTimelines) {
+							const timeline = parseHomeTimeline(row);
 							scyllaClient.execute(
 								prepared.homeTimeline.update.renoteCount,
 								[
@@ -90,6 +92,8 @@ export default async function (
 								{ prepare: true },
 							);
 						}
+					},
+				);
 			}
 		} else {
 			Notes.decrement({ id: note.renoteId }, "renoteCount", 1);
@@ -119,13 +123,15 @@ export default async function (
 					],
 					{ prepare: true },
 				);
-				const homeTimelines = await scyllaClient
+				scyllaClient.eachRow(
-					.execute(prepared.homeTimeline.select.byId, [reply.id], {
+					prepared.homeTimeline.select.byId,
+					[reply.id],
+					{
 						prepare: true,
-					})
+					},
-					.then((result) => result.rows.map(parseHomeTimeline));
+					(_, row) => {
-				// Do not issue BATCH because different home timelines involve different partitions
+						if (scyllaClient) {
-				for (const timeline of homeTimelines) {
+							const timeline = parseHomeTimeline(row);
 							scyllaClient.execute(
 								prepared.homeTimeline.update.repliesCount,
 								[
@@ -138,6 +144,8 @@ export default async function (
 								{ prepare: true },
 							);
 						}
+					},
+				);
 			}
 		} else {
 			await Notes.decrement({ id: note.replyId }, "repliesCount", 1);

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

@@ -1,20 +1,68 @@
 import { publishNoteStream } from "@/services/stream.js";
 import type { CacheableUser } from "@/models/entities/user.js";
-import { User } from "@/models/entities/user.js";
 import type { Note } from "@/models/entities/note.js";
 import { PollVotes, NoteWatchings, Polls, Blockings } from "@/models/index.js";
 import { Not } from "typeorm";
 import { genId } from "@/misc/gen-id.js";
 import { createNotification } from "../../create-notification.js";
+import {
+	type ScyllaNote,
+	type ScyllaPollVote,
+	scyllaClient,
+	prepared,
+	parseScyllaPollVote,
+} from "@/db/scylla.js";
+import { UserBlockingCache } from "@/misc/cache.js";
 
 export default async function (
 	user: CacheableUser,
-	note: Note,
+	note: Note | ScyllaNote,
 	choice: number,
 ) {
-	const poll = await Polls.findOneBy({ noteId: note.id });
+	if (scyllaClient) {
+		const scyllaNote = note as ScyllaNote;
+		if (!scyllaNote.hasPoll || !scyllaNote.poll) {
+			throw new Error("poll not found");
+		}
 
-	if (poll == null) throw new Error("poll not found");
+		if (!Array.from(scyllaNote.poll.choices.keys()).includes(choice)) {
+			throw new Error("invalid choice param");
+		}
+
+		if (scyllaNote.userId !== user.id) {
+			const isBlocking = await UserBlockingCache.init(scyllaNote.userId).then(
+				(cache) => cache.has(user.id),
+			);
+			if (isBlocking) {
+				throw new Error("blocked");
+			}
+		}
+
+		let newChoice: ScyllaPollVote["choice"] = new Set();
+		const result = await scyllaClient.execute(
+			`${prepared.poll.select} AND "userId" = ?`,
+			[scyllaNote.id, user.id],
+			{ prepare: true },
+		);
+		if (result.rowLength > 0) {
+			const vote = parseScyllaPollVote(result.first());
+			if (scyllaNote.poll.multiple && !vote.choice.has(choice)) {
+				newChoice = vote.choice.add(choice);
+			} else {
+				throw new Error("already voted");
+			}
+		} else {
+			newChoice.add(choice);
+		}
+
+		await scyllaClient.execute(
+			prepared.poll.insert,
+			[scyllaNote.id, user.id, newChoice, new Date()],
+			{ prepare: true },
+		);
+	} else {
+		const poll = await Polls.findOneBy({ noteId: note.id });
+		if (!poll) throw new Error("poll not found");
 
 		// Check whether is valid choice
 		if (poll.choices[choice] == null) throw new Error("invalid choice param");
@@ -58,6 +106,7 @@ export default async function (
 		await Polls.query(
 			`UPDATE poll SET votes[${index}] = votes[${index}] + 1 WHERE "noteId" = '${poll.noteId}'`,
 		);
+	}
 
 	publishNoteStream(note.id, "pollVoted", {
 		choice: choice,

packages/backend/src/services/note/reaction/create.ts

@@ -123,13 +123,15 @@ export default async (
 			],
 			{ prepare: true },
 		);
-		const homeTimelines = await scyllaClient
+		scyllaClient.eachRow(
-			.execute(prepared.homeTimeline.select.byId, [note.id], {
+			prepared.homeTimeline.select.byId,
+			[note.id],
+			{
 				prepare: true,
-			})
+			},
-			.then((result) => result.rows.map(parseHomeTimeline));
+			(_, row) => {
-		// Do not issue BATCH because different home timelines involve different partitions
+				if (scyllaClient) {
-		for (const timeline of homeTimelines) {
+					const timeline = parseHomeTimeline(row);
 					scyllaClient.execute(
 						prepared.homeTimeline.update.reactions,
 						[
@@ -144,6 +146,8 @@ export default async (
 						{ prepare: true },
 					);
 				}
+			},
+		);
 	} else {
 		const sql = `jsonb_set("reactions", '{${_reaction}}', (COALESCE("reactions"->>'${_reaction}', '0')::int + 1)::text::jsonb)`;
 		await Notes.createQueryBuilder()

packages/backend/src/services/note/reaction/delete.ts

@@ -79,13 +79,15 @@ export default async (
 			],
 			{ prepare: true },
 		);
-		const homeTimelines = await scyllaClient
+		scyllaClient.eachRow(
-			.execute(prepared.homeTimeline.select.byId, [note.id], {
+			prepared.homeTimeline.select.byId,
+			[note.id],
+			{
 				prepare: true,
-			})
+			},
-			.then((result) => result.rows.map(parseHomeTimeline));
+			(_, row) => {
-		// Do not issue BATCH because different home timelines involve different partitions
+				if (scyllaClient) {
-		for (const timeline of homeTimelines) {
+					const timeline = parseHomeTimeline(row);
 					scyllaClient.execute(
 						prepared.homeTimeline.update.reactions,
 						[
@@ -100,6 +102,8 @@ export default async (
 						{ prepare: true },
 					);
 				}
+			},
+		);
 	} else {
 		const sql = `jsonb_set("reactions", '{${reaction.reaction}}', (COALESCE("reactions"->>'${reaction.reaction}', '0')::int - 1)::text::jsonb)`;
 		await Notes.createQueryBuilder()