Rework API to reflect desired user interface (#75)

Reduce the query space to searching by instrument session for runs and
within runs for data.

Stops the API being as tightly coupled to the internal structure of
tiled.

---------

Co-authored-by: Abigail Emery <[email protected]>

Author: tpoliaw

Cargo.lock

@@ -6,7 +6,7 @@ edition = "2024"
 [dependencies]
 async-graphql = { version = "7.0.17", features = ["uuid"]}
 tokio = { version = "1", features = ["full"]}
-reqwest = { version = "0.12.15", features = ["json", "rustls-tls"], default-features = false }
+reqwest = { version = "0.12.15", features = ["json", "rustls-tls", "stream"], default-features = false }
 serde_json = "1.0.143"
 serde = { version = "1.0.219", features = ["derive"] }
 axum = "0.8.4"

Cargo.toml

@@ -1,24 +1,30 @@
+use std::borrow::Cow;
 use std::fmt;
 
 use axum::http::HeaderMap;
 #[cfg(test)]
 use httpmock::MockServer;
 use reqwest::{Client, Url};
 use serde::de::DeserializeOwned;
-use tracing::{info, instrument};
-use uuid::Uuid;
+use tracing::{debug, info, instrument};
 
-use crate::model::{app, array, container, event_stream, run, table};
+use crate::model::{app, node, table};
 
 pub type ClientResult<T> = Result<T, ClientError>;
 
+#[derive(Clone)]
 pub struct TiledClient {
     client: Client,
     address: Url,
 }
 
 impl TiledClient {
     pub fn new(address: Url) -> Self {
+        if address.cannot_be_a_base() {
+            // Panicking is not great but if we've got this far, nothing else is going to work so
+            // bail out early.
+            panic!("Invalid tiled URL");
+        }
         Self {
             client: Client::new(),
             address,
@@ -29,17 +35,16 @@ impl TiledClient {
         &self,
         endpoint: &str,
         headers: Option<HeaderMap>,
-        query_params: Option<&[(&str, &str)]>,
+        query_params: Option<&[(&str, Cow<'_, str>)]>,
     ) -> ClientResult<T> {
-        info!("Requesting from tiled: {}", endpoint);
         let url = self.address.join(endpoint)?;
 
         let mut request = match headers {
             Some(headers) => self.client.get(url).headers(headers),
             None => self.client.get(url),
         };
         if let Some(params) = query_params {
-            request = request.query(params);
+            request = request.query(&params);
         }
         info!("Querying: {request:?}");
 
@@ -50,84 +55,59 @@ impl TiledClient {
     pub async fn app_metadata(&self) -> ClientResult<app::AppMetadata> {
         self.request("/api/v1/", None, None).await
     }
-    pub async fn run_metadata(&self, id: Uuid) -> ClientResult<run::RunMetadataRoot> {
-        self.request(&format!("/api/v1/metadata/{id}"), None, None)
-            .await
-    }
-    pub async fn event_stream_metadata(
+    pub async fn search(
         &self,
-        id: Uuid,
-        stream: String,
-    ) -> ClientResult<event_stream::EventStreamMetadataRoot> {
-        self.request(&format!("/api/v1/metadata/{id}/{stream}"), None, None)
+        path: &str,
+        query: &[(&str, Cow<'_, str>)],
+    ) -> ClientResult<node::Root> {
+        self.request(&format!("api/v1/search/{}", path), None, Some(query))
             .await
     }
-    pub async fn array_metadata(
-        &self,
-        id: Uuid,
-        stream: String,
-        array: String,
-    ) -> ClientResult<array::ArrayMetadataRoot> {
-        self.request(
-            &format!("/api/v1/metadata/{id}/{stream}/{array}"),
-            None,
-            Some(&[("include_data_sources", "true")]),
-        )
-        .await
-    }
-    pub async fn table_metadata(
-        &self,
-        id: Uuid,
-        stream: String,
-        table: String,
-    ) -> ClientResult<table::TableMetadataRoot> {
-        self.request(
-            &format!("/api/v1/metadata/{id}/{stream}/{table}"),
-            None,
-            Some(&[("include_data_sources", "true")]),
-        )
-        .await
-    }
     pub async fn table_full(
         &self,
-        id: Uuid,
-        stream: String,
-        table: String,
+        path: &str,
+        columns: Option<Vec<String>>,
     ) -> ClientResult<table::Table> {
         let mut headers = HeaderMap::new();
         headers.insert("accept", "application/json".parse().unwrap());
+        let query = columns.map(|columns| {
+            columns
+                .into_iter()
+                .map(|col| ("column", col.into()))
+                .collect::<Vec<_>>()
+        });
 
         self.request(
-            &format!("/api/v1/table/full/{id}/{stream}/{table}"),
+            &format!("/api/v1/table/full/{}", path),
             Some(headers),
-            None,
+            query.as_deref(),
         )
         .await
     }
-    pub async fn search_root(&self) -> ClientResult<run::RunRoot> {
-        self.request("/api/v1/search/", None, None).await
-    }
-    pub async fn search_run_container(
-        &self,
-        id: Uuid,
-    ) -> ClientResult<event_stream::EventStreamRoot> {
-        self.request(&format!("/api/v1/search/{id}"), None, None)
-            .await
-    }
-    pub async fn container_full(
-        &self,
-        id: Uuid,
-        stream: Option<String>,
-    ) -> ClientResult<container::Container> {
-        let mut headers = HeaderMap::new();
-        headers.insert("accept", "application/json".parse().unwrap());
 
-        let endpoint = match stream {
-            Some(stream) => &format!("/api/v1/container/full/{id}/{stream}"),
-            None => &format!("/api/v1/container/full/{id}"),
-        };
+    pub(crate) async fn download(
+        &self,
+        run: String,
+        stream: String,
+        det: String,
+        id: u32,
+    ) -> reqwest::Result<reqwest::Response> {
+        let mut url = self
+            .address
+            .join("/api/v1/asset/bytes")
+            .expect("Base address was cannot_be_a_base");
+        url.path_segments_mut()
+            .expect("Base address was cannot_be_a_base")
+            .push(&run)
+            .push(&stream)
+            .push(&det);
 
-        self.request(endpoint, Some(headers), None).await
+        debug!("Downloading id={id} from {url}");
+        self.client
+            .get(url)
+            .query(&[("id", &id.to_string())])
+            .send()
+            .await
     }
 
     /// Create a new client for the given mock server

src/clients.rs

@@ -0,0 +1,83 @@
+use axum::body::Body;
+use axum::extract::{Path, State};
+use axum::http::{HeaderMap, StatusCode};
+use serde_json::{Value, json};
+use tracing::{error, info};
+
+use crate::clients::TiledClient;
+
+const FORWARDED_HEADERS: [&str; 4] = [
+    "content-disposition",
+    "content-type",
+    "content-length",
+    "last-modified",
+];
+
+pub async fn download(
+    State(client): State<TiledClient>,
+    Path((run, stream, det, id)): Path<(String, String, String, u32)>,
+) -> (StatusCode, HeaderMap, Body) {
+    info!("Downloading {run}/{stream}/{det}/{id}");
+    let req = client.download(run, stream, det, id).await;
+    forward_download_response(req).await
+}
+
+async fn forward_download_response(
+    response: Result<reqwest::Response, reqwest::Error>,
+) -> (StatusCode, HeaderMap, Body) {
+    match response {
+        Ok(mut resp) => match resp.status().as_u16() {
+                200..300  => {
+                    let mut headers = HeaderMap::new();
+                    for key in FORWARDED_HEADERS {
+                        if let Some(value) = resp.headers_mut().remove(key) {
+                            headers.insert(key, value);
+                        }
+                    }
+                    let stream = Body::from_stream(resp.bytes_stream());
+                    (StatusCode::OK, headers, stream)
+                },
+                400..500 => (
+                    // Probably permission error or non-existent file - forward error to client
+                    resp.status(),
+                    HeaderMap::new(),
+                    Body::from_stream(resp.bytes_stream())
+                ),
+                100..200 | // ??? check tiled?
+                300..400 | // should have followed a redirect
+                0..100 | (600..) |  // who needs standards anyway
+                500..600 => {
+                    let status = resp.status().as_u16();
+                    let content = resp.text().await.unwrap_or_else(|e| format!("Unable to read error response: {e}"));
+                    (
+                        // Whatever we got back, it wasn't what we expected so blame it on tiled
+                        StatusCode::SERVICE_UNAVAILABLE,
+                        HeaderMap::new(),
+                        json!({
+                            "detail": "Unexpected response from tiled",
+                            "status": status,
+                            // Try to parse response as json before giving up and passing a string
+                            "response": serde_json::from_str(&content)
+                                .unwrap_or(Value::String(content))
+                        }).to_string().into()
+                    )
+                }
+            },
+        Err(err) => {
+            error!("Error sending request to tiled: {err}");
+            let (status, message) = if err.is_connect() {
+                (
+                    StatusCode::SERVICE_UNAVAILABLE,
+                    "Could not connect to tiled",
+                )
+            } else {
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    "Error making request to tiled",
+                )
+            };
+
+            (status, HeaderMap::new(), message.into())
+        }
+    }
+}

src/download.rs

@@ -1,5 +1,3 @@
-use std::error;
-
 use async_graphql::{EmptyMutation, EmptySubscription, Schema};
 use axum::http::StatusCode;
 use axum::response::{Html, IntoResponse};
@@ -9,6 +7,7 @@ use axum::{Extension, Router};
 mod cli;
 mod clients;
 mod config;
+mod download;
 mod handlers;
 mod model;
 #[cfg(test)]
@@ -25,7 +24,7 @@ use crate::handlers::{graphiql_handler, graphql_handler};
 use crate::model::TiledQuery;
 
 #[tokio::main]
-async fn main() -> Result<(), Box<dyn error::Error>> {
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let subscriber = tracing_subscriber::FmtSubscriber::new();
     tracing::subscriber::set_global_default(subscriber)?;
 
@@ -45,14 +44,18 @@ async fn main() -> Result<(), Box<dyn error::Error>> {
     }
 }
 
-async fn serve(config: GlazedConfig) -> Result<(), Box<dyn error::Error>> {
+async fn serve(config: GlazedConfig) -> Result<(), Box<dyn std::error::Error>> {
+    let client = TiledClient::new(config.tiled_client.address);
     let schema = Schema::build(TiledQuery, EmptyMutation, EmptySubscription)
-        .data(TiledClient::new(config.tiled_client.address))
+        .data(config.bind_address)
+        .data(client.clone())
         .finish();
 
     let app = Router::new()
         .route("/graphql", post(graphql_handler).get(graphql_get_warning))
         .route("/graphiql", get(graphiql_handler))
+        .route("/asset/{run}/{stream}/{det}/{id}", get(download::download))
+        .with_state(client)
         .fallback((
             StatusCode::NOT_FOUND,
             Html(include_str!("../static/404.html")),