Add panel feature updates across API, daemon, and web

apps/api/src/plugins/socket.ts

@@ -20,6 +20,20 @@ declare module 'fastify' {
 }
 
 type ConsolePermission = 'console.read' | 'console.write';
+type ConsoleCommandAck = {
+  requestId: string | null;
+  ok: boolean;
+  error?: string;
+};
+
+interface SharedConsoleStream {
+  handle: DaemonConsoleStreamHandle;
+  subscribers: number;
+}
+
+function roomForServer(serverId: string): string {
+  return `server:console:${serverId}`;
+}
 
 export default fp(async (app: FastifyInstance) => {
   const io = new SocketIOServer(app.server, {
@@ -32,7 +46,16 @@ export default fp(async (app: FastifyInstance) => {
 
   app.decorate('io', io);
 
-  const activeStreams = new Map<string, DaemonConsoleStreamHandle>();
+  const serverStreams = new Map<string, SharedConsoleStream>();
+  const socketSubscriptions = new Map<string, string>();
+
+  const clearServerSubscriptions = (serverId: string) => {
+    for (const [socketId, subscribedServerId] of socketSubscriptions.entries()) {
+      if (subscribedServerId === serverId) {
+        socketSubscriptions.delete(socketId);
+      }
+    }
+  };
 
   io.use((socket, next) => {
     const token = typeof socket.handshake.auth?.token === 'string'
@@ -61,10 +84,20 @@ export default fp(async (app: FastifyInstance) => {
 
   io.on('connection', (socket) => {
     const cleanupSocketStream = () => {
-      const current = activeStreams.get(socket.id);
-      if (!current) return;
-      current.close();
-      activeStreams.delete(socket.id);
+      const subscribedServerId = socketSubscriptions.get(socket.id);
+      if (!subscribedServerId) return;
+
+      socketSubscriptions.delete(socket.id);
+      socket.leave(roomForServer(subscribedServerId));
+
+      const shared = serverStreams.get(subscribedServerId);
+      if (!shared) return;
+
+      shared.subscribers = Math.max(0, shared.subscribers - 1);
+      if (shared.subscribers === 0) {
+        shared.handle.close();
+        serverStreams.delete(subscribedServerId);
+      }
     };
 
     socket.on('server:console:join', async (payload: unknown) => {
@@ -94,34 +127,63 @@ export default fp(async (app: FastifyInstance) => {
         return;
       }
 
+      const previousSubscription = socketSubscriptions.get(socket.id);
+      if (previousSubscription === serverId) {
+        return;
+      }
       cleanupSocketStream();
+      socket.join(roomForServer(serverId));
 
-      try {
-        const streamHandle = await daemonOpenConsoleStream(server.node, server.serverUuid);
-        streamHandle.stream.on('data', (output) => {
-          socket.emit('server:console:output', { line: output.line });
-        });
-        streamHandle.stream.on('end', () => {
-          activeStreams.delete(socket.id);
-          socket.emit('server:console:output', { line: '[console] Stream ended' });
-        });
-        streamHandle.stream.on('error', (error) => {
-          activeStreams.delete(socket.id);
+      let shared = serverStreams.get(serverId);
+      if (!shared) {
+        try {
+          const streamHandle = await daemonOpenConsoleStream(server.node, server.serverUuid);
+          const room = roomForServer(serverId);
+
+          streamHandle.stream.on('data', (output) => {
+            io.to(room).emit('server:console:output', { line: output.line });
+          });
+
+          streamHandle.stream.on('end', () => {
+            const current = serverStreams.get(serverId);
+            if (current?.handle !== streamHandle) return;
+            serverStreams.delete(serverId);
+            clearServerSubscriptions(serverId);
+            io.to(room).emit('server:console:output', { line: '[console] Stream ended' });
+            io.in(room).socketsLeave(room);
+          });
+
+          streamHandle.stream.on('error', (error) => {
+            const current = serverStreams.get(serverId);
+            if (current?.handle !== streamHandle) return;
+            serverStreams.delete(serverId);
+            clearServerSubscriptions(serverId);
+            app.log.warn(
+              { error, serverId, serverUuid: server.serverUuid },
+              'Console stream failed',
+            );
+            io.to(room).emit('server:console:output', { line: '[error] Console stream failed' });
+            io.in(room).socketsLeave(room);
+          });
+
+          shared = {
+            handle: streamHandle,
+            subscribers: 0,
+          };
+          serverStreams.set(serverId, shared);
+        } catch (error) {
           app.log.warn(
             { error, serverId, serverUuid: server.serverUuid, socketId: socket.id },
-            'Console stream failed',
+            'Failed to open console stream',
           );
-          socket.emit('server:console:output', { line: '[error] Console stream failed' });
-        });
-
-        activeStreams.set(socket.id, streamHandle);
-      } catch (error) {
-        app.log.warn(
-          { error, serverId, serverUuid: server.serverUuid, socketId: socket.id },
-          'Failed to open console stream',
-        );
-        socket.emit('server:console:output', { line: '[error] Failed to open console stream' });
+          socket.leave(roomForServer(serverId));
+          socket.emit('server:console:output', { line: '[error] Failed to open console stream' });
+          return;
+        }
       }
+
+      shared.subscribers += 1;
+      socketSubscriptions.set(socket.id, serverId);
     });
 
     socket.on('server:console:leave', () => {
@@ -133,43 +195,67 @@ export default fp(async (app: FastifyInstance) => {
         serverId?: unknown;
         orgId?: unknown;
         command?: unknown;
+        requestId?: unknown;
       };
 
       const serverId = typeof body.serverId === 'string' ? body.serverId : '';
       const orgId = typeof body.orgId === 'string' ? body.orgId : '';
       const command = typeof body.command === 'string' ? body.command.trim() : '';
+      const requestId = typeof body.requestId === 'string' && body.requestId.trim()
+        ? body.requestId.trim()
+        : null;
 
       if (!serverId || !orgId || !command) {
         socket.emit('server:console:output', { line: '[error] Invalid command payload' });
+        const ack: ConsoleCommandAck = {
+          requestId,
+          ok: false,
+          error: 'Invalid command payload',
+        };
+        socket.emit('server:console:command:ack', ack);
         return;
       }
 
       const user = (socket.data as { user?: AccessTokenPayload }).user;
       if (!user) {
         socket.emit('server:console:output', { line: '[error] Unauthorized' });
+        const ack: ConsoleCommandAck = { requestId, ok: false, error: 'Unauthorized' };
+        socket.emit('server:console:command:ack', ack);
         return;
       }
 
       const server = await getServerContext(app, serverId, orgId);
       if (!server) {
         socket.emit('server:console:output', { line: '[error] Server not found' });
+        const ack: ConsoleCommandAck = { requestId, ok: false, error: 'Server not found' };
+        socket.emit('server:console:command:ack', ack);
         return;
       }
 
       const allowed = await hasConsolePermission(app, user, orgId, 'console.write');
       if (!allowed) {
         socket.emit('server:console:output', { line: '[error] Missing permission: console.write' });
+        const ack: ConsoleCommandAck = {
+          requestId,
+          ok: false,
+          error: 'Missing permission: console.write',
+        };
+        socket.emit('server:console:command:ack', ack);
         return;
       }
 
       try {
         await daemonSendCommand(server.node, server.serverUuid, command);
+        const ack: ConsoleCommandAck = { requestId, ok: true };
+        socket.emit('server:console:command:ack', ack);
       } catch (error) {
         app.log.warn(
           { error, serverId, serverUuid: server.serverUuid, socketId: socket.id },
           'Failed to send console command',
         );
         socket.emit('server:console:output', { line: '[error] Failed to send command' });
+        const ack: ConsoleCommandAck = { requestId, ok: false, error: 'Failed to send command' };
+        socket.emit('server:console:command:ack', ack);
       }
     });
 
@@ -179,10 +265,11 @@ export default fp(async (app: FastifyInstance) => {
   });
 
   app.addHook('onClose', async () => {
-    for (const handle of activeStreams.values()) {
-      handle.close();
+    for (const stream of serverStreams.values()) {
+      stream.handle.close();
     }
-    activeStreams.clear();
+    serverStreams.clear();
+    socketSubscriptions.clear();
 
     await new Promise<void>((resolve) => {
       io.close(() => resolve());