heatsync-tui 0.4.0

//! blocking HTTP against the public HeatSync API (no auth). emote sets + image
//! bytes for the TUI, and the archive/corpus reads for the CLI subcommands.

use std::io::Read;
use std::sync::OnceLock;
use std::time::Duration;

use heatsync_core::emote::{EmoteSet, EmoteSetResponse};
use heatsync_core::Platform;
use serde::Deserialize;

pub const BASE: &str = "https://heatsync.org";
const MAX_IMAGE_BYTES: u64 = 8 * 1024 * 1024;

fn platform_q(p: Platform) -> &'static str {
    match p {
        Platform::Twitch => "twitch",
        Platform::Kick => "kick",
        Platform::Youtube => "youtube",
    }
}

/// one shared agent for the whole process — its connection pool reuses TLS
/// sessions, so a burst of 30 emote fetches from one CDN pays one handshake,
/// not thirty. thread-safe; the loader pool and emote-set threads all share it.
fn agent() -> &'static ureq::Agent {
    static AGENT: OnceLock<ureq::Agent> = OnceLock::new();
    AGENT.get_or_init(|| {
        ureq::AgentBuilder::new()
            .timeout(Duration::from_secs(12))
            // the archive search endpoint gates non-browser UAs (anti-scrape); we're
            // first-party, so identify as heatsync-tui behind a Mozilla token so the
            // guard passes. proper fix later: server-side allowlist this UA.
            .user_agent("Mozilla/5.0 (heatsync-tui/0.1; +https://heatsync.org)")
            .build()
    })
}

/// the channel's current pin (`/api/pin`), if it has one. twitch + kick only.
pub fn pin(platform: Platform, channel: &str) -> Option<heatsync_core::Pin> {
    if platform == Platform::Youtube {
        return None;
    }
    let v: serde_json::Value = agent()
        .get(&format!("{BASE}/api/pin"))
        .query("platform", platform_q(platform))
        .query("channel", &channel.to_lowercase())
        .call()
        .ok()?
        .into_json()
        .ok()?;
    heatsync_core::proto::pin_from(v.get("pin")?)
}

/// the pooled global emote set (/api/emotes): twitch globals + event-unlocked
/// emotes, and the bttv/ffz/7tv global sets — merged and deduped server-side.
pub fn global_emotes() -> Option<EmoteSet> {
    let resp: EmoteSetResponse = agent()
        .get(&format!("{BASE}/api/emotes"))
        .call()
        .ok()?
        .into_json()
        .ok()?;
    Some(EmoteSet::from_global_list(resp.emotes))
}

/// `/api/paints` for a batch of paint-space ids (<=50): heatsync paint specs
/// and picked name colors. `None` on any failure so the caller retries.
pub fn paints(ids: &[String]) -> Option<serde_json::Value> {
    let url = format!("{BASE}/api/paints?ids={}", ids.join(","));
    agent().get(&url).call().ok()?.into_json().ok()
}

/// `/api/cosmetics/batch`: 7tv paints for twitch ids and `kick_<id>`s (<=50).
pub fn cosmetics(ids: &[String]) -> Option<serde_json::Value> {
    let url = format!("{BASE}/api/cosmetics/batch?ids={}", ids.join(","));
    agent().get(&url).call().ok()?.into_json().ok()
}

/// channel emote set (the server sends each name's winner last: ffz > 7tv > bttv)
/// plus the streamer's own heatsync emotes, which win their names in their own
/// channel. fetched side by side; a failed owner lookup just leaves them out.
pub fn emote_set(channel: &str, platform: Platform) -> Option<EmoteSet> {
    let owner = std::thread::scope(|s| {
        let h = s.spawn(|| owner_emotes(channel, platform));
        let chan = channel_emotes(channel, platform);
        (chan, h.join().unwrap_or_default())
    });
    let (chan, owner) = owner;
    if chan.is_none() && owner.is_empty() {
        return None;
    }
    Some(EmoteSet::from_list(
        chan.unwrap_or_default().into_iter().chain(owner),
    ))
}

fn channel_emotes(channel: &str, platform: Platform) -> Option<Vec<heatsync_core::emote::Emote>> {
    let url = format!(
        "{BASE}/api/channel/{channel}/emotes?platform={}",
        platform_q(platform)
    );
    let resp: EmoteSetResponse = agent().get(&url).call().ok()?.into_json().ok()?;
    Some(resp.emotes)
}

/// `/api/emotes/user/<channel>?platform=` — the same login is a different
/// person on each platform, so the route is always told which one. youtube
/// tabs are named by video id, not a channel: no owner there.
fn owner_emotes(channel: &str, platform: Platform) -> Vec<heatsync_core::emote::Emote> {
    let plain = !channel.is_empty()
        && channel
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-' || b == b'.');
    if platform == Platform::Youtube || !plain {
        return Vec::new();
    }
    let url = format!(
        "{BASE}/api/emotes/user/{channel}?platform={}",
        platform_q(platform)
    );
    let v: serde_json::Value = match agent()
        .get(&url)
        .call()
        .ok()
        .and_then(|r| r.into_json().ok())
    {
        Some(v) => v,
        None => return Vec::new(),
    };
    heatsync_core::emote::owner_list(&v, BASE)
}

/// raw image bytes for an emote, capped so a hostile CDN can't balloon memory.
pub fn image_bytes(url: &str) -> Option<Vec<u8>> {
    let mut buf = Vec::new();
    agent()
        .get(url)
        .call()
        .ok()?
        .into_reader()
        .take(MAX_IMAGE_BYTES)
        .read_to_end(&mut buf)
        .ok()?;
    Some(buf)
}

// ---- corpus / archive reads (CLI) ----------------------------------------

/// one archived chat row (subset of the archive API shape).
#[derive(Debug, Deserialize)]
pub struct ArchiveMsg {
    #[serde(default)]
    pub platform: String,
    #[serde(default)]
    pub channel: String,
    #[serde(default)]
    pub username: String,
    #[serde(default)]
    pub display_name: Option<String>,
    #[serde(default)]
    pub message: String,
    #[serde(default)]
    pub timestamp: String,
    /// the platform's own message id (null on rows archived before it was kept) —
    /// what lets history and the live line of one message be told as one.
    #[serde(default)]
    pub message_id: Option<String>,
    /// `[{name, version}]` — same shape the kick ws frames use.
    #[serde(default)]
    pub badges: Option<serde_json::Value>,
}

#[derive(Debug, Deserialize)]
pub struct ArchivePage {
    #[serde(default)]
    pub results: Vec<ArchiveMsg>,
}

/// full-text search the archive.
pub fn search(q: &str, channel: Option<&str>, limit: u32) -> Option<ArchivePage> {
    let mut url = format!("{BASE}/api/archive/search?limit={limit}&q={}", urlencode(q));
    if let Some(c) = channel {
        url.push_str(&format!("&channel={}", urlencode(c)));
    }
    agent().get(&url).call().ok()?.into_json().ok()
}

/// the site feed's latest `want` rows, oldest first — the feed tab's backlog.
/// anonymous works; rows pass through the same parser the live socket uses, so
/// they are sanitized at the same trust boundary.
pub fn feed_recent(want: usize) -> Vec<heatsync_core::proto::SpecialLine> {
    let url = format!("{BASE}/api/messages?sort=time&limit={want}");
    let Some(v) = agent()
        .get(&url)
        .call()
        .ok()
        .and_then(|r| r.into_json::<serde_json::Value>().ok())
    else {
        return Vec::new();
    };
    let mut rows: Vec<_> = v
        .get("messages")
        .and_then(serde_json::Value::as_array)
        .map(|a| {
            a.iter()
                .filter_map(heatsync_core::proto::feed_row)
                .collect()
        })
        .unwrap_or_default();
    // the api answers newest first; the tab reads oldest first.
    rows.reverse();
    rows
}

/// a channel's log for a UTC date (YYYY-MM-DD).
pub fn channel_log(channel: &str, from: &str, to: &str, limit: u32) -> Option<ArchivePage> {
    let url =
        format!("{BASE}/api/archive/channel/{channel}/messages?from={from}&to={to}&limit={limit}");
    agent().get(&url).call().ok()?.into_json().ok()
}

/// the last messages a channel saw, oldest first — scrollback seed for a fresh
/// tab. the archive reads ascending, so we ask for a recent window and keep the
/// tail, widening the window when a quiet channel comes back thin. rows are
/// sanitized here — archive content crosses the same trust boundary live chat
/// does.
pub fn recent(channel: &str, platform: Platform, want: usize) -> Vec<heatsync_core::Message> {
    let mut best: Vec<ArchiveMsg> = Vec::new();
    for mins in [10u64, 60, 480] {
        let from = iso_from_unix(unix_now().saturating_sub(mins * 60));
        let to = iso_from_unix(unix_now());
        let Some(page) = channel_log(channel, &from, &to, 200) else {
            continue;
        };
        let rows: Vec<ArchiveMsg> = page
            .results
            .into_iter()
            .filter(|r| r.platform == platform_q(platform))
            .collect();
        let enough = rows.len() >= want;
        best = rows;
        if enough {
            break;
        }
    }
    let skip = best.len().saturating_sub(want);
    best.drain(..skip);
    best.into_iter()
        .map(|r| archived(platform, r))
        .filter(|m| !m.user.is_empty() || !m.text.is_empty())
        .collect()
}

/// one archive row as a scrollback line. it keeps the platform's own message id
/// (when the row has one): that id is how history and the live copy of one
/// message are told apart from two messages with the same words.
fn archived(platform: Platform, r: ArchiveMsg) -> heatsync_core::Message {
    use heatsync_core::{proto::badges_from, sanitize};
    let user = sanitize::clean(r.display_name.as_deref().unwrap_or(&r.username));
    heatsync_core::Message {
        platform,
        user: if user.is_empty() {
            sanitize::clean(&r.username)
        } else {
            user
        },
        text: sanitize::clean(&r.message),
        color: None,
        badges: badges_from(r.badges.as_ref()),
        id: r.message_id.filter(|i| !i.is_empty() && i.len() <= 128),
        gone: None,
        reply_to: None,
        note: None,
        // its own time: stamps show, and a stream-start row can tell
        // whether this history reaches back that far.
        extra: heatsync_core::Extra {
            ts: heatsync_core::iso::parse(&r.timestamp).map_or(0, |t| t.max(0) as u64),
            ..Default::default()
        },
        emotes: None,
        tags: Vec::new(),
        heat: 0.0,
        uid: None,
    }
}

/// twitch's public web gql, the same one the heatsync desktop app's /status
/// asks: no login, 3s at most, answer capped.
const GQL: &str = "https://gql.twitch.tv/gql";
const GQL_CLIENT_ID: &str = "kimne78kx3ncx6brgo4mv6wki5h1ko";
const MAX_GQL: u64 = 1 << 20;

/// (stream start in unix seconds, game) if `login` is live on twitch right now.
/// `None` when offline, unknown, or anything at all goes wrong.
pub fn twitch_live(login: &str) -> Option<(u64, String)> {
    let ok = !login.is_empty()
        && login.len() <= 25
        && login
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b == b'_');
    if !ok {
        return None;
    }
    let body = serde_json::json!({
        "query": "query($login:String!){user(login:$login){displayName stream{type viewersCount createdAt game{name}} broadcastSettings{title language}}}",
        "variables": { "login": login.to_lowercase() },
    });
    let resp = agent()
        .post(GQL)
        .set("Client-Id", GQL_CLIENT_ID)
        .timeout(Duration::from_secs(3))
        .send_json(body)
        .ok()?;
    let v: serde_json::Value = serde_json::from_reader(resp.into_reader().take(MAX_GQL)).ok()?;
    parse_live(&v)
}

/// gql answer → (start, game) of a live stream.
fn parse_live(v: &serde_json::Value) -> Option<(u64, String)> {
    let s = v.pointer("/data/user/stream").filter(|s| s.is_object())?;
    if s.get("type").and_then(|t| t.as_str()) != Some("live") {
        return None;
    }
    let at = s.get("createdAt").and_then(|t| t.as_str())?;
    let start = heatsync_core::iso::parse(at).filter(|t| *t > 0)? as u64;
    let game = s
        .pointer("/game/name")
        .and_then(|g| g.as_str())
        .map(|g| heatsync_core::sanitize::clean(g).chars().take(80).collect())
        .unwrap_or_default();
    Some((start, game))
}

fn unix_now() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// unix seconds → `YYYY-MM-DDThh:mm:ssZ`. hand-rolled civil-date conversion
/// (Hinnant's algorithm) — fifteen tested lines beat a date dependency.
fn iso_from_unix(t: u64) -> String {
    let (h, mi, s) = ((t % 86400) / 3600, (t % 3600) / 60, t % 60);
    let z = (t / 86400) as i64 + 719_468;
    let era = z.div_euclid(146_097);
    let doe = z.rem_euclid(146_097);
    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = yoe + era * 400 + i64::from(m <= 2);
    format!("{y:04}-{m:02}-{d:02}T{h:02}:{mi:02}:{s:02}Z")
}

/// hottest posts across the platform (these carry real heat).
#[derive(Debug, Deserialize)]
pub struct HotMsg {
    #[serde(default)]
    pub content: String,
    #[serde(default)]
    pub heat: f64,
    #[serde(default)]
    pub max_heat: f64,
    #[serde(default)]
    pub display_name: Option<String>,
    #[serde(default)]
    pub username: String,
    #[serde(default)]
    pub subject: Option<String>,
}

#[derive(Debug, Deserialize)]
pub struct HotPage {
    #[serde(default)]
    pub messages: Vec<HotMsg>,
    /// nothing ranked hot, so the server sent its newest posts instead (heat 0)
    #[serde(default)]
    pub fallback: bool,
}

impl HotPage {
    /// the posts that are actually hot: none when the server fell back to newest
    pub fn hot(&self) -> &[HotMsg] {
        if self.fallback {
            &[]
        } else {
            &self.messages
        }
    }
}

pub fn hot(limit: u32, hours: u32) -> Option<HotPage> {
    let url = format!("{BASE}/api/messages/hot?limit={limit}&hours={hours}");
    agent().get(&url).call().ok()?.into_json().ok()
}

// ---- live lists (the channel picker) ---------------------------------------

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct LiveStream {
    #[serde(default)]
    pub username: String,
    #[serde(default)]
    pub platform: String,
    #[serde(default)]
    pub title: String,
    #[serde(default)]
    pub category: String,
    #[serde(default)]
    pub viewer_count: u64,
    #[serde(default)]
    pub video_id: Option<String>,
    #[serde(default)]
    pub platform_usernames: std::collections::HashMap<String, String>,
}

#[derive(Debug, Deserialize)]
struct LivePage {
    #[serde(default)]
    streams: Vec<LiveStream>,
}

/// `/api/live/top` — the busiest live streams across twitch/kick/youtube.
pub fn live_top(limit: u32) -> Option<Vec<LiveStream>> {
    let url = format!("{BASE}/api/live/top?limit={limit}&sort=viewers");
    let page: LivePage = agent().get(&url).call().ok()?.into_json().ok()?;
    Some(page.streams)
}

#[derive(Debug, Deserialize)]
struct LiveStatus {
    #[serde(default)]
    live: Vec<String>,
    #[serde(default, rename = "kickLive")]
    kick_live: Vec<String>,
}

/// `/api/platform/live-status` — which of these twitch / kick channels are live
/// right now (the server caps each list at 20). names only, no viewer counts.
pub fn live_status(twitch: &[String], kick: &[String]) -> Option<(Vec<String>, Vec<String>)> {
    if twitch.is_empty() && kick.is_empty() {
        return Some((Vec::new(), Vec::new()));
    }
    let url = format!(
        "{BASE}/api/platform/live-status?channels={}&kick_channels={}",
        urlencode(&twitch.join(",")),
        urlencode(&kick.join(","))
    );
    let st: LiveStatus = agent().get(&url).call().ok()?.into_json().ok()?;
    Some((st.live, st.kick_live))
}

// ---- admin status (CLI, mellen-only) --------------------------------------

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct HealthData {
    #[serde(default)]
    pub messages_per_minute: f64,
    #[serde(default)]
    pub postgres_status: String,
    #[serde(default)]
    pub redis_status: String,
    #[serde(default)]
    pub ws_connections: i64,
    #[serde(default)]
    pub cpu_usage: f64,
    #[serde(default)]
    pub ram_usage: f64,
    #[serde(default)]
    pub disk_usage: f64,
    #[serde(default)]
    pub error_rate: f64,
}

#[derive(Debug, Deserialize)]
pub struct CountPage {
    #[serde(default)]
    pub total: i64,
}

#[derive(Debug, Deserialize)]
pub struct ReportsPage {
    #[serde(default)]
    pub reports: Vec<serde_json::Value>,
}

#[derive(Debug, Deserialize)]
pub struct NcmecPage {
    #[serde(default)]
    pub count: i64,
}

fn admin_get<T: for<'de> Deserialize<'de>>(path: &str, token: &str) -> Option<T> {
    agent()
        .get(&format!("{BASE}{path}"))
        .set("Authorization", &format!("Bearer {token}"))
        .call()
        .ok()?
        .into_json()
        .ok()
}

pub fn admin_health(token: &str) -> Option<HealthData> {
    admin_get("/api/admin/health", token)
}

/// is the admin token itself refused (401/403)? false for a down server or a
/// good token — only a clear "no" from heatsync counts.
pub fn admin_token_rejected(base: &str, token: &str) -> bool {
    matches!(
        agent()
            .get(&format!("{base}/api/admin/health"))
            .set("Authorization", &format!("Bearer {token}"))
            .call(),
        Err(ureq::Error::Status(401 | 403, _))
    )
}

/// moderation-queue depth (items awaiting review).
pub fn admin_mod_queue_total(token: &str) -> Option<i64> {
    admin_get::<CountPage>("/api/admin/moderation-queue?limit=1", token).map(|p| p.total)
}

/// pending user reports.
pub fn admin_pending_reports(token: &str) -> Option<i64> {
    admin_get::<ReportsPage>("/api/admin/reports?status=pending", token)
        .map(|p| p.reports.len() as i64)
}

/// NCMEC reports stuck in a failed/manual state — never routine, always worth surfacing.
pub fn admin_ncmec_backlog(token: &str) -> Option<i64> {
    admin_get::<NcmecPage>("/api/admin/ncmec-reports", token).map(|p| p.count)
}

/// minimal percent-encoding for query values (alnum + a few safe chars pass).
fn urlencode(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for b in s.bytes() {
        match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                out.push(b as char)
            }
            _ => out.push_str(&format!("%{b:02X}")),
        }
    }
    out
}

#[cfg(test)]
mod tests {
    #[test]
    fn newest_posts_sent_in_place_of_hot_ones_are_not_called_hot() {
        let page = |fallback: bool| -> super::HotPage {
            serde_json::from_value(serde_json::json!({"fallback": fallback,
                "messages": [{"content": "hi", "heat": 0, "max_heat": 0, "username": "a"}]}))
            .unwrap()
        };
        assert!(page(true).hot().is_empty());
        assert_eq!(page(false).hot().len(), 1);
        let old: super::HotPage = serde_json::from_value(serde_json::json!({"messages": [
            {"content": "hi", "heat": 9, "max_heat": 9, "username": "a"}]}))
        .unwrap();
        assert_eq!(
            old.hot().len(),
            1,
            "a server that never says fallback is read as before"
        );
    }

    #[test]
    fn the_gql_answer_gives_a_live_streams_start_and_game() {
        let live = serde_json::json!({"data": {"user": {"stream": {"type": "live",
            "createdAt": "2026-11-03T10:00:00Z", "game": {"name": "Just \u{7}Chatting"}}}}});
        assert_eq!(
            parse_live(&live),
            Some((1_793_700_000, "Just Chatting".into()))
        );
        let no_game = serde_json::json!({"data": {"user": {"stream": {"type": "live",
            "createdAt": "2026-11-03T10:00:00Z", "game": null}}}});
        assert_eq!(parse_live(&no_game).map(|l| l.1), Some(String::new()));
        for bad in [
            serde_json::json!({"data": {"user": {"stream": null}}}),
            serde_json::json!({"data": {"user": null}}),
            serde_json::json!({"data": {"user": {"stream": {"type": "rerun", "createdAt": "2026-11-03T10:00:00Z"}}}}),
            serde_json::json!({"data": {"user": {"stream": {"type": "live", "createdAt": "junk"}}}}),
            serde_json::json!("junk"),
        ] {
            assert_eq!(parse_live(&bad), None);
        }
        assert_eq!(
            twitch_live("a b"),
            None,
            "a bad login never leaves the machine"
        );
    }

    use super::{archived, iso_from_unix, parse_live, twitch_live, ArchiveMsg};
    use heatsync_core::Platform;

    #[test]
    fn an_archive_row_keeps_the_platforms_message_id_for_the_dedup() {
        let row = |extra: &str| -> ArchiveMsg {
            serde_json::from_str(&format!(
                r#"{{"platform":"kick","username":"a","message":"hi","timestamp":"2026-11-03T10:00:00Z"{extra}}}"#
            ))
            .unwrap()
        };
        let id = |extra: &str| archived(Platform::Kick, row(extra)).id;
        assert_eq!(id(r#","message_id":"abc-1""#).as_deref(), Some("abc-1"));
        // legacy rows (null / missing / blank / absurd) have none: never dedup-able, never dropped
        assert_eq!(id(r#","message_id":null"#), None);
        assert_eq!(id(""), None);
        assert_eq!(id(r#","message_id":"""#), None);
        let long = format!(r#","message_id":"{}""#, "x".repeat(129));
        assert_eq!(id(&long), None);
    }

    #[test]
    fn iso_conversion_hits_known_instants() {
        assert_eq!(iso_from_unix(0), "1970-01-01T00:00:00Z");
        assert_eq!(iso_from_unix(1_735_689_600), "2025-01-01T00:00:00Z");
        assert_eq!(iso_from_unix(951_782_400), "2000-02-29T00:00:00Z"); // leap day
        assert_eq!(iso_from_unix(1_787_697_045), "2026-08-25T22:30:45Z");
    }
}