From 8e518c7d4700810d3b6ff949e58e91d7580b8549 Mon Sep 17 00:00:00 2001
From: Christian <christianpavilonis@outlook.com>
Date: Wed, 21 Jan 2026 09:41:36 -0600
Subject: [PATCH 1/5] update trusted-server.toml

---
 AUCTION_ORCHESTRATION_FLOW.md | 242 ++++++++++++++++++++++++++++++++++
 trusted-server.toml           |  10 +-
 2 files changed, 248 insertions(+), 4 deletions(-)
 create mode 100644 AUCTION_ORCHESTRATION_FLOW.md

diff --git a/AUCTION_ORCHESTRATION_FLOW.md b/AUCTION_ORCHESTRATION_FLOW.md
new file mode 100644
index 0000000..4e7dda6
--- /dev/null
+++ b/AUCTION_ORCHESTRATION_FLOW.md
@@ -0,0 +1,242 @@
+# 🎯 Auction Orchestration Flow
+
+## 🔄 System Flow Diagram
+
+```mermaid
+%%{init: {
+  "theme": "base",
+  "themeVariables": {
+    "background": "#ffffff",
+    "primaryColor": "#dbeafe",
+    "primaryTextColor": "#1e3a8a",
+    "primaryBorderColor": "#2563eb",
+    "lineColor": "#334155",
+    "secondaryColor": "#fef3c7",
+    "tertiaryColor": "#d1fae5",
+    "actorBkg": "#eff6ff",
+    "actorBorderColor": "#3b82f6",
+    "actorTextColor": "#1e40af",
+    "actorLineColor": "#64748b",
+    "signalColor": "#1e293b",
+    "signalTextColor": "#0f172a",
+    "labelBoxBkgColor": "#f1f5f9",
+    "labelBoxBorderColor": "#cbd5e1",
+    "labelTextColor": "#1e293b",
+    "loopTextColor": "#1e293b",
+    "noteBkgColor": "#fef3c7",
+    "noteBorderColor": "#d97706",
+    "noteTextColor": "#78350f",
+    "activationBorderColor": "#059669",
+    "activationBkgColor": "#d1fae5",
+    "sequenceNumberColor": "#0f172a"
+  },
+  "themeCSS": ".sequenceNumber{font-size:26px!important;font-weight:900!important;fill:#ffffff!important;paint-order:stroke fill;stroke:#1e293b;stroke-width:1px;} .sequenceNumber circle{r:32px!important;stroke-width:3px!important;stroke:#1e293b!important;fill:#2563eb!important;} .mermaid svg{background:#ffffff!important;border-radius:8px;box-shadow:0 2px 4px rgba(0,0,0,0.06);} .actor{font-weight:600!important;} .messageText{font-weight:600!important;font-size:16px!important;} .activation0{stroke-width:3px!important;} .messageLine0,.messageLine1{stroke-width:3px!important;} .messageText tspan{font-size:16px!important;} path.messageLine0,path.messageLine1{stroke-width:3px!important;} marker#arrowhead path,marker#crosshead path{stroke-width:2px!important;}"
+}}%%
+sequenceDiagram
+  autonumber
+
+  participant Client as 🌐 Browser/TSJS
+  participant TS as 🛡️ Trusted Server
+  participant Orch as 🎯 Orchestrator
+  participant APS as 📦 APS Provider
+  participant Prebid as 🎰 Prebid Provider
+  participant Med as ⚖️ AdServer Mediator
+  participant Mock as 🎭 Mocktioneer
+
+  %% === Auction Request Initiation ===
+  rect rgb(243,244,246)
+    Note over Client,Mock: 🚀 Auction Request Initiation
+    activate Client
+    activate TS
+    Client->>TS: POST /auction<br/>AdRequest with adUnits[]
+    Note right of Client: { "adUnits": [{ "code": "header-banner",<br/>  "mediaTypes": { "banner": { "sizes": [[728,90]] } } }] }
+    
+    TS->>TS: 🔧 Parse AdRequest<br/>🔄 Transform to AuctionRequest<br/>🆔 Generate user IDs<br/>📊 Build context
+    deactivate Client
+    deactivate TS
+  end
+
+  %% === Orchestrator Strategy Detection ===
+  rect rgb(239,246,255)
+    Note over Client,Mock: 🧠 Auction Strategy Detection
+    activate TS
+    activate Orch
+    TS->>Orch: orchestrator.run_auction()
+    Orch->>Orch: 🔍 Detect strategy<br/>mediator? parallel_mediation : parallel_only
+    deactivate TS
+    
+    Note over Orch: Strategy determined by config:<br/>[auction]<br/>mediator = "adserver_mock" → parallel_mediation<br/>No mediator → parallel_only
+  end
+
+  %% === Parallel Provider Execution ===
+  rect rgb(243,232,255)
+    Note over Client,Mock: 🔄 Parallel Provider Execution
+    activate APS
+    activate Prebid
+    activate Mock
+    
+    par Parallel Provider Calls
+      Orch->>APS: POST /e/dtb/bid<br/>APS TAM format
+      Note right of Orch: { "pubId": "5128",<br/>  "slots": [{ "slotID": "header-banner",<br/>    "sizes": [[728,90]] }] }
+      
+      APS->>Mock: APS TAM request
+      Mock-->>APS: APS bid response<br/>(encoded prices, no creative)
+      Note right of Mock: { "contextual": { "slots": [{<br/>  "slotID": "header-banner",<br/>  "amznbid": "Mi41MA==", // "2.50"<br/>  "fif": "1" }] } }
+      
+      APS-->>Orch: AuctionResponse<br/>(APS bids)
+    and
+      Orch->>Prebid: POST /openrtb2/auction<br/>OpenRTB 2.x format
+      Note right of Orch: { "id": "request",<br/>  "imp": [{ "id": "header-banner",<br/>    "banner": { "w": 728, "h": 90 } }] }
+      
+      Prebid->>Mock: OpenRTB request
+      Mock-->>Prebid: OpenRTB response<br/>(clear prices, with creative)
+      Note right of Mock: { "seatbid": [{ "seat": "prebid",<br/>  "bid": [{ "price": 2.00, "adm": "<html>..." }] }] }
+      
+      Prebid-->>Orch: AuctionResponse<br/>(Prebid bids)
+    end
+    
+    Note over Orch: 📊 Collected bids from all providers<br/>APS: encoded prices, no creative<br/>Prebid: clear prices, with creative
+    deactivate Mock
+    deactivate APS
+    deactivate Prebid
+  end
+
+  %% === Winner Selection Strategy ===
+  alt Mediator Configured (parallel_mediation)
+    rect rgb(236,253,245)
+      Note over Client,Mock: ⚖️ Mediation Flow
+      activate Med
+      Orch->>Med: POST /adserver/mediate<br/>All bids for final selection
+      Note right of Orch: { "id": "auction-123",<br/>  "imp": [...],<br/>  "ext": { "bidder_responses": [<br/>    { "bidder": "amazon-aps",<br/>      "bids": [{ "encoded_price": "Mi41MA==" }] },<br/>    { "bidder": "prebid",<br/>      "bids": [{ "price": 2.00 }] }] } }
+      
+      Med->>Med: 🔓 Decode APS encoded prices<br/>📏 Apply floor prices<br/>🏆 Select highest CPM per slot
+      Note right of Med: Base64 decode: "Mi41MA==" → "2.50"<br/>Winner: APS at $2.50 vs Prebid at $2.00
+      
+      Med-->>Orch: OpenRTB response with winners
+      Note right of Med: { "seatbid": [{ "seat": "amazon-aps",<br/>  "bid": [{ "price": 2.50, "impid": "header-banner" }] }] }
+      deactivate Med
+    end
+  else No Mediator (parallel_only)
+    rect rgb(253,243,235)
+      Note over Client,Mock: 🏆 Direct Winner Selection
+      Orch->>Orch: 📏 Compare clear prices only<br/>⚠️  Skip APS (encoded prices)<br/>🏆 Select highest CPM
+      Note right of Orch: APS bids skipped (encoded prices)<br/>Winner: Prebid at $2.00 (only clear price)
+      
+      Note over Orch: 📝 Results: Limited winner selection<br/>Cannot compare encoded APS prices<br/>Prebid wins by default
+    end
+  end
+
+  %% === Response Assembly ===
+  rect rgb(243,244,246)
+    Note over Client,Mock: 📦 Response Assembly
+    activate TS
+    activate Client
+    Orch->>Orch: 🔄 Transform to OpenRTB response<br/>🖼️ Generate iframe creatives<br/>🔏 Rewrite creative URLs<br/>📊 Add orchestrator metadata
+    
+    Orch-->>TS: OpenRTB BidResponse
+    Note right of Orch: { "id": "auction-response",<br/>  "seatbid": [{ "seat": "amazon-aps",<br/>    "bid": [{ "price": 2.50,<br/>      "adm": "<iframe src=\"/first-party/proxy?tsurl=...\">",<br/>      "w": 728, "h": 90 }] }] }<br/>  "ext": { "orchestrator": {<br/>    "strategy": "parallel_mediation",<br/>    "bidders": 2, "time_ms": 150 } }
+    
+    TS-->>Client: 200 OpenRTB response<br/>with winning creative
+    deactivate Orch
+    deactivate TS
+  end
+
+  %% === Creative Rendering ===
+  rect rgb(239,246,255)
+    Note over Client,Mock: 🖼️ Creative Rendering
+    Client->>Client: 💉 Inject winning creative<br/>🖼️ Render iframe<br/>🌐 Load creative through proxy
+    Note right of Client: iframe src="/first-party/proxy?tsurl=...&tstoken=sig"<br/>Ensures first-party serving<br/>Maintains privacy & security
+    deactivate Client
+  end
+
+```
+
+## 📋 Flow Summary
+
+### **Phase 1: Request Initiation**
+- **Browser** sends `POST /auction` with ad units in Prebid.js format
+- **Trusted Server** parses and transforms to internal `AuctionRequest`
+- Generates user IDs (persistent + fresh) and builds auction context
+
+### **Phase 2: Strategy Detection**
+- **Orchestrator** checks configuration for mediator
+- **With mediator** → `parallel_mediation` strategy
+- **Without mediator** → `parallel_only` strategy
+
+### **Phase 3: Parallel Execution**
+- **APS Provider** receives APS TAM format request
+  - Mocktioneer returns APS response with **encoded prices** (`amznbid: "Mi41MA=="`)
+  - **No creative HTML** provided (typical for real APS)
+- **Prebid Provider** receives OpenRTB 2.x request  
+  - Mocktioneer returns OpenRTB response with **clear prices**
+  - **Includes creative HTML** in `adm` field
+
+### **Phase 4: Winner Selection**
+
+#### **🔄 With Mediator (Recommended)**
+1. **AdServer Mediator** receives all bids
+2. **Decodes APS prices** (base64 → actual CPM)
+3. **Applies floor prices** and selects highest CPM per slot
+4. **Returns OpenRTB response** with proper winner selection
+
+#### **⚡ Without Mediator (Limited)**
+1. **Orchestrator** compares only clear prices
+2. **APS bids skipped** (encoded prices can't be compared)
+3. **Prebid wins by default** if no other clear-price bidders
+
+### **Phase 5: Response Assembly**
+- **Creative HTML** rewritten with first-party proxy URLs
+- **Orchestrator metadata** added (strategy, timing, bid counts)
+- **OpenRTB response** returned to browser
+
+### **Phase 6: Creative Rendering**
+- **Winning creative** injected into iframe
+- **Resources proxied** through first-party domain
+- **Privacy & security** maintained throughout
+
+## 🔑 Key Technical Details
+
+### **Price Encoding**
+- **APS Mock**: Uses base64 encoding (`"Mi41MA=="` → `"2.50"`)
+- **Real APS**: Uses proprietary encoding (only Amazon/GAM can decode)
+- **Prebid**: Uses clear decimal prices (`2.50`)
+
+### **Request Formats**
+- **APS TAM**: `{ "pubId": "...", "slots": [...] }`
+- **OpenRTB 2.x**: `{ "imp": [...] }` 
+- **AdRequest**: `{ "adUnits": [...] }`
+
+### **Response Formats**
+- **APS**: `{ "contextual": { "slots": [...] } }` (no `adm`)
+- **OpenRTB**: `{ "seatbid": [{ "seat": "...", "bid": [...] }] }`
+
+### **Configuration Examples**
+
+#### **Parallel Mediation**
+```toml
+[auction]
+enabled = true
+providers = ["prebid", "aps"]
+mediator = "adserver_mock"  # ← Enables mediation
+timeout_ms = 2000
+```
+
+#### **Parallel Only**
+```toml
+[auction]
+enabled = true
+providers = ["prebid", "aps"]
+# No mediator = direct comparison
+timeout_ms = 2000
+```
+
+### **Advantages of Mediation**
+- ✅ **Proper APS integration** - Can decode and compare APS bids
+- ✅ **Fair competition** - All bidders compete on equal footing
+- ✅ **Floor pricing** - Configurable minimum bid thresholds
+- ✅ **Flexibility** - Easy to add new providers
+
+### **Limitations Without Mediation**
+- ❌ **APS bids ignored** - Can't compare encoded prices
+- ❌ **Unfair competition** - Only clear-price bidders compete
+- ❌ **Reduced revenue** - May miss higher APS bids
\ No newline at end of file
diff --git a/trusted-server.toml b/trusted-server.toml
index a546bdb..da92cb4 100644
--- a/trusted-server.toml
+++ b/trusted-server.toml
@@ -40,6 +40,7 @@ timeout_ms = 1000
 bidders = ["kargo", "rubicon", "appnexus", "openx"]
 auto_configure = false
 debug = false
+# debug_query_params = ""
 # script_handler = "/prebid.js"
 
 [integrations.nextjs]
@@ -85,19 +86,20 @@ rewrite_sdk = true
 
 [auction]
 enabled = true
-providers = ["prebid", "aps"]
-mediator = "adserver_mock"  # will use mediator when set
+providers = ["prebid"]
+# mediator = "adserver_mock"  # will use mediator when set
 timeout_ms = 2000
 
 [integrations.aps]
-enabled = true
+enabled = false
 pub_id = "your-aps-publisher-id"
 endpoint = "https://origin-mocktioneer.cdintel.com/e/dtb/bid"
 timeout_ms = 1000
 
 [integrations.adserver_mock]
-enabled = true
+enabled = false
 endpoint = "https://origin-mocktioneer.cdintel.com/adserver/mediate"
 timeout_ms = 1000
 
 
+

From a8220b983373ffaf7da4ecb6a4468aaf5492521f Mon Sep 17 00:00:00 2001
From: Christian <christianpavilonis@outlook.com>
Date: Mon, 19 Jan 2026 17:01:12 -0600
Subject: [PATCH 2/5] fix compression issues for proxy

---
 crates/common/src/proxy.rs | 247 +++++++++++++++++++++++++++++++------
 1 file changed, 210 insertions(+), 37 deletions(-)

diff --git a/crates/common/src/proxy.rs b/crates/common/src/proxy.rs
index 3e782f5..2c737bd 100644
--- a/crates/common/src/proxy.rs
+++ b/crates/common/src/proxy.rs
@@ -3,6 +3,7 @@ use error_stack::{Report, ResultExt};
 use fastly::http::{header, HeaderValue, Method, StatusCode};
 use fastly::{Request, Response};
 use serde::{Deserialize, Serialize};
+use std::io::Read;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use crate::constants::{
@@ -11,6 +12,7 @@ use crate::constants::{
 };
 use crate::error::TrustedServerError;
 use crate::settings::Settings;
+use crate::streaming_processor::Compression;
 use crate::synthetic::get_synthetic_id;
 
 #[derive(Deserialize)]
@@ -77,13 +79,16 @@ impl<'a> ProxyRequestConfig<'a> {
     }
 }
 
+/// Encodings we support decompressing in finalize_proxied_response.
+/// We override the client's Accept-Encoding to only advertise these.
+const SUPPORTED_ENCODINGS: &str = "gzip, deflate, br";
+
 /// Copy a curated set of request headers to a proxied request.
 fn copy_proxy_forward_headers(src: &Request, dst: &mut Request) {
     for header_name in [
         HEADER_USER_AGENT,
         HEADER_ACCEPT,
         HEADER_ACCEPT_LANGUAGE,
-        HEADER_ACCEPT_ENCODING,
         HEADER_REFERER,
         HEADER_X_FORWARDED_FOR,
     ] {
@@ -91,6 +96,54 @@ fn copy_proxy_forward_headers(src: &Request, dst: &mut Request) {
             dst.set_header(&header_name, v);
         }
     }
+    // Only advertise encodings we can decompress (excludes zstd, etc.)
+    dst.set_header(HEADER_ACCEPT_ENCODING, SUPPORTED_ENCODINGS);
+}
+
+/// Decompress body bytes based on Content-Encoding header.
+/// Returns decompressed bytes or error on failure.
+fn decompress_body(
+    bytes: Vec<u8>,
+    content_encoding: &str,
+) -> Result<Vec<u8>, Report<TrustedServerError>> {
+    let compression = Compression::from_content_encoding(content_encoding);
+
+    match compression {
+        Compression::None => Ok(bytes),
+        Compression::Gzip => {
+            use flate2::read::GzDecoder;
+            let mut decoder = GzDecoder::new(&bytes[..]);
+            let mut decompressed = Vec::new();
+            decoder
+                .read_to_end(&mut decompressed)
+                .change_context(TrustedServerError::Proxy {
+                    message: "Failed to decompress gzip body".to_string(),
+                })?;
+            Ok(decompressed)
+        }
+        Compression::Deflate => {
+            use flate2::read::ZlibDecoder;
+            let mut decoder = ZlibDecoder::new(&bytes[..]);
+            let mut decompressed = Vec::new();
+            decoder
+                .read_to_end(&mut decompressed)
+                .change_context(TrustedServerError::Proxy {
+                    message: "Failed to decompress deflate body".to_string(),
+                })?;
+            Ok(decompressed)
+        }
+        Compression::Brotli => {
+            use brotli::Decompressor;
+            let mut decoder = Decompressor::new(&bytes[..], 4096);
+            let mut decompressed = Vec::new();
+            decoder
+                .read_to_end(&mut decompressed)
+                .change_context(TrustedServerError::Proxy {
+                    message: "Failed to decompress brotli body".to_string(),
+                })?;
+            Ok(decompressed)
+        }
+    }
 }
 
 // Transform the backend response into the final response sent to the client.
@@ -103,7 +156,11 @@ fn rebuild_text_response(beresp: Response, content_type: &'static str, body: Str
         .collect();
     let mut resp = Response::from_status(status);
     for (name, value) in headers {
-        if name == header::CONTENT_LENGTH || name == header::CONTENT_TYPE {
+        // Skip headers we're replacing or that are no longer valid (we decompress)
+        if name == header::CONTENT_LENGTH
+            || name == header::CONTENT_TYPE
+            || name == header::CONTENT_ENCODING
+        {
             continue;
         }
         resp.set_header(name, value);
@@ -118,14 +175,19 @@ fn finalize_proxied_response(
     req: &Request,
     target_url: &str,
     mut beresp: Response,
-) -> Response {
-    // Determine content-type from response headers
+) -> Result<Response, Report<TrustedServerError>> {
+    // Determine content-type and content-encoding from response headers
     let status_code = beresp.get_status().as_u16();
     let ct_raw = beresp
         .get_header(header::CONTENT_TYPE)
         .and_then(|h| h.to_str().ok())
         .unwrap_or("")
         .to_string();
+    let content_encoding = beresp
+        .get_header(header::CONTENT_ENCODING)
+        .and_then(|h| h.to_str().ok())
+        .unwrap_or("")
+        .to_string();
     let cl_raw = beresp
         .get_header(header::CONTENT_LENGTH)
         .and_then(|h| h.to_str().ok())
@@ -136,10 +198,16 @@ fn finalize_proxied_response(
         .unwrap_or("-");
 
     let ct_for_log: &str = if ct_raw.is_empty() { "-" } else { &ct_raw };
+    let ce_for_log: &str = if content_encoding.is_empty() {
+        "-"
+    } else {
+        &content_encoding
+    };
     log::info!(
-        "proxy: origin response status={} ct={} cl={} accept={} url={}",
+        "proxy: origin response status={} ct={} ce={} cl={} accept={} url={}",
         status_code,
         ct_for_log,
+        ce_for_log,
         cl_raw,
         accept_raw,
         target_url
@@ -148,17 +216,33 @@ fn finalize_proxied_response(
     let ct = ct_raw.to_ascii_lowercase();
 
     if ct.contains("text/html") {
-        // HTML: rewrite and serve as HTML (safe to read as string)
-        let body = beresp.take_body_str(); // TODO: don't panic!
+        // HTML: decompress if needed, rewrite, and serve as HTML
+        let raw_bytes = beresp.take_body().into_bytes();
+        let decompressed = decompress_body(raw_bytes, &content_encoding)?;
+        let body = String::from_utf8(decompressed).change_context(TrustedServerError::Proxy {
+            message: "HTML response body is not valid UTF-8".to_string(),
+        })?;
         let rewritten = crate::creative::rewrite_creative_html(&body, settings);
-        return rebuild_text_response(beresp, "text/html; charset=utf-8", rewritten);
+        return Ok(rebuild_text_response(
+            beresp,
+            "text/html; charset=utf-8",
+            rewritten,
+        ));
     }
 
     if ct.contains("text/css") {
-        // CSS: rewrite url(...) references in stylesheets (safe to read as string)
-        let body = beresp.take_body_str();
+        // CSS: decompress if needed, rewrite url(...) references in stylesheets
+        let raw_bytes = beresp.take_body().into_bytes();
+        let decompressed = decompress_body(raw_bytes, &content_encoding)?;
+        let body = String::from_utf8(decompressed).change_context(TrustedServerError::Proxy {
+            message: "CSS response body is not valid UTF-8".to_string(),
+        })?;
         let rewritten = crate::creative::rewrite_css_body(&body, settings);
-        return rebuild_text_response(beresp, "text/css; charset=utf-8", rewritten);
+        return Ok(rebuild_text_response(
+            beresp,
+            "text/css; charset=utf-8",
+            rewritten,
+        ));
     }
 
     // Image handling: set generic content-type if missing and log pixel heuristics
@@ -206,11 +290,11 @@ fn finalize_proxied_response(
             );
         }
 
-        return beresp;
+        return Ok(beresp);
     }
 
     // Passthrough for non-text, non-image responses
-    beresp
+    Ok(beresp)
 }
 
 fn finalize_proxied_response_streaming(
@@ -430,11 +514,11 @@ async fn proxy_with_redirects(
             })?;
 
         if !follow_redirects {
-            return Ok(if stream_passthrough {
-                finalize_proxied_response_streaming(req, &current_url, beresp)
+            return if stream_passthrough {
+                Ok(finalize_proxied_response_streaming(req, &current_url, beresp))
             } else {
                 finalize_proxied_response(settings, req, &current_url, beresp)
-            });
+            };
         }
 
         let status = beresp.get_status();
@@ -448,11 +532,11 @@ async fn proxy_with_redirects(
         );
 
         if !is_redirect {
-            return Ok(if stream_passthrough {
-                finalize_proxied_response_streaming(req, &current_url, beresp)
+            return if stream_passthrough {
+                Ok(finalize_proxied_response_streaming(req, &current_url, beresp))
             } else {
                 finalize_proxied_response(settings, req, &current_url, beresp)
-            });
+            };
         }
 
         let Some(location) = beresp
@@ -460,11 +544,11 @@ async fn proxy_with_redirects(
             .and_then(|h| h.to_str().ok())
             .filter(|value| !value.is_empty())
         else {
-            return Ok(if stream_passthrough {
-                finalize_proxied_response_streaming(req, &current_url, beresp)
+            return if stream_passthrough {
+                Ok(finalize_proxied_response_streaming(req, &current_url, beresp))
             } else {
                 finalize_proxied_response(settings, req, &current_url, beresp)
-            });
+            };
         };
 
         if redirect_attempt == MAX_REDIRECTS {
@@ -472,12 +556,7 @@ async fn proxy_with_redirects(
                 "proxy: redirect limit reached for {}; returning redirect response",
                 current_url
             );
-            return Ok(finalize_proxied_response(
-                settings,
-                req,
-                &current_url,
-                beresp,
-            ));
+            return finalize_proxied_response(settings, req, &current_url, beresp);
         }
 
         let next_url = url::Url::parse(location)
@@ -490,11 +569,11 @@ async fn proxy_with_redirects(
 
         let next_scheme = next_url.scheme().to_ascii_lowercase();
         if next_scheme != "http" && next_scheme != "https" {
-            return Ok(if stream_passthrough {
-                finalize_proxied_response_streaming(req, &current_url, beresp)
+            return if stream_passthrough {
+                Ok(finalize_proxied_response_streaming(req, &current_url, beresp))
             } else {
                 finalize_proxied_response(settings, req, &current_url, beresp)
-            });
+            };
         }
 
         log::info!(
@@ -1011,7 +1090,7 @@ mod tests {
     use super::{
         copy_proxy_forward_headers, handle_first_party_click, handle_first_party_proxy,
         handle_first_party_proxy_rebuild, handle_first_party_proxy_sign,
-        reconstruct_and_validate_signed_target, ProxyRequestConfig,
+        reconstruct_and_validate_signed_target, ProxyRequestConfig, SUPPORTED_ENCODINGS,
     };
     use crate::error::{IntoHttpResponse, TrustedServerError};
     use crate::test_support::tests::create_test_settings;
@@ -1363,12 +1442,13 @@ mod tests {
                 .unwrap(),
             "en-US"
         );
+        // Accept-Encoding is overridden to only include supported encodings
         assert_eq!(
             dst.get_header(HEADER_ACCEPT_ENCODING)
                 .unwrap()
                 .to_str()
                 .unwrap(),
-            "gzip"
+            SUPPORTED_ENCODINGS
         );
         assert_eq!(
             dst.get_header(HEADER_REFERER).unwrap().to_str().unwrap(),
@@ -1426,7 +1506,8 @@ mod tests {
         let direct = creative::rewrite_creative_html(html, &settings);
         assert!(direct.contains("/first-party/proxy?tsurl="), "{}", direct);
         let req = Request::new(Method::GET, "https://edge.example/first-party/proxy");
-        let out = finalize(&settings, &req, "https://cdn.example/a.png", beresp);
+        let out = finalize(&settings, &req, "https://cdn.example/a.png", beresp)
+            .expect("finalize should succeed");
         let ct = out
             .get_header(header::CONTENT_TYPE)
             .unwrap()
@@ -1453,7 +1534,8 @@ mod tests {
             .with_header(header::CONTENT_TYPE, "text/css")
             .with_body(css);
         let req = Request::new(Method::GET, "https://edge.example/first-party/proxy");
-        let mut out = finalize(&settings, &req, "https://cdn.example/bg.png", beresp);
+        let mut out = finalize(&settings, &req, "https://cdn.example/bg.png", beresp)
+            .expect("finalize should succeed");
         let body = out.take_body_str();
         assert!(body.contains("/first-party/proxy?tsurl="), "{}", body);
         let ct = out
@@ -1470,7 +1552,8 @@ mod tests {
         let beresp = Response::from_status(StatusCode::OK).with_body("PNG");
         let mut req = Request::new(Method::GET, "https://edge.example/first-party/proxy");
         req.set_header(HEADER_ACCEPT, "image/*");
-        let out = finalize(&settings, &req, "https://cdn.example/pixel.gif", beresp);
+        let out = finalize(&settings, &req, "https://cdn.example/pixel.gif", beresp)
+            .expect("finalize should succeed");
         // Since CT was missing and Accept indicates image, it should set generic image/*
         let ct = out
             .get_header(header::CONTENT_TYPE)
@@ -1487,7 +1570,8 @@ mod tests {
             .with_header(header::CONTENT_TYPE, "application/json")
             .with_body("{\"ok\":true}");
         let req = Request::new(Method::GET, "https://edge.example/first-party/proxy");
-        let mut out = finalize(&settings, &req, "https://api.example/ok", beresp);
+        let mut out = finalize(&settings, &req, "https://api.example/ok", beresp)
+            .expect("finalize should succeed");
         // Should not rewrite, preserve status and content-type
         assert_eq!(out.get_status(), StatusCode::ACCEPTED);
         let ct = out
@@ -1499,4 +1583,93 @@ mod tests {
         let body = out.take_body_str();
         assert_eq!(body, "{\"ok\":true}");
     }
+
+    #[test]
+    fn html_gzip_response_is_decompressed_and_rewritten() {
+        use flate2::write::GzEncoder;
+        use flate2::Compression;
+        use std::io::Write;
+
+        let settings = create_test_settings();
+        let html = r#"<html><body><img src="https://cdn.example/a.png"></body></html>"#;
+
+        // Gzip compress the HTML
+        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
+        encoder.write_all(html.as_bytes()).unwrap();
+        let compressed = encoder.finish().unwrap();
+
+        let beresp = Response::from_status(StatusCode::OK)
+            .with_header(header::CONTENT_TYPE, "text/html; charset=utf-8")
+            .with_header(header::CONTENT_ENCODING, "gzip")
+            .with_body(compressed);
+
+        let req = Request::new(Method::GET, "https://edge.example/first-party/proxy");
+        let mut out = finalize(&settings, &req, "https://cdn.example/a.png", beresp)
+            .expect("finalize should decompress and succeed");
+
+        // Should have removed Content-Encoding header
+        assert!(
+            out.get_header(header::CONTENT_ENCODING).is_none(),
+            "Content-Encoding should be removed"
+        );
+
+        let ct = out
+            .get_header(header::CONTENT_TYPE)
+            .unwrap()
+            .to_str()
+            .unwrap();
+        assert_eq!(ct, "text/html; charset=utf-8");
+
+        let body = out.take_body_str();
+        assert!(
+            body.contains("/first-party/proxy?tsurl="),
+            "HTML should be rewritten: {}",
+            body
+        );
+    }
+
+    #[test]
+    fn css_brotli_response_is_decompressed_and_rewritten() {
+        use brotli::enc::writer::CompressorWriter;
+        use std::io::Write;
+
+        let settings = create_test_settings();
+        let css = "body{background:url(https://cdn.example/bg.png)}";
+
+        // Brotli compress the CSS
+        let mut compressed = Vec::new();
+        {
+            let mut encoder = CompressorWriter::new(&mut compressed, 4096, 4, 22);
+            encoder.write_all(css.as_bytes()).unwrap();
+        }
+
+        let beresp = Response::from_status(StatusCode::OK)
+            .with_header(header::CONTENT_TYPE, "text/css")
+            .with_header(header::CONTENT_ENCODING, "br")
+            .with_body(compressed);
+
+        let req = Request::new(Method::GET, "https://edge.example/first-party/proxy");
+        let mut out = finalize(&settings, &req, "https://cdn.example/bg.png", beresp)
+            .expect("finalize should decompress brotli and succeed");
+
+        // Should have removed Content-Encoding header
+        assert!(
+            out.get_header(header::CONTENT_ENCODING).is_none(),
+            "Content-Encoding should be removed"
+        );
+
+        let ct = out
+            .get_header(header::CONTENT_TYPE)
+            .unwrap()
+            .to_str()
+            .unwrap();
+        assert_eq!(ct, "text/css; charset=utf-8");
+
+        let body = out.take_body_str();
+        assert!(
+            body.contains("/first-party/proxy?tsurl="),
+            "CSS should be rewritten: {}",
+            body
+        );
+    }
 }

From d0b99c13873546ca4e6dddf0cf5af0267836a6e4 Mon Sep 17 00:00:00 2001
From: Christian <christianpavilonis@outlook.com>
Date: Tue, 20 Jan 2026 14:57:05 -0600
Subject: [PATCH 3/5] Reuse StreamProcessor to handle compression

---
 crates/common/src/creative.rs |  80 +++++++++++
 crates/common/src/proxy.rs    | 250 +++++++++++++++++++++-------------
 2 files changed, 232 insertions(+), 98 deletions(-)

diff --git a/crates/common/src/creative.rs b/crates/common/src/creative.rs
index 02041dd..7c0a361 100644
--- a/crates/common/src/creative.rs
+++ b/crates/common/src/creative.rs
@@ -39,8 +39,10 @@
 
 use crate::http_util::compute_encrypted_sha256_token;
 use crate::settings::Settings;
+use crate::streaming_processor::StreamProcessor;
 use crate::tsjs;
 use lol_html::{element, html_content::ContentType, text, HtmlRewriter, Settings as HtmlSettings};
+use std::io;
 
 // Helper: normalize to absolute URL if http/https or protocol-relative. Otherwise None.
 // Checks against the rewrite blacklist to exclude configured domains/patterns from proxying.
@@ -478,6 +480,84 @@ pub fn rewrite_creative_html(markup: &str, settings: &Settings) -> String {
     String::from_utf8(out).unwrap_or_else(|_| markup.to_string())
 }
 
+/// Stream processor for creative HTML that rewrites URLs to first-party proxy.
+///
+/// This processor buffers input chunks and processes the complete HTML document
+/// when the stream ends, using `rewrite_creative_html` internally.
+pub struct CreativeHtmlProcessor {
+    settings: Settings,
+    buffer: Vec<u8>,
+}
+
+impl CreativeHtmlProcessor {
+    /// Create a new HTML processor with the given settings.
+    pub fn new(settings: &Settings) -> Self {
+        Self {
+            settings: settings.clone(),
+            buffer: Vec::new(),
+        }
+    }
+}
+
+impl StreamProcessor for CreativeHtmlProcessor {
+    fn process_chunk(&mut self, chunk: &[u8], is_last: bool) -> Result<Vec<u8>, io::Error> {
+        self.buffer.extend_from_slice(chunk);
+
+        if is_last {
+            let markup = String::from_utf8(std::mem::take(&mut self.buffer))
+                .map_err(|e| io::Error::other(format!("Invalid UTF-8 in HTML: {}", e)))?;
+
+            let rewritten = rewrite_creative_html(&markup, &self.settings);
+            Ok(rewritten.into_bytes())
+        } else {
+            Ok(Vec::new())
+        }
+    }
+
+    fn reset(&mut self) {
+        self.buffer.clear();
+    }
+}
+
+/// Stream processor for CSS that rewrites url() references to first-party proxy.
+///
+/// This processor buffers input chunks and processes the complete CSS
+/// when the stream ends, using `rewrite_css_body` internally.
+pub struct CreativeCssProcessor {
+    settings: Settings,
+    buffer: Vec<u8>,
+}
+
+impl CreativeCssProcessor {
+    /// Create a new CSS processor with the given settings.
+    pub fn new(settings: &Settings) -> Self {
+        Self {
+            settings: settings.clone(),
+            buffer: Vec::new(),
+        }
+    }
+}
+
+impl StreamProcessor for CreativeCssProcessor {
+    fn process_chunk(&mut self, chunk: &[u8], is_last: bool) -> Result<Vec<u8>, io::Error> {
+        self.buffer.extend_from_slice(chunk);
+
+        if is_last {
+            let css = String::from_utf8(std::mem::take(&mut self.buffer))
+                .map_err(|e| io::Error::other(format!("Invalid UTF-8 in CSS: {}", e)))?;
+
+            let rewritten = rewrite_css_body(&css, &self.settings);
+            Ok(rewritten.into_bytes())
+        } else {
+            Ok(Vec::new())
+        }
+    }
+
+    fn reset(&mut self) {
+        self.buffer.clear();
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::{rewrite_creative_html, rewrite_srcset, rewrite_style_urls, to_abs};
diff --git a/crates/common/src/proxy.rs b/crates/common/src/proxy.rs
index 2c737bd..de3911c 100644
--- a/crates/common/src/proxy.rs
+++ b/crates/common/src/proxy.rs
@@ -3,16 +3,16 @@ use error_stack::{Report, ResultExt};
 use fastly::http::{header, HeaderValue, Method, StatusCode};
 use fastly::{Request, Response};
 use serde::{Deserialize, Serialize};
-use std::io::Read;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use crate::constants::{
     HEADER_ACCEPT, HEADER_ACCEPT_ENCODING, HEADER_ACCEPT_LANGUAGE, HEADER_REFERER,
     HEADER_USER_AGENT, HEADER_X_FORWARDED_FOR,
 };
+use crate::creative::{CreativeCssProcessor, CreativeHtmlProcessor};
 use crate::error::TrustedServerError;
 use crate::settings::Settings;
-use crate::streaming_processor::Compression;
+use crate::streaming_processor::{Compression, PipelineConfig, StreamingPipeline};
 use crate::synthetic::get_synthetic_id;
 
 #[derive(Deserialize)]
@@ -100,55 +100,15 @@ fn copy_proxy_forward_headers(src: &Request, dst: &mut Request) {
     dst.set_header(HEADER_ACCEPT_ENCODING, SUPPORTED_ENCODINGS);
 }
 
-/// Decompress body bytes based on Content-Encoding header.
-/// Returns decompressed bytes or error on failure.
-fn decompress_body(
-    bytes: Vec<u8>,
-    content_encoding: &str,
-) -> Result<Vec<u8>, Report<TrustedServerError>> {
-    let compression = Compression::from_content_encoding(content_encoding);
-
-    match compression {
-        Compression::None => Ok(bytes),
-        Compression::Gzip => {
-            use flate2::read::GzDecoder;
-            let mut decoder = GzDecoder::new(&bytes[..]);
-            let mut decompressed = Vec::new();
-            decoder
-                .read_to_end(&mut decompressed)
-                .change_context(TrustedServerError::Proxy {
-                    message: "Failed to decompress gzip body".to_string(),
-                })?;
-            Ok(decompressed)
-        }
-        Compression::Deflate => {
-            use flate2::read::ZlibDecoder;
-            let mut decoder = ZlibDecoder::new(&bytes[..]);
-            let mut decompressed = Vec::new();
-            decoder
-                .read_to_end(&mut decompressed)
-                .change_context(TrustedServerError::Proxy {
-                    message: "Failed to decompress deflate body".to_string(),
-                })?;
-            Ok(decompressed)
-        }
-        Compression::Brotli => {
-            use brotli::Decompressor;
-            let mut decoder = Decompressor::new(&bytes[..], 4096);
-            let mut decompressed = Vec::new();
-            decoder
-                .read_to_end(&mut decompressed)
-                .change_context(TrustedServerError::Proxy {
-                    message: "Failed to decompress brotli body".to_string(),
-                })?;
-            Ok(decompressed)
-        }
-    }
-}
-
-// Transform the backend response into the final response sent to the client.
-// Handles HTML and CSS rewrites and image content-type normalization.
-fn rebuild_text_response(beresp: Response, content_type: &'static str, body: String) -> Response {
+/// Rebuild a response with a new body, preserving headers except Content-Length.
+/// If `preserve_encoding` is true, the Content-Encoding header is kept (for compressed responses).
+/// If false, Content-Encoding is stripped (for decompressed responses).
+fn rebuild_response_with_body(
+    beresp: Response,
+    content_type: &'static str,
+    body: Vec<u8>,
+    preserve_encoding: bool,
+) -> Response {
     let status = beresp.get_status();
     let headers: Vec<(header::HeaderName, HeaderValue)> = beresp
         .get_headers()
@@ -156,11 +116,12 @@ fn rebuild_text_response(beresp: Response, content_type: &'static str, body: Str
         .collect();
     let mut resp = Response::from_status(status);
     for (name, value) in headers {
-        // Skip headers we're replacing or that are no longer valid (we decompress)
-        if name == header::CONTENT_LENGTH
-            || name == header::CONTENT_TYPE
-            || name == header::CONTENT_ENCODING
-        {
+        // Always skip Content-Length (size changed) and Content-Type (we set it)
+        if name == header::CONTENT_LENGTH || name == header::CONTENT_TYPE {
+            continue;
+        }
+        // Skip Content-Encoding only if we're not preserving it
+        if name == header::CONTENT_ENCODING && !preserve_encoding {
             continue;
         }
         resp.set_header(name, value);
@@ -214,34 +175,57 @@ fn finalize_proxied_response(
     );
 
     let ct = ct_raw.to_ascii_lowercase();
+    let compression = Compression::from_content_encoding(&content_encoding);
 
     if ct.contains("text/html") {
-        // HTML: decompress if needed, rewrite, and serve as HTML
-        let raw_bytes = beresp.take_body().into_bytes();
-        let decompressed = decompress_body(raw_bytes, &content_encoding)?;
-        let body = String::from_utf8(decompressed).change_context(TrustedServerError::Proxy {
-            message: "HTML response body is not valid UTF-8".to_string(),
-        })?;
-        let rewritten = crate::creative::rewrite_creative_html(&body, settings);
-        return Ok(rebuild_text_response(
+        // HTML: use streaming pipeline to decompress, rewrite, and re-compress
+        let processor = CreativeHtmlProcessor::new(settings);
+        let config = PipelineConfig {
+            input_compression: compression,
+            output_compression: compression, // Preserve compression
+            chunk_size: 8192,
+        };
+
+        let body = beresp.take_body();
+        let mut output = Vec::new();
+        let mut pipeline = StreamingPipeline::new(config, processor);
+        pipeline
+            .process(body, &mut output)
+            .change_context(TrustedServerError::Proxy {
+                message: "Failed to process HTML response".to_string(),
+            })?;
+
+        return Ok(rebuild_response_with_body(
             beresp,
             "text/html; charset=utf-8",
-            rewritten,
+            output,
+            compression != Compression::None, // preserve encoding if compressed
         ));
     }
 
     if ct.contains("text/css") {
-        // CSS: decompress if needed, rewrite url(...) references in stylesheets
-        let raw_bytes = beresp.take_body().into_bytes();
-        let decompressed = decompress_body(raw_bytes, &content_encoding)?;
-        let body = String::from_utf8(decompressed).change_context(TrustedServerError::Proxy {
-            message: "CSS response body is not valid UTF-8".to_string(),
-        })?;
-        let rewritten = crate::creative::rewrite_css_body(&body, settings);
-        return Ok(rebuild_text_response(
+        // CSS: use streaming pipeline to decompress, rewrite url(...), and re-compress
+        let processor = CreativeCssProcessor::new(settings);
+        let config = PipelineConfig {
+            input_compression: compression,
+            output_compression: compression, // Preserve compression
+            chunk_size: 8192,
+        };
+
+        let body = beresp.take_body();
+        let mut output = Vec::new();
+        let mut pipeline = StreamingPipeline::new(config, processor);
+        pipeline
+            .process(body, &mut output)
+            .change_context(TrustedServerError::Proxy {
+                message: "Failed to process CSS response".to_string(),
+            })?;
+
+        return Ok(rebuild_response_with_body(
             beresp,
             "text/css; charset=utf-8",
-            rewritten,
+            output,
+            compression != Compression::None, // preserve encoding if compressed
         ));
     }
 
@@ -515,7 +499,11 @@ async fn proxy_with_redirects(
 
         if !follow_redirects {
             return if stream_passthrough {
-                Ok(finalize_proxied_response_streaming(req, &current_url, beresp))
+                Ok(finalize_proxied_response_streaming(
+                    req,
+                    &current_url,
+                    beresp,
+                ))
             } else {
                 finalize_proxied_response(settings, req, &current_url, beresp)
             };
@@ -533,7 +521,11 @@ async fn proxy_with_redirects(
 
         if !is_redirect {
             return if stream_passthrough {
-                Ok(finalize_proxied_response_streaming(req, &current_url, beresp))
+                Ok(finalize_proxied_response_streaming(
+                    req,
+                    &current_url,
+                    beresp,
+                ))
             } else {
                 finalize_proxied_response(settings, req, &current_url, beresp)
             };
@@ -545,7 +537,11 @@ async fn proxy_with_redirects(
             .filter(|value| !value.is_empty())
         else {
             return if stream_passthrough {
-                Ok(finalize_proxied_response_streaming(req, &current_url, beresp))
+                Ok(finalize_proxied_response_streaming(
+                    req,
+                    &current_url,
+                    beresp,
+                ))
             } else {
                 finalize_proxied_response(settings, req, &current_url, beresp)
             };
@@ -570,7 +566,11 @@ async fn proxy_with_redirects(
         let next_scheme = next_url.scheme().to_ascii_lowercase();
         if next_scheme != "http" && next_scheme != "https" {
             return if stream_passthrough {
-                Ok(finalize_proxied_response_streaming(req, &current_url, beresp))
+                Ok(finalize_proxied_response_streaming(
+                    req,
+                    &current_url,
+                    beresp,
+                ))
             } else {
                 finalize_proxied_response(settings, req, &current_url, beresp)
             };
@@ -1585,10 +1585,11 @@ mod tests {
     }
 
     #[test]
-    fn html_gzip_response_is_decompressed_and_rewritten() {
+    fn html_gzip_response_is_processed_with_compression_preserved() {
+        use flate2::read::GzDecoder;
         use flate2::write::GzEncoder;
         use flate2::Compression;
-        use std::io::Write;
+        use std::io::{Read, Write};
 
         let settings = create_test_settings();
         let html = r#"<html><body><img src="https://cdn.example/a.png"></body></html>"#;
@@ -1604,14 +1605,16 @@ mod tests {
             .with_body(compressed);
 
         let req = Request::new(Method::GET, "https://edge.example/first-party/proxy");
-        let mut out = finalize(&settings, &req, "https://cdn.example/a.png", beresp)
-            .expect("finalize should decompress and succeed");
+        let out = finalize(&settings, &req, "https://cdn.example/a.png", beresp)
+            .expect("finalize should process and succeed");
 
-        // Should have removed Content-Encoding header
-        assert!(
-            out.get_header(header::CONTENT_ENCODING).is_none(),
-            "Content-Encoding should be removed"
-        );
+        // Content-Encoding should be preserved (gzip in -> gzip out)
+        let ce = out
+            .get_header(header::CONTENT_ENCODING)
+            .expect("Content-Encoding should be preserved")
+            .to_str()
+            .unwrap();
+        assert_eq!(ce, "gzip");
 
         let ct = out
             .get_header(header::CONTENT_TYPE)
@@ -1620,18 +1623,26 @@ mod tests {
             .unwrap();
         assert_eq!(ct, "text/html; charset=utf-8");
 
-        let body = out.take_body_str();
+        // Decompress output to verify content was rewritten
+        let compressed_output = out.into_body().into_bytes();
+        let mut decoder = GzDecoder::new(&compressed_output[..]);
+        let mut decompressed = String::new();
+        decoder
+            .read_to_string(&mut decompressed)
+            .expect("should decompress output");
+
         assert!(
-            body.contains("/first-party/proxy?tsurl="),
+            decompressed.contains("/first-party/proxy?tsurl="),
             "HTML should be rewritten: {}",
-            body
+            decompressed
         );
     }
 
     #[test]
-    fn css_brotli_response_is_decompressed_and_rewritten() {
+    fn css_brotli_response_is_processed_with_compression_preserved() {
         use brotli::enc::writer::CompressorWriter;
-        use std::io::Write;
+        use brotli::Decompressor;
+        use std::io::{Read, Write};
 
         let settings = create_test_settings();
         let css = "body{background:url(https://cdn.example/bg.png)}";
@@ -1649,13 +1660,56 @@ mod tests {
             .with_body(compressed);
 
         let req = Request::new(Method::GET, "https://edge.example/first-party/proxy");
-        let mut out = finalize(&settings, &req, "https://cdn.example/bg.png", beresp)
-            .expect("finalize should decompress brotli and succeed");
+        let out = finalize(&settings, &req, "https://cdn.example/bg.png", beresp)
+            .expect("finalize should process brotli and succeed");
+
+        // Content-Encoding should be preserved (br in -> br out)
+        let ce = out
+            .get_header(header::CONTENT_ENCODING)
+            .expect("Content-Encoding should be preserved")
+            .to_str()
+            .unwrap();
+        assert_eq!(ce, "br");
+
+        let ct = out
+            .get_header(header::CONTENT_TYPE)
+            .unwrap()
+            .to_str()
+            .unwrap();
+        assert_eq!(ct, "text/css; charset=utf-8");
+
+        // Decompress output to verify content was rewritten
+        let compressed_output = out.into_body().into_bytes();
+        let mut decoder = Decompressor::new(&compressed_output[..], 4096);
+        let mut decompressed = String::new();
+        decoder
+            .read_to_string(&mut decompressed)
+            .expect("should decompress brotli output");
 
-        // Should have removed Content-Encoding header
+        assert!(
+            decompressed.contains("/first-party/proxy?tsurl="),
+            "CSS should be rewritten: {}",
+            decompressed
+        );
+    }
+
+    #[test]
+    fn html_uncompressed_response_is_processed_without_encoding() {
+        let settings = create_test_settings();
+        let html = r#"<html><body><img src="https://cdn.example/a.png"></body></html>"#;
+
+        let beresp = Response::from_status(StatusCode::OK)
+            .with_header(header::CONTENT_TYPE, "text/html; charset=utf-8")
+            .with_body(html);
+
+        let req = Request::new(Method::GET, "https://edge.example/first-party/proxy");
+        let mut out = finalize(&settings, &req, "https://cdn.example/a.png", beresp)
+            .expect("finalize should succeed");
+
+        // No Content-Encoding since input was uncompressed
         assert!(
             out.get_header(header::CONTENT_ENCODING).is_none(),
-            "Content-Encoding should be removed"
+            "Content-Encoding should not be set for uncompressed input"
         );
 
         let ct = out
@@ -1663,12 +1717,12 @@ mod tests {
             .unwrap()
             .to_str()
             .unwrap();
-        assert_eq!(ct, "text/css; charset=utf-8");
+        assert_eq!(ct, "text/html; charset=utf-8");
 
         let body = out.take_body_str();
         assert!(
             body.contains("/first-party/proxy?tsurl="),
-            "CSS should be rewritten: {}",
+            "HTML should be rewritten: {}",
             body
         );
     }

From fe07f897a5600acb0b680c6fab1aa7741943dfd6 Mon Sep 17 00:00:00 2001
From: Christian <christianpavilonis@outlook.com>
Date: Wed, 21 Jan 2026 09:45:28 -0600
Subject: [PATCH 4/5] typescript format

---
 crates/js/lib/src/core/request.ts | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/crates/js/lib/src/core/request.ts b/crates/js/lib/src/core/request.ts
index 8aed30e..a84af08 100644
--- a/crates/js/lib/src/core/request.ts
+++ b/crates/js/lib/src/core/request.ts
@@ -135,7 +135,12 @@ function renderCreativeInline(
     // Write creative HTML directly into iframe using srcdoc
     iframe.srcdoc = buildCreativeDocument(creativeHtml);
 
-    log.info('renderCreativeInline: rendered', { slotId, width, height, htmlLength: creativeHtml.length });
+    log.info('renderCreativeInline: rendered', {
+      slotId,
+      width,
+      height,
+      htmlLength: creativeHtml.length,
+    });
   } catch (err) {
     log.warn('renderCreativeInline: failed', { slotId, err });
   }

From 661fff9bc3d272201821633fb24bb1535c40cea9 Mon Sep 17 00:00:00 2001
From: Christian <christianpavilonis@outlook.com>
Date: Thu, 22 Jan 2026 15:50:09 -0600
Subject: [PATCH 5/5] Address pr issues

---
 crates/common/src/creative.rs | 44 +++++++++++------
 crates/common/src/proxy.rs    | 89 +++++++++++++++++++----------------
 2 files changed, 78 insertions(+), 55 deletions(-)

diff --git a/crates/common/src/creative.rs b/crates/common/src/creative.rs
index 7c0a361..862ebb9 100644
--- a/crates/common/src/creative.rs
+++ b/crates/common/src/creative.rs
@@ -44,6 +44,10 @@ use crate::tsjs;
 use lol_html::{element, html_content::ContentType, text, HtmlRewriter, Settings as HtmlSettings};
 use std::io;
 
+/// Maximum size of response body that can be buffered for rewriting.
+/// Responses larger than this will be rejected to prevent memory exhaustion.
+const MAX_REWRITABLE_BODY_SIZE: usize = 10 * 1024 * 1024; // 10 MB
+
 // Helper: normalize to absolute URL if http/https or protocol-relative. Otherwise None.
 // Checks against the rewrite blacklist to exclude configured domains/patterns from proxying.
 pub(super) fn to_abs(u: &str, settings: &Settings) -> Option<String> {
@@ -484,30 +488,36 @@ pub fn rewrite_creative_html(markup: &str, settings: &Settings) -> String {
 ///
 /// This processor buffers input chunks and processes the complete HTML document
 /// when the stream ends, using `rewrite_creative_html` internally.
-pub struct CreativeHtmlProcessor {
-    settings: Settings,
+pub struct CreativeHtmlProcessor<'a> {
+    settings: &'a Settings,
     buffer: Vec<u8>,
 }
 
-impl CreativeHtmlProcessor {
+impl<'a> CreativeHtmlProcessor<'a> {
     /// Create a new HTML processor with the given settings.
-    pub fn new(settings: &Settings) -> Self {
+    pub fn new(settings: &'a Settings) -> Self {
         Self {
-            settings: settings.clone(),
+            settings,
             buffer: Vec::new(),
         }
     }
 }
 
-impl StreamProcessor for CreativeHtmlProcessor {
+impl StreamProcessor for CreativeHtmlProcessor<'_> {
     fn process_chunk(&mut self, chunk: &[u8], is_last: bool) -> Result<Vec<u8>, io::Error> {
+        if self.buffer.len() + chunk.len() > MAX_REWRITABLE_BODY_SIZE {
+            return Err(io::Error::other(format!(
+                "HTML response body exceeds maximum rewritable size of {} bytes",
+                MAX_REWRITABLE_BODY_SIZE
+            )));
+        }
         self.buffer.extend_from_slice(chunk);
 
         if is_last {
             let markup = String::from_utf8(std::mem::take(&mut self.buffer))
                 .map_err(|e| io::Error::other(format!("Invalid UTF-8 in HTML: {}", e)))?;
 
-            let rewritten = rewrite_creative_html(&markup, &self.settings);
+            let rewritten = rewrite_creative_html(&markup, self.settings);
             Ok(rewritten.into_bytes())
         } else {
             Ok(Vec::new())
@@ -523,30 +533,36 @@ impl StreamProcessor for CreativeHtmlProcessor {
 ///
 /// This processor buffers input chunks and processes the complete CSS
 /// when the stream ends, using `rewrite_css_body` internally.
-pub struct CreativeCssProcessor {
-    settings: Settings,
+pub struct CreativeCssProcessor<'a> {
+    settings: &'a Settings,
     buffer: Vec<u8>,
 }
 
-impl CreativeCssProcessor {
+impl<'a> CreativeCssProcessor<'a> {
     /// Create a new CSS processor with the given settings.
-    pub fn new(settings: &Settings) -> Self {
+    pub fn new(settings: &'a Settings) -> Self {
         Self {
-            settings: settings.clone(),
+            settings,
             buffer: Vec::new(),
         }
     }
 }
 
-impl StreamProcessor for CreativeCssProcessor {
+impl StreamProcessor for CreativeCssProcessor<'_> {
     fn process_chunk(&mut self, chunk: &[u8], is_last: bool) -> Result<Vec<u8>, io::Error> {
+        if self.buffer.len() + chunk.len() > MAX_REWRITABLE_BODY_SIZE {
+            return Err(io::Error::other(format!(
+                "CSS response body exceeds maximum rewritable size of {} bytes",
+                MAX_REWRITABLE_BODY_SIZE
+            )));
+        }
         self.buffer.extend_from_slice(chunk);
 
         if is_last {
             let css = String::from_utf8(std::mem::take(&mut self.buffer))
                 .map_err(|e| io::Error::other(format!("Invalid UTF-8 in CSS: {}", e)))?;
 
-            let rewritten = rewrite_css_body(&css, &self.settings);
+            let rewritten = rewrite_css_body(&css, self.settings);
             Ok(rewritten.into_bytes())
         } else {
             Ok(Vec::new())
diff --git a/crates/common/src/proxy.rs b/crates/common/src/proxy.rs
index de3911c..35bad9b 100644
--- a/crates/common/src/proxy.rs
+++ b/crates/common/src/proxy.rs
@@ -12,9 +12,12 @@ use crate::constants::{
 use crate::creative::{CreativeCssProcessor, CreativeHtmlProcessor};
 use crate::error::TrustedServerError;
 use crate::settings::Settings;
-use crate::streaming_processor::{Compression, PipelineConfig, StreamingPipeline};
+use crate::streaming_processor::{Compression, PipelineConfig, StreamProcessor, StreamingPipeline};
 use crate::synthetic::get_synthetic_id;
 
+/// Chunk size used for streaming content through the rewrite pipeline.
+const STREAMING_CHUNK_SIZE: usize = 8192;
+
 #[derive(Deserialize)]
 struct ProxySignReq {
     url: String,
@@ -131,6 +134,40 @@ fn rebuild_response_with_body(
     resp
 }
 
+/// Process a response body through a streaming pipeline with the given processor.
+///
+/// Handles decompression, content processing, and re-compression while preserving
+/// the response status and headers.
+fn process_response_with_pipeline<P: StreamProcessor>(
+    mut beresp: Response,
+    processor: P,
+    compression: Compression,
+    content_type: &'static str,
+    error_context: &'static str,
+) -> Result<Response, Report<TrustedServerError>> {
+    let config = PipelineConfig {
+        input_compression: compression,
+        output_compression: compression,
+        chunk_size: STREAMING_CHUNK_SIZE,
+    };
+
+    let body = beresp.take_body();
+    let mut output = Vec::new();
+    let mut pipeline = StreamingPipeline::new(config, processor);
+    pipeline
+        .process(body, &mut output)
+        .change_context(TrustedServerError::Proxy {
+            message: error_context.to_string(),
+        })?;
+
+    Ok(rebuild_response_with_body(
+        beresp,
+        content_type,
+        output,
+        compression != Compression::None,
+    ))
+}
+
 fn finalize_proxied_response(
     settings: &Settings,
     req: &Request,
@@ -178,55 +215,25 @@ fn finalize_proxied_response(
     let compression = Compression::from_content_encoding(&content_encoding);
 
     if ct.contains("text/html") {
-        // HTML: use streaming pipeline to decompress, rewrite, and re-compress
         let processor = CreativeHtmlProcessor::new(settings);
-        let config = PipelineConfig {
-            input_compression: compression,
-            output_compression: compression, // Preserve compression
-            chunk_size: 8192,
-        };
-
-        let body = beresp.take_body();
-        let mut output = Vec::new();
-        let mut pipeline = StreamingPipeline::new(config, processor);
-        pipeline
-            .process(body, &mut output)
-            .change_context(TrustedServerError::Proxy {
-                message: "Failed to process HTML response".to_string(),
-            })?;
-
-        return Ok(rebuild_response_with_body(
+        return process_response_with_pipeline(
             beresp,
+            processor,
+            compression,
             "text/html; charset=utf-8",
-            output,
-            compression != Compression::None, // preserve encoding if compressed
-        ));
+            "Failed to process HTML response",
+        );
     }
 
     if ct.contains("text/css") {
-        // CSS: use streaming pipeline to decompress, rewrite url(...), and re-compress
         let processor = CreativeCssProcessor::new(settings);
-        let config = PipelineConfig {
-            input_compression: compression,
-            output_compression: compression, // Preserve compression
-            chunk_size: 8192,
-        };
-
-        let body = beresp.take_body();
-        let mut output = Vec::new();
-        let mut pipeline = StreamingPipeline::new(config, processor);
-        pipeline
-            .process(body, &mut output)
-            .change_context(TrustedServerError::Proxy {
-                message: "Failed to process CSS response".to_string(),
-            })?;
-
-        return Ok(rebuild_response_with_body(
+        return process_response_with_pipeline(
             beresp,
+            processor,
+            compression,
             "text/css; charset=utf-8",
-            output,
-            compression != Compression::None, // preserve encoding if compressed
-        ));
+            "Failed to process CSS response",
+        );
     }
 
     // Image handling: set generic content-type if missing and log pixel heuristics