LibreChat/api/server/services/Files/Audio/TTSService.js
Danny Avila 9f6d8c6e93
🧵 feat: ALS Context Middleware, Tenant Threading, and Config Cache Invalidation (#12407)
* feat: add tenant context middleware for ALS-based isolation

Introduces tenantContextMiddleware that propagates req.user.tenantId
into AsyncLocalStorage, activating the Mongoose applyTenantIsolation
plugin for all downstream DB queries within a request.

- Strict mode (TENANT_ISOLATION_STRICT=true) returns 403 if no tenantId
- Non-strict mode passes through for backward compatibility
- No-op for unauthenticated requests
- Includes 6 unit tests covering all paths

* feat: register tenant middleware and wrap startup/auth in runAsSystem()

- Register tenantContextMiddleware in Express app after capability middleware
- Wrap server startup initialization in runAsSystem() for strict mode compat
- Wrap auth strategy getAppConfig() calls in runAsSystem() since they run
  before user context is established (LDAP, SAML, OpenID, social login, AuthService)

* feat: thread tenantId through all getAppConfig callers

Pass tenantId from req.user to getAppConfig() across all callers that
have request context, ensuring correct per-tenant cache key resolution.

Also fixes getBaseConfig admin endpoint to scope to requesting admin's
tenant instead of returning the unscoped base config.

Files updated:
- Controllers: UserController, PluginController
- Middleware: checkDomainAllowed, balance
- Routes: config
- Services: loadConfigModels, loadDefaultModels, getEndpointsConfig, MCP
- Audio services: TTSService, STTService, getVoices, getCustomConfigSpeech
- Admin: getBaseConfig endpoint

* feat: add config cache invalidation on admin mutations

- Add clearOverrideCache(tenantId?) to flush per-principal override caches
  by enumerating Keyv store keys matching _OVERRIDE_: prefix
- Add invalidateConfigCaches() helper that clears base config, override
  caches, tool caches, and endpoint config cache in one call
- Wire invalidation into all 5 admin config mutation handlers
  (upsert, patch, delete field, delete overrides, toggle active)
- Add strict mode warning when __default__ tenant fallback is used
- Add 3 new tests for clearOverrideCache (all/scoped/base-preserving)

* chore: update getUserPrincipals comment to reflect ALS-based tenant filtering

The TODO(#12091) about missing tenantId filtering is resolved by the
tenant context middleware + applyTenantIsolation Mongoose plugin.
Group queries are now automatically scoped by tenantId via ALS.

* fix: replace runAsSystem with baseOnly for pre-tenant code paths

App configs are tenant-owned — runAsSystem() would bypass tenant
isolation and return cross-tenant DB overrides. Instead, add
baseOnly option to getAppConfig() that returns YAML-derived config
only, with zero DB queries.

All startup code, auth strategies, and MCP initialization now use
getAppConfig({ baseOnly: true }) to get the YAML config without
touching the Config collection.

* fix: address PR review findings — middleware ordering, types, cache safety

- Chain tenantContextMiddleware inside requireJwtAuth after passport auth
  instead of global app.use() where req.user is always undefined (Finding 1)
- Remove global tenantContextMiddleware registration from index.js
- Update BalanceMiddlewareOptions to include tenantId, remove redundant cast (Finding 4)
- Add warning log when clearOverrideCache cannot enumerate keys on Redis (Finding 3)
- Use startsWith instead of includes for cache key filtering (Finding 12)
- Use generator loop instead of Array.from for key enumeration (Finding 3)
- Selective barrel export — exclude _resetTenantMiddlewareStrictCache (Finding 5)
- Move isMainThread check to module level, remove per-request check (Finding 9)
- Move mid-file require to top of app.js (Finding 8)
- Parallelize invalidateConfigCaches with Promise.all (Finding 10)
- Remove clearOverrideCache from public app.js exports (internal only)
- Strengthen getUserPrincipals comment re: ALS dependency (Finding 2)

* fix: restore runAsSystem for startup DB ops, consolidate require, clarify baseOnly

- Restore runAsSystem() around performStartupChecks, updateInterfacePermissions,
  initializeMCPs, and initializeOAuthReconnectManager — these make Mongoose
  queries that need system context in strict tenant mode (NEW-3)
- Consolidate duplicate require('@librechat/api') in requireJwtAuth.js (NEW-1)
- Document that baseOnly ignores role/userId/tenantId in JSDoc (NEW-2)

* test: add requireJwtAuth tenant chaining + invalidateConfigCaches tests

- requireJwtAuth: 5 tests verifying ALS tenant context is set after
  passport auth, isolated between concurrent requests, and not set
  when user has no tenantId (Finding 6)
- invalidateConfigCaches: 4 tests verifying all four caches are cleared,
  tenantId is threaded through, partial failure is handled gracefully,
  and operations run in parallel via Promise.all (Finding 11)

* fix: address Copilot review — passport errors, namespaced cache keys, /base scoping

- Forward passport errors in requireJwtAuth before entering tenant
  middleware — prevents silent auth failures from reaching handlers (P1)
- Account for Keyv namespace prefix in clearOverrideCache — stored keys
  are namespaced as "APP_CONFIG:_OVERRIDE_:..." not "_OVERRIDE_:...",
  so override caches were never actually matched/cleared (P2)
- Remove role from getBaseConfig — /base should return tenant-scoped
  base config, not role-merged config that drifts per admin role (P2)
- Return tenantStorage.run() for cleaner async semantics
- Update mock cache in service.spec.ts to simulate Keyv namespacing

* fix: address second review — cache safety, code quality, test reliability

- Decouple cache invalidation from mutation response: fire-and-forget
  with logging so DB mutation success is not masked by cache failures
- Extract clearEndpointConfigCache helper from inline IIFE
- Move isMainThread check to lazy once-per-process guard (no import
  side effect)
- Memoize process.env read in overrideCacheKey to avoid per-request
  env lookups and log flooding in strict mode
- Remove flaky timer-based parallelism assertion, use structural check
- Merge orphaned double JSDoc block on getUserPrincipals
- Fix stale [getAppConfig] log prefix → [ensureBaseConfig]
- Fix import order in tenant.spec.ts (package types before local values)
- Replace "Finding 1" reference with self-contained description
- Use real tenantStorage primitives in requireJwtAuth spec mock

* fix: move JSDoc to correct function after clearEndpointConfigCache extraction

* refactor: remove Redis SCAN from clearOverrideCache, rely on TTL expiry

Redis SCAN causes 60s+ stalls under concurrent load (see #12410).
APP_CONFIG defaults to FORCED_IN_MEMORY_CACHE_NAMESPACES, so the
in-memory store.keys() path handles the standard case. When APP_CONFIG
is Redis-backed, overrides expire naturally via overrideCacheTtl (60s
default) — an acceptable window for admin config mutations.

* fix: remove return from tenantStorage.run to satisfy void middleware signature

* fix: address second review — cache safety, code quality, test reliability

- Switch invalidateConfigCaches from Promise.all to Promise.allSettled
  so partial failures are logged individually instead of producing one
  undifferentiated error (Finding 3)
- Gate overrideCacheKey strict-mode warning behind a once-per-process
  flag to prevent log flooding under load (Finding 4)
- Add test for passport error forwarding in requireJwtAuth — the
  if (err) { return next(err) } branch now has coverage (Finding 5)
- Add test for real partial failure in invalidateConfigCaches where
  clearAppConfigCache rejects (not just the swallowed endpoint error)

* chore: reorder imports in index.js and app.js for consistency

- Moved logger and runAsSystem imports to maintain a consistent import order across files.
- Improved code readability by ensuring related imports are grouped together.
2026-03-26 17:35:00 -04:00

496 lines
15 KiB
JavaScript

const axios = require('axios');
const { logger } = require('@librechat/data-schemas');
const { HttpsProxyAgent } = require('https-proxy-agent');
const { genAzureEndpoint, logAxiosError } = require('@librechat/api');
const { extractEnvVariable, TTSProviders } = require('librechat-data-provider');
const { getRandomVoiceId, createChunkProcessor, splitTextIntoChunks } = require('./streamAudio');
const { getAppConfig } = require('~/server/services/Config');
/**
* Service class for handling Text-to-Speech (TTS) operations.
* @class
*/
class TTSService {
/**
* Creates an instance of TTSService.
*/
constructor() {
this.providerStrategies = {
[TTSProviders.OPENAI]: this.openAIProvider.bind(this),
[TTSProviders.AZURE_OPENAI]: this.azureOpenAIProvider.bind(this),
[TTSProviders.ELEVENLABS]: this.elevenLabsProvider.bind(this),
[TTSProviders.LOCALAI]: this.localAIProvider.bind(this),
};
}
/**
* Creates a singleton instance of TTSService.
* @static
* @async
* @returns {Promise<TTSService>} The TTSService instance.
* @throws {Error} If the custom config is not found.
*/
static async getInstance() {
return new TTSService();
}
/**
* Retrieves the configured TTS provider.
* @param {AppConfig | null | undefined} [appConfig] - The app configuration object.
* @returns {string} The name of the configured provider.
* @throws {Error} If no provider is set or multiple providers are set.
*/
getProvider(appConfig) {
const ttsSchema = appConfig?.speech?.tts;
if (!ttsSchema) {
throw new Error(
'No TTS schema is set. Did you configure TTS in the custom config (librechat.yaml)?',
);
}
const providers = Object.entries(ttsSchema).filter(
([, value]) => Object.keys(value).length > 0,
);
if (providers.length !== 1) {
throw new Error(
providers.length > 1
? 'Multiple providers are set. Please set only one provider.'
: 'No provider is set. Please set a provider.',
);
}
return providers[0][0];
}
/**
* Selects a voice for TTS based on provider schema and request.
* @async
* @param {Object} providerSchema - The schema for the selected provider.
* @param {string} requestVoice - The requested voice.
* @returns {Promise<string>} The selected voice.
*/
async getVoice(providerSchema, requestVoice) {
const voices = providerSchema.voices.filter((voice) => voice && voice.toUpperCase() !== 'ALL');
let voice = requestVoice;
if (!voice || !voices.includes(voice) || (voice.toUpperCase() === 'ALL' && voices.length > 1)) {
voice = getRandomVoiceId(voices);
}
return voice;
}
/**
* Recursively removes undefined properties from an object.
* @param {Object} obj - The object to clean.
*/
removeUndefined(obj) {
Object.keys(obj).forEach((key) => {
if (obj[key] && typeof obj[key] === 'object') {
this.removeUndefined(obj[key]);
if (Object.keys(obj[key]).length === 0) {
delete obj[key];
}
} else if (obj[key] === undefined) {
delete obj[key];
}
});
}
/**
* Prepares the request for OpenAI TTS provider.
* @param {Object} ttsSchema - The TTS schema for OpenAI.
* @param {string} input - The input text.
* @param {string} voice - The selected voice.
* @returns {Array} An array containing the URL, data, and headers for the request.
* @throws {Error} If the selected voice is not available.
*/
openAIProvider(ttsSchema, input, voice) {
const url = ttsSchema?.url || 'https://api.openai.com/v1/audio/speech';
if (
ttsSchema?.voices &&
ttsSchema.voices.length > 0 &&
!ttsSchema.voices.includes(voice) &&
!ttsSchema.voices.includes('ALL')
) {
throw new Error(`Voice ${voice} is not available.`);
}
const data = {
input,
model: ttsSchema?.model,
voice: ttsSchema?.voices && ttsSchema.voices.length > 0 ? voice : undefined,
backend: ttsSchema?.backend,
};
const headers = {
'Content-Type': 'application/json',
Authorization: `Bearer ${extractEnvVariable(ttsSchema?.apiKey)}`,
};
return [url, data, headers];
}
/**
* Prepares the request for Azure OpenAI TTS provider.
* @param {Object} ttsSchema - The TTS schema for Azure OpenAI.
* @param {string} input - The input text.
* @param {string} voice - The selected voice.
* @returns {Array} An array containing the URL, data, and headers for the request.
* @throws {Error} If the selected voice is not available.
*/
azureOpenAIProvider(ttsSchema, input, voice) {
const url = `${genAzureEndpoint({
azureOpenAIApiInstanceName: extractEnvVariable(ttsSchema?.instanceName),
azureOpenAIApiDeploymentName: extractEnvVariable(ttsSchema?.deploymentName),
})}/audio/speech?api-version=${extractEnvVariable(ttsSchema?.apiVersion)}`;
if (
ttsSchema?.voices &&
ttsSchema.voices.length > 0 &&
!ttsSchema.voices.includes(voice) &&
!ttsSchema.voices.includes('ALL')
) {
throw new Error(`Voice ${voice} is not available.`);
}
const data = {
model: extractEnvVariable(ttsSchema?.model),
input,
voice: ttsSchema?.voices && ttsSchema.voices.length > 0 ? voice : undefined,
};
const headers = {
'Content-Type': 'application/json',
'api-key': ttsSchema.apiKey ? extractEnvVariable(ttsSchema.apiKey) : '',
};
return [url, data, headers];
}
/**
* Prepares the request for ElevenLabs TTS provider.
* @param {Object} ttsSchema - The TTS schema for ElevenLabs.
* @param {string} input - The input text.
* @param {string} voice - The selected voice.
* @param {boolean} stream - Whether to use streaming.
* @returns {Array} An array containing the URL, data, and headers for the request.
* @throws {Error} If the selected voice is not available.
*/
elevenLabsProvider(ttsSchema, input, voice, stream) {
let url =
ttsSchema?.url ||
`https://api.elevenlabs.io/v1/text-to-speech/${voice}${stream ? '/stream' : ''}`;
if (!ttsSchema?.voices.includes(voice) && !ttsSchema?.voices.includes('ALL')) {
throw new Error(`Voice ${voice} is not available.`);
}
const data = {
model_id: ttsSchema?.model,
text: input,
voice_settings: {
similarity_boost: ttsSchema?.voice_settings?.similarity_boost,
stability: ttsSchema?.voice_settings?.stability,
style: ttsSchema?.voice_settings?.style,
use_speaker_boost: ttsSchema?.voice_settings?.use_speaker_boost,
},
pronunciation_dictionary_locators: ttsSchema?.pronunciation_dictionary_locators,
};
const headers = {
'Content-Type': 'application/json',
'xi-api-key': extractEnvVariable(ttsSchema?.apiKey),
Accept: 'audio/mpeg',
};
return [url, data, headers];
}
/**
* Prepares the request for LocalAI TTS provider.
* @param {Object} ttsSchema - The TTS schema for LocalAI.
* @param {string} input - The input text.
* @param {string} voice - The selected voice.
* @returns {Array} An array containing the URL, data, and headers for the request.
* @throws {Error} If the selected voice is not available.
*/
localAIProvider(ttsSchema, input, voice) {
const url = ttsSchema?.url;
if (
ttsSchema?.voices &&
ttsSchema.voices.length > 0 &&
!ttsSchema.voices.includes(voice) &&
!ttsSchema.voices.includes('ALL')
) {
throw new Error(`Voice ${voice} is not available.`);
}
const data = {
input,
model: ttsSchema?.voices && ttsSchema.voices.length > 0 ? voice : undefined,
backend: ttsSchema?.backend,
};
const headers = {
'Content-Type': 'application/json',
Authorization: `Bearer ${extractEnvVariable(ttsSchema?.apiKey)}`,
};
if (extractEnvVariable(ttsSchema.apiKey) === '') {
delete headers.Authorization;
}
return [url, data, headers];
}
/**
* Sends a TTS request to the specified provider.
* @async
* @param {string} provider - The TTS provider to use.
* @param {Object} ttsSchema - The TTS schema for the provider.
* @param {Object} options - The options for the TTS request.
* @param {string} options.input - The input text.
* @param {string} options.voice - The voice to use.
* @param {boolean} [options.stream=true] - Whether to use streaming.
* @returns {Promise<Object>} The axios response object.
* @throws {Error} If the provider is invalid or the request fails.
*/
async ttsRequest(provider, ttsSchema, { input, voice, stream = true }) {
const strategy = this.providerStrategies[provider];
if (!strategy) {
throw new Error('Invalid provider');
}
const [url, data, headers] = strategy.call(this, ttsSchema, input, voice, stream);
[data, headers].forEach(this.removeUndefined.bind(this));
const options = { headers, responseType: stream ? 'stream' : 'arraybuffer' };
if (process.env.PROXY) {
options.httpsAgent = new HttpsProxyAgent(process.env.PROXY);
}
try {
return await axios.post(url, data, options);
} catch (error) {
logAxiosError({ message: `TTS request failed for provider ${provider}:`, error });
throw error;
}
}
/**
* Processes a text-to-speech request.
* @async
* @param {ServerRequest} req - The request object.
* @param {ServerResponse} res - The response object.
* @returns {Promise<void>}
*/
async processTextToSpeech(req, res) {
const { input, voice: requestVoice } = req.body;
if (!input) {
return res.status(400).send('Missing text in request body');
}
const appConfig =
req.config ??
(await getAppConfig({
role: req.user?.role,
tenantId: req.user?.tenantId,
}));
try {
res.setHeader('Content-Type', 'audio/mpeg');
const provider = this.getProvider(appConfig);
const ttsSchema = appConfig?.speech?.tts?.[provider];
const voice = await this.getVoice(ttsSchema, requestVoice);
if (input.length < 4096) {
const response = await this.ttsRequest(provider, ttsSchema, { input, voice });
response.data.pipe(res);
return;
}
const textChunks = splitTextIntoChunks(input, 1000);
for (const chunk of textChunks) {
try {
const response = await this.ttsRequest(provider, ttsSchema, {
voice,
input: chunk.text,
stream: true,
});
logger.debug(`[textToSpeech] user: ${req?.user?.id} | writing audio stream`);
await new Promise((resolve) => {
response.data.pipe(res, { end: chunk.isFinished });
response.data.on('end', resolve);
});
if (chunk.isFinished) {
break;
}
} catch (innerError) {
logAxiosError({
message: `[TTS] Error processing manual update for chunk: ${chunk?.text?.substring(0, 50)}...`,
error: innerError,
});
if (!res.headersSent) {
return res.status(500).end();
}
return;
}
}
if (!res.headersSent) {
res.end();
}
} catch (error) {
logAxiosError({ message: '[TTS] Error creating the audio stream:', error });
if (!res.headersSent) {
return res.status(500).send('An error occurred');
}
}
}
/**
* Streams audio data from the TTS provider.
* @async
* @param {ServerRequest} req - The request object.
* @param {ServerResponse} res - The response object.
* @returns {Promise<void>}
*/
async streamAudio(req, res) {
res.setHeader('Content-Type', 'audio/mpeg');
const appConfig =
req.config ??
(await getAppConfig({
role: req.user?.role,
tenantId: req.user?.tenantId,
}));
const provider = this.getProvider(appConfig);
const ttsSchema = appConfig?.speech?.tts?.[provider];
const voice = await this.getVoice(ttsSchema, req.body.voice);
let shouldContinue = true;
req.on('close', () => {
logger.warn('[streamAudio] Audio Stream Request closed by client');
shouldContinue = false;
});
const processChunks = createChunkProcessor(req.user.id, req.body.messageId);
try {
while (shouldContinue) {
const updates = await processChunks();
if (typeof updates === 'string') {
logger.error(`Error processing audio stream updates: ${updates}`);
return res.status(500).end();
}
if (updates.length === 0) {
await new Promise((resolve) => setTimeout(resolve, 1250));
continue;
}
for (const update of updates) {
try {
const response = await this.ttsRequest(provider, ttsSchema, {
voice,
input: update.text,
stream: true,
});
if (!shouldContinue) {
break;
}
logger.debug(`[streamAudio] user: ${req?.user?.id} | writing audio stream`);
await new Promise((resolve) => {
response.data.pipe(res, { end: update.isFinished });
response.data.on('end', resolve);
});
if (update.isFinished) {
shouldContinue = false;
break;
}
} catch (innerError) {
logAxiosError({
message: `[TTS] Error processing audio stream update: ${update?.text?.substring(0, 50)}...`,
error: innerError,
});
if (!res.headersSent) {
return res.status(500).end();
}
return;
}
}
if (!shouldContinue) {
break;
}
}
if (!res.headersSent) {
res.end();
}
} catch (error) {
logAxiosError({ message: '[TTS] Failed to fetch audio:', error });
if (!res.headersSent) {
res.status(500).end();
}
}
}
}
/**
* Factory function to create a TTSService instance.
* @async
* @returns {Promise<TTSService>} A promise that resolves to a TTSService instance.
*/
async function createTTSService() {
return TTSService.getInstance();
}
/**
* Wrapper function for text-to-speech processing.
* @async
* @param {ServerRequest} req - The request object.
* @param {ServerResponse} res - The response object.
* @returns {Promise<void>}
*/
async function textToSpeech(req, res) {
const ttsService = await createTTSService();
await ttsService.processTextToSpeech(req, res);
}
/**
* Wrapper function for audio streaming.
* @async
* @param {Object} req - The request object.
* @param {Object} res - The response object.
* @returns {Promise<void>}
*/
async function streamAudio(req, res) {
const ttsService = await createTTSService();
await ttsService.streamAudio(req, res);
}
/**
* Wrapper function to get the configured TTS provider.
* @async
* @param {AppConfig | null | undefined} appConfig - The app configuration object.
* @returns {Promise<string>} A promise that resolves to the name of the configured provider.
*/
async function getProvider(appConfig) {
const ttsService = await createTTSService();
return ttsService.getProvider(appConfig);
}
module.exports = {
textToSpeech,
streamAudio,
getProvider,
};