codex/codex-rs/message-board/src/local.rs

//! SQLite-backed boards. The board ID scopes every read and write.
//!
//! Immediate transactions serialize mutations across independently opened handles.
//! Live handles in this process share one connection pool per database path.
//! Accepted posts survive runtime unload and process restart, but cannot recreate
//! a board after its root has been permanently deleted.
//! Opening storage adds optional channel metadata without invalidating older readers.
//! Explicit connections can restrict writes; default handles retain legacy access.
//! Fixed read views freeze the channel catalog and restrict reads to a captured post sequence.
//! Posting subscribes and notifies; host appends share persistence without these side effects.

use self::connection::Access;
use self::options::LocalBoardOptions;
use crate::AppendRequest;
use crate::ChannelDescription;
use crate::ChannelSummary;
use crate::CreateChannelRequest;
use crate::MessageBoardHost;
use crate::PostDestination;
use crate::PostMetadata;
use crate::PostRequest;
use crate::SubscriptionChange;
use crate::SubscriptionRequest;
use crate::SubscriptionState;
use crate::SubscriptionTarget;
use crate::api::PostBehavior;
use caseless::default_case_fold_str;
use chrono::DateTime;
use chrono::Utc;
use codex_protocol::AgentPath;
use codex_protocol::SessionId;
use codex_protocol::ThreadId;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result;
use codex_state::SqliteConfig;
use futures::StreamExt;
use serde::Deserialize;
use serde::Serialize;
use sqlx::Row;
use sqlx::SqliteConnection;
use sqlx::SqlitePool;
use std::collections::HashSet;
use std::sync::Arc;
use uuid::Uuid;

mod catalog;
pub(crate) mod connection;
mod json_search;
mod lifecycle;
pub(crate) mod options;
mod paging;
mod pool;
mod queries;
pub(crate) mod setup;
mod subscriptions;

#[cfg(test)]
#[path = "local/pools_tests.rs"]
mod pools_tests;

const MAX_POST_BYTES: usize = 64 * 1024;
const MAX_CHANNEL_BYTES: usize = 128;
const MAX_READ_CHARS: usize = 20_000;

/// Capturing a view makes the handle read-only, regardless of its original access.
#[derive(Clone)]
enum BoardMode {
    Live { access: Option<Access> },
    Fixed { view: Arc<catalog::ReadView> },
}

impl BoardMode {
    const FIXED_ACCESS: Access = Access::ReadOnly;

    fn access(&self) -> Option<Access> {
        match self {
            Self::Live { access } => *access,
            Self::Fixed { .. } => Some(Self::FIXED_ACCESS),
        }
    }

    fn view(&self) -> Option<&catalog::ReadView> {
        match self {
            Self::Live { .. } => None,
            Self::Fixed { view } => Some(view),
        }
    }
}

#[derive(Clone)]
pub struct LocalAgentMessageBoard {
    identity: SessionId,
    storage_namespace: Arc<str>,
    pool: Arc<SqlitePool>,
    host: Arc<dyn MessageBoardHost>,
    mode: BoardMode,
    nested_json_pointer: Option<&'static str>,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
struct StoredPost {
    metadata: PostMetadata,
    text: String,
}

impl LocalAgentMessageBoard {
    /// Reopens the same board for a root, child or resumed runtime. The shared
    /// SQLite configuration preserves the host's connection and journal policy.
    /// The caller selects the storage namespace.
    pub async fn open(
        sqlite: &SqliteConfig,
        options: &LocalBoardOptions,
        identity: SessionId,
        host: Arc<dyn MessageBoardHost>,
    ) -> Result<Self> {
        Ok(Self {
            identity,
            pool: pool::open(sqlite, options).await?,
            storage_namespace: Arc::from(options.storage_namespace.as_str()),
            host,
            mode: BoardMode::Live { access: None },
            nested_json_pointer: None,
        })
    }

    pub async fn create_channel(
        &self,
        caller: ThreadId,
        request: CreateChannelRequest,
    ) -> Result<ChannelSummary> {
        if let Some(access) = self.mode.access() {
            access.require_write(self.identity, "board.create_channel")?;
        }
        validate_channel(&request.channel_name)?;
        let author = self.host.agent_path(caller).await?;
        let now = self.host.current_time(caller).await?;
        let mut tx = self.begin_write().await?;
        insert_channel(
            &mut tx,
            self.identity,
            &request.channel_name,
            request.description.as_ref(),
            &author,
            now,
        )
        .await?;
        if request.subscription == SubscriptionChange::Subscribe {
            self.subscribe(
                &mut tx,
                &SubscriptionTarget::Channel(request.channel_name.clone()),
                caller,
            )
            .await?;
        }
        let summary = catalog::channel_summary(self, &mut tx, &request.channel_name).await?;
        tx.commit().await.map_err(storage_error)?;
        Ok(summary)
    }

    /// Once started, finish the accepted write and fanout even if the tool caller
    /// disconnects. Delivery is attempted once; retries never replay old notices
    /// into a recipient's later turn. Once committed, notification failures are
    /// logged without failing the post; the persisted content remains readable.
    pub async fn post(&self, caller: ThreadId, request: PostRequest) -> Result<PostMetadata> {
        self.write(caller, request, PostBehavior::Discussion).await
    }

    /// Appends evidence without automatic subscriptions or notifications, even on a discussion board.
    /// Uses the same access checks, caller-scoped request IDs and durable write path as posting.
    /// Reusing a request ID for a different payload or operation is an error.
    pub async fn append(&self, caller: ThreadId, request: AppendRequest) -> Result<PostMetadata> {
        self.write(
            caller,
            PostRequest {
                request_id: request.request_id,
                destination: request.destination,
                text: request.text,
                agents_to_notify: Vec::new(),
            },
            PostBehavior::SilentAppend,
        )
        .await
    }

    async fn write(
        &self,
        caller: ThreadId,
        request: PostRequest,
        behavior: PostBehavior,
    ) -> Result<PostMetadata> {
        if let Some(access) = self.mode.access() {
            access.require_write(
                self.identity,
                match &request.destination {
                    PostDestination::Thread(_) => "channel.reply",
                    PostDestination::Channel(_) | PostDestination::NewChannel(_) => "channel.post",
                },
            )?;
        }
        let board = self.clone();
        tokio::spawn(async move { board.post_inner(caller, request, behavior).await })
            .await
            .map_err(storage_error)?
    }

    async fn post_inner(
        &self,
        caller: ThreadId,
        request: PostRequest,
        behavior: PostBehavior,
    ) -> Result<PostMetadata> {
        if request.text.len() > MAX_POST_BYTES
            || request.text.is_empty()
            || request.request_id.is_empty()
            || request.request_id.len() > 512
            || request.agents_to_notify.len() > 256
        {
            return Err(invalid(
                "post text, request ID or recipient count exceeds the board limits",
            ));
        }
        let author = self.host.agent_path(caller).await?;
        let request_id = format!("{caller}:{}", request.request_id);
        let request_json = match behavior {
            // Keep existing post fingerprints compatible with persisted retries.
            PostBehavior::Discussion => serde_json::to_string(&request),
            PostBehavior::SilentAppend => serde_json::to_string(&("append", &request)),
        }
        .map_err(storage_error)?;
        let mut conn = self.pool.acquire().await.map_err(storage_error)?;
        if let Some(post) = self
            .existing_post(&mut conn, &request_id, &request_json)
            .await?
        {
            return Ok(post.metadata);
        }
        drop(conn);
        let mut recipients = HashSet::new();
        if let PostBehavior::Discussion = behavior {
            for path in &request.agents_to_notify {
                recipients.insert(self.host.resolve_agent(path.clone()).await?);
            }
        }
        let now = self.host.current_time(caller).await?;
        let mut tx = self.begin_write().await?;
        if let Some(post) = self
            .existing_post(&mut tx, &request_id, &request_json)
            .await?
        {
            return Ok(post.metadata);
        }
        let id = Uuid::now_v7();
        let (channel, root, target) = match &request.destination {
            PostDestination::Channel(channel) => {
                let exists: bool = sqlx::query_scalar(
                    "SELECT EXISTS(SELECT 1 FROM channels WHERE board=? AND name=?)",
                )
                .bind(self.identity.to_string())
                .bind(channel)
                .fetch_one(&mut *tx)
                .await
                .map_err(storage_error)?;
                if !exists {
                    return Err(invalid("channel not found in this board"));
                }
                (
                    channel.clone(),
                    id,
                    SubscriptionTarget::Channel(channel.clone()),
                )
            }
            PostDestination::NewChannel(channel) => {
                validate_channel(channel)?;
                insert_channel(
                    &mut tx,
                    self.identity,
                    channel,
                    /*description*/ None,
                    &author,
                    now,
                )
                .await?;
                if let PostBehavior::Discussion = behavior {
                    self.subscribe(
                        &mut tx,
                        &SubscriptionTarget::Channel(channel.clone()),
                        caller,
                    )
                    .await?;
                }
                (
                    channel.clone(),
                    id,
                    SubscriptionTarget::Channel(channel.clone()),
                )
            }
            PostDestination::Thread(root) => {
                let post = self.load_post(&mut tx, *root).await?;
                if post.metadata.thread_id != *root {
                    return Err(invalid("thread_id must identify a top-level post"));
                }
                (
                    post.metadata.channel_name,
                    *root,
                    SubscriptionTarget::Thread(*root),
                )
            }
        };
        // Return one JSON array to avoid handing each subscriber row across
        // SQLite's worker thread while the write transaction is held.
        if let PostBehavior::Discussion = behavior {
            let subscribed = sqlx::query_scalar::<_, String>(
                "SELECT json_group_array(agent) FROM subscriptions WHERE board=? AND target=?",
            )
            .bind(self.identity.to_string())
            .bind(target_key(&target)?)
            .fetch_one(&mut *tx)
            .await
            .map_err(storage_error)?;
            let subscribed: Vec<String> =
                serde_json::from_str(&subscribed).map_err(storage_error)?;
            for recipient in subscribed {
                recipients.insert(ThreadId::from_string(&recipient).map_err(storage_error)?);
            }
        }
        recipients.remove(&caller);
        let post = StoredPost {
            metadata: PostMetadata {
                message_id: id,
                channel_name: channel.clone(),
                author: author.clone(),
                thread_id: root,
                created_at: now,
            },
            text: request.text,
        };
        sqlx::query("INSERT INTO posts(board,id,channel,root,author,timestamp,body_search,payload,request_id,request) VALUES(?,?,?,?,?,?,?,?,?,?)")
            .bind(self.identity.to_string()).bind(id.to_string()).bind(channel).bind(root.to_string())
            .bind(author.to_string()).bind(now.timestamp_micros()).bind(default_case_fold_str(&post.text))
            .bind(serde_json::to_string(&post).map_err(storage_error)?).bind(request_id).bind(request_json)
            .execute(&mut *tx).await.map_err(storage_error)?;
        // Participation subscribes by default, without overriding an explicit opt-out.
        if let PostBehavior::Discussion = behavior {
            self.subscribe(&mut tx, &SubscriptionTarget::Thread(root), caller)
                .await?;
        }
        tx.commit().await.map_err(storage_error)?;

        if let PostBehavior::SilentAppend = behavior {
            return Ok(post.metadata);
        }
        // A committed post succeeds even if a best-effort notice cannot be delivered.
        let notice = paging::preview(post.clone(), /*max_chars*/ 150);
        futures::stream::iter(recipients)
            .for_each_concurrent(/*limit*/ 16, |recipient| {
                let notice = notice.clone();
                async move {
                    if let Err(error) = self.host.notify(recipient, notice).await {
                        tracing::warn!(%recipient, %error, "Failed to deliver message-board notification");
                    }
                }
            })
            .await;
        Ok(post.metadata)
    }

    pub async fn set_subscription(
        &self,
        caller: ThreadId,
        request: SubscriptionRequest,
    ) -> Result<SubscriptionState> {
        let caller_path = self.host.agent_path(caller).await?;
        let target_path = request.target_agent.unwrap_or(caller_path);
        let target_agent = self.host.resolve_agent(target_path.clone()).await?;
        if target_agent != caller
            && let Some(access) = self.mode.access()
        {
            access.require_write(self.identity, "board.manage_subscriptions")?;
        }
        let mut tx = self.begin_write().await?;
        let (channel, root, last) = match &request.target {
            SubscriptionTarget::Channel(name) => {
                let last: Option<String> = sqlx::query_scalar(
                    "SELECT (SELECT p.id FROM posts p WHERE p.board=c.board AND p.channel=c.name
                             ORDER BY p.timestamp DESC,p.seq DESC LIMIT 1)
                     FROM channels c WHERE c.board=? AND c.name=?",
                )
                .bind(self.identity.to_string())
                .bind(name)
                .fetch_optional(&mut *tx)
                .await
                .map_err(storage_error)?
                .ok_or_else(|| invalid("channel not found in this board"))?;
                (name.clone(), None, last)
            }
            SubscriptionTarget::Thread(root) => {
                let post = self.load_post(&mut tx, *root).await?;
                if post.metadata.thread_id != *root {
                    return Err(invalid("thread_id must identify a top-level post"));
                }
                let last: String = sqlx::query_scalar("SELECT id FROM posts WHERE board=? AND root=? ORDER BY timestamp DESC, seq DESC LIMIT 1")
                    .bind(self.identity.to_string()).bind(root.to_string()).fetch_one(&mut *tx).await.map_err(storage_error)?;
                (post.metadata.channel_name, Some(*root), Some(last))
            }
        };
        let last_message_id = last
            .as_deref()
            .map(Uuid::parse_str)
            .transpose()
            .map_err(storage_error)?;
        self.change_subscription(&mut tx, &request.target, target_agent, request.change)
            .await?;
        tx.commit().await.map_err(storage_error)?;
        Ok(SubscriptionState {
            channel_name: channel,
            thread_id: root,
            target_agent: target_path,
            enabled: request.change == SubscriptionChange::Subscribe,
            last_message_id,
        })
    }

    async fn subscribe(
        &self,
        conn: &mut SqliteConnection,
        target: &SubscriptionTarget,
        agent: ThreadId,
    ) -> Result<()> {
        sqlx::query("INSERT OR IGNORE INTO subscriptions(board,target,agent) SELECT ?1,?2,?3 WHERE NOT EXISTS(SELECT 1 FROM subscription_opt_outs WHERE board=?1 AND target=?2 AND agent=?3)")
            .bind(self.identity.to_string())
            .bind(target_key(target)?)
            .bind(agent.to_string())
            .execute(conn)
            .await
            .map_err(storage_error)?;
        Ok(())
    }

    async fn existing_post(
        &self,
        conn: &mut SqliteConnection,
        request_id: &str,
        request: &str,
    ) -> Result<Option<StoredPost>> {
        let row = sqlx::query("SELECT payload,request FROM posts WHERE board=? AND request_id=?")
            .bind(self.identity.to_string())
            .bind(request_id)
            .fetch_optional(conn)
            .await
            .map_err(storage_error)?;
        row.map(|row| {
            if row.get::<String, _>("request") != request {
                return Err(invalid("request ID was already used for a different post"));
            }
            serde_json::from_str(row.get("payload")).map_err(storage_error)
        })
        .transpose()
    }
}

// Runtime writes and template setup validate names and board lifecycle before
// inserting within their own transaction.
async fn insert_channel(
    conn: &mut SqliteConnection,
    board: SessionId,
    name: &str,
    description: Option<&ChannelDescription>,
    author: &AgentPath,
    now: DateTime<Utc>,
) -> Result<()> {
    let inserted = sqlx::query("INSERT OR IGNORE INTO channels(board,name,name_search,created_at,timestamp,author,description) VALUES(?,?,?,?,?,?,?)")
        .bind(board.to_string()).bind(name).bind(default_case_fold_str(name)).bind(now.to_rfc3339()).bind(now.timestamp_micros()).bind(author.to_string()).bind(description.map(ChannelDescription::as_str))
        .execute(conn).await.map_err(storage_error)?.rows_affected();
    if inserted == 0 {
        return Err(invalid("channel already exists"));
    }
    Ok(())
}

fn validate_channel(name: &str) -> Result<()> {
    if name.is_empty()
        || name.len() > MAX_CHANNEL_BYTES
        || name.trim() != name
        || name.chars().any(char::is_control)
    {
        return Err(invalid(
            "channel names must contain 1–128 bytes without edge whitespace or control characters",
        ));
    }
    Ok(())
}

fn invalid(message: impl Into<String>) -> CodexErr {
    CodexErr::InvalidRequest(message.into())
}

fn storage_error(error: impl std::fmt::Display) -> CodexErr {
    CodexErr::Io(std::io::Error::other(error.to_string()))
}

fn target_key(target: &SubscriptionTarget) -> Result<String> {
    serde_json::to_string(target).map_err(storage_error)
}