risingwave_frontend/webhook/
mod.rs1use std::net::SocketAddr;
16use std::sync::Arc;
17use std::sync::atomic::AtomicU32;
18
19use anyhow::{Context, anyhow};
20use axum::Router;
21use axum::body::Bytes;
22use axum::extract::{Extension, Path};
23use axum::http::{HeaderMap, Method, StatusCode};
24use axum::routing::post;
25use risingwave_common::array::{Array, ArrayBuilder, DataChunk};
26use risingwave_common::secret::LocalSecretManager;
27use risingwave_common::types::{DataType, JsonbVal, Scalar};
28use risingwave_pb::catalog::WebhookSourceInfo;
29use risingwave_pb::task_service::{FastInsertRequest, FastInsertResponse};
30use tokio::net::TcpListener;
31use tower::ServiceBuilder;
32use tower_http::add_extension::AddExtensionLayer;
33use tower_http::compression::CompressionLayer;
34use tower_http::cors::{self, CorsLayer};
35
36use crate::webhook::utils::{Result, err};
37mod utils;
38use risingwave_rpc_client::ComputeClient;
39
40pub type Service = Arc<WebhookService>;
41
42const USER: &str = "root";
44
45#[derive(Clone)]
46pub struct FastInsertContext {
47 pub webhook_source_info: WebhookSourceInfo,
48 pub fast_insert_request: FastInsertRequest,
49 pub compute_client: ComputeClient,
50}
51
52pub struct WebhookService {
53 webhook_addr: SocketAddr,
54 counter: AtomicU32,
55}
56
57pub(super) mod handlers {
58 use jsonbb::Value;
59 use risingwave_common::array::JsonbArrayBuilder;
60 use risingwave_common::session_config::SearchPath;
61 use risingwave_pb::catalog::WebhookSourceInfo;
62 use risingwave_pb::task_service::fast_insert_response;
63 use utils::{header_map_to_json, verify_signature};
64
65 use super::*;
66 use crate::catalog::root_catalog::SchemaPath;
67 use crate::scheduler::choose_fast_insert_client;
68 use crate::session::SESSION_MANAGER;
69
70 pub async fn handle_post_request(
71 Extension(srv): Extension<Service>,
72 headers: HeaderMap,
73 Path((database, schema, table)): Path<(String, String, String)>,
74 body: Bytes,
75 ) -> Result<()> {
76 let request_id = srv
77 .counter
78 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
79 let FastInsertContext {
80 webhook_source_info,
81 mut fast_insert_request,
82 compute_client,
83 } = acquire_table_info(request_id, &database, &schema, &table).await?;
84
85 let WebhookSourceInfo {
86 signature_expr,
87 secret_ref,
88 wait_for_persistence: _,
89 is_batched,
90 } = webhook_source_info;
91
92 let secret_string = if let Some(secret_ref) = secret_ref {
93 LocalSecretManager::global()
94 .fill_secret(secret_ref)
95 .map_err(|e| err(e, StatusCode::NOT_FOUND))?
96 } else {
97 String::new()
98 };
99
100 let headers_jsonb = header_map_to_json(&headers);
102
103 let is_valid = verify_signature(
105 headers_jsonb,
106 secret_string.as_str(),
107 body.as_ref(),
108 signature_expr.unwrap(),
109 )
110 .await?;
111
112 if !is_valid {
113 return Err(err(
114 anyhow!("Signature verification failed"),
115 StatusCode::UNAUTHORIZED,
116 ));
117 }
118
119 let data_chunk = generate_data_chunk(is_batched, &body)?;
120
121 fast_insert_request.data_chunk = Some(data_chunk.to_protobuf());
123 let res = execute(fast_insert_request, compute_client).await?;
125
126 if res.status == fast_insert_response::Status::Succeeded as i32 {
127 Ok(())
128 } else {
129 Err(err(
130 anyhow!("Failed to fast insert: {}", res.error_message),
131 StatusCode::INTERNAL_SERVER_ERROR,
132 ))
133 }
134 }
135
136 fn generate_data_chunk(is_batched: bool, body: &Bytes) -> Result<DataChunk> {
137 let mut builder = JsonbArrayBuilder::with_type(1, DataType::Jsonb);
138
139 if !is_batched {
140 let json_value = Value::from_text(body).map_err(|e| {
142 err(
143 anyhow!(e).context("Failed to parse body"),
144 StatusCode::UNPROCESSABLE_ENTITY,
145 )
146 })?;
147
148 let jsonb_val = JsonbVal::from(json_value);
149 builder.append(Some(jsonb_val.as_scalar_ref()));
150
151 Ok(DataChunk::new(vec![builder.finish().into_ref()], 1))
152 } else {
153 let rows: Vec<_> = body.split(|&b| b == b'\n').collect();
154
155 for row in &rows {
156 let json_value = Value::from_text(row).map_err(|e| {
157 err(
158 anyhow!(e).context("Failed to parse body"),
159 StatusCode::UNPROCESSABLE_ENTITY,
160 )
161 })?;
162 let jsonb_val = JsonbVal::from(json_value);
163
164 builder.append(Some(jsonb_val.as_scalar_ref()));
165 }
166
167 Ok(DataChunk::new(
168 vec![builder.finish().into_ref()],
169 rows.len(),
170 ))
171 }
172 }
173
174 async fn acquire_table_info(
175 request_id: u32,
176 database: &String,
177 schema: &String,
178 table: &String,
179 ) -> Result<FastInsertContext> {
180 let session_mgr = SESSION_MANAGER
181 .get()
182 .expect("session manager has been initialized");
183
184 let frontend_env = session_mgr.env();
185
186 let search_path = SearchPath::default();
187 let schema_path = SchemaPath::new(Some(schema.as_str()), &search_path, USER);
188
189 let (webhook_source_info, table_id, version_id, row_id_index) = {
190 let reader = frontend_env.catalog_reader().read_guard();
191 let (table_catalog, _schema) = reader
192 .get_any_table_by_name(database.as_str(), schema_path, table)
193 .map_err(|e| err(e, StatusCode::NOT_FOUND))?;
194
195 let webhook_source_info = table_catalog
196 .webhook_info
197 .as_ref()
198 .ok_or_else(|| {
199 err(
200 anyhow!("Table `{}` is not with webhook source", table),
201 StatusCode::FORBIDDEN,
202 )
203 })?
204 .clone();
205 (
206 webhook_source_info,
207 table_catalog.id(),
208 table_catalog.version_id().expect("table must be versioned"),
209 table_catalog.row_id_index.map(|idx| idx as u32),
210 )
211 };
212
213 let fast_insert_request = FastInsertRequest {
214 table_id: table_id.table_id,
215 table_version_id: version_id,
216 column_indices: vec![0],
217 data_chunk: None,
219 row_id_index,
220 request_id,
221 wait_for_persistence: webhook_source_info.wait_for_persistence,
222 };
223
224 let compute_client = choose_fast_insert_client(&table_id, frontend_env, request_id)
225 .await
226 .unwrap();
227
228 Ok(FastInsertContext {
229 webhook_source_info,
230 fast_insert_request,
231 compute_client,
232 })
233 }
234
235 async fn execute(
236 request: FastInsertRequest,
237 client: ComputeClient,
238 ) -> Result<FastInsertResponse> {
239 let response = client.fast_insert(request).await.map_err(|e| {
240 err(
241 anyhow!(e).context("Failed to execute on compute node"),
242 StatusCode::INTERNAL_SERVER_ERROR,
243 )
244 })?;
245 Ok(response)
246 }
247}
248
249impl WebhookService {
250 pub fn new(webhook_addr: SocketAddr) -> Self {
251 Self {
252 webhook_addr,
253 counter: AtomicU32::new(0),
254 }
255 }
256
257 pub async fn serve(self) -> anyhow::Result<()> {
258 use handlers::*;
259 let srv = Arc::new(self);
260
261 let cors_layer = CorsLayer::new()
262 .allow_origin(cors::Any)
263 .allow_methods(vec![Method::POST]);
264
265 let api_router: Router = Router::new()
266 .route("/:database/:schema/:table", post(handle_post_request))
267 .layer(
268 ServiceBuilder::new()
269 .layer(AddExtensionLayer::new(srv.clone()))
270 .into_inner(),
271 )
272 .layer(cors_layer);
273
274 let app: Router = Router::new()
275 .nest("/webhook", api_router)
276 .layer(CompressionLayer::new());
277
278 let listener = TcpListener::bind(&srv.webhook_addr)
279 .await
280 .context("Failed to bind dashboard address")?;
281
282 #[cfg(not(madsim))]
283 axum::serve(listener, app)
284 .await
285 .context("Failed to serve dashboard service")?;
286
287 Ok(())
288 }
289}
290
291#[cfg(test)]
292mod tests {
293 use std::net::SocketAddr;
294
295 #[tokio::test]
296 #[ignore]
297 async fn test_webhook_server() -> anyhow::Result<()> {
298 let addr = SocketAddr::from(([127, 0, 0, 1], 4560));
299 let service = crate::webhook::WebhookService::new(addr);
300 service.serve().await?;
301 Ok(())
302 }
303}