WKVS/websocket/index.js

const { createClient } = require('redis');
const WebSocket = require("ws");
const url = require("url");
const dotenv = require('dotenv');
const path = require('path');

const envPath = path.resolve(__dirname, '..', 'config', '.env.redis');

dotenv.config({ path: envPath });

const redisClient = createClient({
  password: process.env.REDDIS_PASSWORD,
  socket: {
    host: process.env.REDDIS_HOST,
    port: process.env.REDDIS_PORT
  }
});

redisClient.on('error', (err) => console.error('Redis Client Error', err));

async function startRedis() {
  await redisClient.connect();
  console.log('Connected and authenticated with Redis!');
}

startRedis();

const PORT = 8082;
const wss = new WebSocket.Server({ port: PORT });

const groups = new Map();

const authenticatedGroups = new Map();

const clients = new Set();

const HEARTBEAT_INTERVAL = 30000;
const MAX_MESSAGE_SIZE = 10 * 1024; // 10KB


function addToGroup(ws, access, authenticted = false) {
  if (authenticted) {
    if (!authenticatedGroups.has(access)) {
      authenticatedGroups.set(access, new Set());
    }
    authenticatedGroups.get(access).add(ws);
    ws.send("Dieser Benutzer wurde einer Authentifizierten Gruppe hinzugefügt." + access);
  } else {
    if (!groups.has(access)) {
      groups.set(access, new Set());
    }
    groups.get(access).add(ws);
  }
}

function removeFromGroup(ws, access, authenticted = false) {
  let group;
  if (authenticted) {
    group = authenticatedGroups.get(access);
  } else {
    group = groups.get(access);
  }
  if (group) {
    group.delete(ws);
    if (group.size === 0) {
      groups.delete(access);
    }
  }
}

function sendToGroup(access, messageObj, authenticted = false, excludeWs = null) {
  let group;
  if (authenticted) {
    group = authenticatedGroups.get(access);
  } else {
    group = groups.get(access);
  }

  if (!group) return;

  const message = JSON.stringify(messageObj);

  for (const ws of group) {
    if (ws.readyState === WebSocket.OPEN && ws !== excludeWs) {
      ws.send(message);
    }
  }
}

function sendToSelf(ws, messageObj) {

  const message = JSON.stringify(messageObj);

  if (ws.readyState === WebSocket.OPEN) {
    ws.send(message);
  }
}

function sendToGroupsContaining(substring, messageObj, authenticted = false) {
  const message = JSON.stringify(messageObj);

   if (authenticted) {
    for (const [access, group] of authenticatedGroups.entries()) {
      if (access.includes(substring)) {
        for (const ws of group) {
          if (ws.readyState === WebSocket.OPEN) {
            ws.send(message);
          }
        }
      }
    }
  } else {
    for (const [access, group] of groups.entries()) {
      if (access.includes(substring)) {
        for (const ws of group) {
          if (ws.readyState === WebSocket.OPEN) {
            ws.send(message);
          }
        }
      }
    }
  }
}

function safeParse(data) {
  try {
    return JSON.parse(data);
  } catch {
    return null;
  }
}

async function authenticateWithToken(token, ws) {
  const tokenPermissions = await redisClient.get(token);
  if (tokenPermissions === null) {
    ws.send("unauthorized Access");
    ws.terminate();
    return null;
  }
  const authenticatedFreigaben = JSON.parse(tokenPermissions);
  await redisClient.del(token);
  return authenticatedFreigaben;
}

// ----------------------
// Connection handler
// ----------------------

wss.on("connection", async (ws, req) => {
  const params = url.parse(req.url, true).query;
  const access = typeof params.access === "string" ? params.access : "";
  let authenticatedFreigaben = null;
  ws.isAuthenticated = false;
  ws.access = access;
  if (access === 'token' && typeof params.token === "string") {
    authenticatedFreigaben = await authenticateWithToken(params.token, ws);

    if (!authenticatedFreigaben) return;

    ws.send("Authentifizierung mit Token erfolgreich");
    ws.isAuthenticated = true;
    ws.access = authenticatedFreigaben.access ?? authenticatedFreigaben?.type;
  }

  ws.authenticatedFreigaben = authenticatedFreigaben;
  ws.isAlive = true;

  clients.add(ws);

  addToGroup(ws, ws.access, ws.isAuthenticated);

  ws.on("pong", () => {
    ws.isAlive = true;
  });

  ws.on("message", (data) => {
    if (data.length > MAX_MESSAGE_SIZE) return;

    const msg = safeParse(data.toString());
    if (!msg || typeof msg !== "object") return;

    // Expected structure:
    // {
    //   type: string,
    //   payload: object
    // }

    const { type, payload } = msg;

    if (!type) return;

    switch (type) {
      case "DISPLAY_CONTROL":
        if (ws.authenticatedFreigaben.type === "displaycontrol") {
          sendToGroupsContaining("_display", {
            type: "UPDATE_DISPLAYCONTROL",
            payload
          });
        } else {
          ws.send("Unauthorized Request");
          ws.close(4003, "Unauthorized");
        }
        break;

      case "UPDATE_SCORE":
        if (ws.authenticatedFreigaben.type !== ("kampfrichter")) {
          ws.send("Unauthorized Request");
          ws.close(4003, "Unauthorized");
        };

        if (!payload || typeof payload !== "object") return;

        const geraetScreen = payload.geraet;
        if (!geraetScreen) return;

        if (ws.authenticatedFreigaben.access !== geraetScreen && ws.authenticatedFreigaben.access !== 'admin') {
          ws.send("Unauthorized Request");
          ws.close(4003, "Unauthorized");
        }

        sendToGroup(`${geraetScreen}_display`, {
          type: "UPDATE_SCORE",
          payload: payload.data
        });
        break;

      case "SELF":
        sendToSelf(ws, {
          type: "SELF",
          payload
        });
        break;

      case "AUDIO":
        if (ws.authenticatedFreigaben.type !== ("kampfrichter")) {
          ws.send("Unauthorized Request");
          ws.close(4003, "Unauthorized");
        };

        sendToGroup("audio", {
          type: "AUDIO",
          payload
        });
        break;

      case "KAMPFRICHTER_UPDATE":
        if (ws.authenticatedFreigaben.type !== ("kampfrichter")) {
          ws.send("Unauthorized Request");
          ws.close(4003, "Unauthorized");
        };

        if (!payload || typeof payload !== "object") return;

        const discipline = payload.discipline;
        if (!discipline) return;

        if (ws.authenticatedFreigaben.access !== discipline && ws.authenticatedFreigaben.access !== 'admin') {
          ws.send("Unauthorized Request");
          ws.close(4003, "Unauthorized");
        }

        sendToGroup(discipline, {
          type: "UPDATE",
          payload: payload
        }, true, ws);

        sendToGroup('admin', {
          type: "UPDATE",
          payload: payload
        }, true, ws);
        break;


      case "EINSTELLUNGEN_DISPLAY_UPDATE":
        if (ws.authenticatedFreigaben.type !== ("einstellungen")) {
          ws.send("Unauthorized Request");
          ws.close(4003, "Unauthorized");
        };

        if (!payload || typeof payload !== "object") return;

        const { key, value } = payload;
        if (!key) return;

        sendToGroupsContaining("_display", {
          type: "EINSTELLUNGEN_DISPLAY_UPDATE",
          payload: { key, value }
        });
        break;

      default:
        ws.send("Invalid Request");
    }
  });

  ws.on("close", () => {
    clients.delete(ws);
    removeFromGroup(ws, ws.access, ws.isAuthenticated);
  });

  ws.on("error", () => {
    clients.delete(ws);
    removeFromGroup(ws, ws.access, ws.isAuthenticated);
  });
});

// ----------------------
// Heartbeat (cleanup)
// ----------------------

setInterval(() => {
  for (const ws of clients) {
    if (!ws.isAlive) {
      ws.terminate();
      clients.delete(ws);
      removeFromGroup(ws, ws.access, ws.isAuthenticated);
      continue;
    }

    ws.isAlive = false;
    ws.ping();
  }
}, HEARTBEAT_INTERVAL);

// ----------------------

console.log(`WebSocket server running on port ${PORT}`);