Merge 598ddb7b63 into e379a9bb74

packages/text-to-speech/package.json

@@ -27,7 +27,7 @@
     "@types/html-to-text": "^8.1.1",
     "@types/mocha": "^10.0.6",
     "@types/natural": "^5.1.1",
-    "@types/node": "^14.11.2",
+    "@types/node": "^22.14.0",
     "@types/underscore": "^1.11.4",
     "@types/jsonwebtoken": "^9.0.7",
     "chai": "^4.3.6",
@@ -47,7 +47,10 @@
     "microsoft-cognitiveservices-speech-sdk": "1.30",
     "natural": "^6.2.0",
     "nodemon": "^2.0.15",
-    "underscore": "^1.13.4"
+    "underscore": "^1.13.4",
+    "@aws-sdk/client-s3": "^3.679.0",
+    "@aws-sdk/s3-request-presigner": "^3.679.0",
+    "@aws-sdk/lib-storage": "^3.679.0"
   },
   "volta": {
     "extends": "../../package.json"

packages/text-to-speech/src/index.ts

@@ -3,7 +3,6 @@
 /* eslint-disable @typescript-eslint/no-unsafe-argument */
 /* eslint-disable @typescript-eslint/no-unused-vars */
 
-import { File, Storage } from '@google-cloud/storage'
 import { RedisDataSource } from '@omnivore/utils'
 import * as Sentry from '@sentry/serverless'
 import axios from 'axios'
@@ -18,6 +17,8 @@ import {
   TextToSpeechInput,
   TextToSpeechOutput,
 } from './textToSpeech'
+import { S3StorageClient } from './storage/S3StorageClient'
+import { File } from './storage/StorageClient'
 
 interface UtteranceInput {
   text: string
@@ -51,7 +52,10 @@ Sentry.GCPFunction.init({
 })
 
 const MAX_CHARACTER_COUNT = 50000
-const storage = new Storage()
+const storage = new S3StorageClient(
+  process.env.LOCAL_MINIO_URL,
+  process.env.AWS_S3_ENDPOINT_URL
+)
 
 const textToSpeechHandlers = [new OpenAITextToSpeech(), new AzureTextToSpeech()]
 
@@ -73,11 +77,11 @@ const uploadToBucket = async (
   bucket: string,
   options?: { contentType?: string; public?: boolean }
 ): Promise<void> => {
-  await storage.bucket(bucket).file(filePath).save(data, options)
+  await storage.upload(bucket, filePath, data, options || {})
 }
 
 export const createGCSFile = (bucket: string, filename: string): File => {
-  return storage.bucket(bucket).file(filename)
+  return storage.createFile(bucket, filename)
 }
 
 const updateSpeech = async (
@@ -292,10 +296,10 @@ export const textToSpeechStreamingHandler = Sentry.GCPFunction.wrapHttpFunction(
       let audioData: Buffer | undefined
       let speechMarks: SpeechMark[] = []
       // check if audio file already exists
-      const [exists] = await audioFile.exists()
+      const exists = await audioFile.exists()
       if (exists) {
-        ;[audioData] = await audioFile.download()
-        const [speechMarksExists] = await speechMarksFile.exists()
+        audioData = await audioFile.download()
+        const speechMarksExists = await speechMarksFile.exists()
         if (speechMarksExists) {
           speechMarks = JSON.parse(
             (await speechMarksFile.download()).toString()
@@ -321,10 +325,10 @@ export const textToSpeechStreamingHandler = Sentry.GCPFunction.wrapHttpFunction(
         }
 
         // upload audio data to GCS
-        await audioFile.save(audioData)
+        await audioFile.save(audioData, {})
         // upload speech marks to GCS
         if (speechMarks.length > 0) {
-          await speechMarksFile.save(JSON.stringify(speechMarks))
+          await speechMarksFile.save(JSON.stringify(speechMarks), {})
         }
       }
 

packages/text-to-speech/src/openaiTextToSpeech.ts

@@ -91,7 +91,7 @@ export class OpenAITextToSpeech implements TextToSpeech {
     }
 
     const payload = {
-      model: 'tts-1',
+      model: 'gpt-4o-mini-tts',
       voice: voice,
       input: stripEmojis(input.text),
     }

packages/text-to-speech/src/storage/S3StorageClient.ts

@@ -0,0 +1,261 @@
+import {
+  SignedUrlParameters,
+  StorageClient,
+  File,
+  SaveOptions,
+  SaveData,
+} from './StorageClient'
+import { Upload } from '@aws-sdk/lib-storage'
+import {
+  GetObjectCommand,
+  GetObjectCommandOutput,
+  S3Client,
+  ListObjectsV2Command,
+  PutObjectCommand,
+  HeadObjectCommand,
+  S3ServiceException,
+} from '@aws-sdk/client-s3'
+import { getSignedUrl } from '@aws-sdk/s3-request-presigner'
+import { Readable } from 'stream'
+import * as stream from 'node:stream'
+
+// While this is listed as S3, for self hosting we will use MinIO, which is
+// S3 Compatible.
+export class S3StorageClient implements StorageClient {
+  BlankFile = class implements File {
+    bucket: string
+    key: string
+    s3Client: S3StorageClient
+    downloadedFile: File | undefined
+
+    constructor(s3StorageClass: S3StorageClient, bucket: string, file: string) {
+      this.bucket = bucket
+      this.key = file
+      this.s3Client = s3StorageClass
+    }
+
+    isPublic() {
+      return Promise.resolve(true)
+    }
+
+    publicUrl() {
+      return `${this.s3Client.localUrl ?? ''}/${this.bucket}/${this.key}`
+    }
+
+    async download(): Promise<Buffer> {
+      this.downloadedFile = await this.s3Client.downloadFile(
+        this.bucket,
+        this.key
+      )
+      return this.downloadedFile.download()
+    }
+
+    async exists() {
+      try {
+        await this.s3Client.s3Client.send(
+          new HeadObjectCommand({
+            Bucket: this.bucket,
+            Key: this.key,
+          })
+        )
+
+        return true
+      } catch (e) {
+        if (
+          e instanceof S3ServiceException &&
+          e.$metadata.httpStatusCode == 404
+        ) {
+          return false
+        }
+
+        throw e
+      }
+    }
+
+    save(saveData: SaveData, saveOptions: SaveOptions): Promise<void> {
+      return this.s3Client.upload(this.bucket, this.key, saveData, saveOptions)
+    }
+
+    createWriteStream(saveOptions: SaveOptions) {
+      return this.s3Client.createS3UploadStream(
+        this.bucket,
+        this.key,
+        saveOptions
+      )
+    }
+
+    getMetadataMd5() {
+      return this.downloadedFile?.getMetadataMd5() || Promise.resolve('')
+    }
+  }
+
+  private s3Client: S3Client
+  private signingS3Client: S3Client
+
+  private urlOverride: string | undefined
+  private localUrl: string | undefined
+
+  constructor(localUrl: string | undefined, urlOverride: string | undefined) {
+    this.localUrl = localUrl
+    this.urlOverride = urlOverride
+    this.s3Client = new S3Client({
+      forcePathStyle: true,
+      endpoint: urlOverride,
+    })
+
+    this.signingS3Client = new S3Client({
+      forcePathStyle: true,
+      endpoint: localUrl,
+    })
+  }
+
+  private createS3UploadStream = (
+    bucket: string,
+    key: string,
+    saveOptions: SaveOptions
+  ) => {
+    const passThroughStream = new stream.PassThrough()
+
+    const upload = new Upload({
+      client: this.s3Client,
+      params: {
+        Bucket: bucket,
+        Key: key,
+        Body: passThroughStream,
+        ContentType: saveOptions.contentType,
+      },
+    })
+
+    void upload.done().then((res) => {
+      console.log(`Successfully Uploaded File ${res.Key ?? ''}`)
+    })
+
+    return passThroughStream
+  }
+
+  private convertFileToGeneric(
+    s3File: GetObjectCommandOutput,
+    bucket: string,
+    key: string
+  ): File {
+    return {
+      exists: () => {
+        return Promise.resolve(s3File.$metadata.httpStatusCode == 200)
+      },
+      save: async () => Promise.resolve(),
+      isPublic: async () => Promise.resolve(true),
+      download: async () => this.getFileFromReadable(s3File.Body as Readable),
+      getMetadataMd5: () => Promise.resolve(s3File.ETag),
+      createWriteStream: (saveOptions: SaveOptions) =>
+        this.createS3UploadStream(bucket, key, saveOptions),
+      publicUrl: () => `${this.localUrl ?? ''}/${bucket}/${key}`,
+      bucket,
+      key,
+    }
+  }
+
+  private getFileFromReadable(stream: Readable): Promise<Buffer> {
+    return new Promise<Buffer>((resolve, reject) => {
+      const chunks: Buffer[] = []
+      stream.on('data', (chunk) => chunks.push(chunk))
+      stream.once('end', () => resolve(Buffer.concat(chunks)))
+      stream.once('error', reject)
+    })
+  }
+
+  async downloadFile(bucket: string, filePath: string): Promise<File> {
+    const s3File = await this.s3Client.send(
+      new GetObjectCommand({
+        Bucket: bucket,
+        Key: filePath, // path to the file you want to download,
+      })
+    )
+
+    return this.convertFileToGeneric(s3File, bucket, filePath)
+  }
+
+  createFile(bucket: string, filePath: string): File {
+    return new this.BlankFile(this, bucket, filePath) as unknown as File
+  }
+
+  async getFilesFromPrefix(bucket: string, prefix: string): Promise<File[]> {
+    const s3PrefixedFiles = await this.s3Client.send(
+      new ListObjectsV2Command({
+        Bucket: bucket,
+        Prefix: prefix, // path to the file you want to download,
+      })
+    )
+
+    const prefixKeys = s3PrefixedFiles.CommonPrefixes || []
+
+    return prefixKeys
+      .map(({ Prefix }) => Prefix)
+      .map((key: string | undefined) => {
+        return {
+          key: key || '',
+          exists: () => Promise.resolve(true),
+          isPublic: async () => Promise.resolve(true),
+          download: async () => {
+            const s3File = await this.s3Client.send(
+              new GetObjectCommand({
+                Bucket: bucket,
+                Key: key, // path to the file you want to download,
+              })
+            )
+
+            return this.getFileFromReadable(s3File.Body as Readable)
+          },
+          save: () => Promise.resolve(),
+          createWriteStream: (saveOptions: SaveOptions) =>
+            new stream.PassThrough(),
+          getMetadataMd5: () => Promise.resolve(key),
+          bucket: bucket,
+          publicUrl: () => `${this.localUrl ?? ''}/${bucket}/${key ?? ''}`,
+        }
+      })
+  }
+
+  async signedUrl(
+    bucket: string,
+    filePath: string,
+    options: SignedUrlParameters
+  ): Promise<string> {
+    const command =
+      options.action == 'read'
+        ? new GetObjectCommand({
+            Bucket: bucket,
+            Key: filePath, // path to the file you want to download,
+          })
+        : new PutObjectCommand({
+            Bucket: bucket,
+            Key: filePath, // path to the file you want to download,
+          })
+
+    // eslint-disable-next-line @typescript-eslint/no-unsafe-call
+    const url = await getSignedUrl(this.signingS3Client, command, {
+      expiresIn: 900,
+    })
+
+    return url
+  }
+
+  async upload(
+    bucket: string,
+    filePath: string,
+    data: SaveData,
+    options: {
+      contentType?: string
+      public?: boolean
+      timeout?: number
+    }
+  ): Promise<void> {
+    await this.s3Client.send(
+      new PutObjectCommand({
+        Bucket: bucket,
+        Key: filePath,
+        Body: data.toString(),
+        ContentType: options.contentType,
+      })
+    )
+  }
+}

packages/text-to-speech/src/storage/StorageClient.ts

@@ -0,0 +1,49 @@
+import { PipelineSource, Writable } from 'stream'
+
+export type SignedUrlParameters = {
+  action: 'read' | 'write' | 'delete' | 'resumable'
+  expires: number
+}
+
+export type SaveData = string | Buffer | PipelineSource<string | Buffer>
+export type SaveOptions = {
+  contentType?: string
+  gzip?: string | boolean
+  resumable?: boolean
+  timeout?: number
+  validation?: string | boolean
+  private?: boolean | undefined
+}
+
+export type File = {
+  isPublic: () => Promise<boolean>
+  publicUrl: () => string
+  download: () => Promise<Buffer>
+  exists: () => Promise<boolean>
+  save: (saveData: SaveData, saveOptions: SaveOptions) => Promise<void>
+  createWriteStream: (saveOptions: SaveOptions) => Writable
+  getMetadataMd5: () => Promise<string | undefined>
+  bucket: string
+  key: string
+}
+
+export interface StorageClient {
+  downloadFile(bucket: string, filePath: string): Promise<File>
+
+  createFile(bucket: string, filePath: string): File
+
+  getFilesFromPrefix(bucket: string, filePrefix: string): Promise<File[]>
+
+  upload(
+    bucket: string,
+    filePath: string,
+    data: Buffer,
+    options: { contentType?: string; public?: boolean; timeout?: number }
+  ): Promise<void>
+
+  signedUrl(
+    bucket: string,
+    filePath: string,
+    options: SignedUrlParameters
+  ): Promise<string>
+}

self-hosting/docker-compose/.env.example

@@ -36,6 +36,7 @@ LOCAL_MINIO_URL=http://localhost:1010
 
 # Redis
 REDIS_URL=redis://redis:6379/0
+REDIS_TTS_URL=redis://redis:6379/1
 
 #MAIL
 WATCHER_API_KEY=mail-api-key

self-hosting/docker-compose/docker-compose.yml

@@ -44,6 +44,25 @@ services:
       migrate:
         condition: service_completed_successfully
 
+  tts:
+    image: "ghcr.io/omnivore-app/sh-backend:latest"
+    container_name: "omnivore-tts"
+    command: ["yarn", "workspace", "@omnivore/text-to-speech-handler", "start_streaming"]
+    ports:
+      - "5000:8080"
+    healthcheck:
+      test: ["CMD-SHELL", "nc -z 0.0.0.0 8080 || exit 1"]
+      interval: 15s
+      timeout: 90s
+      retries: 6
+    env_file:
+      - .env
+    depends_on:
+      redis:
+        condition: service_healthy
+      migrate:
+        condition: service_completed_successfully
+
   queue-processor:
     image: "ghcr.io/omnivore-app/sh-queue-processor:latest"
     container_name: "omnivore-queue-processor"

self-hosting/docker-compose/self-build/docker-compose.yml

@@ -50,6 +50,27 @@ services:
       migrate:
         condition: service_completed_successfully
 
+  tts:
+    build:
+      context: ../../../
+      dockerfile: ./packages/api/Dockerfile
+    container_name: "omnivore-tts"
+    command: ["yarn", "workspace", "@omnivore/text-to-speech-handler", "start_streaming"]
+    ports:
+      - "5000:8080"
+    healthcheck:
+      test: ["CMD-SHELL", "nc -z 0.0.0.0 8080 || exit 1"]
+      interval: 15s
+      timeout: 90s
+      retries: 6
+    env_file:
+      - .env
+    depends_on:
+      redis:
+        condition: service_healthy
+      migrate:
+        condition: service_completed_successfully
+
   queue-processor:
     build:
       context: ../../../

self-hosting/nginx/nginx.conf

@@ -15,6 +15,11 @@ http {
     	server 127.0.0.1:4000;
      }
 
+     upstream omnivore_tts {
+         ip_hash;
+         server 127.0.0.1:5000;
+     }
+
      upstream omnivore_imageproxy {
     	ip_hash;
     	server 127.0.0.1:7070;
@@ -57,6 +62,12 @@ http {
             proxy_pass http://omnivore_backend;
         }
 
+        # TTS
+        location /tts {
+            rewrite ^/tts(.*)$ /$1 break;
+            proxy_pass http://omnivore_tts;
+        }
+
         # Minio
         location /bucket {
             rewrite ^/bucket/(.*)$ /$1 break;