1use std::sync::Arc;
2
3use axum::body::Bytes;
4use axum::extract::{Path, State};
5use axum::http::header::{CACHE_CONTROL, CONTENT_TYPE};
6use axum::http::{HeaderMap, HeaderValue, StatusCode};
7use axum::response::{IntoResponse, Response};
8use axum::routing::{get, post, put};
9use axum::{Extension, Json, Router};
10use kcode_k1_access_full_audio::{
11 AccessId, FullAudioState, GroupId, K1AccessFullAudio, PersonId, SpeakerLabelV1, UserId,
12};
13use kcode_k1_access_profiles::{ProfileId, ProfileSelection, TxId};
14use kcode_k1_http::Principal;
15use serde::{Deserialize, Serialize};
16use serde_json::Value;
17
18pub fn authenticated_routes(audio: Arc<K1AccessFullAudio>) -> Router<()> {
19 Router::new()
20 .route("/audio", get(list_audio))
21 .route("/audio/profiles/{profile_id}", post(submit_audio))
22 .route("/audio/groups/{group_id}", get(list_group_audio))
23 .route("/audio/{access_id}", get(get_audio))
24 .route(
25 "/audio/{access_id}/fragments/{fragment_id}/audio",
26 get(get_fragment_audio),
27 )
28 .route(
29 "/audio/{access_id}/fragments/{fragment_id}/labels",
30 put(submit_labels),
31 )
32 .route(
33 "/audio/{access_id}/fragments/{fragment_id}/retry",
34 post(retry_fragment),
35 )
36 .route(
37 "/audio/{access_id}/fragments/{fragment_id}/discard",
38 post(discard_fragment),
39 )
40 .with_state(audio)
41}
42
43#[derive(Debug)]
44struct ApiError {
45 status: StatusCode,
46 code: &'static str,
47 message: String,
48}
49type ApiResult<T> = Result<T, ApiError>;
50#[derive(Serialize)]
51struct ErrorDto {
52 error: &'static str,
53 message: String,
54}
55impl IntoResponse for ApiError {
56 fn into_response(self) -> Response {
57 json_response(
58 self.status,
59 ErrorDto {
60 error: self.code,
61 message: self.message,
62 },
63 )
64 }
65}
66fn problem(status: StatusCode, code: &'static str, message: impl Into<String>) -> ApiError {
67 ApiError {
68 status,
69 code,
70 message: message.into(),
71 }
72}
73fn invalid_id(code: &'static str, subject: &str) -> ApiError {
74 problem(
75 StatusCode::BAD_REQUEST,
76 code,
77 format!("{subject} ID is not canonical lowercase hexadecimal"),
78 )
79}
80fn dependency_error(operation: &str, message: String) -> ApiError {
81 if message.contains("fragment does not belong to full audio") {
82 return problem(
83 StatusCode::NOT_FOUND,
84 "fragment_not_found",
85 "fragment is unavailable to the caller",
86 );
87 }
88 if message.contains("cannot access full audio") || message == "access denied" {
89 return problem(
90 StatusCode::NOT_FOUND,
91 "audio_not_found",
92 "audio is unavailable to the caller",
93 );
94 }
95 problem(
96 StatusCode::SERVICE_UNAVAILABLE,
97 "audio_unavailable",
98 format!("{operation}: {message}"),
99 )
100}
101fn json_response<T: Serialize>(status: StatusCode, value: T) -> Response {
102 let mut response = (status, Json(value)).into_response();
103 response
104 .headers_mut()
105 .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
106 response
107}
108fn no_content() -> Response {
109 let mut response = StatusCode::NO_CONTENT.into_response();
110 response
111 .headers_mut()
112 .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
113 response
114}
115async fn run<T, F>(operation: &'static str, task: F) -> ApiResult<T>
116where
117 T: Send + 'static,
118 F: FnOnce() -> Result<T, String> + Send + 'static,
119{
120 match tokio::task::spawn_blocking(task).await {
121 Ok(Ok(value)) => Ok(value),
122 Ok(Err(message)) => Err(dependency_error(operation, message)),
123 Err(_) => Err(problem(
124 StatusCode::SERVICE_UNAVAILABLE,
125 "audio_unavailable",
126 format!("{operation}: blocking task failed"),
127 )),
128 }
129}
130fn caller(principal: &Principal) -> UserId {
131 UserId::from_tx_id(TxId::from_bytes(*principal.user_id()))
132}
133fn parse_txid(value: &str, code: &'static str, subject: &str) -> ApiResult<TxId> {
134 if value.len() != 24 {
135 return Err(invalid_id(code, subject));
136 }
137 let mut output = [0; 12];
138 for (index, pair) in value.as_bytes().chunks_exact(2).enumerate() {
139 let digit = |byte| match byte {
140 b'0'..=b'9' => Some(byte - b'0'),
141 b'a'..=b'f' => Some(byte - b'a' + 10),
142 _ => None,
143 };
144 output[index] = digit(pair[0])
145 .zip(digit(pair[1]))
146 .map(|(high, low)| high << 4 | low)
147 .ok_or_else(|| invalid_id(code, subject))?;
148 }
149 Ok(TxId::from_bytes(output))
150}
151fn parse_access_id(value: &str) -> ApiResult<AccessId> {
152 parse_txid(value, "invalid_access_id", "access").map(AccessId::new)
153}
154fn parse_fragment_id(value: &str) -> ApiResult<TxId> {
155 parse_txid(value, "invalid_fragment_id", "fragment")
156}
157fn parse_group_id(value: &str) -> ApiResult<GroupId> {
158 parse_txid(value, "invalid_group_id", "group").map(GroupId::new)
159}
160fn parse_profile_id(value: &str) -> ApiResult<ProfileId> {
161 parse_txid(value, "invalid_profile_id", "profile").map(ProfileId::new)
162}
163fn hex(bytes: &[u8; 12]) -> String {
164 const DIGITS: &[u8; 16] = b"0123456789abcdef";
165 let mut value = String::with_capacity(24);
166 for byte in bytes {
167 value.push(DIGITS[(byte >> 4) as usize] as char);
168 value.push(DIGITS[(byte & 15) as usize] as char);
169 }
170 value
171}
172fn access_hex(id: &AccessId) -> String {
173 hex(id.txid().as_bytes())
174}
175fn require_content_type(headers: &HeaderMap, expected: &str) -> ApiResult<()> {
176 let mut values = headers.get_all(CONTENT_TYPE).iter();
177 let valid = values
178 .next()
179 .and_then(|value| value.to_str().ok())
180 .is_some_and(|value| {
181 value
182 .split(';')
183 .next()
184 .is_some_and(|media| media.trim().eq_ignore_ascii_case(expected))
185 })
186 && values.next().is_none();
187 if valid {
188 Ok(())
189 } else {
190 Err(problem(
191 StatusCode::UNSUPPORTED_MEDIA_TYPE,
192 "unsupported_media_type",
193 "request content type is unsupported",
194 ))
195 }
196}
197#[derive(Serialize)]
198struct SubmittedDto {
199 access_id: String,
200 full_audio_id: String,
201}
202#[derive(Serialize)]
203struct FragmentDto {
204 fragment_id: String,
205 start_sample_48k: u64,
206 end_sample_48k: u64,
207 status: Value,
208}
209#[derive(Serialize)]
210struct StatusDto {
211 state: &'static str,
212 fragments: Vec<FragmentDto>,
213 final_transcript: Option<String>,
214}
215fn state_name(state: FullAudioState) -> &'static str {
216 match state {
217 FullAudioState::Processing => "processing",
218 FullAudioState::AwaitingLabels => "awaiting_labels",
219 FullAudioState::NeedsAttention => "needs_attention",
220 FullAudioState::Complete => "complete",
221 }
222}
223#[derive(Deserialize)]
224struct LabelsDto {
225 labels: Vec<LabelDto>,
226}
227#[derive(Deserialize)]
228struct LabelDto {
229 speaker: String,
230 person_id: Option<String>,
231}
232fn parse_labels(input: LabelsDto) -> ApiResult<Vec<SpeakerLabelV1>> {
233 input
234 .labels
235 .into_iter()
236 .map(|label| {
237 let person_id = label
238 .person_id
239 .map(|value| {
240 parse_txid(&value, "invalid_person_id", "person").map(PersonId::from_tx_id)
241 })
242 .transpose()?;
243 Ok(SpeakerLabelV1 {
244 speaker: label.speaker.parse().map_err(|_| {
245 problem(
246 StatusCode::BAD_REQUEST,
247 "invalid_labels",
248 "speaker label is invalid",
249 )
250 })?,
251 person_id,
252 })
253 })
254 .collect()
255}
256async fn submit_audio(
257 State(audio): State<Arc<K1AccessFullAudio>>,
258 Extension(principal): Extension<Principal>,
259 Path(value): Path<String>,
260 headers: HeaderMap,
261 body: Bytes,
262) -> ApiResult<Response> {
263 require_content_type(&headers, "application/octet-stream")?;
264 let profile = ProfileSelection::Saved(parse_profile_id(&value)?);
265 let user = caller(&principal);
266 let submitted = run("submit audio", move || {
267 audio.submit_for_user(user, profile, &body)
268 })
269 .await?;
270 Ok(json_response(
271 StatusCode::CREATED,
272 SubmittedDto {
273 access_id: access_hex(&submitted.access_id),
274 full_audio_id: hex(submitted.full_audio_id.as_bytes()),
275 },
276 ))
277}
278async fn list_audio(
279 State(audio): State<Arc<K1AccessFullAudio>>,
280 Extension(principal): Extension<Principal>,
281) -> ApiResult<Response> {
282 let user = caller(&principal);
283 let ids = run("list audio", move || audio.list_for_user(user)).await?;
284 Ok(json_response(
285 StatusCode::OK,
286 ids.iter().map(access_hex).collect::<Vec<_>>(),
287 ))
288}
289async fn list_group_audio(
290 State(audio): State<Arc<K1AccessFullAudio>>,
291 Extension(principal): Extension<Principal>,
292 Path(value): Path<String>,
293) -> ApiResult<Response> {
294 let group = parse_group_id(&value)?;
295 let user = caller(&principal);
296 let ids = run("list group audio", move || {
297 audio.list_group_for_user(user, group)
298 })
299 .await?;
300 Ok(json_response(
301 StatusCode::OK,
302 ids.iter().map(access_hex).collect::<Vec<_>>(),
303 ))
304}
305async fn get_audio(
306 State(audio): State<Arc<K1AccessFullAudio>>,
307 Extension(principal): Extension<Principal>,
308 Path(value): Path<String>,
309) -> ApiResult<Response> {
310 let access_id = parse_access_id(&value)?;
311 let user = caller(&principal);
312 let status = run("get audio status", move || {
313 audio.status_for_user(user, access_id)
314 })
315 .await?;
316 let state = state_name(status.state);
317 let mut fragments = Vec::with_capacity(status.fragments.len());
318 for fragment in status.fragments {
319 let serialized = serde_json::to_value(fragment.status).map_err(|_| {
320 problem(
321 StatusCode::SERVICE_UNAVAILABLE,
322 "audio_unavailable",
323 "serialize audio status: status serialization failed",
324 )
325 })?;
326 fragments.push(FragmentDto {
327 fragment_id: hex(fragment.fragment_id.as_bytes()),
328 start_sample_48k: fragment.start_sample_48k,
329 end_sample_48k: fragment.end_sample_48k,
330 status: serialized,
331 });
332 }
333 Ok(json_response(
334 StatusCode::OK,
335 StatusDto {
336 state,
337 fragments,
338 final_transcript: status.final_transcript,
339 },
340 ))
341}
342async fn get_fragment_audio(
343 State(audio): State<Arc<K1AccessFullAudio>>,
344 Extension(principal): Extension<Principal>,
345 Path((access_value, fragment_value)): Path<(String, String)>,
346) -> ApiResult<Response> {
347 let access_id = parse_access_id(&access_value)?;
348 let fragment_id = parse_fragment_id(&fragment_value)?;
349 let user = caller(&principal);
350 let bytes = run("get fragment audio", move || {
351 audio.fragment_audio_for_user(user, access_id, fragment_id)
352 })
353 .await?;
354 Ok((
355 [
356 (CONTENT_TYPE, HeaderValue::from_static("audio/ogg")),
357 (CACHE_CONTROL, HeaderValue::from_static("no-store")),
358 ],
359 bytes,
360 )
361 .into_response())
362}
363async fn submit_labels(
364 State(audio): State<Arc<K1AccessFullAudio>>,
365 Extension(principal): Extension<Principal>,
366 Path((access_value, fragment_value)): Path<(String, String)>,
367 headers: HeaderMap,
368 body: Bytes,
369) -> ApiResult<Response> {
370 require_content_type(&headers, "application/json")?;
371 let input: LabelsDto = serde_json::from_slice(&body).map_err(|_| {
372 problem(
373 StatusCode::BAD_REQUEST,
374 "invalid_json",
375 "request body is invalid",
376 )
377 })?;
378 let labels = parse_labels(input)?;
379 let access_id = parse_access_id(&access_value)?;
380 let fragment_id = parse_fragment_id(&fragment_value)?;
381 let user = caller(&principal);
382 run("submit fragment labels", move || {
383 audio.submit_labels_for_user(user, access_id, fragment_id, labels)
384 })
385 .await?;
386 Ok(no_content())
387}
388async fn retry_fragment(
389 State(audio): State<Arc<K1AccessFullAudio>>,
390 Extension(principal): Extension<Principal>,
391 Path((access_value, fragment_value)): Path<(String, String)>,
392) -> ApiResult<Response> {
393 let access_id = parse_access_id(&access_value)?;
394 let fragment_id = parse_fragment_id(&fragment_value)?;
395 let user = caller(&principal);
396 run("retry fragment", move || {
397 audio.retry_fragment_for_user(user, access_id, fragment_id)
398 })
399 .await?;
400 Ok(no_content())
401}
402async fn discard_fragment(
403 State(audio): State<Arc<K1AccessFullAudio>>,
404 Extension(principal): Extension<Principal>,
405 Path((access_value, fragment_value)): Path<(String, String)>,
406) -> ApiResult<Response> {
407 let access_id = parse_access_id(&access_value)?;
408 let fragment_id = parse_fragment_id(&fragment_value)?;
409 let user = caller(&principal);
410 run("discard fragment", move || {
411 audio.discard_fragment_for_user(user, access_id, fragment_id)
412 })
413 .await?;
414 Ok(no_content())
415}
416#[cfg(test)]
417mod tests {
418 use super::*;
419 use serde_json::json;
420 #[test]
421 fn canonical_ids_and_user_conversion_preserve_bytes() {
422 let id = "00112233445566778899aabb";
423 assert_eq!(hex(parse_fragment_id(id).unwrap().as_bytes()), id);
424 assert!(parse_fragment_id("00112233445566778899AABB").is_err());
425 assert!(parse_fragment_id("0011").is_err());
426 assert_eq!(
427 hex(caller_bytes([1; 12]).as_tx_id().as_bytes()),
428 "010101010101010101010101"
429 );
430 }
431 fn caller_bytes(bytes: [u8; 12]) -> UserId {
432 UserId::from_tx_id(TxId::from_bytes(bytes))
433 }
434 #[test]
435 fn labels_accept_known_and_unknown_people() {
436 let labels = parse_labels(LabelsDto {
437 labels: vec![
438 LabelDto {
439 speaker: "Speaker 1".to_owned(),
440 person_id: None,
441 },
442 LabelDto {
443 speaker: "Speaker 2".to_owned(),
444 person_id: Some("00112233445566778899aabb".to_owned()),
445 },
446 ],
447 })
448 .unwrap();
449 assert_eq!(labels.len(), 2);
450 assert!(labels[0].person_id.is_none());
451 assert_eq!(
452 hex(labels[1].person_id.unwrap().as_tx_id().as_bytes()),
453 "00112233445566778899aabb"
454 );
455 assert!(
456 parse_labels(LabelsDto {
457 labels: vec![LabelDto {
458 speaker: "Unknown".to_owned(),
459 person_id: None
460 }]
461 })
462 .is_err()
463 );
464 }
465 #[test]
466 fn status_shape_and_hidden_errors_are_stable() {
467 let value = serde_json::to_value(StatusDto {
468 state: "processing",
469 fragments: Vec::new(),
470 final_transcript: None,
471 })
472 .unwrap();
473 assert_eq!(
474 value,
475 json!({"state":"processing","fragments":[],"final_transcript":null})
476 );
477 let error = dependency_error(
478 "get audio status",
479 "principal cannot access full audio".to_owned(),
480 );
481 assert_eq!(error.status, StatusCode::NOT_FOUND);
482 assert_eq!(error.code, "audio_not_found");
483 assert!(!error.message.contains("principal"));
484 }
485 #[test]
486 fn content_types_and_public_route_assembler_are_exact() {
487 let mut headers = HeaderMap::new();
488 headers.insert(
489 CONTENT_TYPE,
490 HeaderValue::from_static("application/json; charset=utf-8"),
491 );
492 assert!(require_content_type(&headers, "application/json").is_ok());
493 assert!(require_content_type(&headers, "application/octet-stream").is_err());
494 let _ = authenticated_routes as fn(Arc<K1AccessFullAudio>) -> Router<()>;
495 }
496}