server.js

"use strict";

const { Router } = require('express');
const { type } = require('os');
const { Transform } = require('stream');

class HTTPSockStream extends Transform {
  _transform(chunk, _enc, callback) {
    const size = chunk.length.toString(16);
    this.push(size + '\r\n');
    this.push(chunk);
    this.push('\r\n');
    callback();
  };
};

/**
 * Returns the httpsock server router.
 *
 * @param {Object} [options]
 * @param {number} [options.maxBody='10mb'] - Maximum allowed body size for POST requests.
 * @param {boolean} [options.auth=false] - Whether to require authentication for clients and broadcasters.
 * @param {function|Array} [options.clients=[]] - List of allowed client credentials. Each item can be an object with `username` and `password` properties, an array of `[username, password]`, or a string in the format 'username:password'. Alternatively, can be a function that returns false or the authenticated username when given parameters: username, password.
 * @param {function|Array} [options.broadcasts=[]] - List of allowed broadcaster credentials. Each item can be an object with `username` and `password` properties, an array of `[username, password]`, or a string in the format 'username:password'. Alternatively, can be a function that returns false or the authenticated username when given parameters: username, password.
 * @param {function} [options.callback=((authenticatedAs, message, req) => console.log(`←→ ${authenticatedAs}:`, message))] - Callback function with successful data sent to the server. Receives three parameters: authenticatedAs, message, req.
 * @param {string|function} [options.welcome] - Optional welcome message to send to clients upon connection. Can be a string or a function that returns a string (or undefined to send nothing) when given the parameter: username.
 * @param {function} [options.error=((err) => console.error('HTTPSockServer error:', (err && err.stack) ? err.stack : err))] - Callback function for errors. Receives one parameter: err.
 * @returns {Router}
 */
function HTTPSockServer(options = {}) {
  const router = Router();
  const requireAuth = Boolean(options.auth);
  const allowedClients = options.clients || [];
  const allowedBroadcasts = options.broadcasts || [];
  const callback = options.callback || ((authenticatedAs, message) => console.log(`←→ ${authenticatedAs}:`, message));
  const errorHandler = options.error || ((err) => console.error('HTTPSockServer error:', (err && err.stack) ? err.stack : err));

  const parseBasic = (header) => {
    if (!header || (typeof header !== 'string')) return null;
    const parts = header.split(' ');
    if (parts.length !== 2) return null;
    if (parts[0] !== 'Basic') return null;
    try {
      const decoded = Buffer.from(parts[1], 'base64').toString();
      const sep = decoded.indexOf(':');
      if (sep === -1) return null;
      return { username: decoded.slice(0, sep), password: decoded.slice(sep + 1) };
    } catch (e) {
      errorHandler(e);
      return null;
    };
  };

  const matches = async (credentials, list) => {
    if (!credentials) return false;
    if (typeof list === 'function') {
      try {
        return await list(credentials.username, credentials.password);
      } catch (e) {
        errorHandler(e);
      };
    } else {
      for (const item of list) {
        if (!item) continue;
        if (item.username && item.password) {
          if ((item.username === credentials.username) && (item.password === credentials.password)) return credentials.username;
        } else if (Array.isArray(item) && (item.length >= 2)) {
          if ((item[0] === credentials.username) && (item[1] === credentials.password)) return credentials.username;
        } else if (typeof item === 'string') {
          const itemCredentials = item.split(':');
          if ((itemCredentials[0] === credentials.username) && (itemCredentials[1] === credentials.password)) return credentials.username;
        };
      };
    };
    return false;
  };

  router.use(require('express').raw({
    type: '*/*',
    limit: options.maxBody || '10mb',
  }));

  router.connections = [];

  const enqueueForConnection = (connection, message) => {
    if (!connection || connection.closed || connection.stream.destroyed || connection.stream.writableEnded) return;
    var buffer;
    if (Buffer.isBuffer(message)) {
      buffer = message;
    } else if (typeof message === 'object') {
      try {
        buffer = Buffer.from(JSON.stringify(message));
      } catch (e) {
        buffer = Buffer.from(String(message));
      };
    } else {
      buffer = Buffer.from(String(message));
    };
    connection.queue.push(buffer);
    drainConnection(connection);
  };

  const drainConnection = (connection) => {
    if (!connection || connection.closed || connection.stream.destroyed || connection.stream.writableEnded) return;
    while (connection.queue.length) {
      const message = connection.queue[0];
      try {
        const ok = connection.stream.write(Buffer.from(message));
        if (!ok) {
          connection.stream.once('drain', () => { if (!connection.closed) drainConnection(connection); });
          return;
        };
      } catch (err) {
        if (err && (err.code === 'ERR_STREAM_WRITE_AFTER_END')) return;
        errorHandler(e);
        return;
      };
      connection.queue.shift();
      if (connection.closed || connection.stream.destroyed || connection.stream.writableEnded) return;
    };
  };

  router.sendTo = (username, message) => {
    for (const connection of router.connections) {
      if (connection.closed) continue;
      if (username && (connection.username !== username)) continue;
      enqueueForConnection(connection, message);
    };
  };

  router.send = (message) => router.sendTo(undefined, message);

  router.get('/', async (req, res) => {
    if (requireAuth) {
      const credentials = parseBasic(req.headers.authorization);
      if (await matches(credentials, allowedClients) === false) {
        res.set('WWW-Authenticate', 'Basic realm="httpsock"');
        return res.status(401).type('text').send('Unauthorized');
      };
    };
    res.set({
      'Content-Type': 'application/octet-stream',
      'Transfer-Encoding': 'chunked',
      Connection: 'keep-alive',
    });
    const stream = new HTTPSockStream();
    stream.pipe(res);

    const connection = {
      stream,
      closed: false,
      queue: [],
      username: undefined,
      room: undefined,
    };

    stream.on('error', (err) => {
      if (err && (err.code === 'ERR_STREAM_WRITE_AFTER_END')) return;
      console.error('HTTPSockStream error:', (err && err.stack) ? err.stack : err);
      try {
        stream.destroy();
      } catch (e) {
        errorHandler(e);
      };
    });

    req.on('close', () => {
      connection.closed = true;
      try {
        if (!stream.writableEnded && !stream.destroyed) stream.end();
      } catch (e) {
        errorHandler(e);
      };
      const idx = router.connections.indexOf(connection);
      if (idx !== -1) router.connections.splice(idx, 1);
    });

    try {
      const credentials = parseBasic(req.headers.authorization);
      const matched = requireAuth ? await matches(credentials, allowedClients) : (credentials ? (credentials.username || undefined) : undefined);
      connection.username = matched || req.headers['x-httpsock-username'] || (req.query && req.query.username) || 'client';
    } catch (e) {
      connection.username = req.headers['x-httpsock-username'] || (req.query && req.query.username) || 'client';
    };
    connection.send = (message) => enqueueForConnection(connection, message);
    router.connections.push(connection);
    if (options.welcome) {
      try {
        const welcomeMsg = (typeof options.welcome === 'function') ? await options.welcome(connection.username) : options.welcome;
        if (welcomeMsg !== undefined) enqueueForConnection(connection, welcomeMsg);
      } catch (e) {
        errorHandler(e);
      };
    };
  });

  router.post('/', async (req, res) => {
    var authenticatedAs = 'broadcaster';
    if (requireAuth) {
      const credentials = parseBasic(req.headers.authorization);
      authenticatedAs = await matches(credentials, allowedBroadcasts);
      if (authenticatedAs === false) {
        res.set('WWW-Authenticate', 'Basic realm="httpsock"');
        return res.status(401).type('text').send('Unauthorized');
      };
    };
    const body = req.body;
    const contentType = req.headers['content-type'] || '';
    const targetUser = req.headers['x-httpsock-username'];
    const noBroadcastHeader = (req.headers['x-httpsock-no-broadcast'] || '').toString().toLowerCase();
    const noBroadcastQuery = (req.query && (req.query.no_broadcast || req.query.noBroadcast)) ? String(req.query.no_broadcast || req.query.noBroadcast).toString().toLowerCase() : undefined;
    const suppressBroadcast = (noBroadcastHeader === '1' || noBroadcastHeader === 'true' || noBroadcastHeader === 'yes' || noBroadcastQuery === '1' || noBroadcastQuery === 'true' || noBroadcastQuery === 'yes');
    var callbackString = '';
    var broadcastReturn = null;
    if (Buffer.isBuffer(body)) {
      if (contentType && ((contentType.indexOf('application/json') !== -1) || ((contentType.indexOf('text/') !== -1)))) {
        callbackString = body.toString();
      } else if (contentType) {
        callbackString = `${contentType} (${body.length} bytes)`;
      } else {
        callbackString = `${body.length} bytes`;
      };
      broadcastReturn = await callback(authenticatedAs, callbackString);
      if (!suppressBroadcast) {
        if (targetUser) {
          router.sendToUser(targetUser, body);
        } else {
          for (const connection of router.connections) enqueueForConnection(connection, body);
        };
      };
    } else {
      callbackString = (body === undefined || body === null) ? '' : String(body);
      broadcastReturn = await callback(authenticatedAs, callbackString);
      const buffer = Buffer.from(callbackString);
      if (!suppressBroadcast) {
        if (targetUser) {
          router.sendToUser(targetUser, buffer);
        } else {
          for (const connection of router.connections) enqueueForConnection(connection, buffer);
        };
      };
    };
    if (broadcastReturn != null) {
      try {
        if (typeof broadcastReturn === 'object') {
          return res.json(broadcastReturn);
        } else {
          return res.type('text').send(String(broadcastReturn));
        };
      } catch (e) {
        errorHandler(e);
        return res.type('text').send('OK');
      };
    } else {
      return res.type('text').send('OK');
    };
  });

  return router;
};

module.exports = HTTPSockServer;