kcode_k1_http_audio/

lib.rs

use std::sync::Arc;

use axum::body::Bytes;
use axum::extract::{DefaultBodyLimit, Path, State};
use axum::http::header::{CACHE_CONTROL, CONTENT_TYPE};
use axum::http::{HeaderMap, HeaderValue, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post, put};
use axum::{Extension, Json, Router};
use kcode_k1_access_full_audio::{
    AccessId, FullAudioState, GroupId, K1AccessFullAudio, PersonId, SpeakerLabelV1, UserId,
};
use kcode_k1_access_profiles::{ProfileId, ProfileSelection, TxId};
use kcode_k1_http::Principal;
use serde::{Deserialize, Serialize};
use serde_json::Value;

pub fn authenticated_routes(audio: Arc<K1AccessFullAudio>) -> Router<()> {
    Router::new()
        .route("/audio", get(list_audio))
        .route(
            "/audio/profiles/{profile_id}",
            post(submit_audio).layer(upload_body_limit()),
        )
        .route("/audio/groups/{group_id}", get(list_group_audio))
        .route("/audio/{access_id}", get(get_audio))
        .route(
            "/audio/{access_id}/fragments/{fragment_id}/audio",
            get(get_fragment_audio),
        )
        .route(
            "/audio/{access_id}/fragments/{fragment_id}/labels",
            put(submit_labels),
        )
        .route(
            "/audio/{access_id}/fragments/{fragment_id}/retry",
            post(retry_fragment),
        )
        .route(
            "/audio/{access_id}/fragments/{fragment_id}/discard",
            post(discard_fragment),
        )
        .with_state(audio)
}

fn upload_body_limit() -> DefaultBodyLimit {
    DefaultBodyLimit::disable()
}

#[derive(Debug)]
struct ApiError {
    status: StatusCode,
    code: &'static str,
    message: String,
}
type ApiResult<T> = Result<T, ApiError>;

#[derive(Serialize)]
struct ErrorDto {
    error: &'static str,
    message: String,
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        json_response(
            self.status,
            ErrorDto {
                error: self.code,
                message: self.message,
            },
        )
    }
}

fn problem(status: StatusCode, code: &'static str, message: impl Into<String>) -> ApiError {
    ApiError {
        status,
        code,
        message: message.into(),
    }
}

fn invalid_id(code: &'static str, subject: &str) -> ApiError {
    problem(
        StatusCode::BAD_REQUEST,
        code,
        format!("{subject} ID is not canonical lowercase hexadecimal"),
    )
}

fn dependency_error(operation: &str, message: String) -> ApiError {
    if message.contains("fragment does not belong to full audio") {
        return problem(
            StatusCode::NOT_FOUND,
            "fragment_not_found",
            "fragment is unavailable to the caller",
        );
    }
    if message.contains("cannot access full audio") || message == "access denied" {
        return problem(
            StatusCode::NOT_FOUND,
            "audio_not_found",
            "audio is unavailable to the caller",
        );
    }
    problem(
        StatusCode::SERVICE_UNAVAILABLE,
        "audio_unavailable",
        format!("{operation}: {message}"),
    )
}

fn json_response<T: Serialize>(status: StatusCode, value: T) -> Response {
    let mut response = (status, Json(value)).into_response();
    response
        .headers_mut()
        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
    response
}

fn no_content() -> Response {
    let mut response = StatusCode::NO_CONTENT.into_response();
    response
        .headers_mut()
        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
    response
}

async fn run<T, F>(operation: &'static str, task: F) -> ApiResult<T>
where
    T: Send + 'static,
    F: FnOnce() -> Result<T, String> + Send + 'static,
{
    match tokio::task::spawn_blocking(task).await {
        Ok(Ok(value)) => Ok(value),
        Ok(Err(message)) => Err(dependency_error(operation, message)),
        Err(_) => Err(problem(
            StatusCode::SERVICE_UNAVAILABLE,
            "audio_unavailable",
            format!("{operation}: blocking task failed"),
        )),
    }
}

fn caller(principal: &Principal) -> UserId {
    UserId::from_tx_id(TxId::from_bytes(*principal.user_id()))
}

fn parse_txid(value: &str, code: &'static str, subject: &str) -> ApiResult<TxId> {
    if value.len() != 24 {
        return Err(invalid_id(code, subject));
    }
    let mut output = [0; 12];
    for (index, pair) in value.as_bytes().chunks_exact(2).enumerate() {
        let digit = |byte| match byte {
            b'0'..=b'9' => Some(byte - b'0'),
            b'a'..=b'f' => Some(byte - b'a' + 10),
            _ => None,
        };
        output[index] = digit(pair[0])
            .zip(digit(pair[1]))
            .map(|(high, low)| high << 4 | low)
            .ok_or_else(|| invalid_id(code, subject))?;
    }
    Ok(TxId::from_bytes(output))
}

fn parse_access_id(value: &str) -> ApiResult<AccessId> {
    parse_txid(value, "invalid_access_id", "access").map(AccessId::new)
}

fn parse_fragment_id(value: &str) -> ApiResult<TxId> {
    parse_txid(value, "invalid_fragment_id", "fragment")
}

fn parse_group_id(value: &str) -> ApiResult<GroupId> {
    parse_txid(value, "invalid_group_id", "group").map(GroupId::new)
}

fn parse_profile_id(value: &str) -> ApiResult<ProfileId> {
    parse_txid(value, "invalid_profile_id", "profile").map(ProfileId::new)
}

fn hex(bytes: &[u8; 12]) -> String {
    const DIGITS: &[u8; 16] = b"0123456789abcdef";
    let mut value = String::with_capacity(24);
    for byte in bytes {
        value.push(DIGITS[(byte >> 4) as usize] as char);
        value.push(DIGITS[(byte & 15) as usize] as char);
    }
    value
}

fn access_hex(id: &AccessId) -> String {
    hex(id.txid().as_bytes())
}

fn require_content_type(headers: &HeaderMap, expected: &str) -> ApiResult<()> {
    let mut values = headers.get_all(CONTENT_TYPE).iter();
    let valid = values
        .next()
        .and_then(|value| value.to_str().ok())
        .is_some_and(|value| {
            value
                .split(';')
                .next()
                .is_some_and(|media| media.trim().eq_ignore_ascii_case(expected))
        })
        && values.next().is_none();
    if valid {
        Ok(())
    } else {
        Err(problem(
            StatusCode::UNSUPPORTED_MEDIA_TYPE,
            "unsupported_media_type",
            "request content type is unsupported",
        ))
    }
}

#[derive(Serialize)]
struct SubmittedDto {
    access_id: String,
    full_audio_id: String,
}

#[derive(Serialize)]
struct FragmentDto {
    fragment_id: String,
    start_sample_48k: u64,
    end_sample_48k: u64,
    status: Value,
}

#[derive(Serialize)]
struct StatusDto {
    state: &'static str,
    fragments: Vec<FragmentDto>,
    final_transcript: Option<String>,
}

fn state_name(state: FullAudioState) -> &'static str {
    match state {
        FullAudioState::Processing => "processing",
        FullAudioState::AwaitingLabels => "awaiting_labels",
        FullAudioState::NeedsAttention => "needs_attention",
        FullAudioState::Complete => "complete",
    }
}

#[derive(Deserialize)]
struct LabelsDto {
    labels: Vec<LabelDto>,
}

#[derive(Deserialize)]
struct LabelDto {
    speaker: String,
    person_id: Option<String>,
}

fn parse_labels(input: LabelsDto) -> ApiResult<Vec<SpeakerLabelV1>> {
    input
        .labels
        .into_iter()
        .map(|label| {
            let person_id = label
                .person_id
                .map(|value| {
                    parse_txid(&value, "invalid_person_id", "person").map(PersonId::from_tx_id)
                })
                .transpose()?;
            Ok(SpeakerLabelV1 {
                speaker: label.speaker.parse().map_err(|_| {
                    problem(
                        StatusCode::BAD_REQUEST,
                        "invalid_labels",
                        "speaker label is invalid",
                    )
                })?,
                person_id,
            })
        })
        .collect()
}

async fn submit_audio(
    State(audio): State<Arc<K1AccessFullAudio>>,
    Extension(principal): Extension<Principal>,
    Path(value): Path<String>,
    headers: HeaderMap,
    body: Bytes,
) -> ApiResult<Response> {
    require_content_type(&headers, "application/octet-stream")?;
    let profile = ProfileSelection::Saved(parse_profile_id(&value)?);
    let user = caller(&principal);
    let submitted = run("submit audio", move || {
        audio.submit_for_user(user, profile, &body)
    })
    .await?;
    Ok(json_response(
        StatusCode::CREATED,
        SubmittedDto {
            access_id: access_hex(&submitted.access_id),
            full_audio_id: hex(submitted.full_audio_id.as_bytes()),
        },
    ))
}

async fn list_audio(
    State(audio): State<Arc<K1AccessFullAudio>>,
    Extension(principal): Extension<Principal>,
) -> ApiResult<Response> {
    let user = caller(&principal);
    let ids = run("list audio", move || audio.list_for_user(user)).await?;
    Ok(json_response(
        StatusCode::OK,
        ids.iter().map(access_hex).collect::<Vec<_>>(),
    ))
}

async fn list_group_audio(
    State(audio): State<Arc<K1AccessFullAudio>>,
    Extension(principal): Extension<Principal>,
    Path(value): Path<String>,
) -> ApiResult<Response> {
    let group = parse_group_id(&value)?;
    let user = caller(&principal);
    let ids = run("list group audio", move || {
        audio.list_group_for_user(user, group)
    })
    .await?;
    Ok(json_response(
        StatusCode::OK,
        ids.iter().map(access_hex).collect::<Vec<_>>(),
    ))
}

async fn get_audio(
    State(audio): State<Arc<K1AccessFullAudio>>,
    Extension(principal): Extension<Principal>,
    Path(value): Path<String>,
) -> ApiResult<Response> {
    let access_id = parse_access_id(&value)?;
    let user = caller(&principal);
    let status = run("get audio status", move || {
        audio.status_for_user(user, access_id)
    })
    .await?;
    let state = state_name(status.state);
    let mut fragments = Vec::with_capacity(status.fragments.len());
    for fragment in status.fragments {
        let serialized = serde_json::to_value(fragment.status).map_err(|_| {
            problem(
                StatusCode::SERVICE_UNAVAILABLE,
                "audio_unavailable",
                "serialize audio status: status serialization failed",
            )
        })?;
        fragments.push(FragmentDto {
            fragment_id: hex(fragment.fragment_id.as_bytes()),
            start_sample_48k: fragment.start_sample_48k,
            end_sample_48k: fragment.end_sample_48k,
            status: serialized,
        });
    }
    Ok(json_response(
        StatusCode::OK,
        StatusDto {
            state,
            fragments,
            final_transcript: status.final_transcript,
        },
    ))
}

async fn get_fragment_audio(
    State(audio): State<Arc<K1AccessFullAudio>>,
    Extension(principal): Extension<Principal>,
    Path((access_value, fragment_value)): Path<(String, String)>,
) -> ApiResult<Response> {
    let access_id = parse_access_id(&access_value)?;
    let fragment_id = parse_fragment_id(&fragment_value)?;
    let user = caller(&principal);
    let bytes = run("get fragment audio", move || {
        audio.fragment_audio_for_user(user, access_id, fragment_id)
    })
    .await?;
    Ok((
        [
            (CONTENT_TYPE, HeaderValue::from_static("audio/ogg")),
            (CACHE_CONTROL, HeaderValue::from_static("no-store")),
        ],
        bytes,
    )
        .into_response())
}

async fn submit_labels(
    State(audio): State<Arc<K1AccessFullAudio>>,
    Extension(principal): Extension<Principal>,
    Path((access_value, fragment_value)): Path<(String, String)>,
    headers: HeaderMap,
    body: Bytes,
) -> ApiResult<Response> {
    require_content_type(&headers, "application/json")?;
    let input: LabelsDto = serde_json::from_slice(&body).map_err(|_| {
        problem(
            StatusCode::BAD_REQUEST,
            "invalid_json",
            "request body is invalid",
        )
    })?;
    let labels = parse_labels(input)?;
    let access_id = parse_access_id(&access_value)?;
    let fragment_id = parse_fragment_id(&fragment_value)?;
    let user = caller(&principal);
    run("submit fragment labels", move || {
        audio.submit_labels_for_user(user, access_id, fragment_id, labels)
    })
    .await?;
    Ok(no_content())
}

async fn retry_fragment(
    State(audio): State<Arc<K1AccessFullAudio>>,
    Extension(principal): Extension<Principal>,
    Path((access_value, fragment_value)): Path<(String, String)>,
) -> ApiResult<Response> {
    let access_id = parse_access_id(&access_value)?;
    let fragment_id = parse_fragment_id(&fragment_value)?;
    let user = caller(&principal);
    run("retry fragment", move || {
        audio.retry_fragment_for_user(user, access_id, fragment_id)
    })
    .await?;
    Ok(no_content())
}

async fn discard_fragment(
    State(audio): State<Arc<K1AccessFullAudio>>,
    Extension(principal): Extension<Principal>,
    Path((access_value, fragment_value)): Path<(String, String)>,
) -> ApiResult<Response> {
    let access_id = parse_access_id(&access_value)?;
    let fragment_id = parse_fragment_id(&fragment_value)?;
    let user = caller(&principal);
    run("discard fragment", move || {
        audio.discard_fragment_for_user(user, access_id, fragment_id)
    })
    .await?;
    Ok(no_content())
}

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use axum::body::{Body, to_bytes};
    use axum::extract::OriginalUri;
    use axum::http::Request;
    use axum::routing::MethodRouter;
    use kcode_k1_http::{
        CanonicalUsername, Config, Identity, IdentityError, IdentityProvider, K1Http,
        RegistrationPrincipal,
    };
    use kcode_k1_http_replay::{ReplayConfig, ReplayWindow};
    use kcode_k1_http_testkit::{Candidate, Fixture, LookupBlock};
    use serde_json::json;
    use tower::ServiceExt;

    #[test]
    fn canonical_ids_and_user_conversion_preserve_bytes() {
        let id = "00112233445566778899aabb";
        assert_eq!(hex(parse_fragment_id(id).unwrap().as_bytes()), id);
        assert!(parse_fragment_id("00112233445566778899AABB").is_err());
        assert!(parse_fragment_id("0011").is_err());
        assert_eq!(
            hex(caller_bytes([1; 12]).as_tx_id().as_bytes()),
            "010101010101010101010101"
        );
    }

    fn caller_bytes(bytes: [u8; 12]) -> UserId {
        UserId::from_tx_id(TxId::from_bytes(bytes))
    }

    #[test]
    fn labels_accept_known_and_unknown_people() {
        let labels = parse_labels(LabelsDto {
            labels: vec![
                LabelDto {
                    speaker: "Speaker 1".to_owned(),
                    person_id: None,
                },
                LabelDto {
                    speaker: "Speaker 2".to_owned(),
                    person_id: Some("00112233445566778899aabb".to_owned()),
                },
            ],
        })
        .unwrap();
        assert_eq!(labels.len(), 2);
        assert!(labels[0].person_id.is_none());
        assert_eq!(
            hex(labels[1].person_id.unwrap().as_tx_id().as_bytes()),
            "00112233445566778899aabb"
        );
        assert!(
            parse_labels(LabelsDto {
                labels: vec![LabelDto {
                    speaker: "Unknown".to_owned(),
                    person_id: None,
                }]
            })
            .is_err()
        );
    }

    #[test]
    fn status_shape_and_hidden_errors_are_stable() {
        let value = serde_json::to_value(StatusDto {
            state: "processing",
            fragments: Vec::new(),
            final_transcript: None,
        })
        .unwrap();
        assert_eq!(
            value,
            json!({"state":"processing","fragments":[],"final_transcript":null})
        );
        let error = dependency_error(
            "get audio status",
            "principal cannot access full audio".to_owned(),
        );
        assert_eq!(error.status, StatusCode::NOT_FOUND);
        assert_eq!(error.code, "audio_not_found");
        assert!(!error.message.contains("principal"));
    }

    #[test]
    fn content_types_and_public_route_assembler_are_exact() {
        let mut headers = HeaderMap::new();
        headers.insert(
            CONTENT_TYPE,
            HeaderValue::from_static("application/json; charset=utf-8"),
        );
        assert!(require_content_type(&headers, "application/json").is_ok());
        assert!(require_content_type(&headers, "application/octet-stream").is_err());
        let _ = authenticated_routes as fn(Arc<K1AccessFullAudio>) -> Router<()>;
    }

    #[tokio::test]
    async fn upload_layer_accepts_more_than_axums_default_limit() {
        let size = 2 * 1024 * 1024 + 1;
        let app = Router::new().route(
            "/audio/profiles/{profile_id}",
            post(|Path(_): Path<String>, body: Bytes| async move { body.len().to_string() })
                .layer(upload_body_limit()),
        );
        let response = app
            .oneshot(
                Request::post("/audio/profiles/00112233445566778899aabb")
                    .body(Body::from(vec![7_u8; size]))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
        assert_eq!(body.as_ref(), size.to_string().as_bytes());
    }

    struct Adapter;

    struct Provider {
        identities: Vec<(CanonicalUsername, Identity)>,
        blocked: Option<(CanonicalUsername, LookupBlock)>,
    }

    #[async_trait]
    impl IdentityProvider for Provider {
        async fn lookup(
            &self,
            username: &CanonicalUsername,
        ) -> Result<Option<Identity>, IdentityError> {
            if let Some((blocked_username, block)) = &self.blocked
                && blocked_username == username
            {
                let release = block.release.notified();
                tokio::pin!(release);
                release.as_mut().enable();
                block.entered.notify_one();
                release.await;
            }
            Ok(self
                .identities
                .iter()
                .find(|(candidate, _)| candidate == username)
                .map(|(_, identity)| *identity))
        }
    }

    #[derive(Deserialize)]
    struct Known {
        known: String,
    }

    async fn register(Extension(principal): Extension<RegistrationPrincipal>) -> String {
        principal.username().as_str().to_owned()
    }

    async fn whoami(Extension(principal): Extension<Principal>) -> String {
        principal.username().as_str().to_owned()
    }

    async fn target(OriginalUri(uri): OriginalUri) -> String {
        uri.path_and_query()
            .map(|value| value.as_str())
            .unwrap_or("/")
            .to_owned()
    }

    async fn echo(body: Bytes) -> Bytes {
        body
    }

    async fn known(Json(value): Json<Known>) -> String {
        value.known
    }

    #[async_trait]
    impl Candidate for Adapter {
        async fn open(&self, fixture: Fixture) -> Result<Router, String> {
            let identities = fixture
                .identities
                .into_iter()
                .map(|identity| {
                    CanonicalUsername::parse(&identity.username)
                        .map(|username| {
                            (
                                username,
                                Identity::new(identity.user_id, identity.public_key),
                            )
                        })
                        .map_err(|_| "invalid fixture identity".to_owned())
                })
                .collect::<Result<Vec<_>, _>>()?;
            let blocked = fixture
                .blocked_lookup
                .map(|block| {
                    CanonicalUsername::parse(&block.username)
                        .map(|username| (username, block))
                        .map_err(|_| "invalid blocked identity".to_owned())
                })
                .transpose()?;
            let replay = ReplayWindow::open(ReplayConfig {
                epoch_file: fixture.epoch_file,
                max_nonces_per_epoch: fixture.max_nonces_per_epoch,
            })
            .await
            .map_err(|_| "replay unavailable".to_owned())?;
            let http = K1Http::new(
                Config {
                    server_id: fixture.server_id,
                    public_origin: fixture.public_origin,
                    max_body_bytes: fixture.max_body_bytes,
                },
                replay,
                Arc::new(Provider {
                    identities,
                    blocked,
                }),
            )
            .map_err(|_| "configuration invalid".to_owned())?;
            let authenticated = Router::new()
                .route("/whoami", get(whoami))
                .route("/target", get(target))
                .route("/echo", post(echo).layer(upload_body_limit()))
                .route("/json", post(known));
            Ok(http.router(
                post(register),
                MethodRouter::new().get(|| async { "terms" }),
                authenticated,
            ))
        }
    }

    #[tokio::test]
    async fn k1_http_configured_body_limit_remains_authoritative() {
        kcode_k1_http_testkit::verify_authentication(&Adapter).await;
    }
}