dtp-base-v1/app/services/core-node.js

// core-node.js
// Copyright (C) 2022 DTP Technologies, LLC
// License: Apache-2.0

'use strict';

const uuidv4 = require('uuid').v4;
const fetch = require('node-fetch'); // jshint ignore:line

const mongoose = require('mongoose');

const CoreNode = mongoose.model('CoreNode');
const CoreUser = mongoose.model('CoreUser');
const CoreNodeRequest = mongoose.model('CoreNodeRequest');

const passport = require('passport');
const OAuth2Strategy = require('passport-oauth2');

const { SiteService, SiteError } = require('../../lib/site-lib');

class CoreAddress {

  constructor (host, port) {
    this.host = host;
    this.port = port;
  }

  parse (host) {
    const tokens = host.split(':');
    this.host = tokens[0];
    if (tokens[1]) {
      this.port = parseInt(tokens[1], 10);
    } else {
      this.port = 443;
    }
    return this;
  }
}

class CoreNodeService extends SiteService {

  constructor (dtp) {
    super(dtp, module.exports);
    this.populateCoreUser = [
      {
        path: 'core'
      },
    ];
  }

  async start ( ) {
    const cores = await this.getConnectedCores(null, true);
    cores.forEach((core) => this.registerPassportCoreOAuth2(core));
  }

  async attachExpressRoutes (router) {
    const cores = await this.getConnectedCores(null, true);
    cores.forEach((core) => {
      const coreAuthStrategyName = this.getCoreAuthStrategyName(core);
      const coreAuthUri = `/core/${core._id}`;
      const coreAuthCallbackUri = `/core/${core._id}/callback`;

      this.log.info('attach Core Auth route', {
        coreId: core._id,
        name: core.meta.name,
        strategy: coreAuthStrategyName,
        auth: coreAuthUri,
        callback: coreAuthCallbackUri,
      });
      router.get(
        coreAuthUri,
        (req, res, next) => {
          this.log.debug('Core auth request', { coreAuthStrategyName, clientId: core.oauth.clientId });
          return next();
        },
        passport.authenticate(coreAuthStrategyName),
      );

      router.get(
        coreAuthCallbackUri,
        (req, res, next) => {
          this.log.debug('Core auth callback', { strategy: coreAuthStrategyName });
          return next();
        },
        passport.authenticate(coreAuthStrategyName, { failureRedirect: '/' }),
        async (req, res, next) => {
          req.session.userType = 'Core';
          req.session.coreId = core._id;
          req.login(req.user, (error) => {
            if (error) {
              return next(error);
            }
            req.session.userType = 'Core';
            req.session.coreId = core._id;
              return res.redirect('/');
          });
        },
      );
    });
  }

  registerPassportCoreOAuth2 (core) {
    const { coreNode: coreNodeService } = this.dtp.services;
    const AUTH_SCHEME = coreNodeService.getCoreRequestScheme();

    const coreAuthStrategyName = this.getCoreAuthStrategyName(core);
    const authorizationHost = `${core.address.host}:${core.address.port}`;
    const authorizationURL = `${AUTH_SCHEME}://${authorizationHost}/oauth2/authorize`;
    const tokenURL = `${AUTH_SCHEME}://${authorizationHost}/oauth2/token`;
    const callbackURL = `${AUTH_SCHEME}://${process.env.DTP_SITE_DOMAIN}/auth/core/${core._id}/callback`;

    const coreAuthStrategy = new OAuth2Strategy(
      {
        authorizationURL,
        tokenURL,
        clientID: core.oauth.clientId.toString(),
        clientSecret: core.oauth.clientSecret,
        callbackURL,
      },
      async (accessToken, refreshToken, params, profile, cb) => {
        const NOW = new Date();
        try {
          const coreUserId = mongoose.Types.ObjectId(params.coreUserId);
          const user = await CoreUser.findOneAndUpdate(
            {
              core: core._id,
              coreUserId,
            },
            {
              $setOnInsert: {
                created: NOW,
                core: core._id,
                coreUserId,
                flags: {
                  isAdmin: false,
                  isModerator: false,
                },
                permissions: {
                  canLogin: true,
                  canChat: true,
                  canComment: true,
                  canReport: true,
                },
                optIn: {
                  system: true,
                  marketing: false,
                },
                theme: 'dtp-light',
                stats: {
                  uniqueVisitCount: 0,
                  totalVisitCount: 0,
                },
              },
              $set: {
                updated: NOW,
                username: params.username,
                username_lc: params.username_lc,
                displayName: params.displayName,
                bio: params.bio,
              },
            },
            {
              upsert: true,
              new: true,
            },
          );
          return cb(null, user.toObject());
        } catch (error) {
          return cb(error);
        }
      },
    );

    this.log.info('registering Core auth strategy', {
      name: coreAuthStrategyName,
      host: core.address.host,
      port: core.address.port,
      clientID: core.oauth.clientId.toString(),
      callbackURL,
    });
    passport.use(coreAuthStrategyName, coreAuthStrategy);
  }

  parseCoreAddress (host) {
    const address = new CoreAddress();
    return address.parse(host);
  }

  async getCoreByAddress (address) {
    const core = await CoreNode
      .findOne({
        'address.host': address.host,
        'address.port': address.port,
      })
      .lean();
    return core;
  }

  /**
   * First ensures that a record exists in the local database for the Core node.
   * Then, calls the node's info services to resolve more metadata about the
   * node, it's operation, policies, and available services.
   *
   * @param {String} host hostname and optional port number of Core node to be
   * resolved.
   * @returns CoreNode document describing the Core node.
   */
  async resolveCore (host) {
    const NOW = new Date();
    this.log.info('resolving Core node', { host });

    const address = this.parseCoreAddress(host);
    let core = await this.getCoreByAddress(address);
    if (!core) {
      core = new CoreNode();
      core.created = NOW;
      core.address.host = address.host;
      core.address.port = address.port;
      await core.save();
      core = core.toObject();
    }

    const txSite = await this.sendRequest(core, {
      method: 'GET',
      url: '/core/info/site',
    });
    const txPackage = await this.sendRequest(core, {
      method: 'GET',
      url: '/core/info/package',
    });

    await CoreNode.updateOne(
      { _id: core._id },
      {
        $set: {
          updated: NOW,
          'meta.name': txSite.response.site.name,
          'meta.description': txSite.response.site.description,
          'meta.domain': txSite.response.site.domain,
          'meta.domainKey': txSite.response.site.domainKey,
          'meta.version': txPackage.response.package.version,
          'meta.admin': txSite.response.site.admin,
          'meta.supportEmail': txSite.response.site.supportEmail,
        },
      },
    );

    core = await CoreNode.findOne({ _id: core._id }).lean();
    this.log.info('resolved Core node', { core });
    return core;
  }

  async getConstellationStats ( ) {
    const connectedCount = await CoreNode.countDocuments({ 'flags.isConnected': true });
    const pendingCount = await CoreNode.countDocuments({ 'flags.isConnected': false });
    const potentialReach = Math.round(Math.random() * 6000000);
    return { connectedCount, pendingCount, potentialReach };
  }

  async broadcast (request) {
    const results = [ ];
    await CoreNode
      .find()
      .cursor()
      .eachAsync(async (core) => {
        try {
          const response = await this.sendRequest(core, request);
          results.push({ coreId: core._id, request, response });
        } catch (error) {
          this.log.error('failed to send Core Node request', { core: core._id, request: request.url, error });
        }
      });
    return results;
  }

  getCoreAuthStrategyName (core) {
    return `dtp:${core.meta.domainKey}`;
  }

  getCoreRequestScheme ( ) {
    return process.env.DTP_CORE_AUTH_SCHEME || 'https';
  }

  getCoreRequestUrl (core, requestUrl) {
    const coreScheme = this.getCoreRequestScheme();
    return `${coreScheme}://${core.address.host}:${core.address.port}${requestUrl}`;
  }

  async sendRequest (core, request) {
    try {
      const req = new CoreNodeRequest();
      const options = {
        headers: {
          'Content-Type': 'application/json',
        },
      };

      req.created = new Date();
      req.core = core._id;
      if (request.tokenized) {
        req.token = {
          value: uuidv4(),
          claimed: false,
        };
        options.headers['X-DTP-Core-Token'] = req.token.value;
      }
      options.method = req.method = request.method || 'GET';
      req.url = request.url;
      await req.save();

      if (request.body) {
        options.body = JSON.stringify(request.body);
      }

      this.log.info('sending Core node request', { request: req });
      const requestUrl = this.getCoreRequestUrl(core, request.url);
      const response = await fetch(requestUrl, options);
      if (!response.ok) {
        throw new SiteError(response.status, response.statusText);
      }

      const json = await response.json();

      /*
       * capture a little inline health monitoring data, which can be used to
       * generate health alerts.
       */
      const DONE = new Date();
      const ELAPSED = DONE.valueOf() - req.created.valueOf();
      await CoreNodeRequest.updateOne(
        { _id: req._id },
        {
          $set: {
            'response.received': DONE,
            'response.elapsed': ELAPSED,
            'response.success': json.success,
          },
        },
      );

      this.log.info('Core node request complete', { request: req });
      return { request: req.toObject(), response: json };
    } catch (error) {
      this.log.error('failed to send Core Node request', { core: core._id, request: request.url, error });
      throw error;
    }
  }

  async setCoreOAuth2Credentials (core, credentials) {
    const { client } = credentials;
    this.log.info('updating Core Connect credentials', { core, client });
    await CoreNode.updateOne(
      { _id: core._id },
      {
        $set: {
          'flags.isConnected': true,
          'oauth.clientId': client._id,
          'oauth.clientSecret': client.secret,
          'oauth.scopes': client.scopes,
          'oauth.redirectUri': client.redirectUri,
        },
      },
    );
  }

  async getConnectedCores (pagination, withOAuth = false) {
    let q = CoreNode.find({ 'flags.isConnected': true });
    if (!withOAuth) {
      q = q.select('-oauth');
    }

    q = q.sort({ 'meta.domain': 1 });
    if (pagination) {
      q = q
        .skip(pagination.skip)
        .limit(pagination.cpp);
    }

    const cores = await q.lean();
    return cores;
  }

  async getUserByLocalId (userId) {
    const user = await CoreUser
      .findOne({ _id: userId })
      .select('+flags +permissions +optIn')
      .populate(this.populateCoreUser)
      .lean();
    return user;
  }
}

module.exports = {
  slug: 'core-node',
  name: 'coreNode',
  create: (dtp) => { return new CoreNodeService(dtp); },
};