use anyhow::Result;
use clap::{Parser, ValueEnum};
use std::collections::HashMap;
use std::fs::{self, OpenOptions};
use std::io::Write;
use std::path::PathBuf;
use tokio::sync::mpsc;
use tracing::{debug, error};
use crate::{error::CliError, CommandContext};
use theater::chain::ChainEvent;
use theater::config::actor_manifest::{
RuntimeHostConfig, StoreHandlerConfig, SupervisorHostConfig, TcpHandlerConfig,
TerminalHandlerConfig, TimerHandlerConfig,
};
use theater::handler::HandlerRegistry;
use theater::messages::TheaterCommand;
use theater::pack_bridge::{Value, ValueType};
use theater::theater_runtime::TheaterRuntime;
use theater::utils::resolve_reference;
use theater::ManifestConfig;
use theater::TheaterId;
use theater_handler_loop::LoopHandler;
use theater_handler_message_server::{MessageRouter, MessageServerHandler};
use theater_handler_rpc::RpcHandler;
use theater_handler_runtime::RuntimeHandler;
use theater_handler_store::StoreHandler;
use theater_handler_supervisor::SupervisorHandler;
use theater_handler_tcp::TcpHandler;
use theater_handler_terminal::TerminalHandler;
use theater_handler_timer::TimerHandler;
#[derive(Debug, Clone, Copy, ValueEnum, Default)]
pub enum EventFormat {
Json,
#[default]
Short,
Full,
}
#[derive(Debug, Parser)]
pub struct StartArgs {
#[arg(default_value = "manifest.toml")]
pub manifest: String,
#[arg(long)]
pub events: bool,
#[arg(long, value_enum, default_value = "short")]
pub events_format: EventFormat,
#[arg(long)]
pub chain_dir: Option<PathBuf>,
#[arg(long)]
pub no_init: bool,
#[arg(long)]
pub no_actor_logs: bool,
}
fn format_event_short(event: &ChainEvent, actor_id: &TheaterId) -> String {
let id_str = actor_id.to_string();
let short_id = &id_str[..8.min(id_str.len())];
format!("[{}] {}\n", short_id, event)
}
fn format_event_full(event: &ChainEvent, actor_id: &TheaterId) -> String {
let id_str = actor_id.to_string();
let short_id = &id_str[..8.min(id_str.len())];
let hash_hex = hex::encode(&event.hash);
let parent_hex = event
.parent_hash
.as_ref()
.map(hex::encode)
.unwrap_or_else(|| "none".to_string());
let data_str = String::from_utf8_lossy(&event.data);
format!(
"EVENT [{}] {}\nparent: {}\ntype: {}\nsize: {}\n{}\n\n",
short_id,
hash_hex,
parent_hex,
event.event_type,
event.data.len(),
data_str
)
}
fn format_event_json(event: &ChainEvent, actor_id: &TheaterId) -> String {
let json = serde_json::json!({
"actor_id": actor_id.to_string(),
"hash": hex::encode(&event.hash),
"parent_hash": event.parent_hash.as_ref().map(hex::encode),
"event_type": event.event_type,
"data": format!("{} bytes (pack-encoded)", event.data.len())
});
serde_json::to_string(&json).unwrap_or_else(|_| "{}".to_string())
}
struct ChainFileManager {
dir: PathBuf,
files: HashMap<TheaterId, std::fs::File>,
}
impl ChainFileManager {
fn new(dir: PathBuf) -> Result<Self, CliError> {
fs::create_dir_all(&dir).map_err(|e| {
CliError::file_operation_failed("create directory", dir.display().to_string(), e)
})?;
Ok(Self {
dir,
files: HashMap::new(),
})
}
fn write_event(&mut self, actor_id: &TheaterId, event: &ChainEvent) -> Result<(), CliError> {
let file = self.files.entry(*actor_id).or_insert_with(|| {
let path = self.dir.join(format!("{}.chain", actor_id));
OpenOptions::new()
.create(true)
.append(true)
.open(&path)
.expect("Failed to open chain file")
});
let block = format_event_full(event, actor_id);
file.write_all(block.as_bytes()).map_err(|e| {
CliError::file_operation_failed("write event", format!("{}.chain", actor_id), e)
})?;
file.flush().map_err(|e| {
CliError::file_operation_failed("flush", format!("{}.chain", actor_id), e)
})?;
Ok(())
}
}
fn create_handler_registry(
theater_tx: mpsc::Sender<TheaterCommand>,
show_actor_logs: bool,
) -> HandlerRegistry {
let mut registry = HandlerRegistry::new();
let runtime_config = RuntimeHostConfig {};
registry.register(
RuntimeHandler::new(runtime_config, theater_tx.clone(), None)
.with_show_logs(show_actor_logs),
);
let store_config = StoreHandlerConfig::default();
registry.register(StoreHandler::new(store_config, None));
let supervisor_config = SupervisorHostConfig {};
registry.register(SupervisorHandler::new(supervisor_config, None));
let message_router = MessageRouter::new();
registry.register(MessageServerHandler::new(None, message_router.clone()));
registry.register(RpcHandler::new(theater_tx.clone()));
let tcp_config = TcpHandlerConfig {
listen: None,
max_connections: None,
..Default::default()
};
registry.register(TcpHandler::new(tcp_config));
let terminal_config = TerminalHandlerConfig::default();
registry.register(TerminalHandler::new(terminal_config));
let timer_config = TimerHandlerConfig::default();
registry.register(TimerHandler::new(timer_config));
registry.register(LoopHandler::new());
registry
}
pub async fn execute_async(args: &StartArgs, ctx: &CommandContext) -> Result<(), CliError> {
debug!("Starting actor from manifest: {}", args.manifest);
let manifest_bytes = resolve_reference(&args.manifest).await.map_err(|e| {
CliError::invalid_manifest(format!(
"Failed to resolve manifest reference '{}': {}",
args.manifest, e
))
})?;
let manifest_content = String::from_utf8(manifest_bytes).map_err(|e| {
CliError::invalid_manifest(format!("Manifest content is not valid UTF-8: {}", e))
})?;
let mut chain_file_manager = if let Some(ref dir) = args.chain_dir {
Some(ChainFileManager::new(dir.clone())?)
} else {
None
};
let (theater_tx, theater_rx) = mpsc::channel::<TheaterCommand>(32);
let handler_registry = create_handler_registry(theater_tx.clone(), !args.no_actor_logs);
let mut runtime = TheaterRuntime::new(
theater_tx.clone(),
theater_rx,
None, handler_registry,
)
.await
.map_err(|e| CliError::server_error(format!("Failed to create runtime: {}", e)))?;
let (global_events_tx, mut global_events_rx) = mpsc::channel(256);
runtime.add_global_subscription(global_events_tx);
let runtime_handle = tokio::spawn(async move {
if let Err(e) = runtime.run().await {
error!("Theater runtime error: {}", e);
}
});
let manifest = ManifestConfig::from_toml_str(&manifest_content)
.map_err(|e| CliError::invalid_manifest(format!("Failed to parse manifest: {}", e)))?;
let wasm_path = if manifest.package.starts_with('/') || manifest.package.contains("://") {
manifest.package.clone()
} else {
let manifest_path = std::path::Path::new(&args.manifest);
if let Some(manifest_dir) = manifest_path.parent() {
manifest_dir
.join(&manifest.package)
.to_string_lossy()
.to_string()
} else {
manifest.package.clone()
}
};
let wasm_bytes = resolve_reference(&wasm_path).await.map_err(|e| {
CliError::server_error(format!("Failed to load WASM from '{}': {}", wasm_path, e))
})?;
let (response_tx, response_rx) = tokio::sync::oneshot::channel();
let (supervisor_tx, mut supervisor_rx) = mpsc::channel(32);
theater_tx
.send(TheaterCommand::SpawnActor {
wasm_bytes,
name: Some(manifest.name.clone()),
manifest: Some(manifest),
init_bytes: None,
response_tx,
supervisor_tx: Some(supervisor_tx),
subscription_tx: None, })
.await
.map_err(|e| CliError::server_error(format!("Failed to send spawn command: {}", e)))?;
let actor_id = match response_rx.await {
Ok(Ok(id)) => {
debug!("Actor started: {}", id);
id
}
Ok(Err(e)) => {
return Err(CliError::server_error(format!(
"Failed to start actor: {}",
e
)));
}
Err(e) => {
return Err(CliError::server_error(format!(
"Failed to receive spawn response: {}",
e
)));
}
};
if !args.no_init {
let (handle_tx, handle_rx) = tokio::sync::oneshot::channel();
theater_tx
.send(TheaterCommand::GetActorHandle {
actor_id,
response_tx: handle_tx,
})
.await
.map_err(|e| CliError::server_error(format!("Failed to get actor handle: {}", e)))?;
let actor_handle = match handle_rx.await {
Ok(Some(handle)) => handle,
Ok(None) => {
return Err(CliError::server_error("Actor handle not found".to_string()));
}
Err(e) => {
return Err(CliError::server_error(format!(
"Failed to receive actor handle: {}",
e
)));
}
};
let init_state = Value::Option {
inner_type: ValueType::List(Box::new(ValueType::U8)),
value: None,
};
let init_params = Value::Tuple(vec![init_state]);
debug!("Calling init on actor {}", actor_id);
let _init_result = actor_handle
.call_function("theater:simple/actor.init".to_string(), init_params)
.await
.map_err(|e| CliError::server_error(format!("Failed to call init: {}", e)))?;
debug!("Init completed");
}
loop {
tokio::select! {
result = supervisor_rx.recv() => {
match result {
Some(actor_result) => {
debug!("Actor exited: {:?}", actor_result);
match actor_result {
theater::messages::ActorResult::Success(success) => {
if let Some(output) = success.result {
let _ = std::io::stdout().write_all(&output);
let _ = std::io::stdout().flush();
}
}
theater::messages::ActorResult::Error(err) => {
eprintln!("Actor error: {}", err.error);
std::process::exit(1);
}
theater::messages::ActorResult::ExternalStop(_) => {
debug!("Actor stopped externally");
}
}
break;
}
None => {
debug!("Supervisor channel closed");
break;
}
}
}
event = global_events_rx.recv() => {
if let Some((event_actor_id, event_result)) = event {
match event_result {
Ok(chain_event) => {
if let Some(ref mut manager) = chain_file_manager {
if let Err(e) = manager.write_event(&event_actor_id, &chain_event) {
eprintln!("Warning: failed to write chain event: {}", e);
}
}
if args.events {
match args.events_format {
EventFormat::Json => {
println!("{}", format_event_json(&chain_event, &event_actor_id));
}
EventFormat::Short => {
print!("{}", format_event_short(&chain_event, &event_actor_id));
}
EventFormat::Full => {
print!("{}", format_event_full(&chain_event, &event_actor_id));
}
}
}
if event_actor_id == actor_id && chain_event.event_type == "shutdown" {
break;
}
}
Err(e) => {
debug!("Actor error event: {:?}", e);
}
}
}
}
_ = tokio::signal::ctrl_c() => {
debug!("Received Ctrl+C, stopping actor {}", actor_id);
eprintln!("\nStopping actor...");
let (stop_tx, stop_rx) = tokio::sync::oneshot::channel();
let _ = theater_tx.send(TheaterCommand::StopActor {
actor_id,
response_tx: stop_tx,
}).await;
match tokio::time::timeout(
tokio::time::Duration::from_secs(5),
stop_rx,
).await {
Ok(Ok(Ok(()))) => debug!("Actor stopped gracefully"),
_ => debug!("Actor stop timed out or failed"),
}
break;
}
_ = ctx.shutdown_token.cancelled() => {
debug!("Shutdown token cancelled");
break;
}
}
}
drop(theater_tx);
let _ = tokio::time::timeout(tokio::time::Duration::from_secs(5), runtime_handle).await;
Ok(())
}