diff --git a/packages/server/modules/previews/index.js b/packages/server/modules/previews/index.js index 242e7eb67..53dadc524 100644 --- a/packages/server/modules/previews/index.js +++ b/packages/server/modules/previews/index.js @@ -10,7 +10,7 @@ const { const { makeOgImage } = require('./ogImage') const { moduleLogger } = require('@/logging/logging') const { - listenForPreviewGenerationUpdates + listenForPreviewGenerationUpdatesFactory } = require('@/modules/previews/services/resultListener') const httpErrorImage = (httpErrorCode) => @@ -29,6 +29,8 @@ const { createObjectPreviewFactory, getPreviewImageFactory } = require('@/modules/previews/repository/previews') +const { publish } = require('@/modules/shared/utils/subscriptions') +const { getObjectCommitsWithStreamIds } = require('@/modules/core/repositories/commits') const noPreviewImage = require.resolve('#/assets/previews/images/no_preview.png') @@ -159,6 +161,10 @@ exports.init = (app, isInitial) => { }) if (isInitial) { + const listenForPreviewGenerationUpdates = listenForPreviewGenerationUpdatesFactory({ + getObjectCommitsWithStreamIds, + publish + }) listenForPreviewGenerationUpdates() } } diff --git a/packages/server/modules/previews/services/resultListener.ts b/packages/server/modules/previews/services/resultListener.ts index 20c513fb3..3b3b712d4 100644 --- a/packages/server/modules/previews/services/resultListener.ts +++ b/packages/server/modules/previews/services/resultListener.ts @@ -1,39 +1,50 @@ import { getObjectCommitsWithStreamIds } from '@/modules/core/repositories/commits' -import { ProjectSubscriptions, publish } from '@/modules/shared/utils/subscriptions' +import { + ProjectSubscriptions, + PublishSubscription +} from '@/modules/shared/utils/subscriptions' import { listenFor, MessageType } from '@/modules/core/utils/dbNotificationListener' const payloadRegexp = /^([\w\d]+):([\w\d]+):([\w\d]+)$/i -async function messageProcessor(msg: MessageType) { - if (msg.channel !== 'preview_generation_update') return - const [, status, streamId, objectId] = payloadRegexp.exec(msg.payload) || [ - null, - null, - null, - null - ] +type MessageProcessorDeps = { + getObjectCommitsWithStreamIds: typeof getObjectCommitsWithStreamIds + publish: PublishSubscription +} - if (status !== 'finished' || !objectId || !streamId) return +const messageProcessorFactory = + (deps: MessageProcessorDeps) => async (msg: MessageType) => { + if (msg.channel !== 'preview_generation_update') return + const [, status, streamId, objectId] = payloadRegexp.exec(msg.payload) || [ + null, + null, + null, + null + ] - // Get all commits with that objectId - const commits = await getObjectCommitsWithStreamIds([objectId], { - streamIds: [streamId] - }) - if (!commits.length) return + if (status !== 'finished' || !objectId || !streamId) return - await Promise.all( - commits.map((c) => - publish(ProjectSubscriptions.ProjectVersionsPreviewGenerated, { - projectVersionsPreviewGenerated: { - versionId: c.id, - projectId: c.streamId, - objectId - } - }) + // Get all commits with that objectId + const commits = await deps.getObjectCommitsWithStreamIds([objectId], { + streamIds: [streamId] + }) + if (!commits.length) return + + await Promise.all( + commits.map((c) => + deps.publish(ProjectSubscriptions.ProjectVersionsPreviewGenerated, { + projectVersionsPreviewGenerated: { + versionId: c.id, + projectId: c.streamId, + objectId + } + }) + ) ) - ) -} + } -export function listenForPreviewGenerationUpdates() { - listenFor('preview_generation_update', messageProcessor) -} +export const listenForPreviewGenerationUpdatesFactory = + (deps: MessageProcessorDeps) => () => { + const messageProcessor = messageProcessorFactory(deps) + listenFor('preview_generation_update', messageProcessor) + }