(
State(WebhookState {
adapter_client_rx,
webhook_cache,
dyncfgs,
}): State<WebhookState>,
Path((database, schema, name)): Path<(String, String, String)>,
heade
| 32 | use crate::http::WebhookState; |
| 33 | |
| 34 | pub async fn handle_webhook( |
| 35 | State(WebhookState { |
| 36 | adapter_client_rx, |
| 37 | webhook_cache, |
| 38 | dyncfgs, |
| 39 | }): State<WebhookState>, |
| 40 | Path((database, schema, name)): Path<(String, String, String)>, |
| 41 | headers: http::HeaderMap, |
| 42 | body: Body, |
| 43 | ) -> impl IntoResponse { |
| 44 | let max_request_size = WEBHOOK_MAX_REQUEST_SIZE_BYTES.get(&dyncfgs); |
| 45 | let body = axum::body::to_bytes(body, max_request_size) |
| 46 | .await |
| 47 | .map_err(|err| { |
| 48 | use std::error::Error; |
| 49 | // axum::Error wraps the underlying cause as its source. If that source is a |
| 50 | // LengthLimitError the body exceeded the configured limit (HTTP 413). Any other |
| 51 | // cause (TCP reset, decompression failure, etc.) is an internal read error (HTTP 500) |
| 52 | // and must not be reported as a size-limit violation. |
| 53 | if err |
| 54 | .source() |
| 55 | .is_some_and(|s| s.is::<http_body_util::LengthLimitError>()) |
| 56 | { |
| 57 | WebhookError::BodyTooLarge { |
| 58 | max_bytes: max_request_size, |
| 59 | } |
| 60 | } else { |
| 61 | WebhookError::Internal(anyhow::anyhow!(err)) |
| 62 | } |
| 63 | })?; |
| 64 | let adapter_client = adapter_client_rx.clone().await.expect("sender not dropped"); |
| 65 | // Collect headers into a map, while converting them into strings. |
| 66 | let mut headers_s = BTreeMap::new(); |
| 67 | for (name, val) in headers.iter() { |
| 68 | if let Ok(val_s) = val.to_str().map(|s| s.to_string()) { |
| 69 | // If a header is included more than once, bail returning an error to the user. |
| 70 | let existing = headers_s.insert(name.as_str().to_string(), val_s); |
| 71 | if existing.is_some() { |
| 72 | let msg = format!("{} provided more than once", name.as_str()); |
| 73 | return Err(WebhookError::InvalidHeaders(msg)); |
| 74 | } |
| 75 | } |
| 76 | } |
| 77 | let headers = Arc::new(headers_s); |
| 78 | |
| 79 | // Append to the webhook source, retrying if we race with a concurrent `ALTER SOURCE` op. |
| 80 | Retry::default() |
| 81 | .max_tries(2) |
| 82 | .retry_async(|_| async { |
| 83 | let result = append_webhook( |
| 84 | &adapter_client, |
| 85 | &webhook_cache, |
| 86 | &database, |
| 87 | &schema, |
| 88 | &name, |
| 89 | &body, |
| 90 | &headers, |
| 91 | ) |
nothing calls this directly
no test coverage detected