[#5] Implement command service skeleton with Kafka producer

AGENTS.md

@@ -36,7 +36,7 @@
 - Hardhat config intentionally omits `accounts` for Ganache when `GANACHE_PRIVATE_KEY` is invalid/missing; do not replace with `accounts: []` (`blockchain/hardhat.config.js`).
 - Ganache chain assumptions are fixed by deploy stack: chainId `1337` and deterministic mnemonic (`deploy/docker-compose.yml`, `deploy/README.md`).
 - Git workflow conventions are documented and reused across docs:
-  - branch: `feature/#XX-description`
+  - branch (mandatory for every issue): `<type>/#XX-description`, where `type` = `feature|fix|docs|test`
   - commit: `[#XX] short message`
   - PR text includes `Closes #XX` (`START_HERE.md`, `CONTRIBUTING.md`).
 

CONTRIBUTING.md

@@ -184,16 +184,26 @@ docs/
 
 ## Branch Naming
 
-Project convention:
+Mandatory convention for all issue branches:
+
+```text
+<type>/#XX-description
+```
 
-| Prefix | Purpose | Example |
-|--------|---------|---------|
-| `feature/` | Feature delivery | `feature/#6-event-store-consumer` |
-| `fix/` | Bug fix | `fix/#7-retry-logic` |
-| `docs/` | Documentation | `docs/#12-architecture-update` |
-| `test/` | Tests | `test/#8-query-filters` |
+Allowed `type` values:
 
-Preferred pattern: `feature/#XX-description`.
+```text
+feature | fix | docs | test
+```
+
+Examples:
+
+```text
+feature/#5-command-service-skeleton
+fix/#7-retry-logic
+docs/#12-architecture-update
+test/#8-query-filters
+```
 
 ---
 

GITHUB_ISSUES_PLAN.md

@@ -139,10 +139,10 @@ backend/
 - Событие публикуется в Kafka topic `user.login.events`
 
 **Subtasks:**
-- [ ] #5.1 - Spring Boot приложение с Kafka producer
-- [ ] #5.2 - REST endpoint для `UserLoggedIn` события
-- [ ] #5.3 - Event DTO класс
-- [ ] #5.4 - application.properties и конфигурация
+- [x] #5.1 - Spring Boot приложение с Kafka producer
+- [x] #5.2 - REST endpoint для `UserLoggedIn` события
+- [x] #5.3 - Event DTO класс
+- [x] #5.4 - application.yml и конфигурация
 
 **Expected PR:** PR-5 (Command Service skeleton)
 
@@ -621,7 +621,7 @@ curl -X POST http://localhost:8081/commands/user/login ...
 ## Notes
 
 1. **Последовательность фаз:** MVP phase должна быть завершена перед Phase 2
-2. **Branching strategy:** Каждый issue → feature branch (`feature/#XX-description`)
+2. **Branching strategy:** Каждый issue → branch по шаблону `<type>/#XX-description`, где `type` = `feature|fix|docs|test`
 3. **PR reviews:** Minimum 1 approval перед merge
 4. **Commit messages:** `[#XX] Brief description` (с номером issue)
 5. **Project board:** Используем GitHub Project для визуализации статуса

START_HERE.md

@@ -19,7 +19,7 @@ Use this file as the single entry point for project setup and work planning.
 ## 3) Work mode
 
 - One feature per issue
-- Branch naming: `feature/#XX-description`
+- Branch naming: `<type>/#XX-description` (`feature|fix|docs|test`)
 - Commit naming: `[#XX] short message`
 - PR includes `Closes #XX`
 

backend/command-service/README.md

@@ -0,0 +1,33 @@
+# Command Service (Issue #5)
+
+MVP command-side service that accepts user login commands and publishes `UserLoggedInEvent` to Kafka.
+
+## Implemented
+
+- WebFlux endpoint: `POST /commands/user/login`
+- Kafka producer to topic `user.login.events`
+- Temporary in-memory event storage for accepted events
+- Validation + error handling responses in `CommandResponse` format
+
+## Run
+
+From `backend/`:
+
+```pwsh
+mvn spring-boot:run -pl command-service -am
+```
+
+## Quick check
+
+```pwsh
+curl -X POST http://localhost:8081/commands/user/login -H "Content-Type: application/json" -d '{"userId":"user1"}'
+```
+
+## Test
+
+From `backend/`:
+
+```pwsh
+mvn -pl command-service -am test
+```
+

backend/command-service/pom.xml

@@ -47,6 +47,11 @@
         </dependency>
 
         <!-- Test -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-webflux-test</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.springframework.kafka</groupId>
             <artifactId>spring-kafka-test</artifactId>

backend/command-service/src/main/java/lt/satsyuk/distributed/audit/command/CommandServiceApplication.java

@@ -2,6 +2,7 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.context.properties.ConfigurationPropertiesScan;
 
 /**
  * Command Service — entry point.
@@ -10,6 +11,7 @@
  * corresponding domain events to Kafka. Runs on port 8081.
  */
 @SpringBootApplication
+@ConfigurationPropertiesScan
 public class CommandServiceApplication {
 
     public static void main(String[] args) {

backend/command-service/src/main/java/lt/satsyuk/distributed/audit/command/api/CommandController.java

@@ -0,0 +1,40 @@
+package lt.satsyuk.distributed.audit.command.api;
+
+import jakarta.validation.Valid;
+import lt.satsyuk.distributed.audit.command.service.UserLoginCommandService;
+import lt.satsyuk.distributed.audit.contracts.command.UserLoginCommand;
+import lt.satsyuk.distributed.audit.contracts.dto.CommandResponse;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestHeader;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.http.server.reactive.ServerHttpRequest;
+import reactor.core.publisher.Mono;
+
+@RestController
+public class CommandController {
+
+    private final UserLoginCommandService userLoginCommandService;
+
+    public CommandController(UserLoginCommandService userLoginCommandService) {
+        this.userLoginCommandService = userLoginCommandService;
+    }
+
+    @PostMapping("/commands/user/login")
+    public Mono<ResponseEntity<CommandResponse>> userLogin(
+            @Valid @RequestBody UserLoginCommand command,
+            @RequestHeader(value = HttpHeaders.USER_AGENT, required = false) String requestUserAgent,
+            ServerHttpRequest request
+    ) {
+        String requestIp = null;
+        if (request.getRemoteAddress() != null && request.getRemoteAddress().getAddress() != null) {
+            requestIp = request.getRemoteAddress().getAddress().getHostAddress();
+        }
+
+        return userLoginCommandService.handleUserLogin(command, requestIp, requestUserAgent)
+                .map(response -> ResponseEntity.accepted().body(response));
+    }
+}
+

backend/command-service/src/main/java/lt/satsyuk/distributed/audit/command/api/GlobalExceptionHandler.java

@@ -0,0 +1,51 @@
+package lt.satsyuk.distributed.audit.command.api;
+
+import lt.satsyuk.distributed.audit.command.service.CommandPublishException;
+import lt.satsyuk.distributed.audit.contracts.dto.CommandResponse;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.validation.FieldError;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+import org.springframework.web.bind.annotation.RestControllerAdvice;
+import org.springframework.web.bind.support.WebExchangeBindException;
+import org.springframework.web.server.ServerWebInputException;
+
+import java.util.Objects;
+
+import java.util.stream.Collectors;
+
+@RestControllerAdvice
+public class GlobalExceptionHandler {
+
+    @ExceptionHandler(WebExchangeBindException.class)
+    public ResponseEntity<CommandResponse> handleValidationError(WebExchangeBindException exception) {
+        String message = exception.getBindingResult()
+                .getFieldErrors()
+                .stream()
+                .map(this::formatFieldError)
+                .collect(Collectors.joining("; "));
+
+        return ResponseEntity.badRequest().body(CommandResponse.rejected(message));
+    }
+
+    @ExceptionHandler(ServerWebInputException.class)
+    public ResponseEntity<CommandResponse> handleWebInputError(ServerWebInputException exception) {
+        String message = Objects.requireNonNullElse(exception.getReason(), "Invalid request payload");
+        return ResponseEntity.badRequest().body(CommandResponse.rejected(message));
+    }
+
+    @ExceptionHandler(CommandPublishException.class)
+    public ResponseEntity<CommandResponse> handlePublishError(CommandPublishException exception) {
+        return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE)
+                .body(CommandResponse.rejected(exception.getMessage()));
+    }
+
+    private String formatFieldError(FieldError error) {
+        String defaultMessage = Objects.requireNonNullElse(error.getDefaultMessage(), "Validation failed");
+        if (defaultMessage.contains(error.getField())) {
+            return defaultMessage;
+        }
+        return error.getField() + " " + defaultMessage;
+    }
+}
+

backend/command-service/src/main/java/lt/satsyuk/distributed/audit/command/config/KafkaTopicsProperties.java

@@ -0,0 +1,21 @@
+package lt.satsyuk.distributed.audit.command.config;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Externalized Kafka topic names used by command-service.
+ */
+@ConfigurationProperties(prefix = "kafka.topics")
+public class KafkaTopicsProperties {
+
+    private String userLoginEvents;
+
+    public String getUserLoginEvents() {
+        return userLoginEvents;
+    }
+
+    public void setUserLoginEvents(String userLoginEvents) {
+        this.userLoginEvents = userLoginEvents;
+    }
+}
+

backend/command-service/src/main/java/lt/satsyuk/distributed/audit/command/repository/InMemoryEventStorage.java

@@ -0,0 +1,33 @@
+package lt.satsyuk.distributed.audit.command.repository;
+
+import lt.satsyuk.distributed.audit.event.AuditEvent;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Temporary in-memory storage for accepted events (MVP stage for issue #5).
+ */
+@Component
+public class InMemoryEventStorage {
+
+    private final ConcurrentLinkedQueue<AuditEvent> events = new ConcurrentLinkedQueue<>();
+    private final AtomicInteger eventsCount = new AtomicInteger(0);
+
+    public void save(AuditEvent event) {
+        events.add(event);
+        eventsCount.incrementAndGet();
+    }
+
+    public List<AuditEvent> findAll() {
+        return List.copyOf(new ArrayList<>(events));
+    }
+
+    public int count() {
+        return eventsCount.get();
+    }
+}
+

backend/command-service/src/main/java/lt/satsyuk/distributed/audit/command/service/CommandPublishException.java

@@ -0,0 +1,12 @@
+package lt.satsyuk.distributed.audit.command.service;
+
+/**
+ * Raised when a command cannot be published to Kafka.
+ */
+public class CommandPublishException extends RuntimeException {
+
+    public CommandPublishException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}
+

backend/command-service/src/main/java/lt/satsyuk/distributed/audit/command/service/UserLoginCommandService.java

@@ -0,0 +1,52 @@
+package lt.satsyuk.distributed.audit.command.service;
+
+import lt.satsyuk.distributed.audit.command.config.KafkaTopicsProperties;
+import lt.satsyuk.distributed.audit.command.repository.InMemoryEventStorage;
+import lt.satsyuk.distributed.audit.contracts.command.UserLoginCommand;
+import lt.satsyuk.distributed.audit.contracts.dto.CommandResponse;
+import lt.satsyuk.distributed.audit.event.AuditEvent;
+import lt.satsyuk.distributed.audit.event.UserLoggedInEvent;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
+@Service
+public class UserLoginCommandService {
+
+    private final KafkaTemplate<String, AuditEvent> kafkaTemplate;
+    private final KafkaTopicsProperties kafkaTopicsProperties;
+    private final InMemoryEventStorage inMemoryEventStorage;
+
+    public UserLoginCommandService(
+            KafkaTemplate<String, AuditEvent> kafkaTemplate,
+            KafkaTopicsProperties kafkaTopicsProperties,
+            InMemoryEventStorage inMemoryEventStorage
+    ) {
+        this.kafkaTemplate = kafkaTemplate;
+        this.kafkaTopicsProperties = kafkaTopicsProperties;
+        this.inMemoryEventStorage = inMemoryEventStorage;
+    }
+
+    public Mono<CommandResponse> handleUserLogin(UserLoginCommand command, String requestIp, String requestUserAgent) {
+        // Prefer server-derived metadata over client-supplied body values to prevent spoofing.
+        String effectiveIp = StringUtils.hasText(requestIp) ? requestIp : command.getIpAddress();
+        String effectiveUserAgent = StringUtils.hasText(requestUserAgent)
+                ? requestUserAgent
+                : command.getUserAgent();
+
+        UserLoggedInEvent event = UserLoggedInEvent.of(command.getUserId(), effectiveIp, effectiveUserAgent);
+
+        return Mono.defer(() -> Mono.fromFuture(kafkaTemplate.send(
+                                kafkaTopicsProperties.getUserLoginEvents(),
+                                event.getEventId(),
+                                event
+                        )))
+                .publishOn(Schedulers.boundedElastic())
+                .doOnNext(ignored -> inMemoryEventStorage.save(event))
+                .map(ignored -> CommandResponse.accepted(event.getEventId()))
+                .onErrorMap(error -> new CommandPublishException("Failed to publish event to Kafka", error));
+    }
+}
+