refactor (backend): port publishInternalEvent to backend-rs

packages/backend-rs/index.d.ts

@@ -574,6 +574,21 @@ export interface Instance {
   faviconUrl: string | null
 }
 
+export declare enum InternalEvent {
+  Suspend = 0,
+  Silence = 1,
+  Moderator = 2,
+  Token = 3,
+  LocalUser = 4,
+  RemoteUser = 5,
+  WebhookCreated = 6,
+  WebhookUpdated = 7,
+  WebhookDeleted = 8,
+  AntennaCreated = 9,
+  AntennaUpdated = 10,
+  AntennaDeleted = 11
+}
+
 /**
  * Checks if a server is allowlisted.
  * Returns `Ok(true)` if private mode is disabled.
@@ -1164,6 +1179,8 @@ export declare function publishToDriveFolderStream(userId: string, kind: DriveFo
 
 export declare function publishToGroupChatStream(groupId: string, kind: ChatEvent, object: any): Promise<void>
 
+export declare function publishToInternalStream(kind: InternalEvent, object: any): Promise<void>
+
 export declare function publishToMainStream(userId: string, kind: Event, object: any): Promise<void>
 
 export declare function publishToModerationStream(moderatorId: string, report: AbuseUserReportLike): Promise<void>

packages/backend-rs/index.js

@@ -394,6 +394,7 @@ module.exports.greet = nativeBinding.greet
 module.exports.hashPassword = nativeBinding.hashPassword
 module.exports.Inbound = nativeBinding.Inbound
 module.exports.initializeRustLogger = nativeBinding.initializeRustLogger
+module.exports.InternalEvent = nativeBinding.InternalEvent
 module.exports.isAllowedServer = nativeBinding.isAllowedServer
 module.exports.isBlockedServer = nativeBinding.isBlockedServer
 module.exports.isOldPasswordAlgorithm = nativeBinding.isOldPasswordAlgorithm
@@ -425,6 +426,7 @@ module.exports.publishToChatStream = nativeBinding.publishToChatStream
 module.exports.publishToDriveFileStream = nativeBinding.publishToDriveFileStream
 module.exports.publishToDriveFolderStream = nativeBinding.publishToDriveFolderStream
 module.exports.publishToGroupChatStream = nativeBinding.publishToGroupChatStream
+module.exports.publishToInternalStream = nativeBinding.publishToInternalStream
 module.exports.publishToMainStream = nativeBinding.publishToMainStream
 module.exports.publishToModerationStream = nativeBinding.publishToModerationStream
 module.exports.publishToNotesStream = nativeBinding.publishToNotesStream

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

@@ -5,6 +5,7 @@ pub mod chat_index;
 pub mod custom_emoji;
 pub mod drive;
 pub mod group_chat;
+pub mod internal;
 pub mod main;
 pub mod moderation;
 pub mod note;

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

@@ -0,0 +1,45 @@
+use crate::service::stream::{publish_to_stream, Error, Stream};
+
+#[macros::export]
+pub enum InternalEvent {
+    Suspend,
+    Silence,
+    Moderator,
+    Token,
+    LocalUser,
+    RemoteUser,
+    WebhookCreated,
+    WebhookUpdated,
+    WebhookDeleted,
+    AntennaCreated,
+    AntennaUpdated,
+    AntennaDeleted,
+}
+
+// We want to merge `kind` and `object` into a single enum
+// https://github.com/napi-rs/napi-rs/issues/2036
+
+#[macros::export(js_name = "publishToInternalStream")]
+pub async fn publish(kind: InternalEvent, object: &serde_json::Value) -> Result<(), Error> {
+    let kind = match kind {
+        InternalEvent::Suspend => "userChangeSuspendedState",
+        InternalEvent::Silence => "userChangeSilencedState",
+        InternalEvent::Moderator => "userChangeModeratorState",
+        InternalEvent::Token => "userTokenRegenerated",
+        InternalEvent::LocalUser => "localUserUpdated",
+        InternalEvent::RemoteUser => "remoteUserUpdated",
+        InternalEvent::WebhookCreated => "webhookCreated",
+        InternalEvent::WebhookUpdated => "webhookUpdated",
+        InternalEvent::WebhookDeleted => "webhookDeleted",
+        InternalEvent::AntennaCreated => "antennaCreated",
+        InternalEvent::AntennaUpdated => "antennaUpdated",
+        InternalEvent::AntennaDeleted => "antennaDeleted",
+    };
+
+    publish_to_stream(
+        &Stream::Internal,
+        Some(kind),
+        Some(serde_json::to_string(&object)?),
+    )
+    .await
+}

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

@@ -16,7 +16,14 @@ import type { IRemoteUser, CacheableUser } from "@/models/entities/user.js";
 import { User } from "@/models/entities/user.js";
 import type { Emoji } from "@/models/entities/emoji.js";
 import { UserNotePining } from "@/models/entities/user-note-pining.js";
-import { genId, genIdAt, isSameOrigin, toPuny } from "backend-rs";
+import {
+	genId,
+	genIdAt,
+	InternalEvent,
+	isSameOrigin,
+	publishToInternalStream,
+	toPuny,
+} from "backend-rs";
 import { UserPublickey } from "@/models/entities/user-publickey.js";
 import { isDuplicateKeyValueError } from "@/misc/is-duplicate-key-value-error.js";
 import { UserProfile } from "@/models/entities/user-profile.js";
@@ -26,7 +33,6 @@ import { normalizeForSearch } from "@/misc/normalize-for-search.js";
 import { truncate } from "@/misc/truncate.js";
 import { StatusError } from "@/misc/fetch.js";
 import { uriPersonCache } from "@/services/user-cache.js";
-import { publishInternalEvent } from "@/services/stream.js";
 import { db } from "@/db/postgre.js";
 import { apLogger } from "../logger.js";
 import { htmlToMfm } from "../misc/html-to-mfm.js";
@@ -611,7 +617,7 @@ export async function updatePerson(
 		},
 	);
 
-	publishInternalEvent("remoteUserUpdated", { id: user.id });
+	publishToInternalStream(InternalEvent.RemoteUser, { id: user.id });
 
 	// Hashtag Update
 	updateUsertags(user, tags);

packages/backend/src/server/api/endpoints/admin/drive-capacity-override.ts

@@ -1,7 +1,7 @@
 import define from "@/server/api/define.js";
 import { Users } from "@/models/index.js";
 import { insertModerationLog } from "@/services/insert-moderation-log.js";
-import { publishInternalEvent } from "@/services/stream.js";
+import { InternalEvent, publishToInternalStream } from "backend-rs";
 export const meta = {
 	tags: ["admin"],
 
@@ -33,7 +33,7 @@ export default define(meta, paramDef, async (ps, me) => {
 		driveCapacityOverrideMb: ps.overrideMb,
 	});
 
-	publishInternalEvent("localUserUpdated", {
+	publishToInternalStream(InternalEvent.LocalUser, {
 		id: user.id,
 	});
 

packages/backend/src/server/api/endpoints/admin/moderators/add.ts

@@ -1,6 +1,6 @@
 import define from "@/server/api/define.js";
 import { Users } from "@/models/index.js";
-import { publishInternalEvent } from "@/services/stream.js";
+import { InternalEvent, publishToInternalStream } from "backend-rs";
 
 export const meta = {
 	tags: ["admin"],
@@ -32,7 +32,7 @@ export default define(meta, paramDef, async (ps) => {
 		isModerator: true,
 	});
 
-	publishInternalEvent("userChangeModeratorState", {
+	publishToInternalStream(InternalEvent.Moderator, {
 		id: user.id,
 		isModerator: true,
 	});

packages/backend/src/server/api/endpoints/admin/moderators/remove.ts

@@ -1,6 +1,6 @@
 import define from "@/server/api/define.js";
 import { Users } from "@/models/index.js";
-import { publishInternalEvent } from "@/services/stream.js";
+import { InternalEvent, publishToInternalStream } from "backend-rs";
 
 export const meta = {
 	tags: ["admin"],
@@ -28,7 +28,7 @@ export default define(meta, paramDef, async (ps) => {
 		isModerator: false,
 	});
 
-	publishInternalEvent("userChangeModeratorState", {
+	publishToInternalStream(InternalEvent.Moderator, {
 		id: user.id,
 		isModerator: false,
 	});

packages/backend/src/server/api/endpoints/admin/set-emoji-moderator.ts

@@ -1,6 +1,5 @@
 import define from "@/server/api/define.js";
 import { Users } from "@/models/index.js";
-import { publishInternalEvent } from "@/services/stream.js";
 import type { EmojiModPerm } from "@/models/entities/user.js";
 import { unsafeCast } from "@/prelude/unsafe-cast.js";
 
@@ -36,9 +35,4 @@ export default define(meta, paramDef, async (ps) => {
 	await Users.update(user.id, {
 		emojiModPerm: unsafeCast<EmojiModPerm>(ps.emojiModPerm),
 	});
-
-	publishInternalEvent("userChangeModeratorState", {
-		id: user.id,
-		isModerator: true,
-	});
 });

packages/backend/src/server/api/endpoints/admin/silence-user.ts

@@ -1,7 +1,7 @@
 import define from "@/server/api/define.js";
 import { Users } from "@/models/index.js";
 import { insertModerationLog } from "@/services/insert-moderation-log.js";
-import { publishInternalEvent } from "@/services/stream.js";
+import { InternalEvent, publishToInternalStream } from "backend-rs";
 
 export const meta = {
 	tags: ["admin"],
@@ -33,7 +33,7 @@ export default define(meta, paramDef, async (ps, me) => {
 		isSilenced: true,
 	});
 
-	publishInternalEvent("userChangeSilencedState", {
+	publishToInternalStream(InternalEvent.Silence, {
 		id: user.id,
 		isSilenced: true,
 	});

packages/backend/src/server/api/endpoints/admin/unsilence-user.ts

@@ -1,7 +1,7 @@
 import define from "@/server/api/define.js";
 import { Users } from "@/models/index.js";
 import { insertModerationLog } from "@/services/insert-moderation-log.js";
-import { publishInternalEvent } from "@/services/stream.js";
+import { InternalEvent, publishToInternalStream } from "backend-rs";
 
 export const meta = {
 	tags: ["admin"],
@@ -29,7 +29,7 @@ export default define(meta, paramDef, async (ps, me) => {
 		isSilenced: false,
 	});
 
-	publishInternalEvent("userChangeSilencedState", {
+	publishToInternalStream(InternalEvent.Silence, {
 		id: user.id,
 		isSilenced: false,
 	});

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

@@ -1,8 +1,13 @@
 import define from "@/server/api/define.js";
-import { fetchMeta, genIdAt, updateAntennaCache } from "backend-rs";
+import {
+	fetchMeta,
+	genIdAt,
+	InternalEvent,
+	publishToInternalStream,
+	updateAntennaCache,
+} from "backend-rs";
 import { Antennas, UserLists, UserGroupJoinings } from "@/models/index.js";
 import { ApiError } from "@/server/api/error.js";
-import { publishInternalEvent } from "@/services/stream.js";
 
 export const meta = {
 	tags: ["antennas"],
@@ -172,7 +177,7 @@ export default define(meta, paramDef, async (ps, user) => {
 		notify: ps.notify,
 	}).then((x) => Antennas.findOneByOrFail(x.identifiers[0]));
 
-	publishInternalEvent("antennaCreated", antenna);
+	await publishToInternalStream(InternalEvent.AntennaCreated, antenna);
 	await updateAntennaCache();
 
 	return await Antennas.pack(antenna);

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

@@ -1,8 +1,11 @@
 import define from "@/server/api/define.js";
 import { ApiError } from "@/server/api/error.js";
 import { Antennas } from "@/models/index.js";
-import { publishInternalEvent } from "@/services/stream.js";
-import { updateAntennaCache } from "backend-rs";
+import {
+	InternalEvent,
+	publishToInternalStream,
+	updateAntennaCache,
+} from "backend-rs";
 
 export const meta = {
 	tags: ["antennas"],
@@ -40,6 +43,6 @@ export default define(meta, paramDef, async (ps, user) => {
 
 	await Antennas.delete(antenna.id);
 
-	publishInternalEvent("antennaDeleted", antenna);
+	await publishToInternalStream(InternalEvent.AntennaDeleted, antenna);
 	await updateAntennaCache();
 });

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

@@ -1,8 +1,11 @@
 import define from "@/server/api/define.js";
 import { ApiError } from "@/server/api/error.js";
 import { Antennas, UserLists, UserGroupJoinings } from "@/models/index.js";
-import { publishInternalEvent } from "@/services/stream.js";
-import { updateAntennaCache } from "backend-rs";
+import {
+	InternalEvent,
+	publishToInternalStream,
+	updateAntennaCache,
+} from "backend-rs";
 
 export const meta = {
 	tags: ["antennas"],
@@ -163,8 +166,8 @@ export default define(meta, paramDef, async (ps, user) => {
 		notify: ps.notify,
 	});
 
-	publishInternalEvent(
-		"antennaUpdated",
+	await publishToInternalStream(
+		InternalEvent.AntennaUpdated,
 		await Antennas.findOneByOrFail({ id: antenna.id }),
 	);
 	await updateAntennaCache();

packages/backend/src/server/api/endpoints/i/regenerate-token.ts

@@ -1,9 +1,11 @@
-import { publishInternalEvent, publishUserEvent } from "@/services/stream.js";
+import { publishUserEvent } from "@/services/stream.js";
 import define from "@/server/api/define.js";
 import { Users, UserProfiles } from "@/models/index.js";
 import {
 	Event,
 	generateUserToken,
+	InternalEvent,
+	publishToInternalStream,
 	publishToMainStream,
 	verifyPassword,
 } from "backend-rs";
@@ -42,7 +44,7 @@ export default define(meta, paramDef, async (ps, user) => {
 	});
 
 	// Publish event
-	publishInternalEvent("userTokenRegenerated", {
+	publishToInternalStream(InternalEvent.Token, {
 		id: user.id,
 		oldToken,
 		newToken,

packages/backend/src/server/api/endpoints/i/webhooks/create.ts

@@ -1,7 +1,6 @@
 import define from "@/server/api/define.js";
-import { genIdAt } from "backend-rs";
+import { genIdAt, InternalEvent, publishToInternalStream } from "backend-rs";
 import { Webhooks } from "@/models/index.js";
-import { publishInternalEvent } from "@/services/stream.js";
 import { webhookEventTypes } from "@/models/entities/webhook.js";
 
 export const meta = {
@@ -41,7 +40,7 @@ export default define(meta, paramDef, async (ps, user) => {
 		on: ps.on,
 	}).then((x) => Webhooks.findOneByOrFail(x.identifiers[0]));
 
-	publishInternalEvent("webhookCreated", webhook);
+	publishToInternalStream(InternalEvent.WebhookCreated, webhook);
 
 	return webhook;
 });

packages/backend/src/server/api/endpoints/i/webhooks/delete.ts

@@ -1,7 +1,7 @@
 import define from "@/server/api/define.js";
 import { ApiError } from "@/server/api/error.js";
 import { Webhooks } from "@/models/index.js";
-import { publishInternalEvent } from "@/services/stream.js";
+import { InternalEvent, publishToInternalStream } from "backend-rs";
 
 export const meta = {
 	tags: ["webhooks"],
@@ -39,5 +39,5 @@ export default define(meta, paramDef, async (ps, user) => {
 
 	await Webhooks.delete(webhook.id);
 
-	publishInternalEvent("webhookDeleted", webhook);
+	publishToInternalStream(InternalEvent.WebhookDeleted, webhook);
 });

packages/backend/src/server/api/endpoints/i/webhooks/update.ts

@@ -1,7 +1,7 @@
 import define from "@/server/api/define.js";
 import { ApiError } from "@/server/api/error.js";
 import { Webhooks } from "@/models/index.js";
-import { publishInternalEvent } from "@/services/stream.js";
+import { InternalEvent, publishToInternalStream } from "backend-rs";
 import { webhookEventTypes } from "@/models/entities/webhook.js";
 
 export const meta = {
@@ -57,5 +57,5 @@ export default define(meta, paramDef, async (ps, user) => {
 		active: ps.active,
 	});
 
-	publishInternalEvent("webhookUpdated", webhook);
+	publishToInternalStream(InternalEvent.WebhookUpdated, webhook);
 });

packages/backend/src/services/stream.ts

@@ -14,7 +14,7 @@ import type {
 	// ChannelStreamTypes,
 	// DriveStreamTypes,
 	// GroupMessagingStreamTypes,
-	InternalStreamTypes,
+	// InternalStreamTypes,
 	// MainStreamTypes,
 	// MessagingIndexStreamTypes,
 	// MessagingStreamTypes,
@@ -45,12 +45,13 @@ class Publisher {
 		);
 	};
 
-	public publishInternalEvent = <K extends keyof InternalStreamTypes>(
-		type: K,
-		value?: InternalStreamTypes[K],
-	): void => {
-		this.publish("internal", type, typeof value === "undefined" ? null : value);
-	};
+	/* ported to backend-rs */
+	// public publishInternalEvent = <K extends keyof InternalStreamTypes>(
+	// 	type: K,
+	// 	value?: InternalStreamTypes[K],
+	// ): void => {
+	// 	this.publish("internal", type, typeof value === "undefined" ? null : value);
+	// };
 
 	public publishUserEvent = <K extends keyof UserStreamTypes>(
 		userId: User["id"],
@@ -215,7 +216,7 @@ const publisher = new Publisher();
 
 export default publisher;
 
-export const publishInternalEvent = publisher.publishInternalEvent;
+// export const publishInternalEvent = publisher.publishInternalEvent;
 export const publishUserEvent = publisher.publishUserEvent;
 // export const publishBroadcastStream = publisher.publishBroadcastStream;
 // export const publishMainStream = publisher.publishMainStream;

packages/backend/src/services/suspend-user.ts

@@ -5,13 +5,13 @@ import { config } from "@/config.js";
 import type { User } from "@/models/entities/user.js";
 import { Users, Followings } from "@/models/index.js";
 import { Not, IsNull } from "typeorm";
-import { publishInternalEvent } from "@/services/stream.js";
+import { InternalEvent, publishToInternalStream } from "backend-rs";
 
 export async function doPostSuspend(user: {
 	id: User["id"];
 	host: User["host"];
 }) {
-	publishInternalEvent("userChangeSuspendedState", {
+	await publishToInternalStream(InternalEvent.Suspend, {
 		id: user.id,
 		isSuspended: true,
 	});

packages/backend/src/services/unsuspend-user.ts

@@ -6,10 +6,10 @@ import { config } from "@/config.js";
 import type { User } from "@/models/entities/user.js";
 import { Users, Followings } from "@/models/index.js";
 import { Not, IsNull } from "typeorm";
-import { publishInternalEvent } from "@/services/stream.js";
+import { InternalEvent, publishToInternalStream } from "backend-rs";
 
 export async function doPostUnsuspend(user: User) {
-	publishInternalEvent("userChangeSuspendedState", {
+	publishToInternalStream(InternalEvent.Suspend, {
 		id: user.id,
 		isSuspended: false,
 	});