//! Remote connections, like T3 Code's environments: one copy of the app can
//! let others on the network see and drive its projects and threads. The
//! host runs a small WebSocket server guarded by a key; a client connects
//! with the host's address and that key, and its threads show up in the
//! client's sidebar under the host's name.
//!
//! The protocol is JSON over WebSocket text messages. A client says
//! [`ClientMessage::Hello`] with the key, gets the host's projects and
//! threads in [`ServerMessage::State`], subscribes to the threads it opens,
//! and receives each change as a [`ServerMessage::Thread`] carrying the
//! transcript from the first item that changed. The host also shares its
//! environment settings in [`ServerMessage::Settings`], which clients can
//! change with [`ClientMessage::UpdateSettings`]. The Add project palette
//! browses and adds the host's folders with [`ClientMessage::ListFolders`]
//! and [`ClientMessage::AddProject`]. A client's terminal drawer runs its
//! shells on the host: [`ClientMessage::TerminalOpen`] starts one in a
//! thread's folder there, keys go over as [`ClientMessage::TerminalInput`],
//! and the shell's output comes back as [`ServerMessage::TerminalOutput`].
//! A client's Device panel uses the host's simulators and emulators: it asks
//! with [`ClientMessage::Device`], gets [`ServerMessage::DeviceReply`], and
//! follows [`ServerMessage::DeviceState`]; the video and touches go through
//! the host's [`device_proxy`] on the same port. Clients without a catalog
//! of their own, like the Android app, ask for the host's agents and models
//! with [`ClientMessage::Catalog`] and for a turn's diff with
//! [`ClientMessage::Diff`].

pub mod client;
pub mod device_proxy;
pub mod server;
pub mod shell;

use std::path::PathBuf;

use serde::{Deserialize, Serialize};

use crate::git::{FileDiff, LineKind, TurnChanges};
use crate::projects::ProjectSource;
use crate::ui::folder_browser::Folder;
use crate::models::{Catalog, InstanceState, ModelEntry, ProbeStatus, Provider};
use crate::session::{QueuedMessage, SavedItem};
use crate::settings::SettingsPatch;
use crate::worktree::PullRequest;

/// The port a host listens on unless told otherwise.
pub const DEFAULT_PORT: u16 = 47_823;
/// Bumped when the messages change incompatibly.
pub const PROTOCOL: u32 = 1;

/// A thread as the sidebar needs it.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(default)]
pub struct ThreadSummary {
    pub id: String,
    pub project: PathBuf,
    pub title: String,
    pub provider: Provider,
    pub instance: String,
    pub model: Option<String>,
    pub permission_mode: String,
    pub reasoning: String,
    pub fast_mode: bool,
    /// `starting`, `idle`, `working` or `exited`.
    pub status: String,
    /// A tool call or question waits for the user.
    pub attention: bool,
    pub settled: bool,
    pub pinned: bool,
    /// Milliseconds since the epoch, like every time here.
    pub snoozed_until: Option<u64>,
    pub created: u64,
    pub updated: u64,
    pub last_message: u64,
    /// The thread's worktree branch, if it runs in one.
    pub branch: Option<String>,
    pub total_cost: f64,
    /// Whether anything was sent in it yet.
    pub has_items: bool,
    /// For a subagent's thread, the thread that delegated its task.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub parent: Option<String>,
}

/// The rest of an open thread's state, sent with its transcript.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(default)]
pub struct ThreadDetail {
    pub cwd: PathBuf,
    pub active_model: Option<String>,
    pub thinking: bool,
    pub preparing: bool,
    /// How long the running turn has been going.
    pub turn_elapsed_ms: Option<u64>,
    pub context_tokens: Option<u64>,
    pub context_window: Option<u64>,
    pub queue: Vec<QueuedMessage>,
    pub limit_resets: Option<u64>,
    pub resume_at: Option<u64>,
    /// Bound for a worktree made with the first message.
    pub pending_worktree: bool,
    pub pull_request: Option<PullRequest>,
    pub turn_changes: Vec<(usize, TurnChanges)>,
    pub background_tasks: Vec<RemoteTask>,
}

/// A command, monitor or subagent the host's agent left running.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(default)]
pub struct RemoteTask {
    pub task_id: String,
    pub description: String,
    pub elapsed_ms: u64,
}

#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(default)]
pub struct ProjectInfo {
    pub path: PathBuf,
    pub name: String,
    /// `None` outside a git repository.
    pub branch: Option<String>,
    pub branches: Vec<String>,
    pub git: bool,
    pub added: u64,
    pub removed: u64,
}

/// A file sent along with a message.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Upload {
    pub name: String,
    /// Base64.
    pub data: String,
}

/// What a client asks of the host's devices. Threads are the host's ids.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "call", rename_all = "snake_case")]
pub enum DeviceCall {
    /// Send the device state now and whenever it changes.
    Watch,
    /// Refresh the devices, starting the hub once device support is on.
    List,
    Configure { enabled: Option<bool>, agent_access: Option<bool>, onboarding: Option<bool> },
    Retry,
    Open { thread: String, device: String, boot: bool },
    Close { thread: String, device: Option<String>, shutdown: bool },
    Detail { device: String },
    Action { device: String, action: crate::device::Action },
    /// A PNG of the screen, base64.
    Screenshot { device: String },
    Accessibility { device: String },
}

/// What a client asks of a thread on the host.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "action", rename_all = "snake_case")]
pub enum ThreadAction {
    Send { text: String, attachments: Vec<Upload> },
    Queue { text: String, attachments: Vec<Upload> },
    SteerQueued { ix: usize },
    RemoveQueued { ix: usize },
    Interrupt,
    /// `allow`, `always` or `deny`.
    Permission { ix: usize, decision: String },
    Answer { ix: usize, option: usize },
    /// Answers an AskUserQuestion form, one answer per question.
    AskSubmit { ix: usize, answers: Vec<String> },
    AskSkip { ix: usize },
    /// Sends a running command or subagent to the background.
    Background { ix: usize },
    StopTask { task_id: String },
    SetModel { model: Option<String> },
    SetPermissionMode { mode: String },
    SetReasoning { reasoning: String },
    SetFastMode { on: bool },
    Settle,
    Unsettle,
    Pin { pinned: bool },
    Snooze { until: Option<u64> },
    Rename { title: String },
    CancelResume,
    ScheduleResume { at: u64 },
    Delete,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum ClientMessage {
    Hello { key: String, name: String, protocol: u32 },
    Subscribe { thread: String },
    Unsubscribe { thread: String },
    Action { thread: String, action: ThreadAction },
    /// Starts a thread in one of the host's projects with its first message.
    Create {
        reference: u64,
        project: PathBuf,
        instance: String,
        provider: Provider,
        model: Option<String>,
        permission_mode: String,
        reasoning: String,
        fast_mode: bool,
        worktree: bool,
        /// The branch a worktree starts from.
        base: Option<String>,
        text: String,
        attachments: Vec<Upload>,
    },
    /// Changes the host's environment settings or its projects' overrides.
    UpdateSettings { patch: SettingsPatch },
    /// Lists a folder on the host for the Add project palette, as typed
    /// there: `~/Work/`. Older hosts don't answer.
    ListFolders { path: String },
    /// Adds a project on the host, cloning or creating it as asked.
    AddProject { source: ProjectSource },
    /// Starts a shell on the host in a thread's folder (the project's, for
    /// a thread not created yet). `terminal` is the client's id for it.
    /// Older hosts don't answer.
    TerminalOpen { terminal: u64, thread: Option<String>, project: PathBuf, columns: u16, lines: u16 },
    /// What was typed in a terminal, base64.
    TerminalInput { terminal: u64, data: String },
    TerminalResize { terminal: u64, columns: u16, lines: u16 },
    /// Hangs up on a terminal's shell.
    TerminalClose { terminal: u64 },
    /// Something for the host's devices; `request` pairs it with its reply.
    /// Older hosts don't answer.
    Device { request: u64, call: DeviceCall },
    /// Asks for the host's agents, models and favorites, answered with
    /// [`ServerMessage::Catalog`]. Older hosts don't answer.
    Catalog,
    /// Asks for the diff between two of a thread's snapshots (a turn's
    /// `base` and `head`), for one file or all of them. Older hosts don't
    /// answer.
    Diff { request: u64, thread: String, base: String, head: String, path: Option<String>, ignore_whitespace: bool },
}

#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum ServerMessage {
    Welcome { host: String, version: String, protocol: u32 },
    Denied { reason: String },
    State { projects: Vec<ProjectInfo>, threads: Vec<ThreadSummary> },
    /// A subscribed thread changed: its state, and its transcript from item
    /// `from` on (earlier items are as last sent); `len` items in all.
    Thread { summary: ThreadSummary, detail: ThreadDetail, from: usize, len: usize, items: Vec<SavedItem> },
    Created { reference: u64, thread: String },
    Error { message: String },
    /// The host's environment settings and project overrides, sent on
    /// connecting and whenever they change. Older hosts don't send it.
    Settings { values: serde_json::Map<String, serde_json::Value> },
    /// The subfolders of a folder a client asked for, or why it can't have them.
    Folders { path: String, folders: Vec<Folder>, truncated: bool, error: Option<String> },
    /// The project a client asked for is in the host's list now.
    ProjectAdded { path: PathBuf },
    /// A terminal's shell printed something, base64.
    TerminalOutput { terminal: u64, data: String },
    /// A terminal's shell exited, or with `error`, couldn't start.
    TerminalExited { terminal: u64, error: Option<String> },
    /// The answer to a [`ClientMessage::Device`]: its result or what went wrong.
    DeviceReply { request: u64, ok: Option<serde_json::Value>, error: Option<String> },
    /// The host's devices, for clients that watch them.
    DeviceState { state: crate::device::State },
    /// An agent on the host opened a device in one of its threads.
    DeviceOpened { thread: String, device: crate::device::DeviceSummary },
    /// An agent closed a thread's device, or all of them without one.
    DeviceClosed { thread: String, device: Option<String> },
    /// The host's agents and the models they offer, as its pickers show
    /// them, and its favorite models by key (`instance:id`).
    Catalog { instances: Vec<RemoteInstance>, favorites: Vec<String> },
    /// The answer to a [`ClientMessage::Diff`]: the changed files, or what
    /// went wrong.
    DiffReply { request: u64, files: Vec<RemoteFileDiff>, error: Option<String> },
}

/// A provider instance on the host and what its last health check found.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(default)]
pub struct RemoteInstance {
    pub id: String,
    pub provider: Provider,
    pub name: String,
    /// A hex color (`#3e9bfa`) marking the instance, if it has one.
    pub accent: Option<String>,
    /// `ready`, `checking`, `missing`, `disabled` or `failed`.
    pub status: String,
    pub account: Option<String>,
    pub plan: Option<String>,
    /// The id of Claude's `default` entry, when it reports one.
    pub default_model: Option<String>,
    pub models: Vec<RemoteModel>,
    pub commands: Vec<RemoteCommand>,
}

/// A model as the host's picker knows it; see [`ModelEntry`].
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(default)]
pub struct RemoteModel {
    pub id: String,
    pub resolved: String,
    pub name: String,
    pub detail: String,
    pub legacy: bool,
    /// Reasoning levels it accepts, as the CLI spells them (`high`).
    pub efforts: Vec<String>,
    pub default_effort: Option<String>,
    pub ultracode: bool,
    pub fast_mode: bool,
    pub auto_mode: bool,
    pub context_window: Option<u64>,
    /// Hidden from the model picker in the host's settings.
    pub hidden: bool,
}

/// A `/` command or skill an instance offers.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(default)]
pub struct RemoteCommand {
    pub name: String,
    pub description: String,
    pub argument_hint: String,
    pub skill: bool,
}

/// One file of a diff.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(default)]
pub struct RemoteFileDiff {
    pub path: String,
    pub binary: bool,
    pub lines: Vec<RemoteDiffLine>,
}

#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(default)]
pub struct RemoteDiffLine {
    /// `hunk`, `context`, `added` or `removed`.
    pub kind: String,
    pub text: String,
    pub old_line: Option<u32>,
    pub new_line: Option<u32>,
}

impl RemoteInstance {
    pub fn from_state(state: &InstanceState, catalog: &Catalog) -> Self {
        let status = match state.status {
            ProbeStatus::Ready => "ready",
            ProbeStatus::Checking => "checking",
            ProbeStatus::Missing => "missing",
            ProbeStatus::Disabled => "disabled",
            ProbeStatus::Failed(_) => "failed",
        };
        let accent = state.config.accent.trim();
        Self {
            id: state.config.id.clone(),
            provider: state.config.kind,
            name: state.label(),
            accent: (!accent.is_empty()).then(|| accent.to_string()),
            status: status.into(),
            account: state.account.clone(),
            plan: state.plan.clone(),
            default_model: state.default.as_ref().map(|model| model.id.clone()),
            models: state.models.iter().map(|model| RemoteModel::from_entry(model, catalog)).collect(),
            commands: state
                .commands
                .iter()
                .map(|command| RemoteCommand {
                    name: command.name.clone(),
                    description: command.description.clone(),
                    argument_hint: command.argument_hint.clone(),
                    skill: command.skill,
                })
                .collect(),
        }
    }
}

impl RemoteModel {
    pub fn from_entry(model: &ModelEntry, catalog: &Catalog) -> Self {
        let profile = model.claude_profile();
        Self {
            id: model.id.clone(),
            resolved: model.resolved.clone(),
            name: model.name.clone(),
            detail: model.detail.clone(),
            legacy: model.legacy,
            efforts: model.efforts.iter().map(|effort| effort.as_cli().to_string()).collect(),
            default_effort: profile.default_effort.map(|effort| effort.as_cli().to_string()),
            ultracode: profile.ultracode,
            fast_mode: model.fast_mode,
            auto_mode: model.auto_mode,
            context_window: model.context_window,
            hidden: catalog.is_hidden(model),
        }
    }
}

impl From<&FileDiff> for RemoteFileDiff {
    fn from(file: &FileDiff) -> Self {
        let lines = file
            .lines
            .iter()
            .map(|line| RemoteDiffLine {
                kind: match line.kind {
                    LineKind::Hunk => "hunk",
                    LineKind::Context => "context",
                    LineKind::Added => "added",
                    LineKind::Removed => "removed",
                }
                .into(),
                text: line.text.clone(),
                old_line: line.old_line,
                new_line: line.new_line,
            })
            .collect();
        Self { path: file.path.clone(), binary: file.binary, lines }
    }
}

/// A new random key: 24 bytes, hex.
pub fn new_key() -> String {
    let mut bytes = [0u8; 24];
    let read = std::fs::File::open("/dev/urandom").and_then(|mut file| std::io::Read::read_exact(&mut file, &mut bytes));
    if read.is_err() {
        // Not as good, but never empty.
        let seed = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_nanos();
        for (ix, byte) in bytes.iter_mut().enumerate() {
            *byte = (seed >> ((ix % 16) * 8)) as u8 ^ (ix as u8).wrapping_mul(151);
        }
    }
    bytes.iter().map(|byte| format!("{byte:02x}")).collect()
}

/// Compares keys in time independent of where they differ.
pub fn keys_match(a: &str, b: &str) -> bool {
    a.len() == b.len() && a.bytes().zip(b.bytes()).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
}

/// A pairing code bundles a host's address and key: `192.168.2.10:47823/<key>`.
pub fn pairing_code(address: &str, key: &str) -> String {
    format!("{address}/{key}")
}

/// Splits a pairing code, or an address alone, into address and key.
pub fn parse_pairing(code: &str) -> (String, Option<String>) {
    let code = code.trim().trim_start_matches("ws://").trim_start_matches("vectrum://");
    match code.split_once('/') {
        Some((address, key)) if !key.trim().is_empty() => (address.to_string(), Some(key.trim().to_string())),
        Some((address, _)) => (address.to_string(), None),
        None => (code.to_string(), None),
    }
}

/// `host` or `host:port`, with the default port filled in.
pub fn with_port(address: &str) -> String {
    let address = address.trim();
    let has_port = match address.rsplit_once(':') {
        // An IPv6 address without brackets has colons but no port.
        Some((host, port)) => !host.is_empty() && port.parse::<u16>().is_ok() && (!host.contains(':') || host.ends_with(']')),
        None => false,
    };
    if has_port { address.to_string() } else { format!("{address}:{DEFAULT_PORT}") }
}

/// This machine's IPv4 addresses on the network, for showing where to connect.
pub fn local_addresses() -> Vec<String> {
    let output = std::process::Command::new("hostname").arg("-I").stderr(std::process::Stdio::null()).output();
    let text = output.map(|output| String::from_utf8_lossy(&output.stdout).to_string()).unwrap_or_default();
    text.split_whitespace().filter(|address| address.parse::<std::net::Ipv4Addr>().is_ok()).map(str::to_string).collect()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[::core::prelude::v1::test]
    fn pairing_codes() {
        assert_eq!(parse_pairing("192.168.2.10:47823/abc"), ("192.168.2.10:47823".into(), Some("abc".into())));
        assert_eq!(parse_pairing(" 10.0.0.2 "), ("10.0.0.2".into(), None));
        assert_eq!(with_port("10.0.0.2"), format!("10.0.0.2:{DEFAULT_PORT}"));
        assert_eq!(with_port("10.0.0.2:9000"), "10.0.0.2:9000");
        assert_eq!(with_port("box.local"), format!("box.local:{DEFAULT_PORT}"));
        assert_eq!(with_port("[::1]:9000"), "[::1]:9000");
    }

    #[::core::prelude::v1::test]
    fn keys() {
        let key = new_key();
        assert_eq!(key.len(), 48);
        assert_ne!(key, new_key());
        assert!(keys_match(&key, &key.clone()));
        assert!(!keys_match(&key, &new_key()));
        assert!(!keys_match("a", "ab"));
    }

    #[::core::prelude::v1::test]
    fn messages_round_trip() {
        let message = ClientMessage::Action {
            thread: "t1".into(),
            action: ThreadAction::Send { text: "hi".into(), attachments: Vec::new() },
        };
        let text = serde_json::to_string(&message).unwrap();
        assert_eq!(serde_json::from_str::<ClientMessage>(&text).unwrap(), message);

        let patch: SettingsPatch = serde_json::from_str(r#"{"values": {"workspace": "worktree"}}"#).unwrap();
        let message = ClientMessage::UpdateSettings { patch };
        let text = serde_json::to_string(&message).unwrap();
        assert_eq!(serde_json::from_str::<ClientMessage>(&text).unwrap(), message);
        let message = ClientMessage::AddProject {
            source: ProjectSource::Forge { forge: crate::projects::Forge::GitHub, repo: "owner/repo".into() },
        };
        let text = serde_json::to_string(&message).unwrap();
        assert_eq!(serde_json::from_str::<ClientMessage>(&text).unwrap(), message);
        let message =
            ClientMessage::TerminalOpen { terminal: 3, thread: None, project: "/p".into(), columns: 80, lines: 24 };
        let text = serde_json::to_string(&message).unwrap();
        assert_eq!(serde_json::from_str::<ClientMessage>(&text).unwrap(), message);
        let text = serde_json::to_string(&ServerMessage::TerminalExited { terminal: 3, error: None }).unwrap();
        assert!(matches!(
            serde_json::from_str::<ServerMessage>(&text).unwrap(),
            ServerMessage::TerminalExited { terminal: 3, error: None }
        ));
        let message = ClientMessage::Device {
            request: 4,
            call: DeviceCall::Action {
                device: "emulator-5554".into(),
                action: crate::device::Action::SetPermission {
                    app_id: "com.example".into(),
                    permission: "camera".into(),
                    decision: "grant".into(),
                },
            },
        };
        let text = serde_json::to_string(&message).unwrap();
        assert_eq!(serde_json::from_str::<ClientMessage>(&text).unwrap(), message);
        let message = ClientMessage::Device { request: 5, call: DeviceCall::Action { device: "d".into(), action: crate::device::Action::SetAppearance(true) } };
        let text = serde_json::to_string(&message).unwrap();
        assert_eq!(serde_json::from_str::<ClientMessage>(&text).unwrap(), message);
        let state = crate::device::State { enabled: true, host_status: crate::device::HostStatus::Ready, ..Default::default() };
        let text = serde_json::to_string(&ServerMessage::DeviceState { state }).unwrap();
        match serde_json::from_str::<ServerMessage>(&text).unwrap() {
            ServerMessage::DeviceState { state } => assert_eq!(state.host_status, crate::device::HostStatus::Ready),
            _ => panic!("not a device state"),
        }
        let values = crate::settings::Settings::default().environment_values();
        let text = serde_json::to_string(&ServerMessage::Settings { values: values.clone() }).unwrap();
        match serde_json::from_str::<ServerMessage>(&text).unwrap() {
            ServerMessage::Settings { values: received } => assert_eq!(received, values),
            _ => panic!("not settings"),
        }

        let text = serde_json::to_string(&ClientMessage::Catalog).unwrap();
        assert_eq!(text, r#"{"type":"catalog"}"#);
        assert_eq!(serde_json::from_str::<ClientMessage>(&text).unwrap(), ClientMessage::Catalog);
        let instance = RemoteInstance {
            id: "claude".into(),
            name: "Claude".into(),
            status: "ready".into(),
            default_model: Some("default".into()),
            models: vec![RemoteModel {
                id: "opus".into(),
                efforts: vec!["low".into(), "high".into()],
                default_effort: Some("medium".into()),
                ..Default::default()
            }],
            commands: vec![RemoteCommand { name: "compact".into(), ..Default::default() }],
            ..Default::default()
        };
        let message = ServerMessage::Catalog { instances: vec![instance.clone()], favorites: vec!["claude:opus".into()] };
        let text = serde_json::to_string(&message).unwrap();
        match serde_json::from_str::<ServerMessage>(&text).unwrap() {
            ServerMessage::Catalog { instances, favorites } => {
                assert_eq!(instances, [instance]);
                assert_eq!(favorites, ["claude:opus"]);
            }
            _ => panic!("not a catalog"),
        }
        // Clients may leave out what they don't know.
        let model: RemoteModel = serde_json::from_str(r#"{"id": "sonnet"}"#).unwrap();
        assert_eq!((model.id.as_str(), model.hidden), ("sonnet", false));

        let message = ClientMessage::Diff {
            request: 6,
            thread: "t1".into(),
            base: "abc".into(),
            head: "def".into(),
            path: Some("src/a.rs".into()),
            ignore_whitespace: true,
        };
        let text = serde_json::to_string(&message).unwrap();
        assert_eq!(serde_json::from_str::<ClientMessage>(&text).unwrap(), message);
        let file = crate::git::FileDiff {
            path: "src/a.rs".into(),
            binary: false,
            lines: vec![
                crate::git::DiffLine { kind: LineKind::Hunk, text: "@@ -1 +1 @@".into(), old_line: None, new_line: None },
                crate::git::DiffLine { kind: LineKind::Added, text: "new".into(), old_line: None, new_line: Some(1) },
            ],
        };
        let files = vec![RemoteFileDiff::from(&file)];
        assert_eq!(files[0].lines[1].kind, "added");
        let text = serde_json::to_string(&ServerMessage::DiffReply { request: 6, files: files.clone(), error: None }).unwrap();
        match serde_json::from_str::<ServerMessage>(&text).unwrap() {
            ServerMessage::DiffReply { request: 6, files: received, error: None } => assert_eq!(received, files),
            _ => panic!("not a diff reply"),
        }
    }

    #[::core::prelude::v1::test]
    fn client_and_server_talk() {
        use std::time::Duration;

        use super::client::{ClientEvent, ConnectionStatus, RemoteClient};
        use super::server::{RemoteServer, ServerEvent};

        let port = 47_900 + (std::process::id() % 90) as u16;
        let key = new_key();
        let server = RemoteServer::start(port, key.clone(), "test-host".into()).unwrap();
        fn wait<T>(receiver: &smol::channel::Receiver<T>) -> Option<T> {
            smol::block_on(async {
                smol::future::or(async { receiver.recv().await.ok() }, async {
                    smol::Timer::after(Duration::from_secs(10)).await;
                    None
                })
                .await
            })
        }

        let client = RemoteClient::connect(&format!("127.0.0.1:{port}"), key, "test-client".into());
        assert!(matches!(wait(&client.events), Some(ClientEvent::Status(ConnectionStatus::Connecting))));
        match wait(&client.events) {
            Some(ClientEvent::Status(ConnectionStatus::Connected(host))) => assert_eq!(host, "test-host"),
            _ => panic!("didn't connect"),
        }
        let conn = match wait(&server.events) {
            Some(ServerEvent::Connected { conn, name, .. }) => {
                assert_eq!(name, "test-client");
                conn
            }
            _ => panic!("server didn't see the client"),
        };
        client.send(&ClientMessage::Subscribe { thread: "t1".into() });
        match wait(&server.events) {
            Some(ServerEvent::Message { message: ClientMessage::Subscribe { thread }, .. }) => assert_eq!(thread, "t1"),
            _ => panic!("server didn't get the message"),
        }
        server.send(conn, &ServerMessage::Created { reference: 7, thread: "t2".into() });
        match wait(&client.events) {
            Some(ClientEvent::Message(ServerMessage::Created { reference, thread })) => {
                assert_eq!((reference, thread.as_str()), (7, "t2"))
            }
            _ => panic!("client didn't get the reply"),
        }

        let intruder = RemoteClient::connect(&format!("127.0.0.1:{port}"), "wrong".into(), "intruder".into());
        assert!(matches!(wait(&intruder.events), Some(ClientEvent::Status(ConnectionStatus::Connecting))));
        assert!(matches!(wait(&intruder.events), Some(ClientEvent::Status(ConnectionStatus::Denied(_)))));
    }
}