rsclaw-runtime 2026.6.26

use std::sync::Arc;

use tokio::sync::mpsc;
use tracing::{debug, error, info, warn};

use super::{
    super::preparse::{btw_direct_call, is_fast_preparse, try_preparse_locally_with_account},
    default_dm_scope,
};
use crate::gateway::session::{MessageKind, SessionKeyParams, derive_session_key};
use rsclaw_agent::{AgentMessage, AgentRegistry};
use rsclaw_channel::{Channel, OutboundMessage};
use rsclaw_config::runtime::RuntimeConfig;

// ---------------------------------------------------------------------------
// Feishu (飞书)
// ---------------------------------------------------------------------------

/// Pick the Feishu identifier the bot should send replies to.
///
/// - Groups address by `chat_id` (`oc_xxx`).
/// - P2P addresses by `open_id` (`ou_xxx` — the same value the runtime tracks
///   as `sender_id` on inbound events).
///
/// P2P over `chat_id` is technically accepted by the Feishu API but
/// returns `230002` ("Bot/User can NOT be out of the chat") whenever
/// the per-user p2p chat session has been GC'd / rebuilt by the
/// platform — most visibly on delayed proactive pushes (plugin
/// notifications fired minutes after the user's last message).
/// `open_id` is identity-keyed and survives p2p session lifecycle
/// changes, so it's the safe address for any deferred outbound work.
fn outbound_addr_for(is_group: bool, chat_id: &str, sender_id: &str) -> String {
    if is_group {
        chat_id.to_owned()
    } else {
        sender_id.to_owned()
    }
}

pub(crate) fn start_feishu_if_configured(
    config: &RuntimeConfig,
    registry: Arc<AgentRegistry>,
    manager: &mut rsclaw_channel::ChannelManager,
    feishu_slot: Arc<tokio::sync::OnceCell<Arc<rsclaw_channel::feishu::FeishuChannel>>>,
    dm_enforcers: Arc<
        std::sync::RwLock<std::collections::HashMap<String, Arc<rsclaw_channel::DmPolicyEnforcer>>>,
    >,
    redb_store: Arc<rsclaw_store::redb_store::RedbStore>,
    _channel_senders: Arc<
        std::sync::RwLock<std::collections::HashMap<String, mpsc::Sender<OutboundMessage>>>,
    >,
    task_queue: Arc<crate::gateway::task_queue::TaskQueueManager>,
    shutdown: crate::gateway::ShutdownCoordinator,
) {
    let fs_cfg = config.channel.channels.feishu.as_ref();
    if let Some(cfg) = fs_cfg {
        if !cfg.base.enabled.unwrap_or(true) {
            return;
        }
    }

    // Collect (account_name, app_id, app_secret, brand) tuples from:
    //   1. channels.feishu.accounts.<name>.{appId, appSecret, brand?}
    //   2. saved auth token (onboard flow fallback, for backward compat)
    let mut fs_accounts: Vec<(String, String, String, String)> = Vec::new();

    // 1. Config file: accounts.<name>.{appId, appSecret, brand?}
    if let Some(accts) = fs_cfg.and_then(|c| c.accounts.as_ref()) {
        for (name, acct) in accts {
            let id = acct.get("appId").and_then(|v| v.as_str()).unwrap_or("");
            let secret = acct.get("appSecret").and_then(|v| v.as_str()).unwrap_or("");
            if !id.is_empty() && !secret.is_empty() {
                let brand = acct
                    .get("brand")
                    .and_then(|v| v.as_str())
                    .unwrap_or("feishu")
                    .to_owned();
                fs_accounts.push((name.clone(), id.to_owned(), secret.to_owned(), brand));
            }
        }
    }

    // 2. Saved auth token from onboard flow (legacy fallback).
    if fs_accounts.is_empty() {
        if let Some(saved) = rsclaw_channel::auth::load_token("feishu") {
            let id = saved["app_id"].as_str().unwrap_or("").to_owned();
            let secret = saved["app_secret"].as_str().unwrap_or("").to_owned();
            let brand = saved["brand"].as_str().unwrap_or("feishu").to_owned();
            if !id.is_empty() && !secret.is_empty() {
                fs_accounts.push(("default".to_owned(), id, secret, brand));
            }
        }
    }

    if fs_accounts.is_empty() {
        if fs_cfg.is_some() {
            warn!("feishu.appId not set in accounts, channel disabled");
        }
        return;
    }

    // Load dmPolicy and groupPolicy from config.
    let dm_policy = fs_cfg
        .and_then(|c| c.base.dm_policy.clone())
        .unwrap_or(rsclaw_config::schema::DmPolicy::Pairing);
    let group_policy = fs_cfg
        .and_then(|c| c.base.group_policy.clone())
        .unwrap_or(rsclaw_config::schema::GroupPolicy::Allowlist);
    let group_allow_from: Vec<String> = fs_cfg
        .and_then(|c| c.base.group_allow_from.clone())
        .unwrap_or_default();
    let allow_from: Vec<String> = fs_cfg
        .and_then(|c| c.base.allow_from.clone())
        .unwrap_or_default();

    let enforcer = Arc::new(
        rsclaw_channel::DmPolicyEnforcer::new(dm_policy, allow_from)
            .with_persistence("feishu", Arc::clone(&redb_store)),
    );
    if let Ok(mut enforcers) = dm_enforcers.write() {
        enforcers.insert("feishu".to_owned(), Arc::clone(&enforcer));
    }

    let feishu_api_base = fs_cfg.and_then(|c| c.api_base.clone());
    let feishu_ws_url = fs_cfg.and_then(|c| c.ws_url.clone());
    let feishu_reconnect_delay = fs_cfg.and_then(|c| c.reconnect_delay_secs).unwrap_or(5);
    let max_file_size = config
        .ext
        .tools
        .as_ref()
        .and_then(|t| t.upload.as_ref())
        .and_then(|u| u.max_file_size)
        .unwrap_or(128_000_000);
    let download_timeout_secs = config
        .ext
        .tools
        .as_ref()
        .and_then(|t| t.upload.as_ref())
        .and_then(|u| u.download_timeout_secs)
        .unwrap_or(600);

    for (acct_name, app_id, app_secret, brand) in fs_accounts {
        let reg = Arc::clone(&registry);
        let cfg = config.clone();
        let acct_for_log = acct_name.clone();
        let enforcer = Arc::clone(&enforcer);
        let gp = Arc::new(group_policy.clone());
        let ga = Arc::new(group_allow_from.clone());
        let tq = Arc::clone(&task_queue);
        let (out_tx, mut out_rx) = mpsc::channel::<OutboundMessage>(64);

        // Register channel sender for notification routing (ACP tools like OpenCode,
        // ClaudeCode).
        // - "feishu/{account}" is the canonical key — multi-account-aware callers use
        //   it.
        // - bare "feishu" is registered ONLY by the first account so legacy single-
        //   account callers still find a sender. Without this guard each account
        //   overwrote the bare key, leaving the last-registered account routing replies
        //   for messages received via every other account → Feishu 230002 "Bot/User can
        //   NOT be out of the chat" because that bot wasn't actually in the originating
        //   chat.
        {
            let mut senders = _channel_senders
                .write()
                .expect("channel_senders lock poisoned");
            senders.insert(format!("feishu/{}", acct_name), out_tx.clone());
            // Bare "feishu" fallback (used when an outbound message carries no
            // account). Must be DETERMINISTIC: bind it to the account named
            // "default", regardless of HashMap iteration order. The old
            // `.or_insert` bound it to whichever account iterated first out of
            // an unordered HashMap, so on multi-account setups the bare key
            // pointed at an arbitrary app's token → Feishu 99992361 "open_id
            // cross app" when that token didn't own the target open_id.
            if acct_name == "default" {
                senders.insert("feishu".to_string(), out_tx.clone());
            } else {
                // Configs without a "default"-named account still need *some*
                // bare fallback; first-wins is fine there since "default"
                // (above) overrides it whenever it exists.
                senders
                    .entry("feishu".to_string())
                    .or_insert_with(|| out_tx.clone());
            }
        }

        // Find binding for this account to determine which agent handles it.
        let bound_agent = config
            .agents
            .bindings
            .iter()
            .find(|b| {
                b.match_.channel.as_deref() == Some("feishu")
                    && b.match_.account_id.as_deref() == Some(&acct_name)
            })
            .map(|b| b.agent_id.clone());
        let bound = bound_agent.clone();
        // Capture acct_name for the on_message closure → per-user worker;
        // the inner spawn only sees what's been moved into the closure.
        let w_acct_outer = acct_name.clone();

        // Per-user inbound queue for Feishu.
        type FsItem = (
            String,
            String,
            String,
            bool,
            Option<String>,
            Vec<rsclaw_agent::registry::ImageAttachment>,
            Vec<rsclaw_agent::registry::FileAttachment>,
        );
        let fs_user_queues: Arc<
            tokio::sync::Mutex<std::collections::HashMap<String, mpsc::Sender<FsItem>>>,
        > = Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new()));

        let on_message = Arc::new(
            move |sender_id: String,
                  text: String,
                  chat_id: String,
                  is_group: bool,
                  images: Vec<rsclaw_agent::registry::ImageAttachment>,
                  file_attachments: Vec<rsclaw_agent::registry::FileAttachment>| {
                let reg = Arc::clone(&reg);
                let cfg = cfg.clone();
                let tx = out_tx.clone();
                let bound = bound.clone();
                let enforcer = Arc::clone(&enforcer);
                let group_policy = Arc::clone(&gp);
                let group_allow = Arc::clone(&ga);
                let queues = Arc::clone(&fs_user_queues);
                let tq = Arc::clone(&tq);
                let w_acct_outer = w_acct_outer.clone();
                tokio::spawn(async move {
                    let outbound_target = outbound_addr_for(is_group, &chat_id, &sender_id);
                    // Group policy check.
                    if is_group {
                        match group_policy.as_ref() {
                            rsclaw_config::schema::GroupPolicy::Disabled => {
                                warn!("feishu group message rejected: groupPolicy=disabled");
                                return;
                            }
                            rsclaw_config::schema::GroupPolicy::Allowlist => {
                                if !group_allow.iter().any(|g| *g == chat_id) {
                                    warn!("feishu group message rejected: not in groupAllowFrom");
                                    return;
                                }
                            }
                            rsclaw_config::schema::GroupPolicy::Open => {}
                        }
                    }
                    // DM policy check.
                    if !is_group {
                        use rsclaw_channel::PolicyResult;
                        match enforcer.check(&sender_id).await {
                            PolicyResult::Allow => {}
                            PolicyResult::Deny => {
                                warn!(peer_id = %sender_id, "feishu DM rejected by policy");
                                return;
                            }
                            PolicyResult::SendPairingCode(code) => {
                                if let Err(e) = tx
                                    .send(OutboundMessage {
                                        target_id: outbound_target.clone(),
                                        is_group: false,
                                        text: rsclaw_i18n::t_fmt(
                                            "pairing_required",
                                            rsclaw_i18n::default_lang(),
                                            &[("code", &code)],
                                        ),
                                        reply_to: None,
                                        images: vec![],
                                        channel: None,

                                        account: Some(w_acct_outer.clone()),
                                        files: vec![],
                                    })
                                    .await
                                {
                                    tracing::warn!("failed to send message: {e}");
                                }
                                return;
                            }
                            PolicyResult::PairingQueueFull => {
                                if let Err(e) = tx
                                    .send(OutboundMessage {
                                        target_id: outbound_target.clone(),
                                        is_group: false,
                                        text: rsclaw_i18n::t(
                                            "pairing_queue_full",
                                            rsclaw_i18n::default_lang(),
                                        )
                                        .to_owned(),
                                        reply_to: None,
                                        images: vec![],
                                        channel: None,

                                        account: Some(w_acct_outer.clone()),
                                        files: vec![],
                                    })
                                    .await
                                {
                                    tracing::warn!("failed to send message: {e}");
                                }
                                return;
                            }
                        }
                    }
                    // Fast preparse bypass: /status, /abort etc. skip per-user queue
                    if is_fast_preparse(&text) {
                        let handle = if let Some(ref agent_id) = bound {
                            match reg.get(agent_id) {
                                Ok(h) => h,
                                Err(_) => match reg.route_account("feishu", Some(&w_acct_outer)) {
                                    Ok(h) => h,
                                    Err(_) => return,
                                },
                            }
                        } else {
                            match reg.route_account("feishu", Some(&w_acct_outer)) {
                                Ok(h) => h,
                                Err(_) => return,
                            }
                        };
                        if let Some(mut reply) = try_preparse_locally_with_account(
                            &text,
                            &handle,
                            "feishu",
                            &sender_id,
                            Some(&w_acct_outer),
                            crate::gateway::preparse::PreparseOrigin::User,
                        )
                        .await
                        {
                            reply.target_id = outbound_target.clone();
                            reply.is_group = is_group;
                            if !reply.text.is_empty() || !reply.images.is_empty() {
                                if let Err(e) = tx.send(reply).await {
                                    tracing::warn!("failed to send message: {e}");
                                }
                            }
                            return;
                        }
                        // try_preparse_locally returned None (e.g. /clear sets
                        // abort then falls through to
                        // agent queue for actual cleanup)
                    }
                    // Get or create a per-user queue.
                    let user_tx = {
                        let mut map = queues.lock().await;
                        let needs_create = match map.get(&sender_id) {
                            Some(existing) if !existing.is_closed() => false,
                            Some(_) => {
                                map.remove(&sender_id);
                                true
                            }
                            None => true,
                        };
                        if needs_create {
                            let (utx, mut urx) = mpsc::channel::<FsItem>(32);
                            map.insert(sender_id.clone(), utx.clone());
                            let w_reg = Arc::clone(&reg);
                            let w_cfg = cfg.clone();
                            let w_uid = sender_id.clone();
                            let w_tq = Arc::clone(&tq);
                            let w_acct = w_acct_outer.clone();
                            tokio::spawn(async move {
                                while let Some((
                                    text,
                                    sender_id,
                                    chat_id,
                                    is_group,
                                    bound,
                                    images,
                                    file_attachments,
                                )) = urx.recv().await
                                {
                                    // No debounce — task queue merge_into_pending
                                    // handles rapid consecutive messages automatically.
                                    info!(user = %w_uid, text_start = %text.chars().take(20).collect::<String>(), "feishu: worker dispatching via task queue");
                                    // Resolve agent for session_key derivation.
                                    let handle = if let Some(ref agent_id) = bound {
                                        match w_reg.get(agent_id) {
                                            Ok(h) => h,
                                            Err(_) => match w_reg.route_account("feishu", Some(&w_acct)) {
                                                Ok(h) => h,
                                                Err(e) => {
                                                    error!("feishu route error: {e:#}");
                                                    continue;
                                                }
                                            },
                                        }
                                    } else {
                                        match w_reg.route_account("feishu", Some(&w_acct)) {
                                            Ok(h) => h,
                                            Err(e) => {
                                                error!("feishu route error: {e:#}");
                                                continue;
                                            }
                                        }
                                    };
                                    let dm_scope = default_dm_scope(&w_cfg);
                                    let session_key = derive_session_key(&SessionKeyParams {
                                        agent_id: handle.id.clone(),
                                        kind: if is_group {
                                            MessageKind::GroupMessage {
                                                group_id: chat_id.clone(),
                                                thread_id: None,
                                            }
                                        } else {
                                            MessageKind::DirectMessage {
                                                account_id: Some(w_acct.clone()),
                                            }
                                        },
                                        channel: "feishu".to_string(),
                                        peer_id: sender_id.clone(),
                                        dm_scope,
                                    });
                                    // Submit to persistent task queue (worker handles dispatch +
                                    // reply). `account` carries
                                    // the originating Feishu app name so the
                                    // task worker can route the reply via the same app's API
                                    // token (multi-account routing fix for 230002).
                                    //
                                    // `outbound_addr_for` resolves the
                                    // identity-keyed outbound target (groups
                                    // → chat_id, p2p → open_id/sender_id).
                                    // `chat_id` and `sender_id` here are the
                                    // per-message values flowing through the
                                    // worker channel — the outer-scope
                                    // `outbound_target` is owned by a
                                    // different spawn and isn't visible from
                                    // inside this worker, so we recompute.
                                    let qmsg = crate::gateway::task_queue::QueuedMessage {
                                        text,
                                        sender: sender_id.clone(),
                                        channel: "feishu".to_string(),
                                        account: Some(w_acct.clone()),
                                        chat_id: outbound_addr_for(is_group, &chat_id, &sender_id),
                                        is_group,
                                        reply_to: None,
                                        timestamp: chrono::Utc::now().timestamp(),
                                        images: images.iter().map(|i| i.data.clone()).collect(),
                                        files: file_attachments
                                            .iter()
                                            .filter_map(|f| {
                                                crate::gateway::task_queue::stage_file(
                                                    &f.filename,
                                                    &f.data,
                                                    &f.mime_type,
                                                )
                                                .ok()
                                            })
                                            .collect(),
                                    };
                                    if let Err(e) = w_tq.submit(
                                        &session_key,
                                        qmsg,
                                        crate::gateway::task_queue::Priority::User,
                                    ) {
                                        error!(user = %w_uid, "feishu: queue submit failed: {e:#}");
                                    }
                                }
                                debug!(user = %w_uid, "feishu: per-user worker stopped");
                            });
                            utx
                        } else {
                            map.get(&sender_id).expect("queue entry must exist").clone()
                        }
                    };
                    // /btw bypass: spawn directly, skip the per-user queue
                    if text.starts_with("/btw ") || text.starts_with("/BTW ") {
                        let reg = Arc::clone(&reg);
                        let tx = tx.clone();
                        let cfg = cfg.clone();
                        let question = text[5..].to_owned();
                        let target = outbound_target.clone();
                        let w_acct_btw = w_acct_outer.clone();
                        tokio::spawn(async move {
                            let handle = match reg.route_account("feishu", Some(&w_acct_btw)) {
                                Ok(h) => h,
                                Err(_) => match reg.route_account("feishu", None) {
                                    Ok(h) => h,
                                    Err(_) => return,
                                },
                            };
                            if let Some(reply_text) = btw_direct_call(
                                &question,
                                &handle.live_status,
                                &handle.providers,
                                &cfg,
                            )
                            .await
                            {
                                if let Err(e) = tx
                                    .send(OutboundMessage {
                                        target_id: target,
                                        is_group: false,
                                        text: format!("[/btw] {}", reply_text),
                                        reply_to: None,
                                        images: vec![],
                                        channel: None,

                                        account: Some(w_acct_btw.clone()),
                                        files: vec![],
                                    })
                                    .await
                                {
                                    tracing::warn!("failed to send message: {e}");
                                }
                            }
                        });
                        return;
                    }
                    // Fast preparse bypass: local commands skip per-user queue
                    if is_fast_preparse(&text) {
                        let reg = Arc::clone(&reg);
                        let tx = tx.clone();
                        let cfg = cfg.clone();
                        let sender_id = sender_id.clone();
                        let chat_id = chat_id.clone();
                        let outbound_target = outbound_target.clone();
                        let bound = bound.clone();
                        let w_acct_for_preparse = w_acct_outer.clone();
                        tokio::spawn(async move {
                            let handle = if let Some(ref agent_id) = bound {
                                match reg.get(agent_id) {
                                    Ok(h) => h,
                                    Err(_) => match reg.route_account("feishu", Some(&w_acct_for_preparse)) {
                                        Ok(h) => h,
                                        Err(_) => return,
                                    },
                                }
                            } else {
                                match reg.route_account("feishu", Some(&w_acct_for_preparse)) {
                                    Ok(h) => h,
                                    Err(_) => return,
                                }
                            };
                            let dm_scope = default_dm_scope(&cfg);
                            let session_key = derive_session_key(&SessionKeyParams {
                                agent_id: handle.id.clone(),
                                kind: if is_group {
                                    MessageKind::GroupMessage {
                                        group_id: chat_id.clone(),
                                        thread_id: None,
                                    }
                                } else {
                                    MessageKind::DirectMessage {
                                        account_id: Some(w_acct_for_preparse.clone()),
                                    }
                                },
                                channel: "feishu".to_string(),
                                peer_id: sender_id.clone(),
                                dm_scope,
                            });
                            if let Some(mut reply) = try_preparse_locally_with_account(
                                &text,
                                &handle,
                                "feishu",
                                &sender_id,
                                Some(&w_acct_for_preparse),
                                crate::gateway::preparse::PreparseOrigin::User,
                            )
                            .await
                            {
                                reply.target_id = outbound_target.clone();
                                reply.is_group = is_group;
                                if !reply.text.is_empty() || !reply.images.is_empty() {
                                    if let Err(e) = tx.send(reply).await {
                                        tracing::warn!("failed to send message: {e}");
                                    }
                                }
                                return;
                            }
                            let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
                            let msg = AgentMessage {
                                session_key,
                                text,
                                channel: "feishu".to_string(),
                                peer_id: sender_id,
                                chat_id: outbound_target.clone(),
                                reply_tx,
                                task_id: None,
                                context_id: None,
                                event_tx: None,
                                cancel_token: None,
                                input_request_tx: None,
                                extra_tools: vec![],
                                images,
                                files: file_attachments,
                                account: Some(w_acct_for_preparse.clone()),
                            };
                            if handle.tx.send(msg).await.is_err() {
                                return;
                            }
                            if let Ok(Ok(r)) =
                                tokio::time::timeout(std::time::Duration::from_secs(10), reply_rx)
                                    .await
                            {
                                if !r.is_empty {
                                    if let Err(e) = tx
                                        .send(OutboundMessage {
                                            target_id: outbound_target,
                                            is_group,
                                            text: r.text,
                                            reply_to: None,
                                            images: r.images,
                                            files: r.files,
                                            channel: None,
                                            account: Some(w_acct_for_preparse.clone()),
                                        })
                                        .await
                                    {
                                        tracing::warn!("failed to send message: {e}");
                                    }
                                }
                            }
                        });
                        return;
                    }
                    if let Err(e) = user_tx.try_send((
                        text,
                        sender_id.clone(),
                        chat_id,
                        is_group,
                        bound,
                        images,
                        file_attachments,
                    )) {
                        warn!(user = %sender_id, error = %e, "feishu: user queue full, dropping message");
                    }
                });
            },
        );

        let mut fs_channel =
            rsclaw_channel::feishu::FeishuChannel::new(app_id, app_secret, vec![], on_message);
        fs_channel.brand = brand;
        fs_channel.api_base_override = feishu_api_base.clone();
        fs_channel.ws_url_override = feishu_ws_url.clone();
        fs_channel.max_file_size = max_file_size;
        fs_channel.download_timeout_secs = download_timeout_secs;
        fs_channel.ws_reconnect_delay_secs = feishu_reconnect_delay;
        let fs = Arc::new(fs_channel);

        // First account fills the webhook slot for backward compatibility.
        if feishu_slot.set(Arc::clone(&fs)).is_err() {
            tracing::debug!("slot already set, skipping");
        }
        // Register under both `feishu/<acct>` (account-keyed; lets /watch
        // route deliveries back through the SAME app that received the
        // inbound message — open_ids are per-app, so cross-app routing
        // gets rejected with 99992361 "open_id cross app") and bare
        // `feishu` (first-account wins; matches the existing
        // `_channel_senders` semantic so legacy single-account callers
        // keep working unchanged).
        let acct_key = format!("feishu/{}", acct_for_log);
        if let Err(e) = manager.register_with_name(
            acct_key.clone(),
            Arc::clone(&fs) as Arc<dyn rsclaw_channel::Channel>,
        ) {
            tracing::warn!("failed to register channel `{acct_key}`: {e}");
        }
        if manager.get("feishu").is_none()
            && let Err(e) = manager.register_with_name(
                "feishu".to_owned(),
                Arc::clone(&fs) as Arc<dyn rsclaw_channel::Channel>,
            )
        {
            tracing::warn!("failed to register bare `feishu` channel: {e}");
        }

        // Outbound throughput engine: hash-partitioned concurrent workers.
        //
        // A single sequential drain caps throughput at ~1/min_interval and, more
        // importantly, blocks all recipients behind the slowest API call. Instead
        // we fan out to N workers keyed by hash(account,target_id): every message
        // for a given recipient always lands on the SAME worker, so per-recipient
        // ordering (including multi-chunk replies) is preserved exactly as a
        // sequential drain would; distinct recipients run concurrently for N×
        // throughput. Each worker still applies the per-account floor interval so
        // a single hot recipient can't trip feishu's per-app rate limit.
        // Override workers via FEISHU_OUTBOUND_WORKERS, floor via
        // FEISHU_OUTBOUND_MIN_MS.
        let min_interval = std::time::Duration::from_millis(
            std::env::var("FEISHU_OUTBOUND_MIN_MS")
                .ok()
                .and_then(|s| s.parse::<u64>().ok())
                .unwrap_or(40),
        );
        let num_workers = std::env::var("FEISHU_OUTBOUND_WORKERS")
            .ok()
            .and_then(|s| s.parse::<usize>().ok())
            .filter(|&n| n >= 1)
            .unwrap_or(8);

        // Spawn N per-worker queues; the dispatcher routes by hash.
        let mut worker_txs: Vec<mpsc::Sender<OutboundMessage>> = Vec::with_capacity(num_workers);
        for w in 0..num_workers {
            let (wtx, mut wrx) = mpsc::channel::<OutboundMessage>(64);
            worker_txs.push(wtx);
            let fs_send = Arc::clone(&fs);
            let shutdown_w = shutdown.clone();
            tokio::spawn(async move {
                let mut last_send: Option<std::time::Instant> = None;
                loop {
                    tokio::select! {
                        () = shutdown_w.notified() => {
                            info!(worker = w, "feishu: drain signaled, stopping outbound worker");
                            break;
                        }
                        msg = wrx.recv() => {
                            let Some(msg) = msg else { break };
                            if let Some(prev) = last_send {
                                let elapsed = prev.elapsed();
                                if elapsed < min_interval {
                                    tokio::time::sleep(min_interval - elapsed).await;
                                }
                            }
                            if let Err(e) = fs_send.send(msg).await {
                                error!("feishu send error: {e:#}");
                            }
                            last_send = Some(std::time::Instant::now());
                        }
                    }
                }
            });
        }

        let shutdown_for_out = shutdown.clone();
        tokio::spawn(async move {
            use std::hash::{Hash, Hasher};
            loop {
                tokio::select! {
                    () = shutdown_for_out.notified() => {
                        info!("feishu: drain signaled, stopping outbound dispatcher");
                        break;
                    }
                    msg = out_rx.recv() => {
                        let Some(msg) = msg else { break };
                        // Same (account,target) → same worker → ordering preserved.
                        let mut h = std::collections::hash_map::DefaultHasher::new();
                        msg.account.as_deref().unwrap_or("").hash(&mut h);
                        msg.target_id.hash(&mut h);
                        let idx = (h.finish() as usize) % worker_txs.len();
                        if let Err(e) = worker_txs[idx].send(msg).await {
                            error!("feishu outbound dispatch error: {e:#}");
                        }
                    }
                }
            }
        });

        let shutdown_for_run = shutdown.clone();
        tokio::spawn(async move {
            tokio::select! {
                res = fs.run() => {
                    if let Err(e) = res {
                        error!("feishu channel error: {e:#}");
                    }
                }
                () = shutdown_for_run.notified() => {
                    info!("feishu: drain signaled, stopping run loop");
                }
            }
        });

        info!(account = %acct_for_log, "feishu channel started");
    }
}