refactor (backend): port publishDriveStream to backend-rs

packages/backend-rs/index.d.ts

@@ -1373,6 +1373,18 @@ export interface PackedEmoji {
   height: number | null
 }
 export function publishToBroadcastStream(emoji: PackedEmoji): Promise<void>
+export enum DriveFileEvent {
+  Create = 0,
+  Update = 1,
+  Delete = 2
+}
+export enum DriveFolderEvent {
+  Create = 0,
+  Update = 1,
+  Delete = 2
+}
+export function publishToDriveFileStream(userId: string, kind: DriveFileEvent, object: any): Promise<void>
+export function publishToDriveFolderStream(userId: string, kind: DriveFolderEvent, object: any): Promise<void>
 export function publishToGroupChatStream(groupId: string, kind: ChatEvent, object: any): Promise<void>
 export interface AbuseUserReportLike {
   id: 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, fetchMeta, updateMetaCache, metaToPugArgs, loadConfig, stringToAcct, acctToString, fetchNodeinfo, nodeinfo_2_1, nodeinfo_2_0, updateNodeinfoCache, Protocol, Inbound, Outbound, greet, initializeRustLogger, showServerInfo, isBlockedServer, isSilencedServer, isAllowedServer, checkWordMute, getFullApAccount, isSelfHost, isSameOrigin, extractHost, toPuny, isUnicodeEmoji, sqlLikeEscape, safeForSql, formatMilliseconds, getImageSizeFromUrl, isQuote, isSafeUrl, latestVersion, toMastodonId, fromMastodonId, getNoteSummary, nyaify, hashPassword, verifyPassword, isOldPasswordAlgorithm, decodeReaction, countReactions, toDbReaction, removeOldAttestationChallenges, cpuInfo, cpuUsage, memoryUsage, storageUsage, AntennaSrc, DriveFileUsageHint, MutedNoteReason, NoteVisibility, NotificationType, PageVisibility, PollNoteVisibility, RelayStatus, UserEmojiModPerm, UserProfileFfvisibility, UserProfileMutingNotificationTypes, updateAntennasOnNewNote, updateAntennaCache, watchNote, unwatchNote, PushNotificationKind, sendPushNotification, publishToChannelStream, publishToChatStream, ChatIndexEvent, publishToChatIndexStream, publishToBroadcastStream, publishToGroupChatStream, publishToModerationStream, publishToNotesStream, ChatEvent, getTimestamp, genId, genIdAt, generateSecureRandomString, generateUserToken } = nativeBinding
+const { SECOND, MINUTE, HOUR, DAY, USER_ONLINE_THRESHOLD, USER_ACTIVE_THRESHOLD, FILE_TYPE_BROWSERSAFE, fetchMeta, updateMetaCache, metaToPugArgs, loadConfig, stringToAcct, acctToString, fetchNodeinfo, nodeinfo_2_1, nodeinfo_2_0, updateNodeinfoCache, Protocol, Inbound, Outbound, greet, initializeRustLogger, showServerInfo, isBlockedServer, isSilencedServer, isAllowedServer, checkWordMute, getFullApAccount, isSelfHost, isSameOrigin, extractHost, toPuny, isUnicodeEmoji, sqlLikeEscape, safeForSql, formatMilliseconds, getImageSizeFromUrl, isQuote, isSafeUrl, latestVersion, toMastodonId, fromMastodonId, getNoteSummary, nyaify, hashPassword, verifyPassword, isOldPasswordAlgorithm, decodeReaction, countReactions, toDbReaction, removeOldAttestationChallenges, cpuInfo, cpuUsage, memoryUsage, storageUsage, AntennaSrc, DriveFileUsageHint, MutedNoteReason, NoteVisibility, NotificationType, PageVisibility, PollNoteVisibility, RelayStatus, UserEmojiModPerm, UserProfileFfvisibility, UserProfileMutingNotificationTypes, updateAntennasOnNewNote, updateAntennaCache, watchNote, unwatchNote, PushNotificationKind, sendPushNotification, publishToChannelStream, publishToChatStream, ChatIndexEvent, publishToChatIndexStream, publishToBroadcastStream, DriveFileEvent, DriveFolderEvent, publishToDriveFileStream, publishToDriveFolderStream, publishToGroupChatStream, publishToModerationStream, publishToNotesStream, ChatEvent, getTimestamp, genId, genIdAt, generateSecureRandomString, generateUserToken } = nativeBinding
 
 module.exports.SECOND = SECOND
 module.exports.MINUTE = MINUTE
@@ -389,6 +389,10 @@ module.exports.publishToChatStream = publishToChatStream
 module.exports.ChatIndexEvent = ChatIndexEvent
 module.exports.publishToChatIndexStream = publishToChatIndexStream
 module.exports.publishToBroadcastStream = publishToBroadcastStream
+module.exports.DriveFileEvent = DriveFileEvent
+module.exports.DriveFolderEvent = DriveFolderEvent
+module.exports.publishToDriveFileStream = publishToDriveFileStream
+module.exports.publishToDriveFolderStream = publishToDriveFolderStream
 module.exports.publishToGroupChatStream = publishToGroupChatStream
 module.exports.publishToModerationStream = publishToModerationStream
 module.exports.publishToNotesStream = publishToNotesStream

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

@@ -3,6 +3,7 @@ pub mod channel;
 pub mod chat;
 pub mod chat_index;
 pub mod custom_emoji;
+pub mod drive;
 pub mod group_chat;
 pub mod moderation;
 pub mod notes;

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

@@ -0,0 +1,58 @@
+use crate::service::stream::{publish_to_stream, Error, Stream};
+
+#[crate::export]
+pub enum DriveFileEvent {
+    Create,
+    Update,
+    Delete,
+}
+
+#[crate::export]
+pub enum DriveFolderEvent {
+    Create,
+    Update,
+    Delete,
+}
+
+// We want to merge `kind` and `object` into a single enum and merge the 2 functions
+// https://github.com/napi-rs/napi-rs/issues/2036
+
+#[crate::export(js_name = "publishToDriveFileStream")]
+pub async fn publish_file(
+    user_id: String,
+    kind: DriveFileEvent,
+    object: &serde_json::Value, // file (create, update) or file id (delete)
+) -> Result<(), Error> {
+    let kind = match kind {
+        DriveFileEvent::Create => "fileCreated",
+        DriveFileEvent::Update => "fileUpdated",
+        DriveFileEvent::Delete => "fileDeleted",
+    };
+
+    publish_to_stream(
+        &Stream::Drive { user_id },
+        Some(kind),
+        Some(serde_json::to_string(object)?),
+    )
+    .await
+}
+
+#[crate::export(js_name = "publishToDriveFolderStream")]
+pub async fn publish_folder(
+    user_id: String,
+    kind: DriveFolderEvent,
+    object: &serde_json::Value, // folder (create, update) or folder id (delete)
+) -> Result<(), Error> {
+    let kind = match kind {
+        DriveFolderEvent::Create => "folderCreated",
+        DriveFolderEvent::Update => "folderUpdated",
+        DriveFolderEvent::Delete => "folderDeleted",
+    };
+
+    publish_to_stream(
+        &Stream::Drive { user_id },
+        Some(kind),
+        Some(serde_json::to_string(object)?),
+    )
+    .await
+}

packages/backend/src/server/api/endpoints/drive/files/delete.ts

@@ -1,5 +1,5 @@
 import { deleteFile } from "@/services/drive/delete-file.js";
-import { publishDriveStream } from "@/services/stream.js";
+import { publishToDriveFileStream, DriveFileEvent } from "backend-rs";
 import define from "@/server/api/define.js";
 import { ApiError } from "@/server/api/error.js";
 import { DriveFiles } from "@/models/index.js";
@@ -51,5 +51,5 @@ export default define(meta, paramDef, async (ps, user) => {
 	await deleteFile(file);
 
 	// Publish fileDeleted event
-	publishDriveStream(user.id, "fileDeleted", file.id);
+	publishToDriveFileStream(user.id, DriveFileEvent.Delete, file.id);
 });

packages/backend/src/server/api/endpoints/drive/files/update.ts

@@ -1,8 +1,9 @@
-import { publishDriveStream } from "@/services/stream.js";
+import { publishToDriveFileStream, DriveFileEvent } from "backend-rs";
 import { DriveFiles, DriveFolders } from "@/models/index.js";
 import { config } from "@/config.js";
 import define from "@/server/api/define.js";
 import { ApiError } from "@/server/api/error.js";
+import { toRustObject } from "@/prelude/undefined-to-null.js";
 
 export const meta = {
 	tags: ["drive"],
@@ -110,7 +111,7 @@ export default define(meta, paramDef, async (ps, user) => {
 	const fileObj = await DriveFiles.pack(file, { self: true });
 
 	// Publish fileUpdated event
-	publishDriveStream(user.id, "fileUpdated", fileObj);
+	publishToDriveFileStream(user.id, DriveFileEvent.Update, toRustObject(file));
 
 	return fileObj;
 });

packages/backend/src/server/api/endpoints/drive/folders/create.ts

@@ -1,8 +1,9 @@
-import { publishDriveStream } from "@/services/stream.js";
+import { publishToDriveFolderStream, DriveFolderEvent } from "backend-rs";
 import define from "@/server/api/define.js";
 import { ApiError } from "@/server/api/error.js";
 import { DriveFolders } from "@/models/index.js";
 import { genIdAt } from "backend-rs";
+import { toRustObject } from "@/prelude/undefined-to-null.js";
 
 export const meta = {
 	tags: ["drive"],
@@ -64,7 +65,11 @@ export default define(meta, paramDef, async (ps, user) => {
 	const folderObj = await DriveFolders.pack(folder);
 
 	// Publish folderCreated event
-	publishDriveStream(user.id, "folderCreated", folderObj);
+	publishToDriveFolderStream(
+		user.id,
+		DriveFolderEvent.Create,
+		toRustObject(folder),
+	);
 
 	return folderObj;
 });

packages/backend/src/server/api/endpoints/drive/folders/delete.ts

@@ -1,5 +1,5 @@
 import define from "@/server/api/define.js";
-import { publishDriveStream } from "@/services/stream.js";
+import { publishToDriveFolderStream, DriveFolderEvent } from "backend-rs";
 import { ApiError } from "@/server/api/error.js";
 import { DriveFolders, DriveFiles } from "@/models/index.js";
 
@@ -56,5 +56,5 @@ export default define(meta, paramDef, async (ps, user) => {
 	await DriveFolders.delete(folder.id);
 
 	// Publish folderCreated event
-	publishDriveStream(user.id, "folderDeleted", folder.id);
+	publishToDriveFolderStream(user.id, DriveFolderEvent.Delete, folder.id);
 });

packages/backend/src/server/api/endpoints/drive/folders/update.ts

@@ -1,7 +1,8 @@
-import { publishDriveStream } from "@/services/stream.js";
+import { publishToDriveFolderStream, DriveFolderEvent } from "backend-rs";
 import define from "@/server/api/define.js";
 import { ApiError } from "@/server/api/error.js";
 import { DriveFolders } from "@/models/index.js";
+import { toRustObject } from "@/prelude/undefined-to-null.js";
 
 export const meta = {
 	tags: ["drive"],
@@ -112,7 +113,11 @@ export default define(meta, paramDef, async (ps, user) => {
 	const folderObj = await DriveFolders.pack(folder);
 
 	// Publish folderUpdated event
-	publishDriveStream(user.id, "folderUpdated", folderObj);
+	publishToDriveFolderStream(
+		user.id,
+		DriveFolderEvent.Update,
+		toRustObject(folder),
+	);
 
 	return folderObj;
 });

packages/backend/src/services/drive/add-file.ts

@@ -5,8 +5,14 @@ import { v4 as uuid } from "uuid";
 import type S3 from "aws-sdk/clients/s3.js"; // TODO: migrate to SDK v3
 import sharp from "sharp";
 import { IsNull } from "typeorm";
-import { publishMainStream, publishDriveStream } from "@/services/stream.js";
-import { FILE_TYPE_BROWSERSAFE, fetchMeta, genId } from "backend-rs";
+import { publishMainStream } from "@/services/stream.js";
+import {
+	DriveFileEvent,
+	FILE_TYPE_BROWSERSAFE,
+	fetchMeta,
+	genId,
+	publishToDriveFileStream,
+} from "backend-rs";
 import { contentDisposition } from "@/misc/content-disposition.js";
 import { getFileInfo } from "@/misc/get-file-info.js";
 import {
@@ -28,6 +34,7 @@ import { driveLogger } from "./logger.js";
 import { GenerateVideoThumbnail } from "./generate-video-thumbnail.js";
 import { deleteFile } from "./delete-file.js";
 import { inspect } from "node:util";
+import { toRustObject } from "@/prelude/undefined-to-null.js";
 
 const logger = driveLogger.createSubLogger("register", "yellow");
 
@@ -656,11 +663,15 @@ export async function addFile({
 
 	logger.info(`drive file has been created ${file.id}`);
 
-	if (user) {
+	if (user != null) {
 		DriveFiles.pack(file, { self: true }).then((packedFile) => {
 			// Publish driveFileCreated event
 			publishMainStream(user.id, "driveFileCreated", packedFile);
-			publishDriveStream(user.id, "fileCreated", packedFile);
+			publishToDriveFileStream(
+				user.id,
+				DriveFileEvent.Create,
+				toRustObject(file),
+			);
 		});
 	}
 

packages/backend/src/services/stream.ts

@@ -12,7 +12,7 @@ import type {
 	// AntennaStreamTypes,
 	// BroadcastTypes,
 	// ChannelStreamTypes,
-	DriveStreamTypes,
+	// DriveStreamTypes,
 	// GroupMessagingStreamTypes,
 	InternalStreamTypes,
 	MainStreamTypes,
@@ -88,17 +88,18 @@ class Publisher {
 		);
 	};
 
-	public publishDriveStream = <K extends keyof DriveStreamTypes>(
-		userId: User["id"],
-		type: K,
-		value?: DriveStreamTypes[K],
-	): void => {
-		this.publish(
-			`driveStream:${userId}`,
-			type,
-			typeof value === "undefined" ? null : value,
-		);
-	};
+	/* ported to backend-rs */
+	// public publishDriveStream = <K extends keyof DriveStreamTypes>(
+	// 	userId: User["id"],
+	// 	type: K,
+	// 	value?: DriveStreamTypes[K],
+	// ): void => {
+	// 	this.publish(
+	// 		`driveStream:${userId}`,
+	// 		type,
+	// 		typeof value === "undefined" ? null : value,
+	// 	);
+	// };
 
 	public publishNoteStream = <K extends keyof NoteStreamTypes>(
 		noteId: Note["id"],
@@ -220,7 +221,7 @@ export const publishInternalEvent = publisher.publishInternalEvent;
 export const publishUserEvent = publisher.publishUserEvent;
 // export const publishBroadcastStream = publisher.publishBroadcastStream;
 export const publishMainStream = publisher.publishMainStream;
-export const publishDriveStream = publisher.publishDriveStream;
+// export const publishDriveStream = publisher.publishDriveStream;
 export const publishNoteStream = publisher.publishNoteStream;
 // export const publishNotesStream = publisher.publishNotesStream;
 // export const publishChannelStream = publisher.publishChannelStream;