dtp-base-v1/app/services/hive.js

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

'use strict';

const mongoose = require('mongoose');

const UserSubscription = mongoose.model('UserSubscription');
const UserNotification = mongoose.model('UserSubscription');

const KaleidoscopeEvent = mongoose.model('KaleidoscopeEvent');

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

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

class HiveService extends SiteService {

  constructor (dtp) {
    super(dtp, module.exports);
  }

  async start ( ) {
    const { oauth2: oauth2Service } = this.dtp.services;

    this.eventHandlers = {
      onOAuth2RemoveClient: this.onOAuth2RemoveClient.bind(this),
    };

    oauth2Service.on(oauth2Service.getEventName('client.remove'), this.eventHandlers.onOAuth2RemoveClient);
  }

  async stop ( ) {
    const { oauth2: oauth2Service } = this.dtp.services;

    oauth2Service.off(oauth2Service.getEventName('client.remove'), this.eventHandlers.onOAuth2RemoveClient);
    delete this.eventHandlers.onOAuth2RemoveClient;

    delete this.eventHandlers;
  }

  async subscribe (user, client, emitterId) {
    await UserSubscription.updateOne(
      { user: user._id },
      {
        $addToSet: {
          subscriptions: {
            client: client._id,
            emitterId,
          },
        },
      },
      {
        upsert: true,
      },
    );
  }

  async unsubscribe (user, subscription) {
    await UserSubscription.updateOne(
      { user: user._id },
      { $pull: { subscriptions: subscription } },
    );
  }

  extractHashtags (content) {
    const hashtags = content
      .split(/ \r\n/g)
      .filter((tag) => tag[0] === '#')
      .map((tag) => slug(tag.slice(1)))
    ;
    return hashtags;
  }

  extractLinks (content) {
    let links = content
      .split(/( |\r|\n)/g)
      .filter((tag) => {
        const test = tag.trim().toLowerCase();
        return test.startsWith('http://') || test.startsWith('https://');
      });

    return links;
  }

  async resolveLink (author, url) {
    const jobData = {
      authorType: author.type,
      author: author._id,
      url,
    };
    this.log.info('creating job to resolve link', { jobData });
    await this.resolver.add('resolve-link', jobData);
  }

  async processKaleidoscopeEvent (eventDefinition) {
    const {
      userNotification: userNotificationService,
      oauth2: oauth2Service,
    } = this.dtp.services;

    const client = await oauth2Service.getClientByDomainKey(eventDefinition.source.site.domainKey);
    if (!client) {
      throw new SiteError(403, 'Unknown client domain key');
    }

    const event = await this.createKaleidoscopeEvent(eventDefinition, client);
    await UserSubscription
      .find({
        'subscriptions.client': client._id,
        'subscriptions.emitterId': eventDefinition.source.emitter._id,
      })
      .select('-subscriptions')
      .cursor()
      .eachAsync(async (subscription) => {
        await userNotificationService.create(subscription.user, event);
      }, 3);

    this.emit('kaleidoscope:event', event, client);
  }

  async createKaleidoscopeEvent (eventDefinition, sourceClient) {
    const NOW = new Date();

    /*
     * Validate general notification data
     */

    if (!eventDefinition.action) {
      throw new SiteError(406, 'Missing action');
    }
    if (!eventDefinition.content) {
      throw new SiteError(406, 'Missing content');
    }
    if (!eventDefinition.href) {
      throw new SiteError(406, 'Missing href');
    }

    /*
     * Validate source data
     */

    if (!eventDefinition.source) {
      throw new SiteError(406, 'Missing source information');
    }

    /*
     * Validate source site
     */

    if (!eventDefinition.source.site) {
      throw new SiteError(406, 'Missing source site information');
    }
    if (!eventDefinition.source.site.name) {
      throw new SiteError(406, 'Missing source site name');
    }
    if (!eventDefinition.source.site.description) {
      throw new SiteError(406, 'Missing source site description');
    }
    if (!eventDefinition.source.site.domain) {
      throw new SiteError(406, 'Missing source site domain');
    }
    if (!eventDefinition.source.site.domainKey) {
      throw new SiteError(406, 'Missing source site domain key');
    }
    if (!eventDefinition.source.site.company) {
      throw new SiteError(406, 'Missing source site company name');
    }
    if (!eventDefinition.source.site.coreAuth ||
        !eventDefinition.source.site.coreAuth.scopes ||
        !Array.isArray(eventDefinition.source.site.coreAuth.scopes) ||
        (eventDefinition.source.site.coreAuth.scopes.length === 0)) {
      throw new SiteError(406, 'Missing source site Core auth or scope information');
    }

    /*
     * Validate source package
     */

    if (!eventDefinition.source.pkg) {
      throw new SiteError(406, 'Missing source package information');
    }
    if (!eventDefinition.source.pkg.name) {
      throw new SiteError(406, 'Missing source package name');
    }
    if (!eventDefinition.source.pkg.version) {
      throw new SiteError(406, 'Missing source package version');
    }

    /*
     * Validate source emitter
     */

    if (!eventDefinition.source.emitter) {
      throw new SiteError(406, 'Missing source emitter information');
    }
    if (!eventDefinition.source.emitter.emitterId) {
      throw new SiteError(406, 'Missing source emitter ID');
    }
    if (!eventDefinition.source.emitter.username) {
      throw new SiteError(406, 'Missing source emitter username');
    }
    if (!eventDefinition.source.emitter.href) {
      throw new SiteError(406, 'Missing source emitter href');
    }

    /*
     * Create the KaleidoscopeEvent document
     */

    const event = new KaleidoscopeEvent();
    if (eventDefinition.created) {
      event.created = new Date(eventDefinition.created);
    } else {
      event.created = NOW;
    }

    if (eventDefinition.recipientType && eventDefinition.recipient) {
      event.recipientType = eventDefinition.recipientType;
      event.recipient = mongoose.Types.ObjectId(eventDefinition.recipient);
    }

    event.action = striptags(eventDefinition.action.trim().toLowerCase());
    if (eventDefinition.label) {
      event.label = striptags(eventDefinition.label.trim());
    }
    event.content = striptags(eventDefinition.content.trim());
    event.href = striptags(eventDefinition.href.trim());

    if (eventDefinition.thumbnail) {
      event.thumbnail = striptags(eventDefinition.thumbnail);
    }

    event.source = {
      pkg: {
        name: striptags(eventDefinition.source.pkg.name.trim().toLowerCase()),
        version: striptags(eventDefinition.source.pkg.version.trim()),
      },
      site: {
        name: striptags(eventDefinition.source.site.name.trim()),
        domain: striptags(eventDefinition.source.site.domain.trim().toLowerCase()),
        domainKey: striptags(eventDefinition.source.site.domainKey.trim().toLowerCase()),
        coreAuth: {
          scopes: eventDefinition.source.site.coreAuth.scopes.map((scope) => scope.trim().toLowerCase()),
        },
      },
    };

    if (sourceClient) {
      event.source.client = sourceClient._id;
    }

    if (eventDefinition.source.emitter) {
      event.source.emitter = {
        emitterType: striptags(eventDefinition.source.emitter.emitterType),
        emitterId: mongoose.Types.ObjectId(eventDefinition.source.emitter.emitterId),
        href: striptags(eventDefinition.source.emitter.href.trim()),
      };
      if (eventDefinition.source.emitter.displayName) {
        event.source.emitter.displayName = striptags(eventDefinition.source.emitter.displayName.trim());
      }
      if (eventDefinition.source.emitter.username) {
        event.source.emitter.username = striptags(eventDefinition.source.emitter.username.trim());
      }
    }

    if (eventDefinition.source.site.company) {
      event.source.site.company = striptags(eventDefinition.source.site.company.trim());
    }
    if (eventDefinition.source.site.description) {
      event.source.site.description = striptags(eventDefinition.source.site.description.trim());
    }

    if (eventDefinition.source.emitter.displayName) {
      event.source.emitter.displayName = striptags(eventDefinition.source.emitter.displayName);
    }

    event.attachmentType = eventDefinition.attachmentType;
    event.attachment = eventDefinition.attachment;

    await event.save();

    return event.toObject();
  }

  async getConstellationTimeline (user, pagination) {
    const totalEventCount = await KaleidoscopeEvent.estimatedDocumentCount();
    const job = { };
    if (user) {
      job.search = {
        $or: [
          { recipient: { $exists: false } },
          { recipient: user._id },
        ],
      };
    } else {
      job.search = { recipient: { $exists: false } };
    }
    const events = await KaleidoscopeEvent
      .find(job.search)
      .sort({ created: -1 })
      .skip(pagination.skip)
      .limit(pagination.cpp)
      .lean();
    return { events, totalEventCount };
  }

  /*
   * OAuth2 event handlers
   */

  /**
   * This event fires when an OAuth2Client is being disconnected and removed by a
   * Core, or a client app is being removed from a Service Node. The Hive service
   * will remove all KaleidoscopeEvent records created on behalf of the client.
   * @param {OAuth2Client} client the client being removed
   */
  async onOAuth2RemoveClient (client) {
    this.log.alert('removing KaleidoscopeEvent records from OAuth2Client', { clientId: client._id, domain: client.site.domain });
    await KaleidoscopeEvent
      .find({ 'source.client': client._id })
      .cursor()
      .eachAsync(async (event) => {
        await UserNotification.deleteMany({ event: event._id });
        await KaleidoscopeEvent.deleteOne({ _id: event._id });
      }, 1);
  }
}

module.exports = {
  slug: 'hive',
  name: 'hive',
  create: (dtp) => { return new HiveService(dtp); },
};