Refactored RPC client and fixed POST body

packages/backend/src/mag/rpc-client.ts

@@ -76,6 +76,7 @@ function getRpcClient(): Socket {
         const rest = buf.readBuffer();
         buf.clear();
         buf.writeBuffer(rest);
+
         client!.emit(`mag-data:${serial}`, dataDecoded);
     });
 
@@ -85,9 +86,10 @@ function getRpcClient(): Socket {
 }
 
 async function rpcCall<D, T>(method: string, data: D): Promise<T> {
-    const currentSerial = serial;
+    const currentSerial = BigInt(new Date().getTime()) * BigInt(100000) + serial;
     serial = serial + BigInt(1);
 
+    return new Promise((resolve, reject) => {
         const header = new Uint8Array([77]);
         const textEncoder = new TextEncoder();
         const methodBuf = textEncoder.encode(method);
@@ -107,34 +109,49 @@ async function rpcCall<D, T>(method: string, data: D): Promise<T> {
         packetDataView.setUint32(header.length + serialLength + sizeLength + methodBuf.length, dataBuf.length);
 
         const client = getRpcClient();
-    client.write(packetBuf);
 
-    const fut = new Promise((resolve, reject) => {
-        client.once(`mag-data:${currentSerial}`, resolve);
-        client.once("close", reject);
-        client.once("error", reject);
-    });
+        const timeout = setTimeout(() => {
+            client.off("close", clearAndReject);
+            client.off("error", clearAndReject);
+            reject(new Error("Promise timeout"))
+        }, 60000);
 
-    const result = await fut as { success: false, data: any } | { success: true, data: T };
+        const clearAndReject = (e: any) => {
+            clearTimeout(timeout);
+            client.removeAllListeners(`mag-data:${currentSerial}`);
+            reject(e);
+        };
 
-    if (!result.success) {
-        throw result.data;
+        client.once(`mag-data:${currentSerial}`, (data: { success: false, data: any } | { success: true, data: T }) => {
+            clearTimeout(timeout);
+            client.off("close", clearAndReject);
+            client.off("error", clearAndReject);
+
+            if (!data.success) {
+                reject(data.data);
             }
 
-    return result.data;
+            resolve(data.data);
+        });
+
+        client.prependOnceListener("close", clearAndReject);
+        client.prependOnceListener("error", clearAndReject);
+
+        client.write(packetBuf);
+    });
 }
 
 
 export async function magApGet(userId: string, url: string): Promise<IObject> {
-    logger.info(`AP GET to: ${url}`);
+    logger.debug(`AP GET to: ${url}`);
     return await rpcCall("/ap/get", {
         user_id: userId,
         url
     });
 }
 
-export async function magApPost(userId: string, url: string, body: string): Promise<IObject> {
-    logger.info(`AP POST to: ${url}`);
+export async function magApPost(userId: string, url: string, body: any): Promise<IObject> {
+    logger.debug(`AP POST to: ${url}`);
     return await rpcCall("/ap/post", {
         user_id: userId,
         url,

packages/backend/src/remote/activitypub/request.ts

@@ -1,9 +1,7 @@
 import type {User} from "@/models/entities/user.js";
 import {magApGet, magApPost} from "@/mag/rpc-client.js";
 
-export default async (user: { id: User["id"] }, url: string, object: any) => {
-	const body = JSON.stringify(object);
-
+export default async (user: { id: User["id"] }, url: string, body: any) => {
     return await magApPost(user.id, url, body);
 };