[#6] Event Store service: Kafka -> PostgreSQL persistence

GITHUB_ISSUES_PLAN.md

@@ -176,11 +176,11 @@ CREATE TABLE audit.events (
 ```
 
 **Subtasks:**
-- [ ] #6.1 - Spring Boot + Spring Data R2DBC setup
-- [ ] #6.2 - Entity класс и repository
-- [ ] #6.3 - Kafka consumer
-- [ ] #6.4 - Liquibase / Flyway миграции
-- [ ] #6.5 - Тесты
+- [x] #6.1 - Spring Boot + Spring Data R2DBC setup
+- [x] #6.2 - Entity класс и repository
+- [x] #6.3 - Kafka consumer
+- [x] #6.4 - Liquibase / Flyway миграции
+- [x] #6.5 - Тесты
 
 **Expected PR:** PR-6 (Event Store service)
 

backend/event-store-service/README.md

@@ -0,0 +1,32 @@
+# Event Store Service
+
+`event-store-service` реализует Issue #6: читает события из Kafka (`user.login.events`) и сохраняет их в PostgreSQL (`audit.events`).
+
+## Что внутри
+
+- Kafka consumer (`UserLoginEventConsumer`)
+- Преобразование `AuditEvent` -> `StoredAuditEvent`
+- Расчет `SHA-256` хэша payload (`event_hash`)
+- Сохранение через Spring Data R2DBC
+- Flyway миграции в `src/main/resources/db/migration`
+
+## Быстрый запуск
+
+Из каталога `backend/`:
+
+```bash
+mvn spring-boot:run -pl event-store-service -am
+```
+
+## Тесты
+
+Из каталога `backend/`:
+
+```bash
+mvn -pl event-store-service test
+```
+
+Интеграционный сценарий Kafka -> PostgreSQL покрыт тестом
+`EventStoreKafkaToPostgresIntegrationTest` на Testcontainers.
+Если Docker недоступен, этот тест будет автоматически пропущен.
+

backend/event-store-service/pom.xml

@@ -51,7 +51,6 @@
         <dependency>
             <groupId>org.postgresql</groupId>
             <artifactId>r2dbc-postgresql</artifactId>
-            <scope>runtime</scope>
         </dependency>
 
         <!-- Flyway for schema migrations -->
@@ -80,6 +79,11 @@
             <artifactId>kafka</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

backend/event-store-service/src/main/java/lt/satsyuk/distributed/audit/eventstore/EventStoreServiceApplication.java

@@ -1,7 +1,10 @@
 package lt.satsyuk.distributed.audit.eventstore;
 
+import lt.satsyuk.distributed.audit.eventstore.config.KafkaTopicsProperties;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.kafka.annotation.EnableKafka;
 
 /**
  * Event Store Service — entry point.
@@ -11,6 +14,8 @@
  * to the {@code audit.events} PostgreSQL table. Runs on port 8082.
  */
 @SpringBootApplication
+@EnableKafka
+@EnableConfigurationProperties(KafkaTopicsProperties.class)
 public class EventStoreServiceApplication {
 
     public static void main(String[] args) {

backend/event-store-service/src/main/java/lt/satsyuk/distributed/audit/eventstore/config/JacksonConfig.java

@@ -0,0 +1,16 @@
+package lt.satsyuk.distributed.audit.eventstore.config;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class JacksonConfig {
+
+    @Bean
+    public ObjectMapper objectMapper() {
+        return JsonMapper.builder().findAndAddModules().build();
+    }
+}
+

backend/event-store-service/src/main/java/lt/satsyuk/distributed/audit/eventstore/config/KafkaListenerConfig.java

@@ -0,0 +1,49 @@
+package lt.satsyuk.distributed.audit.eventstore.config;
+
+import lt.satsyuk.distributed.audit.event.AuditEvent;
+import lt.satsyuk.distributed.audit.event.UserLoggedInEvent;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaListenerConfig {
+
+    @Bean
+    public ConsumerFactory<String, AuditEvent> consumerFactory(
+            @Value("${spring.kafka.bootstrap-servers}") String bootstrapServers,
+            @Value("${spring.kafka.consumer.group-id:event-store-consumer}") String groupId
+    ) {
+        Map<String, Object> props = new HashMap<>();
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
+        props.put(JsonDeserializer.TRUSTED_PACKAGES, "lt.satsyuk.distributed.audit.event");
+        props.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, false);
+        props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, UserLoggedInEvent.class.getName());
+
+        return new DefaultKafkaConsumerFactory<>(props);
+    }
+
+    @Bean(name = "kafkaListenerContainerFactory")
+    public ConcurrentKafkaListenerContainerFactory<String, AuditEvent> kafkaListenerContainerFactory(
+            ConsumerFactory<String, AuditEvent> consumerFactory
+    ) {
+        ConcurrentKafkaListenerContainerFactory<String, AuditEvent> factory =
+                new ConcurrentKafkaListenerContainerFactory<>();
+        factory.setConsumerFactory(consumerFactory);
+        return factory;
+    }
+}
+

backend/event-store-service/src/main/java/lt/satsyuk/distributed/audit/eventstore/config/KafkaTopicsProperties.java

@@ -0,0 +1,14 @@
+package lt.satsyuk.distributed.audit.eventstore.config;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@ConfigurationProperties(prefix = "kafka.topics")
+@Getter
+@Setter
+public class KafkaTopicsProperties {
+
+    private String userLoginEvents = "user.login.events";
+}
+

backend/event-store-service/src/main/java/lt/satsyuk/distributed/audit/eventstore/consumer/UserLoginEventConsumer.java

@@ -0,0 +1,47 @@
+package lt.satsyuk.distributed.audit.eventstore.consumer;
+
+import lt.satsyuk.distributed.audit.event.AuditEvent;
+import lt.satsyuk.distributed.audit.event.UserLoggedInEvent;
+import lt.satsyuk.distributed.audit.eventstore.config.KafkaTopicsProperties;
+import lt.satsyuk.distributed.audit.eventstore.service.EventPersistenceService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.stereotype.Component;
+
+import static org.springframework.kafka.support.KafkaHeaders.RECEIVED_KEY;
+
+@Component
+public class UserLoginEventConsumer {
+
+    private static final Logger log = LoggerFactory.getLogger(UserLoginEventConsumer.class);
+
+    private final EventPersistenceService eventPersistenceService;
+    private final KafkaTopicsProperties kafkaTopicsProperties;
+
+    public UserLoginEventConsumer(
+            EventPersistenceService eventPersistenceService,
+            KafkaTopicsProperties kafkaTopicsProperties
+    ) {
+        this.eventPersistenceService = eventPersistenceService;
+        this.kafkaTopicsProperties = kafkaTopicsProperties;
+    }
+
+    @KafkaListener(topics = "${kafka.topics.user-login-events}")
+    public void consume(AuditEvent event, @Header(value = RECEIVED_KEY, required = false) String key) {
+        if (!(event instanceof UserLoggedInEvent userLoggedInEvent)) {
+            log.warn("Skipping unsupported event type from topic [{}], key=[{}]", kafkaTopicsProperties.getUserLoginEvents(), key);
+            return;
+        }
+
+        try {
+            // Keep Kafka offset handling aligned with DB write result.
+            eventPersistenceService.persist(userLoggedInEvent).block();
+        } catch (RuntimeException ex) {
+            log.error("Failed to persist event key=[{}]", key, ex);
+            throw new IllegalStateException("Failed to persist event key=[" + key + "]", ex);
+        }
+    }
+}
+

backend/event-store-service/src/main/java/lt/satsyuk/distributed/audit/eventstore/model/StoredAuditEvent.java

@@ -0,0 +1,42 @@
+package lt.satsyuk.distributed.audit.eventstore.model;
+
+import io.r2dbc.postgresql.codec.Json;
+import lombok.Getter;
+import lombok.Setter;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.relational.core.mapping.Column;
+import org.springframework.data.relational.core.mapping.Table;
+
+import java.time.LocalDateTime;
+
+@Table(schema = "audit", name = "events")
+@Getter
+@Setter
+public class StoredAuditEvent {
+
+    @Id
+    private Long id;
+
+    @Column("event_id")
+    private String eventId;
+
+    @Column("aggregate_id")
+    private String aggregateId;
+
+    @Column("event_type")
+    private String eventType;
+
+    @Column("user_id")
+    private String userId;
+
+    @Column("payload")
+    private Json payload;
+
+    @Column("event_hash")
+    private String eventHash;
+
+    @Column("created_at")
+    private LocalDateTime createdAt;
+
+}
+

backend/event-store-service/src/main/java/lt/satsyuk/distributed/audit/eventstore/repository/StoredAuditEventRepository.java

@@ -0,0 +1,8 @@
+package lt.satsyuk.distributed.audit.eventstore.repository;
+
+import lt.satsyuk.distributed.audit.eventstore.model.StoredAuditEvent;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+
+public interface StoredAuditEventRepository extends ReactiveCrudRepository<StoredAuditEvent, Long> {
+}
+

backend/event-store-service/src/main/java/lt/satsyuk/distributed/audit/eventstore/service/EventHashService.java

@@ -0,0 +1,30 @@
+package lt.satsyuk.distributed.audit.eventstore.service;
+
+import org.springframework.stereotype.Component;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+
+@Component
+public class EventHashService {
+
+    public String sha256Hex(String payload) {
+        try {
+            MessageDigest digest = MessageDigest.getInstance("SHA-256");
+            byte[] hash = digest.digest(payload.getBytes(StandardCharsets.UTF_8));
+            return bytesToHex(hash);
+        } catch (NoSuchAlgorithmException ex) {
+            throw new IllegalStateException("SHA-256 is not available", ex);
+        }
+    }
+
+    private String bytesToHex(byte[] bytes) {
+        StringBuilder result = new StringBuilder(bytes.length * 2);
+        for (byte value : bytes) {
+            result.append(String.format("%02x", value));
+        }
+        return result.toString();
+    }
+}
+

backend/event-store-service/src/main/java/lt/satsyuk/distributed/audit/eventstore/service/EventPersistenceService.java

@@ -0,0 +1,91 @@
+package lt.satsyuk.distributed.audit.eventstore.service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.r2dbc.postgresql.codec.Json;
+import lt.satsyuk.distributed.audit.event.AuditEvent;
+import lt.satsyuk.distributed.audit.event.UserLoggedInEvent;
+import lt.satsyuk.distributed.audit.eventstore.model.StoredAuditEvent;
+import lt.satsyuk.distributed.audit.eventstore.repository.StoredAuditEventRepository;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Mono;
+import reactor.util.retry.Retry;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+
+@Service
+public class EventPersistenceService {
+
+    private static final Logger log = LoggerFactory.getLogger(EventPersistenceService.class);
+
+    private final ObjectMapper objectMapper;
+    private final EventHashService eventHashService;
+    private final StoredAuditEventRepository repository;
+
+    public EventPersistenceService(
+            ObjectMapper objectMapper,
+            EventHashService eventHashService,
+            StoredAuditEventRepository repository
+    ) {
+        this.objectMapper = objectMapper;
+        this.eventHashService = eventHashService;
+        this.repository = repository;
+    }
+
+    public Mono<StoredAuditEvent> persist(AuditEvent event) {
+        return Mono.fromCallable(() -> toEntity(event))
+                .flatMap(repository::save)
+                .doOnSuccess(saved -> {
+                    if (saved != null) {
+                        log.debug("Saved event [{}] to audit.events", saved.getEventId());
+                    }
+                })
+                .retryWhen(Retry.backoff(3, Duration.ofMillis(200))
+                        .filter(DataAccessResourceFailureException.class::isInstance))
+                .onErrorResume(DuplicateKeyException.class, ignored -> {
+                    log.info("Event [{}] already persisted, skipping duplicate", event.getEventId());
+                    return Mono.empty();
+                });
+    }
+
+    private StoredAuditEvent toEntity(AuditEvent event) throws JsonProcessingException {
+        String payloadJson = objectMapper.writeValueAsString(event);
+        String aggregateId = resolveAggregateId(event);
+
+        StoredAuditEvent entity = new StoredAuditEvent();
+        entity.setEventId(event.getEventId());
+        entity.setAggregateId(aggregateId);
+        entity.setEventType(event.getEventType().name());
+        entity.setUserId(resolveUserId(event));
+        entity.setPayload(Json.of(payloadJson));
+        entity.setEventHash(eventHashService.sha256Hex(payloadJson));
+        entity.setCreatedAt(LocalDateTime.ofInstant(resolveTimestamp(event), ZoneOffset.UTC));
+        return entity;
+    }
+
+    private String resolveAggregateId(AuditEvent event) {
+        if (event instanceof UserLoggedInEvent userLoggedInEvent && userLoggedInEvent.getUserId() != null) {
+            return "user:" + userLoggedInEvent.getUserId();
+        }
+        return event.getEventId();
+    }
+
+    private String resolveUserId(AuditEvent event) {
+        if (event instanceof UserLoggedInEvent userLoggedInEvent) {
+            return userLoggedInEvent.getUserId();
+        }
+        return null;
+    }
+
+    private Instant resolveTimestamp(AuditEvent event) {
+        return event.getOccurredAt() != null ? event.getOccurredAt() : Instant.now();
+    }
+}
+