LinkCloudDev/server/v3.js

import Net from "net";
import "dotenv/config";
import { setup as dbSetup, runPrepQuery, numRows } from "./Utility/db.js";
import { Err, Log, Debug, Warn } from "./Utility/loggerUtility.js";
import crypto from "crypto";
import { botSetup, client } from "./Utility/discordClient.js";
import { event } from "./Utility/eventHandler.js";
import fsConfig from "./config.json" assert { type: "json" };
import { loadModelFromDB, createNewLS } from "./Utility/lsModel.js";
import { listen } from "./webserver.js";
import { getUserFromJwt, setUserSocket, loadUsersFromDB } from "./Utility/userModel.js";
import { getLSModel } from "./Utility/lsModel.js";

/**
 * Application configuration.
 */
const token = fsConfig.token;

// Log initial message
Log("Connecting to MySQL DB...");
dbSetup(fsConfig.db);

// Event listener for successful database connection
event.on("DATABASE_CONNECTED", async () => {
  Log("Ready!");
  try {
    await loadUsersFromDB();
    await loadModelFromDB();
    botSetup();
    listen();
  } catch (error) {
    Err("Error during setup after database connection.", error);
  }
});

// Event listener for failed database connection
event.on("MYSQL_FAILED_TO_CONNECT", () => {
  Err("Failed to launch LinkCloud. Unable to connect to MySQL database. Please check database connection parameters and try again.");
  process.exit(1); // Exit with non-zero status for error indication
});

// Global error handlers
process.on("uncaughtException", (error) => {
  Err("Uncaught Exception:", error);
  //process.exit(1); // Optionally exit with non-zero status
});

process.on("unhandledRejection", (reason, promise) => {
  Err("Unhandled Rejection at Promise:", promise, "Reason:", reason);
  // Optionally exit process if desired:
  // process.exit(1);
});

// Server configuration
const server = Net.createServer();
const discordPacketBuffer = [];
const lsPacketBuffer = [];
const sockets = [];
const MAX_PACKET_BUFFER_SIZE = 1000; // Define constant for buffer size

// Check time difference
const checkTime = (theirTime) => {
  const myTime = Math.floor(Date.now() / 1000);
  return Number(theirTime - myTime);
};

/**
 * Generates a hash value for the given packet data using MD5 algorithm.
 * @param {object} packet - The packet data object containing type, metaData, payload, etc.
 * @returns {string} - The MD5 hash value of the concatenated packet data.
 */
const hashPacketData = (packet) => {
  let base = packet.type;
  base += packet.metaData.gameTime;
  base += packet.metaData.server;
  base += packet.payload.name.replace(/[\n\r]/g, '').trim();
  base += packet.payload.message.replace(/[\n\r]/g, '').trim();
  base += packet.metaData.platform;

  if (packet.payload?.linkshellname) {
    base += packet.payload.linkshellname;
  } else if (packet.payload?.area) {
    base += packet.payload.area;
  }

  return crypto.createHash("md5").update(base).digest("hex");
};

/**
 * Adds a packet to the Discord buffer if it does not already exist in the buffer.
 * If the buffer exceeds 1000 packets, the oldest packet is removed.
 * @param {Object} packet - The packet to add to the buffer.
 * @returns {boolean} - Returns true if packet is added, false if it already exists.
 */
const addToDiscordBuffer = (packet) => {
  const exists = discordPacketBuffer.some(p => p.hash === packet.hash);
  if (exists) {
    Debug(`${packet.hash} exists, skipping...`);
    return false;
  }

  Debug(`New hash created ${packet.hash}`);
  discordPacketBuffer.push(packet);

  if (discordPacketBuffer.length > MAX_PACKET_BUFFER_SIZE) {
    discordPacketBuffer.shift();
  }

  return true;
};

/**
 * Event listener for handling incoming data on a socket connection.
 * Parses the incoming data, processes the packets, and sends responses accordingly.
 * @param {Socket} sock - The socket object representing the connection.
 */
server.on("connection", (sock) => {
  Debug("CONNECTED: " + sock.remoteAddress + ":" + sock.remotePort);

  sock.on("data", async (data) => {
    const response = { error: false };

    try {
      const packets = data.toString().split("\n");
      for (const packetStr of packets) {
        if (packetStr) {
          const packet = JSON.parse(packetStr);
          const differential = checkTime(packet?.metaData?.clientTime);

          if (differential <= process.env.MAX_CLOCK_SYNC_MISMATCH_SECONDS) {
            response.type = packet?.type?.toUpperCase();
            response.packetId = packet?.packetId;

            switch (response.type) {
              case "HANDSHAKE":
                const authed = AuthenticateSocket(packet, sock);
                response.payload = authed ? "ACCEPTED" : "REJECTED";
                response.error = !authed;
                response.errorMsg = "AUTH_FAIL";
                response.errorDetails = "You shall not pass!";
                response.disconnect = true;
                Log(`[${sock.remoteAddress}] connection ${response.payload}.`);
                break;

              case "HEARTBEAT":
                response.payload = "PONG";
                break;

              case "LINKSHELL_MESSAGE":
                packet.hash = hashPacketData(packet);
                Log(`[${sock.remoteAddress}] LS_MESSAGE`)
                ProcessLSMessage(packet, sock);
                break;

              case "LINKSHELL_UPDATE":
                ProcessLSUpdate(packet, sock);
                Log(`[${sock.remoteAddress}] LS_UPDATE`)
                break;

              case "SHOUT":
                packet.hash = hashPacketData(packet);
                addToDiscordBuffer(packet);
                break;

              case "OTHER":
                // Handle other types
                break;

              case "ADD_LINKSHELL":
                Log(`[${sock.remoteAddress}] ADD_LS`)
                await ProcessAddLinkshell(packet, sock);
                break;

              default:
                response.error = true;
                response.errorMsg = "UNKNOWN_PACKET_TYPE";
                response.errorDetails = "UNKNOWN PACKET TYPE";
                break;
            }
          } else {
            response.error = true;
            response.errorMsg = "CLOCK_OUT_OF_SYNC";
            response.errorDetails = `This system and the server clocks are out of sync by ${differential} second(s).`;
          }

          if (sock) {
            writeToClientSocket(sock, JSON.stringify(response) + "\n\r");
          }

          if (response.error) {
            Debug("Error processing packet", packet);
            sock.destroy(); // Use destroy instead of resetAndDestroy
          }
        }
      }

      // Trace (`trace` level logging is not natively supported)
      //Debug(`TRACE: TO ${sock.remoteAddress} ${JSON.stringify(response)}`);
    } catch (ex) {
      Err("Unexpected packet format, unable to parse.");
      Debug(ex);
      Debug("Data:", data.toString());
    }
  });

  sock.on("timeout", () => {
    removeSocketFromPool(sock);
    Debug("TIMEOUT: " + sock.remoteAddress + " " + sock.remotePort);
  });

  sock.on("close", () => {
    removeSocketFromPool(sock);
    Debug("CLOSED: " + sock.remoteAddress + " " + sock.remotePort);
  });

  writeToClientSocket(sock, "CHALLENGE\n");
  sockets.push(sock);
});

/**
 * Removes a socket from the pool.
 * @param {Socket} sock - The socket to remove.
 */
const removeSocketFromPool = (sock) => {
  const index = sockets.findIndex((o) => o.remoteAddress === sock.remoteAddress && o.remotePort === sock.remotePort);
  if (index !== -1) sockets.splice(index, 1);
};

/**
 * Writes data to a client socket with retries.
 * @param {Socket} socket - The socket to write to.
 * @param {string} data - The data to write.
 * @param {number} [retry=0] - The retry count.
 */
const writeToClientSocket = (socket, data, retry = 0) => {
  if (!socket?.destroyed && socket?.readyState === 'open') {
    socket.write(data, (err) => {
      if (err) {
        Debug("Failed to write to socket", err);
      }
    });
  } else {
    if (socket?.destroyed) {
      Debug("The socket is dead and can't be written to");
    } else if (retry < 5) {
      Debug(`Retrying to send packet ${data}`);
      setTimeout(() => writeToClientSocket(socket, data, retry + 1), 1000);
    }
  }
};

/**
 * Processes a linkshell message packet.
 * @param {object} packet - The packet data.
 * @param {Socket} socket - The socket connection.
 * @returns {boolean} - Returns false if packet exists, true otherwise.
 */
const ProcessLSMessage = (packet, socket) => {
  const exists = lsPacketBuffer.some(p => p.hash === packet.hash);
  if (exists) {
    Debug(`${packet.hash} exists, skipping...`);
    return false;
  }

  Debug(`New hash created ${packet.hash}`);
  lsPacketBuffer.push(packet);

  if (lsPacketBuffer.length > MAX_PACKET_BUFFER_SIZE) {
    lsPacketBuffer.shift();
  }

  const linkshell = ProcessLSUpdate(packet, socket);
  if (!linkshell?.channels) return false;

  const re = /^(\[.+\] .+)/;
  if (!packet.payload.message.match(re)) {
    linkshell.webhookQueue.push({
      linkshell,
      packet,
    });
  }

  return true;
};

/**
 * Processes a linkshell update packet.
 * @param {object} packet - The packet data.
 * @param {Socket} socket - The socket connection.
 * @returns {object} - Returns the linkshell object.
 */
const ProcessLSUpdate = (packet, socket) => {
  const linkshell = getLSModel(packet.payload?.linkshellname, packet.metaData.platform, packet.metaData.server);

  if (linkshell) {
    linkshell.socket = socket;
    Log(`Linkshell "${packet.payload.linkshellname}" Registered By ${packet.metaData.character}`);
  }

  return linkshell;
};

/**
 * Sends a system message to the client.
 * @param {Socket} socket - The client socket.
 * @param {string} message - The message to send.
 * @param {boolean} [error=false] - Whether the message is an error.
 */
const sendMessageToClient = (socket, message, error = false) => {
  const packet = {
    type: "SYSTEM_MESSAGE",
    payload: {
      isError: error,
      message,
    },
  };

  if (socket) {
    writeToClientSocket(socket, JSON.stringify(packet) + "\n\r");
  }
};

/**
 * Sends a linkshell echo message to the client.
 * @param {Socket} socket - The client socket.
 * @param {string} message - The message to send.
 * @param {string} from - The sender of the message.
 * @param {string} ls - The linkshell name.
 */
const sendLSMessage = (socket, message, from, ls) => {
  const packet = {
    type: "LS_ECHO",
    payload: {
      from: from.trim(),
      message: message.trim(),
      linkshell: ls,
    },
  };

  if (socket) {
    writeToClientSocket(socket, JSON.stringify(packet) + "\n\r");
  }
};

/**
 * Authenticates a socket connection.
 * @param {object} packet - The packet data containing authentication info.
 * @param {Socket} socket - The socket connection.
 * @returns {boolean} - Returns true if authentication is successful, false otherwise.
 */
const AuthenticateSocket = (packet, socket) => {
  const authId = packet.payload.authId;
  const user = getUserFromJwt(authId);

  if (user) {
    setUserSocket(socket, user.userId);
    return true;
  } else {
    return false;
  }
};

/**
 * Processes a request to add a linkshell.
 * @param {object} packet - The packet data.
 * @param {Socket} socket - The socket connection.
 */
const ProcessAddLinkshell = async (packet, socket) => {
  const linkId = packet.payload.linkId.replace("\r", "").replace("\n", "");
  const serverId = packet.metaData.server;
  const lsName = packet.payload.lsName;

  try {
    const result = await runPrepQuery("SELECT * FROM pendinglinks WHERE linkId = ? LIMIT 0,1", [linkId]);
    if (numRows(result)) {
      const { ffxiver, userId } = result[0];
      const discordUser = client.users.cache.get(userId);

      Debug({ linkId, serverId, lsName, ffxiver, userId });

      try {
        const newLs = await createNewLS(lsName, serverId, ffxiver, userId);

        discordUser.send(
          `# Success!\nYour Linkshell ${lsName} has been added to LinkCloud!\n\n## What's Next?\n- Set up the chat echo channel in your discord server using the \`/lccreateecho\` command in the channel you want to use for the chat echo.\n- Encourage LS members to use the \`/lcjoin\` command to get started streaming data to LinkCloud. The more streamers you have, the more reliable the echo will be.`
        );

        Log(`New Linkshell "${lsName}" has been created!`);
        sendMessageToClient(socket, `Linkshell ${lsName} added successfully to LinkCloud. Check discord for further instruction.`);
      } catch (error) {
        Err(`Failed to create Linkshell "${lsName}".`, error);
        sendMessageToClient(socket, "An error occurred. This is most likely because this Linkshell already exists.", true);

        discordUser.send(
          `# Uh-oh!\nYour Linkshell ${lsName} seems to already exist in our database.\n\n## Need Help?\nYou can reach out to support via discord.\nhttps://discord.gg/n5VYHSQbhA`
        );
      }
    } else {
      sendMessageToClient(socket, "An error occurred. The supplied token is not valid.", true);
    }
  } catch (error) {
    Err("Database query failed during ProcessAddLinkshell.", error);
    sendMessageToClient(socket, "An error occurred during linkshell processing. Please try again later.", true);
  }
};

/**
 * Event listener for new Discord echo received.
 */
event.on('NEW_DISCORD_ECHO_RECEIVED', (message) => {
  const linkshell = getLSModel(message.lsName, message.platform, message.server);
    Log(`[${message.lsName} : ${message.server}] NEW_DISCORD_MESSAGE`)
  if (linkshell?.socket) {
    sendLSMessage(linkshell.socket, message.message, message.from, message.lsName);
  }
});

// Server listening
server.listen(process.env.TCP_LISTEN_PORT, () => {
  Log(`Server Bound to ${process.env.BIND_ADDRESS} on port ${process.env.TCP_LISTEN_PORT}.`);
});