diff --git a/crates/manager-core/src/group_pubsub_service.rs b/crates/manager-core/src/group_pubsub_service.rs index affafd4..9c3b56e 100644 --- a/crates/manager-core/src/group_pubsub_service.rs +++ b/crates/manager-core/src/group_pubsub_service.rs @@ -16,7 +16,10 @@ use std::sync::Arc; use async_trait::async_trait; -use picloud_shared::{GroupId, GroupPubsubError, GroupPubsubService, SdkCallCx, TriggerEvent}; +use picloud_shared::{ + GroupId, GroupPubsubError, GroupPubsubService, RealtimeBroadcaster, RealtimeEvent, SdkCallCx, + TriggerEvent, +}; use crate::authz::{self, AuthzRepo, Capability}; use crate::group_collection_repo::GroupCollectionResolver; @@ -31,6 +34,10 @@ pub struct GroupPubsubServiceImpl { resolver: Arc, authz: Arc, max_message_bytes: usize, + /// §11.6 D2: the realtime broadcaster, attached by the host via + /// [`Self::with_realtime`]. `None` in test bundles → no external SSE fan-out + /// (durable trigger fan-out is unaffected). + realtime: Option>, } impl GroupPubsubServiceImpl { @@ -45,9 +52,19 @@ impl GroupPubsubServiceImpl { resolver, authz, max_message_bytes: pubsub_max_message_bytes_from_env(), + realtime: None, } } + /// §11.6 D2: attach the realtime broadcaster so a shared-topic publish also + /// fans out to external SSE subscribers (in addition to the durable + /// `shared = true` trigger fan-out). Mirrors `PubsubServiceImpl::with_realtime`. + #[must_use] + pub fn with_realtime(mut self, broadcaster: Arc) -> Self { + self.realtime = Some(broadcaster); + self + } + /// The structural boundary: resolve the topic namespace to its owning group /// on the calling app's chain. `CollectionNotShared` when no ancestor group /// declares it — the same outcome a foreign app gets, by construction. @@ -107,10 +124,11 @@ impl GroupPubsubService for GroupPubsubServiceImpl { } else { format!("{namespace}.{subtopic}") }; + let published_at = chrono::Utc::now(); let event = TriggerEvent::Pubsub { topic: topic.clone(), - message, - published_at: chrono::Utc::now(), + message: message.clone(), + published_at, }; let payload = serde_json::to_value(&event) .map_err(|e| GroupPubsubError::Backend(format!("event serialize: {e}")))?; @@ -120,9 +138,183 @@ impl GroupPubsubService for GroupPubsubServiceImpl { trigger_depth: cx.trigger_depth, root_execution_id: cx.root_execution_id, }; - self.repo + // Durable trigger fan-out FIRST (the authoritative delivery path). + let count = self + .repo .fan_out_shared_publish(publish_ctx, group_id, &topic, payload) .await - .map_err(|e| GroupPubsubError::Backend(e.to_string())) + .map_err(|e| GroupPubsubError::Backend(e.to_string()))?; + // Then the best-effort realtime broadcast to external SSE subscribers, + // keyed by the OWNING GROUP so every subtree subscriber shares one + // channel. A broadcast failure never fails the publish. + if let Some(realtime) = &self.realtime { + realtime + .publish_group( + group_id, + &topic, + RealtimeEvent { + topic: topic.clone(), + message, + published_at, + }, + ) + .await; + } + Ok(count) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use picloud_shared::{ + AdminUserId, AppId, BroadcasterError, ExecutionId, InstanceRole, Principal, + RealtimeBroadcaster, RealtimeEvent, RequestId, ScriptId, + }; + use std::sync::Mutex; + use tokio::sync::broadcast; + + use crate::authz::AuthzError; + use crate::pubsub_repo::PubsubRepoError; + + /// Records `publish_group` calls so the bridge can be asserted. + #[derive(Default)] + struct RecordingBroadcaster { + group_publishes: Mutex>, + } + #[async_trait] + impl RealtimeBroadcaster for RecordingBroadcaster { + async fn subscribe( + &self, + _: picloud_shared::AppId, + _: &str, + ) -> Result, BroadcasterError> { + Ok(broadcast::channel(1).1) + } + async fn publish(&self, _: picloud_shared::AppId, _: &str, _: RealtimeEvent) {} + async fn drop_topic(&self, _: picloud_shared::AppId, _: &str) {} + async fn publish_group(&self, group_id: GroupId, topic: &str, event: RealtimeEvent) { + self.group_publishes + .lock() + .unwrap() + .push((group_id, topic.to_string(), event.message)); + } + } + + struct FakeRepo; + #[async_trait] + impl PubsubRepo for FakeRepo { + async fn fan_out_publish( + &self, + _: PublishCtx, + _: &str, + _: serde_json::Value, + ) -> Result { + Ok(0) + } + async fn fan_out_shared_publish( + &self, + _: PublishCtx, + _: GroupId, + _: &str, + _: serde_json::Value, + ) -> Result { + Ok(2) // pretend two shared triggers matched + } + } + + struct FixedResolver(GroupId); + #[async_trait] + impl GroupCollectionResolver for FixedResolver { + async fn resolve_owning_group( + &self, + _: AppId, + _: &str, + kind: &str, + ) -> Result, sqlx::Error> { + Ok((kind == KIND_TOPIC).then_some(self.0)) + } + } + + struct DenyAuthz; + #[async_trait] + impl AuthzRepo for DenyAuthz { + async fn membership( + &self, + _: picloud_shared::UserId, + _: AppId, + ) -> Result, AuthzError> { + Ok(None) + } + } + + fn owner_cx(app_id: AppId) -> SdkCallCx { + SdkCallCx { + app_id, + script_id: ScriptId::new(), + // Instance Owner bypasses the capability check (as in the group_kv + // tests), so DenyAuthz is fine — we're testing the broadcast bridge. + principal: Some(Principal { + user_id: AdminUserId::new(), + instance_role: InstanceRole::Owner, + scopes: None, + app_binding: None, + }), + execution_id: ExecutionId::new(), + request_id: RequestId::new(), + trigger_depth: 0, + root_execution_id: ExecutionId::new(), + is_dead_letter_handler: false, + event: None, + } + } + + #[tokio::test] + async fn publish_bridges_to_the_group_broadcaster() { + // §11.6 D2: a shared-topic publish fans out to the durable triggers AND, + // when a broadcaster is attached, to the OWNING GROUP's realtime channel + // with the full topic (`namespace.subtopic`). + let group = GroupId::new(); + let app = AppId::new(); + let recorder = Arc::new(RecordingBroadcaster::default()); + let svc = GroupPubsubServiceImpl::new( + Arc::new(FakeRepo), + Arc::new(FixedResolver(group)), + Arc::new(DenyAuthz), + ) + .with_realtime(recorder.clone()); + + let count = svc + .publish( + &owner_cx(app), + "events", + "created", + serde_json::json!({ "id": 1 }), + ) + .await + .unwrap(); + assert_eq!(count, 2, "durable fan-out count is returned"); + + let calls = recorder.group_publishes.lock().unwrap(); + assert_eq!(calls.len(), 1, "one realtime broadcast"); + assert_eq!(calls[0].0, group, "keyed by the owning group"); + assert_eq!(calls[0].1, "events.created", "full topic"); + assert_eq!(calls[0].2, serde_json::json!({ "id": 1 })); + } + + #[tokio::test] + async fn publish_without_broadcaster_still_fans_out() { + // No broadcaster attached → durable fan-out only, no panic. + let group = GroupId::new(); + let svc = GroupPubsubServiceImpl::new( + Arc::new(FakeRepo), + Arc::new(FixedResolver(group)), + Arc::new(DenyAuthz), + ); + let count = svc + .publish(&owner_cx(AppId::new()), "events", "", serde_json::json!({})) + .await + .unwrap(); + assert_eq!(count, 2); } } diff --git a/crates/manager-core/src/realtime_authority.rs b/crates/manager-core/src/realtime_authority.rs index 49fac35..c3e210d 100644 --- a/crates/manager-core/src/realtime_authority.rs +++ b/crates/manager-core/src/realtime_authority.rs @@ -22,11 +22,18 @@ use std::collections::HashMap; use std::sync::{Arc, Mutex}; use async_trait::async_trait; -use picloud_shared::{subscriber_token, AppId, RealtimeAuthority, SubscribeDenied, UsersService}; +use picloud_shared::{ + subscriber_token, AppId, GroupId, RealtimeAuthority, SubscribeDenied, UsersService, +}; use crate::app_secrets_repo::AppSecretsRepo; +use crate::group_collection_repo::GroupCollectionResolver; use crate::topic_repo::{TopicAuthMode, TopicRepo}; +/// The registry `kind` a shared topic is declared under (mirrors +/// `group_pubsub_service::KIND_TOPIC`). +const KIND_TOPIC: &str = "topic"; + /// F-S-008: TTL on cached signing-key entries. Once key rotation /// lands, every running process keeps accepting tokens signed by the /// old key until restart — bounded eviction is the simplest defence. @@ -37,6 +44,9 @@ pub struct RealtimeAuthorityImpl { topics: Arc, secrets: Arc, users: Arc, + /// §11.6 D2: resolves a shared-topic name to its owning group on the + /// subscriber app's ancestor chain (the isolation boundary). + collections: Arc, key_cache: Mutex)>>, } @@ -46,11 +56,13 @@ impl RealtimeAuthorityImpl { topics: Arc, secrets: Arc, users: Arc, + collections: Arc, ) -> Self { Self { topics, secrets, users, + collections, key_cache: Mutex::new(HashMap::new()), } } @@ -152,6 +164,24 @@ impl RealtimeAuthority for RealtimeAuthorityImpl { } } } + + async fn authorize_subscribe_shared( + &self, + app_id: AppId, + topic: &str, + ) -> Result { + // The declared collection is the topic's ROOT segment (`events.created` + // → `events`), matching the publish-side validation. Resolving it against + // the subscriber app's chain (kind=`topic`) is both the existence check + // and the authorization — reads are open to the subtree; a foreign app's + // chain never reaches the owning group, so it 404s (isolation boundary). + let root = topic.split('.').next().unwrap_or(topic); + self.collections + .resolve_owning_group(app_id, root, KIND_TOPIC) + .await + .map_err(|e| SubscribeDenied::Backend(e.to_string()))? + .ok_or(SubscribeDenied::NotFound) + } } #[cfg(test)] @@ -222,6 +252,26 @@ mod tests { } } + /// Fake shared-topic resolver: `Some((declared_name, owning_group))` resolves + /// that one topic name (kind=topic) to the group; everything else → None. + #[derive(Default)] + struct FakeCollections(Option<(String, GroupId)>); + #[async_trait] + impl GroupCollectionResolver for FakeCollections { + async fn resolve_owning_group( + &self, + _app_id: AppId, + name: &str, + kind: &str, + ) -> Result, sqlx::Error> { + Ok(self + .0 + .as_ref() + .filter(|(n, _)| n == name && kind == KIND_TOPIC) + .map(|(_, g)| *g)) + } + } + fn authority( topics: Vec<(AppId, Topic)>, key_app: AppId, @@ -231,9 +281,32 @@ mod tests { Arc::new(FakeTopics(topics)), Arc::new(FakeSecrets(key_app, key)), Arc::new(picloud_shared::NoopUsersService), + Arc::new(FakeCollections::default()), ) } + #[tokio::test] + async fn shared_topic_resolves_owning_group_or_404() { + let app = AppId::new(); + let group = GroupId::new(); + let auth = RealtimeAuthorityImpl::new( + Arc::new(FakeTopics(vec![])), + Arc::new(FakeSecrets(app, vec![0u8; 32])), + Arc::new(picloud_shared::NoopUsersService), + Arc::new(FakeCollections(Some(("events".into(), group)))), + ); + // The full topic's ROOT segment resolves; returns the OWNING group. + assert_eq!( + auth.authorize_subscribe_shared(app, "events.created").await, + Ok(group) + ); + // An undeclared topic name → NotFound (the isolation/existence boundary). + assert_eq!( + auth.authorize_subscribe_shared(app, "secrets").await, + Err(SubscribeDenied::NotFound) + ); + } + #[tokio::test] async fn missing_topic_is_not_found() { let app = AppId::new(); @@ -590,6 +663,7 @@ mod tests { app_id, user: stub_user(app_id), }), + Arc::new(FakeCollections::default()), ) } @@ -652,6 +726,7 @@ mod tests { app_id: app_a, user: stub_user(app_a), }), + Arc::new(FakeCollections::default()), ); assert_eq!( a.authorize_subscribe(app_b, "chat", Some("app-a-token")) diff --git a/crates/orchestrator-core/src/realtime.rs b/crates/orchestrator-core/src/realtime.rs index 2b6f4f0..bd30799 100644 --- a/crates/orchestrator-core/src/realtime.rs +++ b/crates/orchestrator-core/src/realtime.rs @@ -25,7 +25,7 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use async_trait::async_trait; -use picloud_shared::{AppId, BroadcasterError, RealtimeBroadcaster, RealtimeEvent}; +use picloud_shared::{AppId, BroadcasterError, GroupId, RealtimeBroadcaster, RealtimeEvent}; use tokio::sync::broadcast; /// Default per-channel broadcast buffer depth. @@ -37,6 +37,9 @@ pub const DEFAULT_GC_INTERVAL_SECS: u64 = 60; pub struct InProcessBroadcaster { inner: Mutex>>, + /// §11.6 D2: group SHARED topic channels, keyed by the OWNING group so every + /// subtree app subscribing to the same shared topic shares one channel. + groups: Mutex>>, capacity: usize, } @@ -45,6 +48,7 @@ impl InProcessBroadcaster { pub fn new(capacity: usize) -> Self { Self { inner: Mutex::new(HashMap::new()), + groups: Mutex::new(HashMap::new()), capacity: capacity.max(1), } } @@ -69,21 +73,29 @@ impl InProcessBroadcaster { Self::new(capacity) } - /// Number of live channels in the map (test/observability helper). + /// Number of live channels in both maps (test/observability helper). #[must_use] pub fn channel_count(&self) -> usize { - self.inner.lock().map(|g| g.len()).unwrap_or(0) + let app = self.inner.lock().map(|g| g.len()).unwrap_or(0); + let grp = self.groups.lock().map(|g| g.len()).unwrap_or(0); + app + grp } - /// Drop senders with zero receivers. Returns how many were removed. - /// Called periodically by [`spawn_realtime_gc`]. + /// Drop senders with zero receivers across both maps. Returns how many were + /// removed. Called periodically by [`spawn_realtime_gc`]. pub fn gc(&self) -> usize { - let Ok(mut g) = self.inner.lock() else { - return 0; - }; - let before = g.len(); - g.retain(|_, tx| tx.receiver_count() > 0); - before - g.len() + let mut removed = 0; + if let Ok(mut g) = self.inner.lock() { + let before = g.len(); + g.retain(|_, tx| tx.receiver_count() > 0); + removed += before - g.len(); + } + if let Ok(mut g) = self.groups.lock() { + let before = g.len(); + g.retain(|_, tx| tx.receiver_count() > 0); + removed += before - g.len(); + } + removed } } @@ -123,6 +135,36 @@ impl RealtimeBroadcaster for InProcessBroadcaster { g.remove(&(app_id, topic.to_string())); } } + + async fn subscribe_group( + &self, + group_id: GroupId, + topic: &str, + ) -> Result, BroadcasterError> { + let mut g = self + .groups + .lock() + .map_err(|_| BroadcasterError::Unavailable("group broadcaster map poisoned".into()))?; + let tx = g + .entry((group_id, topic.to_string())) + .or_insert_with(|| broadcast::channel(self.capacity).0); + Ok(tx.subscribe()) + } + + async fn publish_group(&self, group_id: GroupId, topic: &str, event: RealtimeEvent) { + let Ok(g) = self.groups.lock() else { + return; + }; + if let Some(tx) = g.get(&(group_id, topic.to_string())) { + let _ = tx.send(event); + } + } + + async fn drop_group_topic(&self, group_id: GroupId, topic: &str) { + if let Ok(mut g) = self.groups.lock() { + g.remove(&(group_id, topic.to_string())); + } + } } /// Spawn the background GC sweep that drops empty channels every @@ -239,4 +281,40 @@ mod tests { b.publish(app, "ghost", event("ghost", 1)).await; assert_eq!(b.channel_count(), 0); } + + #[tokio::test] + async fn group_channel_fans_out_and_isolates_by_group() { + // §11.6 D2: a group shared-topic channel is keyed by the OWNING group — + // a publish to group A never reaches a subscriber on group B, and the + // group map is separate from the per-app map. + let b = InProcessBroadcaster::new(16); + let g_a = GroupId::new(); + let g_b = GroupId::new(); + let mut rx_a1 = b.subscribe_group(g_a, "events").await.unwrap(); + let mut rx_a2 = b.subscribe_group(g_a, "events").await.unwrap(); + let mut rx_b = b.subscribe_group(g_b, "events").await.unwrap(); + assert_eq!(b.channel_count(), 2, "one channel per (group, topic)"); + + b.publish_group(g_a, "events", event("events", 7)).await; + // Both of group A's subscribers see it... + assert_eq!(rx_a1.recv().await.unwrap().message, json!({ "n": 7 })); + assert_eq!(rx_a2.recv().await.unwrap().message, json!({ "n": 7 })); + // ...group B's does not (owning-group isolation). + assert!(rx_b.try_recv().is_err()); + + // GC reclaims both maps. + drop((rx_a1, rx_a2, rx_b)); + assert_eq!(b.gc(), 2); + assert_eq!(b.channel_count(), 0); + } + + #[tokio::test] + async fn drop_group_topic_disconnects_subscribers() { + let b = InProcessBroadcaster::new(16); + let g = GroupId::new(); + let mut rx = b.subscribe_group(g, "t").await.unwrap(); + b.drop_group_topic(g, "t").await; + assert!(rx.recv().await.is_err()); + assert_eq!(b.channel_count(), 0); + } } diff --git a/crates/orchestrator-core/src/realtime_api.rs b/crates/orchestrator-core/src/realtime_api.rs index d179980..367be30 100644 --- a/crates/orchestrator-core/src/realtime_api.rs +++ b/crates/orchestrator-core/src/realtime_api.rs @@ -83,6 +83,9 @@ pub fn heartbeat_secs_from_env() -> u64 { pub fn realtime_router(state: RealtimeState) -> Router { Router::new() .route("/realtime/topics/{topic}", get(sse_topic)) + // §11.6 D2: group SHARED topics. Same Host→app dispatch; the owning group + // is resolved from the app's chain by the authority (reads-open model). + .route("/realtime/shared/topics/{topic}", get(sse_shared_topic)) .with_state(state) } @@ -158,6 +161,64 @@ async fn sse_topic( resp } +/// §11.6 D2: SSE for a group SHARED topic. Host→app resolves the subscriber's +/// app; the authority resolves the owning group from that app's chain (reads +/// open — the declaration is the grant; a foreign-subtree app 404s). The stream +/// is keyed by the owning group, so every subtree subscriber shares one channel +/// and sees every subtree app's publishes. +async fn sse_shared_topic( + State(state): State, + Path(topic): Path, + headers: HeaderMap, +) -> Response { + let host = headers + .get("host") + .and_then(|h| h.to_str().ok()) + .unwrap_or(""); + let Some(app_id) = state.app_domains.resolve_app(host) else { + return not_found("no app claims this host"); + }; + + let group_id = match state + .authority + .authorize_subscribe_shared(app_id, &topic) + .await + { + Ok(gid) => gid, + Err(SubscribeDenied::NotFound) => return not_found("shared topic not found"), + Err(SubscribeDenied::Unauthorized) => return unauthorized(), + Err(SubscribeDenied::Backend(e)) => { + tracing::error!(error = %e, "realtime authority backend error"); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + axum::Json(serde_json::json!({ "error": "internal error" })), + ) + .into_response(); + } + }; + + let rx = match state.broadcaster.subscribe_group(group_id, &topic).await { + Ok(rx) => rx, + Err(e) => { + tracing::error!(error = %e, "failed to acquire shared-topic subscription"); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + axum::Json(serde_json::json!({ "error": "internal error" })), + ) + .into_response(); + } + }; + + let sse = Sse::new(event_stream(rx)) + .keep_alive(KeepAlive::new().interval(state.heartbeat).text("heartbeat")); + let mut resp = sse.into_response(); + resp.headers_mut().insert( + "X-Accel-Buffering", + axum::http::HeaderValue::from_static("no"), + ); + resp +} + /// Map the broadcast receiver into a stream of SSE events. Lagged /// notifications (slow consumer) are skipped; a closed channel /// (`drop_topic`, or all senders gone) ends the stream and the SSE @@ -214,8 +275,9 @@ mod tests { use picloud_shared::{AppId, RealtimeEvent}; use tower::ServiceExt; // oneshot - /// Authority stub returning a fixed verdict. - struct StubAuthority(Result<(), SubscribeDenied>); + /// Authority stub returning a fixed verdict. For the shared-topic path it + /// maps `Ok` → a fixed owning group, else the same denial. + struct StubAuthority(Result<(), SubscribeDenied>, picloud_shared::GroupId); #[async_trait] impl RealtimeAuthority for StubAuthority { async fn authorize_subscribe( @@ -226,6 +288,13 @@ mod tests { ) -> Result<(), SubscribeDenied> { self.0.clone() } + async fn authorize_subscribe_shared( + &self, + _: AppId, + _: &str, + ) -> Result { + self.0.clone().map(|()| self.1) + } } /// App-domain table that maps a fixed host to a fixed app. @@ -250,7 +319,7 @@ mod tests { RealtimeState { app_domains: domains(host, app), broadcaster, - authority: Arc::new(StubAuthority(verdict)), + authority: Arc::new(StubAuthority(verdict, picloud_shared::GroupId::new())), heartbeat: Duration::from_millis(100), } } @@ -378,6 +447,74 @@ mod tests { assert!(text.contains("\"hi\":1"), "got: {text}"); } + #[tokio::test] + async fn shared_topic_not_resolved_is_404() { + // §11.6 D2: a subscriber app whose chain reaches no declaring group gets + // 404 (the reads-open resolution IS the authorization; a foreign subtree + // never resolves). + let app = AppId::new(); + let st = state( + app, + "app.example.com", + Err(SubscribeDenied::NotFound), + Arc::new(InProcessBroadcaster::new(8)), + ); + let appr = realtime_router(st); + let req = Request::builder() + .uri("/realtime/shared/topics/events") + .header("host", "app.example.com") + .body(Body::empty()) + .unwrap(); + assert_eq!( + appr.oneshot(req).await.unwrap().status(), + StatusCode::NOT_FOUND + ); + } + + #[tokio::test] + async fn shared_topic_streams_group_publishes() { + // A resolvable shared topic returns an SSE stream subscribed to the + // OWNING GROUP channel; a publish_group on that group is delivered. + let app = AppId::new(); + let group = picloud_shared::GroupId::new(); + let broadcaster = Arc::new(InProcessBroadcaster::new(8)); + let st = RealtimeState { + app_domains: domains("app.example.com", app), + broadcaster: broadcaster.clone(), + authority: Arc::new(StubAuthority(Ok(()), group)), + heartbeat: Duration::from_millis(100), + }; + let appr = realtime_router(st); + let req = Request::builder() + .uri("/realtime/shared/topics/events") + .header("host", "app.example.com") + .body(Body::empty()) + .unwrap(); + let resp = appr.oneshot(req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + let mut body = resp.into_body().into_data_stream(); + tokio::time::sleep(Duration::from_millis(50)).await; + broadcaster + .publish_group( + group, + "events", + RealtimeEvent { + topic: "events".into(), + message: serde_json::json!({ "hi": 2 }), + published_at: chrono::Utc::now(), + }, + ) + .await; + let chunk = tokio::time::timeout(Duration::from_secs(2), body.next()) + .await + .expect("a chunk within timeout") + .expect("stream item") + .expect("chunk ok"); + let text = String::from_utf8_lossy(&chunk); + assert!(text.contains("\"hi\":2"), "got: {text}"); + } + #[tokio::test] async fn heartbeat_fires_on_idle_connection() { let app = AppId::new(); diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index dbf3893..14aa53f 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -265,12 +265,15 @@ pub async fn build_app( // to in-process SSE subscribers. let pubsub_repo = Arc::new(PostgresPubsubRepo::new(pool.clone())); // §11.6 D2: the shared-topic service reuses the same fan-out repo. - let group_pubsub: Arc = - Arc::new(GroupPubsubServiceImpl::new( + let group_pubsub: Arc = Arc::new( + GroupPubsubServiceImpl::new( pubsub_repo.clone(), Arc::new(PostgresGroupCollectionResolver::new(pool.clone())), authz.clone(), - )); + ) + // §11.6 D2: fan a shared-topic publish to external SSE subscribers too. + .with_realtime(broadcaster.clone()), + ); // §11.6 D3: the group shared-queue store + enqueue service. let group_queue_repo = Arc::new(PostgresGroupQueueRepo::new(pool.clone())); let group_dead_letter_repo = Arc::new( @@ -346,6 +349,8 @@ pub async fn build_app( topic_repo.clone(), app_secrets_repo.clone(), users.clone(), + // §11.6 D2: resolves a shared topic to its owning group for SSE subscribe. + Arc::new(PostgresGroupCollectionResolver::new(pool.clone())), )); // v1.1.9 durable per-app queues. Producers write to queue_messages // via QueueService; the dispatcher's queue arm (commit 6) consumes diff --git a/crates/shared/src/realtime.rs b/crates/shared/src/realtime.rs index 16e8dc2..3232af6 100644 --- a/crates/shared/src/realtime.rs +++ b/crates/shared/src/realtime.rs @@ -22,7 +22,7 @@ use chrono::{DateTime, Utc}; use thiserror::Error; use tokio::sync::broadcast; -use crate::AppId; +use crate::{AppId, GroupId}; /// A single realtime event delivered to in-process SSE subscribers. The /// SSE handler serializes this to `data: {...}\n\n` on the wire. @@ -61,6 +61,34 @@ pub trait RealtimeBroadcaster: Send + Sync { /// Drop every subscriber for a topic (called on topic DELETE). Live /// receivers observe a closed channel and disconnect cleanly. async fn drop_topic(&self, app_id: AppId, topic: &str); + + // ---- §11.6 D2: group SHARED topics ------------------------------------- + // A shared topic fans out to every subtree app that subscribes, so it is + // keyed by the OWNING GROUP (not a single app) — one channel per + // `(group_id, topic)`. Default impls are no-ops so non-realtime bundles + // (NoopRealtimeBroadcaster, test doubles) need no change. + + /// Subscribe to a group shared topic's realtime stream. + async fn subscribe_group( + &self, + group_id: GroupId, + topic: &str, + ) -> Result, BroadcasterError> { + let _ = (group_id, topic); + let (_tx, rx) = broadcast::channel(1); + Ok(rx) + } + + /// Publish to a group shared topic's in-process subscribers (best-effort; + /// no live subscriber is a silent no-op). + async fn publish_group(&self, group_id: GroupId, topic: &str, event: RealtimeEvent) { + let _ = (group_id, topic, event); + } + + /// Drop every subscriber for a group shared topic. + async fn drop_group_topic(&self, group_id: GroupId, topic: &str) { + let _ = (group_id, topic); + } } /// Bootstrap / test impl: subscribe yields a receiver on a throwaway diff --git a/crates/shared/src/realtime_authority.rs b/crates/shared/src/realtime_authority.rs index fe21186..1361195 100644 --- a/crates/shared/src/realtime_authority.rs +++ b/crates/shared/src/realtime_authority.rs @@ -14,7 +14,7 @@ use async_trait::async_trait; -use crate::AppId; +use crate::{AppId, GroupId}; /// Why a subscribe attempt was refused. The SSE handler maps these to /// HTTP status codes. @@ -50,6 +50,24 @@ pub trait RealtimeAuthority: Send + Sync { topic: &str, token: Option<&str>, ) -> Result<(), SubscribeDenied>; + + /// §11.6 D2: decide whether a subscriber reaching the platform via `app_id`'s + /// host may subscribe to the group SHARED topic `topic`. Consistent with the + /// shared-collection trust model, **reads are open**: the authorization IS the + /// owning-group resolution — a group on `app_id`'s ancestor chain must declare + /// `topic`'s root segment a shared `kind='topic'` collection. Returns that + /// OWNING GROUP's id (so the caller subscribes to the group channel), or + /// `NotFound` when no ancestor group declares it (the chain walk is the + /// isolation boundary — a foreign-subtree app never resolves it). Default: + /// `NotFound`. + async fn authorize_subscribe_shared( + &self, + app_id: AppId, + topic: &str, + ) -> Result { + let _ = (app_id, topic); + Err(SubscribeDenied::NotFound) + } } /// Bootstrap impl: denies everything as `NotFound`. Replaced in diff --git a/docs/design/groups-and-project-tool.md b/docs/design/groups-and-project-tool.md index 9369813..3c3334d 100644 --- a/docs/design/groups-and-project-tool.md +++ b/docs/design/groups-and-project-tool.md @@ -1314,6 +1314,17 @@ Resolved items now live inline next to their topic. What genuinely remains: > topic pattern's ROOT segment (`events.*` → `events`) be a declared `kind='topic'` collection > (group-only) — the `shared` flag + owning-group chain walk are the isolation boundary. No message store > (topics are events, not persistence). Pinned by `tests/shared_topics.rs` + the `shared_topics` journey. + > **External SSE subscription shipped (Track A M6):** `GET /realtime/shared/topics/{topic}` streams a + > shared topic to external clients. Host→app dispatch (as the per-app SSE route) identifies the subscriber + > app; `RealtimeAuthority::authorize_subscribe_shared` resolves the OWNING GROUP from that app's chain + > (kind=`topic`, the topic's root segment) — the reads-open model: the resolution IS the authorization (a + > foreign-subtree app never resolves → 404, the isolation boundary), consistent with in-script shared-read + > semantics. The `RealtimeBroadcaster` gained a parallel `(group_id, topic)` channel map + > (`subscribe_group`/`publish_group`/`drop_group_topic`, default no-ops so non-realtime bundles are + > unchanged); `GroupPubsubServiceImpl::with_realtime` fans a publish to the owning-group channel (best- + > effort) after the durable trigger fan-out. Pinned by `realtime` broadcaster + `realtime_api` SSE tests + + > `group_pubsub_service::publish_bridges_to_the_group_broadcaster`. Deferred: multi-node propagation + > (cluster mode swaps the in-process broadcaster for LISTEN/NOTIFY behind the same trait). > > **Shipped — D3 shared durable QUEUES (competing consumers).** A group declares a `queue` shared > collection; any subtree app enqueues into ONE group-keyed store (`group_queue_messages`, `0065`) via