use crate::server::http_client::get_http_client;
use deno_core::{OpDecl, OpState, op2};
use deno_error::JsErrorBox;
use serde::Deserialize;
use std::cell::RefCell;
use std::rc::Rc;
use std::sync::Arc;
use tokio::sync::mpsc;
use tracing::error;
#[derive(Deserialize, Debug)]
#[serde(tag = "type")]
enum RscStreamOperation {
#[serde(rename = "module")]
ModuleReference {
row_id: String,
module_id: String,
chunks: Vec<String>,
name: String,
#[serde(default)]
async_module: bool,
},
#[serde(rename = "element")]
ReactElement { row_id: String, element: serde_json::Value },
#[serde(rename = "symbol")]
Symbol { row_id: String, symbol_ref: String },
#[serde(rename = "error")]
Error {
row_id: String,
message: String,
#[serde(default)]
stack: Option<String>,
#[serde(default)]
phase: Option<String>,
#[serde(default)]
digest: Option<String>,
},
#[serde(rename = "complete")]
Complete {
#[serde(default)]
#[allow(unused)]
final_row_id: Option<String>,
},
}
#[derive(Default)]
pub struct StreamOpState {
pub chunk_sender: Option<mpsc::Sender<Result<Vec<u8>, String>>>,
pub current_stream_id: Option<String>,
pub row_counter: u32,
}
impl StreamOpState {
pub fn get_next_row_id(&mut self) -> String {
let id = format!("{:x}", self.row_counter);
self.row_counter += 1;
id
}
}
fn parse_hex_row_id(row_id: &str, context: &str) -> Result<u32, JsErrorBox> {
if !row_id.chars().all(|c| c.is_ascii_hexdigit()) {
error!("op_send_chunk_to_rust: invalid row_id '{}' for {}", row_id, context);
return Err(JsErrorBox::generic(format!("Invalid row_id: {}", row_id)));
}
u32::from_str_radix(row_id, 16).map_err(|e| {
error!("op_send_chunk_to_rust: invalid row_id '{}' for {}: {}", row_id, context, e);
JsErrorBox::generic(format!("Invalid row_id: {}", row_id))
})
}
#[allow(clippy::disallowed_methods)]
#[op2]
pub async fn op_send_chunk_to_rust(
state: Rc<RefCell<OpState>>,
#[string] operation_json: String,
) -> Result<(), JsErrorBox> {
let operation: RscStreamOperation = match serde_json::from_str(&operation_json) {
Ok(op) => op,
Err(e) => {
let err_msg = format!(
"Invalid JSON for RSC operation: {e}. JSON length: {}",
operation_json.len()
);
error!("{err_msg}");
return Err(JsErrorBox::generic(err_msg));
}
};
let sender_option = {
let mut op_state_ref = state.borrow_mut();
let stream_op_state = match op_state_ref.try_borrow_mut::<StreamOpState>() {
Some(sos) => sos,
None => return Err(JsErrorBox::generic("StreamOpState not found.")),
};
match &operation {
RscStreamOperation::Complete { .. } | RscStreamOperation::Error { .. } => {
stream_op_state.chunk_sender.take()
}
_ => stream_op_state.chunk_sender.as_ref().cloned(),
}
};
match (sender_option, operation) {
(
Some(sender),
RscStreamOperation::ModuleReference { row_id, module_id, chunks, name, async_module },
) => {
let module_data = serde_json::json!({
"id": module_id,
"chunks": chunks,
"name": name,
"async": async_module
});
let row_id_num = parse_hex_row_id(&row_id, "module reference")?;
let rsc_row = format!("{:x}:M{}", row_id_num, module_data);
if sender.send(Ok(rsc_row.into_bytes())).await.is_err() {
error!("op_send_chunk_to_rust: receiver dropped for module reference.");
}
}
(Some(sender), RscStreamOperation::ReactElement { row_id, element }) => {
let row_id_num = parse_hex_row_id(&row_id, "React element")?;
let rsc_row = format!("{:x}:J{}", row_id_num, element);
if sender.send(Ok(rsc_row.into_bytes())).await.is_err() {
error!("op_send_chunk_to_rust: receiver dropped for React element.");
}
}
(Some(sender), RscStreamOperation::Symbol { row_id, symbol_ref }) => {
let row_id_num = parse_hex_row_id(&row_id, "symbol reference")?;
let rsc_row = format!("{:x}:S\"{}\"", row_id_num, symbol_ref);
if sender.send(Ok(rsc_row.into_bytes())).await.is_err() {
error!("op_send_chunk_to_rust: receiver dropped for symbol reference.");
}
}
(Some(sender), RscStreamOperation::Error { row_id, message, stack, phase, digest }) => {
error!("Streaming error in row {row_id}: {message}");
if let Some(stack_trace) = &stack {
error!("Stack trace: {stack_trace}");
}
let error_data = serde_json::json!({
"message": message,
"stack": stack,
"phase": phase,
"digest": digest
});
let row_id_num = parse_hex_row_id(&row_id, "error message")?;
let rsc_row = format!("{:x}:E{}", row_id_num, error_data);
if sender.send(Ok(rsc_row.into_bytes())).await.is_err() {
error!("op_send_chunk_to_rust: receiver dropped for error message.");
}
}
(Some(_sender), RscStreamOperation::Complete { final_row_id: _ }) => {}
(None, operation) => {
error!("No sender available for operation: {operation:?}");
return Err(JsErrorBox::generic("No chunk sender available"));
}
}
Ok(())
}
#[allow(clippy::disallowed_methods)]
pub fn create_module_operation(
row_id: &str,
module_id: &str,
chunks: &[&str],
name: &str,
) -> String {
serde_json::json!({
"type": "module",
"row_id": row_id,
"module_id": module_id,
"chunks": chunks,
"name": name,
"async_module": false
})
.to_string()
}
#[allow(clippy::disallowed_methods)]
pub fn create_element_operation(row_id: &str, element: &serde_json::Value) -> String {
serde_json::json!({
"type": "element",
"row_id": row_id,
"element": element
})
.to_string()
}
#[allow(clippy::disallowed_methods)]
pub fn create_symbol_operation(row_id: &str, symbol_ref: &str) -> String {
serde_json::json!({
"type": "symbol",
"row_id": row_id,
"symbol_ref": symbol_ref
})
.to_string()
}
#[allow(clippy::disallowed_methods)]
pub fn create_error_operation(
row_id: &str,
message: &str,
stack: Option<&str>,
phase: Option<&str>,
digest: Option<&str>,
) -> String {
serde_json::json!({
"type": "error",
"row_id": row_id,
"message": message,
"stack": stack,
"phase": phase,
"digest": digest
})
.to_string()
}
pub fn get_streaming_ops() -> Vec<OpDecl> {
vec![
op_send_chunk_to_rust(),
op_internal_log(),
op_sanitize_html(),
op_get_cookies(),
op_set_cookie(),
op_delete_cookie(),
]
}
#[op2(fast)]
pub fn op_internal_log(#[string] message: &str) {
error!("[rari] {message}");
}
#[op2]
#[string]
pub fn op_sanitize_html(#[string] html: &str, #[string] _component_id: &str) -> String {
crate::rsc::rendering::sanitizer::sanitize_component_output(html)
}
fn http_status_text(status: u16) -> &'static str {
match status {
200 => "OK",
201 => "Created",
204 => "No Content",
301 => "Moved Permanently",
302 => "Found",
304 => "Not Modified",
400 => "Bad Request",
401 => "Unauthorized",
403 => "Forbidden",
404 => "Not Found",
500 => "Internal Server Error",
502 => "Bad Gateway",
503 => "Service Unavailable",
_ => "Unknown",
}
}
fn headers_to_json(headers: &axum::http::HeaderMap) -> serde_json::Map<String, serde_json::Value> {
let mut headers_obj = serde_json::Map::new();
for (name, value) in headers.iter() {
if let Ok(value_str) = value.to_str() {
let key = name.as_str().to_string();
if let Some(existing) = headers_obj.get_mut(&key) {
if let Some(s) = existing.as_str() {
*existing = serde_json::Value::String(format!("{}, {}", s, value_str));
}
} else {
headers_obj.insert(key, serde_json::Value::String(value_str.to_string()));
}
}
}
headers_obj
}
#[allow(clippy::disallowed_methods)]
#[op2]
#[serde]
pub async fn op_fetch_with_cache(
state: Rc<RefCell<OpState>>,
#[string] url: String,
#[string] options_json: String,
) -> Result<serde_json::Value, JsErrorBox> {
let options: rustc_hash::FxHashMap<String, String> = serde_json::from_str(&options_json)
.map_err(|e| JsErrorBox::generic(format!("Invalid options JSON: {}", e)))?;
let request_context = {
let op_state_ref = state.borrow();
op_state_ref
.try_borrow::<Arc<crate::server::middleware::request_context::RequestContext>>()
.cloned()
};
if let Some(ctx) = request_context {
match ctx.fetch_with_cache(&url, options).await {
Ok(result) => {
let body_str = String::from_utf8_lossy(&result.body).to_string();
let headers_obj = headers_to_json(&result.headers);
Ok(serde_json::json!({
"ok": true,
"status": result.status,
"statusText": http_status_text(result.status),
"body": body_str,
"headers": headers_obj,
"cached": result.was_cached,
"tags": result.tags
}))
}
Err(e) => {
error!("Fetch failed for {}: {}", url, e);
Ok(serde_json::json!({
"ok": false,
"status": 500,
"statusText": "Internal Server Error",
"error": e.to_string(),
"cached": false,
"tags": Vec::<String>::new()
}))
}
}
} else {
match perform_simple_fetch(&url, &options).await {
Ok((status, body, headers)) => Ok(serde_json::json!({
"ok": (200..300).contains(&status),
"status": status,
"statusText": http_status_text(status),
"body": body,
"headers": headers,
"cached": false,
"tags": Vec::<String>::new()
})),
Err(e) => {
error!("Fetch failed for {}: {}", url, e);
Ok(serde_json::json!({
"ok": false,
"status": 500,
"statusText": "Internal Server Error",
"error": e,
"cached": false,
"tags": Vec::<String>::new()
}))
}
}
}
}
async fn perform_simple_fetch(
url: &str,
options: &rustc_hash::FxHashMap<String, String>,
) -> Result<(u16, String, serde_json::Map<String, serde_json::Value>), String> {
let client = get_http_client()?;
let mut request = client.get(url);
if let Some(headers_str) = options.get("headers")
&& let Ok(pairs) = serde_json::from_str::<Vec<(String, String)>>(headers_str)
{
for (key, value) in pairs {
request = request.header(key.as_str(), value.as_str());
}
}
let timeout = options.get("timeout").and_then(|t| t.parse::<u64>().ok()).unwrap_or(5000);
request = request.timeout(std::time::Duration::from_millis(timeout));
let response = request.send().await.map_err(|e| format!("Request failed: {}", e))?;
let status = response.status().as_u16();
let headers = response.headers().clone();
let headers_obj = headers_to_json(&headers);
let body = response.text().await.map_err(|e| format!("Failed to read response: {}", e))?;
Ok((status, body, headers_obj))
}
#[op2]
#[string]
pub fn op_get_cookies(state: Rc<RefCell<OpState>>) -> String {
let op_state_ref = state.borrow();
let Some(ctx) = op_state_ref
.try_borrow::<Arc<crate::server::middleware::request_context::RequestContext>>()
else {
return String::new();
};
let mut cookies: std::collections::BTreeMap<String, String> = ctx
.cookie_header
.as_deref()
.unwrap_or("")
.split(';')
.filter_map(|pair| {
let mut parts = pair.splitn(2, '=');
let name = parts.next()?.trim().to_string();
let value = parts.next().unwrap_or("").trim().to_string();
if name.is_empty() { None } else { Some((name, value)) }
})
.collect();
let mut pending: Vec<_> =
ctx.pending_cookies.iter().map(|e| (e.key().clone(), e.value().clone())).collect();
pending.sort_by(|(a_key, a_cookie), (b_key, b_cookie)| {
let a_is_delete = a_cookie.max_age == Some(0);
let b_is_delete = b_cookie.max_age == Some(0);
match (a_is_delete, b_is_delete) {
(true, false) => std::cmp::Ordering::Less,
(false, true) => std::cmp::Ordering::Greater,
_ => {
let a_path_len = a_key.path.as_deref().unwrap_or("").len();
let b_path_len = b_key.path.as_deref().unwrap_or("").len();
a_path_len
.cmp(&b_path_len)
.then_with(|| a_key.name.cmp(&b_key.name))
.then_with(|| a_key.domain.cmp(&b_key.domain))
}
}
});
for (key, cookie) in &pending {
if cookie.max_age == Some(0) {
cookies.remove(&key.name);
} else {
cookies.insert(key.name.clone(), cookie.value.clone());
}
}
cookies.iter().map(|(k, v)| format!("{}={}", k, v)).collect::<Vec<_>>().join("; ")
}
#[derive(serde::Deserialize)]
pub struct SetCookieArgs {
name: String,
value: String,
path: Option<String>,
domain: Option<String>,
expires: Option<String>,
#[serde(rename = "maxAge")]
max_age: Option<i64>,
#[serde(rename = "httpOnly", default)]
http_only: bool,
#[serde(default)]
secure: bool,
#[serde(rename = "sameSite")]
same_site: Option<String>,
priority: Option<String>,
#[serde(default)]
partitioned: bool,
}
#[op2]
#[serde]
pub fn op_set_cookie(
state: Rc<RefCell<OpState>>,
#[serde] args: SetCookieArgs,
) -> Result<(), JsErrorBox> {
use crate::server::middleware::request_context::{PendingCookie, PendingCookieKey};
if !crate::server::actions::is_valid_cookie_name(&args.name) {
return Err(JsErrorBox::type_error(format!("Invalid cookie name: '{}'", args.name)));
}
if !crate::server::actions::is_valid_cookie_value(&args.value) {
return Err(JsErrorBox::type_error(format!(
"Invalid cookie value for '{}': contains invalid characters",
args.name
)));
}
if let Some(ref path) = args.path
&& !crate::server::actions::is_valid_attr_value(path)
{
return Err(JsErrorBox::type_error(format!(
"Invalid cookie path for '{}': '{}'",
args.name, path
)));
}
if let Some(ref domain) = args.domain
&& !crate::server::actions::is_valid_attr_value(domain)
{
return Err(JsErrorBox::type_error(format!(
"Invalid cookie domain for '{}': '{}'",
args.name, domain
)));
}
let op_state_ref = state.borrow();
if let Some(ctx) =
op_state_ref.try_borrow::<Arc<crate::server::middleware::request_context::RequestContext>>()
{
let path = args.path.or_else(|| Some("/".to_string()));
ctx.pending_cookies.insert(
PendingCookieKey::new(&args.name, path.as_deref(), args.domain.as_deref()),
PendingCookie {
name: args.name,
value: args.value,
path,
domain: args.domain,
expires: args.expires,
max_age: args.max_age,
http_only: args.http_only,
secure: args.secure,
same_site: args.same_site,
priority: args.priority,
partitioned: args.partitioned,
},
);
}
Ok(())
}
#[op2(fast)]
pub fn op_delete_cookie(state: Rc<RefCell<OpState>>, #[string] name: String) {
use crate::server::middleware::request_context::{PendingCookie, PendingCookieKey};
let op_state_ref = state.borrow();
if let Some(ctx) =
op_state_ref.try_borrow::<Arc<crate::server::middleware::request_context::RequestContext>>()
{
let cookies_to_delete: Vec<(Option<String>, Option<String>)> = ctx
.pending_cookies
.iter()
.filter(|entry| entry.key().name == name)
.map(|entry| (entry.key().path.clone(), entry.key().domain.clone()))
.collect();
ctx.pending_cookies.retain(|k, _| k.name != name);
if cookies_to_delete.is_empty() {
ctx.pending_cookies.insert(
PendingCookieKey::new(&name, Some("/"), None),
PendingCookie {
name,
value: String::new(),
path: Some("/".to_string()),
domain: None,
expires: None,
max_age: Some(0),
http_only: false,
secure: false,
same_site: None,
priority: None,
partitioned: false,
},
);
} else {
for (path, domain) in cookies_to_delete {
let deletion_path = path.clone().unwrap_or_else(|| "/".to_string());
ctx.pending_cookies.insert(
PendingCookieKey::new(&name, Some(&deletion_path), domain.as_deref()),
PendingCookie {
name: name.clone(),
value: String::new(),
path: Some(deletion_path),
domain,
expires: None,
max_age: Some(0),
http_only: false,
secure: false,
same_site: None,
priority: None,
partitioned: false,
},
);
}
}
}
}
#[cfg(test)]
#[allow(clippy::disallowed_methods)]
mod tests {
use super::*;
use tokio::sync::mpsc;
#[test]
fn test_stream_op_state_operations() {
let mut stream_state = StreamOpState::default();
let row_id_1 = stream_state.get_next_row_id();
let row_id_2 = stream_state.get_next_row_id();
assert_eq!(row_id_1, "0");
assert_eq!(row_id_2, "1");
assert_eq!(stream_state.row_counter, 2);
let (sender, _receiver) = mpsc::channel::<Result<Vec<u8>, String>>(32);
stream_state.chunk_sender = Some(sender);
assert!(stream_state.chunk_sender.is_some());
}
#[test]
fn test_operation_creation() {
let module_op = create_module_operation("0", "Button", &["chunk1", "chunk2"], "default");
assert!(module_op.contains("\"type\":\"module\""));
assert!(module_op.contains("\"row_id\":\"0\""));
assert!(module_op.contains("\"module_id\":\"Button\""));
let element_op = create_element_operation("1", &serde_json::json!({"type": "div"}));
assert!(element_op.contains("\"type\":\"element\""));
assert!(element_op.contains("\"row_id\":\"1\""));
let symbol_op = create_symbol_operation("2", "Symbol.for('react.transitional.element')");
assert!(symbol_op.contains("\"type\":\"symbol\""));
assert!(symbol_op.contains("\"row_id\":\"2\""));
let error_op = create_error_operation("3", "Test error", Some("stack trace"), None, None);
assert!(error_op.contains("\"type\":\"error\""));
assert!(error_op.contains("\"message\":\"Test error\""));
assert!(error_op.contains("\"stack\":\"stack trace\""));
}
}
#[op2]
#[serde]
pub fn op_cache_get(
state: Rc<RefCell<OpState>>,
#[string] cache_key: String,
) -> Option<serde_json::Value> {
let op_state_ref = state.borrow();
if let Some(ctx) =
op_state_ref.try_borrow::<Arc<crate::server::middleware::request_context::RequestContext>>()
{
ctx.function_cache.get(&cache_key).map(|entry| entry.value().clone())
} else {
None
}
}
#[op2]
pub fn op_cache_set(
state: Rc<RefCell<OpState>>,
#[string] cache_key: String,
#[serde] value: serde_json::Value,
) {
let op_state_ref = state.borrow();
if let Some(ctx) =
op_state_ref.try_borrow::<Arc<crate::server::middleware::request_context::RequestContext>>()
{
ctx.function_cache.insert(cache_key, value);
}
}