bodyshop/server.js

// Load environment variables THIS MUST BE AT THE TOP
const path = require("path");
require("dotenv").config({
  path: path.resolve(process.cwd(), `.env.${process.env.NODE_ENV || "development"}`)
});

const cors = require("cors");
const http = require("http");
const Redis = require("ioredis");
const express = require("express");
const bodyParser = require("body-parser");
const compression = require("compression");
const cookieParser = require("cookie-parser");
const { Server } = require("socket.io");
const { createAdapter } = require("@socket.io/redis-adapter");
const { instrument } = require("@socket.io/admin-ui");
const { isString, isEmpty } = require("lodash");

const logger = require("./server/utils/logger");
const { applyRedisHelpers } = require("./server/utils/redisHelpers");
const { applyIOHelpers } = require("./server/utils/ioHelpers");
const { redisSocketEvents } = require("./server/web-sockets/redisSocketEvents");
const { ElastiCacheClient, DescribeCacheClustersCommand } = require("@aws-sdk/client-elasticache");
const { InstanceRegion } = require("./server/utils/instanceMgr");
const StartStatusReporter = require("./server/utils/statusReporter");

const cleanupTasks = [];
let isShuttingDown = false;
const CLUSTER_RETRY_BASE_DELAY = 100;
const CLUSTER_RETRY_MAX_DELAY = 5000;
const CLUSTER_RETRY_JITTER = 100;

/**
 * CORS Origin for Socket.IO
 * @type {string[][]}
 */
const SOCKETIO_CORS_ORIGIN = [
  "https://test.imex.online",
  "https://www.test.imex.online",
  "http://localhost:3000",
  "https://localhost:3000",
  "https://imex.online",
  "https://www.imex.online",
  "https://romeonline.io",
  "https://www.romeonline.io",
  "https://test.romeonline.io",
  "https://www.test.romeonline.io",
  "https://beta.romeonline.io",
  "https://www.beta.romeonline.io",
  "https://beta.test.imex.online",
  "https://www.beta.test.imex.online",
  "https://beta.imex.online",
  "https://www.beta.imex.online",
  "https://old.imex.online",
  "https://www.old.imex.online",
  "https://wsadmin.imex.online",
  "https://www.wsadmin.imex.online"
];

const SOCKETIO_CORS_ORIGIN_DEV = ["http://localhost:3333", "https://localhost:3333"];

/**
 * Middleware for Express app
 * @param app
 */
const applyMiddleware = ({ app }) => {
  app.use(compression());
  app.use(cookieParser());
  app.use(bodyParser.json({ limit: "50mb" }));
  app.use(bodyParser.urlencoded({ limit: "50mb", extended: true }));
  app.use(cors({ credentials: true, exposedHeaders: ["set-cookie"] }));

  // Helper middleware
  app.use((req, res, next) => {
    req.logger = logger;
    next();
  });
};

/**
 * Route groupings for Express app
 * @param app
 */
const applyRoutes = ({ app }) => {
  app.use("/", require("./server/routes/miscellaneousRoutes"));
  app.use("/notifications", require("./server/routes/notificationsRoutes"));
  app.use("/render", require("./server/routes/renderRoutes"));
  app.use("/mixdata", require("./server/routes/mixDataRoutes"));
  app.use("/accounting", require("./server/routes/accountingRoutes"));
  app.use("/qbo", require("./server/routes/qboRoutes"));
  app.use("/media", require("./server/routes/mediaRoutes"));
  app.use("/sms", require("./server/routes/smsRoutes"));
  app.use("/job", require("./server/routes/jobRoutes"));
  app.use("/scheduling", require("./server/routes/schedulingRoutes"));
  app.use("/utils", require("./server/routes/utilRoutes"));
  app.use("/data", require("./server/routes/dataRoutes"));
  app.use("/adm", require("./server/routes/adminRoutes"));
  app.use("/tech", require("./server/routes/techRoutes"));
  app.use("/intellipay", require("./server/routes/intellipayRoutes"));
  app.use("/cdk", require("./server/routes/cdkRoutes"));
  app.use("/csi", require("./server/routes/csiRoutes"));
  app.use("/payroll", require("./server/routes/payrollRoutes"));

  // Default route for forbidden access
  app.get("/", (req, res) => {
    res.status(200).send("Access Forbidden.");
  });
};

/**
 * Fetch Redis nodes from AWS ElastiCache
 * @returns {Promise<string[]>}
 */
const getRedisNodesFromAWS = async () => {
  const client = new ElastiCacheClient({
    region: InstanceRegion()
  });

  const params = {
    ReplicationGroupId: process.env.REDIS_CLUSTER_ID,
    ShowCacheNodeInfo: true
  };

  try {
    // Fetch the cache clusters associated with the replication group
    const command = new DescribeCacheClustersCommand(params);
    const response = await client.send(command);
    const cacheClusters = response.CacheClusters;

    return cacheClusters.flatMap((cluster) =>
      cluster.CacheNodes.map((node) => `${node.Endpoint.Address}:${node.Endpoint.Port}`)
    );
  } catch (err) {
    logger.log(`Error fetching Redis nodes from AWS: ${err.message}`, "ERROR", "redis", "api");
    throw err;
  }
};

/**
 * Connect to Redis Cluster
 * @returns {Promise<unknown>}
 */
const connectToRedisCluster = async () => {
  let redisServers;

  if (isString(process.env?.REDIS_CLUSTER_ID) && !isEmpty(process.env?.REDIS_CLUSTER_ID)) {
    // Fetch Redis nodes from AWS if AWS environment variables are present
    redisServers = await getRedisNodesFromAWS();
  } else {
    // Use the Dockerized Redis cluster in development
    if (isEmpty(process.env?.REDIS_URL) || !isString(process.env?.REDIS_URL)) {
      logger.log(`No or Malformed REDIS_URL present.`, "ERROR", "redis", "api");
      process.exit(1);
    }
    try {
      redisServers = JSON.parse(process.env.REDIS_URL);
    } catch (error) {
      logger.log(`Failed to parse REDIS_URL: ${error.message}. Exiting...`, "ERROR", "redis", "api");
      process.exit(1);
    }
  }

  const clusterRetryStrategy = (times) => {
    const delay =
      Math.min(CLUSTER_RETRY_BASE_DELAY + times * 50, CLUSTER_RETRY_MAX_DELAY) + Math.random() * CLUSTER_RETRY_JITTER;
    logger.log(`Redis cluster not yet ready. Retrying in ${delay.toFixed(2)}ms`, "WARN", "redis", "api");
    return delay;
  };

  const redisCluster = new Redis.Cluster(redisServers, {
    clusterRetryStrategy,
    enableAutoPipelining: true,
    enableReadyCheck: true,
    redisOptions: {
      // connectTimeout: 10000, // Timeout for connecting in ms
      // idleTimeoutMillis: 30000, // Close idle connections after 30s
      // maxRetriesPerRequest: 5 // Retry a maximum of 5 times per request
    }
  });

  return new Promise((resolve, reject) => {
    redisCluster.on("ready", () => {
      logger.log(`Redis cluster connection established.`, "INFO", "redis", "api");
      resolve(redisCluster);
    });

    redisCluster.on("error", (err) => {
      logger.log(`Redis cluster connection failed: ${err.message}`, "ERROR", "redis", "api");
      reject(err);
    });
  });
};

/**
 * Apply Redis to the server
 * @param server
 * @param app
 */
const applySocketIO = async ({ server, app }) => {
  const redisCluster = await connectToRedisCluster();

  // Handle errors
  redisCluster.on("error", (err) => {
    logger.log(`Redis ERROR`, "ERROR", "redis", "api");
  });

  const pubClient = redisCluster;
  const subClient = pubClient.duplicate();

  pubClient.on("error", (err) => logger.log(`Redis pubClient error: ${err}`, "ERROR", "redis"));
  subClient.on("error", (err) => logger.log(`Redis subClient error: ${err}`, "ERROR", "redis"));

  process.on("SIGINT", async () => {
    logger.log("Closing Redis connections...", "INFO", "redis", "api");
    try {
      await Promise.all([pubClient.disconnect(), subClient.disconnect()]);
      logger.log("Redis connections closed. Process will exit.", "INFO", "redis", "api");
    } catch (error) {
      logger.log(`Error closing Redis connections: ${error.message}`, "ERROR", "redis", "api");
    }
  });

  const ioRedis = new Server(server, {
    path: "/wss",
    adapter: createAdapter(pubClient, subClient),
    cors: {
      origin:
        process.env?.NODE_ENV === "development"
          ? [...SOCKETIO_CORS_ORIGIN, ...SOCKETIO_CORS_ORIGIN_DEV]
          : SOCKETIO_CORS_ORIGIN,
      methods: ["GET", "POST"],
      credentials: true,
      exposedHeaders: ["set-cookie"]
    }
  });

  if (isString(process.env.REDIS_ADMIN_PASS) && !isEmpty(process.env.REDIS_ADMIN_PASS)) {
    logger.log(`Initializing Redis Admin UI....`, "INFO", "redis", "api");
    instrument(ioRedis, {
      auth: {
        type: "basic",
        username: "admin",
        password: process.env.REDIS_ADMIN_PASS
      },

      mode: process.env.REDIS_ADMIN_MODE || "development"
    });
  }

  const io = new Server(server, {
    path: "/ws",
    cors: {
      origin: SOCKETIO_CORS_ORIGIN,
      methods: ["GET", "POST"],
      credentials: true,
      exposedHeaders: ["set-cookie"]
    }
  });

  const api = {
    pubClient,
    io,
    ioRedis,
    redisCluster
  };

  app.use((req, res, next) => {
    Object.assign(req, api);

    next();
  });

  Object.assign(module.exports, api);

  return api;
};

/**
 * Main function to start the server
 * @returns {Promise<void>}
 */
const main = async () => {
  const app = express();
  const port = process.env.PORT || 5000;

  const server = http.createServer(app);

  const { pubClient, ioRedis } = await applySocketIO({ server, app });
  const redisHelpers = applyRedisHelpers({ pubClient, app, logger });
  const ioHelpers = applyIOHelpers({ app, redisHelpers, ioRedis, logger });

  // Legacy Socket Events
  require("./server/web-sockets/web-socket");

  applyMiddleware({ app });
  applyRoutes({ app });
  redisSocketEvents({ io: ioRedis, redisHelpers, ioHelpers, logger });

  const StatusReporter = StartStatusReporter();
  registerCleanupTask(async () => {
    StatusReporter.end();
  });

  // Add SIGTERM signal handler
  process.on("SIGTERM", handleSigterm);
  process.on("SIGINT", handleSigterm); // Optional: Handle Ctrl+C

  try {
    await server.listen(port);
    logger.log(`Server started on port ${port}`, "INFO", "api");
  } catch (error) {
    logger.log(`Server failed to start on port ${port}`, "ERROR", "api", error);
  }
};

// Start server
main().catch((error) => {
  logger.log(`Main-API-Error: Something was not caught in the application.`, "error", "api", null, {
    error: error.message,
    errorjson: JSON.stringify(error)
  });
  // Note: If we want the app to crash on all uncaught async operations, we would
  // need to put a `process.exit(1);` here
});

// Register a cleanup task
function registerCleanupTask(task) {
  cleanupTasks.push(task);
}

// SIGTERM handler
async function handleSigterm() {
  if (isShuttingDown) {
    logger.log("sigterm-api", "WARN", null, null, { message: "Shutdown already in progress, ignoring signal." });
    return;
  }

  isShuttingDown = true;

  logger.log("sigterm-api", "WARN", null, null, { message: "SIGTERM Received. Starting graceful shutdown." });

  try {
    for (const task of cleanupTasks) {
      logger.log("sigterm-api", "WARN", null, null, { message: `Running cleanup task: ${task.name}` });

      await task();
    }
    logger.log("sigterm-api", "WARN", null, null, { message: `All cleanup tasks completed.` });
  } catch (error) {
    logger.log("sigterm-api-error", "ERROR", null, null, { message: error.message, stack: error.stack });
  }

  process.exit(0);
}