Add a pubsub subscription listening on the new article to automatically synthesize article

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

@@ -1,97 +0,0 @@
-import express from 'express'
-import cors from 'cors'
-import { corsConfig } from '../../utils/corsConfig'
-import { getRepository } from '../../entity/utils'
-import { getPageById } from '../../elastic/pages'
-import { synthesizeTextToSpeech } from '../../utils/textToSpeech'
-import { Speech, SpeechState } from '../../entity/speech'
-import { buildLogger } from '../../utils/logger'
-import { getClaimsByToken } from '../../utils/auth'
-import { setSpeechFailure } from '../../services/speech'
-
-const logger = buildLogger('app.dispatch')
-
-export function speechServiceRouter() {
-  const router = express.Router()
-
-  router.options('/', cors<express.Request>({ ...corsConfig, maxAge: 600 }))
-  // eslint-disable-next-line @typescript-eslint/no-misused-promises
-  router.post('/', async (req, res) => {
-    logger.info('Speech svc request', {
-      // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
-      body: req.body,
-    })
-    const token = req.query.token as string
-    try {
-      if (!(await getClaimsByToken(token))) {
-        logger.info('Unauthorized request', { token })
-        return res.status(200).send('UNAUTHORIZED')
-      }
-    } catch (error) {
-      logger.error('Unauthorized request', { token, error })
-      return res.status(200).send('UNAUTHORIZED')
-    }
-
-    const { userId, speechId } = req.body as {
-      userId: string
-      speechId: string
-    }
-    if (!userId || !speechId) {
-      return res.status(200).send('Invalid data')
-    }
-
-    logger.info(`Create article speech`, {
-      body: {
-        userId,
-        speechId,
-      },
-      labels: {
-        source: 'CreateArticleSpeech',
-      },
-    })
-    const speech = await getRepository(Speech).findOneBy({
-      id: speechId,
-      user: { id: userId },
-    })
-    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: speech.id,
-        text: page.content,
-        languageCode: page.language,
-        voice: speech.voice,
-        textType: 'ssml',
-      })
-      logger.info('Created speech', {
-        audioFileName: speechOutput.audioFileName,
-        speechMarksFileName: speechOutput.speechMarksFileName,
-        duration: Date.now() - startTime,
-      })
-
-      // set state to completed
-      await getRepository(Speech).update(speech.id, {
-        audioFileName: speechOutput.audioFileName,
-        speechMarksFileName: speechOutput.speechMarksFileName,
-        state: SpeechState.COMPLETED,
-      })
-
-      res.status(200).send('OK')
-    } catch (error) {
-      logger.error(`Error creating article speech`, { error })
-      await setSpeechFailure(speech.id)
-      res.status(500).send('Error creating article speech')
-    }
-  })
-
-  return router
-}

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

@@ -0,0 +1,140 @@
+/* eslint-disable @typescript-eslint/no-misused-promises */
+/* eslint-disable @typescript-eslint/no-unsafe-assignment */
+/* eslint-disable @typescript-eslint/explicit-module-boundary-types */
+import express from 'express'
+import cors from 'cors'
+import { corsConfig } from '../../utils/corsConfig'
+import { getRepository } from '../../entity/utils'
+import { getPageById } from '../../elastic/pages'
+import { Speech, SpeechState } from '../../entity/speech'
+import { buildLogger } from '../../utils/logger'
+import { getClaimsByToken } from '../../utils/auth'
+import {
+  setSpeechFailure,
+  shouldSynthesize,
+  synthesize,
+} from '../../services/speech'
+import { readPushSubscription } from '../../datalayer/pubsub'
+
+const logger = buildLogger('app.dispatch')
+
+export function speechServiceRouter() {
+  const router = express.Router()
+
+  router.options(
+    '/auto-synthesis',
+    cors<express.Request>({ ...corsConfig, maxAge: 600 })
+  )
+  // eslint-disable-next-line @typescript-eslint/no-misused-promises
+  router.post('/auto-synthesize', async (req, res) => {
+    logger.info('auto-synthesize')
+    const { message: msgStr, expired } = readPushSubscription(req)
+
+    if (!msgStr) {
+      return res.status(400).send('Bad Request')
+    }
+
+    if (expired) {
+      logger.info('discarding expired message')
+      return res.status(200).send('Expired')
+    }
+
+    try {
+      const data: { userId: string; type: string; id: string } =
+        JSON.parse(msgStr)
+      const { userId, type, id } = data
+      if (!userId || !type) {
+        logger.info('No userId or type found in message')
+        return res.status(400).send('Bad Request')
+      }
+
+      if (type.toUpperCase() !== 'PAGE') {
+        logger.info('Not a page')
+        return res.status(200).send('Not a page')
+      }
+
+      // checks if this page needs to be synthesized automatically
+      const page = await getPageById(id)
+      if (!page) {
+        logger.info('No page found', { id })
+        return res.status(200).send('No page found')
+      }
+
+      if (await shouldSynthesize(userId, page)) {
+        logger.info('page needs to be synthesized')
+        // initialize state
+        const speech = await getRepository(Speech).save({
+          user: { id: userId },
+          elasticPageId: id,
+          state: SpeechState.INITIALIZED,
+        })
+        await synthesize(page, speech)
+        logger.info('page synthesized')
+      }
+
+      res.status(200).send('Page should not synthesize')
+    } catch (err) {
+      logger.error('Auto synthesize failed', err)
+      res.status(500).send(err)
+    }
+  })
+
+  router.options('/', cors<express.Request>({ ...corsConfig, maxAge: 600 }))
+  // eslint-disable-next-line @typescript-eslint/no-misused-promises
+  router.post('/', async (req, res) => {
+    logger.info('Synthesize svc request', {
+      // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
+      body: req.body,
+    })
+    const token = req.query.token as string
+    try {
+      if (!(await getClaimsByToken(token))) {
+        logger.info('Unauthorized request', { token })
+        return res.status(200).send('UNAUTHORIZED')
+      }
+    } catch (error) {
+      logger.error('Unauthorized request', { token, error })
+      return res.status(200).send('UNAUTHORIZED')
+    }
+
+    const { userId, speechId } = req.body as {
+      userId: string
+      speechId: string
+    }
+    if (!userId || !speechId) {
+      return res.status(200).send('Invalid data')
+    }
+
+    logger.info(`Create article speech`, {
+      body: {
+        userId,
+        speechId,
+      },
+      labels: {
+        source: 'CreateArticleSpeech',
+      },
+    })
+    const speech = await getRepository(Speech).findOneBy({
+      id: speechId,
+      user: { id: userId },
+    })
+    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 {
+      await synthesize(page, speech)
+    } catch (error) {
+      logger.error(`Error synthesizing article`, { error })
+      res.status(500).send('Error synthesizing article')
+    }
+  })
+
+  return router
+}

packages/api/src/server.ts

@@ -45,7 +45,7 @@ import { uploadServiceRouter } from './routers/svc/upload'
 import rateLimit from 'express-rate-limit'
 import { webhooksServiceRouter } from './routers/svc/webhooks'
 import { integrationsServiceRouter } from './routers/svc/integrations'
-import { speechServiceRouter } from './routers/svc/speech'
+import { speechServiceRouter } from './routers/svc/text_to_speech'
 
 const PORT = process.env.PORT || 4000
 

packages/api/src/services/speech.ts

@@ -3,6 +3,7 @@ import { Speech, SpeechState } from '../entity/speech'
 import { searchPages } from '../elastic/pages'
 import { Page } from '../elastic/types'
 import { SortBy, SortOrder } from '../utils/search'
+import { synthesizeTextToSpeech } from '../utils/textToSpeech'
 
 export const setSpeechFailure = async (id: string) => {
   // update state
@@ -12,11 +13,11 @@ export const setSpeechFailure = async (id: string) => {
 }
 
 /*
- * We should not transcribe the page when:
+ * We should not synthesize the page when:
  ** 1. User has no recent listens the last 30 days
  ** 2. User has a recent listen but the page was saved after the listen
  */
-export const shouldTranscribe = async (
+export const shouldSynthesize = async (
   userId: string,
   page: Page
 ): Promise<boolean> => {
@@ -44,3 +45,33 @@ export const shouldTranscribe = async (
     page.savedAt < recentListenedPage[0].listenedAt
   )
 }
+
+export const synthesize = async (page: Page, speech: Speech): Promise<void> => {
+  try {
+    console.log('synthesizing', speech.id)
+    const startTime = Date.now()
+    const speechOutput = await synthesizeTextToSpeech({
+      id: speech.id,
+      text: page.content,
+      languageCode: page.language,
+      voice: speech.voice,
+      textType: 'ssml',
+    })
+    console.log('Synthesized article', {
+      audioFileName: speechOutput.audioFileName,
+      speechMarksFileName: speechOutput.speechMarksFileName,
+      duration: Date.now() - startTime,
+    })
+
+    // set state to completed
+    await getRepository(Speech).update(speech.id, {
+      audioFileName: speechOutput.audioFileName,
+      speechMarksFileName: speechOutput.speechMarksFileName,
+      state: SpeechState.COMPLETED,
+    })
+  } catch (error) {
+    console.log('Error synthesize article', error)
+    await setSpeechFailure(speech.id)
+    throw error
+  }
+}