Merge pull request #3128 from specklesystems/alessandro/web-943-save-…

…activity

Activitystream IoC 1 saveActivity

packages/server/modules/activitystream/domain/operations.ts

@@ -133,3 +133,5 @@ export type GetUserActivity = ({
   cursor: string | null
   items: StreamActivityRecord[]
 }>
+
+export type SaveActivity = (args: Omit<StreamActivityRecord, 'time'>) => Promise<void>

packages/server/modules/activitystream/helpers/types.ts

@@ -5,7 +5,7 @@ export type StreamActivityRecord = {
   time: Date
   resourceType: Nullable<(typeof ResourceTypes)[keyof typeof ResourceTypes]>
   resourceId: Nullable<string>
-  actionType: Nullable<AllActivityTypes>
+  actionType: AllActivityTypes
   userId: Nullable<string>
   info: Nullable<Record<string, unknown>>
   message: Nullable<string>

packages/server/modules/activitystream/repositories/index.ts

@@ -8,7 +8,8 @@ import {
   GetStreamActivity,
   GetTimelineCount,
   GetUserActivity,
-  GetUserTimeline
+  GetUserTimeline,
+  SaveActivity
 } from '@/modules/activitystream/domain/operations'
 import {
   StreamActivityRecord,
@@ -17,6 +18,11 @@ import {
 import { StreamAcl, StreamActivity } from '@/modules/core/dbSchema'
 import { Roles } from '@/modules/core/helpers/mainConstants'
 import { StreamAclRecord } from '@/modules/core/helpers/types'
+import { getStream } from '@/modules/core/repositories/streams'
+import { getServerInfo } from '@/modules/core/services/generic'
+import { getUser } from '@/modules/core/repositories/users'
+import { createWebhookEventFactory } from '@/modules/webhooks/repositories/webhooks'
+import { dispatchStreamEventFactory } from '@/modules/webhooks/services/webhooks'
 import { Knex } from 'knex'
 
 const tables = {
@@ -213,3 +219,47 @@ export const getUserActivityFactory =
       cursor: results.length > 0 ? results[results.length - 1].time.toISOString() : null
     }
   }
+
+// TODO: this function should be a service
+export const saveActivityFactory =
+  ({ db }: { db: Knex }): SaveActivity =>
+  async ({ streamId, resourceType, resourceId, actionType, userId, info, message }) => {
+    const 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 tables
+      .streamActivity<Omit<StreamActivityRecord, 'info'> & { info: string }>(db)
+      .insert(dbObject)
+
+    if (streamId) {
+      const webhooksPayload = {
+        streamId,
+        userId,
+        activityMessage: message,
+        event: {
+          // eslint-disable-next-line camelcase
+          event_name: actionType,
+          data: info
+        }
+      }
+
+      await dispatchStreamEventFactory({
+        db,
+        getServerInfo,
+        getStream,
+        createWebhookEvent: createWebhookEventFactory({ db }),
+        getUser
+      })({
+        streamId,
+        event: actionType,
+        eventPayload: webhooksPayload
+      })
+    }
+  }

packages/server/modules/activitystream/services/accessRequestActivity.ts

@@ -1,6 +1,6 @@
-import { saveActivity } from '@/modules/activitystream/services'
-
+import { db } from '@/db/knex'
 import { ActionTypes, ResourceTypes } from '@/modules/activitystream/helpers/types'
+import { saveActivityFactory } from '@/modules/activitystream/repositories'
 
 /**
  * Save a "stream access requested" activity
@@ -10,7 +10,7 @@ export async function addStreamAccessRequestedActivity(params: {
   requesterId: string
 }) {
   const { streamId, requesterId } = params
-  await saveActivity({
+  await saveActivityFactory({ db })({
     streamId,
     resourceType: ResourceTypes.Stream,
     resourceId: streamId,
@@ -30,7 +30,7 @@ export async function addStreamAccessRequestDeclinedActivity(params: {
   declinerId: string
 }) {
   const { streamId, requesterId, declinerId } = params
-  await saveActivity({
+  await saveActivityFactory({ db })({
     streamId,
     resourceType: ResourceTypes.Stream,
     resourceId: streamId,

packages/server/modules/activitystream/services/branchActivity.ts

@@ -1,4 +1,3 @@
-import { saveActivity } from '@/modules/activitystream/services'
 import { ActionTypes, ResourceTypes } from '@/modules/activitystream/helpers/types'
 import { BranchRecord } from '@/modules/core/helpers/types'
 import {
@@ -14,6 +13,8 @@ import {
 } from '@/modules/core/graph/generated/graphql'
 import { ProjectSubscriptions, publish } from '@/modules/shared/utils/subscriptions'
 import { isBranchDeleteInput, isBranchUpdateInput } from '@/modules/core/helpers/branch'
+import { saveActivityFactory } from '@/modules/activitystream/repositories'
+import { db } from '@/db/knex'
 
 /**
  * Save "branch created" activity
@@ -22,7 +23,7 @@ export async function addBranchCreatedActivity(params: { branch: BranchRecord })
   const { branch } = params
 
   await Promise.all([
-    saveActivity({
+    saveActivityFactory({ db })({
       streamId: branch.streamId,
       resourceType: ResourceTypes.Branch,
       resourceId: branch.id,
@@ -56,7 +57,7 @@ export async function addBranchUpdatedActivity(params: {
 
   const streamId = isBranchUpdateInput(update) ? update.streamId : update.projectId
   await Promise.all([
-    saveActivity({
+    saveActivityFactory({ db })({
       streamId,
       resourceType: ResourceTypes.Branch,
       resourceId: update.id,
@@ -90,7 +91,7 @@ export async function addBranchDeletedActivity(params: {
 
   const streamId = isBranchDeleteInput(input) ? input.streamId : input.projectId
   await Promise.all([
-    saveActivity({
+    saveActivityFactory({ db })({
       streamId,
       resourceType: ResourceTypes.Branch,
       resourceId: input.id,

packages/server/modules/activitystream/services/commentActivity.ts

@@ -1,6 +1,6 @@
 import { db } from '@/db/knex'
 import { ActionTypes, ResourceTypes } from '@/modules/activitystream/helpers/types'
-import { saveActivity } from '@/modules/activitystream/services'
+import { saveActivityFactory } from '@/modules/activitystream/repositories'
 import { ViewerResourceItem } from '@/modules/comments/domain/types'
 import { CommentRecord } from '@/modules/comments/helpers/types'
 import { getCommentsResourcesFactory } from '@/modules/comments/repositories/comments'
@@ -96,7 +96,7 @@ export async function addCommentCreatedActivity(params: {
   }
 
   await Promise.all([
-    saveActivity({
+    saveActivityFactory({ db })({
       resourceId: comment.id,
       streamId,
       resourceType: ResourceTypes.Comment,
@@ -156,7 +156,7 @@ export async function addCommentArchivedActivity(params: {
   })
 
   await Promise.all([
-    saveActivity({
+    saveActivityFactory({ db })({
       streamId,
       resourceType: ResourceTypes.Comment,
       resourceId: commentId,
@@ -221,7 +221,7 @@ export async function addReplyAddedActivity(params: {
     ? input.parentComment
     : input.threadId
   await Promise.all([
-    saveActivity({
+    saveActivityFactory({ db })({
       streamId,
       resourceType: ResourceTypes.Comment,
       resourceId: parentCommentId,

packages/server/modules/activitystream/services/commitActivity.ts

@@ -1,4 +1,3 @@
-import { saveActivity } from '@/modules/activitystream/services'
 import { ActionTypes, ResourceTypes } from '@/modules/activitystream/helpers/types'
 import {
   CommitSubscriptions as CommitPubsubEvents,
@@ -14,6 +13,8 @@ import {
 import { CommitRecord } from '@/modules/core/helpers/types'
 import { ProjectSubscriptions, publish } from '@/modules/shared/utils/subscriptions'
 import { has } from 'lodash'
+import { saveActivityFactory } from '@/modules/activitystream/repositories'
+import { db } from '@/db/knex'
 
 /**
  * Save "new commit created" activity item
@@ -29,7 +30,7 @@ export async function addCommitCreatedActivity(params: {
 }) {
   const { commitId, input, streamId, userId, branchName, commit } = params
   await Promise.all([
-    saveActivity({
+    saveActivityFactory({ db })({
       streamId,
       resourceType: ResourceTypes.Commit,
       resourceId: commitId,
@@ -84,7 +85,7 @@ export async function addCommitUpdatedActivity(params: {
       }
 
   await Promise.all([
-    saveActivity({
+    saveActivityFactory({ db })({
       streamId,
       resourceType: ResourceTypes.Commit,
       resourceId: commitId,
@@ -120,7 +121,7 @@ export async function addCommitMovedActivity(params: {
 }) {
   const { commitId, streamId, userId, originalBranchId, newBranchId, commit } = params
   await Promise.all([
-    saveActivity({
+    saveActivityFactory({ db })({
       streamId,
       resourceType: ResourceTypes.Commit,
       resourceId: commitId,
@@ -150,7 +151,7 @@ export async function addCommitDeletedActivity(params: {
 }) {
   const { commitId, streamId, userId, commit, branchId } = params
   await Promise.all([
-    saveActivity({
+    saveActivityFactory({ db })({
       streamId,
       resourceType: ResourceTypes.Commit,
       resourceId: commitId,
@@ -181,7 +182,7 @@ export async function addCommitReceivedActivity(params: {
 }) {
   const { input, userId } = params
 
-  await saveActivity({
+  await saveActivityFactory({ db })({
     streamId: input.streamId,
     resourceType: ResourceTypes.Commit,
     resourceId: input.commitId,

packages/server/modules/activitystream/services/eventListener.ts

@@ -1,10 +1,11 @@
+import { db } from '@/db/knex'
 import {
   AccessRequestsEmitter,
   AccessRequestsEvents,
   AccessRequestsEventsPayloads
 } from '@/modules/accessrequests/events/emitter'
 import { AccessRequestType } from '@/modules/accessrequests/repositories'
-import { saveActivity } from '@/modules/activitystream/services'
+import { saveActivityFactory } from '@/modules/activitystream/repositories'
 import {
   addStreamAccessRequestDeclinedActivity,
   addStreamAccessRequestedActivity
@@ -18,7 +19,7 @@ import {
 async function onUserCreated(payload: UsersEventsPayloads[UsersEvents.Created]) {
   const { user } = payload
 
-  await saveActivity({
+  await saveActivityFactory({ db })({
     streamId: null,
     resourceType: 'user',
     resourceId: user.id,