Simply Patrick Hopes can always go up, tears can only come down.

// 私有模組 - 隱藏內部實作細節
mod job;
mod storage;
mod worker;

// 公開重匯出 - 只暴露需要的類型
pub use job::{Job, JobHandler, JobStatus, Priority};
pub use storage::{Storage, StorageError};
pub use worker::WorkerPool;

pub mod job;  // 暴露整個模組，使用者可存取所有公開項目

pub trait JobHandler: Send + Sync {
    fn handle(&self, payload: &[u8]) -> Result<(), String>;
}

struct EchoHandler;

impl JobHandler for EchoHandler {
    fn handle(&self, payload: &[u8]) -> Result<(), String> {
        let message = String::from_utf8_lossy(payload);
        println!("處理任務：{}", message);
        Ok(())
    }
}

pub struct Job {
    pub payload: Vec<u8>,
    // ...
}

pub struct JobPayload(Vec<u8>);

impl JobPayload {
    pub fn new(bytes: Vec<u8>) -> Self {
        Self(bytes)
    }
    
    pub fn as_bytes(&self) -> &[u8] {
        &self.0
    }
}

impl From<String> for JobPayload {
    fn from(s: String) -> Self {
        Self(s.into_bytes())
    }
}

impl fmt::Display for Priority {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Priority::Low => write!(f, "low"),
            Priority::Normal => write!(f, "normal"),
            Priority::High => write!(f, "high"),
            Priority::Critical => write!(f, "critical"),
        }
    }
}

use strum_macros::Display;

#[derive(Display)]
#[strum(serialize_all = "lowercase")]
pub enum Priority {
    Low,
    Normal,
    High,
    Critical,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // ...
}

for worker_id in 0..self.num_workers {
    let handle = tokio::spawn(async move {
        worker_loop(worker_id, storage, handler, poll_interval).await;
    });
    handles.push(handle);
}

sleep(poll_interval).await;  // 輪詢間隔

pub struct Storage {
    conn: Arc<Mutex<Connection>>,  // ✅ std::sync::Mutex
}

pub fn get_next_pending(&self) -> Result<Option<Job>, StorageError> {
    let conn = self.conn.lock().unwrap();
    // 執行快速的同步資料庫操作
    // 沒有在持鎖期間 await
    Ok(job)
}

// ❌ 錯誤：在持有 std::sync::Mutex 時 await
let guard = std_mutex.lock().unwrap();
async_operation().await;  // 會阻塞整個執行緒！

// ✅ 正確：使用 tokio::sync::Mutex
let guard = tokio_mutex.lock().await;
async_operation().await;  // OK，會讓出執行權

// ❌ 有競爭條件的程式碼
pub fn get_next_pending(&self) -> Result<Option<Job>, StorageError> {
    // 步驟 1：讀取待處理任務
    let job = SELECT ... WHERE status = Pending LIMIT 1;
    // ⚠️ 問題：Worker 2 可能在這裡也讀取到相同任務！
    Ok(job)
}

// Worker 處理邏輯
let job = storage.get_next_pending()?;  // 取得任務
job.mark_running();                     // 標記為執行中
storage.update(&job)?;                  // 更新資料庫

pub fn get_next_pending(&self) -> Result<Option<Job>, StorageError> {
    let conn = self.conn.lock().unwrap();
    
    // ✅ 原子性的 UPDATE + RETURNING（一個 SQL 語句完成）
    let mut stmt = conn.prepare(
        "UPDATE jobs
         SET status = ?1, updated_at = ?2
         WHERE id = (
             SELECT id FROM jobs
             WHERE status = ?3
             ORDER BY priority DESC, created_at ASC
             LIMIT 1
         )
         RETURNING id, payload, priority, status, retry_count, max_retries,
                   created_at, updated_at, error_message",
    )?;

    let now = Utc::now().to_rfc3339();
    let job = stmt
        .query_row(
            params![JobStatus::Running as i32, now, JobStatus::Pending as i32],
            |row| Ok(self.row_to_job(row)?),
        )
        .optional()?;

    Ok(job)
}

// ✅ 修正後
match storage.get_next_pending() {
    Ok(Some(mut job)) => {
        // 任務已經是 Running 狀態，直接處理
        match handler.handle(&job.payload) {
            Ok(()) => job.mark_completed(),
            Err(e) => job.mark_failed(e),
        }
        storage.update(&job)?;
    }
    Ok(None) => {
        sleep(poll_interval).await;  // 無任務，等待
    }
    Err(e) => { /* 處理錯誤 */ }
}

// ✅ 使用者導向的輸出
println!("Job ID: {}", job.id);
println!("Status: {}", job.status);

// ✅ 操作性日誌、除錯資訊
tracing::info!(job_id = %job.id, "Processing job");
tracing::error!("Database error: {}", e);

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 設定日誌格式與過濾級別
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::from_default_env()
                .add_directive(tracing::Level::INFO.into()),
        )
        .init();
    
    // ...
}

# 預設 INFO 級別
./worker

# 設定為 DEBUG 級別
RUST_LOG=debug ./worker

CREATE TABLE jobs (
    id TEXT PRIMARY KEY,
    payload BLOB NOT NULL,
    priority INTEGER NOT NULL,
    status INTEGER NOT NULL,
    retry_count INTEGER NOT NULL,
    max_retries INTEGER NOT NULL,
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    error_message TEXT
);

CREATE INDEX idx_status_priority
ON jobs(status, priority DESC, created_at ASC);

pub fn mark_failed(&mut self, error: String) {
    self.status = if self.can_retry() {
        self.retry_count += 1;
        JobStatus::Pending  // 重新排程
    } else {
        JobStatus::DeadLetter  // 移至死信佇列
    };
    self.error_message = Some(error);
    self.updated_at = Utc::now();
}

# 提交任務
producer submit -p "Hello, World!" -r high -m 3

# 查詢狀態
producer status --job-id <uuid>

# 統計資料
producer stats

#[derive(Parser)]
#[command(name = "producer")]
#[command(about = "Job queue producer - submit and manage jobs")]
struct Cli {
    #[arg(short, long, default_value = "jobs.db")]
    database: String,

    #[command(subcommand)]
    command: Commands,
}

#[derive(Subcommand)]
enum Commands {
    Submit {
        #[arg(short, long)]
        payload: String,
        
        #[arg(short = 'r', long, default_value = "normal")]
        priority: String,
        
        #[arg(short, long, default_value = "3")]
        max_retries: u32,
    },
    Status { /* ... */ },
    Stats,
}

# 啟動 4 個 workers
worker -w 4

# 使用不同的資料庫
worker --database /path/to/jobs.db --workers 8

loop {
    match storage.get_next_pending() {
        Ok(Some(job)) => { /* 處理 */ }
        Ok(None) => sleep(poll_interval).await,  // 等待 1 秒
        Err(e) => { /* 錯誤處理 */ }
    }
}

// 生產者插入任務後通知
notifier.notify_waiters();

// Worker 等待通知
notifier.notified().await;

if job.retry_count > 0 {
    let backoff = Duration::from_secs(2_u64.pow(job.retry_count.min(5)));
    sleep(backoff).await;
}