//! Conversation (spec §10): channels, messages, threads, reactions, each

//! principal's inbox, and search (§13.1). The MCP tools, the API and the

//! CLI all go through here, so the rules are the same whoever asks.

//!

//! A message is redacted by its author or the owner: Spectra stops showing

//! it everywhere, while the ledger keeps the original for audit until a

//! prune removes it (§9.1).

use crate::event::{Auth, Event, NewEvent};

use crate::keys::KeyPair;

use crate::ledger::Ledger;

use crate::principals::{signed_event, Principal};

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

use serde::Serialize;

use serde_json::{json, Value as Json};

use sqlx::SqliteConnection;

/// The most a body or a payload may be (spec §10.3).

pub const MAX_BODY: usize = 64 * 1024;

/// A message identical to one its author posted in the same channel within

/// this long is dropped (spec §20.4).

pub const DUPLICATE_MS: u64 = 60_000;

/// The channels every Spectra has (spec §10.1).

pub const DEFAULT_CHANNELS: &[(&str, &str, &str)] = &[

    ("general", "shared", "Everyone: agents and the human"),

    ("humans", "shared", "Where agents ask the human"),

    ("system", "system", "Guard rails, schedules and hooks; read-only for agents"),

];

/// Message kinds (spec §10.3).

pub const KINDS: &[&str] = &["chat", "request", "answer", "handoff", "status", "decision", "system"];

pub const REACTIONS: &[&str] = &["ack", "done", "eyes", "no"];

/// Who is acting, and how their events are signed.

pub struct Author<'a> {

    pub principal: Principal,

    pub signer: Signer<'a>,

}

/// The CLI signs each event over its payload; the daemon carries the proof

/// of the request that caused it.

pub enum Signer<'a> {

    Cli(&'a KeyPair),

    Request(Auth),

    System,

}

impl Author<'_> {

    fn event(&self, kind: &str, payload: Json) -> NewEvent {

        let actor = self.principal.actor();

        match &self.signer {

            Signer::Cli(pair) => signed_event(pair, &actor, kind, payload),

            Signer::Request(auth) => NewEvent::system(kind, payload).by(actor, auth.clone()),

            Signer::System => NewEvent::system(kind, payload),

        }

    }

}

#[derive(Clone, Debug, Serialize, sqlx::FromRow)]

pub struct Channel {

    pub id: String,

    pub name: String,

    pub kind: String,

    pub project: Option<String>,

    pub purpose: Option<String>,

    pub members: String,

}

impl Channel {

    pub fn members(&self) -> Vec<String> {

        serde_json::from_str(&self.members).unwrap_or_default()

    }

    /// `#general`, `#oxygen`, or `@a+b` for a direct channel.

    pub fn display(&self) -> String {

        match self.name.strip_prefix("dm:") {

            Some(pair) => format!("@{pair}"),

            None => format!("#{}", self.name),

        }

    }

}

#[derive(Clone, Debug, Serialize, sqlx::FromRow)]

pub struct Message {

    pub id: String,

    pub channel: String,

    pub reply_to: Option<String>,

    pub author: String,

    pub kind: String,

    pub body: String,

    pub payload: Option<String>,

    pub mentions: String,

    pub refs: String,

    pub at_ms: i64,

    pub seq: i64,

    pub replies: i64,

    pub redacted: Option<String>,

}

/// What a post asks for.

#[derive(Clone, Debug, Default)]

pub struct Post {

    pub channel: Option<String>,

    pub body: String,

    pub reply_to: Option<String>,

    pub kind: Option<String>,

    pub payload: Option<Json>,

    pub mentions: Vec<String>,

    pub refs: Vec<String>,

}

/// The outcome of a post: the message, or why it was dropped.

#[derive(Clone, Debug, Serialize)]

pub enum Posted {

    Message { id: String, channel: String },

    Duplicate { id: String },

}

/// Make the default channels (at `spectra init`).

pub fn default_channel_events() -> Vec<NewEvent> {

    DEFAULT_CHANNELS.iter().map(|(name, kind, purpose)| channel_event(name, kind, None, Some(purpose), &[])).collect()

}

/// A project's channel, made when the project is added.

pub fn project_channel_event(project: &str) -> NewEvent {

    channel_event(project, "project", Some(project), Some(&format!("The {project} project")), &[])

}

fn channel_event(name: &str, kind: &str, project: Option<&str>, purpose: Option<&str>, members: &[String]) -> NewEvent {

    let id = crate::ids::ulid();

    NewEvent::system(

        "channel.created",

        json!({ "id": id, "name": name, "kind": kind, "project": project, "purpose": purpose, "members": members }),

    )

    .subject(id)

}

/// The name of the direct channel between two actors.

pub fn direct_name(a: &str, b: &str) -> String {

    let short = |x: &str| x.strip_prefix("agent:").unwrap_or(x).to_string();

    let mut v = [short(a), short(b)];

    v.sort();

    format!("dm:{}+{}", v[0], v[1])

}

/// Find a channel by `#name`, `name`, `@a+b` or id.

pub async fn channel(ledger: &Ledger, name: &str) -> Result<Option<Channel>> {

    let n = name.strip_prefix('#').unwrap_or(name);

    let n = match n.strip_prefix('@') {

        Some(pair) => format!("dm:{pair}"),

        None => n.to_string(),

    };

    Ok(sqlx::query_as("SELECT id, name, kind, project, purpose, members FROM channels WHERE name = ? OR id = ?")

        .bind(&n)

        .bind(&n)

        .fetch_optional(ledger.pool())

        .await?)

}

pub async fn channels(ledger: &Ledger) -> Result<Vec<Channel>> {

    Ok(sqlx::query_as("SELECT id, name, kind, project, purpose, members FROM channels ORDER BY kind = 'direct', name")

        .fetch_all(ledger.pool())

        .await?)

}

/// Whether `who` may read `ch`: direct channels are for their members.

pub fn can_read(who: &Principal, ch: &Channel) -> bool {

    ch.kind != "direct" || matches!(who, Principal::Owner { .. }) || ch.members().contains(&who.actor())

}

/// Whether `who` may post in `ch` (spec §10.1, §8.4).

pub fn can_post(who: &Principal, ch: &Channel) -> Result<(), String> {

    match (who, ch.kind.as_str()) {

        (Principal::Instance { .. }, _) => Ok(()),

        (_, "system") => Err("#system is Spectra's: guard rails, schedules and hooks post there".into()),

        (Principal::Owner { .. }, _) => Ok(()),

        (_, "direct") if !ch.members().contains(&who.actor()) => Err(format!("{} is a direct channel you are not in", ch.display())),

        (Principal::Agent { project, scopes, .. }, "project") => {

            if ch.project.as_deref() == Some(project) || scopes.iter().any(|s| s == "projects:all") {

                Ok(())

            } else {

                Err(format!("{} belongs to another project; post in #{project}, #general, or ask for projects:all", ch.display()))

            }

        }

        _ => Ok(()),

    }

}

/// Post a message, checking everything first (spec §10.3).

pub async fn post(ledger: &Ledger, who: &Author<'_>, mut p: Post, now_ms: u64) -> Result<Posted> {

    if p.body.trim().is_empty() {

        bail!("a message needs a body");

    }

    if p.body.len() > MAX_BODY {

        bail!("a message's body is at most 64 KiB; this is {} KiB", p.body.len() / 1024);

    }

    if p.payload.as_ref().is_some_and(|v| v.to_string().len() > MAX_BODY) {

        bail!("a message's payload is at most 64 KiB");

    }

    let kind = p.kind.take().unwrap_or_else(|| "chat".into());

    if !KINDS.contains(&kind.as_str()) {

        bail!("a message's kind is one of {}", KINDS.join(", "));

    }

    let ch = match (&p.channel, &p.reply_to, &who.principal) {

        (Some(c), _, _) => channel(ledger, c).await?.with_context(|| format!("no channel {c}"))?,

        (None, Some(r), _) => {

            let root = message(ledger, r).await?.with_context(|| format!("no message {r}"))?;

            channel(ledger, &root.channel).await?.context("the thread's channel")?

        }

        (None, None, Principal::Agent { project, .. }) => {

            channel(ledger, project).await?.with_context(|| format!("no channel for project {project}"))?

        }

        (None, None, _) => channel(ledger, "general").await?.context("no #general (spectra init makes it)")?,

    };

    can_post(&who.principal, &ch).map_err(anyhow::Error::msg)?;

    let reply_to = match &p.reply_to {

        Some(r) => {

            let m = message(ledger, r).await?.with_context(|| format!("no message {r}"))?;

            if m.channel != ch.id {

                bail!("{r} is in another channel");

            }

            // A reply to a reply joins the root's thread.

            Some(m.reply_to.unwrap_or(m.id))

        }

        None => None,

    };

    let author = who.principal.actor();

    let dup: Option<String> = sqlx::query_scalar("SELECT id FROM messages WHERE channel = ? AND author = ? AND body = ? AND at_ms >= ? AND redacted IS NULL ORDER BY seq DESC LIMIT 1")

        .bind(&ch.id)

        .bind(&author)

        .bind(&p.body)

        .bind(now_ms.saturating_sub(DUPLICATE_MS) as i64)

        .fetch_optional(ledger.pool())

        .await?;

    if let Some(id) = dup {

        return Ok(Posted::Duplicate { id });

    }

    let mut mentions = resolve_mentions(ledger, &p.body, &ch).await?;

    for m in p.mentions.drain(..) {

        if let Some(a) = resolve_name(ledger, &m).await? {

            mentions.push(a);

        }

    }

    mentions.retain(|m| m != &author);

    mentions.sort();

    mentions.dedup();

    let mut refs = parse_refs(&p.body);

    refs.append(&mut p.refs);

    refs.sort();

    refs.dedup();

    let id = crate::ids::ulid();

    let payload = json!({

        "id": id, "channel": ch.id, "reply_to": reply_to, "kind": kind, "body": p.body,

        "payload": p.payload, "mentions": mentions, "refs": refs,

    });

    let mut e = who.event("message.posted", payload).subject(id.clone());

    e.project = ch.project.clone();

    ledger.append_one(e).await?;

    Ok(Posted::Message { id, channel: ch.display() })

}

/// Make the direct channel between `who` and `other` when there is none.

pub async fn ensure_direct(ledger: &Ledger, who: &Principal, other: &str) -> Result<Channel> {

    let other = resolve_name(ledger, other).await?.with_context(|| format!("nobody is called {other}"))?;

    let name = direct_name(&who.actor(), &other);

    if let Some(c) = channel(ledger, &name).await? {

        return Ok(c);

    }

    let mut members = vec![who.actor(), other];

    members.sort();

    ledger.append_one(channel_event(&name, "direct", None, None, &members)).await?;

    channel(ledger, &name).await?.context("the direct channel")

}

pub async fn message(ledger: &Ledger, id: &str) -> Result<Option<Message>> {

    Ok(sqlx::query_as(&format!("{SELECT_MESSAGE} WHERE id = ? OR (length(?) >= 6 AND id LIKE ? || '%') ORDER BY seq LIMIT 1"))

        .bind(id)

        .bind(id)

        .bind(id)

        .fetch_optional(ledger.pool())

        .await?)

}

const SELECT_MESSAGE: &str =

    "SELECT id, channel, reply_to, author, kind, body, payload, mentions, refs, at_ms, seq, replies, redacted FROM messages";

/// A channel's top-level messages (each with its reply count), or a whole

/// thread, oldest first; `since` is a message id or a time in ms.

pub async fn read(

    ledger: &Ledger,

    who: &Principal,

    channel_name: Option<&str>,

    thread: Option<&str>,

    since: Option<&str>,

    limit: u32,

) -> Result<(Channel, Vec<Message>)> {

    let limit = limit.clamp(1, 500) as i64;

    let since_seq = match since {

        None => 0,

        Some(s) => match s.parse::<i64>() {

            Ok(ms) if ms > 1_000_000_000_000 => {

                sqlx::query_scalar("SELECT COALESCE(MIN(seq), 1 << 62) - 1 FROM messages WHERE at_ms > ?")

                    .bind(ms)

                    .fetch_one(ledger.pool())

                    .await?

            }

            _ => message(ledger, s).await?.map(|m| m.seq).with_context(|| format!("no message {s}"))?,

        },

    };

    let (ch, rows) = match thread {

        Some(t) => {

            let root = message(ledger, t).await?.with_context(|| format!("no message {t}"))?;

            let root_id = root.reply_to.clone().unwrap_or(root.id.clone());

            let ch = channel(ledger, &root.channel).await?.context("the thread's channel")?;

            let rows: Vec<Message> =

                sqlx::query_as(&format!("{SELECT_MESSAGE} WHERE (id = ? OR reply_to = ?) AND seq > ? ORDER BY seq LIMIT ?"))

                    .bind(&root_id)

                    .bind(&root_id)

                    .bind(since_seq)

                    .bind(limit)

                    .fetch_all(ledger.pool())

                    .await?;

            (ch, rows)

        }

        None => {

            let name = match (channel_name, who) {

                (Some(c), _) => c.to_string(),

                (None, Principal::Agent { project, .. }) => project.clone(),

                (None, _) => "general".into(),

            };

            let ch = channel(ledger, &name).await?.with_context(|| format!("no channel {name}"))?;

            // The newest `limit`, returned oldest first.

            let mut rows: Vec<Message> =

                sqlx::query_as(&format!("{SELECT_MESSAGE} WHERE channel = ? AND reply_to IS NULL AND seq > ? ORDER BY seq DESC LIMIT ?"))

                    .bind(&ch.id)

                    .bind(since_seq)

                    .bind(limit)

                    .fetch_all(ledger.pool())

                    .await?;

            rows.reverse();

            (ch, rows)

        }

    };

    if !can_read(who, &ch) {

        bail!("{} is a direct channel you are not in", ch.display());

    }

    Ok((ch, rows.into_iter().map(shown).collect()))

}

/// A redacted message shows only that it was.

fn shown(mut m: Message) -> Message {

    if let Some(why) = &m.redacted {

        m.body = format!("[redacted: {why}]");

        m.payload = None;

    }

    m

}

/// One item in a principal's inbox.

#[derive(Clone, Debug, Serialize)]

pub struct Item {

    pub id: String,

    pub kind: String,

    pub at_ms: i64,

    pub from: String,

    pub channel: String,

    pub summary: String,

    pub action_hint: String,

}

/// The unread items of `actor`, oldest first; `deliver` records that they

/// were handed over (`inbox.delivered`).

pub async fn inbox(ledger: &Ledger, actor: &str, deliver: Option<&Author<'_>>) -> Result<Vec<Item>> {

    let rows: Vec<InboxRow> = sqlx::query_as(

        "SELECT i.item AS id, i.kind, m.at_ms, m.author AS from_, c.name AS channel, m.body, m.redacted FROM inbox i JOIN messages m ON m.id = i.item JOIN channels c ON c.id = m.channel \

         WHERE i.principal = ? AND i.read = 0 ORDER BY i.seq",

    )

    .bind(actor)

    .fetch_all(ledger.pool())

    .await?;

    let items: Vec<Item> = rows

        .into_iter()

        .map(|InboxRow { id, kind, at_ms, from_: from, channel: ch, body, redacted }| {

            let channel = match ch.strip_prefix("dm:") {

                Some(p) => format!("@{p}"),

                None => format!("#{ch}"),

            };

            let summary = if redacted.is_some() { "[redacted]".into() } else { summarise(&body, 500) };

            Item {

                action_hint: format!("read(thread: \"{id}\") for context, then say(reply_to: \"{id}\", …)"),

                id,

                kind,

                at_ms,

                from,

                channel,

                summary,

            }

        })

        .collect();

    if let (Some(who), Some(last)) = (deliver, items.last()) {

        let undelivered: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM inbox WHERE principal = ? AND delivered = 0 AND read = 0")

            .bind(actor)

            .fetch_one(ledger.pool())

            .await?;

        if undelivered > 0 {

            ledger.append_one(who.event("inbox.delivered", json!({ "principal": actor, "upto": last.id }))).await?;

        }

    }

    Ok(items)

}

/// The number of unread items, for `_meta`.

pub async fn unread(ledger: &Ledger, actor: &str) -> Result<i64> {

    Ok(sqlx::query_scalar("SELECT COUNT(*) FROM inbox WHERE principal = ? AND read = 0").bind(actor).fetch_one(ledger.pool()).await?)

}

/// Mark items read, with a reaction on each if given.

pub async fn ack(ledger: &Ledger, who: &Author<'_>, ids: &[String], reaction: Option<&str>) -> Result<usize> {

    if let Some(r) = reaction {

        if !REACTIONS.contains(&r) {

            bail!("a reaction is one of {}", REACTIONS.join(", "));

        }

    }

    let actor = who.principal.actor();

    let mut events = Vec::new();

    let mut found = Vec::new();

    for id in ids {

        let m = message(ledger, id).await?.with_context(|| format!("no message {id}"))?;

        found.push(m.id.clone());

        if let Some(r) = reaction {

            events.push(who.event("message.reaction", json!({ "id": m.id, "reaction": r })).subject(m.id.clone()));

        }

    }

    events.push(who.event("inbox.read", json!({ "principal": actor, "ids": found })));

    let n = found.len();

    ledger.append(events).await?;

    Ok(n)

}

/// Redact a message: its author's, or anyone's by the owner.

pub async fn redact(ledger: &Ledger, who: &Author<'_>, id: &str, reason: &str) -> Result<()> {

    let m = message(ledger, id).await?.with_context(|| format!("no message {id}"))?;

    if m.author != who.principal.actor() && !matches!(who.principal, Principal::Owner { .. }) {

        bail!("only its author or the owner may redact a message");

    }

    if reason.trim().is_empty() {

        bail!("say why it is redacted");

    }

    ledger.append_one(who.event("message.redacted", json!({ "id": m.id, "reason": reason })).subject(m.id)).await?;

    Ok(())

}

#[derive(sqlx::FromRow)]

struct InboxRow {

    id: String,

    kind: String,

    at_ms: i64,

    from_: String,

    channel: String,

    body: String,

    redacted: Option<String>,

}

#[derive(sqlx::FromRow)]

struct HitRow {

    id: String,

    name: String,

    author: String,

    at_ms: i64,

    snippet: String,

    reply_to: Option<String>,

    kind: String,

    members: String,

}

/// A search hit (spec §13.1).

#[derive(Clone, Debug, Serialize)]

pub struct Hit {

    pub id: String,

    pub channel: String,

    pub author: String,

    pub at_ms: i64,

    pub snippet: String,

    pub thread: Option<String>,

}

/// Full-text search over messages the caller may read, newest first.

pub async fn search(

    ledger: &Ledger,

    who: &Principal,

    query: &str,

    channel_name: Option<&str>,

    author: Option<&str>,