chore(streak): create streak on habit creation

services/streak/build.gradle.kts

@@ -89,6 +89,7 @@ dependencies {
   testImplementation("org.testcontainers:junit-jupiter")
   testImplementation("org.testcontainers:kafka")
   testImplementation("org.testcontainers:postgresql")
+  intTestImplementation("org.springframework.kafka:spring-kafka-test")
   intTestImplementation("org.springframework.modulith:spring-modulith-starter-test")
   intTestImplementation("org.springframework.boot:spring-boot-starter-test")
   intTestImplementation("org.springframework.security:spring-security-test")

services/streak/src/intTest/kotlin/de/codecentric/streak/HabitConsumerIntegrationTest.kt

@@ -0,0 +1,62 @@
+package de.codecentric.streak
+
+import de.codecentric.streak.domain.Habit.Frequency.WEEKLY
+import de.codecentric.streak.domain.StreakRepository
+import io.kotest.assertions.asClue
+import io.kotest.matchers.shouldBe
+import org.junit.jupiter.api.Test
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration
+import org.springframework.boot.test.context.SpringBootTest
+import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.kafka.test.context.EmbeddedKafka
+import org.springframework.test.annotation.DirtiesContext
+import org.springframework.test.context.ActiveProfiles
+import org.testcontainers.shaded.org.awaitility.Awaitility
+import java.util.Optional
+import java.util.UUID
+import java.util.concurrent.TimeUnit.SECONDS
+
+
+@ActiveProfiles("intTest")
+@SpringBootTest(classes = [HabitConsumer::class, StreakRepository::class, DataJdbcConfiguration::class])
+@EnableAutoConfiguration
+@DirtiesContext
+@EmbeddedKafka(topics = ["habit-events", "habit-deleted"])
+class HabitConsumerIntegrationTest {
+
+  @Autowired
+  private lateinit var habitCreatedKafkaTemplate: KafkaTemplate<String, HabitCreated>
+
+  @Autowired
+  private lateinit var habitDeletedKafkaTemplate: KafkaTemplate<String, HabitDeleted>
+
+  @Autowired
+  private lateinit var streakRepository: StreakRepository
+
+  @Test
+  fun createsAndSavesStreakWhenHabitWasCreated() {
+    val id = UUID.randomUUID()
+    habitCreatedKafkaTemplate.send("habit-events", "key", HabitCreated(id, WEEKLY, 3))
+    habitCreatedKafkaTemplate.flush()
+
+    Awaitility.await().atMost(1, SECONDS).untilAsserted {
+      val result = streakRepository.findByHabitId(id)
+      result.asClue {
+        result.isPresent shouldBe true
+        result.get().habit.id shouldBe id
+      }
+    }
+  }
+
+  @Test
+  fun deletesStreakWhenHabitWasDeleted() {
+    val id = UUID.randomUUID()
+    habitDeletedKafkaTemplate.send("habit-deleted", "key", HabitDeleted(id))
+
+    Awaitility.await().atMost(1, SECONDS).untilAsserted {
+      val result = streakRepository.findByHabitId(id)
+      result shouldBe Optional.empty()
+    }
+  }
+}

services/streak/src/intTest/kotlin/de/codecentric/streak/domain/StreakRepositoryIntegrationTest.kt

@@ -23,4 +23,12 @@ class StreakRepositoryIntegrationTest(@Autowired val subject: StreakRepository)
     subject.save(streak)
     subject.findById(streak.id)
   }
+
+  @Test
+  fun `loads streak from database by habit id`() {
+    val id = UUID.randomUUID()
+    val streak = Streak.from(Habit(id, Schedule(WEEKLY, 3)))
+    subject.save(streak)
+    subject.findByHabitId(id);
+  }
 }

services/streak/src/intTest/resources/application-intTest.yaml

@@ -2,3 +2,8 @@ spring:
   datasource:
     driver-class-name: org.testcontainers.jdbc.ContainerDatabaseDriver
     url: jdbc:tc:postgresql:16://localhost/postgres
+
+  kafka:
+    client-id: streak
+    properties:
+      "security.protocol": PLAINTEXT

services/streak/src/main/kotlin/de/codecentric/streak/HabitConsumer.kt

@@ -0,0 +1,25 @@
+package de.codecentric.streak
+
+import de.codecentric.streak.domain.Habit
+import de.codecentric.streak.domain.Habit.Frequency
+import de.codecentric.streak.domain.Streak
+import de.codecentric.streak.domain.StreakRepository
+import org.springframework.kafka.annotation.KafkaHandler
+import org.springframework.kafka.annotation.KafkaListener
+import org.springframework.stereotype.Component
+import java.util.UUID
+
+@Component
+@KafkaListener(topics = ["habit-events"])
+class HabitConsumer(private val repository: StreakRepository) {
+
+  @KafkaHandler
+  fun consumeHabitCreatedEvent(event: HabitCreated) {
+    val streak = Streak.from(Habit.from(event.habitId, event.frequency, event.repetitions))
+    repository.save(streak)
+  }
+}
+
+data class HabitCreated(val habitId: UUID, val frequency: Frequency, val repetitions: Int)
+
+data class HabitDeleted(val habitId: UUID)

services/streak/src/main/kotlin/de/codecentric/streak/domain/Streak.kt

@@ -20,6 +20,11 @@ data class Habit(
   @Embedded(onEmpty = Embedded.OnEmpty.USE_EMPTY) val schedule: Schedule,
   @Value("null") @Transient private val new: Boolean = false
 ) : Persistable<UUID> {
+  companion object {
+    fun from(id: UUID, frequency: Frequency, repetitions: Int): Habit =
+      Habit(id, Schedule(frequency, repetitions), true)
+  }
+
   override fun getId() = id
   override fun isNew() = new
 

services/streak/src/main/kotlin/de/codecentric/streak/domain/StreakRepository.kt

@@ -1,7 +1,19 @@
 package de.codecentric.streak.domain
 
+import org.springframework.data.jdbc.repository.query.Query
 import org.springframework.data.repository.CrudRepository
+import java.util.Optional
 import java.util.UUID
 
 interface StreakRepository : CrudRepository<Streak, UUID> {
+
+  @Query(
+    """
+    SELECT s.id, h.id habit_id, h.frequency habit_frequency, h.repetitions habit_repetitions
+      FROM hc_streak.streaks s
+      INNER JOIN hc_streak.habits h
+        ON s.id = h.streak WHERE h.id = :id
+    """
+  )
+  fun findByHabitId(id: UUID): Optional<Streak>
 }

services/streak/src/main/resources/application.yaml

@@ -21,6 +21,16 @@ spring:
       "security.protocol": SASL_PLAINTEXT
       "sasl.mechanism": PLAIN
       "sasl.jaas.config": 'org.apache.kafka.common.security.plain.PlainLoginModule required username="${KAFKA_USER:habit}" password="${KAFKA_PASSWORD:habit}";'
+    producer:
+      properties:
+        "spring.json.type.mapping": "habit-created:de.codecentric.streak.HabitCreated"
+      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
+    consumer:
+      properties:
+        "spring.json.type.mapping": "habit-created:de.codecentric.streak.HabitCreated"
+        "auto.offset.reset": earliest
+      group-id: streak
+      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
 
 server:
   port: 9005