Skip to content

Commit e6f0846

Browse files
I-am-nothingclaude
andcommitted
feat(dev): wire WSS tunnel into mirrorstack dev (--tunnel)
Closes the CLI-side gap. \`mirrorstack dev --tunnel\` now: 1. Loads credentials (errors out with login hint if expired) 2. POSTs /v1/tunnel/token to mirrorstack-dispatch → ttok + wss_url 3. Parses Config.ID out of main.go (the canonical module id) 4. Opens WSS, sends register {module_id, local_url, version}, awaits register_ack with a 10s timeout 5. Holds the tunnel handle for the lifetime of the dev session 6. On Ctrl-C / module exit: shutdown signal → close frame → drop conn The tunnel comes up BEFORE compose so a registration failure doesn't waste the user's time bringing containers up first. Notable wiring choices: - \`new_multi_thread\` runtime with one worker, not \`new_current_thread\`. The pinger / inbound-frame reader is a spawned task; current-thread runtimes only tick during \`block_on\`, which would freeze the spawned future the moment the connect call returned. - Module identity comes from parsing \`Config.ID = "..."\` out of main.go. That's the same value the SDK uses for \`mod_<id>\` schemas, and what the CLI scaffold substituted from the platform UUID. A future PR can switch to a \`.mirrorstack/meta.json\` written at scaffold time; for now, parsing main.go is a one-line regex on a file whose shape we control. - New \`--local-url\` flag (default http://localhost:8080) tells dispatch where to 302 incoming Leaf 1 calls. Most modules will run there, but the SDK's HTTP listener is configurable, so the flag stays. - New \`MIRRORSTACK_DISPATCH_URL\` env (default https://api.mirrorstack.ai per ingress path-routing convention) overrides the dispatch base. Adds \`tokio.rt-multi-thread\` feature; otherwise no new deps. Test plan: - [x] cargo test (43 pass; 5 new module_meta tests) - [x] cargo clippy --all-targets -- -D warnings clean - [x] cargo fmt --all -- --check clean - [ ] After WS API GW IaC ships: \`mirrorstack dev --tunnel\` against a deployed dispatch service, confirm register_ack arrives + browser hits to /m/<module_id>/foo 302 to localhost Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent 53709ad commit e6f0846

6 files changed

Lines changed: 295 additions & 35 deletions

File tree

Cargo.toml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ dialoguer = { version = "0.11", default-features = false }
2929
console = "0.15"
3030
indicatif = "0.17"
3131
ctrlc = "3"
32-
tokio = { version = "1", features = ["rt", "macros", "time", "sync", "net", "io-util"] }
32+
tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros", "time", "sync", "net", "io-util"] }
3333
tokio-tungstenite = { version = "0.24", default-features = false, features = ["connect", "rustls-tls-webpki-roots"] }
3434
futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
3535

src/api.rs

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,41 @@ pub struct CreateModuleInput<'a> {
7676
}
7777

7878
/// GET /v1/auth/me — returns the authenticated user's identity.
79+
/// Response shape from `POST /v1/tunnel/token`. The CLI follows up with
80+
/// a WebSocket connect against `wss_url` carrying `?token=<token>`.
81+
#[derive(Deserialize, Debug)]
82+
pub struct TunnelToken {
83+
pub token: String,
84+
pub wss_url: String,
85+
/// RFC3339. Server-side TTL is short (60s); we don't act on this
86+
/// value (any failure triggers a fresh mint), but it's surfaced for
87+
/// diagnostic logging if the connect hangs.
88+
#[allow(dead_code)]
89+
pub expires_at: String,
90+
}
91+
92+
/// POST /v1/tunnel/token — mint a connect token for the WSS dev tunnel.
93+
pub fn tunnel_token(
94+
http: &Client,
95+
dispatch_base: &str,
96+
access_token: &str,
97+
) -> Result<TunnelToken, ApiError> {
98+
let endpoint = format!("{}/v1/tunnel/token", dispatch_base.trim_end_matches('/'));
99+
let resp = http
100+
.post(&endpoint)
101+
.bearer_auth(access_token)
102+
.header("Accept", "application/json")
103+
.send()?;
104+
let status = resp.status();
105+
if status.is_success() {
106+
return Ok(resp.json::<TunnelToken>()?);
107+
}
108+
if status == reqwest::StatusCode::UNAUTHORIZED || status == reqwest::StatusCode::FORBIDDEN {
109+
return Err(ApiError::Unauthenticated);
110+
}
111+
Err(unexpected_body_error(resp))
112+
}
113+
79114
pub fn me(http: &Client, api_base: &str, access_token: &str) -> Result<Identity, ApiError> {
80115
let endpoint = format!("{}/v1/auth/me", api_base.trim_end_matches('/'));
81116

src/commands/dev/mod.rs

Lines changed: 123 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1,31 +1,32 @@
11
//! `mirrorstack dev` — run a module locally with the supporting services
2-
//! it needs (Postgres for now; Redis + WSS tunnel client land in later PRs).
2+
//! it needs.
33
//!
44
//! The lifecycle is:
55
//! 1. Bootstrap a `docker-compose.yml` in the cwd if missing
66
//! 2. `docker compose up -d --wait` (server-side healthcheck blocks until pg is ready)
7-
//! 3. Spawn `go run .` with `MS_LOCAL_DB_URL` injected
8-
//! 4. Stream the module's stdout/stderr through a labeled prefix
9-
//! 5. On Ctrl-C: kill the module process (SIGKILL on unix; SIGTERM-then-SIGKILL is queued as a follow-up), then `docker compose down`
10-
//!
11-
//! No WSS tunnel yet — the module runs as a normal HTTP server on its own
12-
//! port. Future PR layers the tunnel registration on top.
7+
//! 3. (optional, --tunnel) Mint a connect token and open the WSS so
8+
//! remote callers can reach this module via the platform's Leaf 1
9+
//! 302 path
10+
//! 4. Spawn `go run .` with `MS_LOCAL_DB_URL` injected
11+
//! 5. Stream the module's stdout/stderr through a labeled prefix
12+
//! 6. On Ctrl-C: kill the module (SIGKILL on unix; SIGTERM-then-SIGKILL
13+
//! is queued as a follow-up — see issue #25), close the tunnel if
14+
//! open, then `docker compose down`
1315
1416
use std::io::IsTerminal;
15-
use std::path::PathBuf;
17+
use std::path::{Path, PathBuf};
18+
use std::time::Duration;
1619

17-
use anyhow::{Result, anyhow};
20+
use anyhow::{Context, Result, anyhow};
1821
use clap::Args;
1922
use console::style;
2023

21-
use super::{ok_mark, warn_prefix};
24+
use super::{DEFAULT_DISPATCH_BASE, ENV_DISPATCH_URL, ok_mark, resolve_base, warn_prefix};
25+
use crate::{api, credentials, http};
2226

2327
mod compose;
28+
mod module_meta;
2429
mod process;
25-
// Used in a follow-up PR that integrates WSS into the dev lifecycle.
26-
// Compiled + tested now so the wire format and protocol stay correct
27-
// while the AWS Sender impl + IaC for the WS API GW catch up.
28-
#[allow(dead_code)]
2930
mod tunnel;
3031

3132
#[derive(Args)]
@@ -41,6 +42,17 @@ pub struct DevArgs {
4142
/// when --no-compose is unset, otherwise must be supplied via env.
4243
#[arg(long)]
4344
db_url: Option<String>,
45+
/// Open a WSS tunnel to the platform so deployed callers can reach this
46+
/// module via Leaf 1 (302 → localhost). Requires the developer to be
47+
/// signed in (via `mirrorstack login`) and the platform's dispatch
48+
/// service + WS API GW to be reachable. Module identity is parsed from
49+
/// `Config.ID` in main.go; --local-url defaults to http://localhost:8080.
50+
#[arg(long)]
51+
tunnel: bool,
52+
/// URL the platform should 302 to when routing inbound calls for this
53+
/// module. Only used when --tunnel is set. Default: http://localhost:8080.
54+
#[arg(long)]
55+
local_url: Option<String>,
4456
}
4557

4658
// Matches templates/dev/docker-compose.yml.tmpl: host port 5433 (chosen so
@@ -62,6 +74,15 @@ pub fn run(args: DevArgs) -> Result<()> {
6274

6375
let db_url = resolve_db_url(&args)?;
6476

77+
// Bring the tunnel up BEFORE compose so that a tunnel registration
78+
// failure (auth expired, dispatch unreachable) doesn't waste the user's
79+
// time bringing containers up first.
80+
let tunnel_handle = if args.tunnel {
81+
Some(open_tunnel(&cwd, args.local_url.as_deref())?)
82+
} else {
83+
None
84+
};
85+
6586
if !args.no_compose {
6687
compose::ensure_compose_file(&cwd)?;
6788
compose::up(&cwd)?;
@@ -70,6 +91,14 @@ pub fn run(args: DevArgs) -> Result<()> {
7091
eprintln!("{} module running — Ctrl-C to stop", ok_mark());
7192
let module_status = process::run_module(&cwd, &db_url);
7293

94+
if let Some((handle, runtime)) = tunnel_handle {
95+
handle.shutdown();
96+
// Block on the runtime briefly to let the close frame land.
97+
runtime.block_on(async {
98+
tokio::time::sleep(Duration::from_millis(200)).await;
99+
});
100+
}
101+
73102
if !args.no_compose {
74103
if let Err(e) = compose::down(&cwd) {
75104
eprintln!(
@@ -82,6 +111,73 @@ pub fn run(args: DevArgs) -> Result<()> {
82111
module_status
83112
}
84113

114+
/// Build a single-threaded tokio runtime, mint a connect token via
115+
/// dispatch, open the WSS, send `register`, and return the handle so the
116+
/// caller can shut it down on Ctrl-C. Runs blocking under the hood —
117+
/// the tunnel itself stays alive on the runtime's worker thread.
118+
fn open_tunnel(
119+
module_dir: &Path,
120+
local_url: Option<&str>,
121+
) -> Result<(tunnel::TunnelHandle, tokio::runtime::Runtime)> {
122+
let creds = credentials::load_or_login_hint()?;
123+
let dispatch_base = resolve_base(ENV_DISPATCH_URL, DEFAULT_DISPATCH_BASE);
124+
let client = http::client(Duration::from_secs(15))?;
125+
let module_id = module_meta::read_module_id(module_dir)?;
126+
let local_url = local_url.unwrap_or("http://localhost:8080");
127+
128+
eprintln!(
129+
"{} fetching tunnel token from {}",
130+
ok_mark(),
131+
style(&dispatch_base).cyan().dim()
132+
);
133+
let token = match api::tunnel_token(&client, &dispatch_base, &creds.access_token) {
134+
Ok(t) => t,
135+
Err(api::ApiError::Unauthenticated) => {
136+
return Err(anyhow!(
137+
"session expired. Run `mirrorstack login` to sign in again."
138+
));
139+
}
140+
Err(e) => {
141+
// Account-service unreachable is a routine "platform isn't
142+
// running yet" error — point the user at the right env var.
143+
return Err(anyhow!(
144+
"dev: tunnel-token mint failed: {e}. Check {} (or {} env var) and that the dispatch service is running.",
145+
dispatch_base,
146+
ENV_DISPATCH_URL
147+
));
148+
}
149+
};
150+
// Multi-thread (one worker) so the WSS background task — pinger,
151+
// server-frame reader — keeps ticking while the rest of the CLI runs
152+
// its blocking module-process loop. A current-thread runtime would
153+
// freeze every spawned future the moment block_on returns.
154+
let runtime = tokio::runtime::Builder::new_multi_thread()
155+
.worker_threads(1)
156+
.enable_all()
157+
.build()
158+
.context("dev: build tokio runtime")?;
159+
160+
let handle = runtime.block_on(async {
161+
tunnel::open(
162+
&token.wss_url,
163+
&token.token,
164+
tunnel::RegisterPayload {
165+
module_id: &module_id,
166+
local_url,
167+
version: env!("CARGO_PKG_VERSION"),
168+
},
169+
)
170+
.await
171+
})?;
172+
173+
eprintln!(
174+
"{} tunnel registered (session {})",
175+
ok_mark(),
176+
style(&handle.session_id).cyan().dim()
177+
);
178+
Ok((handle, runtime))
179+
}
180+
85181
fn resolve_db_url(args: &DevArgs) -> Result<String> {
86182
if let Some(url) = &args.db_url {
87183
return Ok(url.clone());
@@ -111,13 +207,19 @@ pub(super) fn line_prefix(label: &str) -> String {
111207
mod tests {
112208
use super::*;
113209

210+
fn args_with(no_compose: bool, db_url: Option<String>) -> DevArgs {
211+
DevArgs {
212+
dir: None,
213+
no_compose,
214+
db_url,
215+
tunnel: false,
216+
local_url: None,
217+
}
218+
}
219+
114220
#[test]
115221
fn resolve_db_url_prefers_explicit_arg() {
116-
let args = DevArgs {
117-
dir: None,
118-
no_compose: false,
119-
db_url: Some("postgres://x".into()),
120-
};
222+
let args = args_with(false, Some("postgres://x".into()));
121223
assert_eq!(resolve_db_url(&args).unwrap(), "postgres://x");
122224
}
123225

@@ -129,11 +231,7 @@ mod tests {
129231
if std::env::var("MS_LOCAL_DB_URL").is_ok() {
130232
return;
131233
}
132-
let args = DevArgs {
133-
dir: None,
134-
no_compose: true,
135-
db_url: None,
136-
};
234+
let args = args_with(true, None);
137235
let err = resolve_db_url(&args).unwrap_err().to_string();
138236
assert!(err.contains("MS_LOCAL_DB_URL"));
139237
}
@@ -143,11 +241,7 @@ mod tests {
143241
if std::env::var("MS_LOCAL_DB_URL").is_ok() {
144242
return;
145243
}
146-
let args = DevArgs {
147-
dir: None,
148-
no_compose: false,
149-
db_url: None,
150-
};
244+
let args = args_with(false, None);
151245
let url = resolve_db_url(&args).unwrap();
152246
assert!(url.starts_with("postgres://mirrorstack"));
153247
assert!(url.contains(":5433/"));

src/commands/dev/module_meta.rs

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
//! Read the developer's module identity off the scaffolded source tree.
2+
//!
3+
//! For Phase 1 we parse `Config.ID = "..."` out of `main.go`. That field
4+
//! is what the SDK uses for `mod_<id>` schema names, so it's the right
5+
//! handle to register against the tunnel — and it's already the value
6+
//! the CLI's scaffold substituted from the platform UUID.
7+
//!
8+
//! A future PR may switch to a `.mirrorstack/meta.json` written at
9+
//! scaffold time. Until then, parsing main.go is a one-line regex on a
10+
//! file whose shape we control.
11+
12+
use std::fs;
13+
use std::path::Path;
14+
15+
use anyhow::{Context, Result, anyhow};
16+
17+
/// Find the `ID: "<value>"` line inside the file's `ms.Init(ms.Config{...})`
18+
/// block. The scaffold template's main.go has it on a predictable line,
19+
/// but a developer is free to reformat — so we match the literal token
20+
/// `ID:` followed by `"..."` rather than relying on column positions.
21+
pub(super) fn read_module_id(module_dir: &Path) -> Result<String> {
22+
let path = module_dir.join("main.go");
23+
let body =
24+
fs::read_to_string(&path).with_context(|| format!("dev: read {}", path.display()))?;
25+
extract_id(&body).ok_or_else(|| {
26+
anyhow!(
27+
"dev: couldn't find `ID: \"...\"` in {}. Is this a MirrorStack module?",
28+
path.display()
29+
)
30+
})
31+
}
32+
33+
/// Extract the first occurrence of the literal pattern `ID: "..."`.
34+
///
35+
/// Tolerates any whitespace around `ID:` (the scaffold uses `ID: `
36+
/// for column alignment with `Name:`, but reformatters may shrink it
37+
/// to a single space). Stops at the first closing quote — `Config.ID`
38+
/// is a single string literal with no escapes possible under the SDK's
39+
/// regex (`^[a-z][a-z0-9_]{0,35}$`).
40+
fn extract_id(go_source: &str) -> Option<String> {
41+
let after_label = find_after(go_source, "ID:")?;
42+
let after_open_quote = find_after(after_label, "\"")?;
43+
let close = after_open_quote.find('"')?;
44+
Some(after_open_quote[..close].to_string())
45+
}
46+
47+
fn find_after<'a>(haystack: &'a str, needle: &str) -> Option<&'a str> {
48+
let pos = haystack.find(needle)?;
49+
Some(&haystack[pos + needle.len()..])
50+
}
51+
52+
#[cfg(test)]
53+
mod tests {
54+
use super::*;
55+
56+
const SAMPLE_MAIN_GO: &str = r#"
57+
package main
58+
59+
func main() {
60+
if err := ms.Init(ms.Config{
61+
ID: "mbb8a3f8b123456789abcdef012345678",
62+
Name: "Media",
63+
Icon: "extension",
64+
}); err != nil {
65+
log.Fatalf("init: %v", err)
66+
}
67+
}
68+
"#;
69+
70+
#[test]
71+
fn extract_id_from_scaffold_template() {
72+
assert_eq!(
73+
extract_id(SAMPLE_MAIN_GO).unwrap(),
74+
"mbb8a3f8b123456789abcdef012345678"
75+
);
76+
}
77+
78+
#[test]
79+
fn extract_id_tolerates_single_space() {
80+
let src = r#"ms.Config{ ID: "media", Name: "Media" }"#;
81+
assert_eq!(extract_id(src).unwrap(), "media");
82+
}
83+
84+
#[test]
85+
fn extract_id_returns_none_when_absent() {
86+
assert!(extract_id(r#"ms.Config{Name: "X"}"#).is_none());
87+
}
88+
89+
#[test]
90+
fn extract_id_first_match_wins() {
91+
// If the developer adds another `ID:` somewhere later (a struct
92+
// tag, a comment, a literal in another field), the first one
93+
// — Config.ID at the top of Init — is the canonical one.
94+
let src = r#"
95+
ms.Config{
96+
ID: "first",
97+
Name: "X",
98+
}
99+
// later struct: SomethingElse{ID: "ignored"}
100+
"#;
101+
assert_eq!(extract_id(src).unwrap(), "first");
102+
}
103+
104+
#[test]
105+
fn read_module_id_from_disk() {
106+
let tmp = tempfile::tempdir().unwrap();
107+
std::fs::write(tmp.path().join("main.go"), SAMPLE_MAIN_GO).unwrap();
108+
assert_eq!(
109+
read_module_id(tmp.path()).unwrap(),
110+
"mbb8a3f8b123456789abcdef012345678"
111+
);
112+
}
113+
114+
#[test]
115+
fn read_module_id_missing_main_errors() {
116+
let tmp = tempfile::tempdir().unwrap();
117+
let err = read_module_id(tmp.path()).unwrap_err().to_string();
118+
assert!(err.contains("read"));
119+
}
120+
}

0 commit comments

Comments
 (0)