use crate::cli::fe_runtime;
use crate::server::{self, ServerConfig, ServerHandle, vite_dev};
use anyhow::{Context, Result};
use fn0::cache::BundleCache;
use http_body_util::BodyExt;
use notify_debouncer_mini::{new_debouncer, notify::RecursiveMode};
use std::collections::HashMap;
use std::fs;
use std::net::{SocketAddr, TcpListener};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::Arc;
use std::time::{Duration, SystemTime};
use tokio::sync::mpsc::unbounded_channel;
#[derive(Debug)]
pub struct DevOptions {
pub project_dir: PathBuf,
pub port: Option<u16>,
}
impl Default for DevOptions {
fn default() -> Self {
Self {
project_dir: PathBuf::from("."),
port: None,
}
}
}
fn is_port_available(port: u16) -> bool {
let addr = SocketAddr::from(([0, 0, 0, 0], port));
TcpListener::bind(addr).is_ok()
}
fn find_available_port(start: u16) -> Option<u16> {
(start..=65535).find(|&port| is_port_available(port))
}
const FORTE_RS_TO_TS_VERSION: &str = "0.1.7";
async fn ensure_forte_rs_to_ts() -> Result<PathBuf> {
let url = crate::tools::fn0_release_url("forte-rs-to-ts", FORTE_RS_TO_TS_VERSION)?;
crate::tools::ensure_github_tool_with_libs(
"forte-rs-to-ts",
FORTE_RS_TO_TS_VERSION,
&url,
"forte-rs-to-ts",
)
.await
}
async fn run_codegen(project_dir: &Path) -> Result<()> {
let rs_dir = project_dir.join("rs");
if !rs_dir.exists() {
fe_runtime::ensure(project_dir)?;
return Ok(());
}
let binary = ensure_forte_rs_to_ts().await?;
let status = Command::new(&binary)
.arg(project_dir)
.stdout(Stdio::null())
.status()
.context("Failed to run forte-rs-to-ts")?;
if !status.success() {
anyhow::bail!("forte-rs-to-ts failed with status: {}", status);
}
fe_runtime::ensure(project_dir)?;
generate_frontend_routes(project_dir)?;
Ok(())
}
#[derive(Debug)]
struct RouteInfo {
path: String,
fe_page_path: String,
}
fn generate_frontend_routes(project_dir: &Path) -> Result<()> {
let pages_dir = project_dir.join("rs/src/pages");
if !pages_dir.exists() {
return Ok(());
}
let prefix = fe_runtime::page_import_prefix(project_dir);
let mut routes = Vec::new();
scan_pages_dir(&pages_dir, &pages_dir, prefix, &mut routes)?;
routes.sort_by(|a, b| {
let a_dynamic = a.path.contains(':');
let b_dynamic = b.path.contains(':');
if a_dynamic != b_dynamic {
return a_dynamic.cmp(&b_dynamic);
}
a.path.cmp(&b.path)
});
let mut output = String::new();
output.push_str("// Auto-generated by forte dev\n\n");
output.push_str("export const routes: Array<{ path: string; component: () => Promise<{ default: (props: any) => any }>; schema: () => Promise<{ PropsSchema: any }> }> = [\n");
for route in &routes {
let fe_props_path = route
.fe_page_path
.strip_suffix("/page")
.map(|s| format!("{}/.props", s))
.unwrap_or_else(|| route.fe_page_path.clone());
output.push_str(&format!(
" {{ path: \"{}\", component: () => import(\"{}\"), schema: () => import(\"{}\") }},\n",
route.path, route.fe_page_path, fe_props_path
));
}
output.push_str("];\n");
let output_path = fe_runtime::routes_generated(project_dir);
if let Some(parent) = output_path.parent() {
fs::create_dir_all(parent)?;
}
fs::write(output_path, output)?;
Ok(())
}
fn scan_pages_dir(
base_dir: &Path,
current_dir: &Path,
page_import_prefix: &str,
routes: &mut Vec<RouteInfo>,
) -> Result<()> {
for entry in fs::read_dir(current_dir)? {
let entry = entry?;
let path = entry.path();
if path.is_dir() {
scan_pages_dir(base_dir, &path, page_import_prefix, routes)?;
} else if path.extension().is_some_and(|ext| ext == "rs")
&& has_handler_function(&path)?
&& let Some(route) = path_to_route(base_dir, &path, page_import_prefix)
{
routes.push(route);
}
}
Ok(())
}
fn has_handler_function(path: &Path) -> Result<bool> {
let content = fs::read_to_string(path)?;
if content.contains("type Props = Redirect") {
return Ok(false);
}
Ok(content.contains("pub async fn handler"))
}
fn path_to_route(base_dir: &Path, file_path: &Path, page_import_prefix: &str) -> Option<RouteInfo> {
let relative = file_path.strip_prefix(base_dir).ok()?;
let relative_str = relative.to_string_lossy();
let mut route_path = relative_str
.trim_end_matches(".rs")
.trim_end_matches("/mod")
.replace('\\', "/");
if route_path == "index" || route_path.is_empty() {
route_path = "/".to_string();
} else {
route_path = route_path
.replace("/index", "")
.replace("[", ":")
.replace("]", "");
if !route_path.starts_with('/') {
route_path = format!("/{}", route_path);
}
}
if route_path.starts_with("/api/") {
return None;
}
let fe_page_path = build_fe_page_path(&route_path, page_import_prefix);
Some(RouteInfo {
path: route_path,
fe_page_path,
})
}
fn build_fe_page_path(route_path: &str, prefix: &str) -> String {
if route_path == "/" {
format!("{}/pages/index/page", prefix)
} else {
let path = route_path
.replace(":", "[")
.split('/')
.filter(|s| !s.is_empty())
.map(|s| {
if s.starts_with('[') {
format!("{}]", s)
} else {
s.to_string()
}
})
.collect::<Vec<_>>()
.join("/");
format!("{}/pages/{}/page", prefix, path)
}
}
fn build_backend(project_dir: &Path) -> Result<()> {
let status = Command::new("cargo")
.arg("build")
.arg("--release")
.arg("--quiet")
.arg("--target")
.arg("wasm32-wasip2")
.current_dir(project_dir.join("rs"))
.status()
.context("Failed to run cargo build")?;
if !status.success() {
anyhow::bail!("cargo build failed with status: {}", status);
}
Ok(())
}
fn find_wasm_binary(release_dir: &Path, project_dir: &Path) -> Result<PathBuf> {
let cargo_toml = project_dir.join("rs/Cargo.toml");
let content = fs::read_to_string(&cargo_toml)
.with_context(|| format!("read {}", cargo_toml.display()))?;
#[derive(serde::Deserialize)]
struct CargoToml {
package: CargoPackage,
}
#[derive(serde::Deserialize)]
struct CargoPackage {
name: String,
}
let parsed: CargoToml =
toml::from_str(&content).with_context(|| format!("parse {}", cargo_toml.display()))?;
let wasm_name = format!("{}.wasm", parsed.package.name.replace('-', "_"));
let wasm_path = release_dir.join(&wasm_name);
if !wasm_path.exists() {
anyhow::bail!(
"expected wasm '{}' not found in {}",
wasm_name,
release_dir.display()
);
}
Ok(wasm_path)
}
fn collect_file_mtimes(dir: &Path, extensions: &[&str]) -> HashMap<PathBuf, SystemTime> {
let mut mtimes = HashMap::new();
if let Ok(entries) = fs::read_dir(dir) {
for entry in entries.flatten() {
let path = entry.path();
if path.is_dir() {
mtimes.extend(collect_file_mtimes(&path, extensions));
} else if let Some(ext) = path.extension()
&& extensions.iter().any(|e| ext == *e)
&& let Ok(metadata) = fs::metadata(&path)
&& let Ok(mtime) = metadata.modified()
{
mtimes.insert(path, mtime);
}
}
}
mtimes
}
pub async fn run(options: DevOptions) -> Result<()> {
let project_dir = options.project_dir.canonicalize()?;
let port = match options.port {
Some(p) => {
if !is_port_available(p) {
eprintln!("Error: Port {} is already in use", p);
std::process::exit(1);
}
p
}
None => {
let p = find_available_port(3000)
.ok_or_else(|| anyhow::anyhow!("No available port found starting from 3000"))?;
if p != 3000 {
println!("Port 3000 in use, using port {} instead", p);
}
p
}
};
let sqld_port = find_available_port(8080)
.ok_or_else(|| anyhow::anyhow!("No available port found for sqld starting from 8080"))?;
let mut _sqld = crate::sqld::start(&project_dir, sqld_port).await?;
println!("sqld running on port {}", sqld_port);
run_codegen(&project_dir).await?;
build_backend(&project_dir)?;
let fe_dir = project_dir.join("fe");
let vite_config = fe_runtime::vite_config(&project_dir);
let ssr_module_path = fe_runtime::ssr_load_module_path(&project_dir);
let vite = vite_dev::spawn_vite(&fe_dir, port, vite_config.as_deref(), ssr_module_path)?;
vite_dev::wait_for_vite_ready(&vite.socket_path).await?;
let wasm_path = find_wasm_binary(
&project_dir.join("rs/target/wasm32-wasip2/release"),
&project_dir,
)?
.to_string_lossy()
.to_string();
let js_path = String::new();
let public_dir = project_dir.join("fe/public");
let mut env_vars = server::load_env_file(&project_dir);
if !env_vars.iter().any(|(k, _)| k == "TURSO_URL") {
env_vars.push((
"TURSO_URL".to_string(),
format!("http://127.0.0.1:{}", sqld_port),
));
}
let (queue_tx, queue_rx) =
tokio::sync::mpsc::unbounded_channel::<fn0::queue_hijack::LoopbackMessage>();
let queue_placeholder = "fn0-queue.fn0.dev".to_string();
let queue_tx_for_cron = queue_tx.clone();
let queue_hijack = Arc::new(fn0::QueueHijack::new_loopback(
queue_placeholder.clone(),
queue_tx,
));
if !env_vars.iter().any(|(k, _)| k == "FN0_QUEUE_URL") {
env_vars.push((
"FN0_QUEUE_URL".to_string(),
format!("http://{queue_placeholder}"),
));
}
let config = ServerConfig {
port,
wasm_path,
js_path,
public_dir,
vite_socket_path: Some(vite.socket_path.clone()),
env_vars,
queue_hijack: Some(queue_hijack),
};
let handle = server::run(config).await?;
let _cron_handle = tokio::task::spawn_local({
let project_dir = project_dir.clone();
async move {
run_local_cron_ticker(project_dir, queue_tx_for_cron).await;
}
});
let _consumer_handle = tokio::task::spawn_local({
let executor = handle.executor.clone();
let mut queue_rx = queue_rx;
async move {
while let Some(msg) = queue_rx.recv().await {
let body = serde_json::json!({
"task_name": msg.task_name,
"payload": msg.payload,
});
let body_bytes = match serde_json::to_vec(&body) {
Ok(b) => b,
Err(err) => {
tracing::warn!(?err, "loopback queue: serialize body failed");
continue;
}
};
let req = match hyper::Request::builder()
.method("POST")
.uri("http://localhost/__fn0_queue_task/execute")
.header("content-type", "application/json")
.body(
http_body_util::Full::new(bytes::Bytes::from(body_bytes))
.map_err(|never: std::convert::Infallible| match never {})
.boxed_unsync(),
) {
Ok(r) => r,
Err(err) => {
tracing::warn!(?err, "loopback queue: build request failed");
continue;
}
};
if let Err(err) = executor.run(&msg.subdomain, "", req, None).await {
tracing::warn!(?err, task = %msg.task_name, "loopback queue task failed");
}
}
}
});
let mut vite = vite;
let result = run_watch_loop(&project_dir, handle).await;
let _ = vite.child.kill();
_sqld.kill();
result
}
async fn run_watch_loop(project_dir: &Path, handle: ServerHandle) -> Result<()> {
let (tx, mut rx) = unbounded_channel();
let mut debouncer = new_debouncer(Duration::from_millis(100), move |evt| {
let _ = tx.send(evt);
})?;
let rs_dir = project_dir.join("rs/src");
let env_file = project_dir.join(".env");
debouncer
.watcher()
.watch(&rs_dir, RecursiveMode::Recursive)?;
if env_file.exists() {
debouncer
.watcher()
.watch(&env_file, RecursiveMode::NonRecursive)?;
}
let mut known_rs_mtimes = collect_file_mtimes(&rs_dir, &["rs"]);
let mut known_env_mtime = fs::metadata(&env_file).ok().and_then(|m| m.modified().ok());
while let Some(evt_result) = rx.recv().await {
match evt_result {
Ok(events) => {
let rs_changes: Vec<_> = events
.iter()
.filter(|e| {
e.path.starts_with(&rs_dir)
&& e.path.extension().is_some_and(|ext| ext == "rs")
&& !e.path.ends_with("route_generated.rs")
})
.filter(|e| {
let current_mtime =
fs::metadata(&e.path).ok().and_then(|m| m.modified().ok());
match (current_mtime, known_rs_mtimes.get(&e.path)) {
(Some(current), Some(known)) => current > *known,
(Some(_), None) => true,
_ => false,
}
})
.collect();
let env_changed = events.iter().any(|e| e.path == env_file) && {
let current_mtime =
fs::metadata(&env_file).ok().and_then(|m| m.modified().ok());
match (current_mtime, known_env_mtime) {
(Some(current), Some(known)) => current > known,
(Some(_), None) => true,
_ => false,
}
};
if env_changed {
known_env_mtime = fs::metadata(&env_file).ok().and_then(|m| m.modified().ok());
let new_vars = server::load_env_file(project_dir);
handle.ctx.bundle_cache().set_env(new_vars).await;
}
if !rs_changes.is_empty() {
let result = rebuild_backend(project_dir, &handle).await;
known_rs_mtimes = collect_file_mtimes(&rs_dir, &["rs"]);
while rx.try_recv().is_ok() {}
if let Err(e) = result {
eprintln!("[watch] Backend rebuild failed: {}", e);
}
}
}
Err(error) => {
eprintln!("[watch] Error: {:?}", error);
}
}
}
Ok(())
}
async fn rebuild_backend(project_dir: &Path, handle: &ServerHandle) -> Result<()> {
run_codegen(project_dir).await?;
build_backend(project_dir)?;
handle
.ctx
.bundle_cache()
.invalidate(server::DEV_CODE_ID)
.await;
Ok(())
}
async fn run_local_cron_ticker(
project_dir: PathBuf,
queue_tx: tokio::sync::mpsc::UnboundedSender<fn0::queue_hijack::LoopbackMessage>,
) {
let mut interval = tokio::time::interval_at(next_minute_boundary(), Duration::from_secs(60));
loop {
interval.tick().await;
let now_secs = match std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) {
Ok(d) => d.as_secs() as i64,
Err(_) => continue,
};
let epoch_minute = now_secs / 60;
let jobs = match super::cron::read_and_validate(&project_dir) {
Ok(j) => j,
Err(err) => {
tracing::warn!(?err, "local cron: cron.yaml read/validate failed");
continue;
}
};
for job in jobs {
if job.every_minutes == 0 {
continue;
}
if epoch_minute % (job.every_minutes as i64) != 0 {
continue;
}
let msg = fn0::queue_hijack::LoopbackMessage {
subdomain: server::DEV_CODE_ID.to_string(),
task_name: job.function.clone(),
payload: serde_json::Value::Null,
};
if let Err(err) = queue_tx.send(msg) {
tracing::warn!(?err, "local cron: enqueue failed");
} else {
tracing::info!(function = %job.function, "local cron fired");
}
}
}
}
fn next_minute_boundary() -> tokio::time::Instant {
let now = std::time::SystemTime::now();
let dur = now
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default();
let secs_in_min = dur.as_secs() % 60;
let wait = Duration::from_secs(60 - secs_in_min);
tokio::time::Instant::now() + wait
}