clawmates/crates/cm-db/src/repo/threads.rs

use cm_domain::{AgentId, WorkspaceId};
use sqlx::PgPool;
use time::OffsetDateTime;
use uuid::Uuid;

use crate::DbError;

/// An inter-agent conversation (§7.2 Claw Chat). `kind` is `dm` for the 1:1
/// threads `find_or_create` makes, or `room` for N-way group rooms.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct Thread {
    pub id: Uuid,
    pub workspace_id: Uuid,
    pub subject: String,
    pub sensitivity: String,
    pub kind: String,
    pub created_by: Option<Uuid>,
    pub participants: Vec<Uuid>,
    pub last_preview: Option<String>,
    #[serde(with = "time::serde::rfc3339")]
    pub created_at: OffsetDateTime,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ThreadMessage {
    pub id: Uuid,
    pub thread_id: Uuid,
    pub from_agent: Uuid,
    pub content: serde_json::Value,
    pub taint: Vec<String>,
    #[serde(with = "time::serde::rfc3339")]
    pub created_at: OffsetDateTime,
}

/// Finds the 1:1 thread between two agents, or creates it with the given
/// subject. The match is constrained to `kind = 'dm'` threads with exactly the
/// two active participants, so a DM never resolves to a group room that happens
/// to contain both agents.
pub async fn find_or_create(
    pool: &PgPool,
    workspace_id: WorkspaceId,
    a: AgentId,
    b: AgentId,
    subject: &str,
) -> Result<Uuid, DbError> {
    let existing = sqlx::query_scalar!(
        r#"SELECT t.id FROM threads t
           WHERE t.workspace_id = $1
             AND t.kind = 'dm'
             AND EXISTS (SELECT 1 FROM thread_participants p
                         WHERE p.thread_id = t.id AND p.agent_id = $2 AND p.active)
             AND EXISTS (SELECT 1 FROM thread_participants p
                         WHERE p.thread_id = t.id AND p.agent_id = $3 AND p.active)
             AND (SELECT count(*) FROM thread_participants p
                  WHERE p.thread_id = t.id AND p.active) = 2
           LIMIT 1"#,
        workspace_id.as_uuid(),
        a.as_uuid(),
        b.as_uuid(),
    )
    .fetch_optional(pool)
    .await?;
    if let Some(id) = existing {
        return Ok(id);
    }

    let id = Uuid::now_v7();
    let mut tx = pool.begin().await.map_err(DbError::from)?;
    sqlx::query!(
        "INSERT INTO threads (id, workspace_id, subject, kind) VALUES ($1, $2, $3, 'dm')",
        id,
        workspace_id.as_uuid(),
        subject,
    )
    .execute(&mut *tx)
    .await?;
    for agent in [a, b] {
        sqlx::query!(
            "INSERT INTO thread_participants (thread_id, agent_id, added_by) VALUES ($1, $2, $3)",
            id,
            agent.as_uuid(),
            a.as_uuid(),
        )
        .execute(&mut *tx)
        .await?;
    }
    tx.commit().await.map_err(DbError::from)?;
    Ok(id)
}

/// Creates an N-way group room with the given participants. `created_by` is the
/// agent who opened it (a claw via `room.create`) or `None` for an
/// operator-created room. Participants are deduped.
pub async fn create_room(
    pool: &PgPool,
    workspace_id: WorkspaceId,
    subject: &str,
    created_by: Option<AgentId>,
    participants: &[AgentId],
) -> Result<Uuid, DbError> {
    let id = Uuid::now_v7();
    let creator = created_by.map(|a| a.as_uuid());
    let mut tx = pool.begin().await.map_err(DbError::from)?;
    sqlx::query!(
        "INSERT INTO threads (id, workspace_id, subject, kind, created_by)
         VALUES ($1, $2, $3, 'room', $4)",
        id,
        workspace_id.as_uuid(),
        subject,
        creator,
    )
    .execute(&mut *tx)
    .await?;
    let mut seen = std::collections::HashSet::new();
    for agent in participants {
        if !seen.insert(agent.as_uuid()) {
            continue;
        }
        sqlx::query!(
            "INSERT INTO thread_participants (thread_id, agent_id, added_by) VALUES ($1, $2, $3)",
            id,
            agent.as_uuid(),
            creator,
        )
        .execute(&mut *tx)
        .await?;
    }
    tx.commit().await.map_err(DbError::from)?;
    Ok(id)
}

/// Adds (or re-activates) a participant in a thread. `added_by` is the inviter,
/// or `None` for an operator action.
pub async fn add_participant(
    pool: &PgPool,
    thread_id: Uuid,
    agent: AgentId,
    added_by: Option<AgentId>,
) -> Result<(), DbError> {
    sqlx::query!(
        "INSERT INTO thread_participants (thread_id, agent_id, added_by, active)
         VALUES ($1, $2, $3, true)
         ON CONFLICT (thread_id, agent_id)
         DO UPDATE SET active = true, added_by = EXCLUDED.added_by",
        thread_id,
        agent.as_uuid(),
        added_by.map(|a| a.as_uuid()),
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// Soft-removes a participant (keeps their messages attributable).
pub async fn remove_participant(
    pool: &PgPool,
    thread_id: Uuid,
    agent: AgentId,
) -> Result<(), DbError> {
    sqlx::query!(
        "UPDATE thread_participants SET active = false
         WHERE thread_id = $1 AND agent_id = $2",
        thread_id,
        agent.as_uuid(),
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// Active participant ids in a thread (for message fan-out + event emission).
pub async fn participants(pool: &PgPool, thread_id: Uuid) -> Result<Vec<Uuid>, DbError> {
    let rows = sqlx::query_scalar!(
        "SELECT agent_id FROM thread_participants WHERE thread_id = $1 AND active",
        thread_id,
    )
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// All group rooms in a workspace (human/operator room list).
pub async fn list_rooms_for_workspace(
    pool: &PgPool,
    workspace_id: WorkspaceId,
) -> Result<Vec<Thread>, DbError> {
    let rows = sqlx::query!(
        r#"SELECT t.id, t.workspace_id, t.subject, t.sensitivity, t.kind,
                  t.created_by, t.created_at,
                  ARRAY(SELECT p2.agent_id FROM thread_participants p2
                        WHERE p2.thread_id = t.id AND p2.active) AS "participants!",
                  (SELECT m.content->>'text' FROM thread_messages m
                   WHERE m.thread_id = t.id
                   ORDER BY m.created_at DESC LIMIT 1) AS last_preview
           FROM threads t
           WHERE t.workspace_id = $1 AND t.kind = 'room'
           ORDER BY (SELECT max(m.created_at) FROM thread_messages m
                     WHERE m.thread_id = t.id) DESC NULLS LAST"#,
        workspace_id.as_uuid(),
    )
    .fetch_all(pool)
    .await?;
    Ok(rows
        .into_iter()
        .map(|r| Thread {
            id: r.id,
            workspace_id: r.workspace_id,
            subject: r.subject,
            sensitivity: r.sensitivity,
            kind: r.kind,
            created_by: r.created_by,
            participants: r.participants,
            last_preview: r.last_preview,
            created_at: r.created_at,
        })
        .collect())
}

pub async fn add_message(
    pool: &PgPool,
    thread_id: Uuid,
    from_agent: AgentId,
    content: serde_json::Value,
    taint: &[String],
) -> Result<Uuid, DbError> {
    let id = Uuid::now_v7();
    sqlx::query!(
        "INSERT INTO thread_messages (id, thread_id, from_agent, content, taint)
         VALUES ($1, $2, $3, $4, $5)",
        id,
        thread_id,
        from_agent.as_uuid(),
        content,
        taint,
    )
    .execute(pool)
    .await?;
    Ok(id)
}

/// Threads an agent participates in, most recent message first, with the
/// last message preview (§7.2 thread list).
pub async fn list_for_agent(pool: &PgPool, agent_id: AgentId) -> Result<Vec<Thread>, DbError> {
    let rows = sqlx::query!(
        r#"SELECT t.id, t.workspace_id, t.subject, t.sensitivity, t.kind,
                  t.created_by, t.created_at,
                  ARRAY(SELECT p2.agent_id FROM thread_participants p2
                        WHERE p2.thread_id = t.id AND p2.active) AS "participants!",
                  (SELECT m.content->>'text' FROM thread_messages m
                   WHERE m.thread_id = t.id
                   ORDER BY m.created_at DESC LIMIT 1) AS last_preview
           FROM threads t
           JOIN thread_participants p ON p.thread_id = t.id
           WHERE p.agent_id = $1 AND p.active
           ORDER BY (SELECT max(m.created_at) FROM thread_messages m
                     WHERE m.thread_id = t.id) DESC NULLS LAST"#,
        agent_id.as_uuid(),
    )
    .fetch_all(pool)
    .await?;
    Ok(rows
        .into_iter()
        .map(|r| Thread {
            id: r.id,
            workspace_id: r.workspace_id,
            subject: r.subject,
            sensitivity: r.sensitivity,
            kind: r.kind,
            created_by: r.created_by,
            participants: r.participants,
            last_preview: r.last_preview,
            created_at: r.created_at,
        })
        .collect())
}

pub async fn messages(pool: &PgPool, thread_id: Uuid) -> Result<Vec<ThreadMessage>, DbError> {
    let rows = sqlx::query_as!(
        ThreadMessage,
        r#"SELECT id, thread_id, from_agent, content, taint, created_at
           FROM thread_messages WHERE thread_id = $1 ORDER BY created_at"#,
        thread_id,
    )
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// The workspace a thread belongs to, if it exists (API scoping).
pub async fn workspace_of(pool: &PgPool, thread_id: Uuid) -> Result<Option<Uuid>, DbError> {
    Ok(
        sqlx::query_scalar!("SELECT workspace_id FROM threads WHERE id = $1", thread_id)
            .fetch_optional(pool)
            .await?,
    )
}

/// A thread's subject line, if it exists.
pub async fn subject(pool: &PgPool, thread_id: Uuid) -> Result<Option<String>, DbError> {
    Ok(
        sqlx::query_scalar!("SELECT subject FROM threads WHERE id = $1", thread_id)
            .fetch_optional(pool)
            .await?,
    )
}

/// Whether an agent participates in a thread (API scoping).
pub async fn is_participant(
    pool: &PgPool,
    thread_id: Uuid,
    agent_id: AgentId,
) -> Result<bool, DbError> {
    let row = sqlx::query_scalar!(
        "SELECT 1 AS x FROM thread_participants WHERE thread_id = $1 AND agent_id = $2 AND active",
        thread_id,
        agent_id.as_uuid(),
    )
    .fetch_optional(pool)
    .await?;
    Ok(row.is_some())
}