🦀 Rust 复杂架构设计

/// 连接状态机：编译器保证你不会在 Disconnected 状态下发送数据
enum ConnectionState {
    Disconnected,
    Connecting(TcpStream, Instant),  // stream + 超时时间
    Connected {
        stream: TcpStream,
        last_heartbeat: Instant,
        seq: u64,
    },
    Reconnecting {
        attempts: u8,
        backoff: Duration,
    },
}

struct Connection {
    state: ConnectionState,
}

impl Connection {
    fn send(&mut self, data: &[u8]) -> Result<(), ConnError> {
        match &mut self.state {
            ConnectionState::Connected { stream, seq, .. } => {
                *seq += 1;
                stream.write_all(data)?;
                Ok(())
            }
            // 编译器强制你处理每个状态
            ConnectionState::Disconnected => Err(ConnError::NotConnected),
            ConnectionState::Connecting(_, _) => Err(ConnError::InProgress),
            ConnectionState::Reconnecting { attempts, .. } => {
                Err(ConnError::Reconnecting { attempts: *attempts })
            }
        }
    }
}

// 不要用裸类型——编译器无法区分 UserId 和 OrderId
// struct User { id: u64 }     // ❌ 容易和 Order.id 混淆

// 用 newtype 让编译器帮你守门
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
struct UserId(u64);

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
struct OrderId(u64);

fn get_user(id: UserId) -> User { ... }
fn get_order(id: OrderId) -> Order { ... }

// 现在这样写会编译错误：
// get_user(OrderId(42))  // ❌ type mismatch!
// 只有 get_user(UserId(42)) 才能编译通过

use std::marker::PhantomData;

// 状态编码在类型参数里
struct Unconfigured;
struct Configured;
struct Running;

struct Server<State = Unconfigured> {
    config: Option<ServerConfig>,
    listener: Option<TcpListener>,
    _state: PhantomData<State>,
}

impl Server<Unconfigured> {
    fn new() -> Self { ... }
    
    // configure 消费 Unconfigured，返回 Configured
    fn configure(self, cfg: ServerConfig) -> Server<Configured> {
        Server {
            config: Some(cfg),
            listener: None,
            _state: PhantomData,
        }
    }
}

impl Server<Configured> {
    // start 只在 Configured 状态可用
    fn start(self) -> Result<Server<Running>, Error> {
        let config = self.config.unwrap(); // 编译器保证 Some
        let listener = TcpListener::bind(config.addr)?;
        Ok(Server {
            config: Some(config),
            listener: Some(listener),
            _state: PhantomData,
        })
    }
}

impl Server<Running> {
    fn accept(&self) -> Result<TcpStream, Error> { ... }
}

// 使用：
let server = Server::new()
    .configure(config)       // 必须先 configure
    .start()?;               // 才能 start
// server.accept()           // 才能 accept

// 你无法在 Unconfigured 状态调用 start()——编译错误！
// 你无法跳过 configure 直接 start——编译错误！

use tokio::sync::{mpsc, oneshot};

// Actor 的消息协议——用 enum 强类型化
enum DbActorMsg {
    Get { key: String, reply: oneshot::Sender<Option<String>> },
    Set { key: String, value: String },
    Delete { key: String },
    Flush { reply: oneshot::Sender<()> },
}

struct DbActor {
    rx: mpsc::Receiver<DbActorMsg>,
    data: HashMap<String, String>,     // Actor 独占，无需锁
    pending_writes: Vec<String>,
}

impl DbActor {
    async fn run(&mut self) {
        while let Some(msg) = self.rx.recv().await {
            match msg {
                DbActorMsg::Get { key, reply } => {
                    let _ = reply.send(self.data.get(&key).cloned());
                }
                DbActorMsg::Set { key, value } => {
                    self.data.insert(key.clone(), value);
                    self.pending_writes.push(key);
                }
                DbActorMsg::Delete { key } => {
                    self.data.remove(&key);
                }
                DbActorMsg::Flush { reply } => {
                    self.flush_to_disk().await;
                    self.pending_writes.clear();
                    let _ = reply.send(());
                }
            }
        }
    }
}

// Actor 的 handle——对外暴露的类型安全接口
struct DbHandle {
    tx: mpsc::Sender<DbActorMsg>,
}

impl DbHandle {
    pub async fn get(&self, key: String) -> Option<String> {
        let (tx, rx) = oneshot::channel();
        self.tx.send(DbActorMsg::Get { key, reply: tx }).await.ok()?;
        rx.await.ok()?
    }
    
    pub async fn set(&self, key: String, value: String) {
        self.tx.send(DbActorMsg::Set { key, value }).await.ok();
    }
}

use std::any::{Any, TypeId};
use std::collections::HashMap;

type EntityId = u64;

struct World {
    entities: Vec<EntityId>,
    components: HashMap<TypeId, Box<dyn Any>>,  // TypeId → Vec<Component>
    next_id: EntityId,
}

impl World {
    fn spawn(&mut self) -> EntityId {
        let id = self.next_id;
        self.next_id += 1;
        self.entities.push(id);
        id
    }
    
    fn add_component<T: 'static>(&mut self, entity: EntityId, comp: T) {
        let type_id = TypeId::of::<T>();
        let storage = self.components
            .entry(type_id)
            .or_insert_with(|| Box::new(Vec::<(EntityId, T)>::new()));
        let vec = storage.downcast_mut::<Vec<(EntityId, T)>>().unwrap();
        vec.push((entity, comp));
    }
    
    fn query<T: 'static>(&self) -> Vec<(EntityId, &T)> {
        let type_id = TypeId::of::<T>();
        self.components
            .get(&type_id)
            .and_then(|s| s.downcast_ref::<Vec<(EntityId, T)>>())
            .map(|v| v.iter().map(|(id, c)| (*id, c)).collect())
            .unwrap_or_default()
    }
    
    fn query2<A: 'static, B: 'static>(&self) -> Vec<(EntityId, &A, &B)> {
        // Join two component storages on EntityId
        let a_map: HashMap<_, _> = self.query::<A>().into_iter().collect();
        let b_map: HashMap<_, _> = self.query::<B>().into_iter().collect();
        a_map.keys()
            .filter_map(|&id| {
                Some((id, a_map.get(&id)?, b_map.get(&id)?))
            })
            .collect()
    }
}

// 使用
let mut world = World::new();
let player = world.spawn();
world.add_component(player, Position(0.0, 0.0));
world.add_component(player, Velocity(1.0, 0.0));
world.add_component(player, Health(100));

// Movement system: 只关心有 Position + Velocity 的 entity
for (id, pos, vel) in world.query2::<Position, Velocity>() {
    println!("Entity {} at ({}, {}) moving ({}, {})", 
             id, pos.0, pos.1, vel.0, vel.1);
}

use thiserror::Error;

#[derive(Error, Debug)]
enum AppError {
    #[error("Database error: {0}")]
    Database(#[from] DbError),
    
    #[error("User {user_id} not found")]
    UserNotFound { user_id: UserId },
    
    #[error("Rate limited: {requests} requests in {window:?}")]
    RateLimited { requests: u32, window: Duration },
    
    #[error("Invalid configuration: {message}")]
    Config { message: String },
    
    #[error(transparent)]
    Internal(#[from] anyhow::Error),  // 兜底
}

// 每个错误变体携带足够的上下文用于调试
// #[from] 自动实现 From，? 运算符自动转换

// 核心 crate 定义插件接口
pub trait Plugin: Send + Sync {
    fn name(&self) -> &str;
    fn on_load(&self, ctx: &PluginContext) -> Result<(), PluginError>;
    fn on_request(&self, req: &mut Request) -> Result<(), PluginError>;
    fn on_response(&self, resp: &mut Response) -> Result<(), PluginError>;
    fn on_unload(&self) {}
}

pub struct PluginContext {
    pub config: &Config,
    pub metrics: &MetricsSink,
}

// 插件管理器
struct PluginManager {
    plugins: Vec<Box<dyn Plugin>>,   // 运行期多态
}

impl PluginManager {
    fn register(&mut self, plugin: Box<dyn Plugin>) {
        self.plugins.push(plugin);
    }
    
    fn on_request(&self, req: &mut Request) -> Result<(), PluginError> {
        for plugin in &self.plugins {
            plugin.on_request(req)?;  // 责任链模式
        }
        Ok(())
    }
}

// 具体插件——可以独立 crate
struct AuthPlugin { secret: String }
impl Plugin for AuthPlugin {
    fn name(&self) -> &str { "auth" }
    fn on_request(&self, req: &mut Request) -> Result<(), PluginError> {
        let token = req.header("Authorization")
            .ok_or(PluginError::Unauthorized)?;
        self.verify(token)?;
        Ok(())
    }
    // ...
}

// === 领域类型 ===
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
struct JobId(Uuid);

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
struct WorkerId(Uuid);

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
struct TaskId(Uuid);

// === 状态机 ===
enum JobState {
    Pending,
    Scheduled { worker: WorkerId, task: TaskId },
    Running { worker: WorkerId, started_at: Instant },
    Completed { result: JobResult },
    Failed { error: String, retries: u8 },
    Cancelled,
}

enum TaskState {
    Queued,
    Dispatched { worker: WorkerId },
    Running { heartbeat: Instant },
    Finished { exit_code: i32 },
    Crashed { signal: i32 },
    Lost,  // worker 失联
}

// === 事件溯源 ===
enum JobEvent {
    Created { spec: JobSpec },
    Scheduled { worker: WorkerId, task: TaskId },
    Started { worker: WorkerId, at: Timestamp },
    Heartbeat { worker: WorkerId, progress: f32 },
    Completed { result: JobResult },
    Failed { error: String },
    RetryScheduled { attempt: u8, after: Duration },
    Cancelled { reason: String },
}

// === 插件接口 ===
trait SchedulerPlugin: Send + Sync {
    fn on_job_created(&self, job: &Job) -> Option<ScheduleHint>;
    fn on_task_finished(&self, result: &TaskResult) -> Vec<Action>;
    fn on_worker_lost(&self, worker: WorkerId, tasks: &[TaskId]) -> Vec<Action>;
}

enum Action {
    Reschedule { task: TaskId, priority: Priority },
    Notify { channel: String, message: String },
    ScaleUp { count: u32 },
}

enum ScheduleHint {
    Prefer { worker: WorkerId },
    Avoid { worker: WorkerId },
    Require { resource: String, amount: u64 },
}

enum WorkerMsg {
    Dispatch { task: Task, reply: oneshot::Sender<Result<(), DispatchError>> },
    Cancel { task_id: TaskId },
    HeartbeatCheck,
    Shutdown { reply: oneshot::Sender<()> },
}

struct WorkerActor {
    id: WorkerId,
    capacity: u32,
    running: HashMap<TaskId, JoinHandle<()>>,
    rx: mpsc::Receiver<WorkerMsg>,
    heartbeat_tx: watch::Sender<WorkerStatus>,
}

impl WorkerActor {
    async fn run(&mut self) {
        let mut heartbeat_interval = tokio::time::interval(Duration::from_secs(5));
        
        loop {
            tokio::select! {
                Some(msg) = self.rx.recv() => {
                    match msg {
                        WorkerMsg::Dispatch { task, reply } => {
                            let handle = self.spawn_task(task).await;
                            let _ = reply.send(handle);
                        }
                        WorkerMsg::Cancel { task_id } => {
                            if let Some(h) = self.running.remove(&task_id) {
                                h.abort();
                            }
                        }
                        WorkerMsg::Shutdown { reply } => {
                            for (_, h) in self.running.drain() { h.abort(); }
                            let _ = reply.send(());
                            return;
                        }
                        WorkerMsg::HeartbeatCheck => {
                            self.cleanup_finished();
                        }
                    }
                }
                _ = heartbeat_interval.tick() => {
                    let status = WorkerStatus {
                        id: self.id,
                        running: self.running.len() as u32,
                        capacity: self.capacity,
                    };
                    let _ = self.heartbeat_tx.send(status);
                }
            }
        }
    }
}