From c3a6a3b661c1a2aa8c8c1004df6aae7ede1f20c7 Mon Sep 17 00:00:00 2001 From: Sheng Kun Chang Date: Sun, 5 Jul 2026 00:17:30 +0800 Subject: [PATCH] feat(dev): bring run_inner to parity with dev-runner.sh; log shipper on the live path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The compose runner ran scripts/dev-runner.sh (air hot-reload, dev-proxy, livereload, per-module token files — but no log shipper), while the Rust run_inner had the #36 shipper but none of the dev features, so it was never the live path and the developer Logcat stayed empty. run_inner now carries the full runner: - Polling hot-reload (new reload.rs): mtime scan of .go/.sql every 2s (excludes tmp/web/node_modules — bind mounts deliver no inotify), go build -buildvcs=false -o / ., restart on change, keep the old process on a failed rebuild, and re-tap the new child's stdout/stderr into the forwarder + shipper on every restart. - In-process /_m//* dev-proxy (new proxy.rs) on PROXY_PORT (default 8080): TrimPrefix semantics (empty→/, query preserved), raw duplex relay after the rewritten head so chunked/SSE/upgrades pass through; backend connect refused → 502 (startup race). - Deterministic PORT from 18080 in go.work order; livereload LR_PORT from 8089 per web-enabled module. - Per-module MS_PLATFORM_TOKEN_FILE=/.ms-platform-token- (distinct token per tunnel; a shared file would 401 all but the first). - esbuild web watcher: npm install --silent then node esbuild.config.mjs --watch, output prefixed [:web]. - --watch is honored (was dead surface: run_inner(root, _args)); defaults on so the compose command stays `mirrorstack dev --all`, --watch=false gives the old one-shot behavior. - One-line warning when MS_DISPATCH_URL/MS_INTERNAL_SECRET are unset so a config-disabled Logcat is not silent. - Shutdown tears down supervisors, module children, and esbuild watchers, then lets the shipper threads flush their final batch. log_shipper.rs is untouched — putting run_inner on the live path is the point of mirrorstack-cli#38. Closes #38 Co-Authored-By: Claude Fable 5 --- src/commands/dev/mod.rs | 450 ++++++++++++++++++++++++++++++------- src/commands/dev/proxy.rs | 244 ++++++++++++++++++++ src/commands/dev/reload.rs | 108 +++++++++ 3 files changed, 723 insertions(+), 79 deletions(-) create mode 100644 src/commands/dev/proxy.rs create mode 100644 src/commands/dev/reload.rs diff --git a/src/commands/dev/mod.rs b/src/commands/dev/mod.rs index 0ca21bc..42f3098 100644 --- a/src/commands/dev/mod.rs +++ b/src/commands/dev/mod.rs @@ -3,8 +3,15 @@ //! Two modes: //! - **Outer** (host, default): `docker compose up` — all services //! including Go modules run inside Docker. -//! - **Inner** (`--all`, inside runner container): spawn `go run` -//! per module directly. This is what the compose runner calls. +//! - **Inner** (`--all`, inside runner container): the full dev runner. +//! Per module: build + run with polling hot-reload (see `reload`), +//! deterministic internal ports from 18080 (go.work order), a +//! per-module platform-token file, esbuild web watchers with +//! livereload ports from 8089, and the #36 log shipper tapping +//! stdout/stderr. A `/_m//*` reverse proxy (see `proxy`) +//! multiplexes them on one port. This is what the compose runner +//! calls; it replaces ms-app-modules' scripts/dev-runner.sh + +//! dev-proxy.go. //! //! Tunnel registration always happens on the host side (outer mode). //! @@ -21,12 +28,14 @@ //! legacy InternalAuth fallback enforcing (not bypassing) while //! dispatch round-trips the value. Backward-compat with older SDKs. +use std::collections::HashMap; use std::io::{BufRead, BufReader, IsTerminal, Read}; use std::path::{Path, PathBuf}; use std::process::{Child, Command, Stdio}; -use std::sync::mpsc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, mpsc}; use std::thread; -use std::time::Duration; +use std::time::{Duration, Instant}; use anyhow::{Context, Result, anyhow}; use base64::Engine; @@ -44,6 +53,8 @@ use crate::{api, credentials, http}; mod log_shipper; pub(crate) mod module_meta; +mod proxy; +mod reload; mod tunnel; mod workspace; @@ -61,14 +72,50 @@ pub struct DevArgs { /// Run all registered modules directly (used inside Docker runner). #[arg(long)] all: bool, - /// Enable file watching with hot reload (used with --all). - #[arg(long)] + /// Watch module sources and rebuild/restart on change (used with + /// --all). On by default so the compose runner hot-reloads; pass + /// --watch=false for a one-shot run. + #[arg( + long, + num_args = 0..=1, + default_value_t = true, + default_missing_value = "true", + action = clap::ArgAction::Set + )] watch: bool, } const DEFAULT_LOCAL_URL: &str = "http://localhost"; const DEFAULT_MODULE_PORT: u16 = 9080; +// Inner-runner port layout, matching the retired dev-runner.sh: modules +// bind sequentially from 18080 in go.work order, esbuild livereload +// servers from 8089 (per web-enabled module), and the dev-proxy +// multiplexes them on 8080 (host-published as 9080/9089). +const INTERNAL_PORT_BASE: u16 = 18080; +const LR_PORT_BASE: u16 = 8089; +const PROXY_PORT_DEFAULT: u16 = 8080; + +/// Route prefix multiplexing modules on one port. The proxy parses it off +/// incoming targets and tunnel registration embeds it in each module's +/// local_url — dispatch forwards to exactly what was registered, so the +/// two must agree. +const MODULE_ROUTE_PREFIX: &str = "/_m/"; + +/// Per-module platform-token file. The name is a host↔container contract: +/// the outer run writes it next to go.work, and the inner runner points +/// each module's MS_PLATFORM_TOKEN_FILE at it through the `.:/modules` +/// bind mount. +fn platform_token_file(root: &Path, slug: &str) -> PathBuf { + root.join(format!(".ms-platform-token-{slug}")) +} + +/// How often the supervisor polls module sources for changes — matches the +/// shell runner's air `poll_interval = 2000`. +const REBUILD_POLL: Duration = Duration::from_millis(2000); +/// How often the supervisor checks the stop flag and child liveness. +const SUPERVISE_TICK: Duration = Duration::from_millis(200); + pub fn run(args: DevArgs) -> Result<()> { let cwd = args .dir @@ -141,8 +188,8 @@ fn run_outer(root: &Path, args: &DevArgs) -> Result<()> { // registered first and 401'd every other module on every // platform-initiated call (lifecycle install, manifest read, dev-tunnel // API). The files land in `root` and reach the runner through the - // `.:/modules` bind mount; dev-runner.sh points each module's - // MS_PLATFORM_TOKEN_FILE at `.ms-platform-token-`. + // `.:/modules` bind mount; the inner runner (`run_inner`) points each + // module's MS_PLATFORM_TOKEN_FILE at `.ms-platform-token-`. let mut token_files: Vec = Vec::new(); // Per-session MS_INTERNAL_SECRET. Sent on each register frame (so @@ -166,7 +213,7 @@ fn run_outer(root: &Path, args: &DevArgs) -> Result<()> { // `open_tunnels` pushes handles in `ready` order, so zip is aligned. for (m, handle) in ready.iter().zip(state.0.iter()) { let slug = m.dir.file_name().unwrap().to_string_lossy(); - let f = root.join(format!(".ms-platform-token-{slug}")); + let f = platform_token_file(root, &slug); std::fs::write(&f, &handle.service_token) .with_context(|| format!("dev: write platform token file for {slug}"))?; eprintln!( @@ -190,9 +237,9 @@ fn run_outer(root: &Path, args: &DevArgs) -> Result<()> { .stdout(Stdio::inherit()) .stderr(Stdio::inherit()); - // MS_PLATFORM_TOKEN_FILE is set per-module by dev-runner.sh (each module - // reads its own `.ms-platform-token-`). MS_INTERNAL_SECRET is - // injected once on the compose env; the runner forwards it to every + // MS_PLATFORM_TOKEN_FILE is set per-module by the inner runner (each + // module reads its own `.ms-platform-token-`). MS_INTERNAL_SECRET + // is injected once on the compose env; the runner forwards it to every // module process as the SDK's legacy InternalAuth fallback. if let Some(secret) = &internal_secret { compose.env("MS_INTERNAL_SECRET", secret); @@ -234,13 +281,31 @@ fn run_outer(root: &Path, args: &DevArgs) -> Result<()> { // ── Inner mode: inside Docker, run modules directly ───────────────── -fn run_inner(root: &Path, _args: &DevArgs) -> Result<()> { +/// Everything one module's supervisor thread needs to build, run, watch, +/// and restart its process. +struct ModuleSpec { + slug: String, + /// Absolute module directory (build cwd and process cwd). + dir: PathBuf, + /// Where `go build -o` drops the binary (temp dir, per slug). + bin: PathBuf, + /// Env set on the module process (PORT, MS_PLATFORM_TOKEN_FILE, infra). + envs: Vec<(String, String)>, + /// #36 log shipper sink; None → terminal only. + sink: Option, + watch: bool, + stop: Arc, +} + +fn run_inner(root: &Path, args: &DevArgs) -> Result<()> { let all_modules = workspace::discover_modules(root)?; + // Ready modules carry their platform id so the shipper sink below + // doesn't have to re-parse main.go per module. let mut ready = Vec::new(); for m in &all_modules { match module_meta::read_module_meta(&m.abs_dir) { - Ok(meta) if !meta.id.is_empty() => ready.push(m.clone()), + Ok(meta) if !meta.id.is_empty() => ready.push((m.clone(), meta.id)), _ => { eprintln!("{} skipping {} (no ID)", warn_prefix(), m.dir.display()); } @@ -262,31 +327,61 @@ fn run_inner(root: &Path, _args: &DevArgs) -> Result<()> { // disabled and logs just stay in the terminal. let log_dispatch_url = std::env::var("MS_DISPATCH_URL").unwrap_or_default(); let log_secret = std::env::var("MS_INTERNAL_SECRET").unwrap_or_default(); + if log_dispatch_url.is_empty() || log_secret.is_empty() { + eprintln!( + "{} log shipping disabled (MS_DISPATCH_URL / MS_INTERNAL_SECRET unset) — the dev console Logcat will stay empty", + warn_prefix() + ); + } let log_client = http::client(Duration::from_secs(5)).ok(); eprintln!( - "{} starting {} {}", + "{} starting {} {}{}", ok_mark(), ready.len(), if ready.len() == 1 { "module" } else { "modules" - } + }, + if args.watch { " (hot-reload)" } else { "" } ); - let mut children: Vec<(String, Child)> = Vec::new(); + let stop = Arc::new(AtomicBool::new(false)); + let web_children: Arc>> = Arc::new(Mutex::new(Vec::new())); + let mut supervisors = Vec::new(); + let mut routes: HashMap = HashMap::new(); + let mut lr_port = LR_PORT_BASE; - for m in &ready { + for (i, (m, module_id)) in ready.iter().enumerate() { let slug = m.dir.file_name().unwrap().to_string_lossy().to_string(); + // Deterministic internal port: go.work order, from 18080. + let port = INTERNAL_PORT_BASE + u16::try_from(i).expect("module count fits u16"); + routes.insert(slug.clone(), port); - let mut cmd = Command::new("go"); - cmd.args(["run", &format!("./{}", m.dir.display())]) - .current_dir(root) - .env("MS_LOCAL_DB_URL", &db_url) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()); + let sink = log_client.as_ref().and_then(|client| { + log_shipper::spawn( + client.clone(), + log_dispatch_url.clone(), + module_id.clone(), + log_secret.clone(), + ) + }); + // Each module reads its OWN per-session platform token — dispatch + // mints a distinct service token per tunnel, so a shared file would + // authenticate only the first module and 401 the rest. The files + // are written by the outer run (host) and arrive via the bind mount. + let token_file = platform_token_file(root, &slug); + + let mut envs: Vec<(String, String)> = vec![ + ("MS_LOCAL_DB_URL".into(), db_url.clone()), + ("PORT".into(), port.to_string()), + ( + "MS_PLATFORM_TOKEN_FILE".into(), + token_file.display().to_string(), + ), + ]; // Pass through env vars from compose (MS_INTERNAL_SECRET is the // per-session secret minted by the outer run; the rest are infra). for var in [ @@ -298,48 +393,116 @@ fn run_inner(root: &Path, _args: &DevArgs) -> Result<()> { "AWS_SECRET_ACCESS_KEY", ] { if let Ok(val) = std::env::var(var) { - cmd.env(var, val); + envs.push((var.into(), val)); } } - let mut child = cmd - .spawn() - .with_context(|| format!("dev: spawn module {slug}"))?; + eprintln!(" {} {} on internal :{}", style("✓").green(), slug, port); - let stdout = child.stdout.take().unwrap(); - let stderr = child.stderr.take().unwrap(); - let label: &'static str = Box::leak(slug.clone().into_boxed_str()); - let sink = match (&log_client, module_meta::read_module_id(&m.abs_dir)) { - (Some(client), Ok(module_id)) => log_shipper::spawn( - client.clone(), - log_dispatch_url.clone(), - module_id, - log_secret.clone(), - ), - _ => None, + let spec = ModuleSpec { + slug: slug.clone(), + dir: m.abs_dir.clone(), + bin: std::env::temp_dir().join(format!("ms-dev-{slug}")), + envs, + sink, + watch: args.watch, + stop: stop.clone(), }; - spawn_forwarder(stdout, label, false, sink.clone()); - spawn_forwarder(stderr, label, true, sink); + supervisors.push(thread::spawn(move || supervise_module(spec))); - eprintln!(" {} {}", style("✓").green(), slug); - children.push((slug, child)); + if m.abs_dir.join("web/esbuild.config.mjs").exists() { + eprintln!( + " {} {} web watcher (esbuild + livereload :{})", + style("✓").green(), + slug, + lr_port + ); + spawn_web_watcher( + slug, + m.abs_dir.join("web"), + lr_port, + web_children.clone(), + stop.clone(), + ); + lr_port += 1; + } } - // Wait for Ctrl-C + // The /_m/ multiplexer on the runner's single exposed port. + let proxy_port = std::env::var("PROXY_PORT") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(PROXY_PORT_DEFAULT); + let route_count = routes.len(); + let proxy_port = proxy::spawn(proxy_port, routes)?; + eprintln!( + "{} dev-proxy listening on :{} ({} routes)", + ok_mark(), + proxy_port, + route_count + ); + + // Wait for Ctrl-C, or for every supervisor to finish (a supervisor + // only returns on its own in one-shot mode, after its module exits). let (tx, rx) = mpsc::channel::<()>(); ctrlc::set_handler(move || { let _ = tx.send(()); }) .context("dev: install ctrl-c handler")?; - let mut exited = vec![false; children.len()]; loop { - for (i, (label, child)) in children.iter_mut().enumerate() { - if exited[i] { - continue; - } - if let Ok(Some(status)) = child.try_wait() { - exited[i] = true; + if supervisors.iter().all(|h| h.is_finished()) { + break; + } + match rx.recv_timeout(SUPERVISE_TICK) { + Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => break, + Err(mpsc::RecvTimeoutError::Timeout) => continue, + } + } + + // Tear down the whole tree: supervisors kill their module children, + // then the esbuild watchers go. + stop.store(true, Ordering::SeqCst); + for h in supervisors { + let _ = h.join(); + } + for child in web_children.lock().unwrap().iter_mut() { + kill_wait(child); + } + // All shipper senders die with the supervisors and the forwarder + // threads (EOF on the killed children's pipes); give the shipper + // threads a beat to hit their Disconnected branch and flush the + // final batch before the process exits. + thread::sleep(Duration::from_millis(300)); + + Ok(()) +} + +/// Own one module's build → run → watch → restart loop. +/// +/// Mirrors the air config the shell runner generated: build with +/// `go build -buildvcs=false -o .`, run the binary from the module +/// dir, poll sources every 2s (see `reload` for why polling), rebuild and +/// restart on change, and keep the old process running when a rebuild +/// fails. Every (re)start re-taps stdout/stderr into the forwarder + +/// shipper so logs keep flowing across restarts. +fn supervise_module(spec: ModuleSpec) { + let label: &'static str = Box::leak(spec.slug.clone().into_boxed_str()); + let mut sig = spec.watch.then(|| reload::scan(&spec.dir)); + let mut child = if build_module(&spec, label) { + start_module(&spec, label) + } else { + None + }; + let mut last_scan = Instant::now(); + + loop { + if spec.stop.load(Ordering::SeqCst) { + break; + } + + if let Some(c) = child.as_mut() { + if let Ok(Some(status)) = c.try_wait() { eprintln!( "{} module {} exited ({})", warn_prefix(), @@ -349,34 +512,158 @@ fn run_inner(root: &Path, _args: &DevArgs) -> Result<()> { .map(|c| c.to_string()) .unwrap_or_else(|| "signal".into()) ); + child = None; } } - if exited.iter().all(|e| *e) { - break; + // One-shot mode ends with the module; watch mode keeps polling so + // the next edit revives a crashed (or never-built) module. + if !spec.watch && child.is_none() { + return; } - match rx.recv_timeout(Duration::from_millis(200)) { - Ok(()) => { - for (_, child) in &mut children { - let _ = child.kill(); - } - for (_, child) in &mut children { - let _ = child.wait(); + if let Some(prev) = sig.as_mut() { + if last_scan.elapsed() >= REBUILD_POLL { + last_scan = Instant::now(); + let next = reload::scan(&spec.dir); + if *prev != next { + *prev = next; + eprintln!("{} {} changed — rebuilding…", ok_mark(), label); + if build_module(&spec, label) { + if let Some(mut old) = child.take() { + kill_wait(&mut old); + } + child = start_module(&spec, label); + } } - break; } - Err(mpsc::RecvTimeoutError::Timeout) => continue, - Err(mpsc::RecvTimeoutError::Disconnected) => { - for (_, child) in &mut children { - let _ = child.wait(); - } - break; + } + + thread::sleep(SUPERVISE_TICK); + } + + if let Some(mut c) = child { + kill_wait(&mut c); + } +} + +/// Kill a child process and reap it (kill alone would leave a zombie). +fn kill_wait(c: &mut Child) { + let _ = c.kill(); + let _ = c.wait(); +} + +/// `go build -buildvcs=false -o .` in the module dir. Compile errors +/// are mirrored to the terminal and shipped like module stderr so build +/// failures show up in the dev console Logcat too. +fn build_module(spec: &ModuleSpec, label: &str) -> bool { + let out = Command::new("go") + .args(["build", "-buildvcs=false", "-o"]) + .arg(&spec.bin) + .arg(".") + .current_dir(&spec.dir) + .output(); + match out { + Ok(out) if out.status.success() => true, + Ok(out) => { + let prefix = line_prefix(label); + let text = format!( + "{}{}", + String::from_utf8_lossy(&out.stderr), + String::from_utf8_lossy(&out.stdout) + ); + for line in text.lines().filter(|l| !l.trim().is_empty()) { + forward_line(&prefix, line, true, &spec.sink); } + false + } + Err(e) => { + eprintln!("{} {label}: go build: {e}", warn_prefix()); + false } } +} - Ok(()) +/// Spawn the built module binary and tap its stdout/stderr into the +/// forwarder (terminal prefix + shipper sink). +fn start_module(spec: &ModuleSpec, label: &'static str) -> Option { + let mut cmd = Command::new(&spec.bin); + cmd.current_dir(&spec.dir) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + for (k, v) in &spec.envs { + cmd.env(k, v); + } + match cmd.spawn() { + Ok(mut child) => { + spawn_forwarder( + child.stdout.take().unwrap(), + label, + false, + spec.sink.clone(), + ); + spawn_forwarder(child.stderr.take().unwrap(), label, true, spec.sink.clone()); + Some(child) + } + Err(e) => { + eprintln!("{} spawn {label}: {e}", warn_prefix()); + None + } + } +} + +/// `npm install --silent` then `node esbuild.config.mjs --watch` with +/// LR_PORT, mirroring the shell runner. The module's own esbuild config +/// hosts the SSE livereload server — the runner only sets LR_PORT. Output +/// is prefixed `[:web]` and stays terminal-only (the shell runner +/// never shipped web-watcher lines either). +fn spawn_web_watcher( + slug: String, + web_dir: PathBuf, + lr_port: u16, + children: Arc>>, + stop: Arc, +) { + thread::spawn(move || { + let install = Command::new("npm") + .args(["install", "--silent"]) + .current_dir(&web_dir) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); + if !matches!(install, Ok(s) if s.success()) { + eprintln!( + "{} {slug}: npm install failed — starting esbuild anyway", + warn_prefix() + ); + } + if stop.load(Ordering::SeqCst) { + return; + } + let label: &'static str = Box::leak(format!("{slug}:web").into_boxed_str()); + let spawned = Command::new("node") + .args(["esbuild.config.mjs", "--watch"]) + .current_dir(&web_dir) + .env("LR_PORT", lr_port.to_string()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn(); + match spawned { + Ok(mut child) => { + spawn_forwarder(child.stdout.take().unwrap(), label, false, None); + spawn_forwarder(child.stderr.take().unwrap(), label, true, None); + let mut kids = children.lock().unwrap(); + kids.push(child); + // Shutdown may have raced npm install; don't leave an orphan. + if stop.load(Ordering::SeqCst) { + for c in kids.iter_mut() { + let _ = c.kill(); + } + } + } + Err(e) => eprintln!("{} {slug}: node esbuild: {e}", warn_prefix()), + } + }); } fn spawn_forwarder( @@ -395,16 +682,7 @@ fn spawn_forwarder( Ok(0) => return, Ok(_) => { let trimmed = line.trim_end_matches(['\n', '\r']); - // Ship to the developer console (best-effort), then mirror to - // the terminal as before. - if let Some(s) = &sink { - let _ = s.send(log_shipper::parse_line(trimmed, is_stderr)); - } - if is_stderr { - eprintln!("{prefix} {trimmed}"); - } else { - println!("{prefix} {trimmed}"); - } + forward_line(&prefix, trimmed, is_stderr, &sink); } Err(_) => return, } @@ -412,6 +690,20 @@ fn spawn_forwarder( }); } +/// Ship one line to the developer console (best-effort), then mirror it to +/// the terminal as before. Shared by the live-process forwarders and +/// `build_module`'s compile-error path so both feed the same pipeline. +fn forward_line(prefix: &str, line: &str, is_stderr: bool, sink: &Option) { + if let Some(s) = sink { + let _ = s.send(log_shipper::parse_line(line, is_stderr)); + } + if is_stderr { + eprintln!("{prefix} {line}"); + } else { + println!("{prefix} {line}"); + } +} + fn line_prefix(label: &str) -> String { if std::io::stderr().is_terminal() { format!("{} {}", style(label).cyan().dim(), style("│").dim()) @@ -458,7 +750,7 @@ fn open_tunnels( for m in modules { let module_id = module_meta::read_module_id(&m.abs_dir)?; let slug = m.dir.file_name().unwrap().to_string_lossy(); - let module_local_url = format!("{local_url}/_m/{slug}"); + let module_local_url = format!("{local_url}{MODULE_ROUTE_PREFIX}{slug}"); eprintln!( "{} fetching tunnel token for {} from {}", diff --git a/src/commands/dev/proxy.rs b/src/commands/dev/proxy.rs new file mode 100644 index 0000000..295926b --- /dev/null +++ b/src/commands/dev/proxy.rs @@ -0,0 +1,244 @@ +//! In-process dev reverse proxy for `dev --all` — replaces the shell +//! runner's `scripts/dev-proxy.go`. +//! +//! Routes `/_m//*` → `127.0.0.1:/*` by stripping the +//! `/_m/` prefix (empty result → `/`, query preserved), reproducing +//! dev-proxy.go's `TrimPrefix` semantics. After the rewritten request head +//! is sent, bytes are relayed raw in both directions, so chunked bodies, +//! SSE streams, and protocol upgrades pass through untouched. A module +//! that is still compiling simply refuses the backend connect → 502, the +//! same startup race the shell runner tolerated. + +use std::collections::HashMap; +use std::io::{Read, Write}; +use std::net::{Shutdown, TcpListener, TcpStream}; +use std::sync::Arc; +use std::thread; + +use anyhow::{Context, Result}; + +const MAX_HEAD: usize = 64 * 1024; + +/// Bind the proxy and serve on a background thread. Returns the bound port +/// (`port` 0 picks an ephemeral one, used by tests). +pub(super) fn spawn(port: u16, routes: HashMap) -> Result { + let listener = TcpListener::bind(("0.0.0.0", port)) + .with_context(|| format!("dev: bind dev-proxy on :{port}"))?; + let bound = listener.local_addr().context("dev: dev-proxy local addr")?; + // The table is immutable after spawn and every request is a fresh + // connection (`Connection: close` below), so share it via Arc rather + // than deep-cloning it per connection. + let routes = Arc::new(routes); + thread::spawn(move || { + for stream in listener.incoming() { + let Ok(stream) = stream else { continue }; + let routes = Arc::clone(&routes); + thread::spawn(move || { + let _ = handle(stream, &routes); + }); + } + }); + Ok(bound.port()) +} + +/// Map a request target `/_m/` to `(backend_port, )` with +/// the prefix stripped and an empty path normalized to `/` (query kept). +/// `None` → no matching route (404). +fn rewrite_target(target: &str, routes: &HashMap) -> Option<(u16, String)> { + let rest = target.strip_prefix(super::MODULE_ROUTE_PREFIX)?; + let slug_end = rest.find(['/', '?']).unwrap_or(rest.len()); + let (slug, remainder) = rest.split_at(slug_end); + let port = *routes.get(slug)?; + let rewritten = if remainder.is_empty() || remainder.starts_with('?') { + format!("/{remainder}") + } else { + remainder.to_string() + }; + Some((port, rewritten)) +} + +fn handle(mut client: TcpStream, routes: &HashMap) -> std::io::Result<()> { + // Read the request head (plus whatever body bytes arrived with it). + let mut buf = Vec::new(); + let mut chunk = [0u8; 4096]; + let head_end = loop { + let n = client.read(&mut chunk)?; + if n == 0 { + return Ok(()); + } + // Only scan the new tail (the terminator may straddle the chunk + // boundary by up to 3 bytes) — rescanning from 0 on every read + // would be O(head × chunks). + let scan_from = buf.len().saturating_sub(3); + buf.extend_from_slice(&chunk[..n]); + if let Some(end) = find_head_end(&buf[scan_from..]) { + break scan_from + end; + } + if buf.len() > MAX_HEAD { + return respond(&mut client, "431 Request Header Fields Too Large"); + } + }; + let (head, body_prefix) = buf.split_at(head_end); + let head = String::from_utf8_lossy(head); + let mut lines = head.split("\r\n"); + let mut request_line = lines.next().unwrap_or_default().splitn(3, ' '); + let method = request_line.next().unwrap_or_default(); + let target = request_line.next().unwrap_or_default(); + let version = request_line.next().unwrap_or("HTTP/1.1"); + + let Some((port, rewritten)) = rewrite_target(target, routes) else { + return respond(&mut client, "404 Not Found"); + }; + let Ok(mut backend) = TcpStream::connect(("127.0.0.1", port)) else { + // Module still compiling / crashed — same 502 the Go proxy returned. + return respond(&mut client, "502 Bad Gateway"); + }; + + // Rebuild the head with the stripped path. Force `Connection: close` + // so each proxied request gets its own connection pair — a reused + // client connection would carry the next request's /_m/ prefix past + // this rewrite. Protocol upgrades keep their original Connection + // header (the connection becomes a dedicated duplex stream anyway). + let is_upgrade = lines.clone().any(|l| header_starts(l, "upgrade:")); + let mut out = format!("{method} {rewritten} {version}\r\n"); + for line in lines { + if line.is_empty() { + continue; + } + if !is_upgrade && header_starts(line, "connection:") { + continue; + } + out.push_str(line); + out.push_str("\r\n"); + } + if !is_upgrade { + out.push_str("Connection: close\r\n"); + } + out.push_str("\r\n"); + backend.write_all(out.as_bytes())?; + backend.write_all(body_prefix)?; + + // Relay raw bytes both ways until either side closes. + let mut client_rd = client.try_clone()?; + let mut backend_wr = backend.try_clone()?; + let upstream = thread::spawn(move || { + let _ = std::io::copy(&mut client_rd, &mut backend_wr); + let _ = backend_wr.shutdown(Shutdown::Write); + }); + let _ = std::io::copy(&mut backend, &mut client); + // Unblock the upstream copy (shutdown on a clone reaches the shared fd). + let _ = client.shutdown(Shutdown::Both); + let _ = backend.shutdown(Shutdown::Both); + let _ = upstream.join(); + Ok(()) +} + +fn find_head_end(buf: &[u8]) -> Option { + buf.windows(4).position(|w| w == b"\r\n\r\n").map(|p| p + 4) +} + +/// Case-insensitive "header line starts with `prefix`" without allocating +/// a lowercased copy of every line. +fn header_starts(line: &str, prefix: &str) -> bool { + line.as_bytes() + .get(..prefix.len()) + .is_some_and(|head| head.eq_ignore_ascii_case(prefix.as_bytes())) +} + +fn respond(client: &mut TcpStream, status: &str) -> std::io::Result<()> { + client.write_all( + format!("HTTP/1.1 {status}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").as_bytes(), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn routes() -> HashMap { + HashMap::from([("oauth-core".to_string(), 18080)]) + } + + #[test] + fn rewrite_strips_prefix() { + assert_eq!( + rewrite_target("/_m/oauth-core/foo", &routes()), + Some((18080, "/foo".into())) + ); + assert_eq!( + rewrite_target("/_m/oauth-core/a/b?q=2", &routes()), + Some((18080, "/a/b?q=2".into())) + ); + } + + #[test] + fn rewrite_empty_path_becomes_root() { + assert_eq!( + rewrite_target("/_m/oauth-core", &routes()), + Some((18080, "/".into())) + ); + assert_eq!( + rewrite_target("/_m/oauth-core/", &routes()), + Some((18080, "/".into())) + ); + // Bare slug + query: path is empty, query survives. + assert_eq!( + rewrite_target("/_m/oauth-core?x=1", &routes()), + Some((18080, "/?x=1".into())) + ); + } + + #[test] + fn rewrite_rejects_unknown_slug_and_non_module_paths() { + assert_eq!(rewrite_target("/_m/unknown/foo", &routes()), None); + assert_eq!(rewrite_target("/healthz", &routes()), None); + // Slug must be delimiter-bounded, not prefix-matched. + assert_eq!(rewrite_target("/_m/oauth-corefoo", &routes()), None); + } + + /// End-to-end: proxied request reaches a real backend with the prefix + /// stripped, and the response relays back. + #[test] + fn proxies_to_backend_with_stripped_path() { + // Dummy backend: echo the request line's target in the body. + let backend = TcpListener::bind("127.0.0.1:0").unwrap(); + let backend_port = backend.local_addr().unwrap().port(); + thread::spawn(move || { + for stream in backend.incoming() { + let mut stream = stream.unwrap(); + let mut buf = [0u8; 4096]; + let n = stream.read(&mut buf).unwrap(); + let head = String::from_utf8_lossy(&buf[..n]).to_string(); + let target = head.split(' ').nth(1).unwrap_or("").to_string(); + let body = format!("saw {target}"); + let _ = stream.write_all( + format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ) + .as_bytes(), + ); + } + }); + + let proxy_port = spawn(0, HashMap::from([("test-mod".to_string(), backend_port)])).unwrap(); + + let mut client = TcpStream::connect(("127.0.0.1", proxy_port)).unwrap(); + client + .write_all(b"GET /_m/test-mod/hello?x=1 HTTP/1.1\r\nHost: localhost\r\n\r\n") + .unwrap(); + let mut response = String::new(); + client.read_to_string(&mut response).unwrap(); + assert!(response.starts_with("HTTP/1.1 200 OK"), "got {response}"); + assert!(response.ends_with("saw /hello?x=1"), "got {response}"); + + // Unknown slug → 404 from the proxy itself. + let mut client = TcpStream::connect(("127.0.0.1", proxy_port)).unwrap(); + client + .write_all(b"GET /_m/nope/x HTTP/1.1\r\nHost: localhost\r\n\r\n") + .unwrap(); + let mut response = String::new(); + client.read_to_string(&mut response).unwrap(); + assert!(response.starts_with("HTTP/1.1 404"), "got {response}"); + } +} diff --git a/src/commands/dev/reload.rs b/src/commands/dev/reload.rs new file mode 100644 index 0000000..c1c76c3 --- /dev/null +++ b/src/commands/dev/reload.rs @@ -0,0 +1,108 @@ +//! Polling file watcher for `dev --all` hot-reload. +//! +//! Mirrors the air config the shell runner generated: watch `.go` and +//! `.sql`, exclude `tmp`, `web`, `node_modules`, and POLL mtimes rather +//! than using native fs events — OrbStack/Docker bind mounts deliver no +//! inotify events, so a native watcher would silently miss every edit. + +use std::path::{Path, PathBuf}; +use std::time::SystemTime; + +const WATCH_EXTS: &[&str] = &["go", "sql"]; +const EXCLUDE_DIRS: &[&str] = &["tmp", "web", "node_modules", ".git"]; + +/// Sorted (path, mtime) list of every watched file under `dir`. Two scans +/// compare equal iff no watched file was added, removed, or modified. +pub(super) type Signature = Vec<(PathBuf, SystemTime)>; + +pub(super) fn scan(dir: &Path) -> Signature { + let mut out = Vec::new(); + walk(dir, &mut out); + out.sort(); + out +} + +fn walk(dir: &Path, out: &mut Signature) { + let Ok(entries) = std::fs::read_dir(dir) else { + return; + }; + for entry in entries.flatten() { + let Ok(ft) = entry.file_type() else { continue }; + let path = entry.path(); + if ft.is_dir() { + let name = entry.file_name(); + if EXCLUDE_DIRS.contains(&name.to_string_lossy().as_ref()) { + continue; + } + walk(&path, out); + } else if ft.is_file() { + let watched = path + .extension() + .and_then(|e| e.to_str()) + .is_some_and(|e| WATCH_EXTS.contains(&e)); + if !watched { + continue; + } + if let Ok(md) = entry.metadata() { + out.push((path, md.modified().unwrap_or(SystemTime::UNIX_EPOCH))); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + #[test] + fn scan_includes_watched_extensions_recursively() { + let tmp = tempfile::tempdir().unwrap(); + std::fs::write(tmp.path().join("main.go"), "package main").unwrap(); + std::fs::create_dir(tmp.path().join("sql")).unwrap(); + std::fs::write(tmp.path().join("sql/0001.sql"), "select 1").unwrap(); + std::fs::write(tmp.path().join("notes.txt"), "ignored").unwrap(); + + let sig = scan(tmp.path()); + let names: Vec<_> = sig + .iter() + .map(|(p, _)| p.file_name().unwrap().to_string_lossy().into_owned()) + .collect(); + assert_eq!(names, vec!["main.go", "0001.sql"], "got {names:?}"); + } + + #[test] + fn scan_skips_excluded_dirs() { + let tmp = tempfile::tempdir().unwrap(); + std::fs::write(tmp.path().join("main.go"), "package main").unwrap(); + for dir in ["web", "tmp", "node_modules"] { + std::fs::create_dir(tmp.path().join(dir)).unwrap(); + std::fs::write(tmp.path().join(dir).join("skip.go"), "x").unwrap(); + } + assert_eq!(scan(tmp.path()).len(), 1); + } + + #[test] + fn signature_changes_on_touch_add_remove() { + let tmp = tempfile::tempdir().unwrap(); + let main_go = tmp.path().join("main.go"); + std::fs::write(&main_go, "package main").unwrap(); + let before = scan(tmp.path()); + + // Touch: bump mtime without changing content or file set. + let f = std::fs::File::options().write(true).open(&main_go).unwrap(); + f.set_modified(SystemTime::now() + Duration::from_secs(2)) + .unwrap(); + let touched = scan(tmp.path()); + assert_ne!(before, touched, "mtime bump must change the signature"); + + // Add. + std::fs::write(tmp.path().join("extra.go"), "package main").unwrap(); + let added = scan(tmp.path()); + assert_ne!(touched, added); + + // Remove. + std::fs::remove_file(tmp.path().join("extra.go")).unwrap(); + assert_eq!(scan(tmp.path()).len(), 1); + } +}