impyard 0.1.1

//! Slack — the Web API client for outbound messages and the Socket Mode
//! (WebSocket) client for inbound (docs/slack-channel.md). Trusted-side code
//! holds both tokens (from the vault); the box never does. Socket Mode keeps
//! impyard's posture: dial out, never listen on the internet. Base URL is
//! overridable via SLACK_API_BASE so the executor can be tested against a mock.
//!
//! Channel machinery (trust, mode, memory settings, history) is shared with
//! Discord — it is channel-id keyed and platform-agnostic; Slack ids (C…, D…)
//! drop straight in.

use crate::channel::discord::{channel_mode, channel_trusted, persist_message, set_channel_trust};
use crate::util::now_rfc3339;
use futures_util::{SinkExt, StreamExt};
use serde_json::{json, Value};
use std::collections::HashMap;
use std::sync::{Mutex, OnceLock};
use std::time::Duration;
use tokio_tungstenite::tungstenite::Message as Ws;

fn base() -> String {
    std::env::var("SLACK_API_BASE").unwrap_or_else(|_| "https://slack.com/api".into())
}

/// One Web API POST with the bot token. Slack signals failure inside a 200
/// body (`ok: false`), so both layers are checked.
async fn api(token: &str, method: &str, body: Value) -> Result<Value, String> {
    let res = reqwest::Client::new()
        .post(format!("{}/{method}", base()))
        .header("authorization", format!("Bearer {token}"))
        .json(&body)
        .send()
        .await
        .map_err(|e| e.to_string())?;
    let status = res.status();
    let body: Value = res.json().await.unwrap_or(Value::Null);
    if !status.is_success() || !body["ok"].as_bool().unwrap_or(false) {
        return Err(format!(
            "slack {method} {status}: {}",
            body["error"].as_str().unwrap_or("?")
        ));
    }
    Ok(body)
}

/// Post a message. Returns the message ts (Slack's message id). `thread_ts`
/// replies inside a thread.
pub async fn post_message(
    token: &str,
    channel_id: &str,
    text: &str,
    thread_ts: Option<&str>,
) -> Result<String, String> {
    let mut body = json!({ "channel": channel_id, "text": text });
    if let Some(ts) = thread_ts {
        body["thread_ts"] = json!(ts);
    }
    let res = api(token, "chat.postMessage", body).await?;
    Ok(res["ts"].as_str().unwrap_or("").to_string())
}

/// Open (or fetch) a DM with a user. Returns the DM channel id (D…).
pub async fn open_dm(token: &str, user_id: &str) -> Result<String, String> {
    let res = api(token, "conversations.open", json!({ "users": user_id })).await?;
    res["channel"]["id"]
        .as_str()
        .map(String::from)
        .ok_or_else(|| "DM had no channel id".into())
}

// ── inbound: the Socket Mode client ──────────────────────────────────────────

struct SmError {
    fatal: bool,
    msg: String,
}

/// Run Socket Mode for one imp: dial out, ack envelopes, dispatch events.
/// Reconnects on transient errors (Slack refreshes connections routinely);
/// stops on fatal ones (bad tokens).
pub async fn run_socket_mode(imp: &str, bot_token: &str, app_token: &str) {
    loop {
        match connect_once(imp, bot_token, app_token).await {
            Ok(()) => {} // routine disconnect envelope — reconnect immediately
            Err(e) if e.fatal => {
                eprintln!("slack socket-mode: {} — stopping.", e.msg);
                return;
            }
            Err(e) => {
                eprintln!("slack socket-mode: {} — reconnecting in 5s", e.msg);
                tokio::time::sleep(Duration::from_secs(5)).await;
            }
        }
    }
}

async fn connect_once(imp: &str, bot_token: &str, app_token: &str) -> Result<(), SmError> {
    let transient = |m: String| SmError {
        fatal: false,
        msg: m,
    };
    let fatal = |m: String| SmError {
        fatal: true,
        msg: m,
    };

    // Who are we? (Also validates the bot token.)
    let me = api(bot_token, "auth.test", json!({}))
        .await
        .map_err(|e| fatal(format!("auth.test: {e}")))?;
    let bot_user_id = me["user_id"].as_str().unwrap_or("").to_string();
    let team = me["team"].as_str().unwrap_or("?");

    // A fresh Socket Mode URL each connect (they are single-use).
    let open = api(app_token, "apps.connections.open", json!({}))
        .await
        .map_err(|e| {
            if e.contains("invalid_auth") || e.contains("not_allowed") {
                fatal(format!("apps.connections.open: {e}"))
            } else {
                transient(format!("apps.connections.open: {e}"))
            }
        })?;
    let url = open["url"]
        .as_str()
        .ok_or_else(|| transient("no socket url".into()))?;

    let (ws, _) = tokio_tungstenite::connect_async(url)
        .await
        .map_err(|e| transient(format!("connect: {e}")))?;
    let (mut sink, mut stream) = ws.split();
    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Ws>();
    let writer = tokio::spawn(async move {
        while let Some(m) = rx.recv().await {
            if sink.send(m).await.is_err() {
                break;
            }
        }
    });

    eprintln!("slack: connected as {bot_user_id} to workspace \"{team}\"");

    let mut admins: HashMap<String, bool> = HashMap::new();
    let result = loop {
        let msg = match stream.next().await {
            Some(Ok(Ws::Text(t))) => t,
            Some(Ok(Ws::Ping(p))) => {
                let _ = tx.send(Ws::Pong(p));
                continue;
            }
            Some(Ok(Ws::Close(_))) | None => break Err(transient("socket closed".into())),
            Some(Ok(_)) => continue,
            Some(Err(e)) => break Err(transient(format!("read: {e}"))),
        };
        let v: Value = match serde_json::from_str(&msg) {
            Ok(v) => v,
            Err(_) => continue,
        };
        match v["type"].as_str().unwrap_or("") {
            // Slack refreshes sockets on a schedule; this is routine, not an error.
            "disconnect" => break Ok(()),
            "hello" => {}
            "events_api" => {
                // Ack FIRST — Slack redelivers unacked envelopes, which would
                // double-file work if handling were slow.
                if let Some(id) = v["envelope_id"].as_str() {
                    let _ = tx.send(Ws::text(json!({ "envelope_id": id }).to_string()));
                }
                let event = &v["payload"]["event"];
                if event["type"].as_str() == Some("message") {
                    handle_message(imp, event, &bot_user_id, bot_token, &mut admins).await;
                }
            }
            _ => {}
        }
    };

    writer.abort();
    result
}

/// Is this Slack user a workspace admin/owner? Cached per connection — the
/// answer changes rarely and users.info costs a round trip.
async fn is_workspace_admin(token: &str, user_id: &str, cache: &mut HashMap<String, bool>) -> bool {
    if let Some(known) = cache.get(user_id) {
        return *known;
    }
    let admin = api(token, "users.info", json!({ "user": user_id }))
        .await
        .map(|r| {
            r["user"]["is_admin"].as_bool().unwrap_or(false)
                || r["user"]["is_owner"].as_bool().unwrap_or(false)
        })
        .unwrap_or(false);
    cache.insert(user_id.to_string(), admin);
    admin
}

async fn handle_message(
    imp: &str,
    event: &Value,
    bot_user_id: &str,
    bot_token: &str,
    admins: &mut HashMap<String, bool>,
) {
    // Never react to bots (including ourselves) — avoids reply loops. Skip
    // subtypes too (edits, joins, thread broadcasts): only fresh plain messages.
    if event["bot_id"].as_str().is_some() || event["subtype"].as_str().is_some() {
        return;
    }
    let user_id = event["user"].as_str().unwrap_or("");
    if user_id.is_empty() || user_id == bot_user_id {
        return;
    }
    let channel_id = event["channel"].as_str().unwrap_or("");
    if channel_id.is_empty() {
        return;
    }
    let is_dm = event["channel_type"].as_str() == Some("im");
    if is_dm {
        set_channel_trust(channel_id, true); // DMs are always trusted (1:1, sought-out)
    }
    let role = if is_dm {
        "trusted"
    } else if is_workspace_admin(bot_token, user_id, admins).await {
        "admin"
    } else if channel_trusted(channel_id) {
        "trusted"
    } else {
        "untrusted"
    };
    let text = event["text"].as_str().unwrap_or("");

    persist_message(
        channel_id,
        &json!({
            "ts": now_rfc3339(),
            "slack_ts": event["ts"].as_str().unwrap_or(""),
            "thread_ts": event["thread_ts"].as_str(),
            "author_id": user_id, "author": user_id, "role": role, "content": text,
        }),
    );

    // Wake rule (same as Discord): a DM, an @mention, or a channel in "all"
    // mode. In "mention" mode ambient messages are persisted but don't wake.
    let mentioned = text.contains(&format!("<@{bot_user_id}>"));
    if !(is_dm || mentioned || channel_mode(channel_id) == "all") {
        return;
    }

    let hint = if is_dm || mentioned {
        ""
    } else if crate::channel::discord::distinct_human_authors(channel_id) <= 1 {
        " [you're the only other person here — reply]"
    } else {
        " [group chat; you were not directly addressed — reply only if useful]"
    };
    let context = crate::imp::memory::RunContext {
        provider: "slack".into(),
        channel_id: Some(channel_id.to_string()),
        user_id: Some(user_id.to_string()),
        message_id: event["ts"].as_str().map(String::from),
        role: role.to_string(),
        is_dm,
        inbound: false, // live channel context carries ids; inbound marks relay tasks
    };
    eprintln!("slack: {user_id} ({role}) in {channel_id} → session");
    route_to_session(
        imp,
        channel_id,
        user_id.to_string(),
        format!("{text}{hint}"),
        context,
    )
    .await;
}

// ── conversation sessions: one warm box per active channel ────────────────────

fn sessions(
) -> &'static Mutex<HashMap<String, tokio::sync::mpsc::Sender<crate::run::boxed::SessionMessage>>> {
    static S: OnceLock<
        Mutex<HashMap<String, tokio::sync::mpsc::Sender<crate::run::boxed::SessionMessage>>>,
    > = OnceLock::new();
    S.get_or_init(|| Mutex::new(HashMap::new()))
}

const SESSION_IDLE_SECS: u64 = 90;

/// Deliver a message to the channel's live session, or start a new one —
/// the same warm-box pattern as Discord (idle exit, sender replaced on next
/// message). No typing indicator: Socket Mode has none.
async fn route_to_session(
    imp: &str,
    channel_id: &str,
    author_label: String,
    text: String,
    context: crate::imp::memory::RunContext,
) {
    let start_context = context.clone();
    let message = crate::run::boxed::SessionMessage {
        text,
        author_label,
        context,
    };
    let delivered = {
        let map = sessions().lock().unwrap();
        match map.get(channel_id) {
            Some(tx) => tx
                .try_send(crate::run::boxed::SessionMessage {
                    text: message.text.clone(),
                    author_label: message.author_label.clone(),
                    context: message.context.clone(),
                })
                .is_ok(),
            None => false,
        }
    };
    if delivered {
        return;
    }
    let (tx, rx) = tokio::sync::mpsc::channel::<crate::run::boxed::SessionMessage>(64);
    let _ = tx.try_send(message);
    sessions()
        .lock()
        .unwrap()
        .insert(channel_id.to_string(), tx);
    let (w, run_id) = (imp.to_string(), crate::run::boxed::new_run_id());
    tokio::spawn(async move {
        if let Err(e) = crate::run::boxed::run_session(
            &w,
            &run_id,
            crate::imp::context::RunSurface::SlackSession,
            start_context,
            rx,
            SESSION_IDLE_SECS,
        )
        .await
        {
            eprintln!("slack session error: {e}");
        }
    });
}