From 6f8df4920b9d0de88fb26ca21e5d718c8f52ce06 Mon Sep 17 00:00:00 2001 From: Mystery Date: Tue, 28 Jul 2026 06:20:54 +0000 Subject: [PATCH 1/4] perf(email): process SendGrid webhook batches concurrently handle_sendgrid_webhook looped over events sequentially, awaiting each process_event one at a time. A single batch of hundreds of events could hold a DB pool connection through hundreds of sequential round trips before responding, risking SendGrid webhook timeouts and pool contention. Process events concurrently with bounded parallelism (buffer_unordered) instead. --- services/api/src/email/webhook.rs | 26 +++++++++++++++++++++++--- 1 file changed, 23 insertions(+), 3 deletions(-) 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)); } } } From 907e74e23fa662dc507bb65cdb0a345099e38e54 Mon Sep 17 00:00:00 2001 From: Mystery Date: Tue, 28 Jul 2026 06:23:08 +0000 Subject: [PATCH 2/4] fix(email): derive idempotency key from job creation time, not wall-clock hour bucket idempotency_key rotated its hour bucket off SystemTime::now() on every call. mark_failed's exponential backoff (2/4/8/16/32/~64min...) crosses that 1-hour boundary by the 6th retry, so a retry scheduled more than an hour after the original attempt computed a different key and would not be deduplicated -- risking a duplicate send if the original attempt had actually succeeded at SendGrid before the worker crashed. The hour bucket is now derived from the job's original creation time (job.created_at) instead of the current time, so the key stays stable across a job's full retry lifetime regardless of backoff delay. --- services/api/src/email/queue.rs | 6 +- services/api/src/email/service.rs | 91 ++++++++++++++++++++++++------- 2 files changed, 76 insertions(+), 21 deletions(-) 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() { From 7d1878a1a914066da177f4c6d33fef00c0ff147e Mon Sep 17 00:00:00 2001 From: Mystery Date: Tue, 28 Jul 2026 06:26:55 +0000 Subject: [PATCH 3/4] fix(tts): fire circuit breaker once per request instead of once per retry attempt withRetry() wrapped _callProvider(), which itself called breaker.fire() on every retry attempt. A single failed user request could generate up to maxRetries+1 breaker failures instead of one, tripping the breaker's volumeThreshold/errorThresholdPercentage far earlier than intended. Once open, the breaker's OpenCircuitError was rewrapped as a 503 TTSProviderError, which isRetryable() treats as retryable, so the outer retry loop kept retrying an open breaker with full exponential backoff instead of failing fast -- the opposite of the breaker's purpose. Move the retry/backoff loop inside each breaker's wrapped action so opossum's fire() is called exactly once per external request. When the circuit is open, fire() now rejects immediately without invoking the action at all, so callers fail fast with zero retries. --- services/tts/src/TTSService.ts | 113 +++++++++++++++++++++------------ 1 file changed, 71 insertions(+), 42 deletions(-) diff --git a/services/tts/src/TTSService.ts b/services/tts/src/TTSService.ts index 6adc9c87..6d6cb6f8 100644 --- a/services/tts/src/TTSService.ts +++ b/services/tts/src/TTSService.ts @@ -616,8 +616,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 = { @@ -636,14 +646,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!), + withRetry( + () => generateElevenLabs(text, voice, this.config.elevenlabs!), + this._retryConfig(), + "provider:elevenlabs" + ), { ...opossumOptions, name: "elevenlabs" } ); elBreaker.on("open", () => console.warn("[CircuitBreaker] ElevenLabs circuit OPENED — fast-failing")); @@ -655,7 +670,11 @@ export class TTSService { if (this.config.google) { const gBreaker = new CircuitBreaker( async (text: string, voice: TTSVoice) => - generateGoogle(text, voice, this.config.google!), + withRetry( + () => generateGoogle(text, voice, this.config.google!), + this._retryConfig(), + "provider:google" + ), { ...opossumOptions, name: "google" } ); gBreaker.on("open", () => console.warn("[CircuitBreaker] Google TTS circuit OPENED — fast-failing")); @@ -665,6 +684,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 @@ -819,8 +845,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 { @@ -829,17 +858,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}`); @@ -852,11 +872,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}`); @@ -878,31 +894,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); + return withRetry( + () => generateElevenLabs(text, voice, this.config.elevenlabs!), + 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._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); + throw err; } } From 76708e69d09630561d7af18261986b9a4b0541cb Mon Sep 17 00:00:00 2001 From: Mystery Date: Tue, 28 Jul 2026 06:31:54 +0000 Subject: [PATCH 4/4] fix(tts): scope job lookup/listing to the requesting credential's tenant jobStore was a single global Map with no owner field. GET /tts/jobs and GET /tts/job/:id returned every job in the process -- including other tenants' narration text, voice, and file paths -- to any caller that passed the (optional) auth check, since neither endpoint filtered by the requesting credential. Tag each job with the owner identity resolved from its creating credential (the API key itself, or a JWT's `sub` claim; "anonymous" when no auth is configured), and require getJob/listJobs to match it. getJob returns the same 404 for "not found" and "belongs to another tenant" so a credential can't distinguish the two by probing IDs. Internal operational code (health checks, queue-depth metrics) that needs a cross-tenant view now uses the new listAllJobsUnscoped(). Also fixes /tts/enqueue and /tts/generate, which were never passing the authenticated request's credential down to the service layer at all. --- services/tts/src/HealthCheck.ts | 6 +-- services/tts/src/TTSService.ts | 63 ++++++++++++++++++++++--- services/tts/src/__tests__/auth.test.ts | 38 +++++++++++++++ services/tts/src/server.ts | 46 +++++++++++++----- 4 files changed, 132 insertions(+), 21 deletions(-) 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 6d6cb6f8..f96a31ab 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) // --------------------------------------------------------------------------- @@ -404,12 +414,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("."); @@ -427,6 +443,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; } // --------------------------------------------------------------------------- @@ -725,7 +743,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) { @@ -745,6 +763,7 @@ export class TTSService { createdAt: new Date(), updatedAt: new Date(), bypassCache: bypassCache || false, + owner, }; jobStore.set(id, job); @@ -761,11 +780,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; } - listJobs(status?: TTSJob["status"]): TTSJob[] { + /** 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; + } + + /** + * 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; } 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 da4804fd..9eeb5f86 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) => { // Skip auth for health checks @@ -138,12 +145,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); @@ -224,7 +230,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; @@ -247,11 +254,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); }); /** @@ -268,9 +283,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 }); + } }); /** @@ -308,11 +329,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, );