Implement bulk endpoints for worker

v3xlabs · Dec 29, 2023 · 8e869e5 · 8e869e5
1 parent 1314cbe
commit 8e869e5
Show file tree

Hide file tree

Showing 8 changed files with 161 additions and 16 deletions.
diff --git a/server/src/routes/mod.rs b/server/src/routes/mod.rs
@@ -18,6 +18,8 @@ pub mod name;
 pub mod root;
 pub mod universal;
 
+// TODO (@antony1060): cleanup file
+
 #[derive(Deserialize)]
 pub struct FreshQuery {
     #[serde(default, deserialize_with = "bool_or_false")]

diff --git a/worker/Cargo.lock b/worker/Cargo.lock
diff --git a/worker/Cargo.toml b/worker/Cargo.toml
@@ -39,6 +39,7 @@ thiserror = "1.0.50"
 http = "1.0.0"
 lazy_static = "1.4.0"
 serde_qs = "0.12.0"
+futures-util = "0.3.29"
 
 [build-dependencies]
 chrono = "0.4.31"

diff --git a/worker/src/http_util.rs b/worker/src/http_util.rs
@@ -1,10 +1,14 @@
 use enstate_shared::models::profile::error::ProfileError;
+use enstate_shared::utils::vec::dedup_ord;
 use ethers::prelude::ProviderError;
 use http::status::StatusCode;
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Deserializer, Serialize};
+use thiserror::Error;
 use worker::{Error, Request, Response, Url};
 
+// TODO (@antony1060): cleanup file
+
 #[derive(Deserialize)]
 pub struct FreshQuery {
     #[serde(default, deserialize_with = "bool_or_false")]
@@ -27,6 +31,40 @@ pub fn parse_query<T: DeserializeOwned>(req: &Request) -> worker::Result<T> {
     serde_qs::from_str::<T>(query).map_err(|_| http_simple_status_error(StatusCode::BAD_REQUEST))
 }
 
+#[derive(Error, Debug)]
+pub enum ValidationError {
+    #[error("maximum input length exceeded (expected at most {0})")]
+    MaxLengthExceeded(usize),
+}
+
+impl From<ValidationError> for worker::Error {
+    fn from(value: ValidationError) -> Self {
+        ErrorResponse {
+            status: StatusCode::BAD_REQUEST.as_u16(),
+            error: value.to_string(),
+        }
+        .into()
+    }
+}
+
+pub fn validate_bulk_input(
+    input: &[String],
+    max_len: usize,
+) -> Result<Vec<String>, ValidationError> {
+    let unique = dedup_ord(
+        &input
+            .iter()
+            .map(|entry| entry.to_lowercase())
+            .collect::<Vec<_>>(),
+    );
+
+    if unique.len() > max_len {
+        return Err(ValidationError::MaxLengthExceeded(max_len));
+    }
+
+    Ok(unique)
+}
+
 #[derive(Serialize)]
 pub struct ErrorResponse {
     pub(crate) status: u16,

diff --git a/worker/src/lib.rs b/worker/src/lib.rs
@@ -9,7 +9,7 @@ use enstate_shared::utils::factory::SimpleFactory;
 use ethers::prelude::{Http, Provider};
 use http::StatusCode;
 use lazy_static::lazy_static;
-use worker::{event, Context, Cors, Env, Method, Request, Response, Router};
+use worker::{event, Context, Cors, Env, Headers, Method, Request, Response, Router};
 
 use crate::http_util::http_simple_status_error;
 use crate::kv_cache::CloudflareKVCache;
@@ -62,22 +62,32 @@ async fn main(req: Request, env: Env, _ctx: Context) -> worker::Result<Response>
         .get_async("/u/:name_or_address", routes::universal::get)
         .get_async("/i/:name_or_address", routes::image::get)
         .get_async("/h/:name_or_address", routes::header::get)
-        // .get_async("/bulk/a", main_handler)
-        // .get_async("/bulk/n", main_handler)
-        // .get_async("/bulk/u", main_handler)
-        // .or_else_any_method("*", |_, _| {
-        //     Err(ErrorResponse {
-        //         status: StatusCode::NOT_FOUND.as_u16(),
-        //         error: "Unknown route".to_string(),
-        //     }
-        //     .into())
-        // })
+        .get_async("/bulk/a", routes::address::get_bulk)
+        .get_async("/bulk/n", routes::name::get_bulk)
+        .get_async("/bulk/u", routes::universal::get_bulk)
         .run(req, env)
         .await
         .and_then(|response| response.with_cors(&CORS));
 
-    if let Err(worker::Error::Json(err)) = response {
-        return Response::error(err.0, err.1);
+    if let Err(err) = response {
+        if let worker::Error::Json(json) = err {
+            return Response::error(json.0, json.1).and_then(|response| {
+                response
+                    .with_headers(Headers::from_iter(
+                        [("Content-Type", "application/json")].iter(),
+                    ))
+                    .with_cors(&CORS)
+            });
+        }
+
+        return Response::error(err.to_string(), StatusCode::INTERNAL_SERVER_ERROR.as_u16())
+            .and_then(|response| {
+                response
+                    .with_headers(Headers::from_iter(
+                        [("Content-Type", "application/json")].iter(),
+                    ))
+                    .with_cors(&CORS)
+            });
     }
 
     response

diff --git a/worker/src/routes/address.rs b/worker/src/routes/address.rs
@@ -1,10 +1,13 @@
 use enstate_shared::models::profile::ProfileService;
 use ethers::addressbook::Address;
+use futures_util::future::try_join_all;
 use http::StatusCode;
+use serde::Deserialize;
 use worker::{Request, Response, RouteContext};
 
 use crate::http_util::{
-    http_simple_status_error, parse_query, profile_http_error_mapper, FreshQuery,
+    http_simple_status_error, parse_query, profile_http_error_mapper, validate_bulk_input,
+    FreshQuery,
 };
 
 pub async fn get(req: Request, ctx: RouteContext<ProfileService>) -> worker::Result<Response> {
@@ -26,3 +29,34 @@ pub async fn get(req: Request, ctx: RouteContext<ProfileService>) -> worker::Res
 
     Response::from_json(&profile)
 }
+
+#[derive(Deserialize)]
+pub struct AddressGetBulkQuery {
+    addresses: Vec<String>,
+
+    #[serde(flatten)]
+    fresh: FreshQuery,
+}
+
+pub async fn get_bulk(req: Request, ctx: RouteContext<ProfileService>) -> worker::Result<Response> {
+    let query: AddressGetBulkQuery = parse_query(&req)?;
+
+    let addresses = validate_bulk_input(&query.addresses, 10)?;
+
+    let addresses = addresses
+        .iter()
+        .map(|address| address.parse::<Address>())
+        .collect::<Result<Vec<_>, _>>()
+        .map_err(|_| http_simple_status_error(StatusCode::BAD_REQUEST))?;
+
+    let profiles = addresses
+        .iter()
+        .map(|address| ctx.data.resolve_from_address(*address, query.fresh.fresh))
+        .collect::<Vec<_>>();
+
+    let joined = try_join_all(profiles)
+        .await
+        .map_err(profile_http_error_mapper)?;
+
+    Response::from_json(&joined)
+}
diff --git a/worker/src/routes/name.rs b/worker/src/routes/name.rs
@@ -1,9 +1,12 @@
 use enstate_shared::models::profile::ProfileService;
+use futures_util::future::try_join_all;
 use http::StatusCode;
+use serde::Deserialize;
 use worker::{Request, Response, RouteContext};
 
 use crate::http_util::{
-    http_simple_status_error, parse_query, profile_http_error_mapper, FreshQuery,
+    http_simple_status_error, parse_query, profile_http_error_mapper, validate_bulk_input,
+    FreshQuery,
 };
 
 pub async fn get(req: Request, ctx: RouteContext<ProfileService>) -> worker::Result<Response> {
@@ -21,3 +24,28 @@ pub async fn get(req: Request, ctx: RouteContext<ProfileService>) -> worker::Res
 
     Response::from_json(&profile)
 }
+
+#[derive(Deserialize)]
+pub struct NameGetBulkQuery {
+    names: Vec<String>,
+
+    #[serde(flatten)]
+    fresh: FreshQuery,
+}
+
+pub async fn get_bulk(req: Request, ctx: RouteContext<ProfileService>) -> worker::Result<Response> {
+    let query: NameGetBulkQuery = parse_query(&req)?;
+
+    let names = validate_bulk_input(&query.names, 10)?;
+
+    let profiles = names
+        .iter()
+        .map(|name| ctx.data.resolve_from_name(name, query.fresh.fresh))
+        .collect::<Vec<_>>();
+
+    let joined = try_join_all(profiles)
+        .await
+        .map_err(profile_http_error_mapper)?;
+
+    Response::from_json(&joined)
+}
diff --git a/worker/src/routes/universal.rs b/worker/src/routes/universal.rs
@@ -1,9 +1,12 @@
 use enstate_shared::models::profile::ProfileService;
+use futures_util::future::try_join_all;
 use http::StatusCode;
+use serde::Deserialize;
 use worker::{Request, Response, RouteContext};
 
 use crate::http_util::{
-    http_simple_status_error, parse_query, profile_http_error_mapper, FreshQuery,
+    http_simple_status_error, parse_query, profile_http_error_mapper, validate_bulk_input,
+    FreshQuery,
 };
 
 pub async fn get(req: Request, ctx: RouteContext<ProfileService>) -> worker::Result<Response> {
@@ -21,3 +24,31 @@ pub async fn get(req: Request, ctx: RouteContext<ProfileService>) -> worker::Res
 
     Response::from_json(&profile)
 }
+
+#[derive(Deserialize)]
+pub struct UniversalGetBulkQuery {
+    queries: Vec<String>,
+
+    #[serde(flatten)]
+    fresh: FreshQuery,
+}
+
+pub async fn get_bulk(req: Request, ctx: RouteContext<ProfileService>) -> worker::Result<Response> {
+    let query: UniversalGetBulkQuery = parse_query(&req)?;
+
+    let queries = validate_bulk_input(&query.queries, 10)?;
+
+    let profiles = queries
+        .iter()
+        .map(|input| {
+            ctx.data
+                .resolve_from_name_or_address(input, query.fresh.fresh)
+        })
+        .collect::<Vec<_>>();
+
+    let joined = try_join_all(profiles)
+        .await
+        .map_err(profile_http_error_mapper)?;
+
+    Response::from_json(&joined)
+}