feat: add kafka to award-service (#7)

Author: scottgerring

.gitignore

@@ -1,3 +1,4 @@
-
+.cursor
+**/.cursor
 **/.claude/settings.local.json
 .cursor

sticker-award/pom.xml

@@ -68,16 +68,40 @@
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-rest-jackson</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-messaging-kafka</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-junit5</artifactId>
             <scope>test</scope>
         </dependency>
+        
         <dependency>
             <groupId>io.rest-assured</groupId>
             <artifactId>rest-assured</artifactId>
             <scope>test</scope>
         </dependency>
+        
+        <!-- Kafka Testing Dependencies -->
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-junit5-mockito</artifactId>
+            <scope>test</scope>
+        </dependency>
+        
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <scope>test</scope>
+        </dependency>
+        
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

sticker-award/src/main/java/com/datadoghq/stickerlandia/stickeraward/StickerAwardResource.java

@@ -10,6 +10,7 @@
 import com.datadoghq.stickerlandia.stickeraward.beans.UserStickersResponseApiResponse;
 import com.datadoghq.stickerlandia.stickeraward.entity.Sticker;
 import com.datadoghq.stickerlandia.stickeraward.entity.StickerAssignment;
+import com.datadoghq.stickerlandia.stickeraward.messaging.StickerEventPublisher;
 import io.smallrye.common.constraint.NotNull;
 import jakarta.inject.Inject;
 import jakarta.transaction.Transactional;
@@ -22,6 +23,8 @@
 import jakarta.ws.rs.Produces;
 import jakarta.ws.rs.core.Response;
 import org.eclipse.microprofile.openapi.annotations.Operation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.time.Instant;
 import java.util.ArrayList;
@@ -31,6 +34,11 @@
 
 @Path("/api")
 public class StickerAwardResource {
+    
+    private static final Logger log = LoggerFactory.getLogger(StickerAwardResource.class);
+    
+    @Inject
+    StickerEventPublisher eventPublisher;
 
     @Operation(description = "Get stickers assigned to a user (access controlled based on caller identity)")
     @Path("/award/v1/users/{userId}/stickers")
@@ -100,6 +108,16 @@ public Response assignStickerToUser(@PathParam("userId") String userId,
         // Create new assignment
         StickerAssignment assignment = new StickerAssignment(userId, sticker, data.getReason());
         assignment.persist();
+        
+        // Publish events to Kafka
+        try {
+            log.info("Publishing sticker assignment events for userId={}, stickerId={}", userId, stickerId);
+            eventPublisher.publishStickerAssigned(assignment);
+        } catch (Exception e) {
+            log.error("Failed to publish sticker assignment events", e);
+            // Continue with the response even if the event publishing fails
+            // In a production system, you might want to implement a retry mechanism
+        }
 
         // Create response
         StickerAssignmentResponse response = new StickerAssignmentResponse();
@@ -139,6 +157,16 @@ public Response removeStickerFromUser(@PathParam("userId") String userId,
         // Mark as removed
         assignment.setRemovedAt(Instant.now());
         assignment.persist();
+        
+        // Publish events to Kafka
+        try {
+            log.info("Publishing sticker removal event for userId={}, stickerId={}", userId, stickerId);
+            eventPublisher.publishStickerRemoved(assignment);
+        } catch (Exception e) {
+            log.error("Failed to publish sticker removal event", e);
+            // Continue with the response even if the event publishing fails
+            // In a production system, you might want to implement a retry mechanism
+        }
 
         // Create response
         StickerRemovalResponse response = new StickerRemovalResponse();

sticker-award/src/main/java/com/datadoghq/stickerlandia/stickeraward/events/CertificationCompletedEvent.java

@@ -0,0 +1,38 @@
+package com.datadoghq.stickerlandia.stickeraward.events;
+
+import java.time.Instant;
+
+/**
+ * Event received when a certification is completed.
+ * Subscribed from the 'certifications.certificationCompleted.v1' topic.
+ */
+public class CertificationCompletedEvent {
+    
+    private String accountId;
+    private String certificationId;
+    private Instant completedAt;
+    
+    public String getAccountId() {
+        return accountId;
+    }
+    
+    public void setAccountId(String accountId) {
+        this.accountId = accountId;
+    }
+    
+    public String getCertificationId() {
+        return certificationId;
+    }
+    
+    public void setCertificationId(String certificationId) {
+        this.certificationId = certificationId;
+    }
+    
+    public Instant getCompletedAt() {
+        return completedAt;
+    }
+    
+    public void setCompletedAt(Instant completedAt) {
+        this.completedAt = completedAt;
+    }
+}

sticker-award/src/main/java/com/datadoghq/stickerlandia/stickeraward/events/DomainEvent.java

@@ -0,0 +1,32 @@
+package com.datadoghq.stickerlandia.stickeraward.events;
+
+/**
+ * Base abstract class for all domain events.
+ * This class should be extended by all event types.
+ */
+public abstract class DomainEvent {
+    
+    private String eventName;
+    private String eventVersion;
+    
+    protected DomainEvent(String eventName, String eventVersion) {
+        this.eventName = eventName;
+        this.eventVersion = eventVersion;
+    }
+    
+    public String getEventName() {
+        return eventName;
+    }
+    
+    public void setEventName(String eventName) {
+        this.eventName = eventName;
+    }
+    
+    public String getEventVersion() {
+        return eventVersion;
+    }
+    
+    public void setEventVersion(String eventVersion) {
+        this.eventVersion = eventVersion;
+    }
+}

sticker-award/src/main/java/com/datadoghq/stickerlandia/stickeraward/events/StickerAssignedToUserEvent.java

@@ -0,0 +1,64 @@
+package com.datadoghq.stickerlandia.stickeraward.events;
+
+import java.time.Instant;
+
+import com.datadoghq.stickerlandia.stickeraward.entity.StickerAssignment;
+
+/**
+ * Event published when a sticker is assigned to a user.
+ * Published to the 'stickers.stickerAssignedToUser.v1' topic.
+ */
+public class StickerAssignedToUserEvent extends DomainEvent {
+    
+    private static final String EVENT_NAME = "StickerAssignedToUser";
+    private static final String EVENT_VERSION = "v1";
+    
+    private String accountId;
+    private String stickerId;
+    private Instant assignedAt;
+    
+    /**
+     * Default constructor for serialization frameworks
+     */
+    public StickerAssignedToUserEvent() {
+        super(EVENT_NAME, EVENT_VERSION);
+    }
+    
+    /**
+     * Create a new event from a sticker assignment entity
+     * 
+     * @param assignment The sticker assignment entity
+     * @return A new event instance
+     */
+    public static StickerAssignedToUserEvent fromAssignment(StickerAssignment assignment) {
+        StickerAssignedToUserEvent event = new StickerAssignedToUserEvent();
+        event.setAccountId(assignment.getUserId());
+        event.setStickerId(assignment.getSticker().getStickerId());
+        event.setAssignedAt(assignment.getAssignedAt());
+        return event;
+    }
+    
+    public String getAccountId() {
+        return accountId;
+    }
+    
+    public void setAccountId(String accountId) {
+        this.accountId = accountId;
+    }
+    
+    public String getStickerId() {
+        return stickerId;
+    }
+    
+    public void setStickerId(String stickerId) {
+        this.stickerId = stickerId;
+    }
+    
+    public Instant getAssignedAt() {
+        return assignedAt;
+    }
+    
+    public void setAssignedAt(Instant assignedAt) {
+        this.assignedAt = assignedAt;
+    }
+}

sticker-award/src/main/java/com/datadoghq/stickerlandia/stickeraward/events/StickerClaimedEvent.java

@@ -0,0 +1,52 @@
+package com.datadoghq.stickerlandia.stickeraward.events;
+
+import com.datadoghq.stickerlandia.stickeraward.entity.StickerAssignment;
+
+/**
+ * Event published when a user claims a sticker.
+ * Published to the 'users.stickerClaimed.v1' topic.
+ */
+public class StickerClaimedEvent extends DomainEvent {
+    
+    private static final String EVENT_NAME = "StickerClaimed";
+    private static final String EVENT_VERSION = "v1";
+    
+    private String accountId;
+    private String stickerId;
+    
+    /**
+     * Default constructor for serialization frameworks
+     */
+    public StickerClaimedEvent() {
+        super(EVENT_NAME, EVENT_VERSION);
+    }
+    
+    /**
+     * Create a new event from a sticker assignment entity
+     * 
+     * @param assignment The sticker assignment entity
+     * @return A new event instance
+     */
+    public static StickerClaimedEvent fromAssignment(StickerAssignment assignment) {
+        StickerClaimedEvent event = new StickerClaimedEvent();
+        event.setAccountId(assignment.getUserId());
+        event.setStickerId(assignment.getSticker().getStickerId());
+        return event;
+    }
+    
+    public String getAccountId() {
+        return accountId;
+    }
+    
+    public void setAccountId(String accountId) {
+        this.accountId = accountId;
+    }
+    
+    public String getStickerId() {
+        return stickerId;
+    }
+    
+    public void setStickerId(String stickerId) {
+        this.stickerId = stickerId;
+    }
+}

sticker-award/src/main/java/com/datadoghq/stickerlandia/stickeraward/events/StickerRemovedFromUserEvent.java

@@ -0,0 +1,68 @@
+package com.datadoghq.stickerlandia.stickeraward.events;
+
+import java.time.Instant;
+
+import com.datadoghq.stickerlandia.stickeraward.entity.StickerAssignment;
+
+/**
+ * Event published when a sticker is removed from a user.
+ * Published to the 'stickers.stickerRemovedFromUser.v1' topic.
+ */
+public class StickerRemovedFromUserEvent extends DomainEvent {
+    
+    private static final String EVENT_NAME = "StickerRemovedFromUser";
+    private static final String EVENT_VERSION = "v1";
+    
+    private String accountId;
+    private String stickerId;
+    private Instant removedAt;
+    
+    /**
+     * Default constructor for serialization frameworks
+     */
+    public StickerRemovedFromUserEvent() {
+        super(EVENT_NAME, EVENT_VERSION);
+    }
+    
+    /**
+     * Create a new event from a sticker assignment entity
+     * 
+     * @param assignment The sticker assignment entity with removed status
+     * @return A new event instance
+     */
+    public static StickerRemovedFromUserEvent fromAssignment(StickerAssignment assignment) {
+        if (assignment.getRemovedAt() == null) {
+            throw new IllegalArgumentException("Cannot create removal event from active assignment");
+        }
+        
+        StickerRemovedFromUserEvent event = new StickerRemovedFromUserEvent();
+        event.setAccountId(assignment.getUserId());
+        event.setStickerId(assignment.getSticker().getStickerId());
+        event.setRemovedAt(assignment.getRemovedAt());
+        return event;
+    }
+    
+    public String getAccountId() {
+        return accountId;
+    }
+    
+    public void setAccountId(String accountId) {
+        this.accountId = accountId;
+    }
+    
+    public String getStickerId() {
+        return stickerId;
+    }
+    
+    public void setStickerId(String stickerId) {
+        this.stickerId = stickerId;
+    }
+    
+    public Instant getRemovedAt() {
+        return removedAt;
+    }
+    
+    public void setRemovedAt(Instant removedAt) {
+        this.removedAt = removedAt;
+    }
+}

sticker-award/src/main/java/com/datadoghq/stickerlandia/stickeraward/messaging/CertificationCompletedEventDeserializer.java

@@ -0,0 +1,14 @@
+package com.datadoghq.stickerlandia.stickeraward.messaging;
+
+import io.quarkus.kafka.client.serialization.ObjectMapperDeserializer;
+import com.datadoghq.stickerlandia.stickeraward.events.CertificationCompletedEvent;
+
+/**
+ * Deserializer for the CertificationCompletedEvent from Kafka.
+ */
+public class CertificationCompletedEventDeserializer extends ObjectMapperDeserializer<CertificationCompletedEvent> {
+    
+    public CertificationCompletedEventDeserializer() {
+        super(CertificationCompletedEvent.class);
+    }
+}