From 4fdd44a22855aee0406a6ae921ec135647e756bb Mon Sep 17 00:00:00 2001
From: mibrohimsulaeman-a11y
 <236154532+mibrohimsulaeman-a11y@users.noreply.github.com>
Date: Sat, 6 Jun 2026 17:22:39 +0700
Subject: [PATCH] feat(http): implement HttpFormat::Raw decoder in vil_new_http
 source

---
 crates/vil_new_http/src/source.rs | 48 ++++++++++++++++++++++++++++++-
 1 file changed, 47 insertions(+), 1 deletion(-)

diff --git a/crates/vil_new_http/src/source.rs b/crates/vil_new_http/src/source.rs
index 128cacd4..14d59a4b 100644
--- a/crates/vil_new_http/src/source.rs
+++ b/crates/vil_new_http/src/source.rs
@@ -707,7 +707,53 @@ async fn execute_http_request<T: crate::sink::StreamTokenLike>(
                     }
                 }
             }
-            HttpFormat::Raw => eprintln!("[HttpSource] Raw decoder not implemented yet"),
+            HttpFormat::Raw => {
+                // Raw: read the entire response body and emit it as a single
+                // (non-streaming) message. Mirrors the NDJSON emit path:
+                // optional json_tap extraction, optional transform, then publish.
+                match response.bytes().await {
+                    Ok(body) => {
+                        let final_data = if let Some(ref tap) = json_tap {
+                            apply_json_tap(body, tap)
+                        } else {
+                            body
+                        };
+
+                        if !final_data.is_empty() {
+                            // Apply transform to response data (enrich, filter, classify)
+                            let emit_data = if let Some(ref tf) = transform_fn {
+                                tf(&final_data).map(bytes::Bytes::from)
+                            } else {
+                                Some(final_data)
+                            };
+
+                            if let Some(data) = emit_data {
+                                let msg = T::from_ndjson_line_shm(data, session_id, &world);
+                                let _ = world.publish_value(
+                                    runtime_process.id(),
+                                    out_port,
+                                    msg,
+                                );
+                            }
+                        }
+                    }
+                    Err(err) => {
+                        eprintln!(
+                            "[HttpSource] Raw body error for session {}: {:?}",
+                            session_id, err
+                        );
+                        // CORE PRIMITIVE: emit Error signal immediately on body error
+                        if let Some(ctrl_port) = ctrl_out_port {
+                            let _ = world.publish_value(
+                                runtime_process.id(),
+                                ctrl_port,
+                                ControlSignal::error(session_id, 502, "Body Read Failed"),
+                            );
+                        }
+                        return; // early return
+                    }
+                }
+            }
         },
         Err(err) => {
             eprintln!(