dtp-liberty-links/lib/site-ioserver.js

// site-ioserver.js
// Copyright (C) 2021 Digital Telepresence, LLC
// License: Apache-2.0

'use strict';

const path = require('path');

const Redis = require('ioredis');

const mongoose = require('mongoose');
const ConnectToken = mongoose.model('ConnectToken');
const ChatMessage = mongoose.model('ChatMessage');

const striptags = require('striptags');
const marked = require('marked');

const { SiteLog } = require(path.join(__dirname, 'site-log'));

const Events = require('events');
class SiteIoServer extends Events {

  constructor (dtp) {
    super();
    this.dtp = dtp;
    this.log = new SiteLog(dtp, 'ioserver');
  }

  async start (httpServer) {
    const { domain: domainService } = this.dtp.services;

    const { Server } = require('socket.io');
    const { createAdapter } = require('@socket.io/redis-adapter');

    this.domain = await domainService.getByName(process.env.DTP_SITE_DOMAIN_KEY);
    if (!this.domain) {
      throw new Error('Must define site domain in MongoDB before starting environment');
    }

    this.markedRenderer = new marked.Renderer();
    this.markedRenderer.link = (href, title, text) => { return text; };
    this.markedRenderer.image = (href, title, text) => { return text; };
    this.markedRenderer.image = (href, title, text) => { return text; };

    const hljs = require('highlight.js');
    this.markedConfig = {
      renderer: this.markedRenderer,
      highlight: function(code, lang) {
        const language = hljs.getLanguage(lang) ? lang : 'plaintext';
        return hljs.highlight(code, { language }).value;
      },
      langPrefix: 'hljs language-', // highlight.js css expects a top-level 'hljs' class.
      pedantic: false,
      gfm: true,
      breaks: false,
      sanitize: false,
      smartLists: true,
      smartypants: false,
      xhtml: false,
    };

    const transports = ['websocket'/*, 'polling'*/];

    const pubClient = new Redis({
      host: process.env.REDIS_HOST,
      port: process.env.REDIS_PORT,
      password: process.env.REDIS_PASSWORD,
      key: process.env.REDIS_PREFIX,
    });
    pubClient.on('error', this.onRedisError.bind(this));

    const subClient = pubClient.duplicate();
    subClient.on('error', this.onRedisError.bind(this));

    const adapter = createAdapter(pubClient, subClient);
    this.io = new Server(httpServer, { adapter, transports });

    this.io.on('connection', this.onSocketConnect.bind(this));
  }

  async onRedisError (error) {
    this.log.error('Redis error', { error });
  }

  async stop ( ) {

  }

  async onSocketConnect (socket) {
    this.log.debug('socket connection', { sid: socket.id });
    const token = await ConnectToken.findOne({ token: socket.handshake.auth.token }).populate('user').lean();
    if (!token) {
      this.log.alert('rejecting invalid socket token', {
        sid: socket.sid,
        handshake: socket.handshake,
      });
      socket.close();
      return;
    }

    if (token.claimed) {
      this.log.alert('rejecting use of claimed connect token', {
        sid: socket.id,
        handshake: socket.handshake,
      });
      socket.close();
      return;
    }

    await ConnectToken.updateOne(
      { _id: token._id },
      { $set: { claimed: new Date() } },
    );
    this.log.debug('token claimed', {
      sid: socket.id,
      token: socket.handshake.auth.token,
      user: token.user._id,
    });

    const session = {
      user: {
        _id: token.user._id,
        created: token.user.created,
        username: token.user.username,
        displayName: token.user.displayName,
      },
      socket,
    };

    session.onSocketDisconnect = this.onSocketDisconnect.bind(this, session);
    session.onJoinChannel = this.onJoinChannel.bind(this, session);
    session.onLeaveChannel = this.onLeaveChannel.bind(this, session);
    session.onUserChat = this.onUserChat.bind(this, session);

    socket.on('disconnect', session.onSocketDisconnect);
    socket.on('join', session.onJoinChannel);
    socket.on('leave', session.onLeaveChannel);
    socket.on('user-chat', session.onUserChat);

    socket.emit('authenticated', {
      message: 'token verified',
      user: session.user,
    });
  }

  async onSocketDisconnect (session, reason) {
    this.log.debug('socket disconnect', { sid: session.socket.id, user: session.user._id, reason });
    session.socket.off('disconnect', session.onSocketDisconnect);
    session.socket.off('join', session.onJoinChannel);
    session.socket.off('leave', session.onLeaveChannel);
  }

  async onJoinChannel (session, message) {
    const { channelId } = message;
    this.log.debug('socket joins channel', { sid: session.socket.id, user: session.user._id, channelId });
    session.socket.join(channelId);
    session.socket.emit('join-result', { channelId });
  }

  async onLeaveChannel (session, message) {
    const { channelId } = message;
    this.log.debug('socket leaves channel', { sid: session.socket.id, user: session.user._id, channelId });
    session.socket.leave(channelId);
  }

  async onUserChat (session, message) {
    const { channel: channelService } = this.dtp.services;
    const { channelId } = message;

    if (!message.content || (message.content.length === 0)) {
      return;
    }

    const channel = await channelService.getChannelById(channelId);
    if (!channel) {
      return;
    }

    const stickers = this.findStickers(message.content);
    stickers.forEach((sticker) => {
      const re = new RegExp(`:${sticker}:`, 'gi');
      message.content = message.content.replace(re, '').trim();
    });

    message.content = striptags(message.content);

    await ChatMessage.create({
      created: new Date(),
      domain: this.domain._id,
      channel: mongoose.Types.ObjectId(channelId),
      episode: channel.liveEpisode,
      author: session.user._id,
      content: message.content,
      stickers,
    });

    const renderedContent = marked(message.content, this.markedConfig);

    const payload = {
      user: {
        _id: session.user._id,
        username: session.user.username,
      },
      content: renderedContent,
      stickers,
    };

    session.socket.to(channelId).emit('user-chat', payload);
    session.socket.emit('user-chat', payload);
  }

  findStickers (content) {
    const tokens = content.split(' ');
    const stickers = [ ];
    tokens.forEach((token) => {
      if ((token[0] !== ':') || (token[token.length -1 ] !== ':')) {
        return;
      }

      token = token.slice(1, token.length - 1 ).toLowerCase();
      if (token.includes('/') || token.includes(':') || token.includes(' ')) {
        return; // trimmed token includes invalid characters
      }

      this.log.debug('found sticker request', { token });
      if (!stickers.includes(token)) {
        stickers.push(striptags(token));
      }
    });
    return stickers;
  }
}

module.exports.SiteIoServer = SiteIoServer;