External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

chore: remove redundant inline comments across all crates (#13)

Browse Source
This commit is contained in:
tlongwell-block
2026-03-10 12:06:17 -04:00
committed by GitHub
parent cd3b45f948
commit 431c9f2b46
34 changed files with 2 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/sprout-audit/src/service.rs
-1
View File
@@ -75,7 +75,6 @@ impl AuditService {
.catch_unwind()
.await;
// Always release the lock before returning the connection to the pool.
let _ = sqlx::query("DO RELEASE_LOCK(?)")
.bind(AUDIT_LOCK_NAME)
.execute(&mut *conn)
crates/sprout-auth/src/nip42.rs
-1
View File
@@ -27,7 +27,6 @@ fn normalize_relay_url(raw: &str) -> String {
let _ = parsed.set_host(Some("127.0.0.1"));
}
}
// Remove trailing slash from the path component.
let path = parsed.path().trim_end_matches('/').to_string();
parsed.set_path(&path);
parsed.to_string()
crates/sprout-auth/src/okta.rs
-2
View File
@@ -204,7 +204,6 @@ impl JwksCache {
ttl_secs: u64,
client: &reqwest::Client,
) -> Result<CachedJwks, AuthError> {
// Fast path: read lock, return if fresh.
{
let guard = self.inner.read().await;
if let Some(cached) = guard.as_ref() {
@@ -244,7 +243,6 @@ impl JwksCache {
fetched_at: Instant::now(),
};
// Re-acquire write lock to store the result.
// Final re-check: another thread may have stored a fresh entry while
// we were fetching. If so, discard our result and return theirs.
let mut guard = self.inner.write().await;
crates/sprout-core/src/filter.rs
-4
View File
@@ -91,21 +91,18 @@ mod tests {
let past = Timestamp::from(now_ts.as_u64() - 3600);
let future = Timestamp::from(now_ts.as_u64() + 3600);
// kind
assert!(filters_match(&[Filter::new().kind(Kind::TextNote)], &ev));
assert!(!filters_match(
&[Filter::new().kind(Kind::ContactList)],
&ev
));
// author
assert!(filters_match(&[Filter::new().author(pubkey)], &ev));
assert!(!filters_match(
&[Filter::new().author(Keys::generate().public_key())],
&ev
));
// compound AND
assert!(filters_match(
&[Filter::new().kind(Kind::TextNote).author(pubkey)],
&ev
@@ -115,7 +112,6 @@ mod tests {
&ev
));
// since / until
assert!(filters_match(&[Filter::new().since(past)], &ev));
assert!(!filters_match(&[Filter::new().since(future)], &ev));
assert!(filters_match(&[Filter::new().until(future)], &ev));
crates/sprout-db/src/channel.rs
-7
View File
@@ -204,9 +204,6 @@ pub async fn create_channel(
let id = Uuid::new_v4();
let id_bytes = id.as_bytes().as_slice().to_vec();
// Use a transaction so the INSERT + SELECT are atomic. Without this, a concurrent
// reader could see the channel between the insert and the fetch, or the channel
// could be modified before we read it back.
let mut tx = pool.begin().await?;
sqlx::query(
@@ -329,9 +326,6 @@ pub async fn add_member(
let channel_id_bytes = channel_id.as_bytes().as_slice().to_vec();
// Begin transaction: all role checks and the INSERT run atomically.
// This prevents a TOCTOU race where the inviter is removed between the
// role check and the INSERT.
let mut tx = pool.begin().await?;
let channel = get_channel_tx(&mut tx, channel_id).await?;
@@ -514,7 +508,6 @@ pub async fn get_members(pool: &MySqlPool, channel_id: Uuid) -> Result<Vec<Membe
///
/// Includes channels where the pubkey is an active member AND all open channels.
/// Open channels must be included in REQ filter resolution.
/// Returns IDs of all channels accessible to the given pubkey.
pub async fn get_accessible_channel_ids(pool: &MySqlPool, pubkey: &[u8]) -> Result<Vec<Uuid>> {
let rows = sqlx::query(
r#"
crates/sprout-db/src/feed.rs
-15
View File
@@ -72,12 +72,10 @@ pub async fn query_mentions(
.push_bind(serde_json::json!([["p", pubkey_hex]]).to_string())
.push(", '$')");
// Kinds: stream messages, stream replies, forum posts, forum comments
qb.push(format!(
" AND kind IN ({KIND_STREAM_MESSAGE}, {KIND_STREAM_MESSAGE_V2}, {KIND_FORUM_POST}, {KIND_FORUM_COMMENT})"
));
// Channel access filter
if !accessible_channel_ids.is_empty() {
qb.push(" AND channel_id IN (");
let mut sep = qb.separated(", ");
@@ -130,16 +128,12 @@ pub async fn query_needs_action(
" AND kind IN ({KIND_WORKFLOW_APPROVAL_REQUESTED}, {KIND_STREAM_REMINDER})"
));
// Tag filter: must be tagged to this user.
// Wrap in outer array so MySQL checks for exact sub-array membership — see
// query_mentions for a full explanation of the JSON_CONTAINS semantics.
qb.push(" AND JSON_CONTAINS(tags, ")
.push_bind(serde_json::json!([["p", pubkey_hex]]).to_string())
.push(", '$')");
// Access control: only return events from channels the user can access.
// Identical pattern to query_mentions — prevents leaking events from
// channels the user has been removed from.
if !accessible_channel_ids.is_empty() {
qb.push(" AND channel_id IN (");
let mut sep = qb.separated(", ");
@@ -183,8 +177,6 @@ pub async fn query_activity(
FROM events WHERE 1=1",
);
// Stream messages, forum posts, agent job events.
// KIND_JOB_REQUEST = agent job requested, KIND_JOB_PROGRESS = in-flight progress update, KIND_JOB_RESULT = completed result.
qb.push(format!(
" AND kind IN ({KIND_STREAM_MESSAGE}, {KIND_STREAM_MESSAGE_V2}, {KIND_FORUM_POST}, {KIND_JOB_REQUEST}, {KIND_JOB_PROGRESS}, {KIND_JOB_RESULT})"
));
@@ -233,7 +225,6 @@ mod tests {
#[test]
fn pubkey_hex_encoding_32_byte_key() {
// Simulate a full 32-byte Nostr pubkey.
let pubkey_bytes: Vec<u8> = (0u8..32).collect();
let hex = hex::encode(&pubkey_bytes);
assert_eq!(hex.len(), 64);
@@ -308,8 +299,6 @@ mod tests {
use sprout_core::kind::{
KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_V2,
};
// query_mentions filters for: KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_V2,
// KIND_FORUM_POST, KIND_FORUM_COMMENT
let mention_kinds: &[u32] = &[
KIND_STREAM_MESSAGE,
KIND_STREAM_MESSAGE_V2,
@@ -338,7 +327,6 @@ mod tests {
#[test]
fn needs_action_query_includes_approval_and_reminder_kinds() {
use sprout_core::kind::{KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED};
// query_needs_action filters for: KIND_WORKFLOW_APPROVAL_REQUESTED, KIND_STREAM_REMINDER
let needs_action_kinds: &[u32] = &[KIND_WORKFLOW_APPROVAL_REQUESTED, KIND_STREAM_REMINDER];
assert!(
@@ -357,8 +345,6 @@ mod tests {
KIND_FORUM_POST, KIND_JOB_PROGRESS, KIND_JOB_REQUEST, KIND_JOB_RESULT,
KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_V2,
};
// query_activity filters for: KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_V2,
// KIND_FORUM_POST, KIND_JOB_REQUEST, KIND_JOB_PROGRESS, KIND_JOB_RESULT
let activity_kinds: &[u32] = &[
KIND_STREAM_MESSAGE,
KIND_STREAM_MESSAGE_V2,
@@ -423,7 +409,6 @@ mod tests {
KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_V2, KIND_STREAM_REMINDER,
KIND_WORKFLOW_APPROVAL_REQUESTED,
};
// The two queries serve different purposes — their kind sets should not overlap.
let needs_action_kinds: &[u32] = &[KIND_WORKFLOW_APPROVAL_REQUESTED, KIND_STREAM_REMINDER];
let activity_kinds: &[u32] = &[
KIND_STREAM_MESSAGE,
crates/sprout-db/src/lib.rs
-5
View File
@@ -804,7 +804,6 @@ mod tests {
assert_eq!(channel.description, Some("desc".to_string()));
assert!(db.is_member(channel.id, &owner).await.unwrap());
// Add member via owner invite
db.add_member(
channel.id,
&member,
@@ -818,7 +817,6 @@ mod tests {
let members = db.get_members(channel.id).await.expect("get members");
assert_eq!(members.len(), 2);
// Owner removes member
db.remove_member(channel.id, &member, &owner)
.await
.expect("remove");
@@ -919,17 +917,14 @@ mod tests {
.await
.expect("add rando");
// Rando cannot remove member
let result = db.remove_member(channel.id, &member, &rando).await;
assert!(matches!(result, Err(DbError::AccessDenied(_))));
// Owner can remove member
db.remove_member(channel.id, &member, &owner)
.await
.expect("owner removes");
assert!(!db.is_member(channel.id, &member).await.unwrap());
// Member can remove themselves
db.remove_member(channel.id, &rando, &rando)
.await
.expect("self-remove");
crates/sprout-db/src/workflow.rs
-3
View File
@@ -940,14 +940,12 @@ mod tests {
let mut cloned = record.clone();
cloned.name = "Cloned".to_owned();
// Original is unchanged.
assert_eq!(record.name, "Original");
assert_eq!(cloned.name, "Cloned");
}
#[test]
fn workflow_record_status_variants() {
// Verify all WorkflowStatus variants can be stored in the struct.
let now = Utc::now();
for status in &[
WorkflowStatus::Active,
@@ -1093,7 +1091,6 @@ mod tests {
created_at: now,
};
// Trace is a JSON array with 2 entries.
assert!(record.execution_trace.is_array());
assert_eq!(record.execution_trace.as_array().unwrap().len(), 2);
}
crates/sprout-huddle/src/lib.rs
-1
View File
@@ -139,7 +139,6 @@ mod tests {
HuddleService::create_room_name(id),
"sprout-550e8400-e29b-41d4-a716-446655440000"
);
// Deterministic
assert_eq!(
HuddleService::create_room_name(id),
HuddleService::create_room_name(id)
crates/sprout-huddle/src/webhook.rs
-7
View File
@@ -186,15 +186,12 @@ mod tests {
#[test]
fn test_webhook_event_variants() {
// room_started
let ev = signed_parse(r#"{"event":"room_started","room":{"name":"r1"}}"#).unwrap();
assert_eq!(ev, WebhookEvent::RoomStarted { room: "r1".into() });
// room_finished
let ev = signed_parse(r#"{"event":"room_finished","room":{"name":"r1"}}"#).unwrap();
assert_eq!(ev, WebhookEvent::RoomFinished { room: "r1".into() });
// participant_joined
let ev = signed_parse(
r#"{"event":"participant_joined","room":{"name":"r1"},"participant":{"identity":"alice"}}"#,
)
@@ -207,7 +204,6 @@ mod tests {
}
);
// participant_left
let ev = signed_parse(
r#"{"event":"participant_left","room":{"name":"r1"},"participant":{"identity":"alice"}}"#,
)
@@ -220,7 +216,6 @@ mod tests {
}
);
// track_published — audio (default)
let ev = signed_parse(
r#"{"event":"track_published","room":{"name":"r1"},"participant":{"identity":"alice"},"track":{"type":"audio"}}"#,
)
@@ -234,7 +229,6 @@ mod tests {
}
);
// track_published — video
let ev = signed_parse(
r#"{"event":"track_published","room":{"name":"r1"},"participant":{"identity":"alice"},"track":{"type":"video"}}"#,
)
@@ -248,7 +242,6 @@ mod tests {
}
);
// track_published — screen_share
let ev = signed_parse(
r#"{"event":"track_published","room":{"name":"r1"},"participant":{"identity":"alice"},"track":{"type":"screen_share"}}"#,
)
crates/sprout-mcp/src/relay_client.rs
-3
View File
@@ -435,13 +435,11 @@ impl RelayClient {
let new_inner =
Self::connect_with_retry(&self.relay_url, &self.keys, self.api_token.as_deref()).await;
// Swap the inner connection.
{
let mut inner = self.inner.lock().await;
*inner = new_inner;
}
// Resubscribe to all active subscriptions.
let subs = self.active_subscriptions.lock().await.clone();
if !subs.is_empty() {
tracing::info!("resubscribing to {} active subscription(s)", subs.len());
@@ -879,7 +877,6 @@ mod tests {
let text = "not json at all";
let result = parse_relay_message(text);
assert!(result.is_err());
// Should be a JSON parse error.
assert!(matches!(result.unwrap_err(), RelayClientError::Json(_)));
}
crates/sprout-mcp/src/server.rs
-2
View File
@@ -239,12 +239,10 @@ impl SproutMcpServer {
description = "Send a message to a Sprout channel"
)]
pub async fn send_message(&self, Parameters(p): Parameters<SendMessageParams>) -> String {
// Validate channel_id is a well-formed UUID at the tool boundary.
if let Err(e) = validate_uuid(&p.channel_id) {
return format!("Error: {e}");
}
// Guard against excessively large message content.
if p.content.len() > MAX_CONTENT_BYTES {
return format!(
"Error: content exceeds maximum size of {} bytes (got {})",
crates/sprout-proxy/src/invite.rs
-2
View File
@@ -74,7 +74,6 @@ mod tests {
#[test]
fn test_invite_token_validation() {
// Valid token
let token = InviteToken::new("tok-valid", vec![], future(3600), 5);
assert!(token.validate(Utc::now()).is_ok());
assert!(token.is_valid(Utc::now()));
@@ -108,7 +107,6 @@ mod tests {
// Still valid (uses < max_uses)
assert!(token.is_valid(Utc::now()));
token.consume();
// Now exhausted
assert!(!token.is_valid(Utc::now()));
}
crates/sprout-relay/src/api/agents.rs
-6
View File
@@ -42,7 +42,6 @@ pub async fn agents_handler(
.await
.map_err(|e| internal_error(&format!("db error: {e}")))?;
// Collect pubkeys for bulk presence lookup.
let mut pubkeys_for_presence: Vec<nostr::PublicKey> = Vec::new();
let mut bot_pubkey_hexes: Vec<String> = Vec::new();
@@ -64,7 +63,6 @@ pub async fn agents_handler(
Default::default()
});
// Fetch user records for name resolution.
let user_records = state
.db
.get_users_bulk(&bots.iter().map(|b| b.pubkey.clone()).collect::<Vec<_>>())
@@ -85,7 +83,6 @@ pub async fn agents_handler(
let mut result = Vec::with_capacity(bots.len());
for (bot, hex) in bots.iter().zip(bot_pubkey_hexes.iter()) {
// Resolve display name: users table → bot record → test mapping → fallback.
let name = user_name_map
.get(hex.as_str())
.cloned()
@@ -95,7 +92,6 @@ pub async fn agents_handler(
format!("agent-{}", &hex[..end])
});
// Parse channel names from comma-separated string, filtered to requester's access.
let channels: Vec<&str> = bot
.channel_names
.split(',')
@@ -103,7 +99,6 @@ pub async fn agents_handler(
.filter(|s| !s.is_empty() && accessible_names.contains(*s))
.collect();
// Parse capabilities from JSON value.
let capabilities: Vec<String> = bot
.capabilities
.as_ref()
@@ -115,7 +110,6 @@ pub async fn agents_handler(
})
.unwrap_or_default();
// Presence status.
let status = presence_map
.get(hex.as_str())
.map(|s| s.as_str())
crates/sprout-relay/src/api/channels.rs
-1
View File
@@ -151,7 +151,6 @@ async fn resolve_dm_participants(
let member_pubkeys: Vec<Vec<u8>> = members.iter().map(|m| m.pubkey.clone()).collect();
// Bulk-fetch user records for name resolution.
let user_records = state
.db
.get_users_bulk(&member_pubkeys)
crates/sprout-relay/src/api/feed.rs
-7
View File
@@ -62,14 +62,12 @@ pub async fn feed_handler(
.and_then(|ts| DateTime::from_timestamp(ts, 0))
.unwrap_or_else(|| Utc::now() - Duration::days(7));
// Parse optional type filter.
let type_filter: Option<std::collections::HashSet<&str>> = params
.types
.as_deref()
.map(|t| t.split(',').map(|s| s.trim()).collect());
let wants = |cat: &str| -> bool { type_filter.as_ref().is_none_or(|f| f.contains(cat)) };
// 1. Get accessible channel IDs for this user.
let accessible_ids = state
.db
.get_accessible_channel_ids(&pubkey_bytes)
@@ -93,7 +91,6 @@ pub async fn feed_handler(
})));
}
// 2. Run queries in parallel.
let (mentions_res, needs_action_res, activity_res) = tokio::join!(
state
.db
@@ -111,12 +108,10 @@ pub async fn feed_handler(
let needs_action = needs_action_res.map_err(|e| internal_error(&format!("db error: {e}")))?;
let activity_all = activity_res.map_err(|e| internal_error(&format!("db error: {e}")))?;
// 3. Partition activity into agent activity vs channel activity.
let (agent_activity, channel_activity): (Vec<_>, Vec<_>) = activity_all
.into_iter()
.partition(|e| AGENT_KINDS.contains(&event_kind_u32(&e.event)));
// 4. Enrich events with channel names (batch lookup).
let all_channels = state.db.list_channels(None).await.unwrap_or_else(|e| {
tracing::warn!("feed: failed to load channel names for enrichment: {e}");
vec![]
@@ -124,7 +119,6 @@ pub async fn feed_handler(
let channel_name_map: HashMap<uuid::Uuid, String> =
all_channels.into_iter().map(|c| (c.id, c.name)).collect();
// Helper: convert a StoredEvent to a FeedItem JSON value.
let to_feed_item = |event: &sprout_core::StoredEvent, category: &str| -> serde_json::Value {
let channel_name = event
.channel_id
@@ -155,7 +149,6 @@ pub async fn feed_handler(
})
};
// 5. Build feed sections (apply type filter).
let mentions_items: Vec<serde_json::Value> = if wants("mentions") {
mentions
.iter()
crates/sprout-relay/src/api/mod.rs
-1
View File
@@ -201,7 +201,6 @@ pub(crate) async fn check_channel_access(
if is_member {
return Ok(());
}
// Not an explicit member — check if channel is open.
let is_open = state
.db
.get_channel(channel_id)
crates/sprout-relay/src/api/presence.rs
-2
View File
@@ -52,8 +52,6 @@ pub async fn presence_handler(
.await
.unwrap_or_default();
// Build result: pubkey_hex → status. Include "offline" for any requested
// pubkey not found in the presence map.
let mut result = serde_json::Map::new();
for pk in &pubkeys {
let hex = pk.to_hex();
crates/sprout-relay/src/api/search.rs
-2
View File
@@ -39,7 +39,6 @@ pub async fn search_handler(
let query_str = params.q.unwrap_or_default();
let per_page = params.limit.unwrap_or(20).min(100);
// Get accessible channel IDs to scope the search.
let channel_ids = state
.db
.get_accessible_channel_ids(&pubkey_bytes)
@@ -74,7 +73,6 @@ pub async fn search_handler(
}
};
// Enrich hits with channel names.
let all_channels = state.db.list_channels(None).await.unwrap_or_default();
let channel_name_map: HashMap<String, String> = all_channels
.into_iter()
crates/sprout-relay/src/handlers/event.rs
-2
View File
@@ -449,13 +449,11 @@ async fn derive_reaction_channel(
None => return ReactionChannelResult::NoTarget,
};
// Decode hex to bytes for DB lookup (already validated as 64-char hex above)
let id_bytes = match hex::decode(&target_hex) {
Ok(b) if b.len() == 32 => b,
_ => return ReactionChannelResult::NoTarget,
};
// Look up the target event to get its channel_id
match db.get_event_by_id(&id_bytes).await {
Ok(Some(target_event)) => {
if let Some(ch_id) = target_event.channel_id {
crates/sprout-relay/src/handlers/req.rs
+1 -8
View File
@@ -121,7 +121,6 @@ pub async fn handle_req(
};
for stored in &events {
// Deduplicate across filters by event ID.
if !seen_ids.insert(stored.event.id) {
continue;
}
@@ -255,7 +254,6 @@ mod tests {
fn test_extract_channel_id_mixed_channels_returns_none() {
let channel_a = uuid::Uuid::new_v4();
let channel_b = uuid::Uuid::new_v4();
// Two filters each with a different channel ID — can't index under one channel.
let filters = vec![
filter_with_channel(channel_a),
filter_with_channel(channel_b),
@@ -265,7 +263,6 @@ mod tests {
#[test]
fn test_extract_channel_id_no_channel_tag_returns_none() {
// A filter with no channel tag means "global subscription".
let filters = vec![Filter::new()];
assert_eq!(extract_channel_id_from_filters(&filters), None);
}
@@ -274,16 +271,12 @@ mod tests {
fn test_extract_channel_id_one_filter_missing_channel_returns_none() {
// Even if one filter has a channel, a second filter without one makes it global.
let channel_id = uuid::Uuid::new_v4();
let filters = vec![
filter_with_channel(channel_id),
Filter::new(), // no channel tag → global
];
let filters = vec![filter_with_channel(channel_id), Filter::new()];
assert_eq!(extract_channel_id_from_filters(&filters), None);
}
#[test]
fn test_extract_channel_id_same_channel_multiple_filters() {
// Two filters both scoped to the same channel → returns that channel.
let channel_id = uuid::Uuid::new_v4();
let filters = vec![
filter_with_channel(channel_id),
crates/sprout-relay/src/main.rs
-1
View File
@@ -87,7 +87,6 @@ async fn main() -> anyhow::Result<()> {
let workflow_config = sprout_workflow::WorkflowConfig::default();
let workflow_engine = Arc::new(WorkflowEngine::new(db.clone(), workflow_config));
// Spawn cron scheduler background task.
let wf_cron = Arc::clone(&workflow_engine);
tokio::spawn(async move { wf_cron.run().await });
crates/sprout-relay/src/protocol.rs
-1
View File
@@ -290,7 +290,6 @@ mod tests {
fn parse_req_too_many_filters_is_rejected() {
let filter = Filter::new().kind(Kind::TextNote);
let filter_val = serde_json::to_value(&filter).unwrap();
// Build a REQ with MAX_FILTERS_PER_REQ + 1 filters.
let mut arr: Vec<serde_json::Value> = vec![
serde_json::Value::String("REQ".to_string()),
serde_json::Value::String("sub3".to_string()),
crates/sprout-relay/src/router.rs
-1
View File
@@ -77,7 +77,6 @@ async fn nip11_or_ws_handler(
return Json(info).into_response();
}
// Try WebSocket upgrade from the raw request.
match WebSocketUpgrade::from_request(req, &state).await {
Ok(ws) => ws
.on_upgrade(move |socket| handle_connection(socket, state, addr))
crates/sprout-relay/src/subscription.rs
-9
View File
@@ -606,7 +606,6 @@ mod tests {
let conn_id = Uuid::new_v4();
let channel_id = Uuid::new_v4();
// Build a filter with an explicit empty kinds list.
let filter_empty_kinds = Filter::new().kinds(vec![] as Vec<Kind>);
registry.register(
conn_id,
@@ -615,13 +614,11 @@ mod tests {
Some(channel_id),
);
// The subscription should not appear in the wildcard index.
assert!(
registry.channel_wildcard_index.get(&channel_id).is_none(),
"kinds:[] sub must NOT be in the wildcard index"
);
// The subscription should not appear in any kind-specific index.
let key = IndexKey {
channel_id,
kind: Kind::TextNote,
@@ -631,7 +628,6 @@ mod tests {
"kinds:[] sub must NOT be in the kind-specific index"
);
// Fan-out should produce zero matches for any event kind.
let event = make_stored_event(Kind::TextNote, Some(channel_id));
let matches = registry.fan_out(&event);
assert!(
@@ -657,7 +653,6 @@ mod tests {
let conn_id = Uuid::new_v4();
let channel_id = Uuid::new_v4();
// Register a global (channel-less) subscription that matches all TextNote events.
registry.register(
conn_id,
"global_sub".to_string(),
@@ -665,7 +660,6 @@ mod tests {
None, // global — no channel scope
);
// A channel-scoped event must NOT be delivered to the global subscription.
let channel_event = make_stored_event(Kind::TextNote, Some(channel_id));
let matches = registry.fan_out(&channel_event);
assert!(
@@ -674,7 +668,6 @@ mod tests {
matches
);
// A non-channel event SHOULD still be delivered to the global subscription.
let global_event = make_stored_event(Kind::TextNote, None);
let matches = registry.fan_out(&global_event);
assert_eq!(
@@ -700,10 +693,8 @@ mod tests {
Some(channel_id),
);
// Should not panic.
registry.remove_subscription(conn_id, "sub_empty");
// Indexes remain clean.
assert!(registry.channel_wildcard_index.get(&channel_id).is_none());
let key = IndexKey {
channel_id,
crates/sprout-search/src/collection.rs
-3
View File
@@ -29,7 +29,6 @@ pub async fn ensure_collection(
api_key: &str,
collection_name: &str,
) -> Result<(), SearchError> {
// First, check if the collection already exists.
let check_url = format!("{}/collections/{}", base_url, collection_name);
let resp = client
.get(&check_url)
@@ -43,7 +42,6 @@ pub async fn ensure_collection(
return Ok(());
}
404 => {
// Collection doesn't exist — create it.
debug!(
collection = collection_name,
"Collection not found, creating"
@@ -55,7 +53,6 @@ pub async fn ensure_collection(
}
}
// Create the collection.
let schema = events_schema(collection_name);
let create_url = format!("{}/collections", base_url);
let resp = client
crates/sprout-test-client/src/lib.rs
-1
View File
@@ -14,7 +14,6 @@ use tokio::time::timeout;
use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
use tracing::debug;
// Re-export shared relay wire types from sprout-mcp.
pub use sprout_mcp::relay_client::{parse_relay_message, OkResponse, RelayMessage};
/// Errors returned by [`SproutTestClient`] operations.
crates/sprout-test-client/tests/e2e_mcp.rs
-6
View File
@@ -134,7 +134,6 @@ impl McpSession {
continue;
}
// Check this is our response.
if v["id"] == json!(id) {
return v;
}
@@ -416,12 +415,10 @@ async fn test_mcp_send_and_read_message() {
"get_channel_history returned an error: {history_text}"
);
// The history should be a JSON array of events.
let events: Vec<Value> = serde_json::from_str(&history_text).unwrap_or_else(|e| {
panic!("get_channel_history response is not valid JSON array: {e}\n{history_text}")
});
// Find our message in the history.
let found = events
.iter()
.any(|ev| ev["content"].as_str().unwrap_or("").contains(&unique_token));
@@ -566,7 +563,6 @@ async fn test_mcp_create_and_trigger_workflow() {
return;
}
// Parse the created workflow to get its ID.
let workflow: Value = serde_json::from_str(&create_text).unwrap_or_else(|e| {
panic!("create_workflow response is not valid JSON: {e}\n{create_text}")
});
@@ -577,7 +573,6 @@ async fn test_mcp_create_and_trigger_workflow() {
assert!(!workflow_id.is_empty(), "workflow id must not be empty");
// Verify the workflow name matches.
assert_eq!(
workflow["name"].as_str().unwrap_or(""),
workflow_name,
@@ -679,7 +674,6 @@ async fn test_mcp_create_and_trigger_workflow() {
"expected at least one run after triggering workflow '{workflow_id}'"
);
// Verify our run is in the list.
let found_run = runs.iter().any(|r| r["id"].as_str() == Some(run_id));
assert!(
found_run,
crates/sprout-test-client/tests/e2e_relay.rs
-3
View File
@@ -205,7 +205,6 @@ async fn test_close_subscription_stops_delivery() {
.await
.expect("EOSE");
// Close the subscription.
client
.close_subscription(&sid)
.await
@@ -302,7 +301,6 @@ async fn test_multiple_concurrent_clients() {
.expect("EOSE");
}
// Client 0 sends the event.
let content = format!("broadcast-{}", uuid::Uuid::new_v4());
let ok = clients[0]
.send_text_message(&keys[0], &channel, &content, kind)
@@ -342,7 +340,6 @@ async fn test_stored_events_returned_before_eose() {
.await
.expect("connect");
// Send an event first.
let content = format!("stored-{}", uuid::Uuid::new_v4());
let ok = client
.send_text_message(&keys, &channel, &content, kind)
crates/sprout-test-client/tests/e2e_rest_api.rs
-18
View File
@@ -111,7 +111,6 @@ async fn test_list_channels_returns_expected_fields() {
"expected at least one channel in the list"
);
// Every channel must have the required fields.
for ch in channels {
assert!(ch.get("id").is_some(), "channel missing 'id' field");
assert!(ch.get("name").is_some(), "channel missing 'name' field");
@@ -269,7 +268,6 @@ async fn test_search_returns_results_for_open_channels() {
let hits = body["hits"].as_array().expect("'hits' must be an array");
// Every hit must have the required fields.
for hit in hits {
assert!(hit.get("event_id").is_some(), "hit missing 'event_id'");
assert!(hit.get("content").is_some(), "hit missing 'content'");
@@ -288,11 +286,9 @@ async fn test_search_returns_indexed_event() {
let pubkey_hex = keys.public_key().to_hex();
let ws_url = relay_ws_url();
// Send a message with a unique token via WebSocket so it gets indexed.
let unique_token = format!("e2e-search-{}", uuid::Uuid::new_v4().simple());
let content = format!("E2E REST search test marker: {unique_token}");
// Connect and send the message to an open channel.
let mut ws_client = SproutTestClient::connect(&ws_url, &keys)
.await
.expect("WebSocket connect failed");
@@ -313,7 +309,6 @@ async fn test_search_returns_indexed_event() {
// Wait briefly for the search index to catch up.
tokio::time::sleep(Duration::from_millis(500)).await;
// Search for the unique token.
// The unique_token is UUID simple format (hex only) — safe to use directly in the URL.
let url = format!("{}/api/search?q={unique_token}", relay_http_url());
let resp = authed_get(&client, &url, &pubkey_hex).await;
@@ -328,7 +323,6 @@ async fn test_search_returns_indexed_event() {
"expected at least one search hit for unique token '{unique_token}'"
);
// The first hit should contain our unique token.
let first_content = hits[0]["content"].as_str().unwrap_or("");
assert!(
first_content.contains(&unique_token),
@@ -383,7 +377,6 @@ async fn test_presence_set_and_query() {
let pubkey_hex = keys.public_key().to_hex();
let ws_url = relay_ws_url();
// Send a presence event via WebSocket.
let mut ws_client = SproutTestClient::connect(&ws_url, &keys)
.await
.expect("WebSocket connect failed");
@@ -401,7 +394,6 @@ async fn test_presence_set_and_query() {
// Keep the WebSocket connection alive briefly so presence is registered.
tokio::time::sleep(Duration::from_millis(200)).await;
// Query presence via REST.
let url = format!("{}/api/presence?pubkeys={pubkey_hex}", relay_http_url());
let resp = authed_get(&client, &url, &pubkey_hex).await;
@@ -414,7 +406,6 @@ async fn test_presence_set_and_query() {
"expected 'online' after sending presence event"
);
// Clean up: send offline presence.
let offline_event = nostr::EventBuilder::new(Kind::Custom(20001), "offline", [])
.sign_with_keys(&keys)
.expect("event sign failed");
@@ -444,7 +435,6 @@ async fn test_presence_bulk_query() {
let body: serde_json::Value = resp.json().await.expect("JSON");
assert!(body.is_object(), "presence response must be an object");
// Both pubkeys should appear in the response.
assert!(
body.get(&pk_a).is_some(),
"pk_a missing from presence response"
@@ -454,7 +444,6 @@ async fn test_presence_bulk_query() {
"pk_b missing from presence response"
);
// Both should be offline.
assert_eq!(body[&pk_a].as_str(), Some("offline"));
assert_eq!(body[&pk_b].as_str(), Some("offline"));
}
@@ -499,7 +488,6 @@ async fn test_agents_list() {
.as_array()
.expect("/api/agents must return a JSON array");
// Every agent must have the required fields.
for agent in agents {
assert!(agent.get("pubkey").is_some(), "agent missing 'pubkey'");
assert!(agent.get("name").is_some(), "agent missing 'name'");
@@ -559,7 +547,6 @@ async fn test_agents_scoped_to_accessible_channels() {
let agents: Vec<serde_json::Value> = resp.json().await.expect("JSON");
// Get accessible channels for this user.
let channels_url = format!("{}/api/channels", relay_http_url());
let channels_resp = authed_get(&client, &channels_url, &pubkey_hex).await;
let channels: Vec<serde_json::Value> = channels_resp.json().await.expect("JSON");
@@ -632,17 +619,14 @@ async fn test_feed_returns_activity() {
// Small delay to let the event propagate.
tokio::time::sleep(Duration::from_millis(200)).await;
// Fetch the feed.
let resp = authed_get(&client, &url, &pubkey_hex).await;
assert_eq!(resp.status(), 200, "expected 200 OK from /api/feed");
let body: serde_json::Value = resp.json().await.expect("response must be JSON");
// Top-level structure.
let feed = body.get("feed").expect("response missing 'feed' key");
let meta = body.get("meta").expect("response missing 'meta' key");
// Feed sections must exist.
assert!(feed.get("mentions").is_some(), "feed missing 'mentions'");
assert!(
feed.get("needs_action").is_some(),
@@ -654,7 +638,6 @@ async fn test_feed_returns_activity() {
"feed missing 'agent_activity'"
);
// Meta fields.
assert!(meta.get("since").is_some(), "meta missing 'since'");
assert!(meta.get("total").is_some(), "meta missing 'total'");
assert!(
@@ -662,7 +645,6 @@ async fn test_feed_returns_activity() {
"meta missing 'generated_at'"
);
// Activity must be an array.
assert!(
feed["activity"].is_array(),
"feed 'activity' must be an array"
crates/sprout-test-client/tests/e2e_workflows.rs
-11
View File
@@ -157,7 +157,6 @@ async fn test_list_workflows_empty_channel() {
let body: serde_json::Value = resp.json().await.expect("response must be JSON");
assert!(body.is_array(), "expected JSON array, got: {body}");
// Every workflow in the list must have required fields.
let arr = body.as_array().unwrap();
for wf in arr {
assert!(wf.get("id").is_some(), "workflow missing 'id' field");
@@ -180,7 +179,6 @@ async fn test_create_and_list_workflow() {
let yaml = webhook_workflow_yaml("e2e-create-list-test");
let created = create_workflow(&client, &base, pubkey_hex, CHANNEL_GENERAL, &yaml).await;
// Response must include id, name, channel_id, definition fields.
let workflow_id = created["id"]
.as_str()
.expect("created workflow must have 'id'");
@@ -199,7 +197,6 @@ async fn test_create_and_list_workflow() {
"webhook workflow must return 'webhook_secret' on creation"
);
// Verify it appears in the list.
let list_url = format!("{base}/api/channels/{CHANNEL_GENERAL}/workflows");
let list_resp = client
.get(&list_url)
@@ -216,7 +213,6 @@ async fn test_create_and_list_workflow() {
"newly created workflow {workflow_id} not found in list"
);
// Clean up.
let status = delete_workflow(&client, &base, pubkey_hex, workflow_id).await;
assert_eq!(status, 204, "cleanup DELETE should return 204");
}
@@ -235,7 +231,6 @@ async fn test_trigger_workflow_and_check_run() {
let pubkey_hex: &str = SEEDED_PUBKEY;
let base = relay_http_url();
// Create a webhook workflow.
let yaml = webhook_workflow_yaml("e2e-trigger-test");
let created = create_workflow(&client, &base, pubkey_hex, CHANNEL_GENERAL, &yaml).await;
let workflow_id = created["id"]
@@ -243,7 +238,6 @@ async fn test_trigger_workflow_and_check_run() {
.expect("workflow must have 'id'")
.to_string();
// Manually trigger the workflow via POST /api/workflows/:id/trigger.
let trigger_url = format!("{base}/api/workflows/{workflow_id}/trigger");
let trigger_resp = client
.post(&trigger_url)
@@ -297,7 +291,6 @@ async fn test_trigger_workflow_and_check_run() {
let run = found_run.expect("run must appear in GET /api/workflows/:id/runs within 1 second");
// Run must have required fields.
assert!(run.get("id").is_some(), "run missing 'id'");
assert!(
run.get("workflow_id").is_some(),
@@ -305,14 +298,12 @@ async fn test_trigger_workflow_and_check_run() {
);
assert!(run.get("status").is_some(), "run missing 'status'");
// Status must be one of the valid terminal or in-progress values.
let status = run["status"].as_str().unwrap_or("");
assert!(
matches!(status, "pending" | "running" | "completed" | "failed"),
"run status '{status}' is not a recognized value"
);
// Clean up.
let del_status = delete_workflow(&client, &base, pubkey_hex, &workflow_id).await;
assert_eq!(del_status, 204, "cleanup DELETE should return 204");
}
@@ -402,7 +393,6 @@ steps:
"run status '{status}' is not a recognized value"
);
// Clean up.
let _ = ws_client.disconnect().await;
let del_status = delete_workflow(&client, &base, pubkey_hex, &workflow_id).await;
assert_eq!(del_status, 204, "cleanup DELETE should return 204");
@@ -520,7 +510,6 @@ steps:
"run status '{status}' is not a recognized value"
);
// Clean up.
let _ = ws_client.disconnect().await;
let del_status = delete_workflow(&client, &base, pubkey_hex, &workflow_id).await;
assert_eq!(del_status, 204, "cleanup DELETE should return 204");
crates/sprout-workflow/src/executor.rs
+1 -18
View File
@@ -73,7 +73,6 @@ pub fn resolve_template(
trigger_ctx: &TriggerContext,
step_outputs: &HashMap<String, JsonValue>,
) -> Result<String, WorkflowError> {
// Fast path: no template markers.
if !template.contains("{{") {
return Ok(template.to_owned());
}
@@ -82,11 +81,9 @@ pub fn resolve_template(
let mut remaining = template;
while let Some(start) = remaining.find("{{") {
// Append everything before the `{{`.
result.push_str(&remaining[..start]);
remaining = &remaining[start + 2..];
// Find the closing `}}`.
let end = match remaining.find("}}") {
Some(e) => e,
None => {
@@ -105,10 +102,8 @@ pub fn resolve_template(
let var_path = parts.next().unwrap_or("").trim();
let filter = parts.next().map(|s| s.trim());
// Resolve the variable.
let raw_value = resolve_variable(var_path, trigger_ctx, step_outputs);
// Apply filter (if any).
let value = match (raw_value, filter) {
(Some(v), Some(f)) => apply_filter(v, f)?,
(Some(v), None) => v,
@@ -124,7 +119,6 @@ pub fn resolve_template(
result.push_str(&value);
}
// Append any trailing text after the last `}}`.
result.push_str(remaining);
Ok(result)
}
@@ -141,7 +135,6 @@ fn resolve_variable(
// Pattern: `steps.STEP_ID.output.FIELD`
if let Some(rest) = path.strip_prefix("steps.") {
// rest = "STEP_ID.output.FIELD"
let mut parts = rest.splitn(3, '.');
let step_id = parts.next()?;
let middle = parts.next()?; // must be "output"
@@ -184,7 +177,6 @@ fn json_to_string(v: &JsonValue) -> String {
fn apply_filter(value: String, filter: &str) -> Result<String, WorkflowError> {
let filter = filter.trim();
// `truncate(N)` — truncate to N characters.
if let Some(inner) = filter
.strip_prefix("truncate(")
.and_then(|s| s.strip_suffix(')'))
@@ -331,7 +323,6 @@ pub fn build_eval_context(
}
// ── Step outputs ──────────────────────────────────────────────────────────
// Register as `steps_STEP_ID_output_FIELD`.
for (step_id, output) in step_outputs {
if let JsonValue::Object(map) = output {
@@ -589,7 +580,6 @@ pub async fn dispatch_action(
"RequestApproval from={from} timeout={timeout_str}: {message}"
);
// Generate an approval token.
let token = generate_approval_token(run_id, step_id);
// TODO (WF-08): create approval record in DB, emit kind:46010.
@@ -723,7 +713,6 @@ async fn call_webhook_impl(
use std::time::Duration;
// ── SSRF check ────────────────────────────────────────────────────────────
// Parse the URL to extract host and port before making any connection.
let parsed_url = reqwest::Url::parse(url)
.map_err(|e| WorkflowError::WebhookError(format!("invalid URL: {e}")))?;
@@ -857,7 +846,6 @@ pub async fn execute_run(
)
})?;
// Mark run as Running now that we have a permit.
engine
.db
.update_workflow_run(
@@ -969,7 +957,6 @@ async fn execute_steps(
continue;
}
// 1. Evaluate `if:` condition.
if let Some(expr) = &step.if_expr {
match evaluate_condition(expr, trigger_ctx, &step_outputs).await {
Ok(true) => {
@@ -994,7 +981,6 @@ async fn execute_steps(
}
}
// 2. Resolve template variables.
let resolved_action = match resolve_step_templates(step, trigger_ctx, &step_outputs) {
Ok(a) => a,
Err(e) => {
@@ -1006,7 +992,6 @@ async fn execute_steps(
}
};
// 3. Dispatch action (with per-step timeout).
let timeout_secs = step
.timeout_secs
.unwrap_or(engine.config.default_timeout_secs);
@@ -1139,8 +1124,7 @@ mod tests {
let ctx = make_trigger();
let out =
resolve_template("{{trigger.text | truncate(5)}}", &ctx, &HashMap::new()).unwrap();
assert_eq!(out, "P1 in"); // "P1 incident in production" truncated to 5 chars = "P1 in"
// Actually "P1 in" is 5 chars: 'P','1',' ','i','n'
assert_eq!(out, "P1 in");
assert_eq!(out.chars().count(), 5);
}
@@ -1153,7 +1137,6 @@ mod tests {
&HashMap::new(),
)
.unwrap();
// "abc123def456" → "abc123...def456"
assert_eq!(out, "abc123...def456");
}
crates/sprout-workflow/src/lib.rs
-4
View File
@@ -249,7 +249,6 @@ impl WorkflowEngine {
continue;
}
// Create the workflow_run row (status: pending).
let trigger_event_id_bytes = event.event.id.as_bytes().to_vec();
let run_id = match self
.db
@@ -314,7 +313,6 @@ async fn should_fire_workflow(
trigger_ctx: &executor::TriggerContext,
workflow_id: uuid::Uuid,
) -> bool {
// Enforce reaction emoji filter.
if let TriggerDef::ReactionAdded {
emoji: Some(ref expected),
} = def.trigger
@@ -330,7 +328,6 @@ async fn should_fire_workflow(
}
}
// Evaluate trigger filter expression (MessagePosted only).
if let TriggerDef::MessagePosted {
filter: Some(ref expr),
} = def.trigger
@@ -465,7 +462,6 @@ steps:
let (def, json) = WorkflowEngine::parse_yaml(yaml).expect("parse failed");
assert_eq!(def.name, "Test Workflow");
// JSON must round-trip.
let reparsed: WorkflowDef = serde_json::from_str(&json).expect("json round-trip");
assert_eq!(reparsed.name, def.name);
assert_eq!(reparsed.steps.len(), 1);
crates/sprout-workflow/src/schema.rs
-9
View File
@@ -175,7 +175,6 @@ impl WorkflowDef {
&& id.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
};
// Step IDs must be unique and non-empty.
let mut seen_ids: HashSet<&str> = HashSet::new();
for step in &self.steps {
if step.id.trim().is_empty() {
@@ -197,28 +196,23 @@ impl WorkflowDef {
}
}
// Validate schedule trigger fields.
if let TriggerDef::Schedule { cron, interval } = &self.trigger {
// Must have cron or interval (not neither).
if cron.is_none() && interval.is_none() {
return Err(WorkflowError::InvalidDefinition(
"schedule trigger requires either 'cron' or 'interval'".into(),
));
}
// Must not have both cron and interval simultaneously.
if cron.is_some() && interval.is_some() {
return Err(WorkflowError::InvalidDefinition(
"schedule trigger cannot specify both 'cron' and 'interval'; use one or the other".into(),
));
}
// Validate cron expression syntax.
if let Some(expr) = cron {
validate_cron(expr)?;
}
// Validate interval format (e.g. "30m", "1h", "60s").
if let Some(dur) = interval {
crate::executor::parse_duration_secs(dur).map_err(|_| {
WorkflowError::InvalidDefinition(format!(
@@ -290,7 +284,6 @@ mod tests {
assert_eq!(def.steps.len(), 1);
assert_eq!(def.steps[0].id, "notify");
// Canonical JSON must round-trip.
let reparsed: WorkflowDef = serde_json::from_str(&json).expect("json round-trip");
assert_eq!(reparsed.name, def.name);
}
@@ -360,7 +353,6 @@ mod tests {
let (def, _) = parse_yaml(yaml).expect("parse failed");
assert_eq!(def.steps.len(), 7);
// Verify each action type deserialized correctly.
assert!(matches!(
&def.steps[0].action,
ActionDef::SendMessage { .. }
@@ -672,7 +664,6 @@ mod tests {
);
let (def, json) = parse_yaml(yaml).expect("parse failed");
// Round-trip through JSON.
let reparsed: WorkflowDef = serde_json::from_str(&json).expect("json round-trip");
assert_eq!(reparsed.name, def.name);