break arpart into multiple files

Cargo.lock

@@ -2,6 +2,34 @@
 # It is not intended for manual editing.
 version = 4
 
+[[package]]
+name = "ahash"
+version = "0.8.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75"
+dependencies = [
+ "cfg-if",
+ "getrandom 0.3.4",
+ "once_cell",
+ "version_check",
+ "zerocopy",
+]
+
+[[package]]
+name = "aho-corasick"
+version = "1.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301"
+dependencies = [
+ "memchr",
+]
+
+[[package]]
+name = "allocator-api2"
+version = "0.2.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
+
 [[package]]
 name = "android_system_properties"
 version = "0.1.5"
@@ -67,6 +95,12 @@ version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
 
+[[package]]
+name = "base64"
+version = "0.22.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
+
 [[package]]
 name = "bitflags"
 version = "2.10.0"
@@ -129,6 +163,15 @@ dependencies = [
  "windows-link",
 ]
 
+[[package]]
+name = "chumsky"
+version = "0.9.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8eebd66744a15ded14960ab4ccdbfb51ad3b81f51f3f04a80adac98c985396c9"
+dependencies = [
+ "hashbrown 0.14.5",
+]
+
 [[package]]
 name = "clap"
 version = "4.5.55"
@@ -237,6 +280,18 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "dyn-clone"
+version = "1.0.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"
+
+[[package]]
+name = "equivalent"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
+
 [[package]]
 name = "errno"
 version = "0.3.14"
@@ -360,12 +415,34 @@ dependencies = [
  "wasip2",
 ]
 
+[[package]]
+name = "hashbrown"
+version = "0.14.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
+dependencies = [
+ "ahash",
+ "allocator-api2",
+]
+
+[[package]]
+name = "hashbrown"
+version = "0.16.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
+
 [[package]]
 name = "heck"
 version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
 
+[[package]]
+name = "hifijson"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0a7763b98ba8a24f59e698bf9ab197e7676c640d6455d1580b4ce7dc560f0f0d"
+
 [[package]]
 name = "http"
 version = "1.4.0"
@@ -508,6 +585,16 @@ dependencies = [
  "icu_properties",
 ]
 
+[[package]]
+name = "indexmap"
+version = "2.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7714e70437a7dc3ac8eb7e6f8df75fd8eb422675fc7678aff7364301092b1017"
+dependencies = [
+ "equivalent",
+ "hashbrown 0.16.1",
+]
+
 [[package]]
 name = "is_terminal_polyfill"
 version = "1.70.2"
@@ -520,6 +607,66 @@ version = "1.0.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2"
 
+[[package]]
+name = "jaq-core"
+version = "1.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d6fda09ee08c84c81293fdf811d9ebaa87b327557b5391f290c926d728c2ddd4"
+dependencies = [
+ "aho-corasick",
+ "base64",
+ "chrono",
+ "hifijson",
+ "jaq-interpret",
+ "libm",
+ "log",
+ "regex",
+ "urlencoding",
+]
+
+[[package]]
+name = "jaq-interpret"
+version = "1.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2fe95ec3c24af3fd9f3dd1091593f5e49b003a66c496a8aa39d764d0a06ae17b"
+dependencies = [
+ "ahash",
+ "dyn-clone",
+ "hifijson",
+ "indexmap",
+ "jaq-syn",
+ "once_cell",
+ "serde_json",
+]
+
+[[package]]
+name = "jaq-parse"
+version = "1.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0346d7d3146cdda8acd929581f3d6626a332356c74d5c95aeaffaac2eb6dee82"
+dependencies = [
+ "chumsky",
+ "jaq-syn",
+]
+
+[[package]]
+name = "jaq-std"
+version = "1.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bfbaa55578fd3b70433b594a370741e0c364e4afff92cc0099623fce87311bc1"
+dependencies = [
+ "jaq-syn",
+]
+
+[[package]]
+name = "jaq-syn"
+version = "1.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ba44fe4428c71304604261ecbae047ee9cfb60c4f1a6bd222ebbb31726d3948"
+dependencies = [
+ "serde",
+]
+
 [[package]]
 name = "js-sys"
 version = "0.3.85"
@@ -542,6 +689,12 @@ version = "0.2.180"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bcc35a38544a891a5f7c865aca548a982ccb3b8650a5b06d0fd33a10283c56fc"
 
+[[package]]
+name = "libm"
+version = "0.2.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"
+
 [[package]]
 name = "linux-raw-sys"
 version = "0.11.0"
@@ -569,6 +722,12 @@ version = "0.4.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897"
 
+[[package]]
+name = "memchr"
+version = "2.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f52b00d39961fc5b2736ea853c9cc86238e165017a493d1d5c8eac6bdc4cc273"
+
 [[package]]
 name = "mio"
 version = "1.1.1"
@@ -799,6 +958,35 @@ dependencies = [
  "bitflags",
 ]
 
+[[package]]
+name = "regex"
+version = "1.12.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e10754a14b9137dd7b1e3e5b0493cc9171fdd105e0ab477f51b72e7f3ac0e276"
+dependencies = [
+ "aho-corasick",
+ "memchr",
+ "regex-automata",
+ "regex-syntax",
+]
+
+[[package]]
+name = "regex-automata"
+version = "0.4.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6e1dd4122fc1595e8162618945476892eefca7b88c52820e74af6262213cae8f"
+dependencies = [
+ "aho-corasick",
+ "memchr",
+ "regex-syntax",
+]
+
+[[package]]
+name = "regex-syntax"
+version = "0.8.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a96887878f22d7bad8a3b6dc5b7440e0ada9a245242924394987b21cf2210a4c"
+
 [[package]]
 name = "rustix"
 version = "1.1.3"
@@ -863,6 +1051,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e"
 dependencies = [
  "serde_core",
+ "serde_derive",
 ]
 
 [[package]]
@@ -885,6 +1074,19 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "serde_json"
+version = "1.0.149"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "83fc039473c5595ace860d8c4fafa220ff474b3fc6bfdb4293327f1a37e94d86"
+dependencies = [
+ "itoa",
+ "memchr",
+ "serde",
+ "serde_core",
+ "zmij",
+]
+
 [[package]]
 name = "sha1"
 version = "0.10.6"
@@ -1182,6 +1384,12 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "urlencoding"
+version = "2.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"
+
 [[package]]
 name = "utf-8"
 version = "0.7.6"
@@ -1285,10 +1493,16 @@ dependencies = [
  "chrono",
  "clap",
  "futures-util",
+ "jaq-core",
+ "jaq-interpret",
+ "jaq-parse",
+ "jaq-std",
+ "serde_json",
  "tokio",
  "tokio-tungstenite",
  "tracing",
  "tracing-subscriber",
+ "url",
 ]
 
 [[package]]
@@ -1541,3 +1755,9 @@ dependencies = [
  "quote",
  "syn",
 ]
+
+[[package]]
+name = "zmij"
+version = "1.0.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3ff05f8caa9038894637571ae6b9e29466c1f4f829d26c9b28f869a29cbe3445"

Cargo.toml

@@ -11,3 +11,9 @@ tracing-subscriber = { version = "0.3", features = ["fmt"] }
 clap = { version = "4", features = ["derive"] }
 futures-util = "0.3"
 chrono = "0.4"
+url = "2"
+jaq-core = "1"
+jaq-interpret = "1"
+jaq-parse = "1"
+jaq-std = "1"
+serde_json = "1"

src/cli.rs

@@ -0,0 +1,30 @@
+use clap::Parser;
+
+#[derive(Parser)]
+#[command(name = "websocket-debug")]
+#[command(about = "A WebSocket debugging tool that logs and saves messages")]
+pub struct Args {
+    /// WebSocket URLs to connect to (e.g., ws://localhost:8080 or wss://example.com/ws)
+    #[arg(required = true)]
+    pub urls: Vec<String>,
+
+    /// Bearer token for Authorization header
+    #[arg(long)]
+    pub bearer_token: Option<String>,
+
+    /// Enable debug logging (shows request/response headers)
+    #[arg(long)]
+    pub debug: bool,
+
+    /// Query string parameters to add to all URLs (pre-encoded, e.g., "name=First%20Last&key=value")
+    #[arg(short = 'q', long = "query-string-all")]
+    pub query_string_all: Option<String>,
+
+    /// jq expression(s) to evaluate on JSON text messages for logging (can be specified multiple times)
+    #[arg(short = 'j', long = "jaq")]
+    pub jaq: Vec<String>,
+
+    /// Normalize JSON messages: save as .json with pretty-printing and sorted keys (for easier diffing)
+    #[arg(short = 'n', long = "json-normalize")]
+    pub json_normalize: bool,
+}

src/connection.rs

@@ -0,0 +1,165 @@
+use tokio_tungstenite::{
+    connect_async_with_config, MaybeTlsStream, WebSocketStream,
+    tungstenite::{
+        client::IntoClientRequest,
+        http::header::{HeaderValue, AUTHORIZATION, USER_AGENT},
+    },
+};
+use tracing::{debug, info};
+use url::{form_urlencoded, Url};
+
+/// Process URLs by adding extra query parameters if specified.
+pub fn process_urls(
+    urls: &[String],
+    query_string_all: Option<&String>,
+) -> Result<Vec<String>, url::ParseError> {
+    let extra_params: Vec<(String, String)> = query_string_all
+        .map(|qs| {
+            form_urlencoded::parse(qs.as_bytes())
+                .map(|(k, v)| (k.into_owned(), v.into_owned()))
+                .collect()
+        })
+        .unwrap_or_default();
+
+    let mut processed_urls = Vec::new();
+    for url_str in urls {
+        let mut url = Url::parse(url_str)?;
+
+        if !extra_params.is_empty() {
+            let existing: Vec<(String, String)> = url
+                .query_pairs()
+                .map(|(k, v)| (k.into_owned(), v.into_owned()))
+                .collect();
+
+            let mut query_pairs = url.query_pairs_mut();
+            query_pairs.clear();
+            for (k, v) in &existing {
+                query_pairs.append_pair(k, v);
+            }
+            for (k, v) in &extra_params {
+                query_pairs.append_pair(k, v);
+            }
+        }
+
+        processed_urls.push(url.to_string());
+    }
+
+    Ok(processed_urls)
+}
+
+/// Connect to a WebSocket URL with optional bearer token and debug logging.
+pub async fn connect(
+    url: &str,
+    idx: usize,
+    bearer_token: Option<&str>,
+    debug_enabled: bool,
+) -> Result<(char, String, WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>), String> {
+    let letter = (b'A' + idx as u8) as char;
+
+    let mut request = url
+        .into_client_request()
+        .map_err(|e| format!("[{}] Invalid URL {}: {}", letter, url, e))?;
+
+    if let Some(token) = bearer_token {
+        let auth_value = HeaderValue::from_str(&format!("Bearer {}", token))
+            .map_err(|e| format!("Invalid bearer token: {}", e))?;
+        request.headers_mut().insert(AUTHORIZATION, auth_value);
+    }
+
+    let user_agent = format!("{}/{}", env!("CARGO_PKG_NAME"), env!("CARGO_PKG_VERSION"));
+    request
+        .headers_mut()
+        .insert(USER_AGENT, HeaderValue::from_str(&user_agent).unwrap());
+
+    info!("[{}] Connecting to {}", letter, url);
+
+    if debug_enabled {
+        log_request(&request, letter);
+    }
+
+    let (ws_stream, response) = match connect_async_with_config(request, None, false).await {
+        Ok(result) => result,
+        Err(e) => {
+            if debug_enabled {
+                if let tokio_tungstenite::tungstenite::Error::Http(ref response) = e {
+                    log_error_response(response, letter);
+                }
+            }
+            return Err(format!("[{}] Failed to connect to {}: {}", letter, url, e));
+        }
+    };
+
+    info!(
+        "[{}] Connected successfully (status: {})",
+        letter,
+        response.status()
+    );
+
+    if debug_enabled {
+        log_response(&response, letter);
+    }
+
+    Ok((letter, url.to_string(), ws_stream))
+}
+
+fn log_request(request: &tokio_tungstenite::tungstenite::http::Request<()>, letter: char) {
+    debug!(
+        "[{}] Request: {} {}",
+        letter,
+        request.method(),
+        request.uri()
+    );
+    for (name, value) in request.headers() {
+        debug!(
+            "[{}]   {}: {}",
+            letter,
+            name,
+            value.to_str().unwrap_or("<binary>")
+        );
+    }
+}
+
+fn log_response(
+    response: &tokio_tungstenite::tungstenite::http::Response<Option<Vec<u8>>>,
+    letter: char,
+) {
+    debug!(
+        "[{}] Response: {} {}",
+        letter,
+        response.status().as_u16(),
+        response.status().canonical_reason().unwrap_or("")
+    );
+    for (name, value) in response.headers() {
+        debug!(
+            "[{}]   {}: {}",
+            letter,
+            name,
+            value.to_str().unwrap_or("<binary>")
+        );
+    }
+}
+
+fn log_error_response(
+    response: &tokio_tungstenite::tungstenite::http::Response<Option<Vec<u8>>>,
+    letter: char,
+) {
+    debug!(
+        "[{}] Error response: {} {}",
+        letter,
+        response.status().as_u16(),
+        response.status().canonical_reason().unwrap_or("")
+    );
+    for (name, value) in response.headers() {
+        debug!(
+            "[{}]   {}: {}",
+            letter,
+            name,
+            value.to_str().unwrap_or("<binary>")
+        );
+    }
+    if let Some(body) = response.body() {
+        if let Ok(body_str) = std::str::from_utf8(body) {
+            debug!("[{}] Response body: {}", letter, body_str);
+        }
+    }
+}

src/filter.rs

@@ -0,0 +1,79 @@
+use std::sync::Arc;
+
+use jaq_interpret::{Ctx, Filter, FilterT, ParseCtx, RcIter, Val};
+use tracing::{info, warn};
+
+pub type JaqFilters = Arc<Vec<Filter>>;
+
+/// Compile a list of jaq expressions into filters.
+pub fn compile_filters(exprs: &[String]) -> Result<JaqFilters, String> {
+    let mut filters = Vec::new();
+
+    for expr in exprs {
+        let mut defs = ParseCtx::new(Vec::new());
+        defs.insert_natives(jaq_core::core());
+        defs.insert_defs(jaq_std::std());
+
+        let (parsed, errs) = jaq_parse::parse(expr, jaq_parse::main());
+        if !errs.is_empty() {
+            let err_msgs: Vec<String> = errs.iter().map(|e| format!("{:?}", e)).collect();
+            return Err(format!(
+                "Failed to parse jaq expression '{}': {}",
+                expr,
+                err_msgs.join(", ")
+            ));
+        }
+
+        let parsed = parsed.ok_or_else(|| format!("Failed to parse jaq expression '{}'", expr))?;
+        let filter = defs.compile(parsed);
+
+        if !defs.errs.is_empty() {
+            return Err(format!(
+                "Failed to compile jaq expression '{}' ({} error(s))",
+                expr,
+                defs.errs.len()
+            ));
+        }
+
+        info!("Using jaq filter: {}", expr);
+        filters.push(filter);
+    }
+
+    Ok(Arc::new(filters))
+}
+
+/// Evaluate all filters against a JSON value, returning the outputs for each filter.
+pub fn evaluate_filters(
+    filters: &JaqFilters,
+    json_val: &serde_json::Value,
+    letter: char,
+    seq_num: u64,
+) -> Vec<String> {
+    let mut all_outputs = Vec::new();
+
+    for filter in filters.iter() {
+        let inputs = RcIter::new(core::iter::empty());
+        let ctx = Ctx::new([], &inputs);
+        let out = filter.run((ctx, Val::from(json_val.clone())));
+
+        let mut filter_outputs = Vec::new();
+        for result in out {
+            match result {
+                Ok(val) => {
+                    filter_outputs.push(val.to_string());
+                }
+                Err(e) => {
+                    warn!("[{}:{}] jaq error: {}", letter, seq_num, e);
+                }
+            }
+        }
+
+        if filter_outputs.is_empty() {
+            all_outputs.push("(no output)".to_string());
+        } else {
+            all_outputs.push(filter_outputs.join(", "));
+        }
+    }
+
+    all_outputs
+}

src/handler.rs

@@ -0,0 +1,125 @@
+use std::fs;
+use std::path::PathBuf;
+
+use futures_util::StreamExt;
+use tokio::net::TcpStream;
+use tokio_tungstenite::{tungstenite::Message, MaybeTlsStream, WebSocketStream};
+use tracing::{error, info, warn};
+
+use crate::filter::{evaluate_filters, JaqFilters};
+
+/// Configuration for the connection handler.
+#[derive(Clone)]
+pub struct HandlerConfig {
+    pub session_dir: PathBuf,
+    pub jaq_filters: JaqFilters,
+    pub json_normalize: bool,
+}
+
+/// Handle a WebSocket connection, processing messages and writing them to files.
+pub async fn handle_connection(
+    letter: char,
+    url: String,
+    ws_stream: WebSocketStream<MaybeTlsStream<TcpStream>>,
+    config: HandlerConfig,
+) -> (char, String) {
+    let (_, mut read) = ws_stream.split();
+    let mut seq_num: u64 = 0;
+
+    while let Some(message_result) = read.next().await {
+        match message_result {
+            Ok(message) => {
+                match message {
+                    Message::Text(text) => {
+                        handle_text_message(&text, letter, seq_num, &config);
+                    }
+                    Message::Binary(data) => {
+                        handle_binary_message(&data, letter, seq_num, &config);
+                    }
+                    Message::Ping(data) => {
+                        info!("[{}] Ping: {} bytes", letter, data.len());
+                        continue;
+                    }
+                    Message::Pong(data) => {
+                        info!("[{}] Pong: {} bytes", letter, data.len());
+                        continue;
+                    }
+                    Message::Close(frame) => {
+                        if let Some(cf) = frame {
+                            info!("[{}] Connection closed: {} - {}", letter, cf.code, cf.reason);
+                        } else {
+                            info!("[{}] Connection closed", letter);
+                        }
+                        break;
+                    }
+                    Message::Frame(_) => {
+                        continue;
+                    }
+                }
+                seq_num += 1;
+            }
+            Err(e) => {
+                warn!("[{}] Error receiving message: {}", letter, e);
+            }
+        }
+    }
+
+    info!("[{}] Session ended. Received {} messages.", letter, seq_num);
+    (letter, url)
+}
+
+fn handle_text_message(text: &str, letter: char, seq_num: u64, config: &HandlerConfig) {
+    // Log message based on jaq filters
+    if config.jaq_filters.is_empty() {
+        let preview: String = text.chars().take(50).collect();
+        let truncated = if text.len() > 50 { "..." } else { "" };
+        info!("[{}:{}] Text: {}{}", letter, seq_num, preview, truncated);
+    } else {
+        match serde_json::from_str::<serde_json::Value>(text) {
+            Ok(json_val) => {
+                let outputs = evaluate_filters(&config.jaq_filters, &json_val, letter, seq_num);
+                info!("[{}:{}] {}", letter, seq_num, outputs.join(" | "));
+            }
+            Err(e) => {
+                warn!("[{}:{}] JSON parse error: {}", letter, seq_num, e);
+                let preview: String = text.chars().take(50).collect();
+                let truncated = if text.len() > 50 { "..." } else { "" };
+                info!("[{}:{}] Text: {}{}", letter, seq_num, preview, truncated);
+            }
+        }
+    }
+
+    // Write message to file
+    let (filename, content) = if config.json_normalize {
+        if let Ok(json_val) = serde_json::from_str::<serde_json::Value>(text) {
+            let pretty = serde_json::to_string_pretty(&json_val).unwrap_or_else(|_| text.to_string());
+            (
+                config.session_dir.join(format!("{}{}.json", letter, seq_num)),
+                pretty,
+            )
+        } else {
+            (
+                config.session_dir.join(format!("{}{}.txt", letter, seq_num)),
+                text.to_string(),
+            )
+        }
+    } else {
+        (
+            config.session_dir.join(format!("{}{}.txt", letter, seq_num)),
+            text.to_string(),
+        )
+    };
+
+    if let Err(e) = fs::write(&filename, content) {
+        error!("[{}] Failed to write {:?}: {}", letter, filename, e);
+    }
+}
+
+fn handle_binary_message(data: &[u8], letter: char, seq_num: u64, config: &HandlerConfig) {
+    info!("[{}:{}] Binary: {} bytes", letter, seq_num, data.len());
+
+    let filename = config.session_dir.join(format!("{}{}.bin", letter, seq_num));
+    if let Err(e) = fs::write(&filename, data) {
+        error!("[{}] Failed to write {:?}: {}", letter, filename, e);
+    }
+}

src/main.rs

@@ -1,46 +1,27 @@
+mod cli;
+mod connection;
+mod filter;
+mod handler;
+
 use std::fs;
 use std::path::PathBuf;
 
 use chrono::Local;
 use clap::Parser;
-use futures_util::StreamExt;
 use tokio::task::JoinSet;
-use tokio_tungstenite::{
+use tracing::{error, info, Level};
-    connect_async_with_config,
-    tungstenite::{
-        client::IntoClientRequest,
-        http::header::{HeaderValue, AUTHORIZATION, USER_AGENT},
-        Message,
-    },
-};
-use tracing::{debug, error, info, warn, Level};
 
-#[derive(Parser)]
+use cli::Args;
-#[command(name = "websocket-debug")]
+use connection::{connect, process_urls};
-#[command(about = "A WebSocket debugging tool that logs and saves messages")]
+use filter::compile_filters;
-struct Args {
+use handler::{handle_connection, HandlerConfig};
-    /// WebSocket URLs to connect to (e.g., ws://localhost:8080 or wss://example.com/ws)
-    #[arg(required = true)]
-    urls: Vec<String>,
-
-    /// Bearer token for Authorization header
-    #[arg(long)]
-    bearer_token: Option<String>,
-
-    /// Enable debug logging (shows request/response headers)
-    #[arg(long)]
-    debug: bool,
-}
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let args = Args::parse();
 
-    let log_level = if args.debug {
+    // Initialize logging
-        Level::DEBUG
+    let log_level = if args.debug { Level::DEBUG } else { Level::INFO };
-    } else {
-        Level::INFO
-    };
     tracing_subscriber::fmt()
         .with_target(false)
         .with_thread_ids(false)
@@ -53,105 +34,21 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     fs::create_dir_all(&session_dir)?;
     info!("Created session directory: {}", session_dir.display());
 
+    // Compile jaq filters
+    let jaq_filters = compile_filters(&args.jaq)?;
+
+    // Process URLs with extra query parameters
+    let processed_urls = process_urls(&args.urls, args.query_string_all.as_ref())?;
+
     // Connect to all URLs simultaneously
     let mut connect_futures = Vec::new();
-    for (idx, url) in args.urls.iter().enumerate() {
+    for (idx, url) in processed_urls.iter().enumerate() {
-        let letter = (b'A' + idx as u8) as char;
-        let bearer_token = args.bearer_token.clone();
         let url = url.clone();
+        let bearer_token = args.bearer_token.clone();
         let debug_enabled = args.debug;
 
         connect_futures.push(async move {
-            let mut request = url
+            connect(&url, idx, bearer_token.as_deref(), debug_enabled).await
-                .as_str()
-                .into_client_request()
-                .map_err(|e| format!("[{}] Invalid URL {}: {}", letter, url, e))?;
-
-            if let Some(ref token) = bearer_token {
-                let auth_value = HeaderValue::from_str(&format!("Bearer {}", token))
-                    .map_err(|e| format!("Invalid bearer token: {}", e))?;
-                request.headers_mut().insert(AUTHORIZATION, auth_value);
-            }
-
-            let user_agent = format!("{}/{}", env!("CARGO_PKG_NAME"), env!("CARGO_PKG_VERSION"));
-            request
-                .headers_mut()
-                .insert(USER_AGENT, HeaderValue::from_str(&user_agent).unwrap());
-
-            info!("[{}] Connecting to {}", letter, url);
-
-            if debug_enabled {
-                debug!(
-                    "[{}] Request: {} {}",
-                    letter,
-                    request.method(),
-                    request.uri()
-                );
-                for (name, value) in request.headers() {
-                    debug!(
-                        "[{}]   {}: {}",
-                        letter,
-                        name,
-                        value.to_str().unwrap_or("<binary>")
-                    );
-                }
-            }
-
-            let (ws_stream, response) = match connect_async_with_config(request, None, false).await
-            {
-                Ok(result) => result,
-                Err(e) => {
-                    if debug_enabled {
-                        if let tokio_tungstenite::tungstenite::Error::Http(ref response) = e {
-                            debug!(
-                                "[{}] Error response: {} {}",
-                                letter,
-                                response.status().as_u16(),
-                                response.status().canonical_reason().unwrap_or("")
-                            );
-                            for (name, value) in response.headers() {
-                                debug!(
-                                    "[{}]   {}: {}",
-                                    letter,
-                                    name,
-                                    value.to_str().unwrap_or("<binary>")
-                                );
-                            }
-                            if let Some(body) = response.body() {
-                                if let Ok(body_str) = std::str::from_utf8(body) {
-                                    debug!("[{}] Response body: {}", letter, body_str);
-                                }
-                            }
-                        }
-                    }
-                    return Err(format!("[{}] Failed to connect to {}: {}", letter, url, e));
-                }
-            };
-
-            info!(
-                "[{}] Connected successfully (status: {})",
-                letter,
-                response.status()
-            );
-
-            if debug_enabled {
-                debug!(
-                    "[{}] Response: {} {}",
-                    letter,
-                    response.status().as_u16(),
-                    response.status().canonical_reason().unwrap_or("")
-                );
-                for (name, value) in response.headers() {
-                    debug!(
-                        "[{}]   {}: {}",
-                        letter,
-                        name,
-                        value.to_str().unwrap_or("<binary>")
-                    );
-                }
-            }
-
-            Ok::<_, String>((letter, url, ws_stream))
         });
     }
 
@@ -171,74 +68,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     info!("All {} connections established", connections.len());
 
-    // Spawn tasks for each connection
+    // Spawn handler tasks for each connection
     let mut join_set: JoinSet<(char, String)> = JoinSet::new();
 
     for (letter, url, ws_stream) in connections {
-        let session_dir = session_dir.clone();
+        let config = HandlerConfig {
+            session_dir: session_dir.clone(),
+            jaq_filters: jaq_filters.clone(),
+            json_normalize: args.json_normalize,
+        };
 
-        join_set.spawn(async move {
+        join_set.spawn(handle_connection(letter, url, ws_stream, config));
-            let (_, mut read) = ws_stream.split();
-            let mut seq_num: u64 = 0;
-
-            while let Some(message_result) = read.next().await {
-                match message_result {
-                    Ok(message) => {
-                        match message {
-                            Message::Text(text) => {
-                                let preview: String = text.chars().take(50).collect();
-                                let truncated = if text.len() > 50 { "..." } else { "" };
-                                info!("[{}:{}] Text: {}{}", letter, seq_num, preview, truncated);
-
-                                let filename =
-                                    session_dir.join(format!("{}{}.txt", letter, seq_num));
-                                if let Err(e) = fs::write(&filename, &text) {
-                                    error!("[{}] Failed to write {:?}: {}", letter, filename, e);
-                                }
-                            }
-                            Message::Binary(data) => {
-                                info!("[{}:{}] Binary: {} bytes", letter, seq_num, data.len());
-
-                                let filename =
-                                    session_dir.join(format!("{}{}.bin", letter, seq_num));
-                                if let Err(e) = fs::write(&filename, &data) {
-                                    error!("[{}] Failed to write {:?}: {}", letter, filename, e);
-                                }
-                            }
-                            Message::Ping(data) => {
-                                info!("[{}] Ping: {} bytes", letter, data.len());
-                                continue;
-                            }
-                            Message::Pong(data) => {
-                                info!("[{}] Pong: {} bytes", letter, data.len());
-                                continue;
-                            }
-                            Message::Close(frame) => {
-                                if let Some(cf) = frame {
-                                    info!(
-                                        "[{}] Connection closed: {} - {}",
-                                        letter, cf.code, cf.reason
-                                    );
-                                } else {
-                                    info!("[{}] Connection closed", letter);
-                                }
-                                break;
-                            }
-                            Message::Frame(_) => {
-                                continue;
-                            }
-                        }
-                        seq_num += 1;
-                    }
-                    Err(e) => {
-                        warn!("[{}] Error receiving message: {}", letter, e);
-                    }
-                }
-            }
-
-            info!("[{}] Session ended. Received {} messages.", letter, seq_num);
-            (letter, url)
-        });
     }
 
     // Wait for all connections to finish