Merge pull request #1114 from omnivore-app/feature/queue-tts-in-the-api

feature/queue tts in the api

packages/api/src/resolvers/article/index.ts

@@ -94,7 +94,6 @@ import {
   updatePage,
 } from '../../elastic/pages'
 import { searchHighlights } from '../../elastic/highlights'
-import { enqueueTextToSpeech } from '../../utils/createTask'
 
 export type PartialArticle = Omit<
   Article,
@@ -372,11 +371,6 @@ export const createArticleResolver = authorized<
         }
         articleToSave.id = newPageId
       }
-
-      // enqueue a task to convert text to speech
-      const taskName = await enqueueTextToSpeech(uid, articleToSave.id)
-      log.info('Text to speech task name', { taskName })
-
       log.info(
         'page created in elastic',
         articleToSave.id,

packages/api/src/routers/article_router.ts

@@ -18,9 +18,9 @@ import { Claims } from '../resolvers/types'
 import { getRepository } from '../entity/utils'
 import { Speech, SpeechState } from '../entity/speech'
 import { getPageById } from '../elastic/pages'
-import { synthesizeTextToSpeech } from '../utils/textToSpeech'
-import { UserPersonalization } from '../entity/user_personalization'
 import { generateDownloadSignedUrl } from '../utils/uploads'
+import { enqueueTextToSpeech } from '../utils/createTask'
+import { UserPersonalization } from '../entity/user_personalization'
 
 const logger = buildLogger('app.dispatch')
 
@@ -72,12 +72,13 @@ export function articleRouter() {
   })
 
   router.get(
-    '/:id/:outputFormat',
+    '/:id/:outputFormat/:voice?',
     cors<express.Request>(corsConfig),
     async (req, res) => {
-      const id = req.params.id
+      const articleId = req.params.id
       const outputFormat = req.params.outputFormat
-      if (!id || !['mp3', 'speech-marks'].includes(outputFormat)) {
+      const voice = req.params.voice
+      if (!articleId || !['mp3', 'speech-marks'].includes(outputFormat)) {
         return res.status(400).send('Invalid data')
       }
       const token = req.cookies?.auth || req.headers?.authorization
@@ -94,82 +95,59 @@ export function articleRouter() {
         },
       })
 
-      const existingSpeech = await getRepository(Speech).findOneBy({
-        elasticPageId: id,
+      const existingSpeech = await getRepository(Speech).findOne({
+        where: {
+          elasticPageId: articleId,
+          voice,
+        },
+        order: {
+          createdAt: 'DESC',
+        },
+        relations: ['user'],
       })
-      if (existingSpeech?.state === SpeechState.COMPLETED) {
-        logger.info('Found existing completed speech', {
-          audioUrl: existingSpeech.audioFileName,
-          speechMarksUrl: existingSpeech.speechMarksFileName,
-        })
-        return res.redirect(await redirectUrl(existingSpeech, outputFormat))
-      }
-      if (existingSpeech?.state === SpeechState.INITIALIZED) {
-        logger.info('Found existing in progress speech')
-        // retry later
-        return res.status(429).send('Speech is in progress')
+      if (existingSpeech) {
+        if (existingSpeech.user.id !== uid) {
+          logger.info('User is not allowed to access speech of the article', {
+            userId: uid,
+            articleId,
+          })
+          return res.status(401).send({ errorCode: 'UNAUTHORIZED' })
+        }
+        if (existingSpeech.state === SpeechState.COMPLETED) {
+          logger.info('Found existing completed speech', {
+            audioUrl: existingSpeech.audioFileName,
+            speechMarksUrl: existingSpeech.speechMarksFileName,
+          })
+          return res.redirect(await redirectUrl(existingSpeech, outputFormat))
+        }
+        if (existingSpeech.state === SpeechState.INITIALIZED) {
+          logger.info('Found existing in progress speech')
+          // retry later
+          return res.status(202).send('Speech is in progress')
+        }
       }
 
-      logger.debug('Text to speech request', { articleId: id })
+      logger.info('Create Text to speech task', { articleId })
+      const page = await getPageById(articleId)
+      if (!page) {
+        return res.status(404).send('Page not found')
+      }
       const userPersonalization = await getRepository(
         UserPersonalization
       ).findOneBy({
         user: { id: uid },
       })
-      if (!userPersonalization) {
-        return res.status(404).send('User Personalization not found')
-      }
-
-      const page = await getPageById(id)
-      if (!page) {
-        return res.status(404).send('Page not found')
-      }
-
-      // const text = parseHTML(page.content).document.documentElement.innerText
-      // if (!text) {
-      //   return res.status(404).send('Page has no text')
-      // }
-
       // initialize state
       const speech = await getRepository(Speech).save({
         user: { id: uid },
-        elasticPageId: id,
+        elasticPageId: articleId,
         state: SpeechState.INITIALIZED,
-        voice: userPersonalization.speechVoice,
+        voice: voice || userPersonalization?.speechVoice,
       })
-      try {
-        const startTime = Date.now()
-        const speechOutput = await synthesizeTextToSpeech({
-          id,
-          text: page.content,
-          languageCode: page.language,
-          voice: userPersonalization.speechVoice,
-          textType: 'ssml',
-        })
-        logger.info('Created speech', {
-          audioFileName: speechOutput.audioFileName,
-          speechMarksFileName: speechOutput.speechMarksFileName,
-          duration: Date.now() - startTime,
-        })
-
-        // update state
-        await getRepository(Speech).update(speech.id, {
-          state: SpeechState.COMPLETED,
-          audioFileName: speechOutput.audioFileName,
-          speechMarksFileName: speechOutput.speechMarksFileName,
-        })
-        speech.audioFileName = speechOutput.audioFileName
-        speech.speechMarksFileName = speechOutput.speechMarksFileName
-
-        res.redirect(await redirectUrl(speech, outputFormat))
-      } catch (error) {
-        logger.error('Text to speech error', { error })
-        // update state
-        await getRepository(Speech).update(speech.id, {
-          state: SpeechState.FAILED,
-        })
-        res.status(500).send('Text to speech error')
-      }
+      // enqueue a task to convert text to speech
+      const taskName = await enqueueTextToSpeech(uid, speech.id)
+      logger.info('Start Text to speech task', { taskName })
+      res.status(202).send('Text to speech task started')
     }
   )
 

packages/api/src/routers/svc/speech.ts

@@ -5,9 +5,9 @@ import { getRepository } from '../../entity/utils'
 import { getPageById } from '../../elastic/pages'
 import { synthesizeTextToSpeech } from '../../utils/textToSpeech'
 import { Speech, SpeechState } from '../../entity/speech'
-import { UserPersonalization } from '../../entity/user_personalization'
 import { buildLogger } from '../../utils/logger'
 import { getClaimsByToken } from '../../utils/auth'
+import { setSpeechFailure } from '../../services/speech'
 
 const logger = buildLogger('app.dispatch')
 
@@ -32,57 +32,44 @@ export function speechServiceRouter() {
       return res.status(200).send('UNAUTHORIZED')
     }
 
-    const { userId, pageId } = req.body as {
+    const { userId, speechId } = req.body as {
       userId: string
-      pageId: string
+      speechId: string
     }
-
-    if (!userId || !pageId) {
+    if (!userId || !speechId) {
       return res.status(200).send('Invalid data')
     }
 
-    const userPersonalization = await getRepository(
-      UserPersonalization
-    ).findOneBy({
-      user: { id: userId },
-    })
-    if (!userPersonalization) {
-      return res.status(200).send('User Personalization not found')
-    }
-
-    const page = await getPageById(pageId)
-    if (!page) {
-      return res.status(200).send('Page not found')
-    }
-    // const text = parseHTML(page.content).document.documentElement.innerText
-    // if (!text) {
-    //   return res.status(200).send('Page has no text')
-    // }
     logger.info(`Create article speech`, {
       body: {
         userId,
-        pageId,
+        speechId,
       },
       labels: {
         source: 'CreateArticleSpeech',
       },
     })
-
-    // initialize state
-    const speech = await getRepository(Speech).save({
+    const speech = await getRepository(Speech).findOneBy({
+      id: speechId,
       user: { id: userId },
-      elasticPageId: pageId,
-      state: SpeechState.INITIALIZED,
-      voice: userPersonalization.speechVoice,
     })
+    if (!speech) {
+      return res.status(200).send('Speech not found')
+    }
+
+    const page = await getPageById(speech.elasticPageId)
+    if (!page) {
+      await setSpeechFailure(speech.id)
+      return res.status(200).send('Page not found')
+    }
 
     try {
       const startTime = Date.now()
       const speechOutput = await synthesizeTextToSpeech({
-        id: pageId,
+        id: speech.id,
         text: page.content,
         languageCode: page.language,
-        voice: userPersonalization.speechVoice,
+        voice: speech.voice,
         textType: 'ssml',
       })
       logger.info('Created speech', {
@@ -91,7 +78,7 @@ export function speechServiceRouter() {
         duration: Date.now() - startTime,
       })
 
-      // update state
+      // set state to completed
       await getRepository(Speech).update(speech.id, {
         audioFileName: speechOutput.audioFileName,
         speechMarksFileName: speechOutput.speechMarksFileName,
@@ -101,10 +88,7 @@ export function speechServiceRouter() {
       res.status(200).send('OK')
     } catch (error) {
       logger.error(`Error creating article speech`, { error })
-      // update state
-      await getRepository(Speech).update(speech.id, {
-        state: SpeechState.FAILED,
-      })
+      await setSpeechFailure(speech.id)
       res.status(500).send('Error creating article speech')
     }
   })

packages/api/src/services/speech.ts

@@ -0,0 +1,9 @@
+import { getRepository } from '../entity/utils'
+import { Speech, SpeechState } from '../entity/speech'
+
+export const setSpeechFailure = async (id: string) => {
+  // update state
+  await getRepository(Speech).update(id, {
+    state: SpeechState.FAILED,
+  })
+}

packages/api/src/utils/createTask.ts

@@ -330,12 +330,12 @@ export const enqueueSyncWithIntegration = async (
 
 export const enqueueTextToSpeech = async (
   userId: string,
-  pageId: string
+  speechId: string
 ): Promise<string> => {
   const { GOOGLE_CLOUD_PROJECT } = process.env
   const payload = {
     userId,
-    pageId,
+    speechId,
   }
   // eslint-disable-next-line @typescript-eslint/ban-ts-comment
   // @ts-ignore

packages/api/src/utils/textToSpeech.ts

@@ -128,7 +128,6 @@ export const synthesizeTextToSpeech = async (
           resolve(result)
         },
         (error) => {
-          synthesizer.close()
           reject(error)
         }
       )
@@ -145,7 +144,6 @@ export const synthesizeTextToSpeech = async (
           resolve(result)
         },
         (error) => {
-          synthesizer.close()
           reject(error)
         }
       )
@@ -183,6 +181,10 @@ export const synthesizeTextToSpeech = async (
         )
         logger.debug(`synthesizing ${ssml}`)
         const result = await speakSsmlAsyncPromise(ssml)
+        if (result.reason === ResultReason.Canceled) {
+          synthesizer.close()
+          throw new Error(result.errorDetails)
+        }
         timeOffset = timeOffset + result.audioDuration
         // characterOffset = characterOffset + htmlElement.innerText.length
       }