diff --git a/Cargo.toml b/Cargo.toml index c0ce623..61a7de3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,7 +29,7 @@ dialoguer = { version = "0.11", default-features = false } console = "0.15" indicatif = "0.17" ctrlc = "3" -tokio = { version = "1", features = ["rt", "macros", "time", "sync", "net", "io-util"] } +tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros", "time", "sync", "net", "io-util"] } tokio-tungstenite = { version = "0.24", default-features = false, features = ["connect", "rustls-tls-webpki-roots"] } futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] } diff --git a/src/api.rs b/src/api.rs index 9b0561d..037754f 100644 --- a/src/api.rs +++ b/src/api.rs @@ -76,6 +76,41 @@ pub struct CreateModuleInput<'a> { } /// GET /v1/auth/me — returns the authenticated user's identity. +/// Response shape from `POST /v1/tunnel/token`. The CLI follows up with +/// a WebSocket connect against `wss_url` carrying `?token=`. +#[derive(Deserialize, Debug)] +pub struct TunnelToken { + pub token: String, + pub wss_url: String, + /// RFC3339. Server-side TTL is short (60s); we don't act on this + /// value (any failure triggers a fresh mint), but it's surfaced for + /// diagnostic logging if the connect hangs. + #[allow(dead_code)] + pub expires_at: String, +} + +/// POST /v1/tunnel/token — mint a connect token for the WSS dev tunnel. +pub fn tunnel_token( + http: &Client, + dispatch_base: &str, + access_token: &str, +) -> Result { + let endpoint = format!("{}/v1/tunnel/token", dispatch_base.trim_end_matches('/')); + let resp = http + .post(&endpoint) + .bearer_auth(access_token) + .header("Accept", "application/json") + .send()?; + let status = resp.status(); + if status.is_success() { + return Ok(resp.json::()?); + } + if status == reqwest::StatusCode::UNAUTHORIZED || status == reqwest::StatusCode::FORBIDDEN { + return Err(ApiError::Unauthenticated); + } + Err(unexpected_body_error(resp)) +} + pub fn me(http: &Client, api_base: &str, access_token: &str) -> Result { let endpoint = format!("{}/v1/auth/me", api_base.trim_end_matches('/')); diff --git a/src/commands/dev/mod.rs b/src/commands/dev/mod.rs index 34f97d0..6de6380 100644 --- a/src/commands/dev/mod.rs +++ b/src/commands/dev/mod.rs @@ -1,31 +1,34 @@ //! `mirrorstack dev` — run a module locally with the supporting services -//! it needs (Postgres for now; Redis + WSS tunnel client land in later PRs). +//! it needs. //! //! The lifecycle is: //! 1. Bootstrap a `docker-compose.yml` in the cwd if missing //! 2. `docker compose up -d --wait` (server-side healthcheck blocks until pg is ready) -//! 3. Spawn `go run .` with `MS_LOCAL_DB_URL` injected -//! 4. Stream the module's stdout/stderr through a labeled prefix -//! 5. On Ctrl-C: kill the module process (SIGKILL on unix; SIGTERM-then-SIGKILL is queued as a follow-up), then `docker compose down` -//! -//! No WSS tunnel yet — the module runs as a normal HTTP server on its own -//! port. Future PR layers the tunnel registration on top. +//! 3. (optional, --tunnel) Mint a connect token and open the WSS so +//! remote callers can reach this module via the platform's Leaf 1 +//! 302 path +//! 4. Spawn `go run .` with `MS_LOCAL_DB_URL` injected +//! 5. Stream the module's stdout/stderr through a labeled prefix +//! 6. On Ctrl-C: kill the module (SIGKILL on unix; SIGTERM-then-SIGKILL +//! is queued as a follow-up — see issue #25), close the tunnel if +//! open, then `docker compose down` use std::io::IsTerminal; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; +use std::time::Duration; -use anyhow::{Result, anyhow}; +use anyhow::{Context, Result, anyhow}; use clap::Args; use console::style; -use super::{ok_mark, warn_prefix}; +use super::{ + DEFAULT_DISPATCH_BASE, ENV_DISPATCH_URL, ok_mark, resolve_base, session_expired, warn_prefix, +}; +use crate::{api, credentials, http}; mod compose; +mod module_meta; mod process; -// Used in a follow-up PR that integrates WSS into the dev lifecycle. -// Compiled + tested now so the wire format and protocol stay correct -// while the AWS Sender impl + IaC for the WS API GW catch up. -#[allow(dead_code)] mod tunnel; #[derive(Args)] @@ -41,6 +44,17 @@ pub struct DevArgs { /// when --no-compose is unset, otherwise must be supplied via env. #[arg(long)] db_url: Option, + /// Open a WSS tunnel to the platform so deployed callers can reach this + /// module via Leaf 1 (302 → localhost). Requires the developer to be + /// signed in (via `mirrorstack login`) and the platform's dispatch + /// service + WS API GW to be reachable. Module identity is parsed from + /// `Config.ID` in main.go; --local-url defaults to http://localhost:8080. + #[arg(long)] + tunnel: bool, + /// URL the platform should 302 to when routing inbound calls for this + /// module. Only used when --tunnel is set. Default: http://localhost:8080. + #[arg(long)] + local_url: Option, } // Matches templates/dev/docker-compose.yml.tmpl: host port 5433 (chosen so @@ -48,6 +62,10 @@ pub struct DevArgs { const DEFAULT_DB_URL: &str = "postgres://mirrorstack:mirrorstack@localhost:5433/mirrorstack?sslmode=disable"; +/// Default local URL the platform should 302 to for Leaf 1 routing. Matches +/// the SDK's default HTTP listener port. Override with `--local-url`. +const DEFAULT_LOCAL_URL: &str = "http://localhost:8080"; + pub fn run(args: DevArgs) -> Result<()> { let cwd = args .dir @@ -62,6 +80,15 @@ pub fn run(args: DevArgs) -> Result<()> { let db_url = resolve_db_url(&args)?; + // Bring the tunnel up BEFORE compose so that a tunnel registration + // failure (auth expired, dispatch unreachable) doesn't waste the user's + // time bringing containers up first. + let tunnel_handle = if args.tunnel { + Some(open_tunnel(&cwd, args.local_url.as_deref())?) + } else { + None + }; + if !args.no_compose { compose::ensure_compose_file(&cwd)?; compose::up(&cwd)?; @@ -70,6 +97,14 @@ pub fn run(args: DevArgs) -> Result<()> { eprintln!("{} module running — Ctrl-C to stop", ok_mark()); let module_status = process::run_module(&cwd, &db_url); + if let Some((handle, runtime)) = tunnel_handle { + handle.shutdown(); + // Block on the runtime briefly to let the close frame land. + runtime.block_on(async { + tokio::time::sleep(Duration::from_millis(200)).await; + }); + } + if !args.no_compose { if let Err(e) = compose::down(&cwd) { eprintln!( @@ -82,6 +117,69 @@ pub fn run(args: DevArgs) -> Result<()> { module_status } +/// Build a single-threaded tokio runtime, mint a connect token via +/// dispatch, open the WSS, send `register`, and return the handle so the +/// caller can shut it down on Ctrl-C. Runs blocking under the hood — +/// the tunnel itself stays alive on the runtime's worker thread. +fn open_tunnel( + module_dir: &Path, + local_url: Option<&str>, +) -> Result<(tunnel::TunnelHandle, tokio::runtime::Runtime)> { + let creds = credentials::load_or_login_hint()?; + let dispatch_base = resolve_base(ENV_DISPATCH_URL, DEFAULT_DISPATCH_BASE); + let client = http::client(Duration::from_secs(15))?; + let module_id = module_meta::read_module_id(module_dir)?; + let local_url = local_url.unwrap_or(DEFAULT_LOCAL_URL); + + eprintln!( + "{} fetching tunnel token from {}", + ok_mark(), + style(&dispatch_base).cyan().dim() + ); + let token = match api::tunnel_token(&client, &dispatch_base, &creds.access_token) { + Ok(t) => t, + Err(api::ApiError::Unauthenticated) => return Err(session_expired()), + Err(e) => { + // Dispatch unreachable is a routine "platform isn't running yet" + // error — point the user at the right env var. + return Err(anyhow!( + "dev: tunnel-token mint failed: {e}. Check {} (or {} env var) and that the dispatch service is running.", + dispatch_base, + ENV_DISPATCH_URL + )); + } + }; + // Multi-thread (one worker) so the WSS background task — pinger, + // server-frame reader — keeps ticking while the rest of the CLI runs + // its blocking module-process loop. A current-thread runtime would + // freeze every spawned future the moment block_on returns. + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .build() + .context("dev: build tokio runtime")?; + + let handle = runtime.block_on(async { + tunnel::open( + &token.wss_url, + &token.token, + tunnel::RegisterPayload { + module_id: &module_id, + local_url, + version: env!("CARGO_PKG_VERSION"), + }, + ) + .await + })?; + + eprintln!( + "{} tunnel registered (session {})", + ok_mark(), + style(&handle.session_id).cyan().dim() + ); + Ok((handle, runtime)) +} + fn resolve_db_url(args: &DevArgs) -> Result { if let Some(url) = &args.db_url { return Ok(url.clone()); @@ -111,13 +209,19 @@ pub(super) fn line_prefix(label: &str) -> String { mod tests { use super::*; + fn args_with(no_compose: bool, db_url: Option) -> DevArgs { + DevArgs { + dir: None, + no_compose, + db_url, + tunnel: false, + local_url: None, + } + } + #[test] fn resolve_db_url_prefers_explicit_arg() { - let args = DevArgs { - dir: None, - no_compose: false, - db_url: Some("postgres://x".into()), - }; + let args = args_with(false, Some("postgres://x".into())); assert_eq!(resolve_db_url(&args).unwrap(), "postgres://x"); } @@ -129,11 +233,7 @@ mod tests { if std::env::var("MS_LOCAL_DB_URL").is_ok() { return; } - let args = DevArgs { - dir: None, - no_compose: true, - db_url: None, - }; + let args = args_with(true, None); let err = resolve_db_url(&args).unwrap_err().to_string(); assert!(err.contains("MS_LOCAL_DB_URL")); } @@ -143,11 +243,7 @@ mod tests { if std::env::var("MS_LOCAL_DB_URL").is_ok() { return; } - let args = DevArgs { - dir: None, - no_compose: false, - db_url: None, - }; + let args = args_with(false, None); let url = resolve_db_url(&args).unwrap(); assert!(url.starts_with("postgres://mirrorstack")); assert!(url.contains(":5433/")); diff --git a/src/commands/dev/module_meta.rs b/src/commands/dev/module_meta.rs new file mode 100644 index 0000000..8bb89ed --- /dev/null +++ b/src/commands/dev/module_meta.rs @@ -0,0 +1,120 @@ +//! Read the developer's module identity off the scaffolded source tree. +//! +//! For Phase 1 we parse `Config.ID = "..."` out of `main.go`. That field +//! is what the SDK uses for `mod_` schema names, so it's the right +//! handle to register against the tunnel — and it's already the value +//! the CLI's scaffold substituted from the platform UUID. +//! +//! A future PR may switch to a `.mirrorstack/meta.json` written at +//! scaffold time. Until then, parsing main.go is a one-line regex on a +//! file whose shape we control. + +use std::fs; +use std::path::Path; + +use anyhow::{Context, Result, anyhow}; + +/// Find the `ID: ""` line inside the file's `ms.Init(ms.Config{...})` +/// block. The scaffold template's main.go has it on a predictable line, +/// but a developer is free to reformat — so we match the literal token +/// `ID:` followed by `"..."` rather than relying on column positions. +pub(super) fn read_module_id(module_dir: &Path) -> Result { + let path = module_dir.join("main.go"); + let body = + fs::read_to_string(&path).with_context(|| format!("dev: read {}", path.display()))?; + extract_id(&body).ok_or_else(|| { + anyhow!( + "dev: couldn't find `ID: \"...\"` in {}. Is this a MirrorStack module?", + path.display() + ) + }) +} + +/// Extract the first occurrence of the literal pattern `ID: "..."`. +/// +/// Tolerates any whitespace around `ID:` (the scaffold uses `ID: ` +/// for column alignment with `Name:`, but reformatters may shrink it +/// to a single space). Stops at the first closing quote — `Config.ID` +/// is a single string literal with no escapes possible under the SDK's +/// regex (`^[a-z][a-z0-9_]{0,35}$`). +fn extract_id(go_source: &str) -> Option { + let after_label = find_after(go_source, "ID:")?; + let after_open_quote = find_after(after_label, "\"")?; + let close = after_open_quote.find('"')?; + Some(after_open_quote[..close].to_string()) +} + +fn find_after<'a>(haystack: &'a str, needle: &str) -> Option<&'a str> { + let pos = haystack.find(needle)?; + Some(&haystack[pos + needle.len()..]) +} + +#[cfg(test)] +mod tests { + use super::*; + + const SAMPLE_MAIN_GO: &str = r#" +package main + +func main() { + if err := ms.Init(ms.Config{ + ID: "mbb8a3f8b123456789abcdef012345678", + Name: "Media", + Icon: "extension", + }); err != nil { + log.Fatalf("init: %v", err) + } +} +"#; + + #[test] + fn extract_id_from_scaffold_template() { + assert_eq!( + extract_id(SAMPLE_MAIN_GO).unwrap(), + "mbb8a3f8b123456789abcdef012345678" + ); + } + + #[test] + fn extract_id_tolerates_single_space() { + let src = r#"ms.Config{ ID: "media", Name: "Media" }"#; + assert_eq!(extract_id(src).unwrap(), "media"); + } + + #[test] + fn extract_id_returns_none_when_absent() { + assert!(extract_id(r#"ms.Config{Name: "X"}"#).is_none()); + } + + #[test] + fn extract_id_first_match_wins() { + // If the developer adds another `ID:` somewhere later (a struct + // tag, a comment, a literal in another field), the first one + // — Config.ID at the top of Init — is the canonical one. + let src = r#" +ms.Config{ + ID: "first", + Name: "X", +} +// later struct: SomethingElse{ID: "ignored"} +"#; + assert_eq!(extract_id(src).unwrap(), "first"); + } + + #[test] + fn read_module_id_from_disk() { + let tmp = tempfile::tempdir().unwrap(); + std::fs::write(tmp.path().join("main.go"), SAMPLE_MAIN_GO).unwrap(); + assert_eq!( + read_module_id(tmp.path()).unwrap(), + "mbb8a3f8b123456789abcdef012345678" + ); + } + + #[test] + fn read_module_id_missing_main_errors() { + let tmp = tempfile::tempdir().unwrap(); + let err = read_module_id(tmp.path()).unwrap_err().to_string(); + assert!(err.contains("read")); + } +} diff --git a/src/commands/dev/tunnel.rs b/src/commands/dev/tunnel.rs index 903bc3f..dc5a189 100644 --- a/src/commands/dev/tunnel.rs +++ b/src/commands/dev/tunnel.rs @@ -117,11 +117,6 @@ impl Frame { payload, } } - - pub(super) fn with_corr(mut self, corr_id: String) -> Self { - self.corr_id = Some(corr_id); - self - } } #[derive(Debug, Serialize)] @@ -134,7 +129,14 @@ pub(super) struct RegisterPayload<'a> { #[derive(Debug, Deserialize)] pub(super) struct RegisterAck { pub session_id: String, + /// Carried by future RPC/SQL frames as the per-tunnel service token. + /// Phase 1 doesn't emit those frames yet; the field is held so the + /// returned handle stays a single source of truth for the ack. + #[allow(dead_code)] pub service_token: String, + /// RFC3339 — informational; the L1.5 reconnect contract makes the + /// client retry on any failure rather than expiry-driven refresh. + #[allow(dead_code)] pub expires_at: String, } diff --git a/src/commands/mod.rs b/src/commands/mod.rs index e38d995..d1ffb47 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -1,7 +1,7 @@ //! Top-level CLI surface. Each variant of `Command` maps to a subcommand //! module under this directory. -use anyhow::Result; +use anyhow::{Result, anyhow}; use clap::{Parser, Subcommand}; use console::{StyledObject, style}; @@ -25,9 +25,16 @@ pub(crate) const DEFAULT_APPS_API_BASE: &str = "https://api.mirrorstack.ai"; /// Default web-account host. Override with `MIRRORSTACK_WEB_URL`. pub(crate) const DEFAULT_WEB_BASE: &str = "https://account.mirrorstack.ai"; +/// Default api-platform dispatch-service host. The dispatch Lambda is +/// reached under the same `api.mirrorstack.ai` hostname in prod via +/// path-routed ingress. Local dev uses port 8083 — set via `.env`. +/// Override with `MIRRORSTACK_DISPATCH_URL`. +pub(crate) const DEFAULT_DISPATCH_BASE: &str = "https://api.mirrorstack.ai"; + pub(crate) const ENV_API_URL: &str = "MIRRORSTACK_API_URL"; pub(crate) const ENV_APPS_API_URL: &str = "MIRRORSTACK_APPS_API_URL"; pub(crate) const ENV_WEB_URL: &str = "MIRRORSTACK_WEB_URL"; +pub(crate) const ENV_DISPATCH_URL: &str = "MIRRORSTACK_DISPATCH_URL"; /// Look up a base URL from `env_var`, falling back to `default` when unset. pub(crate) fn resolve_base(env_var: &str, default: &str) -> String { @@ -44,6 +51,13 @@ pub(crate) fn warn_prefix() -> StyledObject<&'static str> { style("warning:").yellow().bold() } +/// Standard error returned by every command that hits a 401 from the +/// platform. Centralized so the wording is identical across `module init`, +/// `dev`, `whoami`, etc. — users see one message, not three slight variants. +pub(crate) fn session_expired() -> anyhow::Error { + anyhow!("session expired. Run `mirrorstack login` to sign in again.") +} + /// Official command-line tool for the MirrorStack platform. #[derive(Parser)] #[command(name = "mirrorstack", version, about, long_about = None)] diff --git a/src/commands/module/mod.rs b/src/commands/module/mod.rs index be8f01f..fe4b7ce 100644 --- a/src/commands/module/mod.rs +++ b/src/commands/module/mod.rs @@ -22,7 +22,7 @@ mod scaffold; use super::{ DEFAULT_API_BASE, DEFAULT_APPS_API_BASE, DEFAULT_WEB_BASE, ENV_API_URL, ENV_APPS_API_URL, - ENV_WEB_URL, ok_mark, resolve_base, + ENV_WEB_URL, ok_mark, resolve_base, session_expired, }; #[derive(Args)] @@ -257,10 +257,6 @@ fn is_cwd(target: &Path) -> bool { matches!(comps.next(), Some(std::path::Component::CurDir)) && comps.next().is_none() } -fn session_expired() -> anyhow::Error { - anyhow!("session expired. Run `mirrorstack login` to sign in again.") -} - fn collect_name_and_slug(theme: &ColorfulTheme, args: &InitArgs) -> Result<(String, String)> { let name = if let Some(n) = args .name