refactor (backend): port publishMessagingIndexStream to backend-rs

packages/backend-rs/index.d.ts

@@ -1157,6 +1157,11 @@ export enum ChatEvent {
   Typing = 'typing'
 }
 export function publishToChatStream(senderUserId: string, receiverUserId: string, kind: ChatEvent, object: any): void
+export enum ChatIndexEvent {
+  Message = 'message',
+  Read = 'read'
+}
+export function publishToChatIndexStream(userId: string, kind: ChatIndexEvent, object: any): void
 export interface AbuseUserReportLike {
   id: string
   targetUserId: string

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, toMastodonId, fromMastodonId, fetchMeta, metaToPugArgs, nyaify, hashPassword, verifyPassword, isOldPasswordAlgorithm, decodeReaction, countReactions, toDbReaction, removeOldAttestationChallenges, AntennaSrcEnum, DriveFileUsageHintEnum, MutedNoteReasonEnum, NoteVisibilityEnum, NotificationTypeEnum, PageVisibilityEnum, PollNotevisibilityEnum, RelayStatusEnum, UserEmojimodpermEnum, UserProfileFfvisibilityEnum, UserProfileMutingnotificationtypesEnum, initializeRustLogger, watchNote, unwatchNote, publishToChannelStream, ChatEvent, publishToChatStream, publishToModerationStream, 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, toMastodonId, fromMastodonId, fetchMeta, metaToPugArgs, nyaify, hashPassword, verifyPassword, isOldPasswordAlgorithm, decodeReaction, countReactions, toDbReaction, removeOldAttestationChallenges, AntennaSrcEnum, DriveFileUsageHintEnum, MutedNoteReasonEnum, NoteVisibilityEnum, NotificationTypeEnum, PageVisibilityEnum, PollNotevisibilityEnum, RelayStatusEnum, UserEmojimodpermEnum, UserProfileFfvisibilityEnum, UserProfileMutingnotificationtypesEnum, initializeRustLogger, watchNote, unwatchNote, publishToChannelStream, ChatEvent, publishToChatStream, ChatIndexEvent, publishToChatIndexStream, publishToModerationStream, getTimestamp, genId, genIdAt, secureRndstr } = nativeBinding
 
 module.exports.SECOND = SECOND
 module.exports.MINUTE = MINUTE
@@ -368,6 +368,8 @@ module.exports.unwatchNote = unwatchNote
 module.exports.publishToChannelStream = publishToChannelStream
 module.exports.ChatEvent = ChatEvent
 module.exports.publishToChatStream = publishToChatStream
+module.exports.ChatIndexEvent = ChatIndexEvent
+module.exports.publishToChatIndexStream = publishToChatIndexStream
 module.exports.publishToModerationStream = publishToModerationStream
 module.exports.getTimestamp = getTimestamp
 module.exports.genId = genId

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

@@ -1,6 +1,7 @@
 pub mod antenna;
 pub mod channel;
 pub mod chat;
+pub mod chat_index;
 pub mod moderation;
 
 use crate::config::CONFIG;
@@ -39,7 +40,7 @@ pub enum Stream {
     #[strum(to_string = "messagingStream:{group_id}")]
     GroupChat { group_id: String },
     #[strum(to_string = "messagingIndexStream:{user_id}")]
-    MessagingIndex { user_id: String },
+    ChatIndex { user_id: String },
 }
 
 #[derive(thiserror::Error, Debug)]

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

@@ -0,0 +1,26 @@
+use crate::service::stream::{publish_to_stream, Error, Stream};
+
+#[derive(strum::Display)]
+#[crate::export(string_enum = "camelCase")]
+pub enum ChatIndexEvent {
+    #[strum(serialize = "message")]
+    Message,
+    #[strum(serialize = "read")]
+    Read,
+}
+
+// We want to merge `kind` and `object` into a single enum
+// https://github.com/napi-rs/napi-rs/issues/2036
+
+#[crate::export(js_name = "publishToChatIndexStream")]
+pub fn publish(
+    user_id: String,
+    kind: ChatIndexEvent,
+    object: &serde_json::Value,
+) -> Result<(), Error> {
+    publish_to_stream(
+        &Stream::ChatIndex { user_id },
+        Some(kind.to_string()),
+        Some(serde_json::to_string(object)?),
+    )
+}

packages/backend/src/server/api/common/read-messaging-message.ts

@@ -2,8 +2,12 @@ import {
 	publishMainStream,
 	publishGroupMessagingStream,
 } from "@/services/stream.js";
-import { publishToChatStream, ChatEvent } from "backend-rs";
+import {
-import { publishMessagingIndexStream } from "@/services/stream.js";
+	publishToChatStream,
+	publishToChatIndexStream,
+	ChatEvent,
+	ChatIndexEvent,
+} from "backend-rs";
 import { pushNotification } from "@/services/push-notification.js";
 import type { User, IRemoteUser } from "@/models/entities/user.js";
 import type { MessagingMessage } from "@/models/entities/messaging-message.js";
@@ -55,7 +59,7 @@ export async function readUserMessagingMessage(
 
 	// Publish event
 	publishToChatStream(otherpartyId, userId, ChatEvent.Read, messageIds);
-	publishMessagingIndexStream(userId, "read", messageIds);
+	publishToChatIndexStream(userId, ChatIndexEvent.Read, messageIds);
 
 	if (!(await Users.getHasUnreadMessagingMessage(userId))) {
 		// 全ての(いままで未読だった)自分宛てのメッセージを(これで)読みましたよというイベントを発行
@@ -130,7 +134,7 @@ export async function readGroupMessagingMessage(
 		ids: reads,
 		userId: userId,
 	});
-	publishMessagingIndexStream(userId, "read", reads);
+	publishToChatIndexStream(userId, ChatIndexEvent.Read, reads);
 
 	if (!(await Users.getHasUnreadMessagingMessage(userId))) {
 		// 全ての(いままで未読だった)自分宛てのメッセージを(これで)読みましたよというイベントを発行

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

@@ -7,10 +7,16 @@ import {
 	Mutings,
 	Users,
 } from "@/models/index.js";
-import { genId, publishToChatStream, toPuny, ChatEvent } from "backend-rs";
+import {
+	genId,
+	publishToChatStream,
+	publishToChatIndexStream,
+	toPuny,
+	ChatEvent,
+	ChatIndexEvent,
+} from "backend-rs";
 import type { MessagingMessage } from "@/models/entities/messaging-message.js";
 import {
-	publishMessagingIndexStream,
 	publishMainStream,
 	publishGroupMessagingStream,
 } from "@/services/stream.js";
@@ -57,7 +63,11 @@ export async function createMessage(
 				ChatEvent.Message,
 				messageObj,
 			);
-			publishMessagingIndexStream(message.userId, "message", messageObj);
+			publishToChatIndexStream(
+				message.userId,
+				ChatIndexEvent.Message,
+				messageObj,
+			);
 			publishMainStream(message.userId, "messagingMessage", messageObj);
 		}
 
@@ -69,7 +79,11 @@ export async function createMessage(
 				ChatEvent.Message,
 				messageObj,
 			);
-			publishMessagingIndexStream(recipientUser.id, "message", messageObj);
+			publishToChatIndexStream(
+				recipientUser.id,
+				ChatIndexEvent.Message,
+				messageObj,
+			);
 			publishMainStream(recipientUser.id, "messagingMessage", messageObj);
 		}
 	} else if (recipientGroup) {
@@ -81,7 +95,11 @@ export async function createMessage(
 			userGroupId: recipientGroup.id,
 		});
 		for (const joining of joinings) {
-			publishMessagingIndexStream(joining.userId, "message", messageObj);
+			publishToChatIndexStream(
+				joining.userId,
+				ChatIndexEvent.Message,
+				messageObj,
+			);
 			publishMainStream(joining.userId, "messagingMessage", messageObj);
 		}
 	}

packages/backend/src/services/stream.ts

@@ -16,7 +16,7 @@ import type {
 	GroupMessagingStreamTypes,
 	InternalStreamTypes,
 	MainStreamTypes,
-	MessagingIndexStreamTypes,
+	// MessagingIndexStreamTypes,
 	// MessagingStreamTypes,
 	NoteStreamTypes,
 	UserListStreamTypes,
@@ -176,19 +176,20 @@ class Publisher {
 		);
 	};
 
-	public publishMessagingIndexStream = <
+	/* ported to backend-rs */
-		K extends keyof MessagingIndexStreamTypes,
+	// public publishMessagingIndexStream = <
-	>(
+	// 	K extends keyof MessagingIndexStreamTypes,
-		userId: User["id"],
+	// >(
-		type: K,
+	// 	userId: User["id"],
-		value?: MessagingIndexStreamTypes[K],
+	// 	type: K,
-	): void => {
+	// 	value?: MessagingIndexStreamTypes[K],
-		this.publish(
+	// ): void => {
-			`messagingIndexStream:${userId}`,
+	// 	this.publish(
-			type,
+	// 		`messagingIndexStream:${userId}`,
-			typeof value === "undefined" ? null : value,
+	// 		type,
-		);
+	// 		typeof value === "undefined" ? null : value,
-	};
+	// 	);
+	// };
 
 	public publishNotesStream = (note: Note): void => {
 		this.publish("notesStream", null, note);
@@ -225,6 +226,5 @@ export const publishUserListStream = publisher.publishUserListStream;
 // export const publishMessagingStream = publisher.publishMessagingStream;
 export const publishGroupMessagingStream =
 	publisher.publishGroupMessagingStream;
-export const publishMessagingIndexStream =
+// export const publishMessagingIndexStream = publisher.publishMessagingIndexStream;
-	publisher.publishMessagingIndexStream;
 // export const publishAdminStream = publisher.publishAdminStream;