feat: add internal event bus, stream events via HTTP endpoint

Cargo.toml

@@ -107,6 +107,7 @@ strum_macros = "0.26"
 temp-env = "0.3"
 thiserror = "1.0"
 tokio = { version = "1.46", features = ["full"] }
+tokio-stream = { version = "0.1", features = ["sync"] }
 tower = { version = "0.4" }
 tower-http = { version = "0.6", features = ["cors", "trace"] }
 tracing = { version = "0.1" }

agent_api_http/Cargo.toml

@@ -24,6 +24,7 @@ axum-macros = "0.5"
 base64.workspace = true
 chrono.workspace = true
 cqrs-es.workspace = true
+futures.workspace = true
 http = "1.3"
 http-api-problem = { version = "0.60", features = ["axum", "api-error"] }
 http-body-util = "0.1"

agent_api_http/openapi-generated.yaml

@@ -437,6 +437,50 @@ paths:
                 $ref: '#/components/schemas/Template'
         '422':
           description: Source Template Not Found
+  /v0/events:
+    get:
+      tags:
+      - Events
+      summary: Stream domain events as CloudEvents via SSE with Catch-Up.
+      operationId: events_sse_handler
+      parameters:
+      - name: types
+        in: query
+        description: Comma-separated list of CloudEvent types to filter
+        required: false
+        schema:
+          type: string
+      - name: sources
+        in: query
+        description: Comma-separated list of sources/aggregate types to filter
+        required: false
+        schema:
+          type: string
+      - name: subject
+        in: query
+        description: Optional aggregate/subject ID filter
+        required: false
+        schema:
+          type: string
+      - name: since
+        in: query
+        description: Filter events after RFC 3339 timestamp
+        required: false
+        schema:
+          type: string
+      - name: until
+        in: query
+        description: Filter events before RFC 3339 timestamp
+        required: false
+        schema:
+          type: string
+      responses:
+        '200':
+          description: Server-Sent Events stream of CloudEvents
+          content:
+            text/event-stream:
+              schema:
+                $ref: '#/components/schemas/CloudEvent'
   /v0/get-all-catalogs:
     get:
       tags:
@@ -1009,6 +1053,42 @@ components:
             name: ''
             description: ''
             icon: null
+    CloudEvent:
+      type: object
+      description: A CNCF CloudEvent envelope (v1.0 spec).
+      required:
+      - id
+      - source
+      - specversion
+      - type
+      properties:
+        data: {}
+        datacontenttype:
+          type:
+          - string
+          - 'null'
+        dataschema:
+          type:
+          - string
+          - 'null'
+        id:
+          type: string
+        source:
+          type: string
+        specversion:
+          type: string
+        subject:
+          type:
+          - string
+          - 'null'
+        time:
+          type:
+          - string
+          - 'null'
+          format: date-time
+        type:
+          type: string
+          example: com.impierce.unicore.offer-created
     Connection:
       type: object
       required:
@@ -2093,6 +2173,8 @@ tags:
   description: Create and manage templates which provide the structure for credentials to be issued.
 - name: Catalog
   description: Create and manage catalogs to organize and share your templates.
+- name: Events
+  description: Stream domain events as CloudEvents via SSE.
 externalDocs:
   url: https://docs.impierce.com/unicore
   description: Official UniCore documentation

agent_api_http/src/lib.rs

@@ -25,6 +25,7 @@ use http::HeaderMap;
 use http_body_util::BodyExt as _;
 use hyper::StatusCode;
 use shared_kernel::authorization::{Actor, ActorExtractor, ToActor};
+use shared_kernel::event_bus::EventBusHandle;
 use std::{sync::Arc, time::Duration};
 use tower::ServiceBuilder;
 use tower_http::trace::TraceLayer;
@@ -42,6 +43,7 @@ pub struct ApiState {
     pub issuance_state: Option<Arc<IssuanceState>>,
     pub holder_state: Option<Arc<HolderState>>,
     pub verification_state: Option<Arc<VerificationState>>,
+    pub event_bus: Option<EventBusHandle>,
 }
 
 /// Build the top-level API router.
@@ -58,6 +60,7 @@ pub fn app<E>(
         issuance_state,
         holder_state,
         verification_state,
+        event_bus,
     }: ApiState,
     actor_extractor: Arc<E>,
 ) -> Router
@@ -84,6 +87,7 @@ where
         )
         .merge(holder_state.map(v0::holder::router).unwrap_or_default())
         .merge(verification_state.map(v0::verification::router).unwrap_or_default())
+        .merge(event_bus.map(v0::events::router).unwrap_or_default())
         .merge(public::router())
         .layer(middleware::from_fn_with_state(actor_extractor, extract_actor::<E>))
         // Trace layers

agent_api_http/src/v0/authorization/authorization_server/authorize.rs

@@ -114,8 +114,15 @@ pub mod tests {
     #[serial_test::serial]
     #[tokio::test]
     async fn test_authorization_endpoint() {
-        let issuance_state =
-            Arc::new(issuance_state(&InMemory, IssuanceServices::default().await, Default::default()).await);
+        let issuance_state = Arc::new(
+            issuance_state(
+                &InMemory,
+                IssuanceServices::default().await,
+                &Default::default(),
+                Default::default(),
+            )
+            .await,
+        );
 
         agent_issuance::state::initialize(&issuance_state).await.unwrap();
 
@@ -133,6 +140,7 @@ pub mod tests {
             authorization_state(
                 &InMemory,
                 AuthorizationServices::default().await,
+                &Default::default(),
                 Default::default(),
                 Default::default(),
             )

agent_api_http/src/v0/authorization/authorization_server/par.rs

@@ -255,8 +255,15 @@ pub mod tests {
     #[serial_test::serial]
     #[tokio::test]
     async fn test_pushed_authorization_request_endpoint() {
-        let issuance_state =
-            Arc::new(issuance_state(&InMemory, IssuanceServices::default().await, Default::default()).await);
+        let issuance_state = Arc::new(
+            issuance_state(
+                &InMemory,
+                IssuanceServices::default().await,
+                &Default::default(),
+                Default::default(),
+            )
+            .await,
+        );
 
         agent_issuance::state::initialize(&issuance_state).await.unwrap();
 
@@ -274,6 +281,7 @@ pub mod tests {
             authorization_state(
                 &InMemory,
                 AuthorizationServices::default().await,
+                &Default::default(),
                 Default::default(),
                 Default::default(),
             )
@@ -295,8 +303,15 @@ pub mod tests {
     #[serial_test::serial]
     #[tokio::test]
     async fn test_interactive_authorization_request_flow() {
-        let issuance_state =
-            Arc::new(issuance_state(&InMemory, IssuanceServices::default().await, Default::default()).await);
+        let issuance_state = Arc::new(
+            issuance_state(
+                &InMemory,
+                IssuanceServices::default().await,
+                &Default::default(),
+                Default::default(),
+            )
+            .await,
+        );
 
         agent_issuance::state::initialize(&issuance_state).await.unwrap();
         let library_state = setup_library_state(&issuance_state).await;
@@ -310,7 +325,13 @@ pub mod tests {
         let issuer_state = issuer_state.unwrap();
 
         let verification_state = Arc::new(
-            agent_store::verification_state(&InMemory, VerificationServices::default().await, Default::default()).await,
+            agent_store::verification_state(
+                &InMemory,
+                VerificationServices::default().await,
+                &Default::default(),
+                Default::default(),
+            )
+            .await,
         );
 
         let oauth2_authorization_request_domain_services = OAuth2AuthorizationRequestDomainServices::new(Box::new(
@@ -321,6 +342,7 @@ pub mod tests {
             authorization_state(
                 &InMemory,
                 AuthorizationServices::default().await,
+                &Default::default(),
                 Default::default(),
                 oauth2_authorization_request_domain_services,
             )

agent_api_http/src/v0/authorization/authorization_server/token.rs

@@ -151,8 +151,15 @@ pub mod tests {
     #[serial_test::serial]
     #[tokio::test]
     async fn test_token_endpoint(#[case] is_pre_authorized: bool) {
-        let issuance_state =
-            Arc::new(issuance_state(&InMemory, IssuanceServices::default().await, Default::default()).await);
+        let issuance_state = Arc::new(
+            issuance_state(
+                &InMemory,
+                IssuanceServices::default().await,
+                &Default::default(),
+                Default::default(),
+            )
+            .await,
+        );
 
         agent_issuance::state::initialize(&issuance_state).await.unwrap();
 
@@ -168,6 +175,7 @@ pub mod tests {
             authorization_state(
                 &InMemory,
                 AuthorizationServices::default().await,
+                &Default::default(),
                 Default::default(),
                 Default::default(),
             )
@@ -186,8 +194,15 @@ pub mod tests {
     #[serial_test::serial]
     #[tokio::test]
     async fn test_pre_authorized_token_redemption_fails_when_public_offer_is_offline() {
-        let issuance_state =
-            Arc::new(issuance_state(&InMemory, IssuanceServices::default().await, Default::default()).await);
+        let issuance_state = Arc::new(
+            issuance_state(
+                &InMemory,
+                IssuanceServices::default().await,
+                &Default::default(),
+                Default::default(),
+            )
+            .await,
+        );
 
         agent_issuance::state::initialize(&issuance_state).await.unwrap();
 
@@ -226,6 +241,7 @@ pub mod tests {
             authorization_state(
                 &InMemory,
                 AuthorizationServices::default().await,
+                &Default::default(),
                 Default::default(),
                 Default::default(),
             )
@@ -273,8 +289,15 @@ pub mod tests {
     #[serial_test::serial]
     #[tokio::test]
     async fn test_pre_authorized_token_redemption_fails_when_public_offer_is_deleted() {
-        let issuance_state =
-            Arc::new(issuance_state(&InMemory, IssuanceServices::default().await, Default::default()).await);
+        let issuance_state = Arc::new(
+            issuance_state(
+                &InMemory,
+                IssuanceServices::default().await,
+                &Default::default(),
+                Default::default(),
+            )
+            .await,
+        );
 
         agent_issuance::state::initialize(&issuance_state).await.unwrap();
 
@@ -313,6 +336,7 @@ pub mod tests {
             authorization_state(
                 &InMemory,
                 AuthorizationServices::default().await,
+                &Default::default(),
                 Default::default(),
                 Default::default(),
             )