jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-31 14:31:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(workflows): scope overview in database

Browse Source 
Resolve the Desktop workflow overview against the authenticated user's active channel memberships in PostgreSQL, with bounded keyset pagination. Remove the multi-channel request array and SQL pushdown path while preserving exact single-channel reads.

Signed-off-by: npub13n66s06epmqf2kc3v373ez8hj65cuzyvxzjf93vwpervxqn2u7jq2qd9je <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz>
This commit is contained in:
npub13n66s06epmqf2kc3v373ez8hj65cuzyvxzjf93vwpervxqn2u7jq2qd9je
2026-07-24 14:51:09 -04:00
committed by Larry
parent 1559c122bf
commit 0e244686e8
9 changed files with 212 additions and 198 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
crates/buzz-db/src/event.rs
View File

@@ -74,11 +74,10 @@ pub struct EventQuery {
/// channel-less global events. Applied before SQL `LIMIT` so access-filtered
/// historical pages have exact exhaustion semantics.
pub channel_ids: Option<Vec<uuid::Uuid>>,
/// Restrict results to events in exactly these channels. Unlike
/// `channel_ids`, this excludes channel-less global events. Used for
/// multi-value `#h` filters so unrelated accessible channels and global
/// events cannot consume the SQL `LIMIT` before NIP-01 post-filtering.
pub exact_channel_ids: Option<Vec<uuid::Uuid>>,
/// Restrict results to events in channels where this pubkey has an active
/// membership. The membership is resolved by PostgreSQL with an indexed
/// `EXISTS` subquery, avoiding client-supplied channel-ID arrays.
pub member_channel_pubkey: Option<Vec<u8>>,
/// Override the default page clamp ([`DEFAULT_MAX_PAGE_LIMIT`]). Used by
/// the COUNT fallback path, which needs to fetch all matching events for
/// post-filter counting. When None, the default clamp applies.
@@ -127,7 +126,7 @@ impl EventQuery {
ids: None,
e_tags: None,
channel_ids: None,
exact_channel_ids: None,
member_channel_pubkey: None,
max_limit: None,
shared_gated_reader: None,
}
@@ -410,7 +409,7 @@ pub(crate) async fn query_events_on(
qb.push(format!(" AND {col_prefix}channel_id IS NULL"));
}
// Multi-channel array pushdown: restrict to events in any of these channels
// Multi-channel IN pushdown: restrict to events in any of these channels
// OR global events (channel_id IS NULL). Used by NIP-45 COUNT to enforce
// channel access at the SQL level without fetching all rows.
//
@@ -422,23 +421,41 @@ pub(crate) async fn query_events_on(
qb.push(format!(" AND {col_prefix}channel_id IS NULL"));
} else {
qb.push(format!(
" AND ({col_prefix}channel_id IS NULL OR {col_prefix}channel_id = ANY("
" AND ({col_prefix}channel_id IS NULL OR {col_prefix}channel_id IN ("
));
qb.push_bind(ch_ids.clone());
let mut sep = qb.separated(", ");
for ch in ch_ids {
sep.push_bind(*ch);
}
qb.push("))");
}
}
// Exact multi-channel pushdown for NIP-01 `#h` filters. Unlike the access
// scope above, an `#h` filter never matches channel-less global events.
if let Some(ref ch_ids) = q.exact_channel_ids {
if ch_ids.is_empty() {
qb.push(" AND FALSE");
// Membership-scoped overview queries resolve channel access inside
// PostgreSQL. The active-membership index starts with (community_id,
// pubkey), and the channel PK check completes the lookup without shipping
// every channel UUID through the client and relay.
if let Some(ref member_pubkey) = q.member_channel_pubkey {
let outer_event_prefix = if q.p_tag_hex.is_some() {
"e."
} else {
qb.push(format!(" AND {col_prefix}channel_id = ANY("));
qb.push_bind(ch_ids.clone());
qb.push(")");
}
"events."
};
qb.push(format!(
" AND {outer_event_prefix}channel_id IS NOT NULL AND EXISTS (\
SELECT 1 FROM channel_members cm \
JOIN channels c \
ON c.community_id = cm.community_id AND c.id = cm.channel_id \
WHERE cm.community_id = {outer_event_prefix}community_id \
AND cm.channel_id = {outer_event_prefix}channel_id \
AND cm.pubkey = "
));
qb.push_bind(member_pubkey.clone());
qb.push(
" AND cm.removed_at IS NULL \
AND c.deleted_at IS NULL \
AND (c.channel_type != 'dm' OR cm.hidden_at IS NULL))",
);
}
if let Some(ks) = q.kinds.as_deref().filter(|k| !k.is_empty()) {
@@ -685,32 +702,23 @@ pub(crate) async fn count_events_on(conn: &mut sqlx::PgConnection, q: &EventQuer
qb.push(format!(" AND {col_prefix}channel_id IS NULL"));
}
// Multi-channel array pushdown for COUNT: restrict to accessible channels + global.
// Multi-channel IN pushdown for COUNT: restrict to accessible channels + global.
// SECURITY: Some(empty vec) = no channel access → global events only.
if let Some(ref ch_ids) = q.channel_ids {
if ch_ids.is_empty() {
qb.push(format!(" AND {col_prefix}channel_id IS NULL"));
} else {
qb.push(format!(
" AND ({col_prefix}channel_id IS NULL OR {col_prefix}channel_id = ANY("
" AND ({col_prefix}channel_id IS NULL OR {col_prefix}channel_id IN ("
));
qb.push_bind(ch_ids.clone());
let mut sep = qb.separated(", ");
for ch in ch_ids {
sep.push_bind(*ch);
}
qb.push("))");
}
}
// Exact multi-channel pushdown for NIP-01 `#h` filters. Unlike the access
// scope above, an `#h` filter never matches channel-less global events.
if let Some(ref ch_ids) = q.exact_channel_ids {
if ch_ids.is_empty() {
qb.push(" AND FALSE");
} else {
qb.push(format!(" AND {col_prefix}channel_id = ANY("));
qb.push_bind(ch_ids.clone());
qb.push(")");
}
}
if let Some(ks) = q.kinds.as_deref().filter(|k| !k.is_empty()) {
qb.push(format!(" AND {col_prefix}kind IN ("));
let mut sep = qb.separated(", ");
@@ -1951,80 +1959,53 @@ mod tests {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn exact_channel_scope_is_applied_before_limit_and_excludes_globals() {
async fn member_channel_scope_is_applied_before_limit_and_excludes_non_members() {
let pool = setup_pool().await;
let community_uuid = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_uuid);
let requested_a = make_test_channel(&pool, community_uuid, None).await;
let requested_b = make_test_channel(&pool, community_uuid, None).await;
let unrequested = make_test_channel(&pool, community_uuid, None).await;
let base = 1_800_100_000;
let member_channel = make_test_channel(&pool, community_uuid, None).await;
let nonmember_channel = make_test_channel(&pool, community_uuid, None).await;
let member_pubkey = vec![42_u8; 32];
sqlx::query(
"INSERT INTO channel_members (community_id, channel_id, pubkey) VALUES ($1, $2, $3)",
)
.bind(community_uuid)
.bind(member_channel)
.bind(&member_pubkey)
.execute(&pool)
.await
.expect("insert active membership");
// Newer rows outside the exact #h set must not consume the SQL page.
let base = 1_800_100_000;
for offset in 10..13 {
let event = make_event_at(39_001, "newer unrequested", base + offset);
insert_event(&pool, community, &event, Some(unrequested))
let event = make_event_at(30_620, "newer nonmember", base + offset);
insert_event(&pool, community, &event, Some(nonmember_channel))
.await
.expect("insert unrequested candidate");
.expect("insert nonmember workflow");
}
let global = make_event_at(39_001, "newer global", base + 9);
let global = make_event_at(30_620, "newer global", base + 9);
insert_event(&pool, community, &global, None)
.await
.expect("insert global candidate");
let requested_newer = make_event_at(39_001, "requested a", base + 2);
insert_event(&pool, community, &requested_newer, Some(requested_a))
.expect("insert global workflow");
let member_workflow = make_event_at(30_620, "older member", base + 1);
insert_event(&pool, community, &member_workflow, Some(member_channel))
.await
.expect("insert requested a candidate");
let requested_older = make_event_at(39_001, "requested b", base + 1);
insert_event(&pool, community, &requested_older, Some(requested_b))
.await
.expect("insert requested b candidate");
.expect("insert member workflow");
let events = query_events(
&pool,
&EventQuery {
kinds: Some(vec![39_001]),
exact_channel_ids: Some(vec![requested_a, requested_b]),
kinds: Some(vec![30_620]),
member_channel_pubkey: Some(member_pubkey),
limit: Some(2),
..EventQuery::for_community(community)
},
)
.await
.expect("query exact-channel page");
.expect("query member workflow page");
assert_eq!(events.len(), 2, "exact-channel page must fill before EOF");
assert_eq!(events[0].event.id, requested_newer.id);
assert_eq!(events[1].event.id, requested_older.id);
assert!(
events.iter().all(|stored| stored.channel_id.is_some()),
"an explicit #h list must exclude channel-less global events"
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn empty_exact_channel_scope_matches_nothing() {
let pool = setup_pool().await;
let community_uuid = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_uuid);
let channel = make_test_channel(&pool, community_uuid, None).await;
let event = make_event_at(39_002, "must not match", 1_800_200_000);
insert_event(&pool, community, &event, Some(channel))
.await
.expect("insert candidate");
let events = query_events(
&pool,
&EventQuery {
kinds: Some(vec![39_002]),
exact_channel_ids: Some(vec![]),
..EventQuery::for_community(community)
},
)
.await
.expect("query empty exact-channel scope");
assert!(events.is_empty());
assert_eq!(events.len(), 1);
assert_eq!(events[0].event.id, member_workflow.id);
}
fn make_text_event(content: &str) -> nostr::Event {

79
crates/buzz-relay/src/api/bridge.rs
View File

@@ -1004,6 +1004,85 @@ async fn query_events_authed(
));
}
// `member_channels` is an authenticated HTTP-bridge extension for global
// overviews of channel-scoped resources. Keep it deliberately narrow: the
// only current contract is workflow definitions, and mixed filters are
// rejected rather than falling back to the generic membership-array path.
let member_workflow_overview = raw_filters
.iter()
.zip(filters.iter())
.filter(|(raw, _)| extension_flag(raw, "member_channels"))
.collect::<Vec<_>>();
if !member_workflow_overview.is_empty() {
if member_workflow_overview.len() != filters.len() {
return Err(api_error(
StatusCode::BAD_REQUEST,
"member_channels filters cannot be mixed with ordinary filters",
));
}
let mut overview_events = Vec::new();
for (raw, filter) in member_workflow_overview {
let workflow_only = filter.kinds.as_ref().is_some_and(|kinds| {
kinds.len() == 1 && kinds.iter().all(|kind| kind.as_u16() == 30_620)
});
if !workflow_only || extract_channel_from_filter(filter).is_some() {
return Err(api_error(
StatusCode::BAD_REQUEST,
"member_channels requires kinds:[30620] without #h",
));
}
let mut query = crate::handlers::req::build_event_query_from_filter(
filter,
&pubkey_bytes,
state,
tenant.community(),
)
.await;
query.member_channel_pubkey = Some(pubkey_bytes.clone());
match extract_before_id(raw) {
BeforeId::Malformed => {
return Err(api_error(
StatusCode::BAD_REQUEST,
"before_id must be a 64-char hex event id",
));
}
BeforeId::Valid(before_id) => {
if query.until.is_none() {
return Err(api_error(
StatusCode::BAD_REQUEST,
"before_id requires until to be set",
));
}
query.before_id = Some(before_id);
}
BeforeId::Absent => {}
}
let stored_events = state
.db
.query_events(&query)
.await
.map_err(|e| internal_error(&format!("member workflow query error: {e}")))?;
for stored in stored_events {
if buzz_core::filter::filters_match(std::slice::from_ref(filter), &stored)
&& buzz_core::filter::reader_authorized_for_event(
&stored.event,
&authed_pubkey_hex,
)
{
overview_events.push(
serde_json::to_value(&stored.event)
.map_err(|e| internal_error(&format!("event serialize: {e}")))?,
);
}
}
}
return Ok(Json(Value::Array(overview_events)));
}
// Get channels this user can access — same enforcement as WS REQ handler.
let accessible_channels = state
.get_accessible_channel_ids_cached(tenant.community(), &pubkey_bytes)

63
crates/buzz-relay/src/handlers/req.rs
View File

@@ -985,23 +985,6 @@ fn filter_to_query_params(
(None, None)
};
// Push multi-value #h into an exact channel list. The caller intersects
// this with the authenticated viewer's accessible channels before the DB
// read; keeping the extraction here preserves per-filter NIP-01 semantics.
let h_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::H);
let exact_channel_ids = filter.generic_tags.get(&h_tag).and_then(|values| {
if values.len() > 1 {
Some(
values
.iter()
.filter_map(|value| value.parse::<uuid::Uuid>().ok())
.collect::<Vec<_>>(),
)
} else {
None
}
});
EventQuery {
channel_id,
kinds,
@@ -1015,7 +998,6 @@ fn filter_to_query_params(
authors,
ids,
e_tags,
exact_channel_ids,
..EventQuery::for_community(community)
}
}
@@ -1029,12 +1011,7 @@ pub(crate) fn apply_access_scope_to_query(
channel_id: Option<uuid::Uuid>,
accessible_channels: &[uuid::Uuid],
) {
if let Some(requested) = query.exact_channel_ids.as_mut() {
requested.retain(|channel| accessible_channels.contains(channel));
// The exact requested/access intersection is sufficient and excludes
// globals by contract; do not also install the broader access scope.
query.channel_ids = None;
} else if channel_id.is_none() {
if channel_id.is_none() {
query.channel_ids = Some(accessible_channels.to_vec());
}
}
@@ -1351,44 +1328,6 @@ mod tests {
assert_eq!(query.channel_id, Some(channel));
}
#[test]
fn multi_h_queries_intersect_requested_and_accessible_before_limit() {
let allowed = uuid::Uuid::new_v4();
let denied = uuid::Uuid::new_v4();
let unrelated = uuid::Uuid::new_v4();
let h_tag = SingleLetterTag::lowercase(Alphabet::H);
let filter = Filter::new().custom_tags(h_tag, [allowed.to_string(), denied.to_string()]);
let mut query = filter_to_query_params(
&filter,
None,
buzz_core::tenant::CommunityId::from_uuid(uuid::Uuid::new_v4()),
);
apply_access_scope_to_query(&mut query, None, &[allowed, unrelated]);
assert_eq!(query.exact_channel_ids, Some(vec![allowed]));
assert!(query.channel_ids.is_none());
}
#[test]
fn multi_h_query_with_no_authorized_channels_matches_no_channels() {
let requested_a = uuid::Uuid::new_v4();
let requested_b = uuid::Uuid::new_v4();
let h_tag = SingleLetterTag::lowercase(Alphabet::H);
let filter =
Filter::new().custom_tags(h_tag, [requested_a.to_string(), requested_b.to_string()]);
let mut query = filter_to_query_params(
&filter,
None,
buzz_core::tenant::CommunityId::from_uuid(uuid::Uuid::new_v4()),
);
apply_access_scope_to_query(&mut query, None, &[uuid::Uuid::new_v4()]);
assert_eq!(query.exact_channel_ids, Some(vec![]));
assert!(query.channel_ids.is_none());
}
/// S2 invariant: the bounded-concurrency pipeline (phase 2) must yield
/// per-filter results in original filter order even when an earlier
/// filter's DB query completes *after* a later one. `buffered` guarantees

56
desktop/src-tauri/src/commands/workflows.rs
View File

@@ -101,34 +101,42 @@ pub async fn get_channel_workflows(
Ok(events.iter().map(workflow_from_event).collect())
}
/// Fetch workflows across many channels in a single relay round-trip.
///
/// The Workflows overview screen previously issued one `get_channel_workflows`
/// query per member channel (`Promise.all` fanout in `WorkflowsView`), i.e. N
/// relay POSTs. A nostr `#h` filter matches ANY of its listed values, so one
/// query with all channel ids returns the same set. Each `WorkflowWire` carries
/// its own `channel_id` (from the event's `h` tag), so the frontend can still
/// group results by channel. Neither this nor the per-channel command sets a
/// `limit`, so batching does not change result completeness.
const WORKFLOW_OVERVIEW_PAGE_SIZE: usize = 500;
fn advance_workflow_cursor(filter: &mut Value, page: &[nostr::Event]) {
let last = page
.last()
.expect("a full workflow overview page always has a last event");
filter["until"] = serde_json::json!(last.created_at.as_secs());
filter["before_id"] = serde_json::json!(last.id.to_hex());
}
/// Fetch every workflow in channels where the authenticated user is an active
/// member. The relay resolves membership server-side; channel UUIDs never cross
/// the Desktop/relay boundary. Composite keyset pagination keeps each DB read
/// bounded while preserving workflows created in the same second.
#[tauri::command]
pub async fn get_channels_workflows(
channel_ids: Vec<String>,
pub async fn get_member_channel_workflows(
state: State<'_, AppState>,
) -> Result<Vec<WorkflowWire>, String> {
if channel_ids.is_empty() {
return Ok(Vec::new());
let mut filter = serde_json::json!({
"kinds": [30620],
"member_channels": true,
"limit": WORKFLOW_OVERVIEW_PAGE_SIZE,
});
let mut workflows = Vec::new();
loop {
let page = query_relay(&state, &[filter.clone()]).await?;
let done = page.len() < WORKFLOW_OVERVIEW_PAGE_SIZE;
if !done {
advance_workflow_cursor(&mut filter, &page);
}
workflows.extend(page.iter().map(workflow_from_event));
if done {
return Ok(workflows);
}
}
let events = query_relay(
&state,
&[serde_json::json!({
"kinds": [30620],
"#h": channel_ids,
})],
)
.await?;
Ok(events.iter().map(workflow_from_event).collect())
}
#[tauri::command]

17
desktop/src-tauri/src/commands/workflows_tests.rs
View File

@@ -188,6 +188,23 @@ fn workflow_wire_serializes_with_snake_case_keys() {
}
}
#[test]
fn workflow_overview_cursor_uses_timestamp_and_event_id() {
let first = wf_event(WF, CHAN, YAML);
let second = wf_event("33333333-3333-3333-3333-333333333333", CHAN, YAML);
let mut filter = serde_json::json!({
"kinds": [30620],
"member_channels": true,
"limit": WORKFLOW_OVERVIEW_PAGE_SIZE,
});
advance_workflow_cursor(&mut filter, &[first, second.clone()]);
assert_eq!(filter["until"], second.created_at.as_secs());
assert_eq!(filter["before_id"], second.id.to_hex());
assert!(filter.get("#h").is_none());
}
#[test]
fn trigger_response_uses_persisted_run_id_contract() {
let wire = trigger_wire_from_message(

2
desktop/src-tauri/src/lib.rs
View File

@@ -822,7 +822,7 @@ pub fn run() {
preview_team_snapshot_import,
confirm_team_snapshot_import,
get_channel_workflows,
get_channels_workflows,
get_member_channel_workflows,
get_workflow,
create_workflow,
update_workflow,

9
desktop/src/features/workflows/ui/WorkflowsView.tsx
View File

@@ -13,7 +13,7 @@ import { WorkflowDialog } from "@/features/workflows/ui/WorkflowDialog";
import type { Channel, Workflow } from "@/shared/api/types";
import {
deleteWorkflow,
getChannelsWorkflows,
getMemberChannelWorkflows,
triggerWorkflow,
} from "@/shared/api/tauriWorkflows";
import { Button } from "@/shared/ui/button";
@@ -86,12 +86,13 @@ export function WorkflowsView({
const allWorkflowsQuery = useQuery({
queryKey: allWorkflowsQueryKey(channelIdKey),
queryFn: async () => {
// Single batched relay query for all member channels, then group by the
// channel_id each workflow carries — replaces the per-channel fanout.
// The relay resolves active memberships server-side and returns bounded,
// keyset-paginated workflow pages. Desktop only retains channel names for
// presentation and never sends its full membership list.
const channelNameById = new Map(
memberChannels.map((channel) => [channel.id, channel.name]),
);
const workflows = await getChannelsWorkflows(channelIds);
const workflows = await getMemberChannelWorkflows();
const results: WorkflowWithChannel[] = [];
for (const workflow of workflows) {
results.push({

16
desktop/src/shared/api/tauriWorkflows.ts
View File

@@ -186,18 +186,12 @@ export async function getChannelWorkflows(
}
/**
* Fetch workflows across many channels in a single relay round-trip.
*
* Replaces the per-channel `Promise.all(getChannelWorkflows)` fanout on the
* Workflows overview: the backend `#h` filter matches any listed channel, and
* each returned workflow carries its own `channelId` so callers can group.
* Fetch workflows from every channel where the authenticated user is a member.
* Channel membership is resolved and paginated by the relay; Desktop does not
* send its full channel list.
*/
export async function getChannelsWorkflows(
channelIds: string[],
): Promise<Workflow[]> {
const raw = await invokeTauri<RawWorkflow[]>("get_channels_workflows", {
channelIds,
});
export async function getMemberChannelWorkflows(): Promise<Workflow[]> {
const raw = await invokeTauri<RawWorkflow[]>("get_member_channel_workflows");
return raw.map(fromRawWorkflow);
}

13
desktop/src/testing/e2eBridge.ts
View File

@@ -3389,11 +3389,8 @@ function handleGetChannelWorkflows(args: { channelId: string }) {
return mockWorkflows.filter((w) => w.channel_id === args.channelId);
}
function handleGetChannelsWorkflows(args: { channelIds: string[] }) {
const ids = new Set(args.channelIds);
return mockWorkflows.filter(
(w) => w.channel_id != null && ids.has(w.channel_id),
);
function handleGetMemberChannelWorkflows() {
return [...mockWorkflows];
}
function handleGetWorkflow(args: { workflowId: string }) {
@@ -12963,10 +12960,8 @@ export function maybeInstallE2eTauriMocks() {
return handleGetChannelWorkflows(
payload as Parameters<typeof handleGetChannelWorkflows>[0],
);
case "get_channels_workflows":
return handleGetChannelsWorkflows(
payload as Parameters<typeof handleGetChannelsWorkflows>[0],
);
case "get_member_channel_workflows":
return handleGetMemberChannelWorkflows();
case "get_workflow":
return handleGetWorkflow(
payload as Parameters<typeof handleGetWorkflow>[0],