diff --git a/packages/text-to-speech/package.json b/packages/text-to-speech/package.json index 7cb2a083e..18375096c 100644 --- a/packages/text-to-speech/package.json +++ b/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" diff --git a/packages/text-to-speech/src/index.ts b/packages/text-to-speech/src/index.ts index 781952b10..b0bffe390 100644 --- a/packages/text-to-speech/src/index.ts +++ b/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 => { - 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), {}) } } diff --git a/packages/text-to-speech/src/openaiTextToSpeech.ts b/packages/text-to-speech/src/openaiTextToSpeech.ts index db9a42764..4b5a337f1 100644 --- a/packages/text-to-speech/src/openaiTextToSpeech.ts +++ b/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), } diff --git a/packages/text-to-speech/src/storage/S3StorageClient.ts b/packages/text-to-speech/src/storage/S3StorageClient.ts new file mode 100644 index 000000000..ef673123f --- /dev/null +++ b/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 { + 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 { + 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 { + return new Promise((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 { + 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 { + 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 { + 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 { + await this.s3Client.send( + new PutObjectCommand({ + Bucket: bucket, + Key: filePath, + Body: data.toString(), + ContentType: options.contentType, + }) + ) + } +} diff --git a/packages/text-to-speech/src/storage/StorageClient.ts b/packages/text-to-speech/src/storage/StorageClient.ts new file mode 100644 index 000000000..12257e0b8 --- /dev/null +++ b/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 +export type SaveOptions = { + contentType?: string + gzip?: string | boolean + resumable?: boolean + timeout?: number + validation?: string | boolean + private?: boolean | undefined +} + +export type File = { + isPublic: () => Promise + publicUrl: () => string + download: () => Promise + exists: () => Promise + save: (saveData: SaveData, saveOptions: SaveOptions) => Promise + createWriteStream: (saveOptions: SaveOptions) => Writable + getMetadataMd5: () => Promise + bucket: string + key: string +} + +export interface StorageClient { + downloadFile(bucket: string, filePath: string): Promise + + createFile(bucket: string, filePath: string): File + + getFilesFromPrefix(bucket: string, filePrefix: string): Promise + + upload( + bucket: string, + filePath: string, + data: Buffer, + options: { contentType?: string; public?: boolean; timeout?: number } + ): Promise + + signedUrl( + bucket: string, + filePath: string, + options: SignedUrlParameters + ): Promise +} diff --git a/self-hosting/docker-compose/.env.example b/self-hosting/docker-compose/.env.example index f414eb7d0..b9bc3b821 100644 --- a/self-hosting/docker-compose/.env.example +++ b/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 diff --git a/self-hosting/docker-compose/docker-compose.yml b/self-hosting/docker-compose/docker-compose.yml index 5ffb9eab7..37719b516 100644 --- a/self-hosting/docker-compose/docker-compose.yml +++ b/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" diff --git a/self-hosting/docker-compose/self-build/docker-compose.yml b/self-hosting/docker-compose/self-build/docker-compose.yml index 81637aa71..f8f4043e7 100644 --- a/self-hosting/docker-compose/self-build/docker-compose.yml +++ b/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: ../../../ diff --git a/self-hosting/nginx/nginx.conf b/self-hosting/nginx/nginx.conf index 0af2f52de..33effd54a 100644 --- a/self-hosting/nginx/nginx.conf +++ b/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;