Merge pull request #3201 from specklesystems/fabians/core-ioc-29

chore(server): IoC 29 - getStream(s)Factory

packages/server/modules/accessrequests/graph/resolvers/index.ts

@@ -20,13 +20,14 @@ import {
 import { Resolvers } from '@/modules/core/graph/generated/graphql'
 import { mapStreamRoleToValue } from '@/modules/core/helpers/graphTypes'
 import { Roles } from '@/modules/core/helpers/mainConstants'
-import { getStream } from '@/modules/core/repositories/streams'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 import {
   addOrUpdateStreamCollaborator,
   validateStreamAccess
 } from '@/modules/core/services/streams/streamAccessService'
 import { LogicError } from '@/modules/shared/errors'
 
+const getStream = getStreamFactory({ db })
 const getUserProjectAccessRequest = getUserProjectAccessRequestFactory({
   getUsersPendingAccessRequest: getUsersPendingAccessRequestFactory({ db })
 })

packages/server/modules/accessrequests/services/stream.ts

@@ -13,7 +13,6 @@ import {
 import { StreamInvalidAccessError } from '@/modules/core/errors/stream'
 import { TokenResourceIdentifier } from '@/modules/core/domain/tokens/types'
 import { Roles, StreamRoles } from '@/modules/core/helpers/mainConstants'
-import { getStream } from '@/modules/core/repositories/streams'
 import {
   addOrUpdateStreamCollaborator,
   validateStreamAccess
@@ -35,6 +34,7 @@ import {
   GetUserStreamAccessRequest,
   RequestProjectAccess
 } from '@/modules/accessrequests/domain/operations'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 
 function buildStreamAccessRequestGraphQLReturn(
   record: ServerAccessRequestRecord<AccessRequestType.Stream, string>
@@ -83,7 +83,7 @@ export const getUserStreamAccessRequestFactory =
 export const requestProjectAccessFactory =
   (deps: {
     getUserStreamAccessRequest: GetUserStreamAccessRequest
-    getStream: typeof getStream
+    getStream: GetStream
     createNewRequest: CreateNewRequest
     accessRequestsEmitter: (typeof AccessRequestsEmitter)['emit']
   }): RequestProjectAccess =>

packages/server/modules/accessrequests/tests/projectAccessRequests.spec.ts

@@ -21,7 +21,10 @@ import {
 import { StreamAccessUpdateError } from '@/modules/core/errors/stream'
 import { mapStreamRoleToValue } from '@/modules/core/helpers/graphTypes'
 import { Roles } from '@/modules/core/helpers/mainConstants'
-import { getStream, getStreamCollaborators } from '@/modules/core/repositories/streams'
+import {
+  getStreamCollaborators,
+  getStreamFactory
+} from '@/modules/core/repositories/streams'
 import {
   addOrUpdateStreamCollaborator,
   removeStreamCollaborator
@@ -48,6 +51,7 @@ import { BasicTestStream, createTestStreams } from '@/test/speckle-helpers/strea
 import { expect } from 'chai'
 import { noop } from 'lodash'
 
+const getStream = getStreamFactory({ db })
 const requestProjectAccess = requestProjectAccessFactory({
   getUserStreamAccessRequest: getUserStreamAccessRequestFactory({
     getUserProjectAccessRequest: getUserProjectAccessRequestFactory({

packages/server/modules/accessrequests/tests/streamAccessRequests.spec.ts

@@ -23,7 +23,10 @@ import {
 import { StreamAccessUpdateError } from '@/modules/core/errors/stream'
 import { mapStreamRoleToValue } from '@/modules/core/helpers/graphTypes'
 import { Roles } from '@/modules/core/helpers/mainConstants'
-import { getStream, getStreamCollaborators } from '@/modules/core/repositories/streams'
+import {
+  getStreamCollaborators,
+  getStreamFactory
+} from '@/modules/core/repositories/streams'
 import {
   addOrUpdateStreamCollaborator,
   removeStreamCollaborator
@@ -50,6 +53,7 @@ import { BasicTestStream, createTestStreams } from '@/test/speckle-helpers/strea
 import { expect } from 'chai'
 import { noop } from 'lodash'
 
+const getStream = getStreamFactory({ db })
 const requestStreamAccess = requestStreamAccessFactory({
   requestProjectAccess: requestProjectAccessFactory({
     getUserStreamAccessRequest: getUserStreamAccessRequestFactory({

packages/server/modules/activitystream/index.ts

@@ -6,7 +6,6 @@ import { activitiesLogger, moduleLogger } from '@/logging/logging'
 import { weeklyEmailDigestEnabled } from '@/modules/shared/helpers/envHelper'
 import { getEventBus } from '@/modules/shared/services/eventBus'
 import { handleServerInvitesActivitiesFactory } from '@/modules/activitystream/services/serverInvitesActivity'
-import { getStream } from '@/modules/core/repositories/streams'
 import { sendActivityNotificationsFactory } from '@/modules/activitystream/services/summary'
 import {
   getActiveUserStreamsFactory,
@@ -15,6 +14,7 @@ import {
 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'
 
 let scheduledTask: ReturnType<typeof scheduleExecution> | null = null
 let quitEventListeners: Optional<ReturnType<typeof initializeEventListeners>> =
@@ -24,7 +24,7 @@ const initializeEventListeners = () => {
   const handleServerInvitesActivities = handleServerInvitesActivitiesFactory({
     eventBus: getEventBus(),
     logger: activitiesLogger,
-    getStream,
+    getStream: getStreamFactory({ db }),
     addStreamInviteSentOutActivity: addStreamInviteSentOutActivityFactory({
       saveActivity: saveActivityFactory({ db }),
       publish

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

@@ -18,12 +18,12 @@ 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'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 
 const tables = {
   streamActivity: <T extends object = StreamActivityRecord>(db: Knex) =>
@@ -253,7 +253,7 @@ export const saveActivityFactory =
       await dispatchStreamEventFactory({
         db,
         getServerInfo,
-        getStream,
+        getStream: getStreamFactory({ db }),
         createWebhookEvent: createWebhookEventFactory({ db }),
         getUser
       })({

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

@@ -7,12 +7,12 @@ import {
   resolveTarget
 } from '@/modules/serverinvites/helpers/core'
 import { EventBus } from '@/modules/shared/services/eventBus'
-import { getStream } from '@/modules/core/repositories/streams'
 import { Logger } from '@/logging/logging'
 import { AddStreamInviteSentOutActivity } from '@/modules/activitystream/domain/operations'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 
 type OnServerInviteCreatedFactoryDeps = {
-  getStream: typeof getStream
+  getStream: GetStream
   logger: Logger
   addStreamInviteSentOutActivity: AddStreamInviteSentOutActivity
 }

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

@@ -3,19 +3,19 @@ import {
   NotificationType
 } from '@/modules/notifications/helpers/types'
 import { getUser } from '@/modules/core/repositories/users'
-import { getStream as getStreamService } from '@/modules/core/services/streams'
 import {
   CreateActivitySummary,
   GetActiveUserStreams,
   GetActivity
 } from '@/modules/activitystream/domain/operations'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 
 export const createActivitySummaryFactory =
   ({
     getStream,
     getActivity
   }: {
-    getStream: typeof getStreamService
+    getStream: GetStream
     getActivity: GetActivity
   }): CreateActivitySummary =>
   async ({

packages/server/modules/activitystream/tests/activitySummary.spec.ts

@@ -6,7 +6,7 @@ import {
   sendActivityNotificationsFactory
 } from '@/modules/activitystream/services/summary'
 import { expect } from 'chai'
-import { createStream, deleteStream, getStream } from '@/modules/core/services/streams'
+import { createStream, deleteStream } from '@/modules/core/services/streams'
 import { ActionTypes, ResourceTypes } from '@/modules/activitystream/helpers/types'
 import {
   ActivityDigestMessage,
@@ -19,11 +19,13 @@ import {
   saveActivityFactory
 } from '@/modules/activitystream/repositories'
 import { db } from '@/db/knex'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 
 const cleanup = async () => {
   await truncateTables([StreamActivity.name, Users.name])
 }
 
+const getStream = getStreamFactory({ db })
 const saveActivity = saveActivityFactory({ db })
 const createActivitySummary = createActivitySummaryFactory({
   getStream,

packages/server/modules/automate/rest/logStream.ts

@@ -6,7 +6,7 @@ import {
   getAutomationRunWithTokenFactory
 } from '@/modules/automate/repositories/automations'
 import { corsMiddleware } from '@/modules/core/configs/cors'
-import { getStream } from '@/modules/core/repositories/streams'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 import {
   validateRequiredStreamFactory,
   validateResourceAccess,
@@ -29,7 +29,7 @@ export default (app: Application) => {
       })({ requiredRole: Roles.Server.Guest }),
       validateScope({ requiredScope: Scopes.Streams.Read }),
       validateRequiredStreamFactory({
-        getStream,
+        getStream: getStreamFactory({ db }),
         getAutomationProject: getAutomationProjectFactory({ db })
       }),
       validateStreamRoleBuilderFactory({ getRoles: getRolesFactory({ db }) })({

packages/server/modules/blobstorage/index.js

@@ -41,13 +41,14 @@ const {
   markUploadOverFileSizeLimitFactory,
   fullyDeleteBlobFactory
 } = require('@/modules/blobstorage/services/management')
-const { getStream } = require('@/modules/core/repositories/streams')
 const { getRolesFactory } = require('@/modules/shared/repositories/roles')
 const {
   getAutomationProjectFactory
 } = require('@/modules/automate/repositories/automations')
 const { adminOverrideEnabled } = require('@/modules/shared/helpers/envHelper')
+const { getStreamFactory } = require('@/modules/core/repositories/streams')
 
+const getStream = getStreamFactory({ db })
 const getAllStreamBlobIds = getAllStreamBlobIdsFactory({ db })
 const updateBlob = updateBlobFactory({ db })
 const uploadFileStream = uploadFileStreamFactory({

packages/server/modules/cli/commands/db/seed/commits.ts

@@ -1,5 +1,6 @@
+import { db } from '@/db/knex'
 import { cliLogger } from '@/logging/logging'
-import { getStream } from '@/modules/core/repositories/streams'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 import { getUser } from '@/modules/core/repositories/users'
 import { BasicTestCommit, createTestCommits } from '@/test/speckle-helpers/commitHelper'
 import dayjs from 'dayjs'
@@ -28,6 +29,8 @@ const command: CommandModule<
     }
   },
   handler: async (argv) => {
+    const getStream = getStreamFactory({ db })
+
     const count = argv.count
     const streamId = argv.streamId
     const authorId = argv.authorId

packages/server/modules/cli/commands/download/commit.ts

@@ -2,8 +2,8 @@ import { CommandModule } from 'yargs'
 import { downloadCommitFactory } from '@/modules/cross-server-sync/services/commit'
 import { cliLogger } from '@/logging/logging'
 import {
-  getStream,
   getStreamCollaborators,
+  getStreamFactory,
   markCommitStreamUpdated
 } from '@/modules/core/repositories/streams'
 import {
@@ -92,6 +92,7 @@ const command: CommandModule<
     }
   },
   handler: async (argv) => {
+    const getStream = getStreamFactory({ db })
     const getObject = getObjectFactory({ db })
     const getStreamObjects = getStreamObjectsFactory({ db })
     const markCommentViewed = markCommentViewedFactory({ db })

packages/server/modules/cli/commands/download/project.ts

@@ -3,8 +3,8 @@ import { cliLogger } from '@/logging/logging'
 import { downloadProjectFactory } from '@/modules/cross-server-sync/services/project'
 import { downloadCommitFactory } from '@/modules/cross-server-sync/services/commit'
 import {
-  getStream,
   getStreamCollaborators,
+  getStreamFactory,
   markCommitStreamUpdated
 } from '@/modules/core/repositories/streams'
 import {
@@ -90,6 +90,7 @@ const command: CommandModule<
     }
   },
   handler: async (argv) => {
+    const getStream = getStreamFactory({ db })
     const getObject = getObjectFactory({ db })
 
     const getStreamObjects = getStreamObjectsFactory({ db })

packages/server/modules/comments/graph/resolvers/comments.ts

@@ -1,6 +1,5 @@
 import { pubsub } from '@/modules/shared/utils/subscriptions'
 import { ForbiddenError } from '@/modules/shared/errors'
-import { getStream } from '@/modules/core/services/streams'
 import { Roles } from '@/modules/core/helpers/mainConstants'
 import {
   streamResourceCheckFactory,
@@ -96,7 +95,9 @@ import {
   getStreamBranchesByNameFactory
 } from '@/modules/core/repositories/branches'
 import { getStreamObjectsFactory } from '@/modules/core/repositories/objects'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 
+const getStream = getStreamFactory({ db })
 const streamResourceCheck = streamResourceCheckFactory({
   checkStreamResourceAccess: checkStreamResourceAccessFactory({ db })
 })

packages/server/modules/comments/services/index.ts

@@ -24,7 +24,7 @@ import {
   ValidateInputAttachments
 } from '@/modules/comments/domain/operations'
 import { ResourceType } from '@/modules/comments/domain/types'
-import { getStream } from '@/modules/core/repositories/streams'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 
 export const streamResourceCheckFactory =
   (deps: {
@@ -237,7 +237,7 @@ export const editCommentFactory =
 export const archiveCommentFactory =
   (deps: {
     getComment: GetComment
-    getStream: typeof getStream
+    getStream: GetStream
     updateComment: UpdateComment
   }) =>
   async ({

packages/server/modules/comments/services/management.ts

@@ -1,7 +1,6 @@
 import { ensureError, Roles, SpeckleViewer } from '@speckle/shared'
 import { AuthContext } from '@/modules/shared/authz'
 import { ForbiddenError } from '@/modules/shared/errors'
-import { getStream } from '@/modules/core/repositories/streams'
 import { StreamInvalidAccessError } from '@/modules/core/errors/stream'
 import {
   CreateCommentInput,
@@ -42,9 +41,10 @@ import {
   UpdateComment,
   ValidateInputAttachments
 } from '@/modules/comments/domain/operations'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 
 type AuthorizeProjectCommentsAccessDeps = {
-  getStream: typeof getStream
+  getStream: GetStream
   adminOverrideEnabled: typeof adminOverrideEnabled
 }
 
@@ -289,7 +289,7 @@ export const editCommentAndNotifyFactory =
 export const archiveCommentAndNotifyFactory =
   (deps: {
     getComment: GetComment
-    getStream: typeof getStream
+    getStream: GetStream
     updateComment: UpdateComment
     addCommentArchivedActivity: typeof addCommentArchivedActivity
   }): ArchiveCommentAndNotify =>

packages/server/modules/comments/tests/comments.spec.js

@@ -56,8 +56,8 @@ const { db } = require('@/db/knex')
 const { getBlobsFactory } = require('@/modules/blobstorage/repositories')
 const { CommentsEmitter } = require('@/modules/comments/events/emitter')
 const {
-  getStream,
-  markCommitStreamUpdated
+  markCommitStreamUpdated,
+  getStreamFactory
 } = require('@/modules/core/repositories/streams')
 const {
   createCommitByBranchIdFactory,
@@ -79,6 +79,7 @@ const {
 } = require('@/modules/activitystream/services/commitActivity')
 const { getObjectFactory } = require('@/modules/core/repositories/objects')
 
+const getStream = getStreamFactory({ db })
 const streamResourceCheck = streamResourceCheckFactory({
   checkStreamResourceAccess: checkStreamResourceAccessFactory({ db })
 })

packages/server/modules/core/domain/streams/operations.ts

@@ -0,0 +1,21 @@
+import { StreamWithOptionalRole } from '@/modules/core/domain/streams/types'
+import { Optional } from '@speckle/shared'
+import { Knex } from 'knex'
+
+export type GetStreams = (
+  streamIds: string[],
+  options?: Partial<{
+    userId: string
+    trx: Knex.Transaction
+  }>
+) => Promise<StreamWithOptionalRole[]>
+
+export type GetStream = (
+  params: {
+    streamId?: string
+    userId?: string
+  },
+  options?: Partial<{
+    trx: Knex.Transaction
+  }>
+) => Promise<Optional<StreamWithOptionalRole>>

packages/server/modules/core/domain/streams/types.ts

@@ -0,0 +1,12 @@
+import { StreamRecord } from '@/modules/core/helpers/types'
+import { StreamRoles } from '@speckle/shared'
+
+export type Stream = StreamRecord
+export type Project = Stream
+
+export type StreamWithOptionalRole = Stream & {
+  /**
+   * Available, if query joined this data StreamAcl
+   */
+  role?: StreamRoles
+}

packages/server/modules/core/graph/directives/hasRole.ts

@@ -18,8 +18,9 @@ import {
   getUserAclRoleFactory,
   getUserServerRoleFactory
 } from '@/modules/shared/repositories/acl'
-import { getStream } from '@/modules/core/repositories/streams'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 
+const getStream = getStreamFactory({ db })
 const throwForNotHavingServerRole = throwForNotHavingServerRoleFactory({
   validateServerRole: validateServerRoleBuilderFactory({
     getRoles: getRolesFactory({ db })

packages/server/modules/core/graph/resolvers/branches.js

@@ -33,8 +33,8 @@ const {
   addBranchDeletedActivity
 } = require('@/modules/activitystream/services/branchActivity')
 const {
-  getStream,
-  markBranchStreamUpdated
+  markBranchStreamUpdated,
+  getStreamFactory
 } = require('@/modules/core/repositories/streams')
 const { ModelsEmitter } = require('@/modules/core/events/modelsEmitter')
 
@@ -43,6 +43,7 @@ const BRANCH_CREATED = BranchPubsubEvents.BranchCreated
 const BRANCH_UPDATED = BranchPubsubEvents.BranchUpdated
 const BRANCH_DELETED = BranchPubsubEvents.BranchDeleted
 
+const getStream = getStreamFactory({ db })
 const getBranchById = getBranchByIdFactory({ db })
 const getStreamBranchByName = getStreamBranchByNameFactory({ db })
 const createBranchAndNotify = createBranchAndNotifyFactory({

packages/server/modules/core/graph/resolvers/commits.js

@@ -57,9 +57,9 @@ const {
 const { db } = require('@/db/knex')
 const {
   markCommitStreamUpdated,
-  getStream,
   getCommitStream,
-  getStreams
+  getStreamFactory,
+  getStreamsFactory
 } = require('@/modules/core/repositories/streams')
 const {
   markCommitBranchUpdatedFactory,
@@ -81,6 +81,8 @@ const COMMIT_CREATED = CommitPubsubEvents.CommitCreated
 const COMMIT_UPDATED = CommitPubsubEvents.CommitUpdated
 const COMMIT_DELETED = CommitPubsubEvents.CommitDeleted
 
+const getStream = getStreamFactory({ db })
+const getStreams = getStreamsFactory({ db })
 const deleteCommitAndNotify = deleteCommitAndNotifyFactory({
   getCommit: getCommitFactory({ db }),
   markCommitStreamUpdated,

packages/server/modules/core/graph/resolvers/models.ts

@@ -52,9 +52,13 @@ import {
   addBranchDeletedActivity,
   addBranchUpdatedActivity
 } from '@/modules/activitystream/services/branchActivity'
-import { getStream, markBranchStreamUpdated } from '@/modules/core/repositories/streams'
+import {
+  getStreamFactory,
+  markBranchStreamUpdated
+} from '@/modules/core/repositories/streams'
 import { ModelsEmitter } from '@/modules/core/events/modelsEmitter'
 
+const getStream = getStreamFactory({ db })
 const getStreamObjects = getStreamObjectsFactory({ db })
 const getViewerResourceGroups = getViewerResourceGroupsFactory({
   getStreamObjects,

packages/server/modules/core/graph/resolvers/projects.ts

@@ -1,3 +1,4 @@
+import { db } from '@/db/knex'
 import { RateLimitError } from '@/modules/core/errors/ratelimit'
 import { StreamNotFoundError } from '@/modules/core/errors/stream'
 import { WorkspacesModuleDisabledError } from '@/modules/core/errors/workspaces'
@@ -13,7 +14,7 @@ import {
   getUserStreamsCount,
   getUserStreams,
   getStreamCollaborators,
-  getStream
+  getStreamFactory
 } from '@/modules/core/repositories/streams'
 import {
   getRateLimitResult,
@@ -36,6 +37,8 @@ import {
 } from '@/modules/shared/utils/subscriptions'
 import { has } from 'lodash'
 
+const getStream = getStreamFactory({ db })
+
 export = {
   Query: {
     async project(_parent, args, context) {

packages/server/modules/core/graph/resolvers/streams.ts

@@ -1,5 +1,4 @@
 import {
-  getStream,
   getStreams,
   getStreamUsers,
   favoriteStream,
@@ -24,7 +23,8 @@ import { getDiscoverableStreams } from '@/modules/core/services/streams/discover
 import { get } from 'lodash'
 import {
   getUserStreamsCount,
-  getUserStreams
+  getUserStreams,
+  getStreamFactory
 } from '@/modules/core/repositories/streams'
 import {
   deleteStreamAndNotify,
@@ -50,6 +50,8 @@ import { getInvitationTargetUsersFactory } from '@/modules/serverinvites/service
 import { getUsers } from '@/modules/core/repositories/users'
 import { BadRequestError } from '@/modules/shared/errors'
 
+const getStream = getStreamFactory({ db })
+
 const getUserStreamsCore = async (
   forOtherUser: boolean,
   parent: { id: string },

packages/server/modules/core/graph/resolvers/versions.ts

@@ -41,8 +41,8 @@ import {
 } from '@/modules/core/repositories/branches'
 import {
   getCommitStream,
-  getStream,
-  getStreams,
+  getStreamFactory,
+  getStreamsFactory,
   markCommitStreamUpdated
 } from '@/modules/core/repositories/streams'
 import { VersionsEmitter } from '@/modules/core/events/versionsEmitter'
@@ -53,6 +53,8 @@ import {
 } from '@/modules/activitystream/services/commitActivity'
 import { getObjectFactory } from '@/modules/core/repositories/objects'
 
+const getStream = getStreamFactory({ db })
+const getStreams = getStreamsFactory({ db })
 const getObject = getObjectFactory({ db })
 const createCommitByBranchId = createCommitByBranchIdFactory({
   createCommit: createCommitFactory({ db }),

packages/server/modules/core/loaders.ts

@@ -3,12 +3,12 @@ import {
   getBatchUserFavoriteData,
   getBatchStreamFavoritesCounts,
   getOwnedFavoritesCountByUserIds,
-  getStreams,
   getStreamRoles,
   getStreamsSourceApps,
   getCommitStreams,
   StreamWithCommitId,
-  getUserStreamCounts
+  getUserStreamCounts,
+  getStreamsFactory
 } from '@/modules/core/repositories/streams'
 import { UserWithOptionalRole, getUsers } from '@/modules/core/repositories/users'
 import { keyBy } from 'lodash'
@@ -88,6 +88,7 @@ import { getAppScopesFactory } from '@/modules/auth/repositories'
 
 const simpleTupleCacheKey = (key: [string, string]) => `${key[0]}:${key[1]}`
 
+const getStreams = getStreamsFactory({ db })
 const getStreamPendingModels = getStreamPendingModelsFactory({ db })
 const getAppScopes = getAppScopesFactory({ db })
 const getAutomations = getAutomationsFactory({ db })

packages/server/modules/core/repositories/streams.ts

@@ -57,7 +57,7 @@ import {
 } from '@/modules/core/errors/stream'
 import { metaHelpers } from '@/modules/core/helpers/meta'
 import { removePrivateFields } from '@/modules/core/helpers/userHelper'
-import { db as defaultKnexInstance } from '@/db/knex'
+import { db, db as defaultKnexInstance } from '@/db/knex'
 import {
   DeleteProjectRole,
   GetProject,
@@ -66,17 +66,13 @@ import {
   GetRolesByUserId,
   UpsertProjectRole
 } from '@/modules/core/domain/projects/operations'
+import { StreamWithOptionalRole } from '@/modules/core/domain/streams/types'
+import { GetStream, GetStreams } from '@/modules/core/domain/streams/operations'
+export type { StreamWithOptionalRole }
 
 const tables = {
-  streams: (db: Knex) => db<StreamRecord>('streams'),
-  streamAcl: (db: Knex) => db<StreamAclRecord>('stream_acl')
-}
-
-export type StreamWithOptionalRole = StreamRecord & {
-  /**
-   * Available, if query joined this data StreamAcl
-   */
-  role?: StreamRoles
+  streams: (db: Knex) => db<StreamRecord>(Streams.name),
+  streamAcl: (db: Knex) => db<StreamAclRecord>(StreamAcl.name)
 }
 
 /**
@@ -130,56 +126,62 @@ const generateStreamName = () => {
 /**
  * Get multiple streams. If userId is specified, the role will be resolved as well.
  */
-export async function getStreams(
-  streamIds: string[],
-  options: Partial<{ userId: string; trx: Knex.Transaction }> = {}
-) {
-  const { userId, trx } = options
-  if (!streamIds?.length) throw new InvalidArgumentError('Empty stream IDs')
+export const getStreamsFactory =
+  (deps: { db: Knex }): GetStreams =>
+  async (
+    streamIds: string[],
+    options: Partial<{ userId: string; trx: Knex.Transaction }> = {}
+  ) => {
+    const { userId, trx } = options
+    if (!streamIds?.length) throw new InvalidArgumentError('Empty stream IDs')
 
-  const q = Streams.knex<StreamWithOptionalRole[]>().whereIn(Streams.col.id, streamIds)
+    const q = tables.streams(deps.db).whereIn(Streams.col.id, streamIds)
 
-  if (userId) {
-    q.select([
-      ...Object.values(Streams.col),
-      // Getting first role from grouped results
-      knex.raw(`(array_agg("stream_acl"."role"))[1] as role`)
-    ])
-    q.leftJoin(StreamAcl.name, function () {
-      this.on(StreamAcl.col.resourceId, Streams.col.id).andOnVal(
-        StreamAcl.col.userId,
-        userId
-      )
-    })
-    q.groupBy(Streams.col.id)
+    if (userId) {
+      q.select<StreamWithOptionalRole[]>([
+        ...Object.values(Streams.col),
+        // Getting first role from grouped results
+        knex.raw(`(array_agg("stream_acl"."role"))[1] as role`)
+      ])
+      q.leftJoin(StreamAcl.name, function () {
+        this.on(StreamAcl.col.resourceId, Streams.col.id).andOnVal(
+          StreamAcl.col.userId,
+          userId
+        )
+      })
+      q.groupBy(Streams.col.id)
+    }
+
+    if (trx) {
+      q.transacting(trx)
+    }
+
+    return await q
   }
 
-  if (trx) {
-    q.transacting(trx)
-  }
-
-  return await q
-}
-
 /**
  * Get a single stream. If userId is specified, the role will be resolved as well.
  */
-export async function getStream(
-  params: { streamId?: string; userId?: string },
-  options?: Partial<{ trx: Knex.Transaction }>
-): Promise<Optional<StreamWithOptionalRole>> {
-  const { streamId, userId } = params
-  if (!streamId) throw new InvalidArgumentError('Invalid stream ID')
+export const getStreamFactory =
+  (deps: { db: Knex }): GetStream =>
+  async (
+    params: { streamId?: string; userId?: string },
+    options?: Partial<{ trx: Knex.Transaction }>
+  ): Promise<Optional<StreamWithOptionalRole>> => {
+    const { streamId, userId } = params
+    if (!streamId) throw new InvalidArgumentError('Invalid stream ID')
 
-  const streams = await getStreams([streamId], { userId, ...(options || {}) })
-  return <Optional<StreamWithOptionalRole>>streams[0]
-}
+    const streams = await getStreamsFactory(deps)([streamId], {
+      userId,
+      ...(options || {})
+    })
+    return <Optional<StreamWithOptionalRole>>streams[0]
+  }
 
-// TODO: Inject db
 export const getProjectFactory =
-  (): GetProject =>
+  (deps: { db: Knex }): GetProject =>
   async ({ projectId }) => {
-    const project = await getStream({ streamId: projectId })
+    const project = await getStreamFactory(deps)({ streamId: projectId })
 
     if (!project) {
       throw new StreamNotFoundError()
@@ -1181,7 +1183,7 @@ export async function revokeStreamPermissions(
  * Mark stream as the onboarding base stream from which user onboarding streams will be cloned
  */
 export async function markOnboardingBaseStream(streamId: string, version: string) {
-  const stream = await getStream({ streamId })
+  const stream = await getStreamFactory({ db })({ streamId })
   if (!stream) {
     throw new Error(`Stream ${streamId} not found`)
   }

packages/server/modules/core/rest/authUtils.js

@@ -1,13 +1,15 @@
 'use strict'
 const { validateScopes, authorizeResolver } = require('@/modules/shared')
 
-const { getStream } = require('../services/streams')
 const { Roles, Scopes } = require('@speckle/shared')
 const { throwForNotHavingServerRole } = require('@/modules/shared/authz')
 const { DatabaseError } = require('@/modules/shared/errors')
+const { getStreamFactory } = require('@/modules/core/repositories/streams')
+const { db } = require('@/db/knex')
 
 module.exports = {
   async validatePermissionsReadStream(streamId, req) {
+    const getStream = getStreamFactory({ db })
     const stream = await getStream({ streamId, userId: req.context.userId })
     if (stream?.isPublic) return { result: true, status: 200 }
 

packages/server/modules/core/services/branch/management.ts

@@ -19,7 +19,7 @@ import {
 } from '@/modules/core/graph/generated/graphql'
 import { BranchRecord } from '@/modules/core/helpers/types'
 
-import { getStream, markBranchStreamUpdated } from '@/modules/core/repositories/streams'
+import { markBranchStreamUpdated } from '@/modules/core/repositories/streams'
 import { has } from 'lodash'
 import { isBranchDeleteInput, isBranchUpdateInput } from '@/modules/core/helpers/branch'
 import { ModelsEmitter, ModelsEventsEmitter } from '@/modules/core/events/modelsEmitter'
@@ -33,6 +33,7 @@ import {
   UpdateBranch,
   UpdateBranchAndNotify
 } from '@/modules/core/domain/branches/operations'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 
 const isBranchCreateInput = (
   i: BranchCreateInput | CreateModelInput
@@ -109,7 +110,7 @@ export const updateBranchAndNotifyFactory =
 
 export const deleteBranchAndNotifyFactory =
   (deps: {
-    getStream: typeof getStream
+    getStream: GetStream
     getBranchById: GetBranchById
     modelsEventsEmitter: ModelsEventsEmitter
     markBranchStreamUpdated: typeof markBranchStreamUpdated

packages/server/modules/core/services/commit/batchCommitActions.ts

@@ -12,6 +12,7 @@ import {
   MoveCommitsToBranch,
   ValidateAndBatchMoveCommits
 } from '@/modules/core/domain/commits/operations'
+import { GetStreams } from '@/modules/core/domain/streams/operations'
 import {
   CommitInvalidAccessError,
   CommitBatchUpdateError
@@ -27,7 +28,7 @@ import {
   deleteCommitsFactory,
   getCommitsFactory
 } from '@/modules/core/repositories/commits'
-import { getStreams } from '@/modules/core/repositories/streams'
+import { getStreamsFactory } from '@/modules/core/repositories/streams'
 import { ensureError } from '@/modules/shared/helpers/errorHelper'
 import { difference, groupBy, has, keyBy } from 'lodash'
 
@@ -38,7 +39,7 @@ const isOldBatchInput = (i: CommitBatchInput): i is OldBatchInput => has(i, 'com
 
 type ValidateBatchBaseRulesDeps = {
   getCommits: GetCommits
-  getStreams: typeof getStreams
+  getStreams: GetStreams
 }
 
 /**
@@ -147,7 +148,7 @@ async function validateCommitsDelete(
 ) {
   const validateBatchBaseRules = validateBatchBaseRulesFactory({
     getCommits: getCommitsFactory({ db }),
-    getStreams
+    getStreams: getStreamsFactory({ db })
   })
   return await validateBatchBaseRules(params, userId)
 }

packages/server/modules/core/services/commit/management.ts

@@ -25,6 +25,7 @@ import {
   UpdateCommitAndNotify
 } from '@/modules/core/domain/commits/operations'
 import { GetObject } from '@/modules/core/domain/objects/operations'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 import {
   CommitCreateError,
   CommitDeleteError,
@@ -45,7 +46,6 @@ import { CommitRecord } from '@/modules/core/helpers/types'
 import { getCommitFactory } from '@/modules/core/repositories/commits'
 import {
   getCommitStream,
-  getStream,
   markCommitStreamUpdated
 } from '@/modules/core/repositories/streams'
 import { ensureError, MaybeNullOrUndefined, Nullable, Roles } from '@speckle/shared'
@@ -253,7 +253,7 @@ const isOldVersionUpdateInput = (
 export const updateCommitAndNotifyFactory =
   (deps: {
     getCommit: GetCommit
-    getStream: typeof getStream
+    getStream: GetStream
     getCommitStream: typeof getCommitStream
     getStreamBranchByName: GetStreamBranchByName
     getCommitBranch: GetCommitBranch

packages/server/modules/core/services/streams.js

@@ -2,7 +2,6 @@
 const _ = require('lodash')
 const { Streams, StreamAcl, knex } = require('@/modules/core/dbSchema')
 const {
-  getStream,
   getFavoritedStreams,
   getFavoritedStreamsCount,
   setStreamFavorited,
@@ -10,7 +9,8 @@ const {
   deleteStream: deleteStreamFromDb,
   updateStream: updateStreamInDb,
   revokeStreamPermissions,
-  grantStreamPermissions
+  grantStreamPermissions,
+  getStreamFactory
 } = require('@/modules/core/repositories/streams')
 const { UnauthorizedError, InvalidArgumentError } = require('@/modules/shared/errors')
 const { dbLogger } = require('@/logging/logging')
@@ -42,8 +42,6 @@ module.exports = {
     return id
   },
 
-  getStream,
-
   /**
    * @deprecated Use updateStreamAndNotify or use the repository function directly
    * @param {import('@/modules/core/graph/generated/graphql').StreamUpdateInput} update
@@ -205,6 +203,8 @@ module.exports = {
     // Favorite/unfavorite the stream
     await setStreamFavorited({ streamId, userId, favorited })
 
+    const getStream = getStreamFactory({ db: knex })
+
     // Get updated stream info
     return await getStream({ streamId, userId })
   },

packages/server/modules/core/services/streams/clone.ts

@@ -6,7 +6,7 @@ import {
 } from '@/modules/core/helpers/types'
 import {
   createStream,
-  getStream,
+  getStreamFactory,
   StreamWithOptionalRole
 } from '@/modules/core/repositories/streams'
 import { getUser, UserWithOptionalRole } from '@/modules/core/repositories/users'
@@ -82,6 +82,7 @@ const prepareState = async (
   userId: string,
   sourceStreamId: string
 ): Promise<CloneStreamInitialState> => {
+  const getStream = getStreamFactory({ db })
   const targetStream = await getStream({ streamId: sourceStreamId })
   if (!targetStream) {
     throw new StreamCloneError('Clonable source stream not found', {

packages/server/modules/core/services/streams/management.ts

@@ -17,7 +17,7 @@ import { StreamRecord } from '@/modules/core/helpers/types'
 import {
   createStream,
   deleteStream,
-  getStream,
+  getStreamFactory,
   updateStream
 } from '@/modules/core/repositories/streams'
 import { createAndSendInviteFactory } from '@/modules/serverinvites/services/creation'
@@ -92,6 +92,7 @@ export async function createStreamReturnRecord(
   // Invite contributors?
   if (!isProjectCreateInput(params) && params.withContributors?.length) {
     // TODO: should be injected in the resolver
+    const getStream = getStreamFactory({ db })
     await inviteUsersToProjectFactory({
       createAndSendInvite: createAndSendInviteFactory({
         findUserByTarget: findUserByTargetFactory(),
@@ -188,6 +189,7 @@ export async function updateStreamAndNotify(
     updaterResourceAccessRules
   )
 
+  const getStream = getStreamFactory({ db })
   const oldStream = await getStream({ streamId: update.id, userId: updaterId })
   if (!oldStream) {
     throw new StreamUpdateError('Stream not found', {

packages/server/modules/core/services/streams/streamAccessService.js

@@ -16,9 +16,9 @@ const {
   addStreamPermissionsAddedActivityFactory
 } = require('@/modules/activitystream/services/streamActivity')
 const {
-  getStream,
   revokeStreamPermissions,
-  grantStreamPermissions
+  grantStreamPermissions,
+  getStreamFactory
 } = require('@/modules/core/repositories/streams')
 
 const { ServerAcl } = require('@/modules/core/dbSchema')
@@ -34,6 +34,7 @@ const { db } = require('@/db/knex')
  * @returns
  */
 async function isStreamCollaborator(userId, streamId) {
+  const getStream = getStreamFactory({ db })
   const stream = await getStream({ streamId, userId })
   return !!stream.role
 }

packages/server/modules/core/tests/branches.spec.js

@@ -30,9 +30,9 @@ const {
   addBranchDeletedActivity
 } = require('@/modules/activitystream/services/branchActivity')
 const {
-  getStream,
   markBranchStreamUpdated,
-  markCommitStreamUpdated
+  markCommitStreamUpdated,
+  getStreamFactory
 } = require('@/modules/core/repositories/streams')
 const { ModelsEmitter } = require('@/modules/core/events/modelsEmitter')
 const {
@@ -52,6 +52,7 @@ const { getObjectFactory } = require('@/modules/core/repositories/objects')
 
 const db = knex
 const Commits = () => knex('commits')
+const getStream = getStreamFactory({ db: knex })
 const getBranchById = getBranchByIdFactory({ db: knex })
 const getStreamBranchByName = getStreamBranchByNameFactory({ db: knex })
 const createBranch = createBranchFactory({ db: knex })

packages/server/modules/core/tests/commits.spec.js

@@ -47,7 +47,7 @@ const {
 const {
   markCommitStreamUpdated,
   getCommitStream,
-  getStream
+  getStreamFactory
 } = require('@/modules/core/repositories/streams')
 const {
   addCommitDeletedActivity,
@@ -57,6 +57,7 @@ const {
 const { VersionsEmitter } = require('@/modules/core/events/versionsEmitter')
 const { getObjectFactory } = require('@/modules/core/repositories/objects')
 
+const getStream = getStreamFactory({ db })
 const createBranch = createBranchFactory({ db })
 const createBranchAndNotify = createBranchAndNotifyFactory({
   createBranch,

packages/server/modules/core/tests/discoverableStreams.spec.ts

@@ -1,6 +1,10 @@
 import { buildApolloServer } from '@/app'
+import { db } from '@/db/knex'
 import { Streams, Users } from '@/modules/core/dbSchema'
-import { getStream, setStreamFavorited } from '@/modules/core/repositories/streams'
+import {
+  getStreamFactory,
+  setStreamFavorited
+} from '@/modules/core/repositories/streams'
 import { Nullable, Optional } from '@/modules/shared/helpers/typeHelper'
 import { BasicTestUser, createTestUsers } from '@/test/authHelper'
 import {
@@ -27,6 +31,7 @@ import { shuffle } from 'lodash'
 const READABLE_DISCOVERABLE_STREAM_COUNT = 15
 
 const cleanup = async () => await truncateTables([Streams.name, Users.name])
+const getStream = getStreamFactory({ db })
 
 describe('Discoverable streams', () => {
   let apollo: ServerAndContext

packages/server/modules/core/tests/streams.spec.ts

@@ -1,7 +1,6 @@
 import { expect } from 'chai'
 import {
   createStream,
-  getStream,
   updateStream,
   deleteStream,
   getStreamUsers,
@@ -29,6 +28,7 @@ import {
 } from '@/test/speckle-helpers/streamHelper'
 import {
   StreamWithOptionalRole,
+  getStreamFactory,
   markBranchStreamUpdated,
   markCommitStreamUpdated,
   revokeStreamPermissions
@@ -74,6 +74,7 @@ import { VersionsEmitter } from '@/modules/core/events/versionsEmitter'
 import { addCommitCreatedActivity } from '@/modules/activitystream/services/commitActivity'
 import { getObjectFactory } from '@/modules/core/repositories/objects'
 
+const getStream = getStreamFactory({ db })
 const getStreamBranchByName = getStreamBranchByNameFactory({ db })
 const createBranch = createBranchFactory({ db })
 const deleteBranchAndNotify = deleteBranchAndNotifyFactory({

packages/server/modules/core/tests/users.spec.js

@@ -21,11 +21,7 @@ const {
   validateToken,
   getUserTokens
 } = require('../services/tokens')
-const {
-  grantPermissionsStream,
-  createStream,
-  getStream
-} = require('../services/streams')
+const { grantPermissionsStream, createStream } = require('../services/streams')
 
 const { getBranchesByStreamId } = require('../services/branches')
 
@@ -52,13 +48,17 @@ const {
   createCommitByBranchIdFactory,
   createCommitByBranchNameFactory
 } = require('@/modules/core/services/commit/management')
-const { markCommitStreamUpdated } = require('@/modules/core/repositories/streams')
+const {
+  markCommitStreamUpdated,
+  getStreamFactory
+} = require('@/modules/core/repositories/streams')
 const { VersionsEmitter } = require('@/modules/core/events/versionsEmitter')
 const {
   addCommitCreatedActivity
 } = require('@/modules/activitystream/services/commitActivity')
 const { getObjectFactory } = require('@/modules/core/repositories/objects')
 
+const getStream = getStreamFactory({ db })
 const createBranch = createBranchFactory({ db })
 const getCommit = getCommitFactory({ db })
 

packages/server/modules/cross-server-sync/index.ts

@@ -42,8 +42,8 @@ import {
 } from '@/modules/core/repositories/objects'
 import {
   getOnboardingBaseStream,
-  getStream,
   getStreamCollaborators,
+  getStreamFactory,
   markCommitStreamUpdated,
   markOnboardingBaseStream
 } from '@/modules/core/repositories/streams'
@@ -68,6 +68,7 @@ const crossServerSyncModule: SpeckleModule = {
   finalize() {
     crossServerSyncLogger.info('⬇️  Ensuring base onboarding stream asynchronously...')
 
+    const getStream = getStreamFactory({ db })
     const getObject = getObjectFactory({ db })
     const getStreamObjects = getStreamObjectsFactory({ db })
     const markCommentViewed = markCommentViewedFactory({ db })

packages/server/modules/cross-server-sync/services/commit.ts

@@ -2,7 +2,7 @@ import fetch from 'cross-fetch'
 import { ApolloClient, NormalizedCacheObject, gql } from '@apollo/client/core'
 import { getFrontendOrigin } from '@/modules/shared/helpers/envHelper'
 import { CreateCommentInput } from '@/test/graphql/generated/graphql'
-import { getStream, getStreamCollaborators } from '@/modules/core/repositories/streams'
+import { getStreamCollaborators } from '@/modules/core/repositories/streams'
 import { Roles, timeoutAt } from '@speckle/shared'
 import { createObject } from '@/modules/core/services/objects'
 import ObjectLoader from '@speckle/objectloader'
@@ -31,6 +31,7 @@ import {
 import { GetStreamBranchByName } from '@/modules/core/domain/branches/operations'
 import { CreateCommitByBranchId } from '@/modules/core/domain/commits/operations'
 import { GetObject } from '@/modules/core/domain/objects/operations'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 
 type LocalResources = Awaited<ReturnType<ReturnType<typeof getLocalResourcesFactory>>>
 type LocalResourcesWithCommit = LocalResources & { newCommitId: string }
@@ -218,7 +219,7 @@ const parseIncomingUrl = async (url: string, token?: string) => {
 }
 
 type GetLocalResourcesDeps = {
-  getStream: typeof getStream
+  getStream: GetStream
   getStreamBranchByName: GetStreamBranchByName
   getStreamCollaborators: typeof getStreamCollaborators
   getUser: typeof getUser

packages/server/modules/fileuploads/index.ts

@@ -12,16 +12,17 @@ import { db } from '@/db/knex'
 import { publish } from '@/modules/shared/utils/subscriptions'
 import { SpeckleModule } from '@/modules/shared/helpers/typeHelper'
 import { streamWritePermissionsPipelineFactory } from '@/modules/shared/authz'
-import { getStream } from '@/modules/core/repositories/streams'
 import { getRolesFactory } from '@/modules/shared/repositories/roles'
 import { getAutomationProjectFactory } from '@/modules/automate/repositories/automations'
 import { getStreamBranchByNameFactory } from '@/modules/core/repositories/branches'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 
 const insertNewUploadAndNotify = insertNewUploadAndNotifyFactory({
   getStreamBranchByName: getStreamBranchByNameFactory({ db }),
   saveUploadFile: saveUploadFileFactory({ db }),
   publish
 })
+const getStream = getStreamFactory({ db })
 
 const saveFileUploads = async ({
   userId,

packages/server/modules/notifications/services/handlers/activityDigest.ts

@@ -30,8 +30,8 @@ import {
   StreamActivitySummary
 } from '@/modules/activitystream/domain/types'
 import { createActivitySummaryFactory } from '@/modules/activitystream/services/summary'
-import { getStream } from '@/modules/core/services/streams'
 import { getActivityFactory } from '@/modules/activitystream/repositories'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 
 const digestNotificationEmailHandlerFactory =
   (
@@ -436,7 +436,7 @@ const digestNotificationEmailHandler = digestNotificationEmailHandlerFactory({
     })
   }),
   createActivitySummary: createActivitySummaryFactory({
-    getStream,
+    getStream: getStreamFactory({ db }),
     getActivity: getActivityFactory({ db })
   }),
   getServerInfo,

packages/server/modules/notifications/services/handlers/mentionedInComment.ts

@@ -2,10 +2,12 @@ import { db } from '@/db/knex'
 import { GetComment } from '@/modules/comments/domain/operations'
 import { ExtendedComment } from '@/modules/comments/domain/types'
 import { getCommentFactory } from '@/modules/comments/repositories/comments'
+import { GetStream } from '@/modules/core/domain/streams/operations'
+import { StreamWithOptionalRole } from '@/modules/core/domain/streams/types'
 import { Roles } from '@/modules/core/helpers/mainConstants'
 import { getCommentRoute } from '@/modules/core/helpers/routeHelper'
 import { ServerInfo } from '@/modules/core/helpers/types'
-import { getStream, StreamWithOptionalRole } from '@/modules/core/repositories/streams'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 import { getUser, UserWithOptionalRole } from '@/modules/core/repositories/users'
 import { getServerInfo } from '@/modules/core/services/generic'
 import {
@@ -163,7 +165,7 @@ function buildEmailTemplateParams(
 const mentionedInCommentHandlerFactory =
   (deps: {
     getUser: typeof getUser
-    getStream: typeof getStream
+    getStream: GetStream
     getComment: GetComment
     getServerInfo: typeof getServerInfo
     renderEmail: typeof renderEmail
@@ -220,7 +222,7 @@ const mentionedInCommentHandlerFactory =
 const handler: NotificationHandler<MentionedInCommentMessage> = async (...args) => {
   const mentionedInCommentHandler = mentionedInCommentHandlerFactory({
     getUser,
-    getStream,
+    getStream: getStreamFactory({ db }),
     getComment: getCommentFactory({ db }),
     getServerInfo,
     renderEmail,

packages/server/modules/notifications/services/handlers/newStreamAccessRequest.ts

@@ -8,7 +8,6 @@ import {
   NotificationHandler
 } from '@/modules/notifications/helpers/types'
 import { NotificationValidationError } from '@/modules/notifications/errors'
-import { getStream } from '@/modules/core/repositories/streams'
 import { Roles } from '@/modules/core/helpers/mainConstants'
 import {
   buildAbsoluteFrontendUrlFromPath,
@@ -22,11 +21,13 @@ import {
 import { getServerInfo } from '@/modules/core/services/generic'
 import { db } from '@/db/knex'
 import { GetPendingAccessRequest } from '@/modules/accessrequests/domain/operations'
+import { GetStream } from '@/modules/core/domain/streams/operations'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 
 type ValidateMessageDeps = {
   getPendingAccessRequest: GetPendingAccessRequest
   getUser: typeof getUser
-  getStream: typeof getStream
+  getStream: GetStream
 }
 
 const validateMessageFactory =
@@ -150,7 +151,7 @@ const handler: NotificationHandler<NewStreamAccessRequestMessage> = (...args) =>
     renderEmail,
     sendEmail,
     getUser,
-    getStream,
+    getStream: getStreamFactory({ db }),
     getPendingAccessRequest: getPendingAccessRequestFactory({ db })
   })
   return newStreamAccessRequestHandler(...args)

packages/server/modules/notifications/services/handlers/streamAccessRequestApproved.ts

@@ -1,8 +1,10 @@
+import { db } from '@/db/knex'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 import {
   buildAbsoluteFrontendUrlFromPath,
   getStreamRoute
 } from '@/modules/core/helpers/routeHelper'
-import { getStream } from '@/modules/core/repositories/streams'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 import { getUser } from '@/modules/core/repositories/users'
 import { getServerInfo } from '@/modules/core/services/generic'
 import {
@@ -18,7 +20,7 @@ import {
 
 type ValidateMessageDeps = {
   getUser: typeof getUser
-  getStream: typeof getStream
+  getStream: GetStream
 }
 
 const validateMessageFactory =
@@ -129,7 +131,7 @@ const handler: NotificationHandler<StreamAccessRequestApprovedMessage> = async (
     renderEmail,
     sendEmail,
     getUser,
-    getStream
+    getStream: getStreamFactory({ db })
   })
   return streamAccessRequestApprovedHandler(...args)
 }

packages/server/modules/previews/index.ts

@@ -18,7 +18,6 @@ import {
   checkStreamPermissionsFactory
 } from '@/modules/previews/services/management'
 import { getObject } from '@/modules/core/services/objects'
-import { getStream } from '@/modules/core/repositories/streams'
 import {
   getObjectPreviewInfoFactory,
   createObjectPreviewFactory,
@@ -30,6 +29,7 @@ import {
   getObjectCommitsWithStreamIdsFactory
 } from '@/modules/core/repositories/commits'
 import { SpeckleModule } from '@/modules/shared/helpers/typeHelper'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 
 const httpErrorImage = (httpErrorCode: number) =>
   require.resolve(`#/assets/previews/images/preview_${httpErrorCode}.png`)
@@ -43,6 +43,7 @@ export const init: SpeckleModule['init'] = (app, isInitial) => {
     moduleLogger.info('📸 Init object preview module')
   }
 
+  const getStream = getStreamFactory({ db })
   const getObjectPreviewBufferOrFilepath = getObjectPreviewBufferOrFilepathFactory({
     getObject,
     getObjectPreviewInfo: getObjectPreviewInfoFactory({ db }),
@@ -56,7 +57,8 @@ export const init: SpeckleModule['init'] = (app, isInitial) => {
   })
   const checkStreamPermissions = checkStreamPermissionsFactory({
     validateScopes,
-    authorizeResolver
+    authorizeResolver,
+    getStream
   })
 
   app.options('/preview/:streamId/:angle?', cors())

packages/server/modules/previews/services/management.ts

@@ -1,5 +1,5 @@
 import { logger } from '@/logging/logging'
-import { getStream } from '@/modules/core/repositories/streams'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 import { getObject } from '@/modules/core/services/objects'
 import {
   CheckStreamPermissions,
@@ -83,7 +83,7 @@ export const getObjectPreviewBufferOrFilepathFactory =
 export const sendObjectPreviewFactory =
   (deps: {
     getObjectPreviewBufferOrFilepath: GetObjectPreviewBufferOrFilepath
-    getStream: typeof getStream
+    getStream: GetStream
     makeOgImage: typeof makeOgImage
   }): SendObjectPreview =>
   async (req, res, streamId, objectId, angle) => {
@@ -136,9 +136,10 @@ export const checkStreamPermissionsFactory =
   (deps: {
     validateScopes: typeof validateScopes
     authorizeResolver: typeof authorizeResolver
+    getStream: GetStream
   }): CheckStreamPermissions =>
   async (req) => {
-    const stream = await getStream({
+    const stream = await deps.getStream({
       streamId: req.params.streamId,
       userId: req.context.userId
     })

packages/server/modules/serverinvites/graph/resolvers/serverInvites.ts

@@ -43,7 +43,6 @@ import {
 } from '@/modules/serverinvites/services/projectInviteManagement'
 import { getUser, getUsers } from '@/modules/core/repositories/users'
 import { collectAndValidateCoreTargetsFactory } from '@/modules/serverinvites/services/coreResourceCollection'
-import { getStream } from '@/modules/core/repositories/streams'
 import { buildCoreInviteEmailContentsFactory } from '@/modules/serverinvites/services/coreEmailContents'
 import { getEventBus } from '@/modules/shared/services/eventBus'
 import {
@@ -73,7 +72,9 @@ import { renderEmail } from '@/modules/emails/services/emailRendering'
 import { sendEmail } from '@/modules/emails/services/sending'
 import { publish } from '@/modules/shared/utils/subscriptions'
 import { saveActivityFactory } from '@/modules/activitystream/repositories'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 
+const getStream = getStreamFactory({ db })
 const requestNewEmailVerification = requestNewEmailVerificationFactory({
   findEmail: findEmailFactory({ db }),
   getUser,

packages/server/modules/serverinvites/services/coreEmailContents.ts

@@ -1,9 +1,9 @@
+import { GetStream } from '@/modules/core/domain/streams/operations'
 import {
   getRegistrationRoute,
   getStreamRoute
 } from '@/modules/core/helpers/routeHelper'
 import { StreamRecord } from '@/modules/core/helpers/types'
-import { getStream } from '@/modules/core/repositories/streams'
 import {
   EmailTemplateParams,
   sanitizeMessage
@@ -162,7 +162,7 @@ const buildProjectEmailTemplateParams = (
   }
 }
 
-type BuildProjectInviteContentsFactoryDeps = { getStream: typeof getStream }
+type BuildProjectInviteContentsFactoryDeps = { getStream: GetStream }
 
 const buildProjectInviteContentsFactory =
   (deps: BuildProjectInviteContentsFactoryDeps): BuildProjectInviteEmailContents =>

packages/server/modules/serverinvites/services/coreFinalization.ts

@@ -1,7 +1,7 @@
 import { AddStreamInviteDeclinedActivity } from '@/modules/activitystream/domain/operations'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 import { StreamInvalidAccessError } from '@/modules/core/errors/stream'
 import { isResourceAllowed } from '@/modules/core/helpers/token'
-import { getStream } from '@/modules/core/repositories/streams'
 import { addOrUpdateStreamCollaborator } from '@/modules/core/services/streams/streamAccessService'
 import { ProjectInviteResourceType } from '@/modules/serverinvites/domain/constants'
 import { InviteFinalizingError } from '@/modules/serverinvites/errors'
@@ -13,7 +13,7 @@ import {
 import { Roles } from '@speckle/shared'
 
 type ValidateProjectInviteBeforeFinalizationFactoryDeps = {
-  getProject: typeof getStream
+  getProject: GetStream
 }
 
 export const validateProjectInviteBeforeFinalizationFactory =
@@ -74,7 +74,7 @@ export const validateProjectInviteBeforeFinalizationFactory =
   }
 
 type ProcessFinalizedProjectInviteFactoryDeps = {
-  getProject: typeof getStream
+  getProject: GetStream
   addInviteDeclinedActivity: AddStreamInviteDeclinedActivity
   addProjectRole: typeof addOrUpdateStreamCollaborator
 }

packages/server/modules/serverinvites/services/coreResourceCollection.ts

@@ -1,4 +1,3 @@
-import { getStream } from '@/modules/core/repositories/streams'
 import {
   ProjectInviteResourceType,
   ServerInviteResourceType
@@ -13,6 +12,7 @@ import {
 import { authorizeResolver } from '@/modules/shared'
 import { Roles } from '@speckle/shared'
 import { flatten } from 'lodash'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 
 const collectAndValidateServerTargetFactory =
   (): CollectAndValidateResourceTargets => (params) => {
@@ -67,7 +67,7 @@ const collectAndValidateServerTargetFactory =
   }
 
 type CollectAndValidateProjectTargetFactoryDeps = {
-  getStream: typeof getStream
+  getStream: GetStream
 }
 
 const collectAndValidateProjectTargetFactory =

packages/server/modules/serverinvites/services/projectInviteManagement.ts

@@ -1,3 +1,4 @@
+import { GetStream } from '@/modules/core/domain/streams/operations'
 import { TokenResourceIdentifier } from '@/modules/core/domain/tokens/types'
 import {
   MutationStreamInviteUseArgs,
@@ -8,7 +9,6 @@ import {
 import { ContextResourceAccessRules } from '@/modules/core/helpers/token'
 import { LimitedUserRecord } from '@/modules/core/helpers/types'
 import { removePrivateFields } from '@/modules/core/helpers/userHelper'
-import { getStream } from '@/modules/core/repositories/streams'
 import { getUser, getUsers } from '@/modules/core/repositories/users'
 import {
   ProjectInviteResourceType,
@@ -58,7 +58,7 @@ const isStreamInviteCreateInput = (
 ): i is StreamInviteCreateInput => has(i, 'streamId')
 
 export const createProjectInviteFactory =
-  (deps: { createAndSendInvite: CreateAndSendInvite; getStream: typeof getStream }) =>
+  (deps: { createAndSendInvite: CreateAndSendInvite; getStream: GetStream }) =>
   async (params: {
     input: StreamInviteCreateInput | FullProjectInviteCreateInput
     inviterId: string

packages/server/modules/shared/index.ts

@@ -1,5 +1,5 @@
 import { db } from '@/db/knex'
-import { getStream } from '@/modules/core/repositories/streams'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 import { adminOverrideEnabled } from '@/modules/shared/helpers/envHelper'
 import {
   getUserAclRoleFactory,
@@ -29,6 +29,6 @@ export const authorizeResolver = authorizeResolverFactory({
   getRoles: getRolesFactory({ db }),
   adminOverrideEnabled,
   getUserServerRole: getUserServerRoleFactory({ db }),
-  getStream,
+  getStream: getStreamFactory({ db }),
   getUserAclRole: getUserAclRoleFactory({ db })
 })

packages/server/modules/shared/services/auth.ts

@@ -1,9 +1,9 @@
+import { GetStream } from '@/modules/core/domain/streams/operations'
 import {
   isResourceAllowed,
   RoleResourceTargets,
   roleResourceTypeToTokenResourceType
 } from '@/modules/core/helpers/token'
-import { getStream } from '@/modules/core/repositories/streams'
 import {
   AuthorizeResolver,
   GetUserAclRole,
@@ -36,7 +36,7 @@ export const authorizeResolverFactory =
     getRoles: GetRoles
     adminOverrideEnabled: typeof adminOverrideEnabled
     getUserServerRole: GetUserServerRole
-    getStream: typeof getStream
+    getStream: GetStream
     getUserAclRole: GetUserAclRole
   }): AuthorizeResolver =>
   async (userId, resourceId, requiredRole, userResourceAccessLimits) => {

packages/server/modules/webhooks/services/webhooks.ts

@@ -11,16 +11,14 @@ import {
 import { Webhook } from '@/modules/webhooks/domain/types'
 import { SetValuesNullable } from '@speckle/shared'
 import crs from 'crypto-random-string'
-import {
-  StreamWithOptionalRole,
-  getStream as getStreamFn
-} from '@/modules/core/repositories/streams'
+import { StreamWithOptionalRole } from '@/modules/core/repositories/streams'
 import {
   getUser as getUserFn,
   UserWithOptionalRole
 } from '@/modules/core/repositories/users'
 import { Knex } from 'knex'
 import { ServerInfo } from '@/modules/core/helpers/types'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 
 const MAX_STREAM_WEBHOOKS = 100
 
@@ -110,7 +108,7 @@ export const dispatchStreamEventFactory =
   }: {
     db: Knex // TODO: this should not be injected here
     getServerInfo: typeof getServerInfoFn
-    getStream: typeof getStreamFn
+    getStream: GetStream
     createWebhookEvent: CreateWebhookEvent
     getUser: typeof getUserFn
   }) =>

packages/server/modules/webhooks/tests/webhooks.spec.js

@@ -31,9 +31,10 @@ const {
 } = require('@/modules/webhooks/services/webhooks')
 const { Users, Streams } = require('@/modules/core/dbSchema')
 const { getServerInfo } = require('@/modules/core/services/generic')
-const { getStream } = require('@/modules/core/repositories/streams')
 const { getUser } = require('@/modules/core/repositories/users')
+const { getStreamFactory } = require('@/modules/core/repositories/streams')
 
+const getStream = getStreamFactory({ db })
 const updateWebhook = updateWebhookFactory({
   updateWebhookConfig: updateWebhookConfigFactory({ db })
 })

packages/server/modules/workspaces/events/eventListener.ts

@@ -5,7 +5,7 @@ import {
 } from '@/modules/core/events/projectsEmitter'
 import {
   deleteProjectRoleFactory,
-  getStream,
+  getStreamFactory,
   upsertProjectRoleFactory
 } from '@/modules/core/repositories/streams'
 import {
@@ -45,6 +45,7 @@ import {
 import { getStreams } from '@/modules/core/services/streams'
 import { withTransaction } from '@/modules/shared/helpers/dbHelper'
 import { findVerifiedEmailsByUserIdFactory } from '@/modules/core/repositories/userEmails'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 
 export const onProjectCreatedFactory =
   ({
@@ -89,7 +90,7 @@ export const onProjectCreatedFactory =
 
 export const onInviteFinalizedFactory =
   (deps: {
-    getStream: typeof getStream
+    getStream: GetStream
     logger: typeof logger
     updateWorkspaceRole: ReturnType<typeof updateWorkspaceRoleFactory>
   }) =>
@@ -230,7 +231,7 @@ export const initializeEventListenersFactory =
       }),
       eventBus.listen(ServerInvitesEvents.Finalized, async ({ payload }) => {
         const onInviteFinalized = onInviteFinalizedFactory({
-          getStream,
+          getStream: getStreamFactory({ db }),
           logger: moduleLogger,
           updateWorkspaceRole: updateWorkspaceRoleFactory({
             getWorkspaceWithDomains: getWorkspaceWithDomainsFactory({ db }),

packages/server/modules/workspaces/graph/resolvers/workspaces.ts

@@ -4,12 +4,12 @@ import { removePrivateFields } from '@/modules/core/helpers/userHelper'
 import {
   getProjectCollaboratorsFactory,
   getProjectFactory,
-  getStream,
   getUserStreams,
   getUserStreamsCount,
   updateProjectFactory,
   upsertProjectRoleFactory,
-  getRolesByUserIdFactory
+  getRolesByUserIdFactory,
+  getStreamFactory
 } from '@/modules/core/repositories/streams'
 import { getUser, getUsers } from '@/modules/core/repositories/users'
 import { getStreams } from '@/modules/core/services/streams'
@@ -136,6 +136,7 @@ import { renderEmail } from '@/modules/emails/services/emailRendering'
 import { sendEmail } from '@/modules/emails/services/sending'
 import { parseDefaultProjectRole } from '@/modules/workspaces/domain/logic'
 
+const getStream = getStreamFactory({ db })
 const requestNewEmailVerification = requestNewEmailVerificationFactory({
   findEmail: findEmailFactory({ db }),
   getUser,
@@ -724,7 +725,7 @@ export = FF_WORKSPACES_MODULE_ENABLED
           const trx = await db.transaction()
 
           const moveProjectToWorkspace = moveProjectToWorkspaceFactory({
-            getProject: getProjectFactory(),
+            getProject: getProjectFactory({ db }),
             updateProject: updateProjectFactory({ db: trx }),
             upsertProjectRole: upsertProjectRoleFactory({ db: trx }),
             getProjectCollaborators: getProjectCollaboratorsFactory(),

packages/server/modules/workspaces/services/invites.ts

@@ -71,7 +71,7 @@ import {
   anyEmailCompliantWithWorkspaceDomains,
   userEmailsCompliantWithWorkspaceDomains
 } from '@/modules/workspaces/domain/logic'
-import { getStream } from '@/modules/core/repositories/streams'
+import { GetStream } from '@/modules/core/domain/streams/operations'
 
 const isWorkspaceResourceTarget = (
   target: InviteResourceTarget
@@ -124,7 +124,7 @@ type CollectAndValidateWorkspaceTargetsFactoryDeps =
     getWorkspace: GetWorkspace
     getWorkspaceDomains: GetWorkspaceDomains
     findVerifiedEmailsByUserId: FindVerifiedEmailsByUserId
-    getStream: typeof getStream
+    getStream: GetStream
   }
 
 export const collectAndValidateWorkspaceTargetsFactory =

packages/server/modules/workspaces/tests/helpers/creation.ts

@@ -1,5 +1,4 @@
 import { db } from '@/db/knex'
-import { getStream } from '@/modules/core/repositories/streams'
 import {
   findEmailsByUserIdFactory,
   findVerifiedEmailsByUserIdFactory
@@ -45,6 +44,7 @@ import {
   StreamRoles,
   WorkspaceRoles
 } from '@speckle/shared'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 
 export type BasicTestWorkspace = {
   /**
@@ -207,6 +207,7 @@ export const createWorkspaceInviteDirectly = async (
   args: CreateWorkspaceInviteMutationVariables,
   inviterId: string
 ) => {
+  const getStream = getStreamFactory({ db })
   const createAndSendInvite = createAndSendInviteFactory({
     findUserByTarget: findUserByTargetFactory(),
     insertInviteAndDeleteOld: insertInviteAndDeleteOldFactory({ db }),

packages/server/modules/workspaces/tests/integration/invites.graph.spec.ts

@@ -65,7 +65,6 @@ import {
 import type { Express } from 'express'
 import { AllScopes } from '@/modules/core/helpers/mainConstants'
 import { getWorkspaceFactory } from '@/modules/workspaces/repositories/workspaces'
-import { getStream } from '@/modules/core/repositories/streams'
 import {
   createUserEmailFactory,
   deleteUserEmailFactory,
@@ -79,6 +78,7 @@ import { addOrUpdateStreamCollaborator } from '@/modules/core/services/streams/s
 import { WorkspaceProtectedError } from '@/modules/workspaces/errors/workspace'
 import { ForbiddenError } from '@/modules/shared/errors'
 import cryptoRandomString from 'crypto-random-string'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 
 enum InviteByTarget {
   Email = 'email',
@@ -87,6 +87,8 @@ enum InviteByTarget {
 
 type TestGraphQLOperations = ReturnType<typeof buildGraphqlOperations>
 
+const getStream = getStreamFactory({ db })
+
 const buildGraphqlOperations = (deps: { apollo: TestApolloServer }) => {
   const { apollo } = deps
 

packages/server/test/speckle-helpers/inviteHelper.ts

@@ -11,7 +11,6 @@ import { createAndSendInviteFactory } from '@/modules/serverinvites/services/cre
 import { BasicTestUser } from '@/test/authHelper'
 import { BasicTestStream } from '@/test/speckle-helpers/streamHelper'
 import { collectAndValidateCoreTargetsFactory } from '@/modules/serverinvites/services/coreResourceCollection'
-import { getStream } from '@/modules/core/repositories/streams'
 import { buildCoreInviteEmailContentsFactory } from '@/modules/serverinvites/services/coreEmailContents'
 import { getEventBus } from '@/modules/shared/services/eventBus'
 import {
@@ -26,7 +25,9 @@ import {
   ServerInviteResourceTarget
 } from '@/modules/serverinvites/domain/types'
 import { EmailSendingServiceMock } from '@/test/mocks/global'
+import { getStreamFactory } from '@/modules/core/repositories/streams'
 
+const getStream = getStreamFactory({ db })
 const createAndSendInvite = createAndSendInviteFactory({
   findUserByTarget: findUserByTargetFactory(),
   insertInviteAndDeleteOld: insertInviteAndDeleteOldFactory({ db }),