use crate::prelude::*;
pub mod github;
pub mod handlers;
pub mod notes;
pub mod ops;
pub mod parser;
pub mod pricing;
pub mod search_sources;
pub mod sse;
pub mod state;
pub mod sync;
use crate::fleet::config::gitmodules::validate_gitmodules_structure;
use anyhow::{bail, Context, Result};
use axum::{
body::{boxed, Full},
http::{header, StatusCode, Uri},
response::{IntoResponse, Response},
routing::{delete, get, patch, post, put},
Router,
};
use rust_embed::RustEmbed;
use std::env;
use std::net::{SocketAddr, TcpListener};
use tower_http::cors::CorsLayer;
use crate::app::AppContext;
use crate::logging::{set_log_broadcast, LogLine, LogRingBuffer};
use state::WebState;
#[derive(RustEmbed)]
#[folder = "frontend/dist"]
struct FrontendAssets;
pub async fn start_server(
port: u16,
debug: bool,
parent_pid: Option<u32>,
ctx: AppContext,
) -> Result<()> {
let workspace_root = ctx.workspace_root.clone();
validate_gitmodules_if_present(&workspace_root)?;
env::set_current_dir(&workspace_root).with_context(|| {
format!(
"failed to switch process directory to workspace root {}",
workspace_root.display()
)
})?;
let addr = SocketAddr::from(([127, 0, 0, 1], port));
let listener = TcpListener::bind(addr).with_context(|| {
format!(
"port {} is already in use — another wsb instance may be running. \
Stop it first or choose a different port with --port.",
port
)
})?;
listener.set_nonblocking(true)?;
if debug {
eprintln!(
"[wsb] starting web server on port {} from {}",
port,
workspace_root.display()
);
}
let (log_tx, _) = tokio::sync::broadcast::channel::<LogLine>(1024);
let log_ring = LogRingBuffer::new();
set_log_broadcast(log_tx.clone(), log_ring.clone());
let db = ctx.db.clone();
let db_service = ctx.db_service.clone();
let state = WebState::new(
db,
db_service,
workspace_root.clone(),
debug,
port,
log_tx,
log_ring,
)
.await?;
let app = create_router(state.clone());
eprintln!("[wsb] listening on http://{}", addr);
let sync_handle = state.sync.clone();
let ops_registry = state.ops.clone();
tokio::spawn(async move {
if let Err(e) = sync_handle
.sync_incremental_tracked(Some(ops_registry))
.await
{
eprintln!("[wsb] initial sync failed: {}", e);
}
});
handlers::fleet::spawn_git_watcher(state.clone());
handlers::fleet::spawn_git_fs_watcher(state.clone());
spawn_workspace_file_watcher(state.clone());
let watcher_state = state;
tokio::spawn(async move {
if let Err(e) = sync::start_watcher(watcher_state).await {
eprintln!("[wsb] file watcher failed: {}", e);
}
});
if let Some(pid) = parent_pid {
tokio::spawn(async move {
loop {
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
if !pid_is_alive(pid) {
eprintln!("[wsb] parent process {} exited — shutting down", pid);
std::process::exit(0);
}
}
});
}
axum::Server::from_tcp(listener)?
.serve(app.into_make_service())
.await?;
Ok(())
}
#[cfg(unix)]
fn pid_is_alive(pid: u32) -> bool {
let result = unsafe { libc::kill(pid as libc::pid_t, 0) };
result == 0
}
#[cfg(windows)]
fn pid_is_alive(pid: u32) -> bool {
let output = std::process::Command::new("tasklist")
.args(["/FI", &format!("PID eq {}", pid), "/FO", "CSV", "/NH"])
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::null())
.output();
match output {
Ok(output) if output.status.success() => {
let stdout = String::from_utf8_lossy(&output.stdout);
stdout
.lines()
.any(|line| !line.trim().is_empty() && !line.starts_with("INFO:"))
}
_ => false,
}
}
pub fn validate_server_workspace(workspace: Option<PathBuf>) -> Result<PathBuf> {
let start_dir = match workspace {
Some(p) => std::fs::canonicalize(&p)
.with_context(|| format!("workspace path '{}' does not exist", p.display()))?,
None => env::current_dir().context("failed to get current directory")?,
};
let workspace_root = crate::git_backend::repo_root(&start_dir)
.map_err(|_| serve_workspace_error_not_git(&start_dir))?;
let required_paths = [workspace_root.join(".wsb")];
let missing: Vec<String> = required_paths
.iter()
.filter(|path| !path.exists())
.map(|path| {
path.strip_prefix(&workspace_root)
.unwrap_or(path)
.display()
.to_string()
})
.collect();
if !missing.is_empty() {
bail!(
"cannot start `wsb serve` from '{}': '{}' is not an initialized wsb workspace.\n\
required workspace files are missing: {}\n\
initialize this repository first:\n wsb init\n\
or change into an existing initialized workspace and rerun `wsb serve`.",
start_dir.display(),
workspace_root.display(),
missing.join(", ")
);
}
Ok(workspace_root)
}
fn serve_workspace_error_not_git(cwd: &Path) -> anyhow::Error {
anyhow::anyhow!(
"cannot start `wsb serve` from '{}': not inside a git repository.\n\
`wsb serve` only works inside an initialized wsb workspace.\n\
initialize this directory first:\n wsb init\n\
or change into an existing workspace and rerun `wsb serve`.",
cwd.display()
)
}
fn validate_gitmodules_if_present(workspace_root: &Path) -> Result<()> {
let path = workspace_root.join(".gitmodules");
if !path.exists() {
return Ok(());
}
let bytes =
std::fs::read(&path).with_context(|| format!("failed to read {}", path.display()))?;
validate_gitmodules_structure(&bytes, &path).map_err(|e| anyhow::anyhow!("{}", e))
}
fn spawn_workspace_file_watcher(state: WebState) {
use notify_debouncer_mini::{new_debouncer, notify::RecursiveMode, DebouncedEvent};
use std::time::Duration;
let workspace_root = state.workspace_root.clone();
let tx = state.file_changed_tx.clone();
let runtime_handle = tokio::runtime::Handle::current();
std::thread::spawn(move || {
let (event_tx, mut event_rx) = tokio::sync::mpsc::channel::<Vec<DebouncedEvent>>(256);
let rt = runtime_handle.clone();
let mut debouncer = match new_debouncer(
Duration::from_millis(300),
move |result: Result<Vec<DebouncedEvent>, notify::Error>| match result {
Ok(events) => {
let etx = event_tx.clone();
rt.spawn(async move {
let _ = etx.send(events).await;
});
}
Err(e) => {
eprintln!("[wsb] workspace file watcher error: {}", e);
}
},
) {
Ok(d) => d,
Err(e) => {
eprintln!("[wsb] failed to create workspace file watcher: {}", e);
return;
}
};
if let Err(e) = debouncer
.watcher()
.watch(&workspace_root, RecursiveMode::Recursive)
{
eprintln!(
"[wsb] failed to watch workspace root {}: {}",
workspace_root.display(),
e
);
return;
}
let root_clone = workspace_root.clone();
runtime_handle.block_on(async move {
while let Some(events) = event_rx.recv().await {
for event in events {
let path = &event.path;
let rel = match path.strip_prefix(&root_clone) {
Ok(r) => r,
Err(_) => continue,
};
let rel_str = rel.to_string_lossy();
if rel_str.split('/').any(|seg| seg.starts_with('.')) {
continue;
}
if rel_str.starts_with("target/") {
continue;
}
let mtime_secs = std::fs::metadata(path)
.ok()
.and_then(|m| m.modified().ok())
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_secs());
let _ = tx.send(crate::web::state::FileChangedEvent {
path: rel_str.into_owned(),
mtime_secs,
});
}
}
});
});
}
pub(crate) fn home_dir() -> Result<PathBuf> {
if let Some(home) = env::var_os("HOME") {
let path = PathBuf::from(home);
if !path.as_os_str().is_empty() {
return Ok(path);
}
}
std::panic::catch_unwind(dirs::home_dir)
.ok()
.flatten()
.context("cannot determine home directory")
}
fn create_router(state: WebState) -> Router {
let api = Router::new()
.route("/sessions", get(handlers::sessions::list_sessions))
.route("/sessions/:id", get(handlers::sessions::get_session))
.route(
"/sessions/:id/messages",
get(handlers::sessions::get_messages),
)
.route(
"/sessions/:id/children",
get(handlers::sessions::get_children),
)
.route(
"/sessions/:id/minimap",
get(handlers::sessions::get_minimap),
)
.route(
"/sessions/:id/search",
get(handlers::sessions::search_session),
)
.route(
"/sessions/:id/compact",
get(handlers::sessions::get_compact_text),
)
.route(
"/sessions/:id/export",
get(handlers::sessions::export_session),
)
.route(
"/sessions/:id/watch",
get(handlers::sessions::watch_session),
)
.route(
"/sessions/:id/rename",
patch(handlers::sessions::rename_session),
)
.route("/sessions/:id", delete(handlers::sessions::delete_session))
.route(
"/sessions/:id/restore",
post(handlers::sessions::restore_session),
)
.route(
"/sessions/:id/permanent",
delete(handlers::sessions::permanent_delete),
)
.route("/projects", get(handlers::sessions::list_projects))
.route("/machines", get(handlers::sessions::list_machines))
.route("/agents", get(handlers::sessions::list_agents))
.route("/stats", get(handlers::sessions::get_stats))
.route("/analytics/summary", get(handlers::analytics::summary))
.route("/analytics/activity", get(handlers::analytics::activity))
.route("/analytics/heatmap", get(handlers::analytics::heatmap))
.route("/analytics/projects", get(handlers::analytics::projects))
.route(
"/analytics/hour-of-week",
get(handlers::analytics::hour_of_week),
)
.route(
"/analytics/top-sessions",
get(handlers::analytics::top_sessions),
)
.route("/usage/summary", get(handlers::analytics::usage_summary))
.route("/usage/daily", get(handlers::analytics::usage_daily))
.route("/usage/by-model", get(handlers::analytics::usage_by_model))
.route(
"/usage/by-project",
get(handlers::analytics::usage_by_project),
)
.route(
"/usage/sync-pricing",
post(handlers::analytics::sync_pricing),
)
.route(
"/overview/stats",
get(handlers::overview::get_overview_stats),
)
.route(
"/overview/stats/refresh",
post(handlers::overview::refresh_overview_stats),
)
.route("/search", get(handlers::search::global_search))
.route("/sync", post(handlers::sync_handler::trigger_sync))
.route("/resync", post(handlers::sync_handler::trigger_resync))
.route("/sync/status", get(handlers::sync_handler::sync_status))
.route("/starred", get(handlers::starred::list_starred))
.route("/sessions/:id/star", put(handlers::starred::star_session))
.route(
"/sessions/:id/star",
delete(handlers::starred::unstar_session),
)
.route("/starred/bulk", post(handlers::starred::bulk_star))
.route("/path-notes", get(handlers::path_notes::list_path_notes))
.route("/path-notes", post(handlers::path_notes::create_path_note))
.route("/pins", get(handlers::pins::list_all_pins))
.route("/sessions/:id/pins", get(handlers::pins::list_session_pins))
.route(
"/sessions/:id/messages/:message_id/pin",
post(handlers::pins::pin_message),
)
.route(
"/sessions/:id/messages/:message_id/pin",
delete(handlers::pins::unpin_message),
)
.route("/trash", get(handlers::trash::list_trash))
.route("/trash", delete(handlers::trash::empty_trash))
.route("/settings", get(handlers::settings::get_settings))
.route("/settings", put(handlers::settings::update_settings))
.route("/workspace/tree", get(handlers::workspace::list_tree))
.route(
"/workspace/watch",
get(handlers::workspace::workspace_watch),
)
.route("/workspace/file", get(handlers::workspace::read_file))
.route("/workspace/file", post(handlers::workspace::save_file))
.route("/workspace/update", post(handlers::workspace::run_update))
.route(
"/workspace/create-file",
post(handlers::workspace::create_file),
)
.route(
"/workspace/create-directory",
post(handlers::workspace::create_directory),
)
.route("/workspace/delete", post(handlers::workspace::delete_entry))
.route("/workspace/rename", post(handlers::workspace::rename_entry))
.route("/workspace/copy", post(handlers::workspace::copy_entry))
.route(
"/workspace/workbench-model",
get(handlers::extensions::get_workbench_model),
)
.route(
"/workspace/extensions/execute",
post(handlers::extensions::execute_extension_command),
)
.route(
"/workspace/extensions/:extension_id/assets",
get(handlers::extensions::get_extension_asset),
)
.route(
"/workspace/extensions/:extension_id/readme",
get(handlers::extensions::get_extension_readme),
)
.route(
"/marketplace/search",
get(handlers::marketplace::search_marketplace),
)
.route(
"/marketplace/readme",
get(handlers::marketplace::get_marketplace_readme),
)
.route(
"/marketplace/install",
post(handlers::marketplace::install_extension),
)
.route(
"/marketplace/inspect",
post(handlers::marketplace::inspect_marketplace_extension),
)
.route(
"/marketplace/uninstall",
post(handlers::marketplace::uninstall_extension),
)
.route(
"/marketplace/disable",
post(handlers::marketplace::disable_extension),
)
.route(
"/marketplace/enable",
post(handlers::marketplace::enable_extension),
)
.route(
"/workbench/state",
get(handlers::workbench_state::get_workbench_state),
)
.route(
"/workbench/state",
put(handlers::workbench_state::put_workbench_state),
)
.route("/plans/list", get(handlers::plans::list_plans))
.route("/plans/content", get(handlers::plans::get_plan_content))
.route("/version", get(handlers::version::get_version))
.route("/issues", get(handlers::issues::list_issues))
.route("/issues/stream", get(handlers::issues::stream_issues))
.route(
"/issues/:owner/:repo",
get(handlers::issues::list_repo_issues),
)
.route("/issues/:owner/:repo", post(handlers::issues::create_issue))
.route(
"/issues/:owner/:repo/:number",
patch(handlers::issues::update_issue),
)
.route(
"/issues/:owner/:repo/:number/comments",
get(handlers::issues::list_comments),
)
.route(
"/issues/:owner/:repo/:number/comments",
post(handlers::issues::add_comment),
)
.route("/gists", get(handlers::gists::list_gists))
.route("/gists", post(handlers::gists::create_gist))
.route("/gists/:id", get(handlers::gists::get_gist))
.route("/gists/:id", patch(handlers::gists::update_gist))
.route("/fleet/status", get(handlers::fleet::fleet_status))
.route("/fleet/repos", get(handlers::fleet::fleet_repos))
.route(
"/fleet/detailed-status",
get(handlers::fleet::fleet_detailed_status),
)
.route("/fleet/branches", get(handlers::fleet::fleet_branches))
.route("/fleet/diff", get(handlers::fleet::fleet_diff))
.route("/fleet/watch", get(handlers::fleet::fleet_watch))
.route("/fleet/focus", post(handlers::fleet::fleet_focus))
.route("/fleet/stage", post(handlers::fleet::fleet_stage))
.route("/fleet/commit", post(handlers::fleet::fleet_commit))
.route("/fleet/push", post(handlers::fleet::fleet_push))
.route("/fleet/pull", post(handlers::fleet::fleet_pull))
.route("/fleet/checkout", post(handlers::fleet::fleet_checkout))
.route(
"/fleet/branch/create",
post(handlers::fleet::fleet_create_branch),
)
.route("/fleet/stage-all", post(handlers::fleet::fleet_stage_all))
.route("/fleet/commit-all", post(handlers::fleet::fleet_commit_all))
.route("/fleet/push-all", post(handlers::fleet::fleet_push_all))
.route("/fleet/pull-all", post(handlers::fleet::fleet_pull_all))
.route(
"/fleet/checkout-all",
post(handlers::fleet::fleet_checkout_all),
)
.route("/fleet/merge-back", post(handlers::fleet::fleet_merge_back))
.route("/logs", get(handlers::logs::stream_logs))
.route("/ops", get(ops::ops_snapshot))
.route("/ops/stream", get(ops::ops_stream));
Router::new()
.nest("/api/v1", api)
.fallback(spa_handler)
.layer(CorsLayer::permissive())
.with_state(state)
}
async fn spa_handler(uri: Uri) -> impl IntoResponse {
let path = uri.path().trim_start_matches('/');
if let Some(content) = FrontendAssets::get(path) {
let mime = mime_guess::from_path(path).first_or_octet_stream();
let cache_control = if path == "index.html" {
"no-store"
} else {
"public, max-age=31536000, immutable"
};
return Response::builder()
.header(header::CONTENT_TYPE, mime.as_ref())
.header(header::CACHE_CONTROL, cache_control)
.body(boxed(Full::from(content.data)))
.unwrap();
}
match FrontendAssets::get("index.html") {
Some(content) => Response::builder()
.header(header::CONTENT_TYPE, "text/html; charset=utf-8")
.header(header::CACHE_CONTROL, "no-store")
.body(boxed(Full::from(content.data)))
.unwrap(),
None => Response::builder()
.status(StatusCode::NOT_FOUND)
.body(boxed(Full::from("frontend not found")))
.unwrap(),
}
}