Learn Rust in 1 Day
— The Keethu Way

# One command — installs rustc, cargo, rustup
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

# Verify
rustc --version    # rustc 1.80+
cargo --version    # cargo 1.80+

fn main() {
    // Variables are immutable by default — Rust forces you to be explicit
    let name = "Keethu";           // type inferred: &str
    let subscribers: u64 = 100_000; // _ as visual separator
    let latency_ms: f64 = 9.7;

    // mut = explicitly opt into mutability
    let mut count = 0u32;
    count += 1;

    println!("{name} handles {subscribers} subs at {latency_ms}ms");

    // Shadowing — redeclare same name, even with different type
    let count = count.to_string();  // now it's a String
    println!("count is now a String: {count}");

    // Functions are first-class
    let result = add(40, 2);
    println!("40 + 2 = {result}");
}

// No 'return' needed — last expression is the return value
fn add(a: i32, b: i32) -> i32 {
    a + b   // no semicolon = expression, not statement
}

fn control_flow() {
    let x = 42;

    // if is an expression
    let label = if x > 100 { "big" } else { "small" };

    // loop returns a value via break
    let found = loop {
        if x == 42 { break true; }
        break false;
    };

    // for over a range
    for i in 0..5 {          // 0,1,2,3,4 (exclusive end)
        print!("{i} ");
    }

    // while let — loop until pattern doesn't match
    let mut stack = vec![1, 2, 3];
    while let Some(top) = stack.pop() {
        println!("{top}");
    }
}

fn main() {
    // String is heap-allocated — it has one owner
    let s1 = String::from("hello");
    let s2 = s1;  // s1 is MOVED into s2 — s1 is now invalid
    // println!("{s1}");  ← COMPILER ERROR: value borrowed after move
    println!("{s2}");  // s2 owns it now

    // Primitives (i32, bool, f64) implement Copy — no move
    let x = 5;
    let y = x;    // x is COPIED, both x and y are valid
    println!("{x} {y}");  // works fine

    // When s2 goes out of scope here, String is freed automatically
    // This is Rust's "drop" — like a destructor, but guaranteed
}

fn takes_ownership(s: String) {
    println!("{s}");
    // s is dropped here — caller can no longer use it
}

fn gives_ownership() -> String {
    String::from("I'm yours now")  // moves out of function
}

fn main() {
    let s = String::from("Keethu");

    // Borrow with & — s is still the owner
    let len = calculate_len(&s);  // pass a reference
    println!("{s} has {len} chars");  // s still valid!

    // Mutable borrow — you can change through it
    let mut msg = String::from("hello");
    append_world(&mut msg);
    println!("{msg}");  // "hello, world!"
}

fn calculate_len(s: &String) -> usize {
    s.len()  // s is borrowed — not moved, not dropped when fn returns
}

fn append_world(s: &mut String) {
    s.push_str(", world!");
}

fn main() {
    let s = String::from("hello world");

    // String slice — a reference to part of a String
    let hello = &s[0..5];   // &str pointing into s
    let world = &s[6..];   // ..11 implicit

    // String literals ARE slices — &str pointing into the binary
    let literal: &str = "I live in the binary";

    // Array slice
    let arr = [1, 2, 3, 4, 5];
    let mid: &[i32] = &arr[1..4];  // [2, 3, 4]

    // &str vs String:
    //   &str  = borrowed view (immutable, fixed-size)
    //   String = owned, growable heap string
    // Prefer &str in function params — accepts both &str and &String
    greet("literal");   // works
    greet(&s);           // works — coerces &String to &str
}

fn greet(name: &str) {
    println!("Hello, {name}!");
}

// Named fields struct
struct Subscriber {
    id:    u64,
    topic: String,
    alive: bool,
}

// Tuple struct — named tuple type
struct Latency(f64);  // Latency(9.7).0

// Unit struct — no fields, useful for traits
struct Marker;

impl Subscriber {
    // Associated function (no self) — like a static method
    fn new(id: u64, topic: &str) -> Self {
        Subscriber { id, topic: topic.to_string(), alive: true }
    }

    // Method — takes &self (shared borrow)
    fn is_active(&self) -> bool {
        self.alive
    }

    // Mutable method — takes &mut self
    fn disconnect(&mut self) {
        self.alive = false;
    }

    // Consuming method — takes self (moves out)
    fn into_id(self) -> u64 {
        self.id  // self is dropped after this
    }
}

fn main() {
    let mut sub = Subscriber::new(1, "/sports");
    println!("Active: {}", sub.is_active());
    sub.disconnect();

    // Struct update syntax — copy remaining fields from another
    let sub2 = Subscriber { id: 2, ..sub };  // sub is partially moved
}

// Enums can hold data — each variant can have different types
enum Message {
    Quit,                          // no data
    Publish { topic: String, payload: Vec<u8> },  // named fields
    Subscribe(String),            // tuple variant
    Heartbeat(u64, u64),          // two values: id, timestamp
}

impl Message {
    fn describe(&self) -> String {
        match self {
            Message::Quit              => "disconnect".to_string(),
            Message::Publish { topic, .. } =>
                format!("publish to {topic}"),
            Message::Subscribe(t)     => format!("subscribe {t}"),
            Message::Heartbeat(id, ts) =>
                format!("heartbeat from {id} at {ts}"),
        }
    }
}

fn process(msg: Message) {
    match msg {
        Message::Quit => println!("bye"),

        Message::Publish { topic, payload }
            if payload.len() > 1024 =>    // match guard
            println!("large msg on {topic}"),

        Message::Publish { topic, .. } =>
            println!("msg on {topic}"),

        Message::Subscribe(t) | Message::Heartbeat(0, _) =>
            println!("special case"),

        _ => {}  // catch-all (must be exhaustive)
    }

    // if let — match one pattern, ignore the rest
    let m2 = Message::Subscribe("/finance".to_string());
    if let Message::Subscribe(topic) = m2 {
        println!("subscribed to {topic}");
    }

    // Destructuring in let
    let (a, b, c) = (1, 2, 3);
    let Latency(ms) = Latency(9.7);
}

// Option is a built-in enum:
// enum Option<T> { Some(T), None }

fn find_subscriber(id: u64) -> Option<String> {
    if id == 1 { Some("alice".to_string()) }
    else        { None }
}

fn main() {
    // Must handle None — compiler won't let you ignore it
    match find_subscriber(1) {
        Some(name) => println!("Found: {name}"),
        None       => println!("Not found"),
    }

    // Ergonomic helpers:
    let name = find_subscriber(99)
        .unwrap_or("anonymous".to_string());  // default if None

    let upper = find_subscriber(1)
        .map(|s| s.to_uppercase());           // Some("ALICE")

    let len = find_subscriber(1)
        .as_ref()
        .map(|s| s.len())
        .unwrap_or(0);

    // ? operator in Option-returning functions
    fn first_char(id: u64) -> Option<char> {
        let s = find_subscriber(id)?;  // returns None if None
        s.chars().next()
    }
}

// Result is a built-in enum:
// enum Result<T, E> { Ok(T), Err(E) }

use std::fs;
use std::io;

fn read_config(path: &str) -> Result<String, io::Error> {
    // fs::read_to_string returns Result — we must handle it
    let content = fs::read_to_string(path)?;  // ? = early return on Err
    Ok(content)
}

fn main() {
    match read_config("config.toml") {
        Ok(text) => println!("Config: {text}"),
        Err(e)   => eprintln!("Failed to read config: {e}"),
    }

    // unwrap() — panics if Err (use only in tests/prototypes)
    let text = read_config("config.toml").unwrap();

    // expect() — panics with a custom message
    let text = read_config("config.toml")
        .expect("config.toml must exist at startup");

    // Chaining with map_err, and_then
    let len: Result<usize, _> = read_config("f")
        .map(|s| s.len());
}

// In Cargo.toml: thiserror = "1"
use thiserror::Error;

#[derive(Debug, Error)]
enum BrokerError {
    #[error("topic '{0}' not found")]
    TopicNotFound(String),

    #[error("subscriber queue full (capacity: {capacity})")]
    QueueFull { capacity: usize },

    #[error("IO error: {0}")]
    Io(#[from] std::io::Error),  // auto-converts from io::Error via ?
}

fn publish(topic: &str) -> Result<(), BrokerError> {
    if topic.is_empty() {
        return Err(BrokerError::TopicNotFound(topic.to_string()));
    }
    // io::Error auto-converts to BrokerError::Io via ?
    std::fs::read_to_string("log")?;
    Ok(())
}

// In Cargo.toml: anyhow = "1"
use anyhow::{Context, Result, bail, ensure};

// anyhow::Result<T> = Result<T, anyhow::Error>
// anyhow::Error wraps any error type — great for application code
fn start_server(port: u16) -> Result<()> {
    ensure!(port > 1024, "port must be > 1024, got {port}");

    let config = std::fs::read_to_string("config.toml")
        .context("failed to load config.toml")?;  // adds context to error

    if config.is_empty() {
        bail!("config file is empty");  // return Err immediately
    }

    Ok(())
}

// Rule of thumb:
// Library code  → thiserror  (precise error types for callers)
// App/bin code  → anyhow     (ergonomic, good error messages)

trait Encoder {
    // Required method — implementors must provide this
    fn encode(&self) -> Vec<u8>;

    // Default method — implementors can override
    fn size_hint(&self) -> usize { 64 }

    // Associated type — implementors set the concrete type
    type Header;
}

struct JsonEncoder;
struct BinaryEncoder;

impl Encoder for JsonEncoder {
    type Header = String;
    fn encode(&self) -> Vec<u8> { b"{ }".to_vec() }
}

impl Encoder for BinaryEncoder {
    type Header = [u8; 16];
    fn encode(&self) -> Vec<u8> { vec![0x4B, 0x45, 0x45, 0x54] }
    fn size_hint(&self) -> usize { 16 }  // override default
}

// Generic function — T must implement Encoder + Debug
fn send<T: Encoder + std::fmt::Debug>(enc: &T) {
    let bytes = enc.encode();
    println!("Sending {:?}: {} bytes", enc, bytes.len());
}

// Where clause — cleaner for complex bounds
fn send_all<T>(encoders: &[T])
where
    T: Encoder + Clone,
{
    for e in encoders {
        let _ = e.clone();
        e.encode();
    }
}

// impl Trait syntax — shorthand in function signatures
fn make_encoder() -> impl Encoder {
    JsonEncoder  // concrete type hidden from caller
}

// Generic struct
struct Channel<T> {
    buffer: Vec<T>,
    capacity: usize,
}

impl<T> Channel<T> {
    fn new(capacity: usize) -> Self {
        Channel { buffer: Vec::with_capacity(capacity), capacity }
    }
    fn push(&mut self, item: T) -> bool {
        if self.buffer.len() < self.capacity {
            self.buffer.push(item); true
        } else { false }
    }
}

// dyn Trait = heap-allocated trait object (vtable dispatch)
// Use when you need a heterogeneous collection or can't use generics

fn make_encoder(json: bool) -> Box<dyn Encoder> {
    if json { Box::new(JsonEncoder) }
    else     { Box::new(BinaryEncoder) }
}

// Vec of different encoder types
let encoders: Vec<Box<dyn Encoder>> = vec![
    Box::new(JsonEncoder),
    Box::new(BinaryEncoder),
];

// impl Trait in params = static dispatch (monomorphized, faster)
// dyn Trait in params = dynamic dispatch (vtable, flexible)
fn fast(e: &impl Encoder) {}   // zero-cost, compiler generates one fn per type
fn flex(e: &dyn Encoder) {}   // one fn, pointer indirection at runtime

let mut v: Vec<u32> = Vec::new();
let mut v = vec![1, 2, 3];          // macro shorthand
let mut v: Vec<u8> = Vec::with_capacity(1024); // pre-allocate

v.push(4);
v.pop();            // Option<u32>
v.insert(0, 99);    // O(n) — shift elements
v.remove(0);        // O(n) — shift elements
v.swap_remove(0);  // O(1) — swap with last, then pop
v.retain(|x| *x > 1);  // keep elements matching predicate
v.sort();
v.dedup();          // remove consecutive duplicates (sort first)
v.extend([10, 20]);

// Indexing panics on out-of-bounds — use .get() for safe access
let first: Option<&u32> = v.get(0);
let third = v[2];   // panics if len < 3

// Slicing
let chunk: &[u32] = &v[1..3];

// Splitting
let (left, right) = v.split_at(2);

use std::collections::HashMap;

let mut subs: HashMap<String, Vec<u64>> = HashMap::new();

// Insert
subs.insert("/sports".to_string(), vec![1, 2, 3]);

// Get — returns Option<&V>
if let Some(ids) = subs.get("/sports") {
    println!("{} subscribers", ids.len());
}

// entry() API — the most ergonomic insert-or-update
subs.entry("/finance".to_string())
    .or_insert_with(Vec::new)
    .push(42);

// Iterate
for (topic, ids) in &subs {
    println!("{topic}: {} subs", ids.len());
}

// Contains / remove
subs.contains_key("/sports");
subs.remove("/sports");

// Build from iterator
let map: HashMap<_, _> = [("a", 1), ("b", 2)].into_iter().collect();

let subs = vec![1u64, 2, 3, 4, 5, 6];

// Adapters are lazy — nothing runs until consumed
let result: Vec<u64> = subs.iter()
    .filter(|&&id| id % 2 == 0)    // keep evens
    .map(|&id| id * 10)             // multiply
    .take(2)                         // first 2
    .collect();                       // [20, 40]

// fold — reduce to a single value
let total: u64 = subs.iter().fold(0, |acc, x| acc + x);

// any / all — short-circuit
let has_large = subs.iter().any(|&x| x > 100);

// enumerate — pairs with index
for (i, id) in subs.iter().enumerate() {
    println!("[{i}] sub_id={id}");
}

// zip — pair two iterators
let topics = vec!["/sports", "/finance"];
let pairs: Vec<_> = subs.iter().zip(topics.iter()).collect();

// chain — concatenate two iterators
let more = vec![7u64, 8];
let all: Vec<_> = subs.iter().chain(more.iter()).collect();

// flat_map — map then flatten
let nested = vec![vec![1, 2], vec![3, 4]];
let flat: Vec<i32> = nested.into_iter().flatten().collect();

// Implement Iterator for your own type
struct Counter { count: u32 }
impl Iterator for Counter {
    type Item = u32;
    fn next(&mut self) -> Option<u32> {
        self.count += 1;
        if self.count < 6 { Some(self.count) } else { None }
    }
}
// Implementing next() gives you all adapters for free!

fn main() {
    let threshold = 100u64;

    // Closure captures threshold from the enclosing scope
    let is_high_load = |subs: u64| subs > threshold;

    println!("{}", is_high_load(200));  // true

    // Type annotation on closure (usually inferred)
    let add = |a: i32, b: i32| -> i32 { a + b };

    // Multi-line closure
    let describe = |n: u64| {
        let label = if n > threshold { "high" } else { "low" };
        format!("{n} is {label}")
    };

    // move closure — takes ownership of captured variables
    // Required when closure outlives its scope (e.g. threads)
    let prefix = String::from("sub");
    let make_id = move |n: u64| format!("{prefix}_{n}");
    // prefix is moved into closure — no longer usable here
    println!("{}", make_id(42));
}

// Fn     — borrows environment, can call multiple times
// FnMut  — mutably borrows, can call multiple times
// FnOnce — consumes environment, can call only once

fn apply<F: Fn(i32) -> i32>(f: F, x: i32) -> i32 { f(x) }
fn run_once<F: FnOnce()>(f: F) { f() }

// Store a closure in a struct
struct Hook<F: Fn(&str)> {
    callback: F,
}

// Store a closure as a trait object (different types OK)
struct Dispatcher {
    hooks: Vec<Box<dyn Fn(&str)>>,
}

impl Dispatcher {
    fn on(&mut self, f: impl Fn(&str) + 'static) {
        self.hooks.push(Box::new(f));
    }
    fn emit(&self, event: &str) {
        for h in &self.hooks { h(event); }
    }
}

// Lifetime annotations start with ' — they're constraints, not types

// Without lifetime annotation this won't compile —
// compiler can't tell if returned ref lives as long as a or b
fn longest<'a>(x: &'a str, y: &'a str) -> &'a str {
    if x.len() >= y.len() { x } else { y }
    // 'a = "the returned ref lives at least as long as both inputs"
}

// Lifetime in structs — struct can't outlive the reference it holds
struct TopicView<'a> {
    topic: &'a str,   // borrows from somewhere with lifetime 'a
    count: usize,
}

impl<'a> TopicView<'a> {
    fn name(&self) -> &str {
        self.topic  // lifetime elision applies — same as &'a str
    }
}

// 'static — lives for the entire program duration
// String literals are 'static. Use in thread closures, error types.
let s: &'static str = "I live forever";

// Lifetime elision rules (compiler infers in most cases):
// 1. Each ¶m gets its own lifetime
// 2. If only one input lifetime → that's the output lifetime
// 3. If &self or &mut self → self's lifetime is the output lifetime

// Box is the simplest: puts T on the heap, owns it, drops when gone

// 1. Recursive types (size unknown at compile time)
enum List {
    Cons(i32, Box<List>),   // Box gives it a known pointer size
    Nil,
}

// 2. Large value — avoid stack copying
let big = Box::new([0u8; 1_000_000]);  // on heap, not stack

// 3. Trait objects
let enc: Box<dyn Encoder> = Box::new(JsonEncoder);

// Deref coercion: *box gives you the T; auto-deref in method calls
let b = Box::new(5i32);
println!("{}", *b + 1);  // 6

use std::rc::Rc;
use std::sync::Arc;

// Rc — reference counted, single-threaded only
let shared = Rc::new(String::from("shared data"));
let clone1 = Rc::clone(&shared);   // increments ref count
let clone2 = Rc::clone(&shared);   // count = 3
// Data dropped when count reaches 0
println!("refs: {}", Rc::strong_count(&shared));  // 3

// Arc — atomic reference counted, safe across threads
let data = Arc::new(vec![1, 2, 3]);
let data2 = Arc::clone(&data);
std::thread::spawn(move || {
    println!("{:?}", data2);  // safe — atomic ref count
});

// Weak — non-owning reference, breaks cycles
use std::rc::Weak;
let weak: Weak<String> = Rc::downgrade(&shared);
if let Some(s) = weak.upgrade() {   // returns Option — may be gone
    println!("{s}");
}

use std::cell::RefCell;

// RefCell lets you mutate T even through a shared reference
// Borrow rules still apply — just checked at runtime (panic on violation)
let v = RefCell::new(vec![1, 2, 3]);

{
    let mut borrow = v.borrow_mut();  // like &mut — panics if already borrowed
    borrow.push(4);
}  // mutable borrow released here

println!("{:?}", v.borrow());       // immutable borrow

// Common pattern: Rc<RefCell<T>> for shared mutable state (single thread)
let shared_list: Rc<RefCell<Vec<i32>>> =
    Rc::new(RefCell::new(vec![]));

// For multi-threaded: Arc<Mutex<T>> (Hour 9)

// Cell<T> — for Copy types, no runtime borrow check
use std::cell::Cell;
let c = Cell::new(0u32);
c.set(42);
println!("{}", c.get());  // 42

use std::thread;

let handle = thread::spawn(|| {
    println!("running in another thread");
    42  // thread return value
});

let result = handle.join().unwrap();  // wait + get return value
println!("thread returned: {result}");

// move closure — take ownership of captured data
let data = vec![1, 2, 3];
let h = thread::spawn(move || {
    println!("{:?}", data);  // data is moved in
});
h.join().unwrap();

// Scoped threads — borrow without move (std::thread::scope)
let shared = vec![10, 20, 30];
thread::scope(|s| {
    s.spawn(|| println!("{:?}", shared));  // borrow OK — scope ensures join
    s.spawn(|| println!("len={}", shared.len()));
});

use std::sync::{Arc, Mutex, RwLock};

// Arc<Mutex<T>> is the canonical thread-safe shared mutable state
let counter = Arc::new(Mutex::new(0u64));

let handles: Vec<_> = (0..4).map(|_| {
    let c = Arc::clone(&counter);
    std::thread::spawn(move || {
        let mut n = c.lock().unwrap();  // acquire lock — MutexGuard
        *n += 1;
        // lock released when n (MutexGuard) drops at end of scope
    })
}).collect();

for h in handles { h.join().unwrap(); }
println!("counter: {}", *counter.lock().unwrap());  // 4

// RwLock — many readers OR one writer
let registry = Arc::new(RwLock::new(Vec::<String>::new()));

// Multiple threads can read simultaneously
let r = registry.read().unwrap();    // RwLockReadGuard
println!("{} entries", r.len());
drop(r);

// Only one can write
let mut w = registry.write().unwrap();  // RwLockWriteGuard
w.push("alice".to_string());

use std::sync::mpsc;  // multi-producer, single-consumer

// Unbounded channel
let (tx, rx) = mpsc::channel();

// Clone sender for multiple producers
let tx2 = tx.clone();

std::thread::spawn(move || { tx.send("msg from thread 1").unwrap(); });
std::thread::spawn(move || { tx2.send("msg from thread 2").unwrap(); });

// Receive blocks until a message arrives
for msg in rx {   // iterates until all senders dropped
    println!("{msg}");
}

// Bounded (backpressure) channel
let (tx, rx) = mpsc::sync_channel(1024);  // blocks sender when full

// For async code, use tokio::sync::mpsc (Hour 10)
// For lock-free queues, consider crossbeam::channel from crates.io

[dependencies]
tokio = { version = "1", features = ["full"] }
# "full" enables: rt-multi-thread, macros, net, io, sync, time, fs

use tokio::time::{sleep, Duration};
use tokio::net::TcpListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

// #[tokio::main] sets up the multi-thread runtime
#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("0.0.0.0:7878").await.unwrap();
    println!("Listening...");

    loop {
        let (socket, addr) = listener.accept().await.unwrap();
        // spawn a new task per connection — costs ~a few KB
        tokio::spawn(async move {
            handle_conn(socket).await;
        });
    }
}

async fn handle_conn(mut socket: tokio::net::TcpStream) {
    let mut buf = [0u8; 1024];
    loop {
        let n = match socket.read(&mut buf).await {
            Ok(0) => break,  // connection closed
            Ok(n) => n,
            Err(_) => break,
        };
        socket.write_all(&buf[..n]).await.unwrap();  // echo
    }
}

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    // Tokio's async mpsc channel
    let (tx, mut rx) = mpsc::channel::<String>(1024);

    // Spawn a task that produces messages
    tokio::spawn(async move {
        for i in 0..5 {
            tx.send(format!("message {i}")).await.unwrap();
            sleep(Duration::from_millis(100)).await;
        }
    });

    // Receive messages
    while let Some(msg) = rx.recv().await {
        println!("{msg}");
    }

    // tokio::select! — race multiple async operations
    let mut shutdown = tokio::signal::ctrl_c();
    tokio::select! {
        _ = sleep(Duration::from_secs(30)) => println!("timeout"),
        _ = tokio::signal::ctrl_c() => println!("shutdown"),
        msg = rx.recv()                => println!("{msg:?}"),
    }

    // Run two futures concurrently (not parallel — same thread)
    let (a, b) = tokio::join!(
        fetch_data("http://api1"),
        fetch_data("http://api2"),
    );
}

// async fn desugars to a state machine implementing Future:
// trait Future { type Output; fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll; }

// poll() returns:
//   Poll::Ready(value)  — computation done
//   Poll::Pending       — not ready, Waker registered to re-poll later

// .await desugar (simplified):
// loop { match future.poll(cx) { Ready(v) => break v, Pending => yield } }

// The runtime's reactor watches file descriptors via epoll/kqueue/IOCP
// When a socket becomes readable, it wakes the task — calls poll again

// Key rules for async code:
// 1. Don't block in async — use tokio::task::spawn_blocking for CPU work
// 2. Don't hold Mutex across .await — deadlock risk
// 3. Prefer tokio::sync types over std::sync in async context
// 4. Clone Arc to share state across tasks, not references

async fn cpu_intensive() -> u64 {
    tokio::task::spawn_blocking(|| {
        // This runs on a dedicated thread pool, won't block async tasks
        (0u64..1_000_000).sum()
    }).await.unwrap()
}

// macro_rules! matches patterns on token trees

macro_rules! keet_log {
    // Pattern: ($level:expr, $msg:expr) means two expressions
    ($level:expr, $($arg:tt)*) => {
        println!("[{}] {}", $level, format!($($arg)*))
    };
}

// Multiple patterns — like match arms
macro_rules! assert_close {
    ($a:expr, $b:expr) => {
        assert_close!($a, $b, 1e-6)
    };
    ($a:expr, $b:expr, $tol:expr) => {
        let diff = ($a - $b).abs();
        assert!(diff < $tol, "{} != {} (diff={})", $a, $b, diff);
    };
}

// Variadic: $($x:expr),* repeats zero or more comma-separated exprs
macro_rules! sum {
    ($($x:expr),*) => {{
        let mut total = 0i64;
        $(total += $x;)*
        total
    }};
}

fn main() {
    keet_log!("INFO", "subscriber {} connected", 42);
    assert_close!(0.1 + 0.2, 0.3, 1e-10);
    println!("{}", sum!(1, 2, 3, 4));  // 10
}

// proc-macros live in their own crate (proc-macro = true in Cargo.toml)
// They receive a TokenStream and return a TokenStream

// ── Using derive macros (most common) ─────────────────────────────────
use serde::{Serialize, Deserialize};

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct Frame {
    pub magic:   u32,
    pub version: u8,
    pub topic:   String,
    pub payload: Vec<u8>,
}

// ── Attribute macro example (from tokio) ──────────────────────────────
#[tokio::main]           // expands to runtime setup + fn main wrapper
async fn main() {}

#[tokio::test]           // async test runner
async fn test_connection() {
    assert!(true);
}

// ── Function-like proc macro ───────────────────────────────────────────
// sql! macro (e.g. sqlx) — validates SQL at compile time
// let row = sqlx::query!("SELECT id FROM users WHERE id = $1", user_id);

// ── Writing a derive macro skeleton ───────────────────────────────────
// In a proc-macro crate (proc-macro = true):
//
// use proc_macro::TokenStream;
// use quote::quote;
// use syn::{parse_macro_input, DeriveInput};
//
// #[proc_macro_derive(MyTrait)]
// pub fn my_trait_derive(input: TokenStream) -> TokenStream {
//     let ast = parse_macro_input!(input as DeriveInput);
//     let name = &ast.ident;
//     quote! {
//         impl MyTrait for #name {
//             fn hello(&self) { println!("Hello from {}!", stringify!(#name)); }
//         }
//     }.into()
// }

[package]
name = "mini-broker"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio     = { version = "1", features = ["full"] }
dashmap   = "6"
bytes     = "1"
thiserror = "1"

use bytes::Bytes;
use dashmap::DashMap;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::net::{TcpListener, TcpStream};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};

// Each subscriber gets a bounded channel
type Tx = mpsc::Sender<Bytes>;

// Topic registry: topic name → list of subscriber senders
// DashMap = concurrent HashMap, no global lock
type Registry = Arc<DashMap<String, Vec<Tx>>>;

#[tokio::main]
async fn main() {
    let registry: Registry = Arc::new(DashMap::new());
    let listener = TcpListener::bind("127.0.0.1:7878").await.unwrap();
    println!("Mini broker on :7878");

    loop {
        let (socket, addr) = listener.accept().await.unwrap();
        println!("connect: {addr}");
        let reg = Arc::clone(®istry);
        tokio::spawn(handle(socket, reg));
    }
}

async fn handle(stream: TcpStream, reg: Registry) {
    let (mut tx_half, mut rx_half) = stream.into_split();
    let (sub_tx, mut sub_rx) = mpsc::channel::<Bytes>(256);

    // Task 1: drain incoming commands from client
    let reg2 = Arc::clone(®);
    tokio::spawn(async move {
        let reader = BufReader::new(rx_half);
        let mut lines = reader.lines();
        while let Ok(Some(line)) = lines.next_line().await {
            let parts: Vec<&str> = line.splitn(3, ' ').collect();
            match parts.as_slice() {
                ["SUB", topic] => {
                    reg2.entry(topic.to_string())
                        .or_default()
                        .push(sub_tx.clone());
                    println!("subscribed to {topic}");
                }
                ["PUB", topic, payload] => {
                    let msg = Bytes::copy_from_slice(
                        format!("{topic}: {payload}\n").as_bytes()
                    );
                    if let Some(mut subs) = reg2.get_mut(*topic) {
                        // Fan-out: send to all subscribers, drop dead ones
                        subs.retain(|tx| tx.try_send(msg.clone()).is_ok());
                    }
                }
                _ => {}
            }
        }
    });

    // Task 2: write outgoing messages to client
    while let Some(msg) = sub_rx.recv().await {
        if tx_half.write_all(&msg).await.is_err() { break; }
    }
}

// Test it:
// cargo run
// nc 127.0.0.1 7878   →  type: SUB /sports
// nc 127.0.0.1 7878   →  type: PUB /sports hello

// Unit tests live in the same file, in a #[cfg(test)] module
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_add() {
        assert_eq!(add(2, 2), 4);
        assert_ne!(add(2, 2), 5);
    }

    #[test]
    #[should_panic(expected = "overflow")]
    fn test_panic() { panic!("overflow!"); }

    // Async test — needs tokio::test
    #[tokio::test]
    async fn test_broker_subscribe() {
        let reg: Registry = Arc::new(DashMap::new());
        let (tx, mut rx) = mpsc::channel(4);
        reg.entry("/test".into()).or_default().push(tx);

        let msg = Bytes::from("hello");
        reg.get_mut("/test").unwrap()
           .retain(|tx| tx.try_send(msg.clone()).is_ok());

        assert_eq!(rx.recv().await, Some(msg));
    }
}
// cargo test              — run all tests
// cargo test test_broker  — run matching tests
// cargo test -- --nocapture — show println output

// unsafe { } block tells the compiler: "I've verified this manually"
// Enables five things: raw pointers, extern C, mutable statics,
// unsafe trait impls, calling unsafe functions

let v = vec![1, 2, 3];
let ptr = v.as_ptr();  // raw pointer *const i32

unsafe {
    // Dereference raw pointer — you must ensure it's valid
    println!("{}", *ptr);
    // get_unchecked skips bounds check — you guarantee i < len
    println!("{}", *v.get_unchecked(0));
}

// FFI — call C from Rust
extern "C" {
    fn strlen(s: *const i8) -> usize;
}
unsafe { strlen(std::ptr::null()); }

// Use unsafe to build safe abstractions
// Unsafe code is not bad — it's a tool. The goal is a safe public API
// that uses unsafe internally where the performance demands it.
// Vec, HashMap, Arc — all use unsafe inside.