diff --git a/services/api/src/email/queue.rs b/services/api/src/email/queue.rs index d6a209a8..877876c6 100644 --- a/services/api/src/email/queue.rs +++ b/services/api/src/email/queue.rs @@ -564,11 +564,15 @@ impl EmailQueue { .await?; // Derive a stable idempotency key for this job so retries never - // produce duplicate sends within the configured TTL window. + // produce duplicate sends within the configured TTL window. The key + // is bucketed off the job's original creation time (not "now") so + // that a retry delayed past an hour boundary by exponential backoff + // still computes the same key as earlier attempts of the same job. let idem = idempotency_key( &job.recipient_email, &job.template_name, &service.idempotency_secret, + job.created_at, ); // Send email (deduplication handled inside send_email_idempotent) diff --git a/services/api/src/email/service.rs b/services/api/src/email/service.rs index 94d7e318..a7c44fb7 100644 --- a/services/api/src/email/service.rs +++ b/services/api/src/email/service.rs @@ -1,9 +1,10 @@ use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; use hmac::{Hmac, Mac}; use redis::AsyncCommands as _; use serde_json::Value; use sha2::Sha256; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::Duration; use validator::ValidateEmail; use crate::cache::RedisCache; @@ -34,16 +35,24 @@ impl Default for IdempotencyConfig { /// - A server-side secret prevents pre-computation by external attackers. /// - The hour bucket bounds the validity window to ~1 hour per key rotation. /// +/// `bucket_time` is the timestamp the hour bucket is derived from. Callers +/// that retry the same logical job (e.g. the email queue worker) must pass +/// the job's original creation time rather than the current time — otherwise +/// a retry whose exponential backoff crosses an hour boundary would compute a +/// different key and defeat deduplication for a job that already succeeded. +/// /// The key format is `email:idem:`. -pub fn idempotency_key(recipient: &str, template_name: &str, secret: &str) -> String { +pub fn idempotency_key( + recipient: &str, + template_name: &str, + secret: &str, + bucket_time: DateTime, +) -> String { type HmacSha256 = Hmac; - // Hour bucket: seconds-since-epoch / 3600, so keys rotate each hour. - let hour_bucket = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_secs() - / 3600; + // Hour bucket: seconds-since-epoch / 3600, so keys rotate each hour + // relative to `bucket_time` rather than wall-clock "now". + let hour_bucket = bucket_time.timestamp() / 3600; let mut mac = HmacSha256::new_from_slice(secret.as_bytes()) .expect("HMAC accepts any key length"); @@ -183,7 +192,7 @@ impl EmailService { template_name: &str, template_data: &Value, ) -> Result { - let idem = idempotency_key(recipient, template_name, &self.idempotency_secret); + let idem = idempotency_key(recipient, template_name, &self.idempotency_secret, Utc::now()); self.send_email_idempotent(recipient, template_name, template_data, Some(&idem)) .await } @@ -441,36 +450,40 @@ mod tests { #[test] fn same_inputs_produce_same_key() { - let k1 = idempotency_key("user@example.com", "welcome_email", "secret"); - let k2 = idempotency_key("user@example.com", "welcome_email", "secret"); + let now = Utc::now(); + let k1 = idempotency_key("user@example.com", "welcome_email", "secret", now); + let k2 = idempotency_key("user@example.com", "welcome_email", "secret", now); assert_eq!(k1, k2); } #[test] fn different_recipient_produces_different_key() { - let k1 = idempotency_key("alice@example.com", "welcome_email", "secret"); - let k2 = idempotency_key("bob@example.com", "welcome_email", "secret"); + let now = Utc::now(); + let k1 = idempotency_key("alice@example.com", "welcome_email", "secret", now); + let k2 = idempotency_key("bob@example.com", "welcome_email", "secret", now); assert_ne!(k1, k2); } #[test] fn different_template_produces_different_key() { - let k1 = idempotency_key("user@example.com", "welcome_email", "secret"); - let k2 = idempotency_key("user@example.com", "newsletter_confirmation", "secret"); + let now = Utc::now(); + let k1 = idempotency_key("user@example.com", "welcome_email", "secret", now); + let k2 = idempotency_key("user@example.com", "newsletter_confirmation", "secret", now); assert_ne!(k1, k2); } /// #932: key changes when the secret changes. #[test] fn different_secret_produces_different_key() { - let k1 = idempotency_key("user@example.com", "welcome_email", "secret-a"); - let k2 = idempotency_key("user@example.com", "welcome_email", "secret-b"); + let now = Utc::now(); + let k1 = idempotency_key("user@example.com", "welcome_email", "secret-a", now); + let k2 = idempotency_key("user@example.com", "welcome_email", "secret-b", now); assert_ne!(k1, k2, "key must change when the HMAC secret changes"); } #[test] fn key_has_expected_prefix() { - let key = idempotency_key("user@example.com", "t", "secret"); + let key = idempotency_key("user@example.com", "t", "secret", Utc::now()); assert!(key.starts_with("email:idem:"), "key should start with email:idem: prefix"); } @@ -489,14 +502,52 @@ mod tests { /// Retry produces the same key (within the same hour bucket). #[test] fn retry_produces_same_idempotency_key() { - let key_attempt_1 = idempotency_key("user@example.com", "newsletter_confirmation", "secret"); - let key_attempt_2 = idempotency_key("user@example.com", "newsletter_confirmation", "secret"); + let created_at = Utc::now(); + let key_attempt_1 = idempotency_key("user@example.com", "newsletter_confirmation", "secret", created_at); + let key_attempt_2 = idempotency_key("user@example.com", "newsletter_confirmation", "secret", created_at); assert_eq!( key_attempt_1, key_attempt_2, "retry must produce the same idempotency key" ); } + /// #1129: a retry delayed past the 1-hour bucket boundary (e.g. by the + /// 6th exponential-backoff attempt, ~64 minutes out) must still compute + /// the same idempotency key as the original attempt, since both are + /// derived from the job's original creation time rather than "now". + #[test] + fn retry_key_is_stable_across_hour_boundary_when_derived_from_job_creation_time() { + let created_at = Utc::now(); + let key_at_creation = idempotency_key("user@example.com", "newsletter_confirmation", "secret", created_at); + + // Simulate a retry scheduled more than an hour after the original + // attempt (e.g. the ~64-minute delay of the 6th backoff attempt). + let delayed_retry_time = created_at + chrono::Duration::minutes(90); + let key_at_delayed_retry = idempotency_key( + "user@example.com", + "newsletter_confirmation", + "secret", + created_at, // bucket is still derived from the original creation time + ); + assert_eq!( + key_at_creation, key_at_delayed_retry, + "idempotency key must remain stable across the job's full retry lifetime" + ); + + // Sanity check: had the key been derived from wall-clock "now" at the + // delayed retry time instead of the job's creation time, it would differ. + let key_from_wall_clock_now = idempotency_key( + "user@example.com", + "newsletter_confirmation", + "secret", + delayed_retry_time, + ); + assert_ne!( + key_at_creation, key_from_wall_clock_now, + "test precondition: 90 minutes must cross an hour bucket boundary" + ); + } + /// Two 429s followed by a 202: the service should succeed on the third attempt. #[tokio::test] async fn retry_succeeds_after_two_429s() { diff --git a/services/api/src/email/webhook.rs b/services/api/src/email/webhook.rs index 17847568..16684aab 100644 --- a/services/api/src/email/webhook.rs +++ b/services/api/src/email/webhook.rs @@ -1,5 +1,6 @@ use anyhow::{anyhow, Result}; use axum::{extract::State, http::{HeaderMap, StatusCode}, response::IntoResponse, Json}; +use futures::stream::{self, StreamExt}; use serde::{Deserialize, Serialize}; use std::{sync::Arc, time::Duration}; @@ -17,6 +18,11 @@ const MAX_TEXT_FIELD_LEN: usize = 1024; /// Maximum length for structured identifier fields (email, event type, message_id). const MAX_ID_FIELD_LEN: usize = 254; +/// Maximum number of webhook events processed concurrently within a single batch. +/// Bounds DB/Redis connection usage while avoiding fully sequential processing +/// of large SendGrid batches (which can contain hundreds of events per POST). +const WEBHOOK_BATCH_CONCURRENCY: usize = 16; + /// Raw deserialization target — accepts any valid JSON so we can validate /// before committing anything to the database. #[derive(Debug, Clone, Deserialize)] @@ -185,19 +191,33 @@ impl WebhookHandler { } /// Process a list of already-sanitized SendGrid webhook events. + /// + /// Events are processed concurrently (bounded by `WEBHOOK_BATCH_CONCURRENCY`) + /// rather than one at a time, since a single POST can contain hundreds of + /// events and fully sequential processing risks holding the request open + /// past the SendGrid webhook timeout budget. pub async fn handle_sendgrid_webhook( &self, events: Vec, ) -> Result { + let results: Vec<(String, Result<()>)> = stream::iter(events) + .map(|event| async move { + let event_desc = event.event.clone(); + (event_desc, self.process_event(event).await) + }) + .buffer_unordered(WEBHOOK_BATCH_CONCURRENCY) + .collect() + .await; + let mut processed = 0; let mut errors = Vec::new(); - for event in events { - match self.process_event(event.clone()).await { + for (event_desc, result) in results { + match result { Ok(_) => processed += 1, Err(e) => { tracing::error!("Error processing webhook event: {}", e); - errors.push(format!("Event {}: {}", event.event, e)); + errors.push(format!("Event {}: {}", event_desc, e)); } } } diff --git a/services/tts/src/HealthCheck.ts b/services/tts/src/HealthCheck.ts index e393fb6e..ff3a2e66 100644 --- a/services/tts/src/HealthCheck.ts +++ b/services/tts/src/HealthCheck.ts @@ -264,7 +264,7 @@ export class HealthChecker { private checkJobStore(): HealthCheckStatus { try { // Verify job store is functional - const jobs = this.service.listJobs(); + const jobs = this.service.listAllJobsUnscoped(); return { status: "ok", @@ -284,8 +284,8 @@ export class HealthChecker { */ private checkJobQueueDepth(): HealthCheckStatus { try { - const pending = this.service.listJobs("pending").length; - const processing = this.service.listJobs("processing").length; + const pending = this.service.listAllJobsUnscoped("pending").length; + const processing = this.service.listAllJobsUnscoped("processing").length; const depth = pending + processing; if (depth >= MAX_QUEUE_DEPTH) { diff --git a/services/tts/src/TTSService.ts b/services/tts/src/TTSService.ts index 162b1792..34f9c9ca 100644 --- a/services/tts/src/TTSService.ts +++ b/services/tts/src/TTSService.ts @@ -46,8 +46,18 @@ export interface TTSJob { createdAt: Date; updatedAt: Date; bypassCache?: boolean; + /** + * Identity of the credential that created this job (the API key itself, + * or the JWT `sub` claim). Used to enforce per-tenant access on + * getJob/listJobs so one credential cannot read another's jobs. + * `ANONYMOUS_OWNER` when no auth is configured (single-tenant deployment). + */ + owner: string; } +/** Owner tag used for jobs created when no auth is configured. */ +export const ANONYMOUS_OWNER = "anonymous"; + // --------------------------------------------------------------------------- // Rate limiting (issue #531) // --------------------------------------------------------------------------- @@ -415,12 +425,18 @@ export class AuthError extends Error { } } -export function authenticate(credential: string | undefined, auth: AuthConfig): void { +/** + * Verify `credential` against `auth` and return a stable tenant identity for it: + * - API key auth: the key itself is the tenant boundary. + * - JWT auth: the `sub` claim if present, otherwise the raw credential. + * Throws `AuthError` if the credential is missing or invalid. + */ +export function authenticate(credential: string | undefined, auth: AuthConfig): string { if (!credential) throw new AuthError("Missing credential"); if (auth.type === "apikey") { if (!auth.keys.includes(credential)) throw new AuthError("Invalid API key"); - return; + return credential; } const parts = credential.split("."); @@ -438,6 +454,8 @@ export function authenticate(credential: string | undefined, auth: AuthConfig): if (payload.exp !== undefined && payload.exp < Math.floor(Date.now() / 1000)) { throw new AuthError("JWT expired"); } + + return typeof payload.sub === "string" && payload.sub ? payload.sub : credential; } // --------------------------------------------------------------------------- @@ -717,8 +735,18 @@ export class TTSService { /** * Build one circuit breaker per configured TTS provider. The breaker - * wraps a thin async action that accepts (text, voice) and delegates to - * the raw provider implementation. + * wraps the *entire retried request* (raw provider call + backoff + * retries), not a single raw call. This is deliberate: firing the + * breaker once per retry attempt would let one failed user request + * inflate the breaker's failure count by up to `maxRetries + 1`, tripping + * it far earlier than the configured threshold intends, and — once + * open — a retry loop wrapped *around* the breaker would keep retrying + * the "circuit open" error with full exponential backoff instead of + * failing fast (see issue #1131). By wrapping retries *inside* the + * breaker action, opossum's `fire()` is called exactly once per external + * request, records exactly one success/failure per request, and — when + * already open — rejects immediately without invoking the action (and + * therefore without running any retry/backoff logic) at all. */ private _initCircuitBreakers(): void { const cbCfg: Required = { @@ -738,14 +766,19 @@ export class TTSService { rollingCountTimeout: cbCfg.rollingWindowMs, // Half-open retry delay resetTimeout: cbCfg.halfOpenIntervalMs, - // Per-call timeout (counted as a failure) + // Per-call timeout (counted as a failure). Bounds the whole retried + // request, since the action below includes the retry/backoff loop. timeout: cbCfg.timeoutMs, }; if (this.config.elevenlabs) { const elBreaker = new CircuitBreaker( async (text: string, voice: TTSVoice) => - generateElevenLabs(text, voice, this.config.elevenlabs!, cbCfg.timeoutMs), + withRetry( + () => generateElevenLabs(text, voice, this.config.elevenlabs!, cbCfg.timeoutMs), + this._retryConfig(), + "provider:elevenlabs" + ), { ...opossumOptions, name: "elevenlabs" } ); elBreaker.on("open", () => console.warn("[CircuitBreaker] ElevenLabs circuit OPENED — fast-failing")); @@ -757,7 +790,11 @@ export class TTSService { if (this.config.google) { const gBreaker = new CircuitBreaker( async (text: string, voice: TTSVoice) => - generateGoogle(text, voice, this.config.google!, cbCfg.timeoutMs), + withRetry( + () => generateGoogle(text, voice, this.config.google!, cbCfg.timeoutMs), + this._retryConfig(), + "provider:google" + ), { ...opossumOptions, name: "google" } ); gBreaker.on("open", () => console.warn("[CircuitBreaker] Google TTS circuit OPENED — fast-failing")); @@ -767,6 +804,13 @@ export class TTSService { } } + private _retryConfig(): RetryConfig { + return { + maxRetries: this.config.retry?.maxRetries ?? 3, + maxDelayMs: this.config.retry?.maxDelayMs ?? 60_000, + }; + } + /** * Returns a snapshot of each provider's circuit breaker state. * Exposed in /health/ready so operators can see the breaker state @@ -801,7 +845,7 @@ export class TTSService { rateLimitKey?: string, bypassCache?: boolean ): string { - if (this.config.auth) authenticate(credential, this.config.auth); + const owner = this._resolveOwner(credential); // Rate limiting if (this.rateLimiter && rateLimitKey) { @@ -821,6 +865,7 @@ export class TTSService { createdAt: new Date(), updatedAt: new Date(), bypassCache: bypassCache || false, + owner, }; jobStore.set(id, job); @@ -837,11 +882,43 @@ export class TTSService { return id; } - getJob(id: string): TTSJob | undefined { - return jobStore.get(id); + /** + * Resolve `credential` to a stable tenant identity, enforcing auth if + * configured. Used both to tag newly created jobs with their owner and to + * check ownership on lookup, so the two paths can never disagree. + */ + private _resolveOwner(credential?: string): string { + if (!this.config.auth) return ANONYMOUS_OWNER; + return authenticate(credential, this.config.auth); + } + + /** + * Look up a job by ID, scoped to the requesting credential's tenant. + * Returns `undefined` both when the job doesn't exist and when it exists + * but belongs to a different tenant — the two cases are indistinguishable + * to the caller so a credential cannot probe for other tenants' job IDs. + */ + getJob(id: string, credential?: string): TTSJob | undefined { + const owner = this._resolveOwner(credential); + const job = jobStore.get(id); + if (!job || job.owner !== owner) return undefined; + return job; + } + + /** List jobs belonging to the requesting credential's tenant, optionally filtered by status. */ + listJobs(status?: TTSJob["status"], credential?: string): TTSJob[] { + const owner = this._resolveOwner(credential); + const all = Array.from(jobStore.values()).filter((j) => j.owner === owner); + return status ? all.filter((j) => j.status === status) : all; } - listJobs(status?: TTSJob["status"]): TTSJob[] { + /** + * Returns jobs across *all* tenants, optionally filtered by status. + * For internal operational use only (health checks, queue-depth metrics) + * — never expose this over a tenant-facing API, since it bypasses the + * ownership scoping that `listJobs`/`getJob` enforce. + */ + listAllJobsUnscoped(status?: TTSJob["status"]): TTSJob[] { const all = Array.from(jobStore.values()); return status ? all.filter((j) => j.status === status) : all; } @@ -921,8 +998,11 @@ export class TTSService { } /** - * Try the requested provider with retry; if it fails and a fallback is available, try that. - * Transient errors (429, 5xx) are retried with exponential backoff + full jitter. + * Try the requested provider; if it fails and a fallback is available, try that. + * Transient errors (429, 5xx) are retried with exponential backoff + full jitter + * *inside* `_callProvider`'s circuit breaker action (see `_initCircuitBreakers`), + * so each provider is attempted at most once per call here — retries are not + * layered on top of the breaker, which would defeat its fast-fail behavior. * Non-retriable errors (400, 401, 403) propagate immediately. */ private async _generateWithFallback(job: TTSJob): Promise { @@ -931,17 +1011,8 @@ export class TTSService { const hasFallback = fallback === "google" ? !!this.config.google : !!this.config.elevenlabs; - const retryConfig: RetryConfig = { - maxRetries: this.config.retry?.maxRetries ?? 3, - maxDelayMs: this.config.retry?.maxDelayMs ?? 60_000, - }; - try { - return await withRetry( - () => this._callProvider(primary, job.text, job.voice), - retryConfig, - `provider:${primary}` - ); + return await this._callProvider(primary, job.text, job.voice); } catch (primaryErr) { const errMsg = primaryErr instanceof Error ? primaryErr.message : String(primaryErr); console.error(`[TTSService] Primary provider "${primary}" failed: ${errMsg}`); @@ -954,11 +1025,7 @@ export class TTSService { console.warn(`[TTSService] Falling back to "${fallback}"`); try { - return await withRetry( - () => this._callProvider(fallback, job.text, job.voice), - retryConfig, - `provider:${fallback}` - ); + return await this._callProvider(fallback, job.text, job.voice); } catch (fallbackErr) { const fbMsg = fallbackErr instanceof Error ? fallbackErr.message : String(fallbackErr); console.error(`[TTSService] Fallback provider "${fallback}" also failed: ${fbMsg}`); @@ -980,31 +1047,44 @@ export class TTSService { if (provider === "elevenlabs") { if (!this.config.elevenlabs) throw new TTSProviderError("elevenlabs", "ElevenLabs config missing"); if (breaker) { - try { - return await breaker.fire(text, voice) as Buffer; - } catch (err) { - // Re-wrap open-circuit errors as TTSProviderError so callers get a - // consistent error type and a meaningful 503 status code. - if (err instanceof Error && err.message.includes("open")) { - throw new TTSProviderError("elevenlabs", `Circuit breaker OPEN: ${err.message}`, 503); - } - throw err; - } + // Retries happen inside the breaker's action (see + // _initCircuitBreakers) — fire() is called exactly once here per + // external request, and rejects immediately without retry/backoff + // when the circuit is already open (fail-fast, not retried by any + // outer caller — see _generateWithFallback). + return await this._fireBreaker("elevenlabs", breaker, text, voice); } - return generateElevenLabs(text, voice, this.config.elevenlabs, this.cbConfig.timeoutMs); + return withRetry( + () => generateElevenLabs(text, voice, this.config.elevenlabs!, this.cbConfig.timeoutMs), + this._retryConfig(), + "provider:elevenlabs" + ); } else { if (!this.config.google) throw new TTSProviderError("google", "Google TTS config missing"); if (breaker) { - try { - return await breaker.fire(text, voice) as Buffer; - } catch (err) { - if (err instanceof Error && err.message.includes("open")) { - throw new TTSProviderError("google", `Circuit breaker OPEN: ${err.message}`, 503); - } - throw err; - } + return await this._fireBreaker("google", breaker, text, voice); + } + return withRetry( + () => generateGoogle(text, voice, this.config.google!, this.cbConfig.timeoutMs), + this._retryConfig(), + "provider:google" + ); + } + } + + private async _fireBreaker( + provider: TTSProvider, + breaker: CircuitBreaker, + text: string, + voice: TTSVoice + ): Promise { + try { + return await breaker.fire(text, voice) as Buffer; + } catch (err) { + if (err instanceof Error && err.message.includes("open")) { + throw new TTSProviderError(provider, `Circuit breaker OPEN: ${err.message}`, 503); } - return generateGoogle(text, voice, this.config.google, this.cbConfig.timeoutMs); + throw err; } } diff --git a/services/tts/src/__tests__/auth.test.ts b/services/tts/src/__tests__/auth.test.ts index bd873263..08c5cd0a 100644 --- a/services/tts/src/__tests__/auth.test.ts +++ b/services/tts/src/__tests__/auth.test.ts @@ -125,3 +125,41 @@ describe("TTSService.enqueue — auth", () => { expect(() => svc.enqueue("hello", VOICE, undefined, "bad.jwt.token")).toThrow(AuthError); }); }); + +// --------------------------------------------------------------------------- +// TTSService — cross-tenant job isolation (#1130) +// --------------------------------------------------------------------------- + +describe("TTSService — cross-tenant job isolation", () => { + it("getJob returns undefined for a job owned by a different credential", () => { + const svc = makeService(API_KEY_CONFIG); + const jobId = svc.enqueue("secret narration for tenant A", VOICE, undefined, "key-abc"); + + expect(svc.getJob(jobId, "key-abc")).toBeDefined(); + expect(svc.getJob(jobId, "key-xyz")).toBeUndefined(); + }); + + it("listJobs only returns jobs belonging to the requesting credential", () => { + const svc = makeService(API_KEY_CONFIG); + const jobA = svc.enqueue("tenant A job", VOICE, undefined, "key-abc"); + const jobB = svc.enqueue("tenant B job", VOICE, undefined, "key-xyz"); + + const jobsForA = svc.listJobs(undefined, "key-abc").map((j) => j.id); + const jobsForB = svc.listJobs(undefined, "key-xyz").map((j) => j.id); + + expect(jobsForA).toContain(jobA); + expect(jobsForA).not.toContain(jobB); + expect(jobsForB).toContain(jobB); + expect(jobsForB).not.toContain(jobA); + }); + + it("listAllJobsUnscoped returns jobs across every tenant", () => { + const svc = makeService(API_KEY_CONFIG); + const jobA = svc.enqueue("tenant A job", VOICE, undefined, "key-abc"); + const jobB = svc.enqueue("tenant B job", VOICE, undefined, "key-xyz"); + + const ids = svc.listAllJobsUnscoped().map((j) => j.id); + expect(ids).toContain(jobA); + expect(ids).toContain(jobB); + }); +}); diff --git a/services/tts/src/server.ts b/services/tts/src/server.ts index 4337abc5..b0aa7785 100644 --- a/services/tts/src/server.ts +++ b/services/tts/src/server.ts @@ -130,6 +130,13 @@ const ttsRateLimiter = rateLimit({ }); app.use(ttsRateLimiter); +/** Extract the bearer credential from the Authorization header, if present. */ +function extractCredential(req: Request): string | undefined { + const authHeader = req.headers.authorization; + if (!authHeader) return undefined; + return authHeader.replace(/^Bearer\s+/i, ""); +} + // Issue #723: Authentication middleware app.use((req: Request, res: Response, next: NextFunction) => { // Only /health/live (liveness probe) is exempt from auth — orchestrators @@ -142,12 +149,11 @@ app.use((req: Request, res: Response, next: NextFunction) => { } if (config.auth) { - const authHeader = req.headers.authorization; - if (!authHeader) { + const credential = extractCredential(req); + if (!credential) { return res.status(401).json({ error: "Missing Authorization header" }); } - const credential = authHeader.replace(/^Bearer\s+/i, ""); try { const { authenticate } = require("./TTSService"); authenticate(credential, config.auth); @@ -228,7 +234,8 @@ app.post("/tts/enqueue", (req: Request, res: Response) => { return res.status(400).json({ error: `Unknown voice: ${voiceId}` }); } - const jobId = service.enqueue(text, voice, provider, undefined, rateLimitKey, bypassCache); + const credential = extractCredential(req); + const jobId = service.enqueue(text, voice, provider, credential, rateLimitKey, bypassCache); res.json({ jobId, status: "pending" }); } catch (error: any) { const statusCode = error.statusCode || 500; @@ -251,11 +258,19 @@ app.post("/tts/enqueue", (req: Request, res: Response) => { * } */ app.get("/tts/job/:id", (req: Request, res: Response) => { - const job = service.getJob(req.params.id); - if (!job) { - return res.status(404).json({ error: "Job not found" }); + try { + const credential = extractCredential(req); + const job = service.getJob(req.params.id, credential); + if (!job) { + // Same response whether the job doesn't exist or belongs to another + // tenant, so a credential can't distinguish the two by probing IDs. + return res.status(404).json({ error: "Job not found" }); + } + res.json(job); + } catch (error: any) { + const statusCode = error.statusCode || 500; + res.status(statusCode).json({ error: error.message }); } - res.json(job); }); /** @@ -272,9 +287,15 @@ app.get("/tts/job/:id", (req: Request, res: Response) => { * ] */ app.get("/tts/jobs", (req: Request, res: Response) => { - const status = req.query.status as any; - const jobs = service.listJobs(status); - res.json(jobs); + try { + const status = req.query.status as any; + const credential = extractCredential(req); + const jobs = service.listJobs(status, credential); + res.json(jobs); + } catch (error: any) { + const statusCode = error.statusCode || 500; + res.status(statusCode).json({ error: error.message }); + } }); /** @@ -312,11 +333,12 @@ app.post("/tts/generate", async (req: Request, res: Response) => { return res.status(400).json({ error: `Unknown voice: ${voiceId}` }); } + const credential = extractCredential(req); const outputPath = await service.generate( text, voice, provider, - undefined, + credential, rateLimitKey, bypassCache, );