test(relay): cover system channels in the workflows API and push matcher (#8509)

🤖
## Summary
Test-only. This adds end-to-end coverage for the last two system channel
read paths that had none. It changes no behavior.

A system channel is a channel type the relay leaves out of ordinary
reads. A client sees one only when it names the channel or opts in. The
system channels spec lists every read path that must follow this rule.
The Postgres end-to-end tests already covered subscriptions, counts,
search, the HTTP bridge and membership notices, but not these two:

- **Workflows API.** A workflow names its channel, so reading a workflow
in a system channel follows ordinary channel access. Members can read
it. Outsiders can read it only when the channel is open. For a private
one, an outsider gets 404, the same answer as for a missing workflow.
- **Push notifications.** This is database-backed matcher coverage: the
test stores real channel rows, memberships and push leases, loads the
matcher's context from the database, and runs the production matcher. It
builds the match batch directly, so it does not cover the outbox or
gateway delivery. A member's push filter that names no channel gets a
wake for a message in an ordinary channel, but not for a message in a
system channel. A member's filter that names the system channel by `#h`
gets a wake for that channel only (the rule since #8517). A non-member's
filter gets neither.

### Related issue
Part of the system channels spec (risks section: end-to-end cases for
the workflows API and push). Rebased on #8507 and #8517, which made push
follow the read rule: a system channel pushes only to a member's filter
that names it by `#h`. No duplicate PR found.

### Testing
- Both tests pass against local Postgres and Redis on this branch.
- Each test fails when the behavior it covers is removed: without the
named-channel check in the workflows API, a member gets 404 for its own
system channel's workflow; if the push matcher lets a filter that does
not name a system channel match it, the member's broad lease also wakes;
if the matcher never pushes system channels, the named lease does not
wake.
- clippy (`--tests -D warnings`) and fmt pass.

---------

Signed-off-by: Logan Johnson <loganj@squareup.com>
Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>

crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs

@@ -1216,3 +1216,107 @@ async fn system_channels_are_found_by_type_or_creator() {
         "bridge /count outsider #t:system"
     );
 }
+
+/// GET a workflow's runs as `reader`, returning the response status.
+async fn workflow_runs_status(f: &Fixture, reader: &Keys, workflow: Uuid) -> u16 {
+    use axum::body::Body;
+    use axum::http::{header, Request};
+    use tower::ServiceExt;
+
+    crate::router::build_router(f.state.clone())
+        .oneshot(
+            Request::get(format!("/workflows/{workflow}/runs"))
+                .header(header::HOST, &f.host)
+                .header("x-pubkey", reader.public_key().to_hex())
+                .body(Body::empty())
+                .expect("request"),
+        )
+        .await
+        .expect("router")
+        .status()
+        .as_u16()
+}
+
+/// A workflow names its channel, so the workflows API reads a workflow in a
+/// system channel under the ordinary rules: members read it, outsiders read
+/// it only when the channel is open, and a private one answers 404 like a
+/// missing workflow.
+/// Mutation: drop the `get_readable_named_only_channels` check from
+/// `authorize_workflow_read` → RED (members get 404 for their own system
+/// channel's workflow).
+#[tokio::test]
+#[ignore = "requires Postgres and Redis"]
+async fn workflows_in_system_channels_follow_channel_access() {
+    let f = fixture().await;
+    let owner = Keys::generate();
+    let outsider = Keys::generate();
+
+    let private = create_channel(&f.state, &f.tenant, &owner, &[&["channel_type", "system"]]).await;
+    let open = create_channel(&f.state, &f.tenant, &owner, &[&["channel_type", "system"]]).await;
+    edit_channel(
+        &f.state,
+        &f.tenant,
+        &owner,
+        open,
+        &[&["visibility", "open"]],
+    )
+    .await
+    .expect("open the system channel");
+    let ordinary = create_channel(&f.state, &f.tenant, &owner, &[&["visibility", "private"]]).await;
+
+    let owner_bytes = owner.public_key().to_bytes();
+    f.state
+        .db
+        .ensure_user(f.tenant.community(), &owner_bytes)
+        .await
+        .expect("workflow owner user");
+    let workflow = |channel: Uuid| {
+        let state = f.state.clone();
+        let community = f.tenant.community();
+        async move {
+            state
+                .db
+                .create_workflow(
+                    community,
+                    Some(channel),
+                    &owner_bytes,
+                    &format!("wf-{}", channel.simple()),
+                    "{}",
+                    &[0; 32],
+                )
+                .await
+                .expect("create workflow")
+        }
+    };
+    let (private_wf, open_wf, ordinary_wf) = (
+        workflow(private).await,
+        workflow(open).await,
+        workflow(ordinary).await,
+    );
+
+    let cases = [
+        (&owner, private_wf, 200, "member, private system channel"),
+        (&owner, open_wf, 200, "member, open system channel"),
+        (&owner, ordinary_wf, 200, "member, ordinary channel"),
+        (
+            &outsider,
+            private_wf,
+            404,
+            "outsider, private system channel",
+        ),
+        (&outsider, open_wf, 200, "outsider, open system channel"),
+        (
+            &outsider,
+            ordinary_wf,
+            404,
+            "outsider, private ordinary channel",
+        ),
+    ];
+    for (reader, workflow, expected, case) in cases {
+        assert_eq!(
+            workflow_runs_status(&f, reader, workflow).await,
+            expected,
+            "{case}"
+        );
+    }
+}

crates/buzz-relay/src/push_runtime.rs

@@ -1007,3 +1007,129 @@ mod shutdown_tests {
         assert!(dropped.load(Ordering::SeqCst));
     }
 }
+
+#[cfg(test)]
+mod postgres_tests {
+    use super::{load_match_context, match_job};
+    use nostr::{EventBuilder, Kind, Tag};
+
+    /// The matcher on real channel rows: a member's lease that names no
+    /// channel wakes for an ordinary channel's message but not for a system
+    /// channel's. A member's lease that names the system channel by `#h`
+    /// wakes for that channel only. A non-member's lease wakes for neither.
+    /// Mutations: let system channel events reach a lease that does not name
+    /// them → RED (the outsider or the broad lease wakes); never push system
+    /// channels → RED (the named lease does not wake).
+    #[tokio::test]
+    #[ignore = "requires Postgres and Redis"]
+    async fn system_channel_messages_wake_only_named_member_leases() {
+        use crate::handlers::group_state_postgres_tests::{create_channel, http_auth};
+        use crate::handlers::ingest::ingest_event;
+
+        let state = crate::api::bridge::postgres_tests::bridge_handler_test_state()
+            .await
+            .expect("local Postgres and Redis");
+        let host = format!("push-system-{}.local", uuid::Uuid::new_v4().simple());
+        let community = state
+            .db
+            .ensure_configured_community(&host)
+            .await
+            .expect("ensure community")
+            .id;
+        let tenant = buzz_core::TenantContext::resolved(community, &host);
+        let member = nostr::Keys::generate();
+        let outsider = nostr::Keys::generate();
+
+        let ordinary = create_channel(&state, &tenant, &member, &[]).await;
+        let system = create_channel(&state, &tenant, &member, &[&["channel_type", "system"]]).await;
+
+        let unnamed = serde_json::json!([{"filter": {"kinds": [9]}, "class": "default"}]);
+        let named = serde_json::json!([{
+            "filter": {"kinds": [9], "#h": [system.to_string()]},
+            "class": "default"
+        }]);
+        let leases = [
+            (&member, "unnamed", &unnamed),
+            (&member, "named", &named),
+            (&outsider, "unnamed", &unnamed),
+        ];
+        for (n, (keys, install, subscriptions)) in leases.into_iter().enumerate() {
+            let source = [n as u8 + 1; 32];
+            assert_eq!(
+                buzz_db::push::replace_active_lease(
+                    state.db.pool(),
+                    community,
+                    &keys.public_key().to_bytes(),
+                    install,
+                    buzz_db::push::LeaseVersion {
+                        source_event_id: &source,
+                        source_created_at: 10,
+                        generation: 1,
+                        expires_at: i64::MAX / 2,
+                    },
+                    buzz_db::push::ActiveLease {
+                        endpoint_hash: &[n as u8 + 42; 32],
+                        endpoint_grant: "opaque-grant",
+                        max_class: "default",
+                        subscriptions,
+                    },
+                )
+                .await
+                .expect("activate lease"),
+                buzz_db::push::ReplaceLeaseOutcome::Accepted
+            );
+        }
+
+        let mut jobs = Vec::new();
+        for channel in [ordinary, system] {
+            let event = EventBuilder::new(Kind::Custom(9), "hello")
+                .tag(Tag::parse(["h", &channel.to_string()]).unwrap())
+                .sign_with_keys(&member)
+                .unwrap();
+            let id = event.id;
+            ingest_event(&state, &tenant, event, http_auth(&member))
+                .await
+                .expect("ingest message");
+            let stored = state
+                .db
+                .get_event_by_id(community, id.as_bytes())
+                .await
+                .expect("load message")
+                .expect("stored message");
+            assert_eq!(stored.channel_id, Some(channel));
+            jobs.push(buzz_db::push::BatchedMatch {
+                event: stored,
+                attempt: 1,
+            });
+        }
+        let batch = buzz_db::push::ClaimedMatchBatch {
+            community,
+            claim_id: uuid::Uuid::new_v4(),
+            jobs,
+        };
+
+        let context = load_match_context(&state, &batch)
+            .await
+            .expect("match context");
+        let woken: Vec<Vec<(Vec<u8>, String)>> = batch
+            .jobs
+            .iter()
+            .map(|job| {
+                match_job(job, &context)
+                    .expect("match")
+                    .into_iter()
+                    .map(|wake| (wake.author, wake.installation_id))
+                    .collect()
+            })
+            .collect();
+        let member_key = member.public_key().to_bytes().to_vec();
+        assert_eq!(
+            woken,
+            vec![
+                vec![(member_key.clone(), "unnamed".to_owned())],
+                vec![(member_key, "named".to_owned())],
+            ],
+            "ordinary channel wakes its member's broad lease; system channel wakes only the lease that names it"
+        );
+    }
+}