feat: add transcription billing via SQS events

pcrDev.json

@@ -1,6 +1,6 @@
 {
   "HashAlgorithm": "Sha384 { ... }",
-  "PCR0": "a5477fffcadd9957ea7175ae8ab548089661d857f51fcd7ee1d3ce6d539d212e97e1692696be812382f9bffd142e57e7",
+  "PCR0": "4d3e952005398e96804a4eb11ee62fbbe133dbc94ba5d36722ebd06a81f5d7e18d9ee394e68e74cf0bffcf490f9e0d59",
   "PCR1": "f004075c672258b499f8e88d59701031a3b451f65c7de60c81d09da2b0799272675481ec390527594dd7069cb7de59d7",
-  "PCR2": "a613344bbcb56a58b2ab25e3aa899ebe0b3638f05380575cec474749c9d583f482e086c6b0469506a2536f069278005c"
+  "PCR2": "d2faa92f1994561371eb6fa70f0d36ecc3ed3ce296c69441f488f6072cb585f9df9fa7466ab1a1074193769a759837fa"
 }

pcrDevHistory.json

@@ -530,5 +530,26 @@
     "PCR2": "a613344bbcb56a58b2ab25e3aa899ebe0b3638f05380575cec474749c9d583f482e086c6b0469506a2536f069278005c",
     "timestamp": 1770149174,
     "signature": "8ULP9CqMUbBFZlbaMr50DlN0pjNs4Os4B6iHAJkej24pZkaNcOYPRAVm9k5sKq3i0GWmomlPdCYXL98rL8CwfP5RVzb6NbDCyNMIwl+O1pzv8FWjQ1vOJhlTALGVBYcg"
+  },
+  {
+    "PCR0": "6e497a9660cdcb398ccaf89b98ab7e86721bd72ed05d013f4cb40448cf520dcc87b7026743ebb5bb364c359c317e674d",
+    "PCR1": "f004075c672258b499f8e88d59701031a3b451f65c7de60c81d09da2b0799272675481ec390527594dd7069cb7de59d7",
+    "PCR2": "543c194849f30bba1c750a4f8d44c2b19b8fee4ebe349398bfe4774a9c228b48bca018b594901f826bfcac23ecc6efd8",
+    "timestamp": 1770270906,
+    "signature": "1xEqoEb7rjt5B6t3oLha1lslI6Swc9W13q+YQW3eeiBlgjEqSZzLXzTTSPBIj/azDIQxiWEInqrnFR/rRZIaCA0QSsd5Yqi0aTpoCztLOCKXAkJGll3nK13DCya4pUyn"
+  },
+  {
+    "PCR0": "2f2a49dafccad29205374c0b1ef780305e931207f9073305f9e90fdaff2803ae53eb3756a1a70f026037a5311f684658",
+    "PCR1": "f004075c672258b499f8e88d59701031a3b451f65c7de60c81d09da2b0799272675481ec390527594dd7069cb7de59d7",
+    "PCR2": "da20625f6b50ff261450831cd2546d20f7b7bd3abd2bb629d0869d93f787ce0d583365064752a290ead813e78848bbec",
+    "timestamp": 1770312495,
+    "signature": "ZKhMycXvDPJys7bGI5J4NRUts6cAsW2/+X2bjQa5ID5k6bMQ6cLpibrgMCWZRjjxUHf6DYylxAw6wnXjGMRr4sLWYPV8Us4GnwiKHUxvaSb0VPxmmoCENTo+W+noeknS"
+  },
+  {
+    "PCR0": "4d3e952005398e96804a4eb11ee62fbbe133dbc94ba5d36722ebd06a81f5d7e18d9ee394e68e74cf0bffcf490f9e0d59",
+    "PCR1": "f004075c672258b499f8e88d59701031a3b451f65c7de60c81d09da2b0799272675481ec390527594dd7069cb7de59d7",
+    "PCR2": "d2faa92f1994561371eb6fa70f0d36ecc3ed3ce296c69441f488f6072cb585f9df9fa7466ab1a1074193769a759837fa",
+    "timestamp": 1770312803,
+    "signature": "ggAM5VObsCtd9qUPhzHDAu/c9OCZBT4eNnhvA8/ZhJQRSoRL/LkGIvEsyZp3pGYy699SOgFTDGj0salA9XqUh4WU3qE3CnsCkIOt7fuTC4X1sA8Vt1x0oAxmAFFtCYNL"
   }
 ]

pcrProd.json

@@ -1,6 +1,6 @@
 {
   "HashAlgorithm": "Sha384 { ... }",
-  "PCR0": "d17ce5561698287055c42b49c94ed1b4b0c55b4626e9841945e51266bfe31b5148e530fb9ba173d28e6939f6ea79f684",
+  "PCR0": "b8a5937060cf2b8abae9a6110e5625c0207c3437c39f0f75ad78a46f19fab89ffd571b1b1df452613840f4c950d0163e",
   "PCR1": "f004075c672258b499f8e88d59701031a3b451f65c7de60c81d09da2b0799272675481ec390527594dd7069cb7de59d7",
-  "PCR2": "985787c8f8510c0e93eda5a302bf43829c0b0bc615458414eb4964ed95924b9201038a0caab023aa25637b5227582504"
+  "PCR2": "09e28e526d7dc0fd2f4c8cb6a6dad434803a357b27b69072a17cb23efab96f1459d75a8c1c82821742f304001311b2b2"
 }

pcrProdHistory.json

@@ -530,5 +530,26 @@
     "PCR2": "985787c8f8510c0e93eda5a302bf43829c0b0bc615458414eb4964ed95924b9201038a0caab023aa25637b5227582504",
     "timestamp": 1770149200,
     "signature": "M9Tw62kKiIPdgMU7gmfcmASEdFEBARef2IIORXzEkJAaoz99Z+QUwhi7JxH97LYfebrtqv8/pv7oBST8OVAlnDEwDwW3mo1C1gILbQgqvF1vXSlMxObrUwFPR55LOI4v"
+  },
+  {
+    "PCR0": "14a227f68d65e7b79a7efb5e37667b9129a3f8ff0572cf5b8f43bfe3115bfe29dc8a262bbf208f6fa9c0503cbd0b1369",
+    "PCR1": "f004075c672258b499f8e88d59701031a3b451f65c7de60c81d09da2b0799272675481ec390527594dd7069cb7de59d7",
+    "PCR2": "0f978e15cdd89c2c305b941e1bc74e695b1f5d5ab2553826283730e12a54cd2da24d502272c30ab8ffbff36ae471b434",
+    "timestamp": 1770270934,
+    "signature": "m7TCqpEuQ9VYfTzU3XU27YT3lGnmjFTVncIwdB9eWt4NeB8e6JpuI951RfEvNd38W+zQmg/NH8JBZJggQkHbwoj/MXgaMMI5iBBYA0FT7aybhyK3sZ1RkjYau6pYvXGe"
+  },
+  {
+    "PCR0": "37f20a0b129ac39b9dc540ff3674ab35e8313275d169085b077f1fe50f8ea6461e847dc9174745421075d7da316c827a",
+    "PCR1": "f004075c672258b499f8e88d59701031a3b451f65c7de60c81d09da2b0799272675481ec390527594dd7069cb7de59d7",
+    "PCR2": "dbe4133d8ec89186a1011106d62df5c3cf36438d95f07276c5f648eb1166b937786f2b51a3af2aebe8d6c1edd8bf42d2",
+    "timestamp": 1770312536,
+    "signature": "ly0HfPkhMynnZSjB2DYwA2CCJiPPuiBOHHehB+CJnFEc2sx713Pc9st47Vxi2nZYcPJPAxbXq5db1ZkG2Du36CXTJe1iGu9IJDALDiJHQHBC0R+ct+OEkw3olHrR4qBd"
+  },
+  {
+    "PCR0": "b8a5937060cf2b8abae9a6110e5625c0207c3437c39f0f75ad78a46f19fab89ffd571b1b1df452613840f4c950d0163e",
+    "PCR1": "f004075c672258b499f8e88d59701031a3b451f65c7de60c81d09da2b0799272675481ec390527594dd7069cb7de59d7",
+    "PCR2": "09e28e526d7dc0fd2f4c8cb6a6dad434803a357b27b69072a17cb23efab96f1459d75a8c1c82821742f304001311b2b2",
+    "timestamp": 1770312844,
+    "signature": "FQZ1WaMtDl4t69dC/WezAW+Pt4lS+iMpBOMPiggPvVYExPdHTofqh7Yz9Ls5NjS97mk/PYnT4j8zc/YPs95mS7C/Ai0QMa7+JSv/jTw++DOk+N5qqaHJkq9FgdCY0QfD"
   }
 ]

src/sqs.rs

@@ -40,6 +40,10 @@ pub struct UsageEvent {
     pub provider_name: String,
     #[serde(default)]
     pub model_name: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub event_type: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub audio_seconds: Option<i32>,
 }
 
 #[derive(Debug, thiserror::Error)]

src/web/openai.rs

@@ -38,6 +38,28 @@ const MAX_AUDIO_SIZE: usize = 100 * 1024 * 1024;
 const REQUEST_TIMEOUT_SECS: u64 = 120; // Request timeout (generous for large non-streaming responses)
 const STREAM_CHUNK_TIMEOUT_SECS: u64 = 120; // Per-chunk timeout for streaming reads
 
+/// Estimate audio duration in seconds based on file size and content type.
+/// Uses average bitrate assumptions for common formats.
+fn estimate_audio_duration_seconds(file_size: usize, content_type: &str) -> f64 {
+    // Bytes per second for common audio formats (approximate averages)
+    let bytes_per_second: f64 = match content_type.to_lowercase().as_str() {
+        // MP3: typically 128-192kbps, use ~160kbps = 20KB/s
+        "audio/mpeg" | "audio/mp3" => 20_000.0,
+        // WAV: 44.1kHz, 16-bit stereo = 176.4KB/s
+        "audio/wav" | "audio/x-wav" | "audio/wave" => 176_400.0,
+        // OGG/Opus/WebM: typically ~96-128kbps = ~12-16KB/s, use 14KB/s
+        "audio/ogg" | "audio/opus" | "audio/webm" => 14_000.0,
+        // FLAC: lossless, roughly half of WAV = ~88KB/s
+        "audio/flac" | "audio/x-flac" => 88_000.0,
+        // AAC/M4A: similar to MP3, ~160kbps = 20KB/s
+        "audio/aac" | "audio/m4a" | "audio/mp4" | "audio/x-m4a" => 20_000.0,
+        // Default: assume MP3-like compression
+        _ => 20_000.0,
+    };
+
+    file_size as f64 / bytes_per_second
+}
+
 /// Parameters for transcription requests
 struct TranscriptionParams<'a> {
     audio_data: &'a [u8],
@@ -904,6 +926,8 @@ async fn publish_usage_event_internal(
                 is_api_request,
                 provider_name,
                 model_name,
+                event_type: None,
+                audio_seconds: None,
             };
 
             match publisher.publish_event(event).await {
@@ -914,6 +938,54 @@ async fn publish_usage_event_internal(
     });
 }
 
+/// Publish transcription usage event for billing
+/// Uses file size and content type to estimate audio duration
+fn publish_transcription_usage_event(
+    state: &Arc<AppState>,
+    user: &User,
+    file_size: usize,
+    content_type: &str,
+    model_name: &str,
+    provider_name: &str,
+    is_api_request: bool,
+) {
+    let estimated_seconds = estimate_audio_duration_seconds(file_size, content_type).ceil() as i32;
+
+    info!(
+        "Transcription usage for user {}: model={}, provider={}, file_size={} bytes, estimated_duration={}s",
+        user.uuid, model_name, provider_name, file_size, estimated_seconds,
+    );
+
+    // Spawn background task for SQS (no DB storage for transcription)
+    let state_clone = state.clone();
+    let user_id = user.uuid;
+    let model_name = model_name.to_string();
+    let provider_name = provider_name.to_string();
+
+    tokio::spawn(async move {
+        if let Some(publisher) = &state_clone.sqs_publisher {
+            let event = UsageEvent {
+                event_id: Uuid::new_v4(),
+                user_id,
+                input_tokens: 0,
+                output_tokens: 0,
+                estimated_cost: BigDecimal::from(0),
+                chat_time: Utc::now(),
+                is_api_request,
+                provider_name,
+                model_name,
+                event_type: Some("transcription".to_string()),
+                audio_seconds: Some(estimated_seconds),
+            };
+
+            match publisher.publish_event(event).await {
+                Ok(_) => debug!("published transcription usage event successfully"),
+                Err(e) => error!("error publishing transcription usage event: {e}"),
+            }
+        }
+    });
+}
+
 /// Helper to encrypt an SSE event
 async fn encrypt_sse_event(
     state: &AppState,
@@ -956,13 +1028,14 @@ async fn proxy_models(
 }
 
 /// Helper function to send transcription request with retries to primary and fallback providers
+/// Returns (response, provider_name) on success
 async fn send_transcription_with_retries(
     client: &Client<HttpsConnector<hyper::client::HttpConnector>, HyperBody>,
     route: &crate::proxy_config::ModelRoute,
     state: &Arc<AppState>,
     model_name: &str,
     params: &TranscriptionParams<'_>,
-) -> Result<Value, String> {
+) -> Result<(Value, String), String> {
     let max_cycles = 3;
     let mut last_error = None;
 
@@ -991,7 +1064,7 @@ async fn send_transcription_with_retries(
                     route.primary.provider_name,
                     cycle + 1
                 );
-                return Ok(response);
+                return Ok((response, route.primary.provider_name.clone()));
             }
             Err(err) => {
                 error!(
@@ -1023,7 +1096,7 @@ async fn send_transcription_with_retries(
                         fallback.provider_name,
                         cycle + 1
                     );
-                    return Ok(response);
+                    return Ok((response, fallback.provider_name.clone()));
                 }
                 Err(err) => {
                     error!(
@@ -1049,7 +1122,7 @@ async fn proxy_transcription(
     _headers: HeaderMap,
     axum::Extension(session_id): axum::Extension<Uuid>,
     axum::Extension(user): axum::Extension<User>,
-    axum::Extension(_auth_method): axum::Extension<AuthMethod>,
+    axum::Extension(auth_method): axum::Extension<AuthMethod>,
     axum::Extension(transcription_request): axum::Extension<TranscriptionRequest>,
 ) -> Result<Json<EncryptedResponse<Value>>, ApiError> {
     debug!("Entering proxy_transcription function");
@@ -1197,9 +1270,12 @@ async fn proxy_transcription(
             match send_transcription_with_retries(&client, &route, &state, &model_name, &params)
                 .await
             {
-                Ok(response) => {
-                    info!("Chunk {} transcribed successfully", chunk.index);
-                    Ok((chunk.index, response))
+                Ok((response, provider)) => {
+                    info!(
+                        "Chunk {} transcribed successfully via {}",
+                        chunk.index, provider
+                    );
+                    Ok((chunk.index, response, provider))
                 }
                 Err(err) => {
                     error!("Chunk {} failed: {}", chunk.index, err);
@@ -1212,13 +1288,21 @@ async fn proxy_transcription(
     }
 
     // Execute all futures in parallel
-    let results: Vec<Result<(usize, Value), String>> = futures::future::join_all(futures).await;
+    let results: Vec<Result<(usize, Value, String), String>> =
+        futures::future::join_all(futures).await;
 
     // Check if all chunks succeeded
     let mut successful_results = Vec::new();
+    let mut successful_provider = String::new();
     for result in results {
         match result {
-            Ok(r) => successful_results.push(r),
+            Ok((index, value, provider)) => {
+                successful_results.push((index, value));
+                // Track the provider (use the first one, or could use the last)
+                if successful_provider.is_empty() {
+                    successful_provider = provider;
+                }
+            }
             Err(e) => {
                 error!("Chunk processing failed: {}", e);
                 return Err(ApiError::InternalServerError);
@@ -1265,8 +1349,16 @@ async fn proxy_transcription(
 
     debug!("Exiting proxy_transcription function");
 
-    // TODO: Add SQS-based billing events for transcription usage
-    // Should track: audio duration/size, model used, user ID, timestamp, provider
+    // Publish transcription usage for billing (runs in background)
+    publish_transcription_usage_event(
+        &state,
+        &user,
+        file_size,
+        &transcription_request.content_type,
+        &transcription_request.model,
+        &successful_provider,
+        auth_method == AuthMethod::ApiKey,
+    );
 
     // Encrypt and return the response
     encrypt_response(&state, &session_id, &response).await
@@ -1819,3 +1911,48 @@ async fn try_provider(
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_estimate_audio_duration_mp3() {
+        // MP3 at ~20KB/s, so 1MB should be ~50 seconds
+        let one_mb = 1024 * 1024;
+        let duration = estimate_audio_duration_seconds(one_mb, "audio/mpeg");
+        assert!((duration - 52.4).abs() < 1.0); // ~52.4s for 1MB at 20KB/s
+    }
+
+    #[test]
+    fn test_estimate_audio_duration_wav() {
+        // WAV at ~176KB/s, so 1MB should be ~6 seconds
+        let one_mb = 1024 * 1024;
+        let duration = estimate_audio_duration_seconds(one_mb, "audio/wav");
+        assert!((duration - 5.9).abs() < 1.0); // ~5.9s for 1MB at 176KB/s
+    }
+
+    #[test]
+    fn test_estimate_audio_duration_ogg() {
+        // OGG at ~14KB/s, so 1MB should be ~75 seconds
+        let one_mb = 1024 * 1024;
+        let duration = estimate_audio_duration_seconds(one_mb, "audio/ogg");
+        assert!((duration - 74.9).abs() < 1.0);
+    }
+
+    #[test]
+    fn test_estimate_audio_duration_unknown_defaults_to_mp3() {
+        let one_mb = 1024 * 1024;
+        let duration_unknown = estimate_audio_duration_seconds(one_mb, "audio/unknown");
+        let duration_mp3 = estimate_audio_duration_seconds(one_mb, "audio/mpeg");
+        assert_eq!(duration_unknown, duration_mp3);
+    }
+
+    #[test]
+    fn test_estimate_audio_duration_case_insensitive() {
+        let one_mb = 1024 * 1024;
+        let duration_lower = estimate_audio_duration_seconds(one_mb, "audio/mpeg");
+        let duration_upper = estimate_audio_duration_seconds(one_mb, "AUDIO/MPEG");
+        assert_eq!(duration_lower, duration_upper);
+    }
+}