Implement message type inference and enhance dashboard statistics

compass/todolist.md

@@ -14,6 +14,8 @@
 | 04-30 | Token 认证系统 | 启动生成 UUID → 登录遮罩 → localStorage → API/WS 携带 |
 | 04-30 | CI 格式修复 × 3 | import order / 行宽 / 链式调用换行 |
 | 05-01 | 集成测试补全 | 39 个新测试: broadcaster(11) + protocol(21) + auth(7) |
+| 05-01 | Phase B: API 补全 | list/stats/PATCH/DELETE 全部完成；status Canceled 拆分修复 |
+| 05-01 | Phase C1-C2: 消息流 + Dispatcher | 图片/文件类型扩展、sender 补全、Canceled 跳过、Web 前端同步 |
 
 ---
 
@@ -31,21 +33,21 @@
 | A4 | 调度器生命周期管理 | 🔴 | main.rs + dispatcher.rs | start / graceful shutdown |
 | A5 | 集成测试 | 🔴 | tests/dispatcher.rs | 模拟发送成功/失败/超时/最大重试 |
 
-### 🟡 Phase B: API 补全
+### 🟡 Phase B: API 补全 ✅ 已完成
 
 | # | 任务 | 优先级 | 说明 |
 |:--|------|:---:|------|
-| B1 | `GET /api/messages` 列表分页 | 🟡 | 支持 `?limit=&offset=&status=` 筛选 |
-| B2 | `PATCH /api/messages/{id}` 取消/重发 | 🟡 | `{"status": "Failed"}` 取消待发送消息 |
-| B3 | `GET /api/stats` 统计概览 | 🟡 | 今日消息数/各状态计数/平台分布 |
-| B4 | `DELETE /api/messages/{id}` | 🟢 | 删除消息记录 |
+| B1 | `GET /api/messages` 列表分页 | ✅ | 支持 `?limit=&offset=&status=` 筛选 |
+| B2 | `PATCH /api/messages/{id}` 取消/重发 | ✅ | 支持 Pending/Sending/Sent/Failed/Canceled |
+| B3 | `GET /api/stats` 统计概览 | ✅ | 5种状态计数 + total |
+| B4 | `DELETE /api/messages/{id}` | ✅ | 删除消息记录 |
 
-### 🟡 Phase C: 消息流完善
+### 🟡 Phase C: 消息流完善 🔴 进行中
 
 | # | 任务 | 优先级 | 说明 |
 |:--|------|:---:|------|
-| C1 | OneBot 图片/文件消息类型扩展 | 🟡 | 当前只处理 Text 类型 → 扩展 Image/File MessageType |
-| C2 | 入站消息 sender 信息补全 | 🟡 | group/private conversation_name 自动生成 |
+| C1 | OneBot 图片/文件消息类型扩展 | ✅ | 根据消息段自动识别 Image/File/Text 类型 |
+| C2 | 入站消息 sender 信息补全 | ✅ | conversation_name 从 nickname/群号自动生成 |
 | C3 | 多平台支持框架就绪 | 🟢 | adapter 注册机制验证 (QQ 已有，可加 mock 测试微信) |
 
 ### 🟢 Phase D: 运维与部署
@@ -102,7 +104,10 @@
 └── [ ] GET /api/stats
 
 ⏳ Phase C-E: 完善 + 部署 → 目标 2026-05-05
-├── [ ] 图片/文件类型
+├── [✅] 图片/文件类型识别
+├── [✅] sender 信息补全
+├── [✅] Dispatcher Canceled 跳过
+├── [✅] Web 前端状态同步
 ├── [ ] Docker 部署
 └── [ ] 架构优化
 ```

src/adapters/onebot/protocol.rs

@@ -136,6 +136,20 @@ impl MessageSegment {
             .collect::<Vec<_>>()
             .join("")
     }
+
+    /// 从消息段数组推断消息类型
+    pub fn segments_message_type(segments: &[MessageSegment]) -> &'static str {
+        for seg in segments {
+            match seg.seg_type.as_str() {
+                "image" => return "Image",
+                "video" => return "File",
+                "record" => return "File",
+                "file" => return "File",
+                _ => {}
+            }
+        }
+        "Text"
+    }
 }
 
 // ========== 通知事件 ==========

src/adapters/onebot/ws.rs

@@ -132,17 +132,38 @@ fn handle_onebot_event(broadcaster: &MessageBroadcaster, event: OneBotEvent) {
 
 fn handle_message_event(broadcaster: &MessageBroadcaster, event: MessageEvent) {
     let platform = "qq".to_string();
-    let conversation = if let Some(gid) = event.group_id {
-        format!("group_{}", gid)
+    let (conversation, conversation_name) = if let Some(gid) = event.group_id {
+        (
+            format!("group_{}", gid),
+            event
+                .sender
+                .as_ref()
+                .map(|s| s.card.clone().unwrap_or_else(|| s.nickname.clone()))
+                .or_else(|| Some(format!("群聊 {}", gid))),
+        )
     } else {
-        format!("private_{}", event.user_id)
+        (
+            format!("private_{}", event.user_id),
+            event
+                .sender
+                .as_ref()
+                .map(|s| s.nickname.clone())
+                .or_else(|| Some(format!("用户 {}", event.user_id))),
+        )
     };
 
     let raw_text = super::protocol::MessageSegment::segments_to_text(&event.message);
+    let msg_type_str = super::protocol::MessageSegment::segments_message_type(&event.message);
+
+    let message_type = match msg_type_str {
+        "Image" => MessageType::Image,
+        "File" => MessageType::File,
+        _ => MessageType::Text,
+    };
 
     let message = Message {
         id: uuid::Uuid::new_v4().to_string(),
-        message_type: MessageType::Text,
+        message_type,
         content: raw_text.clone(),
         recipient: conversation.clone(),
         status: MessageStatus::Pending,
@@ -168,9 +189,9 @@ fn handle_message_event(broadcaster: &MessageBroadcaster, event: MessageEvent) {
             id: message.id.clone(),
             platform: platform.clone(),
             conversation: conversation.clone(),
-            conversation_name: None,
+            conversation_name,
             content: raw_text,
-            message_type: "Text".into(),
+            message_type: msg_type_str.into(),
             sender: event.sender.map(|s| crate::core::broadcaster::SenderInfo {
                 id: s.user_id.to_string(),
                 name: s.card.unwrap_or(s.nickname),

src/core/dispatcher.rs

@@ -96,6 +96,26 @@ impl MessageDispatcher {
                             }
                         };
 
+                        // Atomically claim: only proceeds if Pending/Sending, not Canceled
+                        match repo.try_claim_message(&message.id) {
+                            Ok(true) => {}
+                            Ok(false) => {
+                                tracing::info!(
+                                    message_id = %message.id,
+                                    "Message already claimed or canceled, skipping"
+                                );
+                                return;
+                            }
+                            Err(e) => {
+                                tracing::error!(
+                                    error = %e,
+                                    message_id = %message.id,
+                                    "Failed to claim message"
+                                );
+                                return;
+                            }
+                        }
+
                         let mut sent = false;
 
                         for attempt in 0..=max_retries {

src/core/message.rs

@@ -219,6 +219,20 @@ impl MessageRepository {
         Ok(())
     }
 
+    /// Atomically claim a message for sending: only succeeds if status is Pending or Sending.
+    /// Returns true if claimed, false if the message was already canceled/failed/sent.
+    pub fn try_claim_message(&self, id: &str) -> Result<bool> {
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_secs() as i64;
+        let affected = self.conn.execute(
+            "UPDATE messages SET status = 'Sending', updated_at = ?1 WHERE id = ?2 AND status IN ('Pending', 'Sending')",
+            rusqlite::params![now, id],
+        )?;
+        Ok(affected > 0)
+    }
+
     pub fn increment_retry(&self, id: &str) -> Result<()> {
         let now = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)

src/web/templates/app.js

@@ -164,17 +164,17 @@ function updateWSStatus(connected) {
 // ========== Dashboard ==========
 
 function refreshDashboard() {
-    try {
-        j('/api/health').then(function () {
-            document.getElementById('statToday').textContent = messageCache.length;
-            document.getElementById('statPlatforms').textContent = '--';
-            document.getElementById('statClients').textContent = '--';
-            document.getElementById('statPending').textContent = '--';
-            renderRecentMessages();
-        });
-    } catch (e) {
-        document.getElementById('statToday').textContent = '错误';
-    }
+    j('/api/stats').then(function (stats) {
+        document.getElementById('statPending').textContent = stats.pending || 0;
+        document.getElementById('statSending').textContent = stats.sending || 0;
+        document.getElementById('statSent').textContent = stats.sent || 0;
+        document.getElementById('statFailed').textContent = stats.failed || 0;
+        document.getElementById('statCanceled').textContent = stats.canceled || 0;
+        document.getElementById('statTotal').textContent = stats.total || 0;
+    }).catch(function () {
+        document.getElementById('statPending').textContent = 'ERR';
+    });
+    renderRecentMessages();
 }
 
 function renderRecentMessages() {
@@ -188,9 +188,9 @@ function renderRecentMessages() {
         var time = new Date(m.created_at * 1000).toLocaleTimeString('zh-CN');
         return '<tr>' +
             '<td><span class="badge info">' + esc(m.platform) + '</span></td>' +
-            '<td>' + esc(m.conversation.substring(0, 20)) + '</td>' +
-            '<td>' + esc(m.content.substring(0, 50)) + '</td>' +
-            '<td><span class="badge success">已接收</span></td>' +
+            '<td>' + esc((m.conversation || '').substring(0, 20)) + '</td>' +
+            '<td>' + esc((m.content || '').substring(0, 50)) + '</td>' +
+            '<td>' + statusBadge(m.status || 'Pending') + '</td>' +
             '<td>' + time + '</td>' +
             '</tr>';
     }).join('');
@@ -214,11 +214,11 @@ function renderMessages() {
         var sender = m.sender ? m.sender.name : '--';
         return '<tr>' +
             '<td><span class="badge info">' + esc(m.platform) + '</span></td>' +
-            '<td>' + esc(m.conversation.substring(0, 24)) + '</td>' +
+            '<td>' + esc((m.conversation || '').substring(0, 24)) + '</td>' +
             '<td>' + esc(sender) + '</td>' +
-            '<td>' + esc(m.content.substring(0, 40)) + '</td>' +
+            '<td>' + esc((m.content || '').substring(0, 40)) + '</td>' +
             '<td>' + esc(m.message_type) + '</td>' +
-            '<td><span class="badge success">已接收</span></td>' +
+            '<td>' + statusBadge(m.status || 'Pending') + '</td>' +
             '<td>' + time + '</td>' +
             '</tr>';
     }).join('');
@@ -299,6 +299,25 @@ function esc(s) {
     return String(s).replace(/&/g, '&amp;').replace(/</g, '&lt;').replace(/>/g, '&gt;').replace(/"/g, '&quot;');
 }
 
+function statusBadge(status) {
+    var map = {
+        Sent: 'success',
+        Pending: 'warning',
+        Sending: 'info',
+        Failed: 'danger',
+        Canceled: 'danger'
+    };
+    var label = {
+        Sent: '已发送',
+        Pending: '待发送',
+        Sending: '发送中',
+        Failed: '失败',
+        Canceled: '已取消'
+    };
+    var cls = map[status] || 'info';
+    return '<span class="badge ' + cls + '">' + (label[status] || status || '未知') + '</span>';
+}
+
 // ========== Boot ==========
 
 if (accessToken) {

src/web/templates/index.html

@@ -69,20 +69,28 @@ <h2>ReChat 登录</h2>
         <h2 class="section-title">仪表盘</h2>
         <div class="stats-grid" id="statsGrid">
             <div class="card stat-card">
-                <div class="stat-label">今日消息</div>
-                <div class="stat-value" id="statToday">--</div>
+                <div class="stat-label">待发送</div>
+                <div class="stat-value" id="statPending">--</div>
             </div>
             <div class="card stat-card">
-                <div class="stat-label">活跃平台</div>
-                <div class="stat-value" id="statPlatforms">--</div>
+                <div class="stat-label">发送中</div>
+                <div class="stat-value" id="statSending">--</div>
             </div>
             <div class="card stat-card">
-                <div class="stat-label">WS 客户端</div>
-                <div class="stat-value" id="statClients">--</div>
+                <div class="stat-label">已发送</div>
+                <div class="stat-value" id="statSent">--</div>
             </div>
             <div class="card stat-card">
-                <div class="stat-label">待发送</div>
-                <div class="stat-value" id="statPending">--</div>
+                <div class="stat-label">失败</div>
+                <div class="stat-value" id="statFailed">--</div>
+            </div>
+            <div class="card stat-card">
+                <div class="stat-label">已取消</div>
+                <div class="stat-value" id="statCanceled">--</div>
+            </div>
+            <div class="card stat-card">
+                <div class="stat-label">总计</div>
+                <div class="stat-value" id="statTotal">--</div>
             </div>
         </div>