Add admin moderation and sync acknowledgements

server.js

@@ -25,12 +25,12 @@ app.prepare().then(() => {
   });
 
   io.on("connection", (socket) => {
-    socket.on("room:join", async ({ roomSlug } = {}) => {
-      await safeSocket(socket, async () => {
+    socket.on("room:join", async ({ roomSlug } = {}, acknowledge) => {
+      await safeSocket(socket, acknowledge, async () => {
         const session = await getSocketSession(socket);
-        if (!session || !roomSlug) return reject(socket, "Sign in to join this room.");
+        if (!session || !roomSlug) return reject(socket, "Sign in to join this room.", acknowledge);
         const context = await getRoomContext(roomSlug, session.user.id);
-        if (!context.allowed) return reject(socket, "You do not have access to this room.");
+        if (!context.allowed) return reject(socket, "You do not have access to this room.", acknowledge);
 
         socket.data.user = session.user;
         socket.data.roomSlug = context.room.slug;
@@ -40,15 +40,16 @@ app.prepare().then(() => {
 
         socket.emit("room:state", await buildRoomSnapshot(context.room.id));
         io.to(context.room.slug).emit("presence:list", getPresence(context.room.slug));
+        ok(acknowledge);
       });
     });
 
-    socket.on("queue:add", (payload) => safeRoomAction(socket, async ({ room, user }) => {
+    socket.on("queue:add", (payload, acknowledge) => safeRoomAction(socket, acknowledge, async ({ room, user }) => {
       const sourceUrl = String(payload?.sourceUrl || "").trim();
       if (!sourceUrl) return;
       const settings = await getAppSettings();
       const media = normalizeMediaUrl(sourceUrl);
-      if (!settings.allowedProviders.includes(media.provider)) return reject(socket, "This media provider is disabled.");
+      if (!settings.allowedProviders.includes(media.provider)) return reject(socket, "This media provider is disabled.", acknowledge);
 
       const nextPosition = await prisma.mediaSource.count({ where: { roomId: room.id } });
       const created = await prisma.mediaSource.create({
@@ -77,7 +78,7 @@ app.prepare().then(() => {
       await broadcastRoom(io, room.slug, room.id);
     }));
 
-    socket.on("queue:play", (payload) => safeRoomAction(socket, async ({ room, user }) => {
+    socket.on("queue:play", (payload, acknowledge) => safeRoomAction(socket, acknowledge, async ({ room, user }) => {
       const mediaSourceId = String(payload?.mediaSourceId || "");
       const media = await prisma.mediaSource.findFirst({ where: { id: mediaSourceId, roomId: room.id } });
       if (!media) return;
@@ -86,7 +87,7 @@ app.prepare().then(() => {
       await broadcastRoom(io, room.slug, room.id);
     }));
 
-    socket.on("queue:remove", (payload) => safeRoomAction(socket, async ({ room, user }) => {
+    socket.on("queue:remove", (payload, acknowledge) => safeRoomAction(socket, acknowledge, async ({ room, user }) => {
       const mediaSourceId = String(payload?.mediaSourceId || "");
       const media = await prisma.mediaSource.findFirst({ where: { id: mediaSourceId, roomId: room.id } });
       if (!media) return;
@@ -113,7 +114,7 @@ app.prepare().then(() => {
       await broadcastRoom(io, room.slug, room.id);
     }));
 
-    socket.on("queue:move", (payload) => safeRoomAction(socket, async ({ room, user }) => {
+    socket.on("queue:move", (payload, acknowledge) => safeRoomAction(socket, acknowledge, async ({ room, user }) => {
       const mediaSourceId = String(payload?.mediaSourceId || "");
       const direction = payload?.direction === "down" ? 1 : -1;
       await moveMedia(room.id, mediaSourceId, direction);
@@ -121,7 +122,7 @@ app.prepare().then(() => {
       await broadcastRoom(io, room.slug, room.id);
     }));
 
-    socket.on("playback:play", (payload) => safeRoomAction(socket, async ({ room, user }) => {
+    socket.on("playback:play", (payload, acknowledge) => safeRoomAction(socket, acknowledge, async ({ room, user }) => {
       await persistPlaybackState(room.id, user, {
         mediaSourceId: String(payload?.mediaSourceId || parseState(room.currentState)?.mediaSourceId || ""),
         status: "PLAYING",
@@ -131,7 +132,7 @@ app.prepare().then(() => {
       await broadcastRoom(io, room.slug, room.id);
     }));
 
-    socket.on("playback:pause", (payload) => safeRoomAction(socket, async ({ room, user }) => {
+    socket.on("playback:pause", (payload, acknowledge) => safeRoomAction(socket, acknowledge, async ({ room, user }) => {
       await persistPlaybackState(room.id, user, {
         mediaSourceId: String(payload?.mediaSourceId || parseState(room.currentState)?.mediaSourceId || ""),
         status: "PAUSED",
@@ -141,7 +142,7 @@ app.prepare().then(() => {
       await broadcastRoom(io, room.slug, room.id);
     }));
 
-    socket.on("playback:seek", (payload) => safeRoomAction(socket, async ({ room, user }) => {
+    socket.on("playback:seek", (payload, acknowledge) => safeRoomAction(socket, acknowledge, async ({ room, user }) => {
       const previous = parseState(room.currentState);
       await persistPlaybackState(room.id, user, {
         mediaSourceId: String(payload?.mediaSourceId || previous?.mediaSourceId || ""),
@@ -152,7 +153,7 @@ app.prepare().then(() => {
       await broadcastRoom(io, room.slug, room.id);
     }));
 
-    socket.on("chat:message", (payload) => safeRoomAction(socket, async ({ room, user }) => {
+    socket.on("chat:message", (payload, acknowledge) => safeRoomAction(socket, acknowledge, async ({ room, user }) => {
       const body = String(payload?.body || "").trim().slice(0, 1000);
       if (!body) return;
       await prisma.roomMessage.create({ data: { roomId: room.id, userId: user.id, body } });
@@ -160,8 +161,8 @@ app.prepare().then(() => {
       await broadcastRoom(io, room.slug, room.id);
     }));
 
-    socket.on("chat:delete", (payload) => safeRoomAction(socket, async ({ room, user }) => {
-      if (!canManageRoom(room, user.id, user)) return reject(socket, "Only room managers can moderate chat.");
+    socket.on("chat:delete", (payload, acknowledge) => safeRoomAction(socket, acknowledge, async ({ room, user }) => {
+      if (!canManageRoom(room, user.id, user)) return reject(socket, "Only room managers can moderate chat.", acknowledge);
       const messageId = String(payload?.messageId || "");
       if (!messageId) return;
       const result = await prisma.roomMessage.deleteMany({ where: { id: messageId, roomId: room.id } });
@@ -185,28 +186,35 @@ app.prepare().then(() => {
   });
 });
 
-async function safeSocket(socket, action) {
+async function safeSocket(socket, acknowledge, action) {
   try {
     await action();
   } catch (error) {
     console.error(error);
-    reject(socket, "Realtime action failed.");
+    reject(socket, "Realtime action failed.", acknowledge);
   }
 }
 
-async function safeRoomAction(socket, action) {
-  await safeSocket(socket, async () => {
+async function safeRoomAction(socket, acknowledge, action) {
+  await safeSocket(socket, acknowledge, async () => {
     const user = socket.data.user || (await getSocketSession(socket))?.user;
     const roomSlug = socket.data.roomSlug;
-    if (!user || !roomSlug) return reject(socket, "Join a room before sending actions.");
+    if (!user || !roomSlug) return reject(socket, "Join a room before sending actions.", acknowledge);
     const context = await getRoomContext(roomSlug, user.id);
-    if (!context.allowed) return reject(socket, "You do not have access to this room.");
-    await action({ room: context.room, user });
+    if (!context.allowed) return reject(socket, "You do not have access to this room.", acknowledge);
+    const result = await action({ room: context.room, user });
+    if (result !== false) ok(acknowledge);
   });
 }
 
-function reject(socket, message) {
+function ok(callback) {
+  if (typeof callback === "function") callback({ ok: true, at: Date.now() });
+}
+
+function reject(socket, message, callback) {
   socket.emit("room:error", { message });
+  if (typeof callback === "function") callback({ ok: false, message, at: Date.now() });
+  return false;
 }
 
 async function broadcastRoom(io, roomSlug, roomId) {