Coverage / crates/spectra-core/src/conversation.rs 90.7%

    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()),

    }

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

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

    }

    pub fn display(&self) -> String {

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

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

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

    }

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

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

}

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)

}

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])

}

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('@') {

        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?)

}

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

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

}

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

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

        (_, "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") => {

            let granted = format!("channel:{}", ch.name);

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

                Ok(())

                Err(format!(

                    "{} belongs to another project; post in #{project} or #general, mention its agents from there, or ask the owner to run `spectra agent allow NAME --channel {}`",

                    ch.display(),

                    ch.name

                ))

        _ => Ok(()),

}

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) {

    }

    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}"))?

    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 {

            }

            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(..) {

    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() })

}

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? {

    }

    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?)

}

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,

    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)

            let name = match (channel_name, who) {

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

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

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

            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()))

}

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

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

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

        m.payload = None;

    }

    m

}

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)

}

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?)

}

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

    if let Some(r) = reaction {

        if !REACTIONS.contains(&r) {

        }

    }

    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)

}

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() {