Merge branch 'fix/pagination' into 'develop'

Draft: fix(backend): requested limit to be fulfilled if possible

Closes #10867

See merge request firefish/firefish!10696

.gitlab-ci.yml

@@ -3,8 +3,10 @@ image: docker.io/rust:slim-bookworm
 services:
   - name: docker.io/groonga/pgroonga:latest-alpine-12-slim
     alias: postgres
+    pull_policy: if-not-present
   - name: docker.io/redis:7-alpine
     alias: redis
+    pull_policy: if-not-present
 
 workflow:
   rules:
@@ -84,10 +86,7 @@ client_build_test:
           - packages/client/*
           - packages/firefish-js/*
           - packages/sw/*
-          - scripts/**/*
           - locales/**/*
-          - package.json
-          - pnpm-lock.yaml
     - if: $CI_PIPELINE_SOURCE == 'push' || $CI_PIPELINE_SOURCE == 'merge_request_event'
       changes:
         paths:
@@ -98,6 +97,15 @@ client_build_test:
           - Cargo.toml
           - Cargo.lock
       when: never
+  services: []
+  before_script:
+    - apt-get update && apt-get -y upgrade
+    - apt-get -y --no-install-recommends install curl
+    - curl -fsSL 'https://deb.nodesource.com/setup_18.x' | bash -
+    - apt-get install -y --no-install-recommends build-essential python3 perl nodejs
+    - corepack enable
+    - corepack prepare pnpm@latest --activate
+    - cp .config/ci.yml .config/default.yml
   script:
     - pnpm install --frozen-lockfile
     - pnpm --filter 'firefish-js' --filter 'client' --filter 'sw' run build:debug
@@ -175,8 +183,12 @@ cargo_clippy:
           - Cargo.lock
     - if: $CI_MERGE_REQUEST_TARGET_BRANCH_NAME == 'main'
       when: never
-  script:
+  services: []
+  before_script:
+    - apt-get install -y --no-install-recommends build-essential clang mold perl
+    - cp ci/cargo/config.toml /usr/local/cargo/config.toml
     - rustup component add clippy
+  script:
     - cargo clippy -- -D warnings
 
 renovate:

packages/backend-rs/index.d.ts

@@ -1292,7 +1292,6 @@ export interface AbuseUserReportLike {
   comment: string
 }
 export function publishToModerationStream(moderatorId: string, report: AbuseUserReportLike): void
-export function publishToNotesStream(note: Note): void
 export function getTimestamp(id: string): number
 /**
  * The generated ID results in the form of `[8 chars timestamp] + [cuid2]`.

packages/backend-rs/index.js

@@ -310,7 +310,7 @@ if (!nativeBinding) {
   throw new Error(`Failed to load native binding`)
 }
 
-const { SECOND, MINUTE, HOUR, DAY, USER_ONLINE_THRESHOLD, USER_ACTIVE_THRESHOLD, FILE_TYPE_BROWSERSAFE, loadEnv, loadConfig, stringToAcct, acctToString, addNoteToAntenna, isBlockedServer, isSilencedServer, isAllowedServer, checkWordMute, getFullApAccount, isSelfHost, isSameOrigin, extractHost, toPuny, isUnicodeEmoji, sqlLikeEscape, safeForSql, formatMilliseconds, getImageSizeFromUrl, getNoteSummary, isSafeUrl, latestVersion, toMastodonId, fromMastodonId, fetchMeta, metaToPugArgs, nyaify, hashPassword, verifyPassword, isOldPasswordAlgorithm, decodeReaction, countReactions, toDbReaction, removeOldAttestationChallenges, AntennaSrcEnum, DriveFileUsageHintEnum, MutedNoteReasonEnum, NoteVisibilityEnum, NotificationTypeEnum, PageVisibilityEnum, PollNotevisibilityEnum, RelayStatusEnum, UserEmojimodpermEnum, UserProfileFfvisibilityEnum, UserProfileMutingnotificationtypesEnum, initializeRustLogger, fetchNodeinfo, nodeinfo_2_1, nodeinfo_2_0, Protocol, Inbound, Outbound, watchNote, unwatchNote, publishToChannelStream, ChatEvent, publishToChatStream, ChatIndexEvent, publishToChatIndexStream, publishToBroadcastStream, publishToGroupChatStream, publishToModerationStream, publishToNotesStream, getTimestamp, genId, genIdAt, secureRndstr } = nativeBinding
+const { SECOND, MINUTE, HOUR, DAY, USER_ONLINE_THRESHOLD, USER_ACTIVE_THRESHOLD, FILE_TYPE_BROWSERSAFE, loadEnv, loadConfig, stringToAcct, acctToString, addNoteToAntenna, isBlockedServer, isSilencedServer, isAllowedServer, checkWordMute, getFullApAccount, isSelfHost, isSameOrigin, extractHost, toPuny, isUnicodeEmoji, sqlLikeEscape, safeForSql, formatMilliseconds, getImageSizeFromUrl, getNoteSummary, isSafeUrl, latestVersion, toMastodonId, fromMastodonId, fetchMeta, metaToPugArgs, nyaify, hashPassword, verifyPassword, isOldPasswordAlgorithm, decodeReaction, countReactions, toDbReaction, removeOldAttestationChallenges, AntennaSrcEnum, DriveFileUsageHintEnum, MutedNoteReasonEnum, NoteVisibilityEnum, NotificationTypeEnum, PageVisibilityEnum, PollNotevisibilityEnum, RelayStatusEnum, UserEmojimodpermEnum, UserProfileFfvisibilityEnum, UserProfileMutingnotificationtypesEnum, initializeRustLogger, fetchNodeinfo, nodeinfo_2_1, nodeinfo_2_0, Protocol, Inbound, Outbound, watchNote, unwatchNote, publishToChannelStream, ChatEvent, publishToChatStream, ChatIndexEvent, publishToChatIndexStream, publishToBroadcastStream, publishToGroupChatStream, publishToModerationStream, getTimestamp, genId, genIdAt, secureRndstr } = nativeBinding
 
 module.exports.SECOND = SECOND
 module.exports.MINUTE = MINUTE
@@ -381,7 +381,6 @@ module.exports.publishToChatIndexStream = publishToChatIndexStream
 module.exports.publishToBroadcastStream = publishToBroadcastStream
 module.exports.publishToGroupChatStream = publishToGroupChatStream
 module.exports.publishToModerationStream = publishToModerationStream
-module.exports.publishToNotesStream = publishToNotesStream
 module.exports.getTimestamp = getTimestamp
 module.exports.genId = genId
 module.exports.genIdAt = genIdAt

packages/backend-rs/src/service/stream.rs

@@ -5,7 +5,6 @@ pub mod chat_index;
 pub mod custom_emoji;
 pub mod group_chat;
 pub mod moderation;
-pub mod new_note;
 
 use crate::config::CONFIG;
 use crate::database::redis_conn;
@@ -26,7 +25,7 @@ pub enum Stream {
     #[strum(to_string = "noteStream:{note_id}")]
     Note { note_id: String },
     #[strum(serialize = "notesStream")]
-    NewNote,
+    Notes,
     #[strum(to_string = "userListStream:{list_id}")]
     UserList { list_id: String },
     #[strum(to_string = "mainStream:{user_id}")]

packages/backend-rs/src/service/stream/new_note.rs

@@ -1,10 +0,0 @@
-use crate::model::entity::note;
-use crate::service::stream::{publish_to_stream, Error, Stream};
-
-// for napi export (https://github.com/napi-rs/napi-rs/issues/2060)
-type Note = note::Model;
-
-#[crate::export(js_name = "publishToNotesStream")]
-pub fn publish(note: &Note) -> Result<(), Error> {
-    publish_to_stream(&Stream::NewNote, None, Some(serde_json::to_string(note)?))
-}

packages/backend/src/models/repositories/note-favorite.ts

@@ -2,6 +2,7 @@ import { db } from "@/db/postgre.js";
 import { NoteFavorite } from "@/models/entities/note-favorite.js";
 import { Notes } from "../index.js";
 import type { User } from "@/models/entities/user.js";
+import Logger from "@/services/logger.js";
 
 export const NoteFavoriteRepository = db.getRepository(NoteFavorite).extend({
 	async pack(
@@ -23,9 +24,16 @@ export const NoteFavoriteRepository = db.getRepository(NoteFavorite).extend({
 	packMany(favorites: any[], me: { id: User["id"] }) {
 		return Promise.allSettled(favorites.map((x) => this.pack(x, me))).then(
 			(promises) =>
-				promises.flatMap((result) =>
-					result.status === "fulfilled" ? [result.value] : [],
-				),
+				promises.flatMap((result, i) => {
+					if (result.status === "fulfilled") {
+						return [result.value];
+					}
+					const logger = new Logger("models-note-favorite");
+					logger.error(
+						`dropping note favorite due to violating visibility restrictions, note favorite ${favorites[i].id} user ${me.id}`,
+					);
+					return [];
+				}),
 		);
 	},
 });

packages/backend/src/models/repositories/note-reaction.ts

@@ -4,6 +4,7 @@ import { Notes, Users } from "../index.js";
 import type { Packed } from "@/misc/schema.js";
 import { decodeReaction } from "backend-rs";
 import type { User } from "@/models/entities/user.js";
+import Logger from "@/services/logger.js";
 
 export const NoteReactionRepository = db.getRepository(NoteReaction).extend({
 	async pack(
@@ -49,8 +50,15 @@ export const NoteReactionRepository = db.getRepository(NoteReaction).extend({
 		);
 
 		// filter out rejected promises, only keep fulfilled values
-		return reactions.flatMap((result) =>
-			result.status === "fulfilled" ? [result.value] : [],
-		);
+		return reactions.flatMap((result, i) => {
+			if (result.status === "fulfilled") {
+				return [result.value];
+			}
+			const logger = new Logger("models-note-reaction");
+			logger.error(
+				`dropping note reaction due to violating visibility restrictions, reason is ${result.reason}`,
+			);
+			return [];
+		});
 	},
 });

packages/backend/src/models/repositories/note.ts

@@ -23,6 +23,7 @@ import {
 } from "@/misc/populate-emojis.js";
 import { db } from "@/db/postgre.js";
 import { IdentifiableError } from "@/misc/identifiable-error.js";
+import Logger from "@/services/logger.js";
 
 export async function populatePoll(note: Note, meId: User["id"] | null) {
 	const poll = await Polls.findOneByOrFail({ noteId: note.id });
@@ -343,8 +344,15 @@ export const NoteRepository = db.getRepository(Note).extend({
 		);
 
 		// filter out rejected promises, only keep fulfilled values
-		return promises.flatMap((result) =>
-			result.status === "fulfilled" ? [result.value] : [],
-		);
+		return promises.flatMap((result, i) => {
+			if (result.status === "fulfilled") {
+				return [result.value];
+			}
+			const logger = new Logger("models-note");
+			logger.error(
+				`dropping note due to violating visibility restrictions, note ${notes[i].id} user ${meId}`,
+			);
+			return [];
+		});
 	},
 });

packages/backend/src/server/activitypub/outbox.ts

@@ -73,7 +73,21 @@ export default async (ctx: Router.RouterContext) => {
 			)
 			.andWhere("note.localOnly = FALSE");
 
-		const notes = await query.take(limit).getMany();
+		// We fetch more than requested because some may be filtered out, and if there's less than
+		// requested, this is not normal behavior of any API.
+		const notes = [];
+		const take = Math.floor(limit * 1.5);
+		let skip = 0;
+		while (notes.length < limit) {
+			const notes_query = await query.take(take).skip(skip).getMany();
+			notes.push(...(await Notes.packMany(notes_query)));
+			skip += take;
+			if (notes_query.length < take) break;
+		}
+
+		if (notes.length > limit) {
+			notes.length = limit;
+		}
 
 		if (sinceId) notes.reverse();
 

packages/backend/src/server/api/endpoints/antennas/notes.ts

@@ -117,11 +117,25 @@ export default define(meta, paramDef, async (ps, user) => {
 	generateMutedUserQuery(query, user);
 	generateBlockedUserQuery(query, user);
 
-	const notes = await query.take(limit).getMany();
-
-	if (notes.length > 0) {
-		readNote(user.id, notes);
+	// We fetch more than requested because some may be filtered out, and if there's less than
+	// requested, the pagination stops.
+	const found = [];
+	const take = Math.floor(ps.limit * 1.5);
+	let skip = 0;
+	while (found.length < ps.limit) {
+		const notes = await query.take(take).skip(skip).getMany();
+		found.push(...(await Notes.packMany(notes)));
+		skip += take;
+		if (notes.length < take) break;
 	}
 
-	return await Notes.packMany(notes, user);
+	if (found.length > ps.limit) {
+		found.length = ps.limit;
+	}
+
+	if (found.length > 0) {
+		readNote(user.id, found);
+	}
+
+	return found;
 });

packages/backend/src/server/api/endpoints/channels/timeline.ts

@@ -76,9 +76,23 @@ export default define(meta, paramDef, async (ps, user) => {
 		.leftJoinAndSelect("note.channel", "channel");
 	//#endregion
 
-	const timeline = await query.take(ps.limit).getMany();
+	// We fetch more than requested because some may be filtered out, and if there's less than
+	// requested, the pagination stops.
+	const found = [];
+	const take = Math.floor(ps.limit * 1.5);
+	let skip = 0;
+	while (found.length < ps.limit) {
+		const timeline = await query.take(take).skip(skip).getMany();
+		found.push(...(await Notes.packMany(timeline, user)));
+		skip += take;
+		if (timeline.length < take) break;
+	}
+
+	if (found.length > ps.limit) {
+		found.length = ps.limit;
+	}
 
 	if (user) activeUsersChart.read(user);
 
-	return await Notes.packMany(timeline, user);
+	return found;
 });

packages/backend/src/server/api/endpoints/clips/notes.ts

@@ -88,7 +88,21 @@ export default define(meta, paramDef, async (ps, user) => {
 		generateBlockedUserQuery(query, user);
 	}
 
-	const notes = await query.take(ps.limit).getMany();
+	// We fetch more than requested because some may be filtered out, and if there's less than
+	// requested, the pagination stops.
+	const found = [];
+	const take = Math.floor(ps.limit * 1.5);
+	let skip = 0;
+	while (found.length < ps.limit) {
+		const notes = await query.take(take).skip(skip).getMany();
+		found.push(...(await Notes.packMany(notes, user)));
+		skip += take;
+		if (notes.length < take) break;
+	}
 
-	return await Notes.packMany(notes, user);
+	if (found.length > ps.limit) {
+		found.length = ps.limit;
+	}
+
+	return found;
 });

packages/backend/src/server/api/endpoints/notes.ts

@@ -85,7 +85,21 @@ export default define(meta, paramDef, async (ps) => {
 	//	query.isBot = bot;
 	//}
 
-	const notes = await query.take(ps.limit).getMany();
+	// We fetch more than requested because some may be filtered out, and if there's less than
+	// requested, the pagination stops.
+	const found = [];
+	const take = Math.floor(ps.limit * 1.5);
+	let skip = 0;
+	while (found.length < ps.limit) {
+		const notes = await query.take(take).skip(skip).getMany();
+		found.push(...(await Notes.packMany(notes)));
+		skip += take;
+		if (notes.length < take) break;
+	}
 
-	return await Notes.packMany(notes);
+	if (found.length > ps.limit) {
+		found.length = ps.limit;
+	}
+
+	return found;
 });

packages/backend/src/server/api/endpoints/notes/children.ts

@@ -57,7 +57,25 @@ export default define(meta, paramDef, async (ps, user) => {
 		generateBlockedUserQuery(query, user);
 	}
 
-	const notes = await query.getMany();
+	// We fetch more than requested because some may be filtered out, and if there's less than
+	// requested, the pagination stops.
+	const found = [];
+	const take = Math.floor(ps.limit * 1.5);
+	let skip = 0;
+	while (found.length < ps.limit) {
+		const notes = await query.take(take).skip(skip).getMany();
+		found.push(
+			...(await Notes.packMany(notes, user, {
+				detail: false,
+			})),
+		);
+		skip += take;
+		if (notes.length < take) break;
+	}
 
-	return await Notes.packMany(notes, user, { detail: false });
+	if (found.length > ps.limit) {
+		found.length = ps.limit;
+	}
+
+	return found;
 });

packages/backend/src/server/api/endpoints/users/notes.ts

@@ -138,7 +138,21 @@ export default define(meta, paramDef, async (ps, me) => {
 
 	//#endregion
 
-	const timeline = await query.take(ps.limit).getMany();
+	// We fetch more than requested because some may be filtered out, and if there's less than
+	// requested, the pagination stops.
+	const found = [];
+	const take = Math.floor(ps.limit * 1.5);
+	let skip = 0;
+	while (found.length < ps.limit) {
+		const timeline = await query.take(take).skip(skip).getMany();
+		found.push(...(await Notes.packMany(timeline, user)));
+		skip += take;
+		if (timeline.length < take) break;
+	}
 
-	return await Notes.packMany(timeline, me);
+	if (found.length > ps.limit) {
+		found.length = ps.limit;
+	}
+
+	return found;
 });

packages/backend/src/server/api/endpoints/users/reactions.ts

@@ -70,7 +70,23 @@ export default define(meta, paramDef, async (ps, me) => {
 
 	generateVisibilityQuery(query, me);
 
-	const reactions = await query.take(ps.limit).getMany();
+	// We fetch more than requested because some may be filtered out, and if there's less than
+	// requested, the pagination stops.
+	const found = [];
+	const take = Math.floor(ps.limit * 1.5);
+	let skip = 0;
+	while (found.length < ps.limit) {
+		const reactions = await query.take(take).skip(skip).getMany();
+		found.push(
+			...(await NoteReactions.packMany(reactions, me, { withNote: true })),
+		);
+		skip += take;
+		if (reactions.length < take) break;
+	}
 
-	return await NoteReactions.packMany(reactions, me, { withNote: true });
+	if (found.length > ps.limit) {
+		found.length = ps.limit;
+	}
+
+	return found;
 });

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

@@ -1,5 +1,9 @@
 import * as mfm from "mfm-js";
-import { publishMainStream, publishNoteStream } from "@/services/stream.js";
+import {
+	publishMainStream,
+	publishNotesStream,
+	publishNoteStream,
+} from "@/services/stream.js";
 import DeliverManager from "@/remote/activitypub/deliver-manager.js";
 import renderNote from "@/remote/activitypub/renderer/note.js";
 import renderCreate from "@/remote/activitypub/renderer/create.js";
@@ -45,7 +49,6 @@ import {
 	genId,
 	genIdAt,
 	isSilencedServer,
-	publishToNotesStream,
 } from "backend-rs";
 import { countSameRenotes } from "@/misc/count-same-renotes.js";
 import { deliverToRelays, getCachedRelays } from "../relay.js";
@@ -508,7 +511,7 @@ export default async (
 								30,
 							);
 						}
-						publishToNotesStream(toRustObject(noteToPublish));
+						publishNotesStream(noteToPublish);
 					}
 				} finally {
 					await lock.release();

packages/backend/src/services/stream.ts

@@ -193,10 +193,9 @@ class Publisher {
 	// 	);
 	// };
 
-	/* ported to backend-rs */
-	// public publishNotesStream = (note: Note): void => {
-	// 	this.publish("notesStream", null, note);
-	// };
+	public publishNotesStream = (note: Note): void => {
+		this.publish("notesStream", null, note);
+	};
 
 	/* ported to backend-rs */
 	// public publishAdminStream = <K extends keyof AdminStreamTypes>(
@@ -222,7 +221,7 @@ export const publishUserEvent = publisher.publishUserEvent;
 export const publishMainStream = publisher.publishMainStream;
 export const publishDriveStream = publisher.publishDriveStream;
 export const publishNoteStream = publisher.publishNoteStream;
-// export const publishNotesStream = publisher.publishNotesStream;
+export const publishNotesStream = publisher.publishNotesStream;
 // export const publishChannelStream = publisher.publishChannelStream;
 export const publishUserListStream = publisher.publishUserListStream;
 // export const publishAntennaStream = publisher.publishAntennaStream;