use std::cell::Cell;
use std::collections::BTreeMap;
use std::io::{self, BufRead, Write};
use std::time::{Duration, Instant};

use lazily::stdlib::{
    RevisionBarrier, RevisionCheck, RevisionWaitOutcome, Timeout, TimeoutCancellation,
    TimeoutOperation, TimeoutPoll, TimeoutUnavailableReason, Timer, TimerError, TimerPoll,
};
use lazily::{
    Context, CrdtOp, CrdtPlaneRuntime, CrdtSync, IpcMessage, IpcValue, NodeId, NodeKey, PeerId,
    WireStamp,
};
use serde_json::{Value, json};

const PROTOCOL_VERSION: u64 = 1;
const STDLIB_FEATURES: [&str; 3] = [
    "stdlib_timer_v1",
    "stdlib_timeout_v1",
    "stdlib_revision_barrier_v1",
];

struct Peer {
    peer_id: Option<PeerId>,
    context: Context,
    runtime: Option<CrdtPlaneRuntime>,
    stdlib: BTreeMap<String, StdlibFeature>,
}

impl Peer {
    fn new() -> Self {
        Self {
            peer_id: None,
            context: Context::new(),
            runtime: None,
            stdlib: BTreeMap::new(),
        }
    }

    fn handle(&mut self, request: &Value) -> Value {
        let Some(command) = request.get("cmd").and_then(Value::as_str) else {
            return error("missing cmd");
        };
        match command {
            "hello" => self.hello(request),
            "local_set" => self.local_set(request),
            "deliver" => self.deliver(request),
            "snapshot" => self.snapshot(),
            "feature_reset" => self.feature_reset(request),
            "feature_step" => self.feature_step(request),
            "feature_observe" => self.feature_observe(request),
            "bye" => json!({"ok": true}),
            command if command.starts_with("link_") => json!({
                "ok": false,
                "error": "unsupported channel",
                "unsupported": true
            }),
            _ => error(format!("unknown command {command}")),
        }
    }

    fn hello(&mut self, request: &Value) -> Value {
        if request.get("protocol_version").and_then(Value::as_u64) != Some(PROTOCOL_VERSION) {
            return error(format!(
                "unsupported protocol_version {}",
                request.get("protocol_version").unwrap_or(&Value::Null)
            ));
        }
        let Some(peer) = request.get("peer").and_then(Value::as_u64) else {
            return error("hello requires peer");
        };
        let peer = PeerId(peer);
        self.peer_id = Some(peer);
        self.runtime = Some(CrdtPlaneRuntime::new(peer));
        json!({
            "ok": true,
            "binding": "lazily-rs",
            "version": env!("CARGO_PKG_VERSION"),
            "protocol_version": PROTOCOL_VERSION,
            "features": [
                "distributed_crdt",
                "stdlib_timer_v1",
                "stdlib_timeout_v1",
                "stdlib_revision_barrier_v1"
            ],
            "codecs": ["json", "msgpack"],
            "channels": [],
            "channel_variants": {},
            "platform_profile": "portable",
            "carve_outs": ["transport_links"]
        })
    }

    fn local_set(&mut self, request: &Value) -> Value {
        let Some(peer) = self.peer_id else {
            return error("hello must run first");
        };
        let Some(node) = request.get("node").and_then(Value::as_u64) else {
            return error("local_set requires node");
        };
        let key = match request.get("key") {
            Some(Value::Null) => None,
            Some(Value::String(key)) => match NodeKey::new(key) {
                Ok(key) => Some(key),
                Err(error) => return self::error(format!("invalid key: {error}")),
            },
            _ => return error("local_set requires nullable key"),
        };
        let state: IpcValue = match request.get("state").cloned().map(serde_json::from_value) {
            Some(Ok(state)) => state,
            Some(Err(error)) => return self::error(format!("invalid IpcValue: {error}")),
            None => return error("local_set requires state"),
        };
        let Some(at) = request.get("at").and_then(Value::as_u64) else {
            return error("local_set requires at");
        };
        let runtime = match self.runtime.as_mut() {
            Some(runtime) => runtime,
            None => return error("hello must run first"),
        };
        let stamp = runtime.plane_mut().tick(at);
        let wire_stamp = WireStamp::from(stamp);
        let op = match key {
            Some(key) => CrdtOp::keyed(NodeId(node), key, wire_stamp, state),
            None => CrdtOp::new(NodeId(node), wire_stamp, state),
        };
        let sync = CrdtSync::new(vec![(peer.0, wire_stamp)], vec![op]);
        if runtime.ingest(&self.context, &sync, at) != 1 {
            return error("production runtime rejected its fresh local op");
        }
        match serde_json::to_value(IpcMessage::CrdtSync(sync)) {
            Ok(frame) => json!({"ok": true, "frame": frame}),
            Err(error) => self::error(format!("encode CrdtSync: {error}")),
        }
    }

    fn deliver(&mut self, request: &Value) -> Value {
        let Some(frame) = request.get("frame").cloned() else {
            return error("deliver requires frame");
        };
        let message: IpcMessage = match serde_json::from_value(frame) {
            Ok(message) => message,
            Err(error) => return self::error(format!("decode IpcMessage: {error}")),
        };
        let IpcMessage::CrdtSync(sync) = message else {
            return error("deliver requires CrdtSync");
        };
        let Some(at) = request.get("at").and_then(Value::as_u64) else {
            return error("deliver requires at");
        };
        let Some(runtime) = self.runtime.as_mut() else {
            return error("hello must run first");
        };
        let applied = runtime.ingest(&self.context, &sync, at);
        json!({"ok": true, "applied": applied})
    }

    fn snapshot(&self) -> Value {
        let Some(runtime) = self.runtime.as_ref() else {
            return error("hello must run first");
        };
        let cells = runtime
            .converged()
            .into_iter()
            .map(|entry| {
                json!({
                    "node": entry.node.0,
                    "key": entry.key.map(|key| key.as_str().to_owned()),
                    "state": entry.state
                })
            })
            .collect::<Vec<_>>();
        json!({"ok": true, "cells": cells})
    }

    fn feature_reset(&mut self, request: &Value) -> Value {
        let Some(feature) = request.get("feature").and_then(Value::as_str) else {
            return error("feature_reset requires feature");
        };
        if !STDLIB_FEATURES.contains(&feature) {
            return json!({
                "ok": false,
                "error": format!("unsupported feature {feature}"),
                "unsupported": true
            });
        }
        let Some(adapter) = StdlibFeature::new(feature) else {
            unreachable!("validated stdlib feature has an adapter");
        };
        self.stdlib.insert(feature.to_owned(), adapter);
        json!({"ok": true, "feature": feature})
    }

    fn feature_step(&mut self, request: &Value) -> Value {
        let Some(feature) = request.get("feature").and_then(Value::as_str) else {
            return error("feature_step requires feature");
        };
        let Some(step) = request.get("step") else {
            return error("feature_step requires step");
        };
        let Some(adapter) = self.stdlib.get_mut(feature) else {
            return error(format!("feature {feature} must be reset before stepping"));
        };
        match adapter.step(step) {
            Ok(observation) => json!({
                "ok": true,
                "feature": feature,
                "observation": observation
            }),
            Err(message) => error(message),
        }
    }

    fn feature_observe(&self, request: &Value) -> Value {
        let Some(feature) = request.get("feature").and_then(Value::as_str) else {
            return error("feature_observe requires feature");
        };
        let Some(adapter) = self.stdlib.get(feature) else {
            return error(format!(
                "feature {feature} must be reset before observation"
            ));
        };
        match adapter.observation() {
            Some(observation) => json!({
                "ok": true,
                "feature": feature,
                "observation": observation
            }),
            None => error(format!("feature {feature} has no observation")),
        }
    }
}

enum StdlibFeature {
    Timer(TimerFeature),
    Timeout(TimeoutFeature),
    Barrier(BarrierFeature),
}

impl StdlibFeature {
    fn new(feature: &str) -> Option<Self> {
        match feature {
            "stdlib_timer_v1" => Some(Self::Timer(TimerFeature::new())),
            "stdlib_timeout_v1" => Some(Self::Timeout(TimeoutFeature::new())),
            "stdlib_revision_barrier_v1" => Some(Self::Barrier(BarrierFeature::new())),
            _ => None,
        }
    }

    fn step(&mut self, step: &Value) -> Result<Value, String> {
        match self {
            Self::Timer(adapter) => adapter.step(step),
            Self::Timeout(adapter) => adapter.step(step),
            Self::Barrier(adapter) => adapter.step(step),
        }
    }

    fn observation(&self) -> Option<&Value> {
        match self {
            Self::Timer(adapter) => adapter.last.as_ref(),
            Self::Timeout(adapter) => adapter.last.as_ref(),
            Self::Barrier(adapter) => adapter.last.as_ref(),
        }
    }
}

struct TimerFeature {
    base: Instant,
    timer: Option<Timer>,
    deadline: Option<u64>,
    last: Option<Value>,
}

impl TimerFeature {
    fn new() -> Self {
        Self {
            base: Instant::now(),
            timer: None,
            deadline: None,
            last: None,
        }
    }

    fn step(&mut self, step: &Value) -> Result<Value, String> {
        let observation = match string_field(step, "op")? {
            "start" => {
                let now = u64_field(step, "now")?;
                let duration = u64_field(step, "duration")?;
                match Timer::checked_deadline_ticks(now, duration) {
                    Ok(deadline) => {
                        self.deadline = Some(deadline);
                        self.timer = Some(
                            Timer::try_after_at(
                                logical_instant(self.base, now)?,
                                Duration::from_nanos(duration),
                            )
                            .map_err(|error| error.to_string())?,
                        );
                        json!({"outcome": "pending", "deadline": deadline})
                    }
                    Err(TimerError::DeadlineOverflow) => {
                        json!({"outcome": "unavailable", "reason": "deadline_overflow"})
                    }
                    Err(TimerError::ClockRegression) => unreachable!(),
                }
            }
            "observe" => {
                let now = u64_field(step, "now")?;
                let timer = self.timer.as_mut().ok_or("timer feature is not started")?;
                match timer.try_poll_at(logical_instant(self.base, now)?) {
                    Ok(TimerPoll::Pending { .. }) => {
                        json!({"outcome": "pending", "deadline": self.deadline})
                    }
                    Ok(TimerPoll::Fired { .. }) => json!({
                        "outcome": "fired",
                        "fired_at": timer
                            .fired_at()
                            .expect("production timer records fire edge")
                            .duration_since(self.base)
                            .as_nanos() as u64
                    }),
                    Err(TimerError::ClockRegression) => json!({
                        "outcome": "unavailable",
                        "reason": "clock_regression",
                        "deadline": self.deadline
                    }),
                    Err(TimerError::DeadlineOverflow) => unreachable!(),
                }
            }
            op => return Err(format!("unsupported timer feature step {op}")),
        };
        self.last = Some(observation.clone());
        Ok(observation)
    }
}

struct TimeoutFeature {
    base: Instant,
    timeout: Option<Timeout<String>>,
    deadline: Option<u64>,
    last: Option<Value>,
}

impl TimeoutFeature {
    fn new() -> Self {
        Self {
            base: Instant::now(),
            timeout: None,
            deadline: None,
            last: None,
        }
    }

    fn step(&mut self, step: &Value) -> Result<Value, String> {
        let observation = match string_field(step, "op")? {
            "start" => {
                let now = u64_field(step, "now")?;
                let duration = u64_field(step, "duration")?;
                let deadline = Timer::checked_deadline_ticks(now, duration)
                    .map_err(|error| error.to_string())?;
                self.deadline = Some(deadline);
                self.timeout = Some(
                    Timeout::try_after_at(
                        logical_instant(self.base, now)?,
                        Duration::from_nanos(duration),
                    )
                    .map_err(|error| error.to_string())?,
                );
                json!({"outcome": "pending", "deadline": deadline})
            }
            "poll" => {
                let now = u64_field(step, "now")?;
                let operation = string_field(step, "operation")?;
                let cancellation = string_field(step, "cancellation")?;
                if !matches!(operation, "pending" | "completed" | "unavailable") {
                    return Err(format!("unsupported operation {operation}"));
                }
                if !matches!(cancellation, "pending" | "cancelled" | "unavailable") {
                    return Err(format!("unsupported cancellation {cancellation}"));
                }
                let value = step.get("value").and_then(Value::as_str).map(str::to_owned);
                let operation_calls = Cell::new(0_u64);
                let cancellation_calls = Cell::new(0_u64);
                let timeout = self
                    .timeout
                    .as_mut()
                    .ok_or("timeout feature is not started")?;
                let poll = timeout.poll_at_with_cancellation(
                    logical_instant(self.base, now)?,
                    || {
                        operation_calls.set(operation_calls.get() + 1);
                        match operation {
                            "pending" => TimeoutOperation::Pending,
                            "completed" => TimeoutOperation::Completed(
                                value.clone().expect("completed step has value"),
                            ),
                            "unavailable" => TimeoutOperation::Unavailable,
                            _ => unreachable!("operation validated below"),
                        }
                    },
                    || {
                        cancellation_calls.set(cancellation_calls.get() + 1);
                        match cancellation {
                            "pending" => TimeoutCancellation::Pending,
                            "cancelled" => TimeoutCancellation::Cancelled,
                            "unavailable" => TimeoutCancellation::Unavailable,
                            _ => unreachable!("cancellation validated below"),
                        }
                    },
                );
                let operation_calls = operation_calls.get();
                let cancellation_calls = cancellation_calls.get();
                match poll {
                    TimeoutPoll::Pending { .. } => json!({
                        "outcome": "pending",
                        "deadline": self.deadline,
                        "operation_calls": operation_calls,
                        "cancellation_calls": cancellation_calls
                    }),
                    TimeoutPoll::Completed(value) => json!({
                        "outcome": "completed",
                        "value": value,
                        "operation_calls": operation_calls,
                        "cancellation_calls": cancellation_calls
                    }),
                    TimeoutPoll::TimedOut => json!({
                        "outcome": "timed_out",
                        "operation_calls": operation_calls,
                        "cancellation_calls": cancellation_calls
                    }),
                    TimeoutPoll::Cancelled => json!({
                        "outcome": "cancelled",
                        "operation_calls": operation_calls,
                        "cancellation_calls": cancellation_calls
                    }),
                    TimeoutPoll::Unavailable => {
                        let reason = match timeout.unavailable_reason() {
                            Some(TimeoutUnavailableReason::Operation) | None => {
                                "operation_unavailable"
                            }
                            Some(TimeoutUnavailableReason::Cancellation) => {
                                "cancellation_unavailable"
                            }
                            Some(TimeoutUnavailableReason::ClockRegression) => "clock_regression",
                        };
                        json!({
                            "outcome": "unavailable",
                            "reason": reason,
                            "operation_calls": operation_calls,
                            "cancellation_calls": cancellation_calls
                        })
                    }
                }
            }
            op => return Err(format!("unsupported timeout feature step {op}")),
        };
        self.last = Some(observation.clone());
        Ok(observation)
    }
}

struct BarrierFeature {
    barrier: Option<RevisionBarrier>,
    required_revision: u64,
    deadline: Option<u64>,
    last: Option<Value>,
}

impl BarrierFeature {
    fn new() -> Self {
        Self {
            barrier: None,
            required_revision: 0,
            deadline: None,
            last: None,
        }
    }

    fn step(&mut self, step: &Value) -> Result<Value, String> {
        let observation = match string_field(step, "op")? {
            "start" => {
                self.required_revision = u64_field(step, "required_revision")?;
                self.deadline = step.get("deadline").and_then(value_u64);
                self.barrier = Some(RevisionBarrier::new(u64_field(step, "revision")?));
                barrier_value("pending", self.barrier.as_ref().unwrap(), None)
            }
            "register_recheck" => {
                let barrier = self
                    .barrier
                    .as_ref()
                    .ok_or("barrier feature is not started")?
                    .clone();
                if !barrier.observe_clock(u64_field(step, "now")?) {
                    if barrier.clock_regressed() {
                        barrier_value("unavailable", &barrier, Some("clock_regression"))
                    } else {
                        barrier_value("disposed", &barrier, None)
                    }
                } else {
                    let observed = u64_field(step, "observed_revision")?;
                    let predicate = bool_field(step, "predicate")?;
                    let after = self.required_revision.saturating_sub(1);
                    let _ = barrier.advance(observed);
                    if predicate && barrier.revision() >= self.required_revision {
                        barrier_wait_value(
                            barrier.wait_after(after, |_| RevisionCheck::Satisfied, None, None),
                            &barrier,
                        )
                    } else {
                        barrier_value("pending", &barrier, None)
                    }
                }
            }
            "advance" => {
                let barrier = self
                    .barrier
                    .as_ref()
                    .ok_or("barrier feature is not started")?;
                let _ = barrier.advance(u64_field(step, "revision")?);
                if barrier.is_disposed() {
                    barrier_value("disposed", barrier, None)
                } else if bool_field(step, "predicate")?
                    && barrier.revision() >= self.required_revision
                {
                    barrier_wait_value(
                        barrier.wait_after(
                            self.required_revision.saturating_sub(1),
                            |_| RevisionCheck::Satisfied,
                            None,
                            None,
                        ),
                        barrier,
                    )
                } else {
                    barrier_value("pending", barrier, None)
                }
            }
            "observe" => {
                let barrier = self
                    .barrier
                    .as_ref()
                    .ok_or("barrier feature is not started")?;
                let now = u64_field(step, "now")?;
                if !barrier.observe_clock(now) {
                    let (outcome, reason) = if barrier.clock_regressed() {
                        ("unavailable", Some("clock_regression"))
                    } else {
                        ("disposed", None)
                    };
                    let mut value = barrier_value(outcome, barrier, reason);
                    value["cancellation_calls"] = json!(0);
                    value
                } else {
                    let reached = self.deadline.is_some_and(|deadline| now >= deadline);
                    let predicate = bool_field(step, "predicate")?;
                    let cancellation = string_field(step, "cancellation")?;
                    let mut timer = reached.then(|| Timer::after(Duration::ZERO));
                    let owned = barrier.cancellation();
                    let foreign_barrier = RevisionBarrier::new(0);
                    let foreign = foreign_barrier.cancellation();
                    if !reached && cancellation == "cancelled" {
                        let _ = owned.cancel();
                    }
                    let token = match cancellation {
                        "cancelled" | "pending" => Some(&owned),
                        "unavailable" => Some(&foreign),
                        value => return Err(format!("unsupported cancellation {value}")),
                    };
                    let predicate_satisfied =
                        predicate && barrier.revision() >= self.required_revision;
                    let mut value = if reached || predicate_satisfied || cancellation != "pending" {
                        barrier_wait_value(
                            barrier.wait_after(
                                self.required_revision.saturating_sub(1),
                                |_| {
                                    if predicate {
                                        RevisionCheck::Satisfied
                                    } else {
                                        RevisionCheck::Pending
                                    }
                                },
                                timer.as_mut(),
                                token,
                            ),
                            barrier,
                        )
                    } else {
                        barrier_value("pending", barrier, None)
                    };
                    value["cancellation_calls"] =
                        json!(if reached || predicate_satisfied { 0 } else { 1 });
                    value
                }
            }
            "dispose" => {
                let barrier = self
                    .barrier
                    .as_ref()
                    .ok_or("barrier feature is not started")?;
                let _ = barrier.dispose();
                barrier_value("disposed", barrier, None)
            }
            "receipt" => {
                let barrier = self
                    .barrier
                    .as_ref()
                    .ok_or("barrier feature is not started")?;
                barrier.notify();
                barrier_value("pending", barrier, None)
            }
            op => return Err(format!("unsupported revision barrier feature step {op}")),
        };
        self.last = Some(observation.clone());
        Ok(observation)
    }
}

fn logical_instant(base: Instant, tick: u64) -> Result<Instant, String> {
    base.checked_add(Duration::from_nanos(tick))
        .ok_or_else(|| "logical instant exceeds the platform Instant range".to_owned())
}

fn string_field<'a>(value: &'a Value, field: &str) -> Result<&'a str, String> {
    value
        .get(field)
        .and_then(Value::as_str)
        .ok_or_else(|| format!("feature step requires string {field}"))
}

fn u64_field(value: &Value, field: &str) -> Result<u64, String> {
    value
        .get(field)
        .and_then(value_u64)
        .ok_or_else(|| format!("feature step requires u64 {field}"))
}

fn value_u64(value: &Value) -> Option<u64> {
    value
        .as_u64()
        .or_else(|| value.as_str().and_then(|text| text.parse().ok()))
}

fn bool_field(value: &Value, field: &str) -> Result<bool, String> {
    value
        .get(field)
        .and_then(Value::as_bool)
        .ok_or_else(|| format!("feature step requires boolean {field}"))
}

fn barrier_value(outcome: &str, barrier: &RevisionBarrier, reason: Option<&str>) -> Value {
    let mut value = json!({
        "outcome": outcome,
        "revision": barrier.revision(),
        "generation": barrier.generation()
    });
    if let Some(reason) = reason {
        value["reason"] = json!(reason);
    }
    value
}

fn barrier_wait_value(outcome: RevisionWaitOutcome, barrier: &RevisionBarrier) -> Value {
    match outcome {
        RevisionWaitOutcome::Satisfied { .. } => barrier_value("satisfied", barrier, None),
        RevisionWaitOutcome::TimedOut { .. } => barrier_value("timed_out", barrier, None),
        RevisionWaitOutcome::Cancelled { .. } => barrier_value("cancelled", barrier, None),
        RevisionWaitOutcome::Disposed { .. } => barrier_value("disposed", barrier, None),
        RevisionWaitOutcome::Unavailable { .. } => barrier_value(
            "unavailable",
            barrier,
            Some(if barrier.clock_regressed() {
                "clock_regression"
            } else {
                "cancellation_unavailable"
            }),
        ),
    }
}

fn error(message: impl Into<String>) -> Value {
    json!({"ok": false, "error": message.into()})
}

fn self_check() -> Result<(), String> {
    let mut peer = Peer::new();
    let hello = peer.handle(&json!({
        "cmd": "hello",
        "peer": 1,
        "protocol_version": PROTOCOL_VERSION
    }));
    if hello["ok"] != true {
        return Err(format!("hello failed: {hello}"));
    }
    let local = peer.handle(&json!({
        "cmd": "local_set",
        "node": 7,
        "key": null,
        "state": {"Inline": [65]},
        "at": 10
    }));
    if local["ok"] != true || local.pointer("/frame/CrdtSync/ops/0/key") != Some(&Value::Null) {
        return Err(format!("local_set failed canonical key check: {local}"));
    }
    let delivered = peer.handle(&json!({
        "cmd": "deliver",
        "frame": local["frame"],
        "at": 11
    }));
    if delivered["applied"] != 0 {
        return Err(format!(
            "duplicate delivery was not idempotent: {delivered}"
        ));
    }
    let snapshot = peer.handle(&json!({"cmd": "snapshot"}));
    if snapshot.pointer("/cells/0/state/Inline/0") != Some(&json!(65)) {
        return Err(format!("snapshot mismatch: {snapshot}"));
    }

    let feature_cases = [
        (
            "stdlib_timer_v1",
            vec![
                json!({"op": "start", "now": 0, "duration": 0}),
                json!({"op": "observe", "now": 0}),
            ],
            "fired",
        ),
        (
            "stdlib_timeout_v1",
            vec![
                json!({"op": "start", "now": 0, "duration": 1}),
                json!({
                    "op": "poll",
                    "now": 0,
                    "operation": "completed",
                    "value": "ok",
                    "cancellation": "pending"
                }),
            ],
            "completed",
        ),
        (
            "stdlib_revision_barrier_v1",
            vec![
                json!({
                    "op": "start",
                    "revision": 1,
                    "required_revision": 1,
                    "deadline": null
                }),
                json!({
                    "op": "observe",
                    "now": 0,
                    "predicate": true,
                    "cancellation": "pending"
                }),
            ],
            "satisfied",
        ),
    ];
    for (feature, steps, expected) in feature_cases {
        let reset = peer.handle(&json!({"cmd": "feature_reset", "feature": feature}));
        if reset["ok"] != true {
            return Err(format!("{feature} reset failed: {reset}"));
        }
        for step in steps {
            let reply =
                peer.handle(&json!({"cmd": "feature_step", "feature": feature, "step": step}));
            if reply["ok"] != true {
                return Err(format!("{feature} step failed: {reply}"));
            }
        }
        let observed = peer.handle(&json!({"cmd": "feature_observe", "feature": feature}));
        if observed.pointer("/observation/outcome") != Some(&json!(expected)) {
            return Err(format!("{feature} observation mismatch: {observed}"));
        }
    }
    Ok(())
}

fn main() {
    if std::env::args().any(|argument| argument == "--self-check") {
        match self_check() {
            Ok(()) => {
                println!("lazily-rs interop peer self-check: ok");
                return;
            }
            Err(error) => {
                eprintln!("lazily-rs interop peer self-check: {error}");
                std::process::exit(1);
            }
        }
    }

    let stdin = io::stdin();
    let mut stdout = io::stdout().lock();
    let mut peer = Peer::new();
    for line in stdin.lock().lines() {
        let response = match line {
            Ok(line) => match serde_json::from_str::<Value>(&line) {
                Ok(request) => {
                    let bye = request.get("cmd").and_then(Value::as_str) == Some("bye");
                    let response = peer.handle(&request);
                    if let Err(error) = write_response(&mut stdout, &response) {
                        eprintln!("write response: {error}");
                        std::process::exit(1);
                    }
                    if bye {
                        return;
                    }
                    continue;
                }
                Err(error) => self::error(format!("invalid JSON: {error}")),
            },
            Err(error) => {
                eprintln!("read request: {error}");
                std::process::exit(1);
            }
        };
        if let Err(error) = write_response(&mut stdout, &response) {
            eprintln!("write response: {error}");
            std::process::exit(1);
        }
    }
}

fn write_response(output: &mut impl Write, response: &Value) -> io::Result<()> {
    serde_json::to_writer(&mut *output, response)?;
    output.write_all(b"\n")?;
    output.flush()
}
