speckle-server/packages/server/modules/activitystream/services/index.js

'use strict'

const appRoot = require( 'app-root-path' )
const knex = require( `${appRoot}/db/knex` )

const { dispatchStreamEvent } = require( '../../webhooks/services/webhooks' )
const StreamActivity = () => knex( 'stream_activity' )
const StreamAcl = ( ) => knex( 'stream_acl' )

module.exports = {

  async saveActivity( { streamId, resourceType, resourceId, actionType, userId, info, message } ) {
    let dbObject = {
      streamId, // abc
      resourceType, // "commit"
      resourceId, // commit id
      actionType, // "commit_receive"
      userId, // populated by the api
      info: JSON.stringify( info ), // can be anything with conventions! (TBD) 
      message, // something human understandable for frontend purposes mostly
    }
    await StreamActivity( ).insert( dbObject )
    if ( streamId ) {
      let webhooksPayload = {
        streamId: streamId,
        userId: userId,
        activityMessage: message,
        event: {
          'event_name': actionType,
          'data': info
        }
      }
      dispatchStreamEvent( { streamId, event: actionType, eventPayload: webhooksPayload } )
    }
  },

  async getStreamActivity( { streamId, actionType, after, before, cursor, limit } ) {
    if ( !limit ) {
      limit = 200
    }

    let dbQuery = StreamActivity().where( { streamId: streamId } )
    if ( actionType ) dbQuery.andWhere( { actionType: actionType } )
    if ( after ) dbQuery.andWhere( 'time', '>', after )
    if ( before ) dbQuery.andWhere( 'time', '<', before )
    if ( cursor ) dbQuery.andWhere( 'time', '<', cursor )
    dbQuery.orderBy( 'time', 'desc' ).limit( limit )

    let results = await dbQuery.select( '*' )

    return { items: results, cursor: results.length > 0 ? results[ results.length - 1 ].time.toISOString() : null }
  },

  async getUserActivity( { userId, actionType, after, before, cursor, limit } ) {
    if ( !limit ) {
      limit = 200
    }

    let dbQuery = StreamActivity().where( { userId: userId } )
    if ( actionType ) dbQuery.andWhere( { actionType: actionType } )
    if ( after ) dbQuery.andWhere( 'time', '>', after )
    if ( before ) dbQuery.andWhere( 'time', '<', before )
    if ( cursor ) dbQuery.andWhere( 'time', '<', cursor )
    dbQuery.orderBy( 'time', 'desc' ).limit( limit )

    let results = await dbQuery.select( '*' )
    return { items: results, cursor: results.length > 0 ? results[ results.length - 1 ].time.toISOString() : null }
  },

  async getResourceActivity( { resourceType, resourceId, actionType, after, before, cursor, limit } ) {
    if ( !limit ) {
      limit = 200
    }

    let dbQuery = StreamActivity().where( { resourceType, resourceId } )
    if ( actionType ) dbQuery.andWhere( { actionType: actionType } )
    if ( after ) dbQuery.andWhere( 'time', '>', after )
    if ( before ) dbQuery.andWhere( 'time', '<', before )
    if ( cursor ) dbQuery.andWhere( 'time', '<', cursor )
    dbQuery.orderBy( 'time', 'desc' ).limit( limit )

    let results = await dbQuery.select( '*' )
    return { items: results, cursor: results.length > 0 ? results[ results.length - 1 ].time.toISOString() : null }
  },

  async getUserTimeline( { userId, after, before, cursor, limit } ) {
    if ( !limit ) {
      limit = 200
    }

    let sqlFilters = ''
    let sqlVariables = []
    if ( after ) {
      sqlFilters += ' AND time > ?'
      sqlVariables.push( after )
    }
    if ( before ) {
      sqlFilters += ' AND time < ?'
      sqlVariables.push( before )
    }
    if ( cursor ) {
      sqlFilters += ' AND time < ?'
      sqlVariables.push( cursor )
    }

    let dbRawQuery = `
      SELECT act.*
      FROM stream_acl acl
      INNER JOIN stream_activity act ON acl."resourceId" = act."streamId"
      WHERE acl."userId" = ? ${sqlFilters}
      ORDER BY time DESC
      LIMIT ?
    `

    sqlVariables.unshift( userId )
    sqlVariables.push( limit )
    let results = ( await knex.raw( dbRawQuery, sqlVariables ) ).rows
    return { items: results, cursor: results.length > 0 ? results[ results.length - 1 ].time.toISOString() : null }
  },

  async getActivityCountByResourceId( { resourceId, actionType, after, before } ) {
    let query = StreamActivity().count().where( { resourceId } )
    if ( actionType ) query.andWhere( { actionType } )
    if ( after ) query.andWhere( 'time', '>', after )
    if ( before ) query.andWhere( 'time', '<', before )
    let [ res ] = await query
    return parseInt( res.count )
  },

  async getActivityCountByStreamId( { streamId, actionType, after, before } ) {
    let query = StreamActivity().count().where( { streamId } )
    if ( actionType ) query.andWhere( { actionType } )
    if ( after ) query.andWhere( 'time', '>', after )
    if ( before ) query.andWhere( 'time', '<', before )
    let [ res ] = await query
    return parseInt( res.count )
  },

  async getActivityCountByUserId( { userId, actionType, after, before } ) {
    let query = StreamActivity().count().where( { userId } )
    if ( actionType ) query.andWhere( { actionType } )
    if ( after ) query.andWhere( 'time', '>', after )
    if ( before ) query.andWhere( 'time', '<', before )
    let [ res ] = await query
    return parseInt( res.count )
  },

  async getTimelineCount( { userId, after, before } ) {
    let query = StreamAcl().count()
      .innerJoin( 'stream_activity', { 'stream_acl.resourceId': 'stream_activity.streamId' } )
      .where( { 'stream_acl.userId': userId } )
    if ( after ) query.andWhere( 'stream_activity.time', '>', after )
    if ( before ) query.andWhere( 'stream_activity.time', '<', before )
    let [ res ] = await query
    return parseInt( res.count )
  }
}