feat: integrate Kafka event bus for task lifecycle events

docker-compose.yml

@@ -58,6 +58,47 @@ services:
     networks:
       - backend
 
+  # -------------------------------------------------------------------
+  # Kafka broker — event backbone for task lifecycle events.
+  # Apache Kafka official image in KRaft mode (no Zookeeper).
+  # Single-node cluster: fine for development, not for production.
+  # Two client listeners:
+  #   INTERNAL://kafka:9092 — used by services inside the compose network
+  #   EXTERNAL://localhost:9094 — used by kcat/kafka-console tools from host
+  # -------------------------------------------------------------------
+  kafka:
+    image: apache/kafka:3.7.0
+    container_name: tmm-kafka
+    ports:
+      - "9094:9094"
+    environment:
+      # KRaft node identity
+      KAFKA_NODE_ID: 0
+      KAFKA_PROCESS_ROLES: controller,broker
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 0@kafka:9093
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      # Listeners
+      KAFKA_LISTENERS: INTERNAL://:9092,CONTROLLER://:9093,EXTERNAL://:9094
+      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,EXTERNAL://localhost:9094
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
+      # Log directory (must match the volume mount path)
+      KAFKA_LOG_DIRS: /var/lib/kafka/data
+      # Single-node replication settings (no sensible defaults from apache/kafka)
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+    volumes:
+      - kafka-data:/var/lib/kafka/data
+    healthcheck:
+      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 >/dev/null 2>&1"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+      start_period: 30s
+    networks:
+      - backend
+
   # -------------------------------------------------------------------
   # Auth service — registro, login, JWT issuance, refresh tokens.
   # Depende de Postgres saludable antes de arrancar Flyway.
@@ -106,6 +147,8 @@ services:
         condition: service_healthy
       mongo:
         condition: service_healthy
+      kafka:
+        condition: service_healthy
     healthcheck:
       test: ["CMD-SHELL", "wget -qO- http://localhost:8082/actuator/health | grep -q '\"status\":\"UP\"'"]
       interval: 10s
@@ -153,6 +196,7 @@ services:
 volumes:
   postgres-data:
   mongo-data:
+  kafka-data:
 
 # -----------------------------------------------------------------------
 # Red: bridge dedicada. Los contenedores se resuelven por nombre de

task-service/pom.xml

@@ -67,6 +67,11 @@
             <artifactId>spring-boot-starter-data-mongodb</artifactId>
         </dependency>
 
+        <!-- Kafka -->
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
         <!-- MapStruct -->
         <dependency>
             <groupId>org.mapstruct</groupId>

task-service/src/main/java/com/mtole/task/activity/ActivityEvent.java

@@ -67,6 +67,26 @@ public ActivityEvent(
         this.after = after;
         this.timestamp = timestamp;
     }
+    // Constructor con id explícito. Usado por el consumer para poblar
+    // el _id de Mongo con el eventId del contrato Kafka (idempotencia).
+    public ActivityEvent(
+            String id,
+            Long userId,
+            String action,
+            String resourceType,
+            Long resourceId,
+            Map<String, Object> before,
+            Map<String, Object> after,
+            Instant timestamp) {
+        this.id = id;
+        this.userId = userId;
+        this.action = action;
+        this.resourceType = resourceType;
+        this.resourceId = resourceId;
+        this.before = before;
+        this.after = after;
+        this.timestamp = timestamp;
+    }
 
     public String getId() {
         return id;

task-service/src/main/java/com/mtole/task/kafka/ActivityLogKafkaConsumer.java

@@ -0,0 +1,97 @@
+package com.mtole.task.kafka;
+
+import org.springframework.dao.DuplicateKeyException;
+import com.mtole.task.activity.ActivityEvent;
+import com.mtole.task.kafka.events.CategoryEvent;
+import com.mtole.task.kafka.events.TaskEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+
+import static com.mtole.task.kafka.KafkaConfig.TOPIC_CATEGORY_EVENTS;
+import static com.mtole.task.kafka.KafkaConfig.TOPIC_TASK_EVENTS;
+
+/**
+ * Consumer del audit log. Escucha los topics `task.events` y
+ * `category.events`, y persiste un documento ActivityEvent en Mongo
+ * por cada mensaje recibido.
+ *
+ * Idempotencia: el _id del documento en Mongo es el eventId del
+ * contrato Kafka. Si el mismo mensaje se procesa más de una vez
+ * (por reintentos, reset de offsets, etc.), el insert lanza
+ * DuplicateKeyException y se ignora silenciosamente. Efecto:
+ * exactamente un documento por eventId. Patrón "effectively
+ * exactly-once".
+ *
+ * Semántica: at-least-once. El commit del offset se hace manualmente
+ * tras confirmación de escritura en Mongo. Si la escritura falla,
+ * el offset no se commitea y Kafka reintenta más tarde.
+ *
+ * Detalle de negocio: los eventos Kafka son thin (solo IDs y
+ * metadatos). El activity log resultante no guarda title/status/name.
+ * Trade-off documentado en ADR-007.
+ */
+@Component
+public class ActivityLogKafkaConsumer {
+
+    private static final Logger log = LoggerFactory.getLogger(ActivityLogKafkaConsumer.class);
+
+    private final MongoTemplate mongoTemplate;
+
+    public ActivityLogKafkaConsumer(MongoTemplate mongoTemplate) {
+        this.mongoTemplate = mongoTemplate;
+    }
+
+    @KafkaListener(topics = TOPIC_TASK_EVENTS, groupId = "activity-log-writer")
+    public void onTaskEvent(TaskEvent event, Acknowledgment ack) {
+        log.debug("Received TaskEvent {} type={} task={}", event.eventId(), event.type(), event.taskId());
+
+        ActivityEvent doc = new ActivityEvent(
+                event.eventId().toString(),
+                event.userId(),
+                "TASK_" + event.type().name(),
+                "TASK",
+                event.taskId(),
+                Map.of(),
+                Map.of(),
+                event.timestamp()
+        );
+
+        persist(doc, event.eventId().toString());
+        ack.acknowledge();
+    }
+
+    @KafkaListener(topics = TOPIC_CATEGORY_EVENTS, groupId = "activity-log-writer")
+    public void onCategoryEvent(CategoryEvent event, Acknowledgment ack) {
+        log.debug("Received CategoryEvent {} type={} category={}",
+                event.eventId(), event.type(), event.categoryId());
+
+        ActivityEvent doc = new ActivityEvent(
+                event.eventId().toString(),
+                event.userId(),
+                "CATEGORY_" + event.type().name(),
+                "CATEGORY",
+                event.categoryId(),
+                Map.of(),
+                Map.of(),
+                event.timestamp()
+        );
+
+        persist(doc, event.eventId().toString());
+        ack.acknowledge();
+    }
+
+    private void persist(ActivityEvent doc, String eventId) {
+        try {
+            mongoTemplate.insert(doc);
+            log.debug("Persisted ActivityEvent id={}", eventId);
+        } catch (DuplicateKeyException ex) {
+            log.info("Duplicate ActivityEvent skipped id={}", eventId);
+        }
+    }
+}