mirror of
https://github.com/block/buzz.git
synced 2026-10-27 12:40:30 +00:00
feat(messages): add since parameter to get_messages API and MCP tool (#154)
This commit is contained in:
@@ -69,6 +69,7 @@ pub async fn cmd_get_messages(
|
||||
channel_id: &str,
|
||||
limit: Option<u32>,
|
||||
before: Option<i64>,
|
||||
since: Option<i64>,
|
||||
kinds: Option<&str>,
|
||||
) -> Result<(), CliError> {
|
||||
validate_uuid(channel_id)?;
|
||||
@@ -77,6 +78,9 @@ pub async fn cmd_get_messages(
|
||||
if let Some(b) = before {
|
||||
path.push_str(&format!("&before={b}"));
|
||||
}
|
||||
if let Some(s) = since {
|
||||
path.push_str(&format!("&since={s}"));
|
||||
}
|
||||
if let Some(k) = kinds {
|
||||
path.push_str(&format!("&kinds={}", percent_encode(k)));
|
||||
}
|
||||
|
||||
@@ -97,6 +97,8 @@ enum Cmd {
|
||||
#[arg(long)]
|
||||
before: Option<i64>,
|
||||
#[arg(long)]
|
||||
since: Option<i64>,
|
||||
#[arg(long)]
|
||||
kinds: Option<String>,
|
||||
},
|
||||
/// Get a message thread
|
||||
@@ -531,10 +533,18 @@ async fn run(cli: Cli) -> Result<(), CliError> {
|
||||
channel,
|
||||
limit,
|
||||
before,
|
||||
since,
|
||||
kinds,
|
||||
} => {
|
||||
commands::messages::cmd_get_messages(&client, &channel, limit, before, kinds.as_deref())
|
||||
.await
|
||||
commands::messages::cmd_get_messages(
|
||||
&client,
|
||||
&channel,
|
||||
limit,
|
||||
before,
|
||||
since,
|
||||
kinds.as_deref(),
|
||||
)
|
||||
.await
|
||||
}
|
||||
Cmd::GetThread {
|
||||
channel,
|
||||
|
||||
@@ -641,6 +641,7 @@ impl Db {
|
||||
channel_id: Uuid,
|
||||
limit: u32,
|
||||
before_cursor: Option<DateTime<Utc>>,
|
||||
since_cursor: Option<DateTime<Utc>>,
|
||||
kind_filter: Option<&[u32]>,
|
||||
) -> Result<Vec<thread::TopLevelMessage>> {
|
||||
thread::get_channel_messages_top_level(
|
||||
@@ -648,6 +649,7 @@ impl Db {
|
||||
channel_id,
|
||||
limit,
|
||||
before_cursor,
|
||||
since_cursor,
|
||||
kind_filter,
|
||||
)
|
||||
.await
|
||||
|
||||
@@ -480,14 +480,19 @@ pub async fn get_thread_summary(pool: &PgPool, event_id: &[u8]) -> Result<Option
|
||||
/// - At depth 0 (root messages), OR
|
||||
/// - At depth 1 with `broadcast = true` (replies surfaced to the channel)
|
||||
///
|
||||
/// Results are ordered newest-first for a standard channel view.
|
||||
/// `before_cursor` enables keyset pagination (pass the `created_at` of the
|
||||
/// last item from the previous page).
|
||||
/// Default ordering is newest-first (DESC). When `since_cursor` is provided
|
||||
/// without `before_cursor`, ordering flips to oldest-first (ASC) for
|
||||
/// chronological polling.
|
||||
///
|
||||
/// `before_cursor` enables backward keyset pagination (pass the `created_at`
|
||||
/// of the last item from the previous page). `since_cursor` enables forward
|
||||
/// polling (returns only messages created after the given timestamp).
|
||||
pub async fn get_channel_messages_top_level(
|
||||
pool: &PgPool,
|
||||
channel_id: Uuid,
|
||||
limit: u32,
|
||||
before_cursor: Option<DateTime<Utc>>,
|
||||
since_cursor: Option<DateTime<Utc>>,
|
||||
kind_filter: Option<&[u32]>,
|
||||
) -> Result<Vec<TopLevelMessage>> {
|
||||
let mut param_idx = 2u32; // $1 is channel_id
|
||||
@@ -520,6 +525,11 @@ pub async fn get_channel_messages_top_level(
|
||||
param_idx += 1;
|
||||
}
|
||||
|
||||
if since_cursor.is_some() {
|
||||
sql.push_str(&format!(" AND e.created_at > ${param_idx}"));
|
||||
param_idx += 1;
|
||||
}
|
||||
|
||||
if let Some(kinds) = kind_filter {
|
||||
if !kinds.is_empty() {
|
||||
let list = kinds
|
||||
@@ -531,13 +541,23 @@ pub async fn get_channel_messages_top_level(
|
||||
}
|
||||
}
|
||||
|
||||
sql.push_str(&format!(" ORDER BY e.created_at DESC LIMIT ${param_idx}"));
|
||||
let order = if since_cursor.is_some() && before_cursor.is_none() {
|
||||
"ASC"
|
||||
} else {
|
||||
"DESC"
|
||||
};
|
||||
sql.push_str(&format!(
|
||||
" ORDER BY e.created_at {order} LIMIT ${param_idx}"
|
||||
));
|
||||
|
||||
let mut q = sqlx::query(&sql).bind(channel_id);
|
||||
|
||||
if let Some(cursor) = before_cursor {
|
||||
q = q.bind(cursor);
|
||||
}
|
||||
if let Some(cursor) = since_cursor {
|
||||
q = q.bind(cursor);
|
||||
}
|
||||
q = q.bind(limit as i32);
|
||||
|
||||
let rows = q.fetch_all(pool).await?;
|
||||
|
||||
@@ -109,6 +109,11 @@ pub struct GetMessagesParams {
|
||||
/// Unix timestamp cursor for pagination. Returns messages before this time.
|
||||
#[serde(default)]
|
||||
pub before: Option<i64>,
|
||||
/// Unix timestamp cursor. Returns messages created strictly after this time.
|
||||
/// When used without `before`, results are ordered oldest-first (chronological).
|
||||
/// Useful for polling: pass the timestamp of the last seen message to get only newer ones.
|
||||
#[serde(default)]
|
||||
pub since: Option<i64>,
|
||||
/// Comma-separated event kind numbers to filter by (e.g. "45001" for forum posts,
|
||||
/// "45002" for votes). When omitted, all kinds are returned.
|
||||
#[serde(default)]
|
||||
@@ -1072,8 +1077,10 @@ Default kind is 9 (stream message)."
|
||||
/// Get recent messages from a Sprout channel.
|
||||
#[tool(
|
||||
name = "get_messages",
|
||||
description = "Fetch recent top-level messages from a Sprout channel. Use `before` for pagination \
|
||||
(Unix timestamp). Use `kinds` to filter by event type (e.g. \"45001\" for forum posts, \
|
||||
description = "Fetch recent top-level messages from a Sprout channel. Use `before` for backward \
|
||||
pagination and `since` for forward pagination (both Unix timestamps). When `since` is \
|
||||
provided without `before`, results are ordered oldest-first — useful for polling new \
|
||||
messages. Use `kinds` to filter by event type (e.g. \"45001\" for forum posts, \
|
||||
\"45002\" for votes). Thread summaries are included automatically. Threaded replies \
|
||||
are not returned — use `get_thread` to fetch the full reply tree for a specific message."
|
||||
)]
|
||||
@@ -1096,6 +1103,9 @@ are not returned — use `get_thread` to fetch the full reply tree for a specifi
|
||||
if let Some(before) = p.before {
|
||||
query_parts.push(format!("before={before}"));
|
||||
}
|
||||
if let Some(since) = p.since {
|
||||
query_parts.push(format!("since={since}"));
|
||||
}
|
||||
if let Some(ref kinds) = p.kinds {
|
||||
query_parts.push(format!("kinds={}", percent_encode(kinds)));
|
||||
}
|
||||
|
||||
@@ -387,6 +387,10 @@ pub struct ListMessagesParams {
|
||||
/// Pagination cursor — Unix timestamp (seconds). Returns messages created
|
||||
/// strictly before this time.
|
||||
pub before: Option<i64>,
|
||||
/// Pagination cursor — Unix timestamp (seconds). Returns messages created
|
||||
/// strictly after this time. Results are ordered oldest-first when `since`
|
||||
/// is provided without `before`.
|
||||
pub since: Option<i64>,
|
||||
/// Legacy parameter (thread summaries are now always included). Kept for backward compatibility.
|
||||
#[serde(default)]
|
||||
pub with_threads: bool,
|
||||
@@ -395,7 +399,10 @@ pub struct ListMessagesParams {
|
||||
pub kinds: Option<String>,
|
||||
}
|
||||
|
||||
/// List top-level messages in a channel (newest first).
|
||||
/// List top-level messages in a channel.
|
||||
///
|
||||
/// Default ordering is newest-first (DESC). When `since` is provided without
|
||||
/// `before`, ordering flips to oldest-first (ASC) for chronological polling.
|
||||
///
|
||||
/// Returns root messages and broadcast replies. Thread summaries (reply counts,
|
||||
/// participant pubkeys) are always included. Thread replies themselves are excluded —
|
||||
@@ -423,6 +430,10 @@ pub async fn list_messages(
|
||||
.before
|
||||
.and_then(|ts| chrono::DateTime::from_timestamp(ts, 0));
|
||||
|
||||
let since_cursor: Option<chrono::DateTime<Utc>> = params
|
||||
.since
|
||||
.and_then(|ts| chrono::DateTime::from_timestamp(ts, 0));
|
||||
|
||||
let kind_filter: Option<Vec<u32>> = params
|
||||
.kinds
|
||||
.as_deref()
|
||||
@@ -441,7 +452,13 @@ pub async fn list_messages(
|
||||
|
||||
let mut messages = state
|
||||
.db
|
||||
.get_channel_messages_top_level(channel_id, limit, before_cursor, kind_filter.as_deref())
|
||||
.get_channel_messages_top_level(
|
||||
channel_id,
|
||||
limit,
|
||||
before_cursor,
|
||||
since_cursor,
|
||||
kind_filter.as_deref(),
|
||||
)
|
||||
.await
|
||||
.map_err(|e| internal_error(&format!("db error: {e}")))?;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user