synaps 0.10.2

//! `synaps send` — deliver an event via Unix socket, falling back to inbox file-drop.
//!
//! ## Driver-aware remote steering (P9, C2)
//! `send` is a dumb pipe: it delivers an [`Event`] to the target session's
//! per-session event socket, which lands in the runtime `event_queue`. It does
//! NOT speak `SessionCommand` (no `Submit`/`Steer`), and it never inspects
//! driver state. ALL Submit-vs-Steer routing — including remote steering of a
//! *driver-armed* session — is owned by the actor's event-queue wake path
//! (`session/actor.rs::on_queue_wake` → `engine/reactor.rs::drain_event_queue`
//! + `wake_action`):
//!
//! * session **busy** (a driver-owned turn is streaming): the event is
//!   **Steered** into the live stream via the steer channel — this is exactly
//!   the "route to Steer while driver-armed" behavior, achieved server-side.
//! * session **idle + driver armed**: the event is injected as context but the
//!   actor **inhibits the competing turn** (`on_queue_wake`: "RunTurn inhibited
//!   — driver armed") so the driver's own tick owns scheduling — no rogue
//!   Submit-turn races the driver.
//! * session **idle + unarmed**: injected + a normal auto-turn (the historical
//!   `Submit` behavior).
//!
//! Because armed-state routing is the actor's job, `send` needs no armed-state
//! discovery and stays transport-simple. See `tests/send_remote_steering.rs`.

use chrono::Utc;
use synaps_cli::events::registry::{
    find_session_registration, list_active_sessions, SessionRegistration,
};
use synaps_cli::events::{Event, EventChannel, EventContent, EventSource, Severity};
use tokio::io::AsyncWriteExt;
#[cfg(unix)]
use tokio::net::UnixStream;
use uuid::Uuid;

pub async fn run(
    message: String,
    source: String,
    severity: String,
    channel: Option<String>,
    content_type: String,
    session: Option<String>,
    broadcast: bool,
) -> anyhow::Result<()> {
    let event = build_event(message, source, severity, channel, content_type);
    let json = serde_json::to_string(&event)?;

    if broadcast {
        let sessions = list_active_sessions();
        if sessions.is_empty() {
            eprintln!("No active sessions found for broadcast.");
        }
        for reg in sessions {
            send_to_socket_or_inbox(&reg, &json, &event).await;
        }
        return Ok(());
    }

    if let Some(query) = session {
        match find_session_registration(&query) {
            Some(reg) => send_to_socket_or_inbox(&reg, &json, &event).await,
            None => {
                eprintln!("No session matching {:?} found — dropping to inbox.", query);
                write_inbox(&event)?;
            }
        }
        return Ok(());
    }

    // No flags: try to auto-resolve to exactly one active session.
    let sessions = list_active_sessions();
    match sessions.len() {
        1 => send_to_socket_or_inbox(&sessions[0], &json, &event).await,
        _ => {
            if sessions.is_empty() {
                eprintln!("No active sessions — writing to inbox.");
            } else {
                eprintln!(
                    "{} active sessions, ambiguous — writing to inbox.",
                    sessions.len()
                );
            }
            write_inbox(&event)?;
        }
    }

    Ok(())
}

// ── helpers ──────────────────────────────────────────────────────────────────

fn build_event(
    message: String,
    source: String,
    severity: String,
    channel: Option<String>,
    content_type: String,
) -> Event {
    Event {
        id: Uuid::new_v4().to_string(),
        timestamp: Utc::now(),
        source: EventSource {
            source_type: source.clone(),
            name: source,
            callback: None,
        },
        channel: channel.map(|c| EventChannel {
            id: c.clone(),
            name: c,
        }),
        sender: None,
        content: EventContent {
            text: message,
            content_type,
            severity: Some(Severity::from_str(&severity)),
            data: None,
            disclosure: None,
        },
        expects_response: false,
        reply_to: None,
    }
}

async fn send_to_socket_or_inbox(reg: &SessionRegistration, json: &str, event: &Event) {
    match send_via_socket(&reg.socket_path, json).await {
        Ok(()) => eprintln!("Event sent to session {} via socket.", reg.session_id),
        Err(e) => {
            eprintln!(
                "Warning: socket send to session {} failed ({}), falling back to inbox.",
                reg.session_id, e
            );
            if let Err(ie) = write_inbox(event) {
                eprintln!("Inbox fallback also failed: {}", ie);
            }
        }
    }
}

async fn send_via_socket(socket_path: &str, json: &str) -> anyhow::Result<()> {
    // Validate socket_path is inside the registry dir — prevents a crafted
    // registration JSON from redirecting sends to an attacker-controlled socket.
    let sock = std::path::Path::new(socket_path);
    let run_dir = synaps_cli::events::registry::registry_dir();
    if !sock.starts_with(&run_dir) {
        anyhow::bail!(
            "socket path {:?} is outside registry dir — refusing to connect",
            socket_path
        );
    }
    let mut stream = connect_session_transport(socket_path).await?;
    stream.write_all(json.as_bytes()).await?;
    stream.shutdown().await?;
    Ok(())
}

#[cfg(unix)]
async fn connect_session_transport(socket_path: &str) -> anyhow::Result<UnixStream> {
    Ok(UnixStream::connect(socket_path).await?)
}

/// Windows: the socket_path is only an identifier — connect to the named pipe
/// derived from it (must match the engine's `pipe_name_for` derivation).
#[cfg(windows)]
async fn connect_session_transport(
    socket_path: &str,
) -> anyhow::Result<tokio::net::windows::named_pipe::NamedPipeClient> {
    use tokio::net::windows::named_pipe::ClientOptions;
    let pipe_name = synaps_cli::events::socket::pipe_name_for(socket_path);
    Ok(ClientOptions::new().open(&pipe_name)?)
}

fn write_inbox(event: &Event) -> anyhow::Result<()> {
    let inbox_dir = synaps_cli::config::base_dir().join("inbox");
    std::fs::create_dir_all(&inbox_dir)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        if let Ok(meta) = std::fs::metadata(&inbox_dir) {
            let mut perms = meta.permissions();
            perms.set_mode(0o700);
            let _ = std::fs::set_permissions(&inbox_dir, perms);
        }
    }
    let filename = format!(
        "{}-{}.json",
        Utc::now().timestamp_nanos_opt().unwrap_or(0),
        Uuid::new_v4().simple()
    );
    let path = inbox_dir.join(&filename);
    let tmp_path = inbox_dir.join(format!("{}.tmp", filename));
    std::fs::write(&tmp_path, serde_json::to_string_pretty(event)?)?;
    std::fs::rename(&tmp_path, &path)?;
    eprintln!("Event written to inbox: {}", path.display());
    Ok(())
}