初步完成channel->queueu->[tool]->[queue]->[assistant]->channel的流程

【待处理】上下文管理和错误处理。

service/Cargo.toml

@@ -46,3 +46,7 @@ sqlx = { version = "0.9.0", features = [
 # ── 错误处理 ──
 anyhow = "1"
 thiserror = "2"
+# ── cli ──
+clap = { version = "4.6.1", features = ["derive"] }
+# ── 时间处理 ──
+chrono = { version = "0.4", features = ["serde"] }             # 日期时间

service/migrations/20260601000001_init.sql

@@ -1,29 +0,0 @@
--- iAs 初始数据库迁移
--- 认证信息和聊天记录存储
-
--- 应用状态表（key-value 存储，用于认证信息等）
-CREATE TABLE IF NOT EXISTS app_state (
-    key TEXT PRIMARY KEY,
-    value JSONB NOT NULL,
-    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
-);
-
--- 聊天记录表
-CREATE TABLE IF NOT EXISTS chat_records (
-    id BIGSERIAL PRIMARY KEY,
-    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-    direction TEXT NOT NULL,          -- 'inbound' | 'outbound'
-    user_id TEXT NOT NULL,            -- 微信用户 ID
-    account_id TEXT NOT NULL,         -- 机器人账号 ID
-    text TEXT NOT NULL,               -- 消息内容
-    source TEXT NOT NULL DEFAULT '',  -- 消息来源
-    context_token TEXT NOT NULL DEFAULT '',  -- 上下文令牌
-    message_id TEXT NOT NULL DEFAULT '',
-    meta JSONB NOT NULL DEFAULT '{}'::JSONB
-);
-
-CREATE INDEX IF NOT EXISTS idx_chat_records_user_created
-    ON chat_records (user_id, created_at DESC);
-
-CREATE INDEX IF NOT EXISTS idx_chat_records_account_created
-    ON chat_records (account_id, created_at DESC);

service/migrations/20260601000002_pending_approvals.sql

@@ -1,17 +0,0 @@
--- 待审批的工具调用
-CREATE TABLE IF NOT EXISTS pending_approvals (
-    id UUID PRIMARY KEY,
-    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-    expires_at TIMESTAMPTZ NOT NULL,
-    user_id TEXT NOT NULL,
-    skill_name TEXT NOT NULL,
-    params JSONB NOT NULL,
-    code_hash TEXT NOT NULL,
-    attempts_left INTEGER NOT NULL DEFAULT 3,
-    status TEXT NOT NULL DEFAULT 'pending',
-    result JSONB,
-    consumed_at TIMESTAMPTZ
-);
-
-CREATE INDEX IF NOT EXISTS idx_pending_approvals_user_status
-    ON pending_approvals (user_id, status);

service/migrations/20260601000003_scheduled_tasks.sql

@@ -1,20 +0,0 @@
--- 定时任务表（Phase 6）
-CREATE TABLE IF NOT EXISTS scheduled_tasks (
-    id UUID PRIMARY KEY,
-    name TEXT NOT NULL UNIQUE,
-    description TEXT NOT NULL DEFAULT '',
-    user_id TEXT NOT NULL,
-    command TEXT NOT NULL,
-    shell TEXT NOT NULL DEFAULT '/bin/bash',
-    cwd TEXT NOT NULL DEFAULT '',
-    interval_seconds INTEGER NOT NULL,
-    enabled BOOLEAN NOT NULL DEFAULT true,
-    next_run_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-    last_run_at TIMESTAMPTZ,
-    last_status TEXT,
-    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
-);
-
-CREATE INDEX IF NOT EXISTS idx_scheduled_tasks_due
-    ON scheduled_tasks (enabled, next_run_at);

service/migrations/20260601000004_llm_usage.sql

@@ -1,15 +0,0 @@
--- LLM Token 用量统计
-CREATE TABLE IF NOT EXISTS llm_usage (
-    id BIGSERIAL PRIMARY KEY,
-    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-    user_id TEXT NOT NULL DEFAULT '',
-    model TEXT NOT NULL DEFAULT '',
-    provider TEXT NOT NULL DEFAULT 'deepseek',
-    prompt_tokens INTEGER NOT NULL DEFAULT 0,
-    completion_tokens INTEGER NOT NULL DEFAULT 0,
-    total_tokens INTEGER NOT NULL DEFAULT 0,
-    cache_hit_tokens INTEGER NOT NULL DEFAULT 0,
-    cache_miss_tokens INTEGER NOT NULL DEFAULT 0
-);
-
-CREATE INDEX IF NOT EXISTS idx_llm_usage_created ON llm_usage (created_at DESC);

service/migrations/20260601000005_user_memories.sql

@@ -1,9 +0,0 @@
--- 长期记忆表
-CREATE TABLE IF NOT EXISTS user_memories (
-    id SERIAL PRIMARY KEY,
-    user_id TEXT NOT NULL DEFAULT 'default',
-    content TEXT NOT NULL,
-    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
-);
-
-CREATE INDEX IF NOT EXISTS idx_user_memories_user ON user_memories (user_id);

service/migrations/20260601000006_session_summaries.sql

@@ -1,12 +0,0 @@
--- 会话摘要持久化
-CREATE TABLE IF NOT EXISTS session_summaries (
-    id SERIAL PRIMARY KEY,
-    user_id TEXT NOT NULL DEFAULT 'default',
-    summary_text TEXT NOT NULL,
-    reason TEXT NOT NULL DEFAULT 'overflow',
-    message_count INTEGER NOT NULL DEFAULT 0,
-    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
-);
-
-CREATE INDEX IF NOT EXISTS idx_session_summaries_created
-    ON session_summaries (created_at DESC);

service/migrations/20260601000007_fix_llm_usage.sql

@@ -1,11 +0,0 @@
--- 修复 llm_usage 表：从 iPet 旧格式迁移到 iAs 新格式
--- 旧表有 kind/entry 字段，新表需要 prompt_tokens/completion_tokens 等
-
-ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS prompt_tokens INTEGER NOT NULL DEFAULT 0;
-ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS completion_tokens INTEGER NOT NULL DEFAULT 0;
-ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS total_tokens INTEGER NOT NULL DEFAULT 0;
-ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS cache_hit_tokens INTEGER NOT NULL DEFAULT 0;
-ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS cache_miss_tokens INTEGER NOT NULL DEFAULT 0;
-ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS user_id TEXT NOT NULL DEFAULT '';
-ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS model TEXT NOT NULL DEFAULT '';
-ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS provider TEXT NOT NULL DEFAULT 'deepseek';

service/migrations/20260601000008_fix_scheduled_tasks_user_id.sql

@@ -1,3 +0,0 @@
--- scheduled_tasks.user_id 默认值修复
-ALTER TABLE scheduled_tasks ALTER COLUMN user_id SET DEFAULT '';
-ALTER TABLE scheduled_tasks ALTER COLUMN user_id DROP NOT NULL;

service/migrations/20260601000009_chat_dedup.sql

@@ -1,3 +0,0 @@
--- 聊天记录去重索引（普通索引，不用 UNIQUE 避免存量重复数据冲突）
-CREATE INDEX IF NOT EXISTS idx_chat_records_message_id
-    ON chat_records (message_id) WHERE message_id != '';

service/migrations/20260601000010_fix_llm_usage_legacy_columns.sql

@@ -1,4 +0,0 @@
--- 移除 llm_usage 表中旧 iPet 遗留列的 NOT NULL 约束
--- kind/entry 是旧格式字段，新代码不使用它们
-ALTER TABLE llm_usage ALTER COLUMN kind DROP NOT NULL;
-ALTER TABLE llm_usage ALTER COLUMN entry DROP NOT NULL;

service/migrations/20260601000011_fix_chat_records_context_token.sql

@@ -1,26 +0,0 @@
--- 修复 chat_records 表中 iPet 遗留的 context_token_present (boolean) → context_token (text)
--- 迁移 0001 创建的是 context_token TEXT，但旧库中列名是 context_token_present BOOLEAN
-DO $$
-BEGIN
-    IF EXISTS (SELECT 1 FROM information_schema.columns
-               WHERE table_name = 'chat_records' AND column_name = 'context_token_present') THEN
-
-        IF EXISTS (SELECT 1 FROM information_schema.columns
-                   WHERE table_name = 'chat_records' AND column_name = 'context_token') THEN
-            -- 两列都存在（混合/半迁移库）：回填数据后删除旧列
-            UPDATE chat_records
-                SET context_token = CASE WHEN context_token_present::boolean THEN '1' ELSE '' END
-                WHERE context_token = '';
-            ALTER TABLE chat_records DROP COLUMN context_token_present;
-        ELSE
-            -- 只有旧列：类型转换 boolean → text (true → '1', false → '')，再重命名
-            ALTER TABLE chat_records
-                ALTER COLUMN context_token_present TYPE TEXT
-                USING CASE WHEN context_token_present::boolean THEN '1' ELSE '' END;
-            ALTER TABLE chat_records RENAME COLUMN context_token_present TO context_token;
-            ALTER TABLE chat_records ALTER COLUMN context_token SET DEFAULT '';
-        END IF;
-
-    END IF;
-END;
-$$;

service/migrations/20260601000012_fix_chat_dedup.sql

@@ -1,12 +0,0 @@
--- 聊天记录去重：清理存量重复 → 建 partial UNIQUE 索引支持 ON CONFLICT
--- 1) 删除重复记录（保留每个 message_id 最早的那条）
-DELETE FROM chat_records a
-USING chat_records b
-WHERE a.message_id = b.message_id
-  AND a.message_id NOT IN ('', '0')
-  AND a.id > b.id;
-
--- 2) 替换旧普通索引为 UNIQUE partial index（排除空和占位值）
-DROP INDEX IF EXISTS idx_chat_records_message_id;
-CREATE UNIQUE INDEX idx_chat_records_message_id
-    ON chat_records (message_id) WHERE message_id NOT IN ('', '0');

service/migrations/20260601000013_fix_llm_usage_kind_entry_newdb.sql

@@ -1,3 +0,0 @@
--- 确保 llm_usage 表在全新数据库上也有 kind/entry 列（迁移 0010 仅解除了旧库的 NOT NULL）
-ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS kind TEXT;
-ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS entry JSONB;

service/migrations/20260601000014_scheduled_tasks_type.sql

@@ -1,6 +0,0 @@
--- 定时任务增加 task_type 字段：system（系统任务）或 model（模型任务）
--- 模型只能管理 model 类型的任务，不能修改 system 任务
-ALTER TABLE scheduled_tasks ADD COLUMN IF NOT EXISTS task_type TEXT NOT NULL DEFAULT 'model';
-
-CREATE INDEX IF NOT EXISTS idx_scheduled_tasks_type
-    ON scheduled_tasks (task_type, enabled, next_run_at);

service/migrations/20260601000015_dummy_noop.sql

@@ -1,4 +0,0 @@
--- 空迁移：用于对齐 _sqlx_migrations 表中已记录但文件丢失的迁移
--- 该迁移已被之前的某次运行应用过，但对应文件被删除/重命名
--- 此文件仅用于让 sqlx 迁移校验通过，不执行任何 DDL
-SELECT 1;

service/migrations/20260601000016_session_summaries_user_idx.sql

@@ -1,4 +0,0 @@
--- 为"按用户查询最近的会话边界（clear/timeout 摘要）"添加索引。
--- list_recent_chat_records_for_context 会据此过滤掉已过期会话的消息。
-CREATE INDEX IF NOT EXISTS idx_session_summaries_user_created
-    ON session_summaries (user_id, created_at DESC);

service/migrations/20260625031750_create_account.sql

@@ -1,10 +1,12 @@
 -- Add migration script here
-CREATE TABLE ias_wechat_users (
+CREATE TABLE ias_wechat_accounts (
     id BIGSERIAL PRIMARY KEY,
     user_name VARCHAR(100) NOT NULL,
     token VARCHAR(255),
     account_id VARCHAR(255),
     base_url VARCHAR(255),
+    user_id VARCHAR(255),
+    updates_buf VARCHAR(255),
     user_status INTEGER,
     created_at TIMESTAMP NOT NULL DEFAULT NOW()
 );

service/migrations/20260625035946_wecaht_user_add_id.sql

@@ -1,2 +0,0 @@
--- Add migration script here
-ALTER TABLE ias_wechat_users ADD COLUMN user_id VARCHAR(255);

service/src/channel/wechat.rs

@@ -1,63 +0,0 @@
-use std::time::Duration;
-use wechat::{client::WeChatClient, types::WeixinMessage};
-
-use crate::AppState;
-
-pub async fn init_wechat(state: Option<AppState>) -> Result<(), Box<dyn std::error::Error>> {
-    match state {
-        Some(state)=>{
-            load_config(state);
-        }
-        None=>{}
-    }
-    
-    let client = WeChatClient::new(None);
-    
-
-    let login = client
-        .login(
-            |qrcode_url| {
-                println!("二维码地址: {qrcode_url}");
-            },
-            120,
-        )
-        .await?;
-
-    // println!("登录成功: account_id={}", login.account_id);
-
-    // client.notify_start().await?;
-
-    // loop {
-    //     let updates = client.receive_messages().await?;
-
-    //     for msg in updates.msgs {
-    //         if msg.msg_type != Some(WeixinMessage::TYPE_USER) {
-    //             continue;
-    //         }
-
-    //         let Some(from_user_id) = msg.from_user_id.as_deref() else {
-    //             continue;
-    //         };
-
-    //         let Some(text) = msg.text_content() else {
-    //             continue;
-    //         };
-
-    //         client
-    //             .send_text(
-    //                 from_user_id,
-    //                 &format!("收到: {text}"),
-    //                 msg.context_token.as_deref(),
-    //             )
-    //             .await?;
-    //     }
-
-    //     tokio::time::sleep(Duration::from_secs(1)).await;
-    // }
-    Ok(())
-}
-
-fn load_config(state: AppState){
-    let db = state.db;
-    
-}

service/src/channel/wechat/looper.rs

@@ -0,0 +1,67 @@
+use std::sync::Arc;
+
+use crate::{AppState, repository::user_repository::UserRepositor};
+use common::queue::{ChannelMessage, QueueMessage};
+use sqlx::PgPool;
+use tokio::sync::mpsc::Receiver;
+use wechat::{manager::WeChatEvent, types::WeixinMessage};
+
+pub async fn start_loop(
+    state: Arc<AppState>,
+    mut event_rx: Receiver<WeChatEvent>,
+) -> Result<(), String> {
+    tokio::spawn(async move {
+        while let Some(event) = event_rx.recv().await {
+            match event {
+                WeChatEvent::UpdatesBuf {
+                    account_id,
+                    updates_buf,
+                } => {
+                    persist_updates_buf(&state.db, account_id, updates_buf).await;
+                }
+                WeChatEvent::Message {
+                    account_id,
+                    message,
+                    updates_buf,
+                } => {
+                    persist_updates_buf(&state.db, account_id.clone(), updates_buf).await;
+
+                    if message.msg_type != Some(WeixinMessage::TYPE_USER) {
+                        continue;
+                    }
+
+                    let Some(from_user_id) = message.from_user_id.as_deref() else {
+                        continue;
+                    };
+                    let Some(text) = message.text_content() else {
+                        continue;
+                    };
+                    println!("Message account_id:{}=>{}", account_id.clone(), text);
+
+                    state
+                        .sender
+                        .clone()
+                        .send(QueueMessage::Channel(ChannelMessage {
+                            channel: "wechat".to_string(),
+                            account_id: account_id,
+                            from_user_id: from_user_id.to_string(),
+                            content: text.to_string(),
+                            context: message.context_token,
+                        }))
+                        .await
+                        .unwrap();
+                }
+                WeChatEvent::PollError { account_id, error } => {
+                    println!("微信账号轮询失败 account_id={account_id}: {error}");
+                }
+            }
+        }
+    });
+    Ok(())
+}
+
+async fn persist_updates_buf(pool: &PgPool, account_id: String, buf: String) {
+    UserRepositor::update_account_buf(pool, account_id, buf)
+        .await
+        .unwrap();
+}

service/src/channel/wechat/manager.rs

@@ -0,0 +1,123 @@
+use crate::model::user::{WechatAccount, WechatAccountCreate};
+use sqlx::PgPool;
+use std::io;
+use std::sync::Arc;
+use wechat::client::WeChatClient;
+use wechat::manager::{WeChatAccountConfig, WeChatMultiAccountManager};
+
+use crate::AppState;
+use crate::channel::wechat::looper::start_loop;
+use crate::repository::user_repository::UserRepositor;
+
+pub async fn init_wechat(state: Option<AppState>) -> Option<WeChatMultiAccountManager> {
+    if let Some(state) = state {
+        println!("init_wechat");
+        if let Some(accounts) = load_accounts(&state.db).await {
+            println!("accounts:{}", accounts.len());
+            let (mut manager, event_rx) = WeChatMultiAccountManager::new(1024);
+            for account in accounts {
+                println!("account:{}", account.account_id.clone().unwrap_or_default());
+                manager
+                    .start_account(WeChatAccountConfig {
+                        token: account.token.unwrap_or_default(),
+                        base_url: account.base_url.unwrap_or_default(),
+                        account_id: account.account_id.unwrap_or_default(),
+                        updates_buf: account.updates_buf.unwrap_or_default(),
+                    })
+                    .await
+                    .unwrap();
+            }
+
+            start_loop(Arc::new(state.clone()), event_rx).await.unwrap();
+            return Some(manager);
+        }
+    }
+    None
+}
+
+pub async fn login_user(pool: &PgPool) {
+    let client = WeChatClient::new(None);
+
+    let login = client
+        .login(
+            |qrcode_url| {
+                println!("二维码地址: {qrcode_url}");
+            },
+            120,
+        )
+        .await;
+    match login {
+        Ok(account) => {
+            let account_id = account.account_id;
+            if let Ok(exist) = UserRepositor::find_wechat_account(pool, account_id.clone()).await {
+                eprintln!(
+                    "账号已存在:{}:{}",
+                    exist.account_id.unwrap_or_default(),
+                    exist.user_name.unwrap_or_default()
+                );
+            } else {
+                let mut user_name = String::new();
+
+                io::stdin().read_line(&mut user_name).unwrap();
+
+                let user_name = if user_name.trim().is_empty() {
+                    let user_start: String = account_id.clone().chars().take(6).collect();
+                    format!("user{}", user_start).to_string()
+                } else {
+                    user_name.trim().to_string()
+                };
+                let user = WechatAccountCreate {
+                    account_id: account_id,
+                    token: account.token,
+                    base_url: account.base_url,
+                    user_id: account.user_id,
+                    user_name: user_name.to_string(),
+                };
+
+                // todo 后续需要处理错误
+                UserRepositor::add_wechat_account(&pool, user)
+                    .await
+                    .unwrap();
+            }
+        }
+        Err(err) => {
+            eprintln!("登录失败，请重试:{}", err)
+        }
+    }
+}
+
+pub async fn list_accounts(pool: &PgPool) {
+    println!("已登录用户如下：");
+    if let Some(accounts) = load_accounts(pool).await {
+        for account in accounts {
+            let user_name = account.user_name.unwrap_or_default();
+            let account_id = account.account_id.unwrap_or_default();
+            println!("  - {}:{}", user_name, account_id);
+        }
+    }
+}
+
+pub async fn rename_account(pool: &PgPool, account: String, name: String) {
+    if let Ok(lines) = UserRepositor::update_account_name(pool, account, name).await
+        && lines > 0
+    {
+        println!("修改成功")
+    }
+}
+pub async fn delete_account(pool: &PgPool, account: String) {
+    if let Ok(lines) = UserRepositor::delete_account(pool, account).await
+        && lines > 0
+    {
+        println!("删除成功")
+    }
+}
+
+async fn load_accounts(pool: &PgPool) -> Option<Vec<WechatAccount>> {
+    let accounts = UserRepositor::list_wechat_accounts(pool)
+        .await
+        .unwrap_or_default();
+    if !accounts.is_empty() {
+        return Some(accounts);
+    }
+    None
+}

service/src/channel/wechat/mod.rs

@@ -0,0 +1,2 @@
+pub mod manager;
+pub mod looper;

service/src/core/cli.rs

@@ -0,0 +1,52 @@
+use clap::{Parser, Subcommand};
+
+#[derive(Parser)]
+#[command(name = "ias")]
+#[command(version = "1.0")]
+#[command(about = "Rust驱动的智能AI助手。")]
+pub struct Cli {
+    #[command(subcommand)]
+    pub command: Commands,
+}
+
+#[derive(Debug, Subcommand)]
+pub enum Commands {
+    // 启动服务
+    Service,
+    // 配置渠道
+    Channel {
+        #[command(subcommand)]
+        command: Channels,
+    },
+}
+
+#[derive(Debug, Subcommand)]
+pub enum Channels {
+    // 微信渠道配置
+    Wechat {
+        #[command(subcommand)]
+        command: WechatAction,
+    },
+}
+#[derive(Debug, Subcommand)]
+pub enum WechatAction {
+    /// 登录新的微信账号
+    Login,
+    /// 查看已经登录的微信账号
+    List,
+    /// 删除已登录的微信账号
+    Delete {
+        /// 要删除的微信账号
+        #[arg(short, long)]
+        account: String,
+    },
+    /// 修改账号名称 - 仅做标记，与微信名无关
+    Rename {
+        /// 要修改的微信账号
+        #[arg(short, long)]
+        account: String,
+        /// 名称 仅做标记
+        #[arg(short, long)]
+        name: String,
+    },
+}

service/src/core/context.rs

@@ -2,7 +2,7 @@ use std::sync::OnceLock;
 
 use tokio::sync::Mutex;
 
-use common::ai::ChatCompletionRequestMessage;
+use common::ai::{ChatCompletionRequestMessage, ChatCompletionToolCall};
 
 static CHAT_CONTEXT: OnceLock<Mutex<Vec<ChatCompletionRequestMessage>>> = OnceLock::new();
 
@@ -20,8 +20,15 @@ pub async fn snapshot_context() -> Vec<ChatCompletionRequestMessage> {
     context.clone()
 }
 
-pub async fn add_assistant_message(content: String) {
-    let message = ChatCompletionRequestMessage::new("assistant".to_string(), content);
+pub async fn add_assistant_message(
+    content: String,
+    tool_calls: Option<Vec<ChatCompletionToolCall>>,
+) {
+    let message = ChatCompletionRequestMessage::assistant(
+        "assistant".to_string(),
+        content,
+        tool_calls,
+    );
     push_context_message(message).await;
 }
 
@@ -31,6 +38,6 @@ pub async fn add_user_message(content: String) {
 }
 
 pub async fn add_tool_message(call_id: String, result: String) {
-    let message = ChatCompletionRequestMessage::tool("user".to_string(), result, call_id);
+    let message = ChatCompletionRequestMessage::tool("tool".to_string(), result, call_id);
     push_context_message(message).await;
 }

service/src/core/mod.rs

@@ -1,2 +1,3 @@
-pub mod queue;
-pub mod context;
+pub mod cli;
+pub mod context;
+pub mod queue;

service/src/core/queue.rs

@@ -1,62 +1,84 @@
-use crate::{
-    AppState,
-    core::context::{add_assistant_message, add_tool_message, add_user_message, snapshot_context},
+use crate::core::context::{
+    add_assistant_message, add_tool_message, add_user_message, snapshot_context,
 };
 use ai::core::send_message;
-use axum::Json;
-use axum::extract::State;
-use common::queue::{ChannelMessage, QueueMessage};
+use common::queue::QueueMessage;
 use serde_json::error;
-use std::sync::Arc;
-use tokio::sync::{mpsc, mpsc::Sender};
+use tokio::sync::{mpsc::Receiver, mpsc::Sender};
+use wechat::manager::WeChatMultiAccountManager;
 
-pub async fn create_queue() -> Result<Sender<QueueMessage>, error::Error> {
-    let (sender, mut receiver) = mpsc::channel::<QueueMessage>(100);
-    let result_sender = sender.clone();
+pub async fn create_queue(
+    sender: Sender<QueueMessage>,
+    mut receiver: Receiver<QueueMessage>,
+    manager: WeChatMultiAccountManager,
+) -> Result<(), error::Error> {
     tokio::spawn(async move {
         while let Some(msg) = receiver.recv().await {
             let queue_sender = sender.clone();
             match msg {
-                QueueMessage::Aissitant(reasoning, content) => {
+                QueueMessage::Aissitant(channel, assistant) => {
+                    let content = assistant.content.clone();
+                    let reasoning = assistant.reasoning.clone().unwrap_or_default();
                     println!("\n\n收到llm回复：\n{}\n-------\n{}", reasoning, content);
-                    add_assistant_message(content).await;
+                    if !content.trim().is_empty() || !reasoning.trim().is_empty() {
+                        let text = if !content.trim().is_empty() {
+                            content.trim()
+                        } else {
+                            reasoning.trim()
+                        };
+                        manager
+                            .send_text(
+                                &channel.account_id,
+                                &channel.from_user_id,
+                                &text,
+                                channel.context.as_deref(),
+                            )
+                            .await
+                            .unwrap();
+                    }
+                    add_assistant_message(assistant.content, assistant.tool_calls).await;
                 }
-                QueueMessage::Channel(name, text) => {
-                    println!("\n\n收到来自{}消息：\n{}", name, text);
-                    add_user_message(text).await;
-                    send_message(queue_sender, snapshot_context().await)
+                QueueMessage::Channel(bundle) => {
+                    println!(
+                        "\n\n收到来自{}消息：\n{}",
+                        bundle.from_user_id.clone(),
+                        bundle.content.clone()
+                    );
+                    // todo 目前按单用户处理
+                    add_user_message(bundle.content.clone()).await;
+                    send_message(queue_sender, snapshot_context().await, bundle)
                         .await
                         .unwrap();
                 }
-                QueueMessage::Tool(tool_call_id, name, result) => {
+                QueueMessage::Tool(channel, tool_call_id, name, result) => {
                     println!("\n\n调用{}工具的结果:\n{}", name, result);
                     add_tool_message(tool_call_id, result).await;
-                    send_message(queue_sender, snapshot_context().await)
+                    send_message(queue_sender, snapshot_context().await, channel)
                         .await
                         .unwrap();
                 }
             }
         }
     });
-    Ok(result_sender)
+    Ok(())
 }
 
-pub async fn send_to_queue(
-    State(state): State<Arc<AppState>>,
-    Json(body): Json<ChannelMessage>,
-) -> Json<serde_json::Value> {
-    match state
-        .sender
-        .send(QueueMessage::Channel(body.channel, body.content))
-        .await
-    {
-        Ok(_) => Json(serde_json::json!({
-            "success":true,
-            "message":"queued"
-        })),
-        Err(_) => Json(serde_json::json!({
-            "success":true,
-            "message":"queue closed"
-        })),
-    }
-}
+// pub async fn send_to_queue(
+//     State(state): State<Arc<AppState>>,
+//     Json(body): Json<ChannelMessageBundle>,
+// ) -> Json<serde_json::Value> {
+// match state
+//     .sender
+//     .send(QueueMessage::Channel(body.channel, body.content))
+//     .await
+// {
+//     Ok(_) => Json(serde_json::json!({
+//         "success":true,
+//         "message":"queued"
+//     })),
+//     Err(_) => Json(serde_json::json!({
+//         "success":true,
+//         "message":"queue closed"
+//     })),
+// }
+// }

service/src/error.rs

@@ -5,7 +5,7 @@ use axum::{
 };
 
 use serde_json::json;
-
+#[allow(unused)]
 #[derive(Debug, thiserror::Error)]
 pub enum AppError {
     #[error("not found")]
@@ -30,4 +30,5 @@ impl IntoResponse for AppError {
     }
 }
 
+#[allow(unused)]
 pub type AppResult<T> = Result<T, AppError>;

service/src/http/app.rs

@@ -1,20 +1,21 @@
 use std::sync::Arc;
 
-use axum::routing::{get, post};
+use anyhow::Ok;
+use axum::routing::{get};
 use axum::{Json, Router};
 
 use crate::AppState;
-use crate::core::queue::send_to_queue;
+// use crate::core::queue::send_to_queue;
 
-pub async fn create_http(state: AppState, host: String) {
+pub async fn create_http(state: AppState, host: String) -> anyhow::Result<()> {
     let app = Router::new()
-        .route("/message", post(send_to_queue))
+        // .route("/message", post(send_to_queue))
         .route("/health", get(health))
         .with_state(Arc::new(state));
-
-    let listener = tokio::net::TcpListener::bind(host.clone()).await.unwrap();
-    axum::serve(listener, app).await.unwrap();
-    println!("服务启动在：{}", host);
+    let listener = tokio::net::TcpListener::bind(host.clone()).await?;
+    println!("服务启动在：http://{}", listener.local_addr()?);
+    axum::serve(listener, app).await?;
+    Ok(())
 }
 
 async fn health() -> Json<serde_json::Value> {

service/src/main.rs

@@ -2,8 +2,10 @@ mod channel;
 mod core;
 mod error;
 mod http;
+mod model;
 mod repository;
 
+use clap::Parser;
 use common::queue::QueueMessage;
 use core::queue::create_queue;
 use http::app::create_http;
@@ -12,7 +14,10 @@ use std::env;
 
 use tokio::sync::mpsc;
 
-use crate::channel::wechat::init_wechat;
+use crate::channel::wechat::manager::{
+    delete_account, init_wechat, list_accounts, login_user, rename_account,
+};
+use crate::core::cli::{Channels, Cli, Commands, WechatAction};
 
 #[derive(Clone)]
 pub struct AppState {
@@ -21,35 +26,49 @@ pub struct AppState {
 }
 
 #[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    if std::env::var("RUST_ANALYZER").is_err() {
-        dotenv::dotenv().ok();
-    }
-    // dotenvy::dotenv().ok();
-    let name = env::var("NAME").unwrap_or_else(|_| "ias".to_string());
-    let host = env::var("HOST").unwrap_or_else(|_| "0.0.0.0:9003".to_string());
+async fn main() -> anyhow::Result<()> {
+    dotenvy::dotenv().ok();
     let db_url = env::var("DATABASE_URL").expect("必须设置DATABASE_URL");
-
     let pool = PgPoolOptions::new()
         .max_connections(10)
         .connect(&db_url)
         .await?;
-
-    let sender = create_queue().await.unwrap();
-    let state = AppState {
-        sender: sender,
-        db: pool,
-    };
-
-    println!("{} start in : {}", name, host);
-    create_http(state.clone(), host).await;
-    init_wechat(Some(state)).await.unwrap();
+    let cli = Cli::parse();
+    match cli.command {
+        Commands::Service => {
+            service(pool).await.unwrap();
+        }
+        Commands::Channel { command } => match command {
+            Channels::Wechat { command } => match command {
+                WechatAction::Login => login_user(&pool).await,
+                WechatAction::List => list_accounts(&pool).await,
+                WechatAction::Delete { account } => delete_account(&pool, account).await,
+                WechatAction::Rename { account, name } => {
+                    rename_account(&pool, account, name).await
+                }
+            },
+        },
+    }
 
     Ok(())
 }
 
-#[test]
-fn test() {
-    let api_key = std::env::var("ROUTER_API_KEY").unwrap();
-    println!("{}", api_key);
+async fn service(pool: PgPool) -> anyhow::Result<()> {
+    let name: String = env::var("NAME").unwrap_or_else(|_| "ias".to_string());
+    let host = env::var("HOST").unwrap_or_else(|_| "0.0.0.0:9003".to_string());
+
+    println!("{}已启动...", name);
+    let (sender, receiver) = mpsc::channel::<QueueMessage>(100);
+    let state = AppState {
+        sender: sender.clone(),
+        db: pool,
+    };
+    let manger = init_wechat(Some(state.clone())).await.unwrap();
+    create_queue(sender.clone(), receiver, manger)
+        .await
+        .unwrap();
+
+    // 启动http服务器，务必放在最下面执行
+    create_http(state.clone(), host.clone()).await.unwrap();
+    Ok(())
 }

service/src/model/mod.rs

@@ -0,0 +1 @@
+pub mod user;

service/src/model/user.rs

@@ -0,0 +1,25 @@
+use chrono::NaiveDateTime;
+use serde::Serialize;
+use sqlx::prelude::FromRow;
+
+#[derive(Debug, Clone, Serialize, FromRow)]
+pub struct WechatAccount {
+    pub id: i64,
+    pub user_name: Option<String>,
+    pub token: Option<String>,
+    pub account_id: Option<String>,
+    pub base_url: Option<String>,
+    pub user_id: Option<String>,
+    pub updates_buf: Option<String>,
+    pub user_status: Option<i32>,
+    pub created_at: Option<NaiveDateTime>,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct WechatAccountCreate {
+    pub user_name: String,
+    pub token: String,
+    pub account_id: String,
+    pub base_url: String,
+    pub user_id: String,
+}

service/src/repository/user_repository.rs

@@ -1,39 +1,108 @@
-use sqlx::PgPool;
+use anyhow::Ok;
+use sqlx::{PgPool, Row};
 
-use common::db::user::{WechatUser, WechatUserCreate};
+use crate::model::user::{WechatAccount, WechatAccountCreate};
 
 pub struct UserRepositor;
 
 impl UserRepositor {
-    pub async fn create(pool: &PgPool, user: WechatUserCreate) -> anyhow::Result<isize> {
-        let record = sqlx::query!(
+    pub async fn add_wechat_account(
+        pool: &PgPool,
+        user: WechatAccountCreate,
+    ) -> anyhow::Result<i64> {
+        let row = sqlx::query(
             r#"
-            INSERT INTO ias_wechat_users (user_name,token,account_id,base_url,user_id,user_status)
-            VALUE ($1,$2,$3,$4,$5,1)
+            INSERT INTO ias_wechat_accounts (user_name,token,account_id,base_url,user_id,user_status)
+            VALUES ($1,$2,$3,$4,$5,1)
             RETURNING id
-            "#,
-            user.user_name,
-            user.token,
-            user.account_id,
-            user.base_url,
-            user.user_id,
+            "#
         )
+          .bind ( user.user_name)
+          .bind ( user.token)
+          .bind ( user.account_id)
+          .bind ( user.base_url)
+          .bind ( user.user_id)
         .fetch_one(pool)
         .await?;
-        Ok(record.id)
+        let id = row.try_get::<i64, _>("id");
+        Ok(id.unwrap_or_default())
     }
-    pub async fn find_by_user_id(pool: &PgPool, user_id: isize) -> anyhow::Result<isize> {
-        let user = sqlx::query_as!(
-            WechatUser,
+
+    pub async fn find_wechat_account(
+        pool: &PgPool,
+        account_id: String,
+    ) -> anyhow::Result<WechatAccount> {
+        let row = sqlx::query_as::<_, WechatAccount>(
             r#"
-            SELECT * 
-            FROM ias_wechat_users
-            WHERE user_id = $1
+            SELECT *
+            FROM ias_wechat_accounts
+            WHERE account_id = $1
+            AND user_status = 1
             "#,
-            user_id,
         )
-        .fetch_optional(pool)
+        .bind(account_id)
+        .fetch_one(pool)
         .await?;
-        Ok(user)
+        Ok(row)
+    }
+    pub async fn list_wechat_accounts(pool: &PgPool) -> anyhow::Result<Vec<WechatAccount>> {
+        let row = sqlx::query_as::<_, WechatAccount>(
+            r#"
+            SELECT *
+            FROM ias_wechat_accounts
+            WHERE user_status = 1
+            "#,
+        )
+        .fetch_all(pool)
+        .await?;
+        Ok(row)
+    }
+    pub async fn update_account_name(
+        pool: &PgPool,
+        account_id: String,
+        new_name: String,
+    ) -> anyhow::Result<u64> {
+        let result = sqlx::query(
+            r#"
+            UPDATE ias_wechat_accounts 
+            SET user_name = $1 
+            WHERE account_id = $2 
+            "#,
+        )
+        .bind(new_name)
+        .bind(account_id)
+        .execute(pool)
+        .await?;
+        Ok(result.rows_affected())
+    }
+    pub async fn update_account_buf(
+        pool: &PgPool,
+        account_id: String,
+        buf: String,
+    ) -> anyhow::Result<u64> {
+        let result = sqlx::query(
+            r#"
+            UPDATE ias_wechat_accounts 
+            SET updates_buf = $1 
+            WHERE account_id = $2 
+            "#,
+        )
+        .bind(buf)
+        .bind(account_id)
+        .execute(pool)
+        .await?;
+        Ok(result.rows_affected())
+    }
+    pub async fn delete_account(pool: &PgPool, account_id: String) -> anyhow::Result<u64> {
+        let result = sqlx::query(
+            r#"
+            DELETE from ias_wechat_accounts 
+            WHERE account_id = $1
+            "#,
+        )
+        .bind(account_id)
+        .execute(pool)
+        .await?;
+        Ok(result.rows_affected())
     }
 }