use std::io;
use std::net::SocketAddr;
use std::path::Path;
use std::sync::Arc;
use zebflow::infra::cluster::config::ClusterRole;
use zebflow::infra::health::{
HealthState, spawn_main_runtime_heartbeat, start_dedicated_health_server,
};
use zebflow::platform::boot::{
configured_host, configured_port, default_data_root, load_platform_config,
};
use zebflow::platform::cli;
use zebflow::platform::services::PlatformService;
use zebflow::platform::services::project::webhook_triggers_from_source;
use zebflow::platform::services::{
DependencyLockService, ProjectConfigurationService, ProjectService,
};
use zebflow::platform::web;
use zebflow::platform::{FileAdapterKind, PlatformConfig, build_router};
use zebflow::provision::k8s as k8s_provision;
use zebflow::version::APP_VERSION;
async fn shutdown_signal(health_state: Option<Arc<HealthState>>) {
use tokio::signal;
let ctrl_c = async {
signal::ctrl_c()
.await
.expect("failed to install Ctrl+C handler");
};
#[cfg(unix)]
let terminate = async {
signal::unix::signal(signal::unix::SignalKind::terminate())
.expect("failed to install SIGTERM handler")
.recv()
.await;
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
tokio::select! {
_ = ctrl_c => {},
_ = terminate => {},
}
if let Some(state) = health_state {
state.mark_shutdown_requested();
}
eprintln!("Zebflow: graceful shutdown initiated; draining in-flight requests...");
tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
eprintln!("Zebflow: drain deadline passed with connections still open; exiting.");
std::process::exit(0);
});
}
fn configured_health_addr(default_host: &str) -> Result<Option<SocketAddr>, io::Error> {
let Some(raw_port) = std::env::var("ZEBFLOW_HEALTH_PORT").ok() else {
return Ok(None);
};
let trimmed = raw_port.trim();
if trimmed.is_empty() || trimmed == "0" {
return Ok(None);
}
let port = trimmed.parse::<u16>().map_err(|err| {
io::Error::new(
io::ErrorKind::InvalidInput,
format!("invalid ZEBFLOW_HEALTH_PORT '{trimmed}': {err}"),
)
})?;
let host = std::env::var("ZEBFLOW_HEALTH_HOST").unwrap_or_else(|_| default_host.to_string());
let addr = format!("{host}:{port}")
.parse::<SocketAddr>()
.map_err(|err| {
io::Error::new(
io::ErrorKind::InvalidInput,
format!("invalid health listen address {host}:{port}: {err}"),
)
})?;
Ok(Some(addr))
}
fn maybe_start_dedicated_health_server(
host: &str,
) -> Result<Option<(Arc<HealthState>, std::thread::JoinHandle<()>, SocketAddr)>, io::Error> {
let Some(addr) = configured_health_addr(host)? else {
return Ok(None);
};
let state = HealthState::new();
let handle = start_dedicated_health_server(addr, state.clone())?;
spawn_main_runtime_heartbeat(state.clone());
Ok(Some((state, handle, addr)))
}
fn display_mode(role: ClusterRole) -> &'static str {
match role {
ClusterRole::Standalone => "standalone (controller + office)",
ClusterRole::Master => "controller",
ClusterRole::Worker => "office",
}
}
fn top_level_help() -> String {
format!(
"Zebflow {version}
Usage:
{zeb} [standalone]
{zeb} install <ref> [--repo <repository-id>] [--yes]
{zeb} remove <owner>/<project>
{zeb} list
{zeb} status
{zeb} login <instance-url> [--user <name>] [--password <pw>]
{zeb} use <owner>/<project>
{zeb} logout
{zeb} project install <ref>
{zeb} project remove <owner>/<project>
{zeb} project list
{zeb} run <owner>/<project>
{zeb} project config migrate <owner> <project>
{zeb} project lock migrate <owner> <project>
{zeb} project pipelines migrate <owner> <project>
{zeb} admin reset-password <owner>
{zeb} admin break-glass [<owner>]
{zeb} admin detach
{zeb} controller
{zeb} office
{zeb} k8s cluster <command> ...
{zeb} help
{zeb} --help
{zeb} --version
Runtime Modes:
standalone Start the combined controller + office server (default)
run Serve one installed project's public route (install it first)
controller Start the control-plane oriented server
office Start the execution-plane oriented server
`master` and `worker` are deprecated spellings of `controller` and `office`.
They still start the same role, print a warning, and will be removed.
{client_help}
Project Maintenance (offline: these run against a data directory with no server,
because they are what you run when the server will not start):
{zeb} project config migrate <owner> <project>
Explicitly migrate repo/zebflow.json to repo/zebflow.yaml
{zeb} project lock migrate <owner> <project>
Explicitly migrate the pre-v1 repo/zeb.lock format
{zeb} project pipelines migrate <owner> <project>
Rewrite persisted pipeline ids to the source-relative form
{zeb} admin reset-password <owner>
Rotate the account's password to a fresh random one, printed
once. The account is marked `generated` again, so the next
browser login is forced to choose a new password. Refuses when
a server answers on this machine's configured address; a server
listening on an unusual port is not detected, so make sure none
holds the data root before running this.
{zeb} admin break-glass [<owner>]
Re-enable local password login on an office that has joined a
controller. Needs no controller and no quorum: host access is
the protection, so stop the server and run this on the office
itself. With an owner, that account's password is rotated in the
same act and printed once. The office stays joined -- this
changes who may open the door, not who the office belongs to.
The use is recorded locally, readable at
GET /api/platform/office/local-authority, and reported to the controller
on the next successful registration.
{zeb} admin detach
Leave the controller. The office keeps its projects, data,
files, blessed shelf, and public surface, and its local accounts
go live again -- they were disabled while joined, never deleted.
Unset ZEBFLOW_CLUSTER_JOIN_TOKEN afterwards or the variable
joins the instance straight back on its next start.
Kubernetes:
{zeb} k8s cluster init <path>
{zeb} k8s cluster add-office <path> <office-id>
{zeb} k8s cluster set-controller <path> <office-id>
{zeb} k8s cluster set-namespace <path> <namespace>
{zeb} k8s cluster set-resource-suffix <path> <suffix>
{zeb} k8s cluster set-image <path> <image>
{zeb} k8s cluster use-secret <path> <secret-name>
{zeb} k8s cluster set-replicas <path> <replicas>
{zeb} k8s cluster enable-precreate-pvcs <path>
{zeb} k8s cluster disable-precreate-pvcs <path>
{zeb} k8s cluster enable-auto-update <path>
{zeb} k8s cluster disable-auto-update <path>
{zeb} k8s cluster render-copy-jobs <source-path> <target-path> <output-file>
{zeb} k8s cluster describe <path>
{zeb} k8s cluster validate <path>
Environment - process shape (where this process listens and stores):
ZEBFLOW_PLATFORM_HOST Listen host (default: 127.0.0.1)
ZEBFLOW_PLATFORM_PORT Listen port (default: 10610)
ZEBFLOW_PLATFORM_DATA_DIR Data root. Unset, the OS user data path is used:
~/.local/share/zebflow, ~/Library/Application Support/Zebflow,
or %LOCALAPPDATA%\\Zebflow. Never a path relative to the
working directory
ZEBFLOW_PLATFORM_BASE_URL External base URL used for OAuth redirect and MCP session
URLs (default: derived from the request headers)
ZEBFLOW_HEALTH_PORT Optional dedicated liveness port, e.g. 10611
ZEBFLOW_HEALTH_HOST Dedicated liveness host (default: ZEBFLOW_PLATFORM_HOST)
Environment - first-boot bootstrap (server state created on first start; not CLI context):
ZEBFLOW_PLATFORM_DEFAULT_OWNER Owner account created on first boot (default: superadmin)
ZEBFLOW_PLATFORM_DEFAULT_PROJECT Project created for that owner on first boot
(default: default)
ZEBFLOW_PLATFORM_DEFAULT_PASSWORD Initial password for that owner; when unset a random one is
generated into <data-dir>/.bootstrap/superadmin-password
ZEBFLOW_PLATFORM_ALLOW_INSECURE_DEFAULT_PASSWORD
Set to 1 to permit the literal password 'secret'
(disposable local development only)
Environment - cluster membership (controller and office):
ZEBFLOW_CLUSTER_JOIN_TOKEN This office's own join token, shaped
zfjoin2:<office_id>:<controller_verify_key>:<secret>, minted
on the controller (POST /api/platform/cluster/join-tokens); required
by office for its first join only. It is then stored at
<data-dir>/platform/office-join-token (0600) and read from
there, so a joined office restarts with this unset. A
controller needs no token of its own; it keeps the private
half of the verification key at
<data-dir>/platform/cluster-signing-key (0600).
ZEBFLOW_CLUSTER_MASTER_URL Controller base URL an office registers with; required
by office
ZEBFLOW_CLUSTER_ADVERTISE_URL Base URL this node advertises to the control plane
(default: this process's own listen URL)
ZEBFLOW_CLUSTER_NODE_ID Stable node id (default: the role name)
ZEBFLOW_CLUSTER_NODE_LABEL Human-readable node label (default: the node id)
A controller or office missing any variable marked required above refuses to start and
names every missing one at once.
Environment - sessions and tokens:
ZEBFLOW_COOKIE_SECURE Force the Secure attribute on session cookies
(default: on unless the listen host is loopback)
ZEBFLOW_SECRET_ROTATION_EPOCH Unix timestamp; invalidate older platform-issued tokens
Environment - hub:
ZEBFLOW_HUB_DEFAULT_BASE_URL Default platform hub API URL
(default: https://hub.zebflow.com/api)
ZEBFLOW_HUB_ALLOW_LOCALHOST_REMOTE Set to 1 to allow localhost hub remotes (local development
only; production must leave this unset)
Environment - rendering engine (advanced):
ZEBFLOW_PLATFORM_RWE_ENGINE_ID Reactive web engine id for the platform admin UI
ZEBFLOW_RWE_ENGINE_ID Reactive web engine id for project pipeline rendering
ZEBFLOW_RWE_PREWARM Set to 0 to disable post-compile SSR warmup
Use `{zeb} k8s --help` for the file-based Kubernetes cluster manager.",
zeb = cli::program(),
version = APP_VERSION,
client_help = cli::help_section()
)
}
fn print_top_level_help() {
println!("{}", top_level_help());
}
fn print_version() {
println!("{APP_VERSION}");
}
async fn project_command(args: &[String]) -> Result<(), Box<dyn std::error::Error>> {
match args.first().map(String::as_str) {
Some("install") => cli::run_install(&args[1..]).await.map_err(Into::into),
Some("remove") => cli::run_remove(&args[1..]).await.map_err(Into::into),
Some("list") => cli::run_list(&args[1..]).await.map_err(Into::into),
Some("config") | Some("lock") | Some("pipelines") => project_maintenance(args),
_ => Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"usage: {zeb} project <install|remove|list> ...\n \
{zeb} project <config|lock|pipelines> migrate <owner> <project>",
zeb = cli::program()
),
)
.into()),
}
}
fn project_maintenance(args: &[String]) -> Result<(), Box<dyn std::error::Error>> {
if args.len() != 4 || args[1] != "migrate" {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"usage: {} project <config|lock|pipelines> migrate <owner> <project>",
cli::program()
),
)
.into());
}
let data_root = default_data_root();
match args[0].as_str() {
"config" => {
let service = ProjectConfigurationService::new(data_root.join("users"));
let result = service.migrate_legacy_json(&args[2], &args[3])?;
println!("Project configuration migrated ({})", result.source_format);
println!("Canonical: {}", result.canonical_path.display());
println!("Recovery: {}", result.recovery_path.display());
Ok(())
}
"lock" => {
let service = DependencyLockService::new(data_root.join("users"));
let result = service.migrate_legacy(&args[2], &args[3])?;
println!("Dependency lock migrated ({})", result.source_format);
println!("Canonical: {}", result.canonical_path.display());
println!("Recovery: {}", result.recovery_path.display());
Ok(())
}
"pipelines" => {
let result = pipeline_identity_service(&data_root)?
.migrate_pipeline_identity(&args[2], &args[3])?;
println!(
"Pipeline identity migrated ({} converted, {} already source-relative)",
result.rewrites.len(),
result.already_canonical
);
for rewrite in &result.rewrites {
println!(" {} -> {}", rewrite.from, rewrite.to);
}
if !result.bootstrap_activate.is_empty() {
println!("spec.bootstrap.activate: {:?}", result.bootstrap_activate);
}
match result.recovery_path {
Some(path) => println!("Recovery: {}", path.display()),
None => println!("Nothing to convert; no recovery copy written."),
}
Ok(())
}
_ => Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"usage: {} project <config|lock|pipelines> migrate <owner> <project>",
cli::program()
),
)
.into()),
}
}
fn pipeline_identity_service(
data_root: &Path,
) -> Result<ProjectService, Box<dyn std::error::Error>> {
let users_root = data_root.join("users");
let data = zebflow::platform::adapters::data::build_data_adapter(
PlatformConfig::default().data_adapter,
data_root,
)?;
let configs = Arc::new(ProjectConfigurationService::new(users_root.clone()));
let file = zebflow::platform::adapters::file::build_file_adapter(
FileAdapterKind::Filesystem,
data_root.to_path_buf(),
configs.clone(),
Some(data.clone()),
);
file.initialize()?;
Ok(ProjectService::new(
data,
file,
zebflow::platform::adapters::project_data::build_project_data_factory(data_root),
configs,
Arc::new(DependencyLockService::new(users_root)),
))
}
async fn run_server(role: ClusterRole) -> Result<(), Box<dyn std::error::Error>> {
let host = configured_host();
let port = configured_port();
let health = maybe_start_dedicated_health_server(&host)?;
let config = load_platform_config(role)?;
let data_root = config.data_root.clone();
let app = build_router(config).await.map_err(io::Error::other)?;
let addr: SocketAddr = format!("{host}:{port}").parse()?;
let listener = tokio::net::TcpListener::bind(addr).await?;
println!("Zebflow v{APP_VERSION} listening on http://{addr}");
if let Some((_, _, health_addr)) = &health {
println!("Dedicated liveness: http://{health_addr}/health/runtime");
}
println!("Mode: {}", display_mode(role));
println!("Data: {}", data_root.display());
println!("Flow: /login -> /home -> /projects/{{owner}}/{{project}}");
axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>())
.with_graceful_shutdown(shutdown_signal(
health.as_ref().map(|(state, _, _)| state.clone()),
))
.await?;
eprintln!("Zebflow: shutdown complete.");
Ok(())
}
#[derive(Debug, Clone)]
struct RunRequest {
owner: Option<String>,
project: String,
}
fn parse_run_request(args: &[String]) -> Result<RunRequest, io::Error> {
let Some(target) = args.first() else {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"usage: {} run <owner>/<project>\n\nA project has to be installed before it can \
be served: `{} install <ref>`.",
cli::program(),
cli::program()
),
));
};
let mut owner_flag = None;
let mut project_flag = None;
let mut index = 1;
while index < args.len() {
match args[index].as_str() {
"--owner" => {
index += 1;
let Some(value) = args.get(index) else {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"missing value for --owner",
));
};
owner_flag = Some(value.clone());
}
"--project" => {
index += 1;
let Some(value) = args.get(index) else {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"missing value for --project",
));
};
project_flag = Some(value.clone());
}
other => {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("unknown run flag '{other}'"),
));
}
}
index += 1;
}
let (reference_owner, project) = match target.split_once('/') {
Some((left, right)) => (Some(left.trim().to_string()), right.trim().to_string()),
None => (None, target.trim().to_string()),
};
if project.is_empty() || project.contains('/') {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("'{target}' is not <owner>/<project>"),
));
}
if let Some(flag) = owner_flag.as_deref()
&& reference_owner
.as_deref()
.is_some_and(|named| named != flag.trim())
{
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("'{target}' and --owner name different owners; give the owner once"),
));
}
if let Some(flag) = project_flag.as_deref()
&& flag.trim() != project
{
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("'{target}' and --project name different projects; give it once"),
));
}
Ok(RunRequest {
owner: reference_owner.or(owner_flag),
project,
})
}
fn choose_public_app_path(
platform: &PlatformService,
owner: &str,
project: &str,
) -> Result<String, io::Error> {
let metas = platform
.projects
.list_pipeline_meta_rows(owner, project)
.map_err(|err| io::Error::other(err.to_string()))?;
let mut preferred_root = None::<String>;
let mut preferred_get = None::<String>;
let mut fallback = None::<String>;
for meta in metas {
let source = platform
.projects
.read_pipeline_source(owner, project, &meta.file_rel_path)
.map_err(|err| io::Error::other(err.to_string()))?;
if let Some(triggers) = webhook_triggers_from_source(&source) {
for trigger in triggers {
if trigger.method == "GET" && trigger.path == "/" {
preferred_root = Some(trigger.path);
break;
}
if preferred_get.is_none() && trigger.method == "GET" {
preferred_get = Some(trigger.path.clone());
}
if fallback.is_none() {
fallback = Some(trigger.path.clone());
}
}
}
if preferred_root.is_some() {
break;
}
}
preferred_root
.or(preferred_get)
.or(fallback)
.ok_or_else(|| io::Error::other("no webhook-triggered public route found for project"))
}
async fn run_project(req: RunRequest) -> Result<(), Box<dyn std::error::Error>> {
let config = load_platform_config(ClusterRole::Standalone)?;
let host = configured_host();
let port = configured_port();
let health = maybe_start_dedicated_health_server(&host)?;
let Some(owner) = req.owner.clone() else {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"run needs an owner; write `{zeb} run <owner>/{project}` or pass `--owner <name>`",
zeb = cli::program(),
project = req.project
),
)
.into());
};
let project = req.project.clone();
let platform = Arc::new(PlatformService::from_config(config)?);
let exists = platform
.projects
.list_projects(&owner)?
.into_iter()
.any(|item| item.project == project);
if !exists {
return Err(io::Error::new(
io::ErrorKind::NotFound,
format!(
"project '{owner}/{project}' is not installed in this data root; install it \
first with `{} install <ref>`",
cli::program()
),
)
.into());
}
platform
.cluster_runtime_sync
.refresh_local_repo_state(&owner, &project)?;
let public_path = choose_public_app_path(platform.as_ref(), &owner, &project)?;
let app_url = format!(
"http://{}:{}/wh/{owner}/{project}{}",
if host == "0.0.0.0" {
"127.0.0.1"
} else {
host.as_str()
},
port,
public_path
);
let app = web::router(platform).await;
let addr: SocketAddr = format!("{host}:{port}").parse()?;
let listener = tokio::net::TcpListener::bind(addr).await?;
println!("Zebflow app running for {owner}/{project}");
if let Some((_, _, health_addr)) = &health {
println!("Dedicated liveness: http://{health_addr}/health/runtime");
}
println!("Public route: {app_url}");
println!("Mode: standalone app runtime");
axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>())
.with_graceful_shutdown(shutdown_signal(
health.as_ref().map(|(state, _, _)| state.clone()),
))
.await?;
eprintln!("Zebflow: shutdown complete.");
Ok(())
}
#[tokio::main]
pub async fn main() {
if let Err(err) = run().await {
eprintln!("Error: {err}");
std::process::exit(1);
}
}
async fn run() -> Result<(), Box<dyn std::error::Error>> {
let mut argv = std::env::args();
cli::set_program_name(argv.next().as_deref());
let mut args = argv;
let mode = args.next();
match mode.as_deref() {
None => run_server(ClusterRole::Standalone).await,
Some("run") => run_project(parse_run_request(&args.collect::<Vec<_>>())?).await,
Some("project") => project_command(&args.collect::<Vec<_>>()).await,
Some("admin") => cli::admin::run(&args.collect::<Vec<_>>())
.await
.map_err(Into::into),
Some("install") => cli::run_install(&args.collect::<Vec<_>>())
.await
.map_err(Into::into),
Some("remove") => cli::run_remove(&args.collect::<Vec<_>>())
.await
.map_err(Into::into),
Some("list") => cli::run_list(&args.collect::<Vec<_>>())
.await
.map_err(Into::into),
Some("status") => cli::run_status(&args.collect::<Vec<_>>())
.await
.map_err(Into::into),
Some("login") => cli::run_login(&args.collect::<Vec<_>>())
.await
.map_err(Into::into),
Some("use") => cli::run_use(&args.collect::<Vec<_>>()).map_err(Into::into),
Some("logout") => cli::run_logout(&args.collect::<Vec<_>>())
.await
.map_err(Into::into),
Some("help") | Some("--help") | Some("-h") => {
print_top_level_help();
Ok(())
}
Some("version") | Some("--version") | Some("-V") => {
print_version();
Ok(())
}
Some("k8s") => k8s_provision::run_cli(&args.collect::<Vec<_>>()),
Some(other) => match ClusterRole::from_mode_arg(other) {
Some(mode) => {
if let Some(canonical) = mode.deprecated_alias_of {
eprintln!(
"Zebflow: `{zeb} {other}` is a deprecated spelling of `{zeb} {canonical}` and will be removed; use `{zeb} {canonical}`.",
zeb = cli::program()
);
}
run_server(mode.role).await
}
None => Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"unknown {} mode '{other}'\n\n{}",
cli::program(),
top_level_help()
),
)
.into()),
},
}
}