//! pickle - CSP/Actor building blocks for Rust
//!
//! Generic channel types that eliminate per-actor boilerplate. Actors own
//! their state inside a spawned thread; callers hold typed channel handles
//! that are Clone + Send. No `Arc<Mutex<T>>`, no lifetime annotations.
//!
//! # Channel types
//!
//! | Type | Pattern | Caller | Actor |
//! |------|---------|--------|-------|
//! | [`Func<Req, Resp>`] | function call | `call(req) -> resp` | `msg.req`, `msg.reply(resp)` |
//! | [`Query<Resp>`] | getter | `call() -> resp` | `msg.reply(resp)` |
//! | [`Proc<Req>`] | command | `call(req)` blocks | `msg.req`, `msg.done()` |
//! | [`Signal`] | trigger | `call()` blocks | `msg.done()` |
//! | [`Inbox<T>`] | fire-and-forget | `send(v)` / `try_send(v)` | recv from `receiver()` |
//! | [`Lifecycle`] | shutdown | `stop()` | `stopping()` in select |

pub use crossbeam_channel::{self, Receiver, Sender, select};
use crossbeam_channel::bounded;
use std::thread::JoinHandle;
use std::time::Duration;

// ---------------------------------------------------------------------------
// CallError
// ---------------------------------------------------------------------------

/// Error returned by `call_timeout` methods.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CallError {
    /// The operation timed out.
    Timeout,
    /// The receiving end was disconnected.
    Disconnected,
}

// ---------------------------------------------------------------------------
// Func<Req, Resp>
// ---------------------------------------------------------------------------

/// Message sent through a [`Func`] channel.
pub struct FuncMsg<Req, Resp> {
    pub req: Req,
    resp_tx: Sender<Resp>,
}

impl<Req, Resp> FuncMsg<Req, Resp> {
    /// Reply to the caller. Consumes self - double-reply is a compile error.
    pub fn reply(self, resp: Resp) {
        let _ = self.resp_tx.send(resp);
    }
}

/// Synchronous request-response channel. Rendezvous (zero-capacity) on the
/// request side; the caller blocks until the actor picks up the message and
/// replies.
#[derive(Clone)]
pub struct Func<Req, Resp> {
    tx: Sender<FuncMsg<Req, Resp>>,
    rx: Receiver<FuncMsg<Req, Resp>>,
}

impl<Req: Send + 'static, Resp: Send + 'static> Func<Req, Resp> {
    pub fn new() -> Self {
        let (tx, rx) = bounded(0);
        Func { tx, rx }
    }

    /// Call the actor with `req` and block until a response arrives.
    pub fn call(&self, req: Req) -> Resp {
        let (resp_tx, resp_rx) = bounded(1);
        let msg = FuncMsg { req, resp_tx };
        self.tx.send(msg).expect("actor disconnected");
        resp_rx.recv().expect("actor dropped response channel")
    }

    /// Call with a timeout on both send and receive phases.
    pub fn call_timeout(&self, req: Req, timeout: Duration) -> Result<Resp, CallError> {
        let (resp_tx, resp_rx) = bounded(1);
        let msg = FuncMsg { req, resp_tx };
        self.tx
            .send_timeout(msg, timeout)
            .map_err(|e| match e {
                crossbeam_channel::SendTimeoutError::Timeout(_) => CallError::Timeout,
                crossbeam_channel::SendTimeoutError::Disconnected(_) => CallError::Disconnected,
            })?;
        resp_rx.recv_timeout(timeout).map_err(|e| match e {
            crossbeam_channel::RecvTimeoutError::Timeout => CallError::Timeout,
            crossbeam_channel::RecvTimeoutError::Disconnected => CallError::Disconnected,
        })
    }

    /// Clone of the receive end for the actor's `select!`.
    pub fn receiver(&self) -> Receiver<FuncMsg<Req, Resp>> {
        self.rx.clone()
    }
}

impl<Req: Send + 'static, Resp: Send + 'static> Default for Func<Req, Resp> {
    fn default() -> Self {
        Self::new()
    }
}

// ---------------------------------------------------------------------------
// Query<Resp>
// ---------------------------------------------------------------------------

/// Message sent through a [`Query`] channel.
pub struct QueryMsg<Resp> {
    resp_tx: Sender<Resp>,
}

impl<Resp> QueryMsg<Resp> {
    /// Reply to the caller. Consumes self.
    pub fn reply(self, resp: Resp) {
        let _ = self.resp_tx.send(resp);
    }
}

/// Synchronous getter channel. Like [`Func`] but with no request payload.
#[derive(Clone)]
pub struct Query<Resp> {
    tx: Sender<QueryMsg<Resp>>,
    rx: Receiver<QueryMsg<Resp>>,
}

impl<Resp: Send + 'static> Query<Resp> {
    pub fn new() -> Self {
        let (tx, rx) = bounded(0);
        Query { tx, rx }
    }

    /// Block until the actor replies.
    pub fn call(&self) -> Resp {
        let (resp_tx, resp_rx) = bounded(1);
        let msg = QueryMsg { resp_tx };
        self.tx.send(msg).expect("actor disconnected");
        resp_rx.recv().expect("actor dropped response channel")
    }

    /// Call with a timeout.
    pub fn call_timeout(&self, timeout: Duration) -> Result<Resp, CallError> {
        let (resp_tx, resp_rx) = bounded(1);
        let msg = QueryMsg { resp_tx };
        self.tx
            .send_timeout(msg, timeout)
            .map_err(|e| match e {
                crossbeam_channel::SendTimeoutError::Timeout(_) => CallError::Timeout,
                crossbeam_channel::SendTimeoutError::Disconnected(_) => CallError::Disconnected,
            })?;
        resp_rx.recv_timeout(timeout).map_err(|e| match e {
            crossbeam_channel::RecvTimeoutError::Timeout => CallError::Timeout,
            crossbeam_channel::RecvTimeoutError::Disconnected => CallError::Disconnected,
        })
    }

    pub fn receiver(&self) -> Receiver<QueryMsg<Resp>> {
        self.rx.clone()
    }
}

impl<Resp: Send + 'static> Default for Query<Resp> {
    fn default() -> Self {
        Self::new()
    }
}

// ---------------------------------------------------------------------------
// Proc<Req>
// ---------------------------------------------------------------------------

/// Message sent through a [`Proc`] channel.
pub struct ProcMsg<Req> {
    pub req: Req,
    done_tx: Sender<()>,
}

impl<Req> ProcMsg<Req> {
    /// Signal that processing is complete. Consumes self.
    pub fn done(self) {
        let _ = self.done_tx.send(());
    }
}

/// Synchronous command channel. Caller sends a request and blocks until the
/// actor calls `done()`.
#[derive(Clone)]
pub struct Proc<Req> {
    tx: Sender<ProcMsg<Req>>,
    rx: Receiver<ProcMsg<Req>>,
}

impl<Req: Send + 'static> Proc<Req> {
    pub fn new() -> Self {
        let (tx, rx) = bounded(0);
        Proc { tx, rx }
    }

    /// Send `req` and block until the actor calls `done()`.
    pub fn call(&self, req: Req) {
        let (done_tx, done_rx) = bounded(1);
        let msg = ProcMsg { req, done_tx };
        self.tx.send(msg).expect("actor disconnected");
        done_rx.recv().expect("actor dropped done channel");
    }

    /// Call with a timeout.
    pub fn call_timeout(&self, req: Req, timeout: Duration) -> Result<(), CallError> {
        let (done_tx, done_rx) = bounded(1);
        let msg = ProcMsg { req, done_tx };
        self.tx
            .send_timeout(msg, timeout)
            .map_err(|e| match e {
                crossbeam_channel::SendTimeoutError::Timeout(_) => CallError::Timeout,
                crossbeam_channel::SendTimeoutError::Disconnected(_) => CallError::Disconnected,
            })?;
        done_rx.recv_timeout(timeout).map_err(|e| match e {
            crossbeam_channel::RecvTimeoutError::Timeout => CallError::Timeout,
            crossbeam_channel::RecvTimeoutError::Disconnected => CallError::Disconnected,
        })
    }

    pub fn receiver(&self) -> Receiver<ProcMsg<Req>> {
        self.rx.clone()
    }
}

impl<Req: Send + 'static> Default for Proc<Req> {
    fn default() -> Self {
        Self::new()
    }
}

// ---------------------------------------------------------------------------
// Signal
// ---------------------------------------------------------------------------

/// Message sent through a [`Signal`] channel.
pub struct SignalMsg {
    done_tx: Sender<()>,
}

impl SignalMsg {
    /// Signal that processing is complete. Consumes self.
    pub fn done(self) {
        let _ = self.done_tx.send(());
    }
}

/// Synchronous trigger channel. Like [`Proc`] with no payload.
#[derive(Clone)]
pub struct Signal {
    tx: Sender<SignalMsg>,
    rx: Receiver<SignalMsg>,
}

impl Signal {
    pub fn new() -> Self {
        let (tx, rx) = bounded(0);
        Signal { tx, rx }
    }

    /// Block until the actor calls `done()`.
    pub fn call(&self) {
        let (done_tx, done_rx) = bounded(1);
        let msg = SignalMsg { done_tx };
        self.tx.send(msg).expect("actor disconnected");
        done_rx.recv().expect("actor dropped done channel");
    }

    /// Call with a timeout.
    pub fn call_timeout(&self, timeout: Duration) -> Result<(), CallError> {
        let (done_tx, done_rx) = bounded(1);
        let msg = SignalMsg { done_tx };
        self.tx
            .send_timeout(msg, timeout)
            .map_err(|e| match e {
                crossbeam_channel::SendTimeoutError::Timeout(_) => CallError::Timeout,
                crossbeam_channel::SendTimeoutError::Disconnected(_) => CallError::Disconnected,
            })?;
        done_rx.recv_timeout(timeout).map_err(|e| match e {
            crossbeam_channel::RecvTimeoutError::Timeout => CallError::Timeout,
            crossbeam_channel::RecvTimeoutError::Disconnected => CallError::Disconnected,
        })
    }

    pub fn receiver(&self) -> Receiver<SignalMsg> {
        self.rx.clone()
    }
}

impl Default for Signal {
    fn default() -> Self {
        Self::new()
    }
}

// ---------------------------------------------------------------------------
// Inbox<T>
// ---------------------------------------------------------------------------

/// Buffered fire-and-forget channel.
#[derive(Clone)]
pub struct Inbox<T> {
    tx: Sender<T>,
    rx: Receiver<T>,
}

impl<T: Send + 'static> Inbox<T> {
    /// Create an inbox with the given buffer capacity.
    pub fn new(buffer: usize) -> Self {
        let (tx, rx) = bounded(buffer);
        Inbox { tx, rx }
    }

    /// Blocking send.
    pub fn send(&self, v: T) {
        self.tx.send(v).expect("actor disconnected");
    }

    /// Non-blocking send. Returns `false` if the buffer is full.
    pub fn try_send(&self, v: T) -> bool {
        self.tx.try_send(v).is_ok()
    }

    /// Clone of the receive end for the actor's `select!`.
    pub fn receiver(&self) -> Receiver<T> {
        self.rx.clone()
    }
}

// ---------------------------------------------------------------------------
// Lifecycle
// ---------------------------------------------------------------------------

#[allow(dead_code)] // held for Drop side-effect only
struct DoneGuard(Option<Sender<()>>);

impl Drop for DoneGuard {
    fn drop(&mut self) {
        // Sender drops, done_rx sees Disconnected, stop() unblocks.
    }
}

/// Manages actor startup and shutdown.
///
/// The pattern:
/// 1. Create a `Lifecycle` with `new()`.
/// 2. Clone `stopping()` for the actor's select loop.
/// 3. Call `spawn()` to start the actor thread.
/// 4. Call `stop()` to shut down - drops the stop sender (actor's `stopping()`
///    receiver fires on disconnect), then blocks until the actor thread exits.
///
/// `stop()` is idempotent after the first call (the second `take()` is a no-op
/// on the already-consumed Option). `spawn()` panics if called twice.
pub struct Lifecycle {
    stop_tx: Option<Sender<()>>,
    stop_rx: Receiver<()>,
    done_tx: Option<Sender<()>>,
    done_rx: Receiver<()>,
}

impl Lifecycle {
    pub fn new() -> Self {
        let (stop_tx, stop_rx) = bounded(0);
        let (done_tx, done_rx) = bounded(0);
        Lifecycle {
            stop_tx: Some(stop_tx),
            stop_rx,
            done_tx: Some(done_tx),
            done_rx,
        }
    }

    /// Returns a receiver that fires (with `Err(Disconnected)`) when `stop()`
    /// is called. Use in `select!` with `-> _` to catch the disconnect:
    ///
    /// ```ignore
    /// select! {
    ///     recv(stopping) -> _ => return,
    ///     // ...
    /// }
    /// ```
    pub fn stopping(&self) -> Receiver<()> {
        self.stop_rx.clone()
    }

    /// Returns a receiver that fires when the actor thread exits (the
    /// `DoneGuard` is dropped). For external observers that need to wait
    /// on actor completion without calling `stop()`.
    pub fn stopped(&self) -> Receiver<()> {
        self.done_rx.clone()
    }

    /// Signal the actor to stop and block until it exits.
    ///
    /// Drops the stop sender (causing the actor's `stopping()` receiver to
    /// disconnect), then waits for the done channel to disconnect (which
    /// happens when the `DoneGuard` inside the actor thread is dropped).
    pub fn stop(&mut self) {
        debug_assert!(
            self.done_tx.is_none(),
            "stop() called before spawn()"
        );
        self.stop_tx.take(); // drop sends disconnect to stopping()
        let _ = self.done_rx.recv(); // blocks until DoneGuard drops
    }
}

impl Default for Lifecycle {
    fn default() -> Self {
        Self::new()
    }
}

/// Spawn an actor thread with lifecycle management.
///
/// Takes the done sender out of the lifecycle and wraps it in a `DoneGuard`
/// whose `Drop` impl closes the done channel - whether the closure returns
/// normally or panics. Panics if called twice on the same lifecycle.
pub fn spawn<F>(lc: &mut Lifecycle, f: F) -> JoinHandle<()>
where
    F: FnOnce() + Send + 'static,
{
    let done_tx = lc.done_tx.take().expect("spawn() called twice on same Lifecycle");
    let guard = DoneGuard(Some(done_tx));
    std::thread::spawn(move || {
        let _guard = guard;
        f();
    })
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap;
    use std::time::Duration;

    // Counter: Signal + Query
    #[test]
    fn counter() {
        let inc = Signal::new();
        let get = Query::<i32>::new();
        let mut lc = Lifecycle::new();

        let inc_rx = inc.receiver();
        let get_rx = get.receiver();
        let stopping = lc.stopping();

        spawn(&mut lc, move || {
            let mut n: i32 = 0;
            loop {
                select! {
                    recv(stopping) -> _ => return,
                    recv(inc_rx) -> msg => {
                        msg.unwrap().done();
                        n += 1;
                    },
                    recv(get_rx) -> msg => {
                        msg.unwrap().reply(n);
                    },
                }
            }
        });

        inc.call();
        inc.call();
        inc.call();
        assert_eq!(get.call(), 3);
        lc.stop();
    }

    // KV store: Func + Proc
    #[test]
    fn kv_store() {
        let set = Proc::<(String, String)>::new();
        let get = Func::<String, Option<String>>::new();
        let mut lc = Lifecycle::new();

        let set_rx = set.receiver();
        let get_rx = get.receiver();
        let stopping = lc.stopping();

        spawn(&mut lc, move || {
            let mut store = HashMap::<String, String>::new();
            loop {
                select! {
                    recv(stopping) -> _ => return,
                    recv(set_rx) -> msg => {
                        let msg = msg.unwrap();
                        store.insert(msg.req.0.clone(), msg.req.1.clone());
                        msg.done();
                    },
                    recv(get_rx) -> msg => {
                        let msg = msg.unwrap();
                        let val = store.get(&msg.req).cloned();
                        msg.reply(val);
                    },
                }
            }
        });

        set.call(("a".into(), "1".into()));
        set.call(("b".into(), "2".into()));
        assert_eq!(get.call("a".into()), Some("1".into()));
        assert_eq!(get.call("b".into()), Some("2".into()));
        assert_eq!(get.call("c".into()), None);
        lc.stop();
    }

    // Logger: Inbox + Query
    #[test]
    fn logger() {
        let log = Inbox::<String>::new(64);
        let count = Query::<usize>::new();
        let mut lc = Lifecycle::new();

        let log_rx = log.receiver();
        let count_rx = count.receiver();
        let stopping = lc.stopping();

        spawn(&mut lc, move || {
            let mut n: usize = 0;
            loop {
                select! {
                    recv(stopping) -> _ => return,
                    recv(log_rx) -> msg => {
                        let _ = msg.unwrap();
                        n += 1;
                    },
                    recv(count_rx) -> msg => {
                        msg.unwrap().reply(n);
                    },
                }
            }
        });

        log.send("one".into());
        log.send("two".into());
        log.send("three".into());
        // sync on count to ensure all three are processed
        assert_eq!(count.call(), 3);
        lc.stop();
    }

    // Inbox try_send: buffer full returns false
    #[test]
    fn inbox_try_send() {
        let inbox = Inbox::<i32>::new(2);
        assert!(inbox.try_send(1));
        assert!(inbox.try_send(2));
        assert!(!inbox.try_send(3)); // buffer full
    }

    // Lifecycle stopped: observable from outside
    #[test]
    fn lifecycle_stopped() {
        let mut lc = Lifecycle::new();
        let stopped = lc.stopped();
        let stopping = lc.stopping();

        spawn(&mut lc, move || {
            // wait for stop signal then exit
            let _ = stopping.recv();
        });

        // not stopped yet
        assert!(stopped.recv_timeout(Duration::from_millis(50)).is_err());

        lc.stop();

        // now the done channel is disconnected - recv returns Err
        // (it already returned from the recv in stop(), so this should
        // also see Disconnected)
        assert!(stopped.recv_timeout(Duration::from_millis(50)).is_err());
    }

    // call_timeout: unlistened channel times out
    #[test]
    fn call_timeout_expires() {
        let sig = Signal::new();
        // nobody is listening on the receiver
        let result = sig.call_timeout(Duration::from_millis(50));
        assert_eq!(result, Err(CallError::Timeout));
    }

    // call_timeout on Func
    #[test]
    fn func_call_timeout() {
        let f = Func::<i32, i32>::new();
        let result = f.call_timeout(42, Duration::from_millis(50));
        assert_eq!(result, Err(CallError::Timeout));
    }

    // call_timeout on Query
    #[test]
    fn query_call_timeout() {
        let q = Query::<i32>::new();
        let result = q.call_timeout(Duration::from_millis(50));
        assert_eq!(result, Err(CallError::Timeout));
    }

    // call_timeout on Proc
    #[test]
    fn proc_call_timeout() {
        let p = Proc::<i32>::new();
        let result = p.call_timeout(42, Duration::from_millis(50));
        assert_eq!(result, Err(CallError::Timeout));
    }

    // Concurrent callers: 100 threads incrementing, final count correct
    #[test]
    fn concurrent_callers() {
        let inc = Signal::new();
        let get = Query::<i32>::new();
        let mut lc = Lifecycle::new();

        let inc_rx = inc.receiver();
        let get_rx = get.receiver();
        let stopping = lc.stopping();

        spawn(&mut lc, move || {
            let mut n: i32 = 0;
            loop {
                select! {
                    recv(stopping) -> _ => return,
                    recv(inc_rx) -> msg => {
                        msg.unwrap().done();
                        n += 1;
                    },
                    recv(get_rx) -> msg => {
                        msg.unwrap().reply(n);
                    },
                }
            }
        });

        let mut handles = Vec::new();
        for _ in 0..100 {
            let inc = inc.clone();
            handles.push(std::thread::spawn(move || {
                inc.call();
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
        assert_eq!(get.call(), 100);
        lc.stop();
    }

    // Clone callers: clone a Func handle, call from both
    #[test]
    fn clone_callers() {
        let f = Func::<i32, i32>::new();
        let mut lc = Lifecycle::new();

        let f_rx = f.receiver();
        let stopping = lc.stopping();

        spawn(&mut lc, move || {
            loop {
                select! {
                    recv(stopping) -> _ => return,
                    recv(f_rx) -> msg => {
                        let msg = msg.unwrap();
                        let result = msg.req * 2;
                        msg.reply(result);
                    },
                }
            }
        });

        let f2 = f.clone();
        assert_eq!(f.call(5), 10);
        assert_eq!(f2.call(7), 14);
        lc.stop();
    }
}