🧑‍💻 chore(dx): add RestApi for read events and reset processors (#17)

Author: MateuszNaKodach

src/main/java/com/dddheroes/heroesofddd/creaturerecruitment/read/DwellingReadModelProjector.java

@@ -6,6 +6,7 @@
 import com.dddheroes.heroesofddd.shared.GameMetaData;
 import org.axonframework.config.ProcessingGroup;
 import org.axonframework.eventhandling.EventHandler;
+import org.axonframework.eventhandling.ResetHandler;
 import org.axonframework.messaging.annotation.MetaDataValue;
 import org.springframework.stereotype.Component;
 
@@ -44,4 +45,9 @@ void on(CreatureRecruited event) {
                   .map(state -> state.withAvailableCreaturesDecreasedBy(event.quantity()))
                   .ifPresent(repository::save);
     }
+
+    @ResetHandler
+    void onReset() {
+        repository.deleteAll();
+    }
 }

src/main/java/com/dddheroes/heroesofddd/maintenance/read/geteventstream/EventStreamsRestApi.java

@@ -0,0 +1,28 @@
+package com.dddheroes.heroesofddd.maintenance.read.geteventstream;
+
+import org.axonframework.eventhandling.DomainEventMessage;
+import org.axonframework.eventsourcing.eventstore.EventStore;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.web.bind.annotation.CrossOrigin;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+@ConditionalOnProperty(name = "application.maintenance.enabled", havingValue = "true")
+@RestController
+class EventStreamsRestApi {
+
+    private final EventStore eventStore;
+
+    EventStreamsRestApi(EventStore eventStore) {
+        this.eventStore = eventStore;
+    }
+
+    @CrossOrigin
+    @GetMapping("/maintenance/event-store/streams/{streamId}/events")
+    List<? extends DomainEventMessage<?>> readEvents(@PathVariable String streamId) {
+        return eventStore.readEvents(streamId).asStream().toList();
+    }
+}

src/main/java/com/dddheroes/heroesofddd/maintenance/write/resetprocessor/ResetStreamProcessorRestApi.java

@@ -0,0 +1,24 @@
+package com.dddheroes.heroesofddd.maintenance.write.resetprocessor;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.web.bind.annotation.CrossOrigin;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@ConditionalOnProperty(name = "application.maintenance.enabled", havingValue = "true")
+@RestController
+class ResetStreamProcessorRestApi {
+
+    private final StreamProcessorsOperations streamProcessorsOperations;
+
+    ResetStreamProcessorRestApi(StreamProcessorsOperations streamProcessorsOperations) {
+        this.streamProcessorsOperations = streamProcessorsOperations;
+    }
+
+    @CrossOrigin
+    @PostMapping("/maintenance/processors/{name}/resets")
+    void resetProcessor(@PathVariable String name) {
+        streamProcessorsOperations.reset(name);
+    }
+}

src/main/java/com/dddheroes/heroesofddd/maintenance/write/resetprocessor/StreamProcessorsOperations.java

@@ -0,0 +1,69 @@
+package com.dddheroes.heroesofddd.maintenance.write.resetprocessor;
+
+import org.axonframework.config.EventProcessingConfiguration;
+import org.axonframework.eventhandling.ReplayToken;
+import org.axonframework.eventhandling.TrackingEventProcessor;
+import org.axonframework.eventhandling.tokenstore.TokenStore;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.stream.IntStream;
+
+@Component
+public class StreamProcessorsOperations {
+
+    private final EventProcessingConfiguration eventProcessingConfiguration;
+    private final TokenStore tokenStore;
+
+    StreamProcessorsOperations(EventProcessingConfiguration eventProcessingConfiguration, TokenStore tokenStore) {
+        this.eventProcessingConfiguration = eventProcessingConfiguration;
+        this.tokenStore = tokenStore;
+    }
+
+    public void reset(String processor) {
+        eventProcessingConfiguration
+                .eventProcessorByProcessingGroup(processor, TrackingEventProcessor.class)
+                .ifPresent(eventProcessor -> {
+                    if (eventProcessor.supportsReset()) {
+                        eventProcessor.shutDown();
+                        eventProcessor.resetTokens();
+                        eventProcessor.start();
+                    }
+                });
+    }
+
+    @Transactional
+    public Optional<Progress> progressOf(String processor) {
+        var segments = tokenStore.fetchSegments(processor);
+
+        if (segments.length == 0) {
+            return Optional.empty();
+        } else {
+            var accumulatedProgress = IntStream.of(segments).mapToObj(segment -> {
+                var token = tokenStore.fetchToken(processor, segment);
+
+                var maybeCurrent = token.position();
+                var maybePositionAtReset = token instanceof ReplayToken replayToken
+                        ? replayToken.getTokenAtReset().position()
+                        : OptionalLong.empty();
+
+                return new Progress(maybeCurrent.orElse(0L), maybePositionAtReset.orElse(0L));
+            }).reduce(new Progress(0, 0), (acc, progress) ->
+                    new Progress(acc.current + progress.current, acc.tail + progress.tail));
+
+            return (accumulatedProgress.tail == 0L) ? Optional.empty() : Optional.of(accumulatedProgress);
+        }
+    }
+
+    public record Progress(long current, long tail) {
+
+        public BigDecimal progress() {
+            return BigDecimal.valueOf(current, 2)
+                             .divide(BigDecimal.valueOf(tail, 2), RoundingMode.HALF_UP);
+        }
+    }
+}

src/main/resources/application.yaml

@@ -43,4 +43,7 @@ axon:
         sequencing-policy: gameIdSequencingPolicy
       Read_GetAllDwellings_QueryCache:
         mode: subscribing
-        sequencing-policy: gameIdSequencingPolicy
+        sequencing-policy: gameIdSequencingPolicy
+application:
+  maintenance:
+    enabled: true

src/test/java/com/dddheroes/heroesofddd/astrologers/automation/whenweekstartedthenproclaimweeksymbol/WhenWeekStartedThenProclaimWeekSymbolTest.java

@@ -5,6 +5,7 @@
 import com.dddheroes.heroesofddd.calendar.write.CalendarEvent;
 import com.dddheroes.heroesofddd.calendar.write.CalendarId;
 import com.dddheroes.heroesofddd.calendar.write.startday.DayStarted;
+import com.dddheroes.heroesofddd.maintenance.write.resetprocessor.StreamProcessorsOperations;
 import com.dddheroes.heroesofddd.shared.GameId;
 import com.dddheroes.heroesofddd.shared.GameMetaData;
 import com.dddheroes.heroesofddd.shared.PlayerId;
@@ -34,6 +35,9 @@ class WhenWeekStartedThenProclaimWeekSymbolTest {
     @Autowired
     private EventGateway eventGateway;
 
+    @Autowired
+    private StreamProcessorsOperations streamProcessorsOperations;
+
     @MockitoSpyBean
     private CommandGateway commandGateway;
 
@@ -56,6 +60,32 @@ void whenDayStartedForFirstDayOfTheWeek_ThenProclaimWeekSymbol() {
         );
     }
 
+    @Test
+    void givenDisallowedReplay_WhenReplayed_ThenShouldNotResendTheCommand() {
+        // given
+        var gameId = UUID.randomUUID().toString();
+        var calendarId = CalendarId.of(gameId);
+        givenCalendarEvents(
+                gameId,
+                new DayStarted(calendarId.raw(), 1, 1, 1)
+        );
+
+        // when
+        // processed by the automation
+
+        // then
+        awaitUntilAsserted(() -> verify(commandGateway, times(1))
+                .sendAndWait(ProclaimWeekSymbol.command(gameId, 1, 1, "angel", any()), eq(gameMetaData()))
+        );
+
+        // when
+        streamProcessorsOperations.reset("Automation_WhenWeekStartedThenProclaimWeekSymbol_Processor");
+
+        // then
+        verify(commandGateway, times(1))
+                .sendAndWait(ProclaimWeekSymbol.command(gameId, 1, 1, "angel", any()), eq(gameMetaData()));
+    }
+
     private void givenCalendarEvents(String gameId, CalendarEvent... events) {
         for (int i = 0; i < events.length; i++) {
             eventGateway.publish(calendarDomainEvent(gameId, i, events[i]));

src/test/java/com/dddheroes/heroesofddd/creaturerecruitment/read/DwellingReadModelTest.java

@@ -0,0 +1,44 @@
+package com.dddheroes.heroesofddd.creaturerecruitment.read;
+
+import com.dddheroes.heroesofddd.creaturerecruitment.write.DwellingEvent;
+import com.dddheroes.heroesofddd.shared.GameId;
+import com.dddheroes.heroesofddd.shared.GameMetaData;
+import com.dddheroes.heroesofddd.shared.PlayerId;
+import com.dddheroes.heroesofddd.shared.ResourceType;
+import org.axonframework.eventhandling.DomainEventMessage;
+import org.axonframework.eventhandling.GenericDomainEventMessage;
+import org.axonframework.eventhandling.gateway.EventGateway;
+
+import java.util.Map;
+
+public abstract class DwellingReadModelTest {
+
+    protected final String GAME_ID = GameId.random().raw();
+    protected static final String PLAYER_ID = PlayerId.random().raw();
+    protected static final Map<String, Integer> PHOENIX_COST = Map.of(
+            ResourceType.GOLD.name(), 2000,
+            ResourceType.MERCURY.name(), 1
+    );
+
+    protected EventGateway eventGateway;
+
+    protected DwellingReadModelTest(EventGateway eventGateway) {
+        this.eventGateway = eventGateway;
+    }
+
+    protected void givenDwellingEvents(String dwellingId, DwellingEvent... events) {
+        for (int i = 0; i < events.length; i++) {
+            eventGateway.publish(dwellingDomainEvent(dwellingId, i, events[i]));
+        }
+    }
+
+    protected DomainEventMessage<?> dwellingDomainEvent(String dwellingId, int sequenceNumber,
+                                                             DwellingEvent payload) {
+        return new GenericDomainEventMessage<>(
+                "Dwelling",
+                dwellingId,
+                sequenceNumber,
+                payload
+        ).andMetaData(GameMetaData.with(GAME_ID, PLAYER_ID));
+    }
+}

src/test/java/com/dddheroes/heroesofddd/creaturerecruitment/read/getalldwellings/GetAllDwellingsTest.java

@@ -0,0 +1,91 @@
+package com.dddheroes.heroesofddd.creaturerecruitment.read.getalldwellings;
+
+import com.dddheroes.heroesofddd.TestcontainersConfiguration;
+import com.dddheroes.heroesofddd.creaturerecruitment.read.DwellingReadModelTest;
+import com.dddheroes.heroesofddd.creaturerecruitment.write.DwellingId;
+import com.dddheroes.heroesofddd.creaturerecruitment.write.builddwelling.DwellingBuilt;
+import com.dddheroes.heroesofddd.shared.CreatureIds;
+import org.axonframework.eventhandling.gateway.EventGateway;
+import org.axonframework.queryhandling.QueryGateway;
+import org.junit.jupiter.api.*;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Import;
+
+import static com.dddheroes.heroesofddd.utils.AwaitilityUtils.awaitUntilAsserted;
+import static org.assertj.core.api.Assertions.assertThat;
+
+@Import(TestcontainersConfiguration.class)
+@SpringBootTest
+class GetAllDwellingsTest extends DwellingReadModelTest {
+
+    private final QueryGateway queryGateway;
+
+    @Autowired
+    GetAllDwellingsTest(
+            EventGateway eventGateway,
+            QueryGateway queryGateway
+    ) {
+        super(eventGateway);
+        this.queryGateway = queryGateway;
+    }
+
+    @Test
+    void projectingDwellingReadModel_TestCase1() {
+        // when
+        var query = getAllDwellings();
+
+        // then
+        awaitUntilAsserted(() -> {
+            var result = getDwellingReadModel(query).dwellings();
+            assertThat(result).isEmpty();
+        });
+    }
+
+    @Test
+    void projectingDwellingReadModel_TestCase2() {
+        // given
+        var creatureId = CreatureIds.phoenix().raw();
+
+        var dwellingId1 = DwellingId.random().raw();
+        givenDwellingEvents(
+                dwellingId1,
+                new DwellingBuilt(dwellingId1, creatureId, PHOENIX_COST)
+        );
+        var dwellingId2 = DwellingId.random().raw();
+        givenDwellingEvents(
+                dwellingId2,
+                new DwellingBuilt(dwellingId2, creatureId, PHOENIX_COST)
+        );
+
+        // when
+        var query = getAllDwellings();
+
+        // then
+        awaitUntilAsserted(() -> {
+            var result = getDwellingReadModel(query).dwellings();
+            assertThat(result).hasSize(2);
+            assertThat(result).anySatisfy(dwelling -> {
+                assertThat(dwelling.getDwellingId()).isEqualTo(dwellingId1);
+                assertThat(dwelling.getCreatureId()).isEqualTo(creatureId);
+                assertThat(dwelling.getCostPerTroop()).isEqualTo(PHOENIX_COST);
+                assertThat(dwelling.getAvailableCreatures()).isEqualTo(0);
+            });
+            assertThat(result).anySatisfy(dwelling -> {
+                assertThat(dwelling.getDwellingId()).isEqualTo(dwellingId2);
+                assertThat(dwelling.getCreatureId()).isEqualTo(creatureId);
+                assertThat(dwelling.getCostPerTroop()).isEqualTo(PHOENIX_COST);
+                assertThat(dwelling.getAvailableCreatures()).isEqualTo(0);
+            });
+        });
+    }
+
+    private GetAllDwellings getAllDwellings() {
+        System.out.println("GET ALL DWELLINGS: " + GAME_ID);
+        return GetAllDwellings.query(GAME_ID);
+    }
+
+    private GetAllDwellings.Result getDwellingReadModel(GetAllDwellings query) {
+        return queryGateway.query(query, GetAllDwellings.Result.class).join();
+    }
+}