P1f: self-contained Tauri host; complete P1
This commit is contained in:
Emil
@@ -4,6 +4,10 @@ version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
|
||||
[lib]
|
||||
name = "zhila_daemon"
|
||||
path = "src/lib.rs"
|
||||
|
||||
[[bin]]
|
||||
name = "zhila-daemon"
|
||||
path = "src/main.rs"
|
||||
|
||||
@@ -0,0 +1,597 @@
|
||||
//! ZHILA local daemon: one WebSocket endpoint, one session, one engine.
|
||||
//!
|
||||
//! Text frames are JSON control messages; meter frames are binary. The daemon
|
||||
//! binds to loopback only and requires a token in the hello.
|
||||
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
|
||||
use anyhow::Context;
|
||||
use futures_util::stream::SplitSink;
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use tokio::net::{TcpListener, TcpStream};
|
||||
use tokio::sync::{Mutex, broadcast, mpsc};
|
||||
use tokio::time::{interval, timeout};
|
||||
use tokio_tungstenite::tungstenite::Message;
|
||||
use tokio_tungstenite::tungstenite::handshake::server::{ErrorResponse, Request, Response};
|
||||
use tokio_tungstenite::{WebSocketStream, accept_hdr_async};
|
||||
use tracing::{info, warn};
|
||||
use zhila_core::{CoreError, Engine, Session, default_pattern};
|
||||
use zhila_engine::{AudioEngine, NullEngine};
|
||||
use zhila_project::{ProjectSnapshot, ProjectStore};
|
||||
use zhila_protocol::{
|
||||
Actor, ClientMessage, Command, CommandResult, ErrorCode, PROTOCOL_VERSION, Seal, ServerMessage,
|
||||
};
|
||||
|
||||
const HELLO_TIMEOUT: Duration = Duration::from_secs(5);
|
||||
const METER_INTERVAL: Duration = Duration::from_millis(33);
|
||||
|
||||
type WsSink = SplitSink<WebSocketStream<TcpStream>, Message>;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct HostConfig {
|
||||
pub port: u16,
|
||||
pub token: String,
|
||||
pub no_audio: bool,
|
||||
pub project: Option<PathBuf>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
enum Outbound {
|
||||
Text(String),
|
||||
Binary(Vec<u8>),
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct Shared {
|
||||
session: Arc<Mutex<Session>>,
|
||||
store: Arc<std::sync::Mutex<ProjectStore>>,
|
||||
out: broadcast::Sender<Outbound>,
|
||||
token: Arc<String>,
|
||||
session_id: Arc<String>,
|
||||
}
|
||||
|
||||
pub async fn run(config: HostConfig) -> anyhow::Result<()> {
|
||||
let HostConfig {
|
||||
port,
|
||||
token,
|
||||
no_audio,
|
||||
project,
|
||||
} = config;
|
||||
|
||||
let pattern = default_pattern();
|
||||
let audio_holder = if no_audio {
|
||||
None
|
||||
} else {
|
||||
match AudioEngine::start(&pattern) {
|
||||
Ok(engine) => Some(engine),
|
||||
Err(error) => {
|
||||
warn!("audio offline: {error}");
|
||||
None
|
||||
}
|
||||
}
|
||||
};
|
||||
let control: Arc<dyn Engine> = match &audio_holder {
|
||||
Some(engine) => engine.control(),
|
||||
None => Arc::new(NullEngine::new()),
|
||||
};
|
||||
let shutdown = control.clone();
|
||||
let info = control.info();
|
||||
|
||||
let project_path = project.clone().unwrap_or_else(default_project_dir);
|
||||
let store = ProjectStore::open(&project_path).context("could not open the project")?;
|
||||
let loaded = store.load().context("could not load the project")?;
|
||||
let (session, project_name) = match loaded {
|
||||
Some(loaded) => {
|
||||
for warning in &loaded.warnings {
|
||||
warn!("project: {warning}");
|
||||
}
|
||||
let name = loaded.data.name.clone();
|
||||
info!(project = %name, path = %project_path.display(), "project loaded");
|
||||
(Session::from_project(control.clone(), loaded.data), name)
|
||||
}
|
||||
None => {
|
||||
let name = project_path
|
||||
.file_name()
|
||||
.and_then(|value| value.to_str())
|
||||
.unwrap_or("descent-04")
|
||||
.to_string();
|
||||
let mut session = Session::with_pattern(control.clone(), pattern.clone());
|
||||
session.set_project_name(name.clone());
|
||||
info!(project = %name, path = %project_path.display(), "new project");
|
||||
(session, name)
|
||||
}
|
||||
};
|
||||
let session = Arc::new(Mutex::new(session));
|
||||
let store = Arc::new(std::sync::Mutex::new(store));
|
||||
let (out, _) = broadcast::channel::<Outbound>(128);
|
||||
let session_id = format!("zhila-{:x}", now_ms());
|
||||
let shared = Shared {
|
||||
session,
|
||||
store,
|
||||
out: out.clone(),
|
||||
token: Arc::new(token.clone()),
|
||||
session_id: Arc::new(session_id.clone()),
|
||||
};
|
||||
save_now(&shared).await;
|
||||
|
||||
let meter_engine = control.clone();
|
||||
let meter_out = out.clone();
|
||||
tokio::spawn(async move {
|
||||
let mut ticker = interval(METER_INTERVAL);
|
||||
loop {
|
||||
ticker.tick().await;
|
||||
let frame = meter_engine.meter();
|
||||
let _ = meter_out.send(Outbound::Binary(frame.encode()));
|
||||
}
|
||||
});
|
||||
|
||||
let listener = TcpListener::bind(("127.0.0.1", port))
|
||||
.await
|
||||
.with_context(|| format!("could not bind 127.0.0.1:{port}"))?;
|
||||
|
||||
print_banner(
|
||||
port,
|
||||
&token,
|
||||
&info,
|
||||
&session_id,
|
||||
&project_name,
|
||||
&project_path,
|
||||
);
|
||||
if token == "dev" {
|
||||
warn!("using the development token; pass --token or ZHILA_TOKEN to change it");
|
||||
}
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
accepted = listener.accept() => {
|
||||
let (stream, peer) = accepted.context("accept failed")?;
|
||||
let shared = shared.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(error) = handle(stream, shared).await {
|
||||
warn!("connection {peer}: {error:#}");
|
||||
}
|
||||
});
|
||||
}
|
||||
_ = tokio::signal::ctrl_c() => {
|
||||
info!("shutting down");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
shutdown.stop();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn print_banner(
|
||||
port: u16,
|
||||
token: &str,
|
||||
info: &zhila_protocol::AudioInfo,
|
||||
session_id: &str,
|
||||
project_name: &str,
|
||||
project_path: &std::path::Path,
|
||||
) {
|
||||
println!();
|
||||
println!(" ZHILA engine");
|
||||
println!(" ws://127.0.0.1:{port} · token {token}");
|
||||
println!(" project {project_name} · {}", project_path.display());
|
||||
if info.running {
|
||||
let buffer = if info.buffer_frames == 0 {
|
||||
"default".to_string()
|
||||
} else {
|
||||
info.buffer_frames.to_string()
|
||||
};
|
||||
println!(
|
||||
" audio {} · {} Hz · {} ch · buffer {}",
|
||||
info.device, info.sample_rate, info.channels, buffer
|
||||
);
|
||||
} else {
|
||||
println!(" audio offline · the surface will connect, but nothing will sound");
|
||||
}
|
||||
println!(" session {session_id}");
|
||||
println!();
|
||||
}
|
||||
|
||||
async fn save_now(shared: &Shared) {
|
||||
let session = shared.session.lock().await;
|
||||
let state = session.snapshot();
|
||||
let gains = session.gains();
|
||||
let snapshot = ProjectSnapshot {
|
||||
name: session.project_name(),
|
||||
revision: state.revision,
|
||||
bpm: session.bpm(),
|
||||
gains: &gains,
|
||||
pattern: session.pattern(),
|
||||
cores: &state.cores,
|
||||
journal: session.journal(),
|
||||
history: session.history(),
|
||||
redo: session.redo(),
|
||||
};
|
||||
let result = {
|
||||
let mut store = shared.store.lock().expect("project store lock poisoned");
|
||||
store.save(&snapshot)
|
||||
};
|
||||
drop(session);
|
||||
if let Err(error) = result {
|
||||
warn!("project save failed: {error}");
|
||||
}
|
||||
}
|
||||
|
||||
fn default_project_dir() -> PathBuf {
|
||||
let base = std::env::var_os("XDG_DATA_HOME")
|
||||
.map(PathBuf::from)
|
||||
.or_else(|| std::env::var_os("HOME").map(|home| PathBuf::from(home).join(".local/share")))
|
||||
.unwrap_or_else(|| PathBuf::from("."));
|
||||
base.join("zhila").join("descent-04")
|
||||
}
|
||||
|
||||
async fn handle(stream: TcpStream, shared: Shared) -> anyhow::Result<()> {
|
||||
let peer = stream.peer_addr().ok();
|
||||
let ws = accept_hdr_async(stream, handshake)
|
||||
.await
|
||||
.context("websocket handshake failed")?;
|
||||
|
||||
let (mut sink, mut source) = ws.split();
|
||||
|
||||
let first = match timeout(HELLO_TIMEOUT, source.next()).await {
|
||||
Ok(Some(Ok(message))) => message,
|
||||
Ok(Some(Err(error))) => return Err(error.into()),
|
||||
Ok(None) => return Ok(()),
|
||||
Err(_) => {
|
||||
let _ = send_out(
|
||||
&mut sink,
|
||||
error_message(
|
||||
None,
|
||||
ErrorCode::Protocol,
|
||||
"expected a hello within 5 seconds",
|
||||
),
|
||||
)
|
||||
.await;
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
let Message::Text(text) = first else {
|
||||
let _ = send_out(
|
||||
&mut sink,
|
||||
error_message(None, ErrorCode::Protocol, "the first frame must be a hello"),
|
||||
)
|
||||
.await;
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
let hello: ClientMessage = match serde_json::from_str(text.as_str()) {
|
||||
Ok(message) => message,
|
||||
Err(_) => {
|
||||
let _ = send_out(
|
||||
&mut sink,
|
||||
error_message(None, ErrorCode::BadRequest, "malformed hello"),
|
||||
)
|
||||
.await;
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
let (client, protocol, token) = match hello {
|
||||
ClientMessage::Hello {
|
||||
client,
|
||||
protocol,
|
||||
token,
|
||||
} => (client, protocol, token),
|
||||
ClientMessage::Command { .. } | ClientMessage::ImportCore { .. } => {
|
||||
let _ = send_out(
|
||||
&mut sink,
|
||||
error_message(None, ErrorCode::Protocol, "expected hello before commands"),
|
||||
)
|
||||
.await;
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
if protocol != PROTOCOL_VERSION {
|
||||
let _ = send_out(
|
||||
&mut sink,
|
||||
error_message(
|
||||
None,
|
||||
ErrorCode::Protocol,
|
||||
format!(
|
||||
"protocol mismatch: this daemon speaks {PROTOCOL_VERSION}, the client sent {protocol}"
|
||||
),
|
||||
),
|
||||
)
|
||||
.await;
|
||||
return Ok(());
|
||||
}
|
||||
if token.as_str() != shared.token.as_str() {
|
||||
let _ = send_out(
|
||||
&mut sink,
|
||||
error_message(None, ErrorCode::Unauthorized, "the token was not accepted"),
|
||||
)
|
||||
.await;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let state = shared.session.lock().await.snapshot();
|
||||
info!(?peer, client = %client, "wire opened");
|
||||
send_out(
|
||||
&mut sink,
|
||||
Outbound::Text(
|
||||
serde_json::to_string(&ServerMessage::Hello {
|
||||
protocol: PROTOCOL_VERSION,
|
||||
session: shared.session_id.as_ref().clone(),
|
||||
engine: state.audio.clone(),
|
||||
})
|
||||
.expect("hello serializes"),
|
||||
),
|
||||
)
|
||||
.await?;
|
||||
let revision = state.revision;
|
||||
send_out(
|
||||
&mut sink,
|
||||
Outbound::Text(
|
||||
serde_json::to_string(&ServerMessage::Snapshot { revision, state })
|
||||
.expect("snapshot serializes"),
|
||||
),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let (direct_tx, mut direct_rx) = mpsc::unbounded_channel::<Outbound>();
|
||||
let mut events = shared.out.subscribe();
|
||||
let writer = tokio::spawn(async move {
|
||||
loop {
|
||||
tokio::select! {
|
||||
direct = direct_rx.recv() => {
|
||||
match direct {
|
||||
Some(outbound) => {
|
||||
if send_out(&mut sink, outbound).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
broadcast = events.recv() => {
|
||||
match broadcast {
|
||||
Ok(outbound) => {
|
||||
if send_out(&mut sink, outbound).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(broadcast::error::RecvError::Lagged(_)) => continue,
|
||||
Err(broadcast::error::RecvError::Closed) => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let mut pending_import: Option<(u64, String, Seal)> = None;
|
||||
while let Some(frame) = source.next().await {
|
||||
let message = match frame {
|
||||
Ok(Message::Text(text)) => text,
|
||||
Ok(Message::Binary(bytes)) => {
|
||||
let Some((id, name, seal)) = pending_import.take() else {
|
||||
let _ = direct_tx.send(error_message(
|
||||
None,
|
||||
ErrorCode::BadRequest,
|
||||
"a binary frame arrived without an import header",
|
||||
));
|
||||
continue;
|
||||
};
|
||||
let source_name = name.clone();
|
||||
let audio = bytes.to_vec();
|
||||
let asset_audio = audio.clone();
|
||||
let import = tokio::task::spawn_blocking(move || {
|
||||
zhila_material::import_bytes(&name, seal, audio)
|
||||
})
|
||||
.await;
|
||||
match import {
|
||||
Ok(Ok(core)) => {
|
||||
let summary = core.summary.clone();
|
||||
let asset = {
|
||||
let mut store =
|
||||
shared.store.lock().expect("project store lock poisoned");
|
||||
store.write_asset(&summary, &source_name, &asset_audio)
|
||||
};
|
||||
if let Err(error) = asset {
|
||||
warn!("project asset failed: {error}");
|
||||
let _ = direct_tx.send(error_message(
|
||||
Some(id),
|
||||
ErrorCode::Internal,
|
||||
"the asset could not be saved",
|
||||
));
|
||||
continue;
|
||||
}
|
||||
let state = {
|
||||
let mut session = shared.session.lock().await;
|
||||
session.import_core(core, Actor::Hand)
|
||||
};
|
||||
save_now(&shared).await;
|
||||
let _ = direct_tx.send(Outbound::Text(
|
||||
serde_json::to_string(&ServerMessage::Imported { id, core: summary })
|
||||
.expect("import serializes"),
|
||||
));
|
||||
let snapshot = ServerMessage::Snapshot {
|
||||
revision: state.revision,
|
||||
state,
|
||||
};
|
||||
let _ = shared
|
||||
.out
|
||||
.send(Outbound::Text(serde_json::to_string(&snapshot).unwrap()));
|
||||
}
|
||||
Ok(Err(error)) => {
|
||||
let _ = direct_tx.send(error_message(
|
||||
Some(id),
|
||||
ErrorCode::BadRequest,
|
||||
error.to_string(),
|
||||
));
|
||||
}
|
||||
Err(error) => {
|
||||
warn!("import task failed: {error}");
|
||||
let _ = direct_tx.send(error_message(
|
||||
Some(id),
|
||||
ErrorCode::Internal,
|
||||
"the import task failed",
|
||||
));
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
Ok(Message::Close(_)) => break,
|
||||
Ok(_) => continue,
|
||||
Err(error) => {
|
||||
warn!("wire read error: {error}");
|
||||
break;
|
||||
}
|
||||
};
|
||||
|
||||
let message: ClientMessage = match serde_json::from_str(message.as_str()) {
|
||||
Ok(message) => message,
|
||||
Err(_) => {
|
||||
let _ = direct_tx.send(error_message(
|
||||
None,
|
||||
ErrorCode::BadRequest,
|
||||
"malformed message",
|
||||
));
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
match message {
|
||||
ClientMessage::Hello { .. } => {
|
||||
let _ = direct_tx.send(error_message(None, ErrorCode::Protocol, "already greeted"));
|
||||
}
|
||||
ClientMessage::ImportCore { id, name, seal } => {
|
||||
if pending_import.is_some() {
|
||||
let _ = direct_tx.send(error_message(
|
||||
Some(id),
|
||||
ErrorCode::BadRequest,
|
||||
"the previous import has not received its bytes yet",
|
||||
));
|
||||
} else {
|
||||
pending_import = Some((id, name, seal));
|
||||
}
|
||||
}
|
||||
ClientMessage::Command { id, command } => {
|
||||
let should_save = !matches!(
|
||||
&command,
|
||||
Command::Play
|
||||
| Command::Stop
|
||||
| Command::Strike { .. }
|
||||
| Command::StrikeFace { .. }
|
||||
| Command::GetSnapshot
|
||||
);
|
||||
let outcome = {
|
||||
let mut session = shared.session.lock().await;
|
||||
session.apply(command, Actor::Hand)
|
||||
};
|
||||
match outcome {
|
||||
Ok((result, emitted)) => {
|
||||
if should_save {
|
||||
save_now(&shared).await;
|
||||
}
|
||||
let revision = match &result {
|
||||
CommandResult::Snapshot { state } | CommandResult::State { state } => {
|
||||
state.revision
|
||||
}
|
||||
};
|
||||
let _ = direct_tx.send(Outbound::Text(
|
||||
serde_json::to_string(&ServerMessage::Result { id, result })
|
||||
.expect("result serializes"),
|
||||
));
|
||||
for event in emitted {
|
||||
let message = ServerMessage::Event { revision, event };
|
||||
let _ = shared
|
||||
.out
|
||||
.send(Outbound::Text(serde_json::to_string(&message).unwrap()));
|
||||
}
|
||||
}
|
||||
Err(error) => {
|
||||
let _ = direct_tx.send(error_message(
|
||||
Some(id),
|
||||
error_code(&error),
|
||||
error.to_string(),
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
drop(direct_tx);
|
||||
let _ = writer.await;
|
||||
info!(?peer, "wire closed");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[allow(clippy::result_large_err)]
|
||||
fn handshake(request: &Request, response: Response) -> Result<Response, ErrorResponse> {
|
||||
if let Some(origin) = request
|
||||
.headers()
|
||||
.get("origin")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
&& !origin_allowed(origin)
|
||||
{
|
||||
return Err(ErrorResponse::new(Some(
|
||||
"origin is not allowed to talk to ZHILA".into(),
|
||||
)));
|
||||
}
|
||||
Ok(response)
|
||||
}
|
||||
|
||||
fn origin_allowed(origin: &str) -> bool {
|
||||
if origin == "null" {
|
||||
return true;
|
||||
}
|
||||
let Some((scheme, rest)) = origin.split_once("://") else {
|
||||
return false;
|
||||
};
|
||||
if !matches!(scheme, "http" | "https" | "tauri") {
|
||||
return false;
|
||||
}
|
||||
let authority = rest.split('/').next().unwrap_or_default();
|
||||
let host = authority.split(':').next().unwrap_or_default();
|
||||
matches!(host, "localhost" | "127.0.0.1" | "tauri.localhost")
|
||||
}
|
||||
|
||||
fn error_code(error: &CoreError) -> ErrorCode {
|
||||
match error {
|
||||
CoreError::InvalidTempo
|
||||
| CoreError::InvalidVelocity
|
||||
| CoreError::InvalidGain
|
||||
| CoreError::InvalidRow
|
||||
| CoreError::InvalidStep
|
||||
| CoreError::UnknownFace
|
||||
| CoreError::InvalidFaces
|
||||
| CoreError::UnsupportedCommand => ErrorCode::BadRequest,
|
||||
}
|
||||
}
|
||||
|
||||
fn error_message(id: Option<u64>, code: ErrorCode, message: impl Into<String>) -> Outbound {
|
||||
Outbound::Text(
|
||||
serde_json::to_string(&ServerMessage::Error {
|
||||
id,
|
||||
code,
|
||||
message: message.into(),
|
||||
})
|
||||
.expect("error serializes"),
|
||||
)
|
||||
}
|
||||
|
||||
async fn send_out(sink: &mut WsSink, outbound: Outbound) -> anyhow::Result<()> {
|
||||
let message = match outbound {
|
||||
Outbound::Text(text) => Message::Text(text.into()),
|
||||
Outbound::Binary(bytes) => Message::Binary(bytes.into()),
|
||||
};
|
||||
sink.send(message).await.context("websocket send failed")?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn now_ms() -> u64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|duration| duration.as_millis() as u64)
|
||||
.unwrap_or_default()
|
||||
}
|
||||
@@ -1,41 +1,11 @@
|
||||
//! ZHILA local daemon: one WebSocket endpoint, one session, one engine.
|
||||
//!
|
||||
//! Text frames are JSON control messages; meter frames are binary. The daemon
|
||||
//! binds to loopback only and requires a token in the hello.
|
||||
//! ZHILA daemon binary: parse flags and run the local engine host.
|
||||
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
|
||||
use anyhow::Context;
|
||||
use clap::Parser;
|
||||
use futures_util::stream::SplitSink;
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use tokio::net::{TcpListener, TcpStream};
|
||||
use tokio::sync::{Mutex, broadcast, mpsc};
|
||||
use tokio::time::{interval, timeout};
|
||||
use tokio_tungstenite::tungstenite::Message;
|
||||
use tokio_tungstenite::tungstenite::handshake::server::{ErrorResponse, Request, Response};
|
||||
use tokio_tungstenite::{WebSocketStream, accept_hdr_async};
|
||||
use tracing::{info, warn};
|
||||
use zhila_core::{CoreError, Engine, Session, default_pattern};
|
||||
use zhila_engine::{AudioEngine, NullEngine};
|
||||
use zhila_project::{ProjectSnapshot, ProjectStore};
|
||||
use zhila_protocol::{
|
||||
Actor, ClientMessage, Command, CommandResult, ErrorCode, PROTOCOL_VERSION, Seal, ServerMessage,
|
||||
};
|
||||
|
||||
const HELLO_TIMEOUT: Duration = Duration::from_secs(5);
|
||||
const METER_INTERVAL: Duration = Duration::from_millis(33);
|
||||
|
||||
type WsSink = SplitSink<WebSocketStream<TcpStream>, Message>;
|
||||
|
||||
#[derive(Debug, Parser)]
|
||||
#[command(
|
||||
name = "zhila-daemon",
|
||||
version,
|
||||
about = "ZHILA local engine daemon (P0)"
|
||||
)]
|
||||
#[command(name = "zhila-daemon", version, about = "ZHILA local engine host")]
|
||||
struct Args {
|
||||
/// WebSocket port on loopback.
|
||||
#[arg(long, default_value_t = 4317)]
|
||||
@@ -51,21 +21,6 @@ struct Args {
|
||||
project: Option<PathBuf>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
enum Outbound {
|
||||
Text(String),
|
||||
Binary(Vec<u8>),
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct Shared {
|
||||
session: Arc<Mutex<Session>>,
|
||||
store: Arc<std::sync::Mutex<ProjectStore>>,
|
||||
out: broadcast::Sender<Outbound>,
|
||||
token: Arc<String>,
|
||||
session_id: Arc<String>,
|
||||
}
|
||||
|
||||
#[tokio::main(flavor = "current_thread")]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
let filter = tracing_subscriber::EnvFilter::try_from_default_env()
|
||||
@@ -77,539 +32,11 @@ async fn main() -> anyhow::Result<()> {
|
||||
|
||||
let args = Args::parse();
|
||||
let token = std::env::var("ZHILA_TOKEN").unwrap_or(args.token);
|
||||
|
||||
let pattern = default_pattern();
|
||||
let audio_holder = if args.no_audio {
|
||||
None
|
||||
} else {
|
||||
match AudioEngine::start(&pattern) {
|
||||
Ok(engine) => Some(engine),
|
||||
Err(error) => {
|
||||
warn!("audio offline: {error}");
|
||||
None
|
||||
}
|
||||
}
|
||||
};
|
||||
let control: Arc<dyn Engine> = match &audio_holder {
|
||||
Some(engine) => engine.control(),
|
||||
None => Arc::new(NullEngine::new()),
|
||||
};
|
||||
let shutdown = control.clone();
|
||||
let info = control.info();
|
||||
|
||||
let project_path = args.project.clone().unwrap_or_else(default_project_dir);
|
||||
let store = ProjectStore::open(&project_path).context("could not open the project")?;
|
||||
let loaded = store.load().context("could not load the project")?;
|
||||
let (session, project_name) = match loaded {
|
||||
Some(loaded) => {
|
||||
for warning in &loaded.warnings {
|
||||
warn!("project: {warning}");
|
||||
}
|
||||
let name = loaded.data.name.clone();
|
||||
info!(project = %name, path = %project_path.display(), "project loaded");
|
||||
(Session::from_project(control.clone(), loaded.data), name)
|
||||
}
|
||||
None => {
|
||||
let name = project_path
|
||||
.file_name()
|
||||
.and_then(|value| value.to_str())
|
||||
.unwrap_or("descent-04")
|
||||
.to_string();
|
||||
let mut session = Session::with_pattern(control.clone(), pattern.clone());
|
||||
session.set_project_name(name.clone());
|
||||
info!(project = %name, path = %project_path.display(), "new project");
|
||||
(session, name)
|
||||
}
|
||||
};
|
||||
let session = Arc::new(Mutex::new(session));
|
||||
let store = Arc::new(std::sync::Mutex::new(store));
|
||||
let (out, _) = broadcast::channel::<Outbound>(128);
|
||||
let session_id = format!("zhila-{:x}", now_ms());
|
||||
let shared = Shared {
|
||||
session,
|
||||
store,
|
||||
out: out.clone(),
|
||||
token: Arc::new(token.clone()),
|
||||
session_id: Arc::new(session_id.clone()),
|
||||
};
|
||||
save_now(&shared).await;
|
||||
|
||||
let meter_engine = control.clone();
|
||||
let meter_out = out.clone();
|
||||
tokio::spawn(async move {
|
||||
let mut ticker = interval(METER_INTERVAL);
|
||||
loop {
|
||||
ticker.tick().await;
|
||||
let frame = meter_engine.meter();
|
||||
let _ = meter_out.send(Outbound::Binary(frame.encode()));
|
||||
}
|
||||
});
|
||||
|
||||
let listener = TcpListener::bind(("127.0.0.1", args.port))
|
||||
.await
|
||||
.with_context(|| format!("could not bind 127.0.0.1:{}", args.port))?;
|
||||
|
||||
print_banner(
|
||||
args.port,
|
||||
&token,
|
||||
&info,
|
||||
&session_id,
|
||||
&project_name,
|
||||
&project_path,
|
||||
);
|
||||
if token == "dev" {
|
||||
warn!("using the development token; pass --token or ZHILA_TOKEN to change it");
|
||||
}
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
accepted = listener.accept() => {
|
||||
let (stream, peer) = accepted.context("accept failed")?;
|
||||
let shared = shared.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(error) = handle(stream, shared).await {
|
||||
warn!("connection {peer}: {error:#}");
|
||||
}
|
||||
});
|
||||
}
|
||||
_ = tokio::signal::ctrl_c() => {
|
||||
info!("shutting down");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
shutdown.stop();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn print_banner(
|
||||
port: u16,
|
||||
token: &str,
|
||||
info: &zhila_protocol::AudioInfo,
|
||||
session_id: &str,
|
||||
project_name: &str,
|
||||
project_path: &std::path::Path,
|
||||
) {
|
||||
println!();
|
||||
println!(" ZHILA engine");
|
||||
println!(" ws://127.0.0.1:{port} · token {token}");
|
||||
println!(" project {project_name} · {}", project_path.display());
|
||||
if info.running {
|
||||
let buffer = if info.buffer_frames == 0 {
|
||||
"default".to_string()
|
||||
} else {
|
||||
info.buffer_frames.to_string()
|
||||
};
|
||||
println!(
|
||||
" audio {} · {} Hz · {} ch · buffer {}",
|
||||
info.device, info.sample_rate, info.channels, buffer
|
||||
);
|
||||
} else {
|
||||
println!(" audio offline · the surface will connect, but nothing will sound");
|
||||
}
|
||||
println!(" session {session_id}");
|
||||
println!();
|
||||
}
|
||||
|
||||
async fn save_now(shared: &Shared) {
|
||||
let session = shared.session.lock().await;
|
||||
let state = session.snapshot();
|
||||
let gains = session.gains();
|
||||
let snapshot = ProjectSnapshot {
|
||||
name: session.project_name(),
|
||||
revision: state.revision,
|
||||
bpm: session.bpm(),
|
||||
gains: &gains,
|
||||
pattern: session.pattern(),
|
||||
cores: &state.cores,
|
||||
journal: session.journal(),
|
||||
history: session.history(),
|
||||
redo: session.redo(),
|
||||
};
|
||||
let result = {
|
||||
let mut store = shared.store.lock().expect("project store lock poisoned");
|
||||
store.save(&snapshot)
|
||||
};
|
||||
drop(session);
|
||||
if let Err(error) = result {
|
||||
warn!("project save failed: {error}");
|
||||
}
|
||||
}
|
||||
|
||||
fn default_project_dir() -> PathBuf {
|
||||
let base = std::env::var_os("XDG_DATA_HOME")
|
||||
.map(PathBuf::from)
|
||||
.or_else(|| std::env::var_os("HOME").map(|home| PathBuf::from(home).join(".local/share")))
|
||||
.unwrap_or_else(|| PathBuf::from("."));
|
||||
base.join("zhila").join("descent-04")
|
||||
}
|
||||
|
||||
async fn handle(stream: TcpStream, shared: Shared) -> anyhow::Result<()> {
|
||||
let peer = stream.peer_addr().ok();
|
||||
let ws = accept_hdr_async(stream, handshake)
|
||||
.await
|
||||
.context("websocket handshake failed")?;
|
||||
|
||||
let (mut sink, mut source) = ws.split();
|
||||
|
||||
let first = match timeout(HELLO_TIMEOUT, source.next()).await {
|
||||
Ok(Some(Ok(message))) => message,
|
||||
Ok(Some(Err(error))) => return Err(error.into()),
|
||||
Ok(None) => return Ok(()),
|
||||
Err(_) => {
|
||||
let _ = send_out(
|
||||
&mut sink,
|
||||
error_message(
|
||||
None,
|
||||
ErrorCode::Protocol,
|
||||
"expected a hello within 5 seconds",
|
||||
),
|
||||
)
|
||||
.await;
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
let Message::Text(text) = first else {
|
||||
let _ = send_out(
|
||||
&mut sink,
|
||||
error_message(None, ErrorCode::Protocol, "the first frame must be a hello"),
|
||||
)
|
||||
.await;
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
let hello: ClientMessage = match serde_json::from_str(text.as_str()) {
|
||||
Ok(message) => message,
|
||||
Err(_) => {
|
||||
let _ = send_out(
|
||||
&mut sink,
|
||||
error_message(None, ErrorCode::BadRequest, "malformed hello"),
|
||||
)
|
||||
.await;
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
let (client, protocol, token) = match hello {
|
||||
ClientMessage::Hello {
|
||||
client,
|
||||
protocol,
|
||||
token,
|
||||
} => (client, protocol, token),
|
||||
ClientMessage::Command { .. } | ClientMessage::ImportCore { .. } => {
|
||||
let _ = send_out(
|
||||
&mut sink,
|
||||
error_message(None, ErrorCode::Protocol, "expected hello before commands"),
|
||||
)
|
||||
.await;
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
if protocol != PROTOCOL_VERSION {
|
||||
let _ = send_out(
|
||||
&mut sink,
|
||||
error_message(
|
||||
None,
|
||||
ErrorCode::Protocol,
|
||||
format!(
|
||||
"protocol mismatch: this daemon speaks {PROTOCOL_VERSION}, the client sent {protocol}"
|
||||
),
|
||||
),
|
||||
)
|
||||
.await;
|
||||
return Ok(());
|
||||
}
|
||||
if token.as_str() != shared.token.as_str() {
|
||||
let _ = send_out(
|
||||
&mut sink,
|
||||
error_message(None, ErrorCode::Unauthorized, "the token was not accepted"),
|
||||
)
|
||||
.await;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let state = shared.session.lock().await.snapshot();
|
||||
info!(?peer, client = %client, "wire opened");
|
||||
send_out(
|
||||
&mut sink,
|
||||
Outbound::Text(
|
||||
serde_json::to_string(&ServerMessage::Hello {
|
||||
protocol: PROTOCOL_VERSION,
|
||||
session: shared.session_id.as_ref().clone(),
|
||||
engine: state.audio.clone(),
|
||||
})
|
||||
.expect("hello serializes"),
|
||||
),
|
||||
)
|
||||
.await?;
|
||||
let revision = state.revision;
|
||||
send_out(
|
||||
&mut sink,
|
||||
Outbound::Text(
|
||||
serde_json::to_string(&ServerMessage::Snapshot { revision, state })
|
||||
.expect("snapshot serializes"),
|
||||
),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let (direct_tx, mut direct_rx) = mpsc::unbounded_channel::<Outbound>();
|
||||
let mut events = shared.out.subscribe();
|
||||
let writer = tokio::spawn(async move {
|
||||
loop {
|
||||
tokio::select! {
|
||||
direct = direct_rx.recv() => {
|
||||
match direct {
|
||||
Some(outbound) => {
|
||||
if send_out(&mut sink, outbound).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
broadcast = events.recv() => {
|
||||
match broadcast {
|
||||
Ok(outbound) => {
|
||||
if send_out(&mut sink, outbound).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(broadcast::error::RecvError::Lagged(_)) => continue,
|
||||
Err(broadcast::error::RecvError::Closed) => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let mut pending_import: Option<(u64, String, Seal)> = None;
|
||||
while let Some(frame) = source.next().await {
|
||||
let message = match frame {
|
||||
Ok(Message::Text(text)) => text,
|
||||
Ok(Message::Binary(bytes)) => {
|
||||
let Some((id, name, seal)) = pending_import.take() else {
|
||||
let _ = direct_tx.send(error_message(
|
||||
None,
|
||||
ErrorCode::BadRequest,
|
||||
"a binary frame arrived without an import header",
|
||||
));
|
||||
continue;
|
||||
};
|
||||
let source_name = name.clone();
|
||||
let audio = bytes.to_vec();
|
||||
let asset_audio = audio.clone();
|
||||
let import = tokio::task::spawn_blocking(move || {
|
||||
zhila_material::import_bytes(&name, seal, audio)
|
||||
})
|
||||
.await;
|
||||
match import {
|
||||
Ok(Ok(core)) => {
|
||||
let summary = core.summary.clone();
|
||||
let asset = {
|
||||
let mut store =
|
||||
shared.store.lock().expect("project store lock poisoned");
|
||||
store.write_asset(&summary, &source_name, &asset_audio)
|
||||
};
|
||||
if let Err(error) = asset {
|
||||
warn!("project asset failed: {error}");
|
||||
let _ = direct_tx.send(error_message(
|
||||
Some(id),
|
||||
ErrorCode::Internal,
|
||||
"the asset could not be saved",
|
||||
));
|
||||
continue;
|
||||
}
|
||||
let state = {
|
||||
let mut session = shared.session.lock().await;
|
||||
session.import_core(core, Actor::Hand)
|
||||
};
|
||||
save_now(&shared).await;
|
||||
let _ = direct_tx.send(Outbound::Text(
|
||||
serde_json::to_string(&ServerMessage::Imported { id, core: summary })
|
||||
.expect("import serializes"),
|
||||
));
|
||||
let snapshot = ServerMessage::Snapshot {
|
||||
revision: state.revision,
|
||||
state,
|
||||
};
|
||||
let _ = shared
|
||||
.out
|
||||
.send(Outbound::Text(serde_json::to_string(&snapshot).unwrap()));
|
||||
}
|
||||
Ok(Err(error)) => {
|
||||
let _ = direct_tx.send(error_message(
|
||||
Some(id),
|
||||
ErrorCode::BadRequest,
|
||||
error.to_string(),
|
||||
));
|
||||
}
|
||||
Err(error) => {
|
||||
warn!("import task failed: {error}");
|
||||
let _ = direct_tx.send(error_message(
|
||||
Some(id),
|
||||
ErrorCode::Internal,
|
||||
"the import task failed",
|
||||
));
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
Ok(Message::Close(_)) => break,
|
||||
Ok(_) => continue,
|
||||
Err(error) => {
|
||||
warn!("wire read error: {error}");
|
||||
break;
|
||||
}
|
||||
};
|
||||
|
||||
let message: ClientMessage = match serde_json::from_str(message.as_str()) {
|
||||
Ok(message) => message,
|
||||
Err(_) => {
|
||||
let _ = direct_tx.send(error_message(
|
||||
None,
|
||||
ErrorCode::BadRequest,
|
||||
"malformed message",
|
||||
));
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
match message {
|
||||
ClientMessage::Hello { .. } => {
|
||||
let _ = direct_tx.send(error_message(None, ErrorCode::Protocol, "already greeted"));
|
||||
}
|
||||
ClientMessage::ImportCore { id, name, seal } => {
|
||||
if pending_import.is_some() {
|
||||
let _ = direct_tx.send(error_message(
|
||||
Some(id),
|
||||
ErrorCode::BadRequest,
|
||||
"the previous import has not received its bytes yet",
|
||||
));
|
||||
} else {
|
||||
pending_import = Some((id, name, seal));
|
||||
}
|
||||
}
|
||||
ClientMessage::Command { id, command } => {
|
||||
let should_save = !matches!(
|
||||
&command,
|
||||
Command::Play
|
||||
| Command::Stop
|
||||
| Command::Strike { .. }
|
||||
| Command::StrikeFace { .. }
|
||||
| Command::GetSnapshot
|
||||
);
|
||||
let outcome = {
|
||||
let mut session = shared.session.lock().await;
|
||||
session.apply(command, Actor::Hand)
|
||||
};
|
||||
match outcome {
|
||||
Ok((result, emitted)) => {
|
||||
if should_save {
|
||||
save_now(&shared).await;
|
||||
}
|
||||
let revision = match &result {
|
||||
CommandResult::Snapshot { state } | CommandResult::State { state } => {
|
||||
state.revision
|
||||
}
|
||||
};
|
||||
let _ = direct_tx.send(Outbound::Text(
|
||||
serde_json::to_string(&ServerMessage::Result { id, result })
|
||||
.expect("result serializes"),
|
||||
));
|
||||
for event in emitted {
|
||||
let message = ServerMessage::Event { revision, event };
|
||||
let _ = shared
|
||||
.out
|
||||
.send(Outbound::Text(serde_json::to_string(&message).unwrap()));
|
||||
}
|
||||
}
|
||||
Err(error) => {
|
||||
let _ = direct_tx.send(error_message(
|
||||
Some(id),
|
||||
error_code(&error),
|
||||
error.to_string(),
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
drop(direct_tx);
|
||||
let _ = writer.await;
|
||||
info!(?peer, "wire closed");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[allow(clippy::result_large_err)]
|
||||
fn handshake(request: &Request, response: Response) -> Result<Response, ErrorResponse> {
|
||||
if let Some(origin) = request
|
||||
.headers()
|
||||
.get("origin")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
&& !origin_allowed(origin)
|
||||
{
|
||||
return Err(ErrorResponse::new(Some(
|
||||
"origin is not allowed to talk to ZHILA".into(),
|
||||
)));
|
||||
}
|
||||
Ok(response)
|
||||
}
|
||||
|
||||
fn origin_allowed(origin: &str) -> bool {
|
||||
if origin == "null" {
|
||||
return true;
|
||||
}
|
||||
let Some((scheme, rest)) = origin.split_once("://") else {
|
||||
return false;
|
||||
};
|
||||
if !matches!(scheme, "http" | "https" | "tauri") {
|
||||
return false;
|
||||
}
|
||||
let authority = rest.split('/').next().unwrap_or_default();
|
||||
let host = authority.split(':').next().unwrap_or_default();
|
||||
matches!(host, "localhost" | "127.0.0.1" | "tauri.localhost")
|
||||
}
|
||||
|
||||
fn error_code(error: &CoreError) -> ErrorCode {
|
||||
match error {
|
||||
CoreError::InvalidTempo
|
||||
| CoreError::InvalidVelocity
|
||||
| CoreError::InvalidGain
|
||||
| CoreError::InvalidRow
|
||||
| CoreError::InvalidStep
|
||||
| CoreError::UnknownFace
|
||||
| CoreError::InvalidFaces
|
||||
| CoreError::UnsupportedCommand => ErrorCode::BadRequest,
|
||||
}
|
||||
}
|
||||
|
||||
fn error_message(id: Option<u64>, code: ErrorCode, message: impl Into<String>) -> Outbound {
|
||||
Outbound::Text(
|
||||
serde_json::to_string(&ServerMessage::Error {
|
||||
id,
|
||||
code,
|
||||
message: message.into(),
|
||||
})
|
||||
.expect("error serializes"),
|
||||
)
|
||||
}
|
||||
|
||||
async fn send_out(sink: &mut WsSink, outbound: Outbound) -> anyhow::Result<()> {
|
||||
let message = match outbound {
|
||||
Outbound::Text(text) => Message::Text(text.into()),
|
||||
Outbound::Binary(bytes) => Message::Binary(bytes.into()),
|
||||
};
|
||||
sink.send(message).await.context("websocket send failed")?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn now_ms() -> u64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|duration| duration.as_millis() as u64)
|
||||
.unwrap_or_default()
|
||||
zhila_daemon::run(zhila_daemon::HostConfig {
|
||||
port: args.port,
|
||||
token,
|
||||
no_audio: args.no_audio,
|
||||
project: args.project,
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user