urus/examples/crud.rs

//! CRUD example: a tiny user database with JSON persistence.
//!
//! Demonstrates urus and the actor model together:
//!   - The pipeline is shared (Arc) across all connection actors.
//!   - Handlers do NOT take a lock or share mutable state directly.
//!   - A single "store" actor owns the data; handlers send it a request
//!     via a channel and block on the reply. Serialization is structural —
//!     the store processes one request at a time, no Mutex needed.
//!   - On every mutating request the store writes the JSON file. Read
//!     requests don't touch disk.
//!
//! Endpoints:
//!   GET    /users        — list
//!   POST   /users        — create
//!   GET    /users/:id    — fetch
//!   PUT    /users/:id    — replace
//!   DELETE /users/:id    — delete
//!
//! Try it:
//!   cargo run --example crud
//!   curl -s http://localhost:8080/users
//!   curl -s -X POST -d '{"name":"alice","email":"a@x"}' http://localhost:8080/users
//!   curl -s http://localhost:8080/users/1

use serde::{Deserialize, Serialize};
use smarm::{channel, Sender};
use std::sync::OnceLock;
use urus::{serve_with_shutdown, shutdown_handle, Config, Conn, Next, Pipeline, Router};

// ---------------------------------------------------------------------------
// Domain
// ---------------------------------------------------------------------------

#[derive(Clone, Debug, Serialize, Deserialize)]
struct User {
    id:    u64,
    name:  String,
    email: String,
}

#[derive(Deserialize)]
struct NewUser {
    name:  String,
    email: String,
}

// ---------------------------------------------------------------------------
// Store actor protocol
// ---------------------------------------------------------------------------
//
// One enum per request kind. Each carries a `reply` Sender for the response.
// Reply types are kept simple: most are JSON byte vectors + HTTP status. The
// store does the serialisation; the handler just writes the bytes.

enum Request {
    List   { reply: Sender<(u16, Vec<u8>)> },
    Get    { id: u64, reply: Sender<(u16, Vec<u8>)> },
    Create { body: Vec<u8>, reply: Sender<(u16, Vec<u8>)> },
    Update { id: u64, body: Vec<u8>, reply: Sender<(u16, Vec<u8>)> },
    Delete { id: u64, reply: Sender<(u16, Vec<u8>)> },
}

const DB_PATH: &str = "/tmp/urus-crud.json";

// ---------------------------------------------------------------------------
// Store actor body
// ---------------------------------------------------------------------------

fn store_loop(rx: smarm::Receiver<Request>) {
    // Load on start. Missing file = empty store. Corrupt file = panic; we
    // don't auto-rebuild because silently losing data is worse than failing
    // loud.
    let mut users: Vec<User> = match std::fs::read(DB_PATH) {
        Ok(bytes) if !bytes.is_empty() =>
            serde_json::from_slice(&bytes).expect("urus-crud: db file is not valid JSON"),
        _ => Vec::new(),
    };
    let mut next_id: u64 = users.iter().map(|u| u.id).max().unwrap_or(0) + 1;

    loop {
        // recv with a timeout rather than a bare recv: the store must be
        // stoppable at shutdown, but a cross-thread Sender::send (from the
        // stdin thread) can't wake a parked actor — same smarm limitation
        // that motivates urus's SHUTDOWN_POLL. So we wake on our own timer
        // and poll the flag. This poll dies with that limitation too.
        let req = match rx.recv_timeout(std::time::Duration::from_millis(250)) {
            Ok(r) => r,
            Err(smarm::channel::RecvTimeoutError::Timeout) => {
                if SHUTTING_DOWN.load(std::sync::atomic::Ordering::Relaxed) {
                    return;
                }
                continue;
            }
            Err(_) => return, // all senders dropped
        };
        match req {
            Request::List { reply } => {
                let body = serde_json::to_vec(&users).unwrap();
                let _ = reply.send((200, body));
            }
            Request::Get { id, reply } => {
                match users.iter().find(|u| u.id == id) {
                    Some(u) => {
                        let body = serde_json::to_vec(u).unwrap();
                        let _ = reply.send((200, body));
                    }
                    None => {
                        let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec()));
                    }
                }
            }
            Request::Create { body, reply } => {
                match serde_json::from_slice::<NewUser>(&body) {
                    Ok(nu) => {
                        let u = User { id: next_id, name: nu.name, email: nu.email };
                        next_id += 1;
                        users.push(u.clone());
                        persist(&users);
                        let _ = reply.send((201, serde_json::to_vec(&u).unwrap()));
                    }
                    Err(_) => {
                        let _ = reply.send((400, b"{\"error\":\"invalid body\"}".to_vec()));
                    }
                }
            }
            Request::Update { id, body, reply } => {
                match serde_json::from_slice::<NewUser>(&body) {
                    Ok(nu) => match users.iter_mut().find(|u| u.id == id) {
                        Some(u) => {
                            u.name  = nu.name;
                            u.email = nu.email;
                            let snapshot = u.clone();
                            persist(&users);
                            let _ = reply.send((200, serde_json::to_vec(&snapshot).unwrap()));
                        }
                        None => {
                            let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec()));
                        }
                    },
                    Err(_) => {
                        let _ = reply.send((400, b"{\"error\":\"invalid body\"}".to_vec()));
                    }
                }
            }
            Request::Delete { id, reply } => {
                let before = users.len();
                users.retain(|u| u.id != id);
                if users.len() < before {
                    persist(&users);
                    let _ = reply.send((204, Vec::new()));
                } else {
                    let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec()));
                }
            }
        }
    }
}

fn persist(users: &[User]) {
    let json = serde_json::to_vec_pretty(users).unwrap();
    // Atomic-ish: write to temp then rename. Avoids half-written files on
    // crash. /tmp is on the same filesystem so rename is atomic.
    let tmp = format!("{DB_PATH}.tmp");
    std::fs::write(&tmp, &json).expect("urus-crud: write tmp failed");
    std::fs::rename(&tmp, DB_PATH).expect("urus-crud: rename failed");
}

// ---------------------------------------------------------------------------
// Handler helpers
// ---------------------------------------------------------------------------
//
// Once-cell trick: the store actor is spawned the first time a handler
// runs (smarm requires `spawn` to be called from inside an actor — which
// connection actors are). After that all handlers share the same Sender.
// Simpler than threading the Sender through the pipeline at startup.

static STORE_TX: OnceLock<Sender<Request>> = OnceLock::new();

// Set by the stdin thread at shutdown; the store actor polls it (see
// store_loop). An always-on app actor that never returns would otherwise
// block smarm's AllDone and keep serve_with_shutdown from returning.
static SHUTTING_DOWN: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

fn store() -> &'static Sender<Request> {
    STORE_TX.get_or_init(|| {
        let (tx, rx) = channel::<Request>();
        smarm::spawn(move || store_loop(rx));
        tx
    })
}

fn json(conn: Conn, status: u16, body: Vec<u8>) -> Conn {
    conn.put_status(status)
        .put_header("content-type", "application/json")
        .put_body(body)
}

fn parse_id(s: &str) -> Option<u64> {
    s.parse().ok()
}

// ---------------------------------------------------------------------------
// Handlers
// ---------------------------------------------------------------------------

/// SSE demo: `curl -N localhost:8080/ticker` streams a tick every second
/// (with `: keep-alive` comments if it ever goes quiet). The producer
/// exits on SseClosed (client gone / write timeout / shutdown drain) or
/// when the example is shutting down.
fn ticker(conn: Conn, _next: Next) -> Conn {
    let (conn, events) = conn.sse();
    smarm::spawn(move || {
        let mut n: u64 = 0;
        loop {
            if SHUTTING_DOWN.load(std::sync::atomic::Ordering::Relaxed) {
                return; // dropping `events` ends the stream cleanly
            }
            if events.send("tick", &n.to_string()).is_err() {
                return; // SseClosed
            }
            n += 1;
            smarm::sleep(std::time::Duration::from_secs(1));
        }
    });
    conn
}

fn list(conn: Conn, _next: Next) -> Conn {
    let (tx, rx) = channel::<(u16, Vec<u8>)>();
    store().send(Request::List { reply: tx }).ok();
    let (status, body) = rx.recv().expect("store dropped");
    json(conn, status, body)
}

fn create(conn: Conn, _next: Next) -> Conn {
    let body = conn.body.as_bytes().to_vec();
    let (tx, rx) = channel::<(u16, Vec<u8>)>();
    store().send(Request::Create { body, reply: tx }).ok();
    let (status, body) = rx.recv().expect("store dropped");
    json(conn, status, body)
}

fn get_one(conn: Conn, _next: Next) -> Conn {
    let id = match conn.params.get("id").and_then(parse_id) {
        Some(id) => id,
        None     => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()),
    };
    let (tx, rx) = channel::<(u16, Vec<u8>)>();
    store().send(Request::Get { id, reply: tx }).ok();
    let (status, body) = rx.recv().expect("store dropped");
    json(conn, status, body)
}

fn update(conn: Conn, _next: Next) -> Conn {
    let id = match conn.params.get("id").and_then(parse_id) {
        Some(id) => id,
        None     => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()),
    };
    let body = conn.body.as_bytes().to_vec();
    let (tx, rx) = channel::<(u16, Vec<u8>)>();
    store().send(Request::Update { id, body, reply: tx }).ok();
    let (status, body) = rx.recv().expect("store dropped");
    json(conn, status, body)
}

fn delete(conn: Conn, _next: Next) -> Conn {
    let id = match conn.params.get("id").and_then(parse_id) {
        Some(id) => id,
        None     => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()),
    };
    let (tx, rx) = channel::<(u16, Vec<u8>)>();
    store().send(Request::Delete { id, reply: tx }).ok();
    let (status, body) = rx.recv().expect("store dropped");
    json(conn, status, body)
}

// ---------------------------------------------------------------------------
// Minimal request logger
// ---------------------------------------------------------------------------

fn logger(conn: Conn, next: Next) -> Conn {
    let method = conn.method.as_str().to_string();
    let path   = conn.path.clone();
    let conn   = next.run(conn);
    println!("{method} {path} -> {}", conn.status.unwrap_or(0));
    conn
}

// ---------------------------------------------------------------------------
// main
// ---------------------------------------------------------------------------

fn main() {
    let pipeline = Pipeline::new()
        .plug(logger)
        .plug(
            Router::new()
                .get(   "/users",     list)
                .post(  "/users",     create)
                .get(   "/users/:id", get_one)
                .put(   "/users/:id", update)
                .delete("/users/:id", delete)
                .get(   "/ticker",    ticker),
        );

    let cfg = Config::new("127.0.0.1:8080".parse().unwrap());
    println!("urus-crud: DB at {DB_PATH}");
    println!("urus-crud: listening on 127.0.0.1:8080 — press Enter to shut down");

    // Graceful shutdown on stdin-Enter: a plain OS thread blocks on
    // read_line and fires the handle. No signal handling crate needed.
    let (handle, signal) = shutdown_handle();
    std::thread::spawn(move || {
        let mut line = String::new();
        let _ = std::io::stdin().read_line(&mut line);
        println!("urus-crud: shutting down (draining in-flight requests)…");
        SHUTTING_DOWN.store(true, std::sync::atomic::Ordering::Relaxed);
        handle.shutdown();
    });

    serve_with_shutdown(cfg, pipeline, signal).unwrap();
    println!("urus-crud: bye");
}