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;
diff --git a/task-service/src/main/java/com/mtole/task/activity/ActivityEventListener.java b/task-service/src/main/java/com/mtole/task/activity/ActivityEventListener.java
deleted file mode 100644
index d27f88c..0000000
--- a/task-service/src/main/java/com/mtole/task/activity/ActivityEventListener.java
+++ /dev/null
@@ -1,178 +0,0 @@
-package com.mtole.task.activity;
-
-import com.mtole.task.categories.events.CategoryCreatedEvent;
-import com.mtole.task.categories.events.CategoryDeletedEvent;
-import com.mtole.task.categories.events.CategoryUpdatedEvent;
-import com.mtole.task.tasks.events.TaskCreatedEvent;
-import com.mtole.task.tasks.events.TaskDeletedEvent;
-import com.mtole.task.tasks.events.TaskStatusChangedEvent;
-import com.mtole.task.tasks.events.TaskUpdatedEvent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.context.event.EventListener;
-import org.springframework.stereotype.Component;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * Listener de eventos de dominio. Traduce cada evento publicado por
- * los services de PostgreSQL en un documento ActivityEvent que se
- * persiste en MongoDB.
- *
- * Síncrono por defecto: los métodos se ejecutan en el mismo hilo y
- * dentro de la misma transacción que el publisher. Si la escritura
- * en MongoDB falla, se produce rollback de la transacción JPA
- * completa (la operación original en PostgreSQL también se revierte).
- *
- * Si en el futuro hace falta desacoplar latencias (audit logging
- * lento bloqueando los endpoints), se puede pasar a asíncrono
- * añadiendo @Async + configurando un TaskExecutor, aceptando que el
- * audit log puede perderse si la escritura en Mongo falla.
- */
-@Component
-public class ActivityEventListener {
-
- private static final Logger log = LoggerFactory.getLogger(ActivityEventListener.class);
-
- private final ActivityEventRepository repository;
-
- public ActivityEventListener(ActivityEventRepository repository) {
- this.repository = repository;
- }
-
- @EventListener
- public void onTaskCreated(TaskCreatedEvent event) {
- log.debug("Recording TaskCreated event for taskId={}", event.taskId());
-
- Map after = new HashMap<>();
- after.put("title", event.title());
- after.put("status", event.status());
- if (event.categoryId() != null) {
- after.put("categoryId", event.categoryId());
- }
-
- ActivityEvent doc = new ActivityEvent(
- event.userId(),
- "TASK_CREATED",
- "TASK",
- event.taskId(),
- Map.of(),
- after,
- event.occurredAt()
- );
- repository.save(doc);
- }
-
- // TODO: implementar el resto de listeners:
- // - onTaskUpdated(TaskUpdatedEvent)
-
- @EventListener
- public void onTaskUpdated(TaskUpdatedEvent event) {
- log.debug("Recording TaskUpdated event for taskId={}", event.taskId());
-
- ActivityEvent doc = new ActivityEvent(
- event.userId(),
- "TASK_UPDATED",
- "TASK",
- event.taskId(),
- Map.of(),
- Map.of(),
- event.occurredAt()
- );
- repository.save(doc);
- }
-
- @EventListener
- public void onTaskStatusChanged(TaskStatusChangedEvent event) {
- log.debug("Recording TaskStatusChanged event for taskId={}", event.taskId());
-
- // before y after vacíos
- ActivityEvent doc = new ActivityEvent(
- event.userId(),
- "TASK_STATUS_CHANGED",
- "TASK",
- event.taskId(),
- Map.of("status", event.oldStatus()),
- Map.of("status", event.newStatus()),
- event.occurredAt()
- );
- repository.save(doc);
- }
-
- // - onTaskDeleted(TaskDeletedEvent)
- @EventListener
- public void onTaskDeleted(TaskDeletedEvent event) {
- log.debug("Recording TaskDeleted event for taskId={}", event.taskId());
-
- Map before = new HashMap<>();
- before.put("title", event.title());
- before.put("status", event.status());
-
- ActivityEvent doc = new ActivityEvent(
- event.userId(),
- "TASK_DELETED",
- "TASK",
- event.taskId(),
- before,
- Map.of(),
- event.occurredAt()
- );
- repository.save(doc);
- }
-
- // - onCategoryCreated(CategoryCreatedEvent)
- @EventListener
- public void onCategoryCreated(CategoryCreatedEvent event) {
- log.debug("Recording CategoryCreated event for categoryId={}", event.categoryId());
-
- Map after = new HashMap<>();
- after.put("name", event.name());
-
- ActivityEvent doc = new ActivityEvent(
- event.userId(),
- "CATEGORY_CREATED",
- "CATEGORY",
- event.categoryId(),
- Map.of(),
- after,
- event.occurredAt()
- );
- repository.save(doc);
- }
-
- // - onCategoryUpdated(CategoryUpdatedEvent)
- @EventListener
- public void onCategoryUpdated(CategoryUpdatedEvent event) {
- log.debug("Recording CategoryUpdated event for categoryId={}", event.categoryId());
- ActivityEvent doc = new ActivityEvent(
- event.userId(),
- "CATEGORY_UPDATED",
- "CATEGORY",
- event.categoryId(),
- Map.of(),
- Map.of(),
- event.occurredAt()
- );
- repository.save(doc);
- }
-
- // - onCategoryDeleted(CategoryDeletedEvent)
- @EventListener
- public void onCategoryDeleted(CategoryDeletedEvent event) {
- log.debug("Recording CategoryDeleted event for categoryId={}", event.categoryId());
- Map before = new HashMap<>();
- before.put("name", event.name());
-
- ActivityEvent doc = new ActivityEvent(
- event.userId(),
- "CATEGORY_DELETED",
- "CATEGORY",
- event.categoryId(),
- before,
- Map.of(),
- event.occurredAt()
- );
- repository.save(doc);
- }
-}
diff --git a/task-service/src/main/java/com/mtole/task/kafka/ActivityLogKafkaConsumer.java b/task-service/src/main/java/com/mtole/task/kafka/ActivityLogKafkaConsumer.java
new file mode 100644
index 0000000..1d5e6dc
--- /dev/null
+++ b/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);
+ }
+ }
+}
diff --git a/task-service/src/main/java/com/mtole/task/kafka/CategoryEventKafkaPublisher.java b/task-service/src/main/java/com/mtole/task/kafka/CategoryEventKafkaPublisher.java
new file mode 100644
index 0000000..c2cac9a
--- /dev/null
+++ b/task-service/src/main/java/com/mtole/task/kafka/CategoryEventKafkaPublisher.java
@@ -0,0 +1,71 @@
+package com.mtole.task.kafka;
+
+import com.mtole.task.categories.events.CategoryCreatedEvent;
+import com.mtole.task.categories.events.CategoryDeletedEvent;
+import com.mtole.task.categories.events.CategoryUpdatedEvent;
+import com.mtole.task.kafka.events.CategoryEvent;
+import com.mtole.task.kafka.events.CategoryEventType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
+
+import java.util.UUID;
+
+import static com.mtole.task.kafka.KafkaConfig.TOPIC_CATEGORY_EVENTS;
+
+/**
+ * Publica eventos de dominio de Category al topic Kafka `category.events`
+ * tras el commit exitoso de la transacción JPA.
+ *
+ * Escucha los ApplicationEvent locales publicados por CategoryService y
+ * los traduce a CategoryEvent (contrato Kafka thin) con eventId nuevo.
+ * El eventId es generado aquí porque pertenece al contrato de Kafka,
+ * no al evento local Spring.
+ *
+ * Consistencia: @TransactionalEventListener(AFTER_COMMIT) garantiza que
+ * eventos de transacciones fallidas no se publican. No garantiza publicación
+ * ante fallo del broker (deuda técnica: transactional outbox, ADR-007).
+ */
+@Component
+public class CategoryEventKafkaPublisher {
+
+ private static final Logger log = LoggerFactory.getLogger(CategoryEventKafkaPublisher.class);
+
+ private final KafkaTemplate kafkaTemplate;
+
+ public CategoryEventKafkaPublisher(KafkaTemplate kafkaTemplate) {
+ this.kafkaTemplate = kafkaTemplate;
+ }
+
+ @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+ public void onCategoryCreated(CategoryCreatedEvent event) {
+ publish(event.userId(), event.categoryId(), CategoryEventType.CREATED, event.occurredAt());
+ }
+
+ @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+ public void onCategoryUpdated(CategoryUpdatedEvent event) {
+ publish(event.userId(), event.categoryId(), CategoryEventType.UPDATED, event.occurredAt());
+ }
+
+ @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+ public void onCategoryDeleted(CategoryDeletedEvent event) {
+ publish(event.userId(), event.categoryId(), CategoryEventType.DELETED, event.occurredAt());
+ }
+
+ private void publish(Long userId, Long categoryId, CategoryEventType type, java.time.Instant occurredAt) {
+ CategoryEvent event = new CategoryEvent(
+ UUID.randomUUID(),
+ type,
+ userId,
+ categoryId,
+ occurredAt
+ );
+ String key = userId.toString();
+ kafkaTemplate.send(TOPIC_CATEGORY_EVENTS, key, event);
+ log.debug("Published CategoryEvent {} for user={} category={} type={}",
+ event.eventId(), userId, categoryId, type);
+ }
+}
diff --git a/task-service/src/main/java/com/mtole/task/kafka/KafkaConfig.java b/task-service/src/main/java/com/mtole/task/kafka/KafkaConfig.java
new file mode 100644
index 0000000..e1d9bc5
--- /dev/null
+++ b/task-service/src/main/java/com/mtole/task/kafka/KafkaConfig.java
@@ -0,0 +1,50 @@
+package com.mtole.task.kafka;
+
+import org.apache.kafka.clients.admin.NewTopic;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.TopicBuilder;
+import org.springframework.kafka.listener.DefaultErrorHandler;
+import org.springframework.util.backoff.ExponentialBackOff;
+
+@Configuration
+public class KafkaConfig {
+
+ public static final String TOPIC_TASK_EVENTS = "task.events";
+ public static final String TOPIC_CATEGORY_EVENTS = "category.events";
+
+ @Bean
+ public NewTopic taskEventsTopic() {
+ return TopicBuilder.name(TOPIC_TASK_EVENTS)
+ .partitions(3)
+ .replicas(1)
+ .build();
+ }
+
+ @Bean
+ public NewTopic categoryEventsTopic() {
+ return TopicBuilder.name(TOPIC_CATEGORY_EVENTS)
+ .partitions(3)
+ .replicas(1)
+ .build();
+ }
+
+
+ /**
+ * ErrorHandler personalizado para consumers Kafka.
+ *
+ * Sustituye el DefaultErrorHandler por defecto de Spring Kafka, que aplica
+ * FixedBackOff(0ms, 9 intentos) y descarta el mensaje al agotar reintentos.
+ * Aquí usamos ExponentialBackOff sin techo total: reintentos indefinidos
+ * con espera creciente hasta 30s por intento. Preferimos consumer bloqueado
+ * en partición (incidente visible vía lag) a pérdida silenciosa.
+ *
+ * Aplica a todos los @KafkaListener del contexto por resolución por tipo.
+ */
+ @Bean
+ public DefaultErrorHandler kafkaErrorHandler() {
+ ExponentialBackOff backOff = new ExponentialBackOff(1000L, 2.0);
+ backOff.setMaxInterval(30_000L);
+ return new DefaultErrorHandler(backOff);
+ }
+}
diff --git a/task-service/src/main/java/com/mtole/task/kafka/TaskEventKafkaPublisher.java b/task-service/src/main/java/com/mtole/task/kafka/TaskEventKafkaPublisher.java
new file mode 100644
index 0000000..5beba12
--- /dev/null
+++ b/task-service/src/main/java/com/mtole/task/kafka/TaskEventKafkaPublisher.java
@@ -0,0 +1,77 @@
+package com.mtole.task.kafka;
+
+import com.mtole.task.kafka.events.TaskEvent;
+import com.mtole.task.kafka.events.TaskEventType;
+import com.mtole.task.tasks.events.TaskCreatedEvent;
+import com.mtole.task.tasks.events.TaskDeletedEvent;
+import com.mtole.task.tasks.events.TaskStatusChangedEvent;
+import com.mtole.task.tasks.events.TaskUpdatedEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
+
+import java.util.UUID;
+
+import static com.mtole.task.kafka.KafkaConfig.TOPIC_TASK_EVENTS;
+
+/**
+ * Publica eventos de dominio de Task al topic Kafka `task.events`
+ * tras el commit exitoso de la transacción JPA.
+ *
+ * Escucha los ApplicationEvent locales publicados por TaskService y
+ * los traduce a TaskEvent (contrato Kafka thin) con eventId nuevo.
+ * El eventId es generado aquí porque pertenece al contrato de Kafka,
+ * no al evento local Spring.
+ *
+ * Consistencia: @TransactionalEventListener(AFTER_COMMIT) garantiza que
+ * eventos de transacciones fallidas no se publican. No garantiza publicación
+ * ante fallo del broker (deuda técnica: transactional outbox, ADR-007).
+ */
+@Component
+public class TaskEventKafkaPublisher {
+
+ private static final Logger log = LoggerFactory.getLogger(TaskEventKafkaPublisher.class);
+
+ private final KafkaTemplate kafkaTemplate;
+
+ public TaskEventKafkaPublisher(KafkaTemplate kafkaTemplate) {
+ this.kafkaTemplate = kafkaTemplate;
+ }
+
+ @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+ public void onTaskCreated(TaskCreatedEvent event) {
+ publish(event.userId(), event.taskId(), TaskEventType.CREATED, event.occurredAt());
+ }
+
+ @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+ public void onTaskUpdated(TaskUpdatedEvent event) {
+ publish(event.userId(), event.taskId(), TaskEventType.UPDATED, event.occurredAt());
+ }
+
+ @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+ public void onTaskStatusChanged(TaskStatusChangedEvent event) {
+ publish(event.userId(), event.taskId(), TaskEventType.STATUS_CHANGED, event.occurredAt());
+ }
+
+ @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+ public void onTaskDeleted(TaskDeletedEvent event) {
+ publish(event.userId(), event.taskId(), TaskEventType.DELETED, event.occurredAt());
+ }
+
+ private void publish(Long userId, Long taskId, TaskEventType type, java.time.Instant occurredAt) {
+ TaskEvent event = new TaskEvent(
+ UUID.randomUUID(),
+ type,
+ userId,
+ taskId,
+ occurredAt
+ );
+ String key = userId.toString();
+ kafkaTemplate.send(TOPIC_TASK_EVENTS, key, event);
+ log.debug("Published TaskEvent {} for user={} task={} type={}",
+ event.eventId(), userId, taskId, type);
+ }
+}
diff --git a/task-service/src/main/java/com/mtole/task/kafka/events/CategoryEvent.java b/task-service/src/main/java/com/mtole/task/kafka/events/CategoryEvent.java
new file mode 100644
index 0000000..0ce5ce8
--- /dev/null
+++ b/task-service/src/main/java/com/mtole/task/kafka/events/CategoryEvent.java
@@ -0,0 +1,12 @@
+package com.mtole.task.kafka.events;
+
+import java.time.Instant;
+import java.util.UUID;
+
+public record CategoryEvent(
+ UUID eventId,
+ CategoryEventType type,
+ Long userId,
+ Long categoryId,
+ Instant timestamp
+) {}
diff --git a/task-service/src/main/java/com/mtole/task/kafka/events/CategoryEventType.java b/task-service/src/main/java/com/mtole/task/kafka/events/CategoryEventType.java
new file mode 100644
index 0000000..f09ed79
--- /dev/null
+++ b/task-service/src/main/java/com/mtole/task/kafka/events/CategoryEventType.java
@@ -0,0 +1,7 @@
+package com.mtole.task.kafka.events;
+
+public enum CategoryEventType {
+ CREATED,
+ UPDATED,
+ DELETED
+}
diff --git a/task-service/src/main/java/com/mtole/task/kafka/events/TaskEvent.java b/task-service/src/main/java/com/mtole/task/kafka/events/TaskEvent.java
new file mode 100644
index 0000000..d092341
--- /dev/null
+++ b/task-service/src/main/java/com/mtole/task/kafka/events/TaskEvent.java
@@ -0,0 +1,12 @@
+package com.mtole.task.kafka.events;
+
+import java.time.Instant;
+import java.util.UUID;
+
+public record TaskEvent(
+ UUID eventId,
+ TaskEventType type,
+ Long userId,
+ Long taskId,
+ Instant timestamp
+){}
diff --git a/task-service/src/main/java/com/mtole/task/kafka/events/TaskEventType.java b/task-service/src/main/java/com/mtole/task/kafka/events/TaskEventType.java
new file mode 100644
index 0000000..8106d7b
--- /dev/null
+++ b/task-service/src/main/java/com/mtole/task/kafka/events/TaskEventType.java
@@ -0,0 +1,8 @@
+package com.mtole.task.kafka.events;
+
+public enum TaskEventType {
+ CREATED,
+ UPDATED,
+ STATUS_CHANGED,
+ DELETED
+}
diff --git a/task-service/src/main/resources/application-docker.yml b/task-service/src/main/resources/application-docker.yml
index b3c5105..8f2c319 100644
--- a/task-service/src/main/resources/application-docker.yml
+++ b/task-service/src/main/resources/application-docker.yml
@@ -13,3 +13,6 @@ spring:
data:
mongodb:
uri: mongodb://mongo:27017/taskmanager
+
+ kafka:
+ bootstrap-servers: kafka:9092
diff --git a/task-service/src/main/resources/application.yml b/task-service/src/main/resources/application.yml
index 748cccb..59dec40 100644
--- a/task-service/src/main/resources/application.yml
+++ b/task-service/src/main/resources/application.yml
@@ -26,6 +26,22 @@ spring:
database: taskmanager
auto-index-creation: true
+ kafka:
+ bootstrap-servers: localhost:9092
+ producer:
+ key-serializer: org.apache.kafka.common.serialization.StringSerializer
+ value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
+ consumer:
+ group-id: activity-log-writer
+ key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+ value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
+ auto-offset-reset: earliest
+ enable-auto-commit: false
+ properties:
+ spring.json.trusted.packages: "com.mtole.task.kafka.events"
+ listener:
+ ack-mode: MANUAL_IMMEDIATE
+
management:
endpoints:
web: