Activitystream IoC 1 addStreamAccessRequestedActivity (#3227)

* chore(activitystream): addStreamUpdatedActivity refactor multi region

* chore(activitystream): addStreamAccessRequestedActivity refactor multiregion

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

@@ -186,3 +186,8 @@ export type AddStreamUpdatedActivity = (params: {
   newStream: StreamRecord
   update: ProjectUpdateInput | StreamUpdateInput
 }) => Promise<void>
+
+export type AddStreamAccessRequestedActivity = (params: {
+  streamId: string
+  requesterId: string
+}) => Promise<void>

packages/server/modules/activitystream/index.ts

@@ -1,5 +1,5 @@
 import { Optional, SpeckleModule } from '@/modules/shared/helpers/typeHelper'
-import { initializeEventListener } from '@/modules/activitystream/services/eventListener'
+import { initializeEventListenerFactory } from '@/modules/activitystream/services/eventListener'
 import { publishNotification } from '@/modules/notifications/services/publication'
 import { scheduleExecution } from '@/modules/core/services/taskScheduler'
 import { activitiesLogger, moduleLogger } from '@/logging/logging'
@@ -15,6 +15,7 @@ import { db } from '@/db/knex'
 import { addStreamInviteSentOutActivityFactory } from '@/modules/activitystream/services/streamActivity'
 import { publish } from '@/modules/shared/utils/subscriptions'
 import { getStreamFactory } from '@/modules/core/repositories/streams'
+import { addStreamAccessRequestedActivityFactory } from '@/modules/activitystream/services/accessRequestActivity'
 
 let scheduledTask: ReturnType<typeof scheduleExecution> | null = null
 let quitEventListeners: Optional<ReturnType<typeof initializeEventListeners>> =
@@ -68,7 +69,11 @@ const activityModule: SpeckleModule = {
   init: async (_, isInitial) => {
     moduleLogger.info('🤺 Init activity module')
     if (isInitial) {
-      initializeEventListener()
+      initializeEventListenerFactory({
+        addStreamAccessRequestedActivity: addStreamAccessRequestedActivityFactory({
+          saveActivity: saveActivityFactory({ db })
+        })
+      })()
       if (weeklyEmailDigestEnabled())
         scheduledTask = scheduleWeeklyActivityNotifications()
     }

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

@@ -1,25 +1,32 @@
 import { db } from '@/db/knex'
+import {
+  AddStreamAccessRequestedActivity,
+  SaveActivity
+} from '@/modules/activitystream/domain/operations'
 import { ActionTypes, ResourceTypes } from '@/modules/activitystream/helpers/types'
 import { saveActivityFactory } from '@/modules/activitystream/repositories'
 
 /**
  * Save a "stream access requested" activity
  */
-export async function addStreamAccessRequestedActivity(params: {
-  streamId: string
-  requesterId: string
-}) {
-  const { streamId, requesterId } = params
-  await saveActivityFactory({ db })({
-    streamId,
-    resourceType: ResourceTypes.Stream,
-    resourceId: streamId,
-    userId: requesterId,
-    actionType: ActionTypes.Stream.AccessRequestSent,
-    message: `User ${requesterId} has requested access to stream ${streamId}`,
-    info: { requesterId }
-  })
-}
+export const addStreamAccessRequestedActivityFactory =
+  ({
+    saveActivity
+  }: {
+    saveActivity: SaveActivity
+  }): AddStreamAccessRequestedActivity =>
+  async (params: { streamId: string; requesterId: string }) => {
+    const { streamId, requesterId } = params
+    await saveActivity({
+      streamId,
+      resourceType: ResourceTypes.Stream,
+      resourceId: streamId,
+      userId: requesterId,
+      actionType: ActionTypes.Stream.AccessRequestSent,
+      message: `User ${requesterId} has requested access to stream ${streamId}`,
+      info: { requesterId }
+    })
+  }
 
 /**
  * Save a "stream acccess request declined/denied" activity

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

@@ -5,11 +5,9 @@ import {
   AccessRequestsEventsPayloads
 } from '@/modules/accessrequests/events/emitter'
 import { AccessRequestType } from '@/modules/accessrequests/repositories'
+import { AddStreamAccessRequestedActivity } from '@/modules/activitystream/domain/operations'
 import { saveActivityFactory } from '@/modules/activitystream/repositories'
-import {
-  addStreamAccessRequestDeclinedActivity,
-  addStreamAccessRequestedActivity
-} from '@/modules/activitystream/services/accessRequestActivity'
+import { addStreamAccessRequestDeclinedActivity } from '@/modules/activitystream/services/accessRequestActivity'
 import {
   UsersEmitter,
   UsersEvents,
@@ -30,21 +28,25 @@ async function onUserCreated(payload: UsersEventsPayloads[UsersEvents.Created])
   })
 }
 
-async function onServerAccessRequestCreated(
-  payload: AccessRequestsEventsPayloads[AccessRequestsEvents.Created]
-) {
-  const {
-    request: { resourceId, resourceType, requesterId }
-  } = payload
-  if (!resourceId) return
+const onServerAccessRequestCreatedFactory =
+  ({
+    addStreamAccessRequestedActivity
+  }: {
+    addStreamAccessRequestedActivity: AddStreamAccessRequestedActivity
+  }) =>
+  async (payload: AccessRequestsEventsPayloads[AccessRequestsEvents.Created]) => {
+    const {
+      request: { resourceId, resourceType, requesterId }
+    } = payload
+    if (!resourceId) return
 
-  if (resourceType === AccessRequestType.Stream) {
-    await addStreamAccessRequestedActivity({
-      streamId: resourceId,
-      requesterId
-    })
+    if (resourceType === AccessRequestType.Stream) {
+      await addStreamAccessRequestedActivity({
+        streamId: resourceId,
+        requesterId
+      })
+    }
   }
-}
 
 async function onServerAccessRequestFinalized(
   payload: AccessRequestsEventsPayloads[AccessRequestsEvents.Finalized]
@@ -72,18 +74,24 @@ async function onServerAccessRequestFinalized(
  * Initialize event listener for tracking various Speckle events and responding
  * to them by creating activitystream entries
  */
-export function initializeEventListener() {
-  const quitCbs = [
-    UsersEmitter.listen(UsersEvents.Created, onUserCreated),
-    AccessRequestsEmitter.listen(
-      AccessRequestsEvents.Created,
-      onServerAccessRequestCreated
-    ),
-    AccessRequestsEmitter.listen(
-      AccessRequestsEvents.Finalized,
-      onServerAccessRequestFinalized
-    )
-  ]
+export const initializeEventListenerFactory =
+  ({
+    addStreamAccessRequestedActivity
+  }: {
+    addStreamAccessRequestedActivity: AddStreamAccessRequestedActivity
+  }) =>
+  () => {
+    const quitCbs = [
+      UsersEmitter.listen(UsersEvents.Created, onUserCreated),
+      AccessRequestsEmitter.listen(
+        AccessRequestsEvents.Created,
+        onServerAccessRequestCreatedFactory({ addStreamAccessRequestedActivity })
+      ),
+      AccessRequestsEmitter.listen(
+        AccessRequestsEvents.Finalized,
+        onServerAccessRequestFinalized
+      )
+    ]
 
-  return () => quitCbs.forEach((quit) => quit())
-}
+    return () => quitCbs.forEach((quit) => quit())
+  }