//! The human at a terminal (spec §19): `spectra say`, `spectra inbox

//! [--follow]`, `spectra search`, and `spectra api` for anything else. Each

//! signs with the owner's own key: through the daemon when it is serving,

//! else straight into the ledger.

use super::identity::owner_home;

use super::{ui, Ctx};

use anyhow::{bail, Context, Result};

use serde_json::{json, Value};

use spectra_core::conversation::{self as conv, Author, Post, Posted, Signer};

use spectra_core::keys::KeyPair;

use spectra_core::ledger::Ledger;

use spectra_core::principals::Principals;

use spectra_mcp::client::Client;

use std::path::PathBuf;

async fn owner_client(ctx: &Ctx, home: &std::path::Path) -> Result<Option<Client>> {

    ctx.owner_client(home).await

}

async fn api(c: &Client, method: &str, path: &str, body: Option<Value>) -> Result<Value> {

    super::call(c, method, path, body).await

}

use super::encode;

pub async fn say(

    ctx: &Ctx,

    channel: Option<String>,

    reply_to: Option<String>,

    body: String,

    owner_home_flag: Option<PathBuf>,

    as_json: bool,

) -> Result<i32> {

    let home = owner_home(ctx, owner_home_flag);

    let (id, where_) = match owner_client(ctx, &home).await? {

        Some(c) => {

            let ch = channel.clone().unwrap_or_else(|| "general".into());

            let v = api(

                &c,

                "POST",

                &format!("/api/v1/channels/{}/messages", encode(ch.trim_start_matches('#'))),

                Some(json!({ "body": body, "reply_to": reply_to })),

            )

            .await?;

            if v["duplicate"] == true {

                bail!("you said exactly this in the last minute ({})", v["id"].as_str().unwrap_or(""));

            }

            (v["id"].as_str().unwrap_or("").to_string(), v["channel"].as_str().unwrap_or("").to_string())

        }

        None => {

            let ledger = ctx.writer().await?;

            let owner = KeyPair::load_home(&home)?;

            let principal = Principals::load(&ledger)

                .await?

                .principal(&owner.key, spectra_core::ids::now_ms())

                .map_err(anyhow::Error::msg)

                .context("run `spectra identity refresh` first")?;

            let author = Author { principal, signer: Signer::Cli(&owner) };

            let post = Post {

                channel: if reply_to.is_some() { None } else { Some(channel.unwrap_or_else(|| "general".into())) },

                body,

                reply_to,

                ..Default::default()

            };

            let posted = conv::post(&ledger, &author, post, spectra_core::ids::now_ms()).await?;

            ledger.checkpoint().await?;

            match posted {

                Posted::Message { id, channel } => (id, channel),

                Posted::Duplicate { id } => bail!("you said exactly this in the last minute ({id})"),

            }

        }

    };

    if as_json {

        println!("{}", json!({ "id": id, "channel": where_ }));

    } else {

        println!("{} {}{}{}", ui::done("Posted"), ui::subject(&where_), ui::sep(), ui::id(&id));

    }

    Ok(0)

}

fn print_item(i: &Value) {

    let kind = i["kind"].as_str().unwrap_or("");

    let mark = if kind == "dm" { ui::warn(ui::glyph("✉", "@")) } else { ui::ok(ui::glyph("●", "*")) };

    let from = i["from_display"].as_str().or(i["from"].as_str()).unwrap_or("");

    let at = i["at_ms"].as_u64().map(ui::ago).unwrap_or_default();

    println!(

        "{mark} {} {}{}{}{}{}",

        ui::dim(kind),

        ui::author(from),

        ui::sep(),

        i["channel"].as_str().unwrap_or(""),

        ui::sep(),

        ui::dim(at)

    );

    println!("  {}", i["summary"].as_str().unwrap_or(""));

    println!("  {}", ui::dim(format!("reply: spectra say --reply-to {} \"…\"", i["id"].as_str().unwrap_or(""))));

}

/// The owner's unread items: from the daemon when one serves, else the ledger.

async fn inbox_items(ctx: &Ctx, client: Option<&Client>) -> Result<Vec<Value>> {

    if let Some(c) = client {

        return Ok(api(c, "GET", "/api/v1/inbox", None).await?.as_array().cloned().unwrap_or_default());

    }

    let ledger = Ledger::open_read_only(&ctx.db()?).await?;

    let mut items = Vec::new();

    for i in conv::inbox(&ledger, "human", None).await? {

        let from_display = conv::display_name(&ledger, &i.from).await;

        let mut v = serde_json::to_value(&i)?;

        v["from_display"] = from_display.into();

        items.push(v);

    }

    Ok(items)

}

pub async fn inbox(ctx: &Ctx, follow: bool, owner_home_flag: Option<PathBuf>, as_json: bool) -> Result<i32> {

    let home = owner_home(ctx, owner_home_flag);

    let client = owner_client(ctx, &home).await?;

    // Following: subscribe first, and read the inbox once the server says

    // the subscription holds, so nothing posted in between is missed.

    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<bool>();

    let watcher = if follow {

        let stream_client = owner_client(ctx, &home).await?.context("--follow needs `spectra serve` running")?;

        let w = tokio::spawn(async move {

            stream_client

                .stream("/api/v1/events?follow=1&kind=message.posted", |l| {

                    if l.starts_with(':') {

                        let _ = tx.send(false);

                    } else if l.starts_with("data:") {

                        let _ = tx.send(true);

                    }

                    true

                })

                .await

        });

        match tokio::time::timeout(std::time::Duration::from_secs(10), rx.recv()).await {

            Ok(Some(_)) => {}

            _ => bail!("the server did not start the stream of new messages"),

        }

        Some(w)

    } else {

        None

    };

    let items = inbox_items(ctx, client.as_ref()).await?;

    if as_json && !follow {

        println!("{}", Value::Array(items));

        return Ok(0);

    }

    if items.is_empty() && !follow {

        println!("{} {}", ui::note(), ui::dim("nothing unread"));

    }

    let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();

    for i in &items {

        seen.insert(i["id"].as_str().unwrap_or("").to_string());

        if as_json {

            println!("{i}");

        } else {

            print_item(i);

        }

    }

    let Some(watcher) = watcher else { return Ok(0) };

    let client = client.context("--follow needs `spectra serve` running")?;

    if !as_json {

        println!("{} {}", ui::note(), ui::dim("following; Ctrl-C stops"));

    }

    // Every new message is a cue to look at the inbox again.

    while let Some(changed) = rx.recv().await {

        if !changed {

            continue;

        }

        for v in inbox_items(ctx, Some(&client)).await? {

            if seen.insert(v["id"].as_str().unwrap_or("").to_string()) {

                if as_json {

                    println!("{v}");

                } else {

                    print_item(&v);

                }

            }

        }

    }

    watcher.await??;

    Ok(0)

}

pub async fn search(

    ctx: &Ctx,

    query: &str,

    channel: Option<String>,

    author: Option<String>,

    since: Option<String>,

    limit: u32,

    as_json: bool,

) -> Result<i32> {

    let hits: Vec<Value> = match owner_client(ctx, &owner_home(ctx, None)).await? {

        Some(c) => {

            let mut q = format!("/api/v1/search?q={}&limit={limit}", encode(query));

            for (k, v) in [("channel", &channel), ("author", &author), ("since", &since)] {

                if let Some(v) = v {

                    q.push_str(&format!("&{k}={}", encode(v)));

                }

            }

            api(&c, "GET", &q, None).await?.as_array().cloned().unwrap_or_default()

        }

        None => {

            let ledger = Ledger::open_read_only(&ctx.db()?).await?;

            let owner = Principals::load(&ledger).await?;

            let me = owner.owner().context("no owner is recorded")?;

            let principal = spectra_core::principals::Principal::Owner { key: me.hash(), name: me.name.clone() };

            let since = since.as_deref().and_then(|s| spectra_mcp::tools::parse_since(s, spectra_core::ids::now_ms()));

            let mut out = Vec::new();

            for h in conv::search(&ledger, &principal, query, channel.as_deref(), author.as_deref(), since, limit).await? {

                let mut v = serde_json::to_value(&h)?;

                v["author_display"] = conv::display_name(&ledger, &h.author).await.into();

                out.push(v);

            }

            out

        }

    };

    if as_json {

        println!("{}", Value::Array(hits));

        return Ok(0);

    }

    if hits.is_empty() {

        println!("{} {}", ui::note(), ui::dim("no matches"));

    }

    for h in hits {

        let id = h["id"].as_str().unwrap_or("");

        println!(

            "{} {}{}{}{}{}",

            ui::id(&id[..12.min(id.len())]),

            ui::author(h["author_display"].as_str().or(h["author"].as_str()).unwrap_or("")),

            ui::sep(),

            h["channel"].as_str().unwrap_or(""),

            ui::sep(),

            ui::dim(ui::ago(h["at_ms"].as_u64().unwrap_or(0)))

        );

        println!("  {}", h["snippet"].as_str().unwrap_or(""));

    }

    Ok(0)

}

/// `spectra api METHOD PATH [-d JSON]`: any API call, signed as the owner,

/// JSON out (the signing proxy for scripts and the editor extensions).

pub async fn raw(ctx: &Ctx, method: &str, path: &str, data: Option<String>, owner_home_flag: Option<PathBuf>) -> Result<i32> {

    let home = owner_home(ctx, owner_home_flag);

    let c = owner_client(ctx, &home).await?.context("`spectra api` needs `spectra serve` running, or `spectra connect`")?;

    let r = c.send(&method.to_uppercase(), path, &[], data.map(String::into_bytes).unwrap_or_default()).await?;

    println!("{}", String::from_utf8_lossy(&r.body));

    Ok(if r.status < 400 { 0 } else { 2 })

}