feat: migrate messaging to Spring Cloud Stream and add comprehensive API documentation

README.md

@@ -119,6 +119,7 @@ PCM uses a centralized configuration model via **Spring Cloud Config**. All core
 ## 📚 Documentation
 
 - [**Quick Start Guide**](docs/QUICKSTART.md) - Get PCM running locally in 5 minutes.
+- [**API Reference**](docs/API_REFERENCE.md) - Endpoints, payloads, and Examples.
 - [**Architecture Decision Records**](docs/architecture/) - Design decisions and rationale.
 ---
 

config-service/src/main/resources/config/application.yml

@@ -32,21 +32,50 @@ spring:
       port: ${REDIS_PORT:6779}
       timeout: 2000ms
 
-  # Shared Kafka
-  kafka:
-    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
-    producer:
-      key-serializer: org.apache.kafka.common.serialization.StringSerializer
-      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
-      acks: all
-      retries: 3
-    consumer:
-      group-id: ${spring.application.name}-group
-      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
-      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
-    properties:
-      schema.registry.url: ${SCHEMA_REGISTRY_URL:http://localhost:8081}
-      specific.avro.reader: true
+  # Shared Spring Cloud Stream (Messaging Abstraction)
+  cloud:
+    function:
+      definition: ${PCM_STREAM_FUNCTIONS:} # To be overridden in service-specific configs if needed
+    stream:
+      kafka:
+        binder:
+          brokers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
+          configuration:
+            schema.registry.url: ${SCHEMA_REGISTRY_URL:http://localhost:8081}
+            specific.avro.reader: true
+      bindings:
+        # Producers (profile-service)
+        profileCreated-out-0:
+          destination: profile-events
+          content-type: application/*+avro
+        profileUpdated-out-0:
+          destination: profile-events
+          content-type: application/*+avro
+        profileErased-out-0:
+          destination: profile-events
+          content-type: application/*+avro
+
+        # Consumers
+        profileCreated-in-0:
+          destination: profile-events
+          group: ${spring.application.name}-group
+          content-type: application/*+avro
+        profileUpdated-in-0:
+          destination: profile-events
+          group: ${spring.application.name}-group
+          content-type: application/*+avro
+        profileErased-in-0:
+          destination: profile-events
+          group: ${spring.application.name}-group
+          content-type: application/*+avro
+
+        # Producers (consent-service)
+        consentGranted-out-0:
+          destination: consent-events
+          content-type: application/*+avro
+        consentRevoked-out-0:
+          destination: consent-events
+          content-type: application/*+avro
 
   # Shared Vault
   cloud:

config-service/src/main/resources/config/preference-service.yml

@@ -7,6 +7,9 @@ spring:
   kafka:
     consumer:
       group-id: preference-service-group
+  cloud:
+    function:
+      definition: profileErased
 
 grpc:
   server:

config-service/src/main/resources/config/segment-service.yml

@@ -9,6 +9,9 @@ spring:
   kafka:
     consumer:
       group-id: segment-service-group
+  cloud:
+    function:
+      definition: profileCreated;profileUpdated
 
 grpc:
   server:

consent-service/pom.xml

@@ -61,10 +61,10 @@
             <version>2.15.0.RELEASE</version>
         </dependency>
 
-        <!-- Spring Kafka -->
+        <!-- Spring Cloud Stream Kafka Binder -->
         <dependency>
-            <groupId>org.springframework.kafka</groupId>
-            <artifactId>spring-kafka</artifactId>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
         </dependency>
 
         <!-- Spring Cloud Config -->

consent-service/src/main/java/dev/vibeafrika/pcm/consent/infrastructure/messaging/KafkaConsentEventPublisher.java

@@ -6,8 +6,8 @@
 import dev.vibeafrika.pcm.events.ConsentPurpose;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.kafka.core.KafkaTemplate;
+
+import org.springframework.cloud.stream.function.StreamBridge;
 import org.springframework.stereotype.Component;
 
 /**
@@ -19,53 +19,50 @@
 @RequiredArgsConstructor
 public class KafkaConsentEventPublisher implements ConsentEventPublisher {
 
-    private final KafkaTemplate<String, Object> kafkaTemplate;
-
-    @Value("${pcm.topics.consent-events:consent-events}")
-    private String consentEventsTopic;
+    private final StreamBridge streamBridge;
 
     @Override
     public void publish(ConsentGrantedEvent domainEvent) {
-        dev.vibeafrika.pcm.events.ConsentGrantedEvent avroEvent = dev.vibeafrika.pcm.events.ConsentGrantedEvent.newBuilder()
-            .setEventId(domainEvent.getEventId())
-            .setEventType(domainEvent.getEventType())
-            .setOccurredAt(domainEvent.getOccurredAt().toEpochMilli())
-            .setVersion(1)
-            .setTenantId(domainEvent.getTenantId())
-            .setConsentId(domainEvent.getAggregateId())
-            .setProfileId(domainEvent.getProfileId())
-            .setPurpose(ConsentPurpose.valueOf(domainEvent.getPurpose()))
-            .setConsentVersion(domainEvent.getConsentVersion())
-            .setProofHash(domainEvent.getProofHash())
-            .build();
+        dev.vibeafrika.pcm.events.ConsentGrantedEvent avroEvent = dev.vibeafrika.pcm.events.ConsentGrantedEvent
+                .newBuilder()
+                .setEventId(domainEvent.getEventId())
+                .setEventType(domainEvent.getEventType())
+                .setOccurredAt(domainEvent.getOccurredAt().toEpochMilli())
+                .setVersion(1)
+                .setTenantId(domainEvent.getTenantId())
+                .setConsentId(domainEvent.getAggregateId())
+                .setProfileId(domainEvent.getProfileId())
+                .setPurpose(ConsentPurpose.valueOf(domainEvent.getPurpose()))
+                .setConsentVersion(domainEvent.getConsentVersion())
+                .setProofHash(domainEvent.getProofHash())
+                .build();
 
-        publish(avroEvent.getProfileId().toString(), avroEvent);
+        send("consentGranted-out-0", avroEvent.getProfileId().toString(), avroEvent);
     }
 
     @Override
     public void publish(ConsentRevokedEvent domainEvent) {
-        dev.vibeafrika.pcm.events.ConsentRevokedEvent avroEvent = dev.vibeafrika.pcm.events.ConsentRevokedEvent.newBuilder()
-            .setEventId(domainEvent.getEventId())
-            .setEventType(domainEvent.getEventType())
-            .setOccurredAt(domainEvent.getOccurredAt().toEpochMilli())
-            .setVersion(1)
-            .setTenantId(domainEvent.getTenantId())
-            .setConsentId(domainEvent.getAggregateId())
-            .setProfileId(domainEvent.getProfileId())
-            .setPurpose(ConsentPurpose.valueOf(domainEvent.getPurpose()))
-            .build();
+        dev.vibeafrika.pcm.events.ConsentRevokedEvent avroEvent = dev.vibeafrika.pcm.events.ConsentRevokedEvent
+                .newBuilder()
+                .setEventId(domainEvent.getEventId())
+                .setEventType(domainEvent.getEventType())
+                .setOccurredAt(domainEvent.getOccurredAt().toEpochMilli())
+                .setVersion(1)
+                .setTenantId(domainEvent.getTenantId())
+                .setConsentId(domainEvent.getAggregateId())
+                .setProfileId(domainEvent.getProfileId())
+                .setPurpose(ConsentPurpose.valueOf(domainEvent.getPurpose()))
+                .build();
 
-        publish(avroEvent.getProfileId().toString(), avroEvent);
+        send("consentRevoked-out-0", avroEvent.getProfileId().toString(), avroEvent);
     }
 
-    private void publish(String key, Object event) {
-        kafkaTemplate.send(consentEventsTopic, key, event)
-            .whenComplete((result, ex) -> {
-                if (ex == null) {
-                    log.debug("Published consent event to topic {}: {}", consentEventsTopic, event);
-                } else {
-                    log.error("Failed to publish consent event to topic {}: {}", consentEventsTopic, ex.getMessage());
-                }
-            });
+    private void send(String bindingName, String key, Object event) {
+        log.debug("Sending consent event to binding {}: {}", bindingName, event);
+        streamBridge.send(bindingName,
+                org.springframework.messaging.support.MessageBuilder
+                        .withPayload(event)
+                        .setHeader("partitionKey", key)
+                        .build());
     }
 }

docs/API_REFERENCE.md

@@ -0,0 +1,178 @@
+# API Reference & Curl Examples
+
+This document provide a comprehensive reference for the PCM (Profile & Consent Manager) REST APIs, including expected payloads and `curl` examples.
+
+## Base Infrastructure
+| Service | Port | Description |
+| :--- | :--- | :--- |
+| **API Gateway** | `9880` | Entry point (Auth & Aggregation) |
+| **Profile Service** | `18081` | PII & Identity Management |
+| **Preference Service** | `18082` | UX & Application Settings |
+| **Consent Service** | `18083` | Consent Ledger & GDPR Compliance |
+| **Segment Service** | `18084` | User Classification |
+
+---
+
+## Authentication & Headers
+- **X-Tenant-Id**: Required for all requests. Default is `default`.
+- **Authorization**: Bearer token required for Gateway endpoints (Keycloak JWT).
+
+---
+
+## 1. Profile Service (`:18081`)
+
+### Create a Profile
+Used to initialize a new user record.
+
+**Endpoint**: `POST /api/v1/profiles`
+
+**Payload**:
+```json
+{
+  "id": "550e8400-e29b-41d4-a716-446655440000",
+  "handle": "jdoe",
+  "attributes": {
+    "fullName": "John Doe",
+    "email": "john.doe@example.com",
+    "country": "FR"
+  }
+}
+```
+
+**Curl**:
+```bash
+curl -X POST http://localhost:18081/api/v1/profiles \
+  -H "Content-Type: application/json" \
+  -H "X-Tenant-Id: default" \
+  -d '{
+    "id": "550e8400-e29b-41d4-a716-446655440000",
+    "handle": "jdoe",
+    "attributes": {
+      "fullName": "John Doe",
+      "email": "john.doe@example.com",
+      "country": "FR"
+    }
+  }'
+```
+
+### Get Profile
+Retrieve a profile. Sensitive attributes are decrypted automatically if Vault is enabled.
+
+**Endpoint**: `GET /api/v1/profiles/{id}`
+
+**Curl**:
+```bash
+curl http://localhost:18081/api/v1/profiles/550e8400-e29b-41d4-a716-446655440000 \
+  -H "X-Tenant-Id: default"
+```
+
+---
+
+## 2. Consent Service (`:18083`)
+
+### Grant Consent
+Records a positive consent action in the ledger.
+
+**Endpoint**: `POST /api/v1/consents/{profileId}/grant`
+
+**Payload**:
+```json
+{
+  "purpose": "MARKETING",
+  "version": "v1.2",
+  "consentText": "I agree to receive marketing emails.",
+  "metadata": {
+    "source": "web-form-footer"
+  }
+}
+```
+
+**Curl**:
+```bash
+curl -X POST http://localhost:18083/api/v1/consents/550e8400-e29b-41d4-a716-446655440000/grant \
+  -H "Content-Type: application/json" \
+  -H "X-Tenant-ID: default" \
+  -d '{
+    "purpose": "MARKETING",
+    "version": "v1.2",
+    "consentText": "I agree to receive marketing emails."
+  }'
+```
+
+### Verify Consent
+Check if a user currently has granted permission for a specific purpose.
+
+**Endpoint**: `GET /api/v1/consents/{profileId}/verify?purpose=MARKETING`
+
+**Curl**:
+```bash
+curl "http://localhost:18083/api/v1/consents/550e8400-e29b-41d4-a716-446655440000/verify?purpose=MARKETING" \
+  -H "X-Tenant-ID: default"
+```
+
+---
+
+## 3. Preference Service (`:18082`)
+
+### Update Preferences
+Update key-value settings for a user.
+
+**Endpoint**: `PATCH /api/v1/preferences/{profileId}`
+
+**Payload**:
+```json
+{
+  "theme": "dark",
+  "language": "fr",
+  "notifications_enabled": "true"
+}
+```
+
+**Curl**:
+```bash
+curl -X PATCH http://localhost:18082/api/v1/preferences/550e8400-e29b-41d4-a716-446655440000 \
+  -H "Content-Type: application/json" \
+  -H "X-Tenant-Id: default" \
+  -d '{"theme": "dark", "language": "fr"}'
+```
+
+---
+
+## 4. Segment Service (`:18084`)
+
+### Get User Segments
+Retrieve computed segments (classification) for a user.
+
+**Endpoint**: `GET /api/v1/segments/{profileId}`
+
+**Curl**:
+```bash
+curl http://localhost:18084/api/v1/segments/550e8400-e29b-41d4-a716-446655440000
+```
+
+---
+
+## 5. API Gateway (Aggregation & Auth) (`:9880`)
+
+### Unified "Me" Endpoint
+Returns an aggregated view of the authenticated user (Profile + Preferences + Segments).
+
+**Endpoint**: `GET /api/v1/users/me` (or `GET /api/v1/me`)
+
+**Curl**:
+```bash
+curl http://localhost:9880/api/v1/users/me \
+  -H "Authorization: Bearer <JWT_TOKEN>" \
+  -H "X-Tenant-Id: default"
+```
+
+---
+
+## Purpose Reference
+Standard values for `purpose` in Consent & Segments:
+- `MARKETING`
+- `ANALYTICS`
+- `PERSONALIZATION`
+- `THIRD_PARTY_SHARING`
+- `TERMS_AND_CONDITIONS`
+- `PRIVACY_POLICY`

preference-service/pom.xml

@@ -40,10 +40,10 @@
             <artifactId>spring-boot-starter-data-redis</artifactId>
         </dependency>
 
-        <!-- Messaging -->
+        <!-- Spring Cloud Stream Kafka Binder -->
         <dependency>
-            <groupId>org.springframework.kafka</groupId>
-            <artifactId>spring-kafka</artifactId>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
         </dependency>
         <dependency>
             <groupId>dev.vibe-afrika</groupId>