Merge pull request #1148 from omnivore-app/feature/separate-queues-for-synthesis

feature/separate queues for synthesis

packages/api/src/routers/article_router.ts

@@ -73,13 +73,18 @@ export function articleRouter() {
   })
 
   router.get(
-    '/:id/:outputFormat/:voice?',
+    '/:id/:outputFormat/:priority/:voice?',
     cors<express.Request>(corsConfig),
     async (req, res) => {
       const articleId = req.params.id
       const outputFormat = req.params.outputFormat
       const voice = req.params.voice
-      if (!articleId || !['mp3', 'speech-marks'].includes(outputFormat)) {
+      const priority = req.params.priority
+      if (
+        !articleId ||
+        !['mp3', 'speech-marks'].includes(outputFormat) ||
+        !['low', 'high'].includes(priority)
+      ) {
         return res.status(400).send('Invalid data')
       }
       const token = req.cookies?.auth || req.headers?.authorization
@@ -153,14 +158,13 @@ export function articleRouter() {
         voice: voice || userPersonalization?.speechVoice || 'en-US-JennyNeural',
       })
       // enqueue a task to convert text to speech
-      const taskName = await enqueueTextToSpeech(
-        uid,
-        speech.id,
-        page.content,
-        'ssml',
-        speech.voice,
-        env.fileUpload.gcsUploadBucket
-      )
+      const taskName = await enqueueTextToSpeech({
+        userId: uid,
+        speechId: speech.id,
+        text: page.content,
+        voice: speech.voice,
+        priority: priority as 'low' | 'high',
+      })
       logger.info('Start Text to speech task', { taskName })
       res.status(202).send('Text to speech task started')
     }

packages/api/src/util.ts

@@ -94,6 +94,9 @@ interface BackendEnv {
     speechKey: string
     speechRegion: string
   }
+  gcp: {
+    location: string
+  }
 }
 
 /***
@@ -148,6 +151,7 @@ const nullableEnvVars = [
   'TEXT_TO_SPEECH_TASK_HANDLER_URL',
   'AZURE_SPEECH_KEY',
   'AZURE_SPEECH_REGION',
+  'GCP_LOCATION',
 ] // Allow some vars to be null/empty
 
 /* If not in GAE and Prod/QA/Demo env (f.e. on localhost/dev env), allow following env vars to be null */
@@ -273,6 +277,10 @@ export function getEnv(): BackendEnv {
     speechRegion: parse('AZURE_SPEECH_REGION'),
   }
 
+  const gcp = {
+    location: parse('GCP_LOCATION'),
+  }
+
   return {
     pg,
     client,
@@ -292,6 +300,7 @@ export function getEnv(): BackendEnv {
     sendgrid,
     readwise,
     azure,
+    gcp,
   }
 }
 

packages/api/src/utils/createTask.ts

@@ -45,11 +45,7 @@ const createHttpTaskWithToken = async ({
   ]
 > => {
   // Construct the fully qualified queue name.
-  if (priority === 'low') {
-    queue = `${queue}-low`
-    // use GCF url for low priority tasks
-    taskHandlerUrl = env.queue.contentFetchGCFUrl
-  }
+  priority === 'low' && (queue = `${queue}-low`)
 
   const parent = client.queuePath(project, location, queue)
   console.log(`Task creation options: `, {
@@ -211,13 +207,15 @@ export const deleteTask = async (
  * @param userId - Id of the user authorized
  * @param saveRequestId - Id of the article_saving_request table record
  * @param priority - Priority of the task
+ * @param queue - Queue name
  * @returns Name of the task created
  */
 export const enqueueParseRequest = async (
   url: string,
   userId: string,
   saveRequestId: string,
-  priority: 'low' | 'high' = 'high'
+  priority: 'low' | 'high' = 'high',
+  queue = env.queue.name
 ): Promise<string> => {
   const { GOOGLE_CLOUD_PROJECT } = process.env
   const payload = {
@@ -240,10 +238,18 @@ export const enqueueParseRequest = async (
     return ''
   }
 
+  // use GCF url for low priority tasks
+  const taskHandlerUrl =
+    priority === 'low'
+      ? env.queue.contentFetchGCFUrl
+      : env.queue.contentFetchUrl
+
   const createdTasks = await createHttpTaskWithToken({
     project: GOOGLE_CLOUD_PROJECT,
     payload,
     priority,
+    taskHandlerUrl,
+    queue,
   })
   if (!createdTasks || !createdTasks[0].name) {
     logger.error(`Unable to get the name of the task`, {
@@ -328,14 +334,27 @@ export const enqueueSyncWithIntegration = async (
   return createdTasks[0].name
 }
 
-export const enqueueTextToSpeech = async (
-  userId: string,
-  speechId: string,
-  text: string,
-  textType: 'text' | 'ssml',
-  voice: string,
-  bucket: string
-): Promise<string> => {
+export const enqueueTextToSpeech = async ({
+  userId,
+  text,
+  speechId,
+  voice,
+  priority,
+  textType = 'ssml',
+  bucket = env.fileUpload.gcsUploadBucket,
+  queue = 'omnivore-demo-text-to-speech-queue',
+  location = env.gcp.location,
+}: {
+  userId: string
+  speechId: string
+  text: string
+  voice: string
+  priority: 'low' | 'high'
+  bucket?: string
+  textType?: 'text' | 'ssml'
+  queue?: string
+  location?: string
+}): Promise<string> => {
   const { GOOGLE_CLOUD_PROJECT } = process.env
   const payload = {
     id: speechId,
@@ -364,6 +383,9 @@ export const enqueueTextToSpeech = async (
     project: GOOGLE_CLOUD_PROJECT,
     payload,
     taskHandlerUrl,
+    queue,
+    location,
+    priority,
   })
 
   if (!createdTasks || !createdTasks[0].name) {