feat(stats): update user stats via video event SQS handlers

src/main/java/com/accountabilityatlas/userservice/event/VideoStatusChangedEvent.java

@@ -0,0 +1,13 @@
+package com.accountabilityatlas.userservice.event;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.UUID;
+
+public record VideoStatusChangedEvent(
+    UUID videoId,
+    UUID submittedBy,
+    List<UUID> locationIds,
+    String previousStatus,
+    String newStatus,
+    Instant timestamp) {}

src/main/java/com/accountabilityatlas/userservice/event/VideoStatusChangedHandler.java

@@ -0,0 +1,35 @@
+package com.accountabilityatlas.userservice.event;
+
+import com.accountabilityatlas.userservice.service.UserStatsService;
+import io.awspring.cloud.sqs.annotation.SqsListener;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class VideoStatusChangedHandler {
+
+  private final UserStatsService userStatsService;
+
+  @SqsListener("${app.sqs.user-video-status-events-queue:user-video-status-events}")
+  public void handleVideoStatusChanged(VideoStatusChangedEvent event) {
+    log.info(
+        "Received VideoStatusChanged event for video {} ({} -> {})",
+        event.videoId(),
+        event.previousStatus(),
+        event.newStatus());
+    try {
+      userStatsService.handleStatusChange(
+          event.submittedBy(), event.previousStatus(), event.newStatus());
+    } catch (Exception e) {
+      log.error(
+          "Failed to handle VideoStatusChanged event for video {}: {}",
+          event.videoId(),
+          e.getMessage(),
+          e);
+      throw e;
+    }
+  }
+}

src/main/java/com/accountabilityatlas/userservice/event/VideoSubmittedEvent.java

@@ -0,0 +1,15 @@
+package com.accountabilityatlas.userservice.event;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+
+public record VideoSubmittedEvent(
+    UUID videoId,
+    UUID submitterId,
+    String submitterTrustTier,
+    String title,
+    Set<String> amendments,
+    List<UUID> locationIds,
+    Instant timestamp) {}

src/main/java/com/accountabilityatlas/userservice/event/VideoSubmittedHandler.java

@@ -0,0 +1,33 @@
+package com.accountabilityatlas.userservice.event;
+
+import com.accountabilityatlas.userservice.service.UserStatsService;
+import io.awspring.cloud.sqs.annotation.SqsListener;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class VideoSubmittedHandler {
+
+  private final UserStatsService userStatsService;
+
+  @SqsListener("${app.sqs.user-video-events-queue:user-video-events}")
+  public void handleVideoSubmitted(VideoSubmittedEvent event) {
+    log.info(
+        "Received VideoSubmitted event for video {} from submitter {}",
+        event.videoId(),
+        event.submitterId());
+    try {
+      userStatsService.incrementSubmissionCount(event.submitterId());
+    } catch (Exception e) {
+      log.error(
+          "Failed to handle VideoSubmitted event for video {}: {}",
+          event.videoId(),
+          e.getMessage(),
+          e);
+      throw e;
+    }
+  }
+}

src/main/java/com/accountabilityatlas/userservice/service/UserStatsService.java

@@ -0,0 +1,62 @@
+package com.accountabilityatlas.userservice.service;
+
+import com.accountabilityatlas.userservice.repository.UserStatsRepository;
+import java.time.Instant;
+import java.util.UUID;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class UserStatsService {
+
+  private final UserStatsRepository userStatsRepository;
+
+  @Transactional
+  public void incrementSubmissionCount(UUID userId) {
+    userStatsRepository
+        .findById(userId)
+        .ifPresentOrElse(
+            stats -> {
+              stats.setSubmissionCount(stats.getSubmissionCount() + 1);
+              stats.setUpdatedAt(Instant.now());
+              userStatsRepository.save(stats);
+              log.debug("Incremented submission count for user {}", userId);
+            },
+            () -> log.warn("UserStats not found for user {}, skipping submission count", userId));
+  }
+
+  @Transactional
+  public void handleStatusChange(UUID userId, String previousStatus, String newStatus) {
+    userStatsRepository
+        .findById(userId)
+        .ifPresentOrElse(
+            stats -> {
+              if ("APPROVED".equals(newStatus)) {
+                stats.setApprovedCount(stats.getApprovedCount() + 1);
+              } else if ("APPROVED".equals(previousStatus)) {
+                stats.setApprovedCount(Math.max(0, stats.getApprovedCount() - 1));
+              }
+
+              if ("REJECTED".equals(newStatus)) {
+                stats.setRejectedCount(stats.getRejectedCount() + 1);
+              } else if ("REJECTED".equals(previousStatus)) {
+                stats.setRejectedCount(Math.max(0, stats.getRejectedCount() - 1));
+              }
+
+              stats.setUpdatedAt(Instant.now());
+              userStatsRepository.save(stats);
+              log.debug(
+                  "Updated stats for user {} ({} -> {}): approved={}, rejected={}",
+                  userId,
+                  previousStatus,
+                  newStatus,
+                  stats.getApprovedCount(),
+                  stats.getRejectedCount());
+            },
+            () -> log.warn("UserStats not found for user {}, skipping status change", userId));
+  }
+}

src/main/resources/application.yml

@@ -47,3 +47,5 @@ logging:
 app:
   sqs:
     user-events-queue: user-events
+    user-video-events-queue: user-video-events
+    user-video-status-events-queue: user-video-status-events

src/main/resources/db/devdata/R__dev_seed_users.sql

@@ -22,6 +22,6 @@ INSERT INTO users.user_stats (user_id, submission_count, approved_count, rejecte
 VALUES
     ('00000000-0000-0000-0000-000000000001', 0, 0, 0),
     ('00000000-0000-0000-0000-000000000002', 0, 0, 0),
-    ('00000000-0000-0000-0000-000000000003', 10, 8, 2),
+    ('00000000-0000-0000-0000-000000000003', 10, 10, 0),
     ('00000000-0000-0000-0000-000000000004', 0, 0, 0)
 ON CONFLICT (user_id) DO NOTHING;

src/test/java/com/accountabilityatlas/userservice/event/VideoStatusChangedHandlerTest.java

@@ -0,0 +1,55 @@
+package com.accountabilityatlas.userservice.event;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.verify;
+
+import com.accountabilityatlas.userservice.service.UserStatsService;
+import java.time.Instant;
+import java.util.List;
+import java.util.UUID;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class VideoStatusChangedHandlerTest {
+
+  @Mock private UserStatsService userStatsService;
+  @InjectMocks private VideoStatusChangedHandler videoStatusChangedHandler;
+
+  @Test
+  void handleVideoStatusChanged_delegatesToUserStatsService() {
+    UUID submittedBy = UUID.randomUUID();
+    VideoStatusChangedEvent event =
+        new VideoStatusChangedEvent(
+            UUID.randomUUID(),
+            submittedBy,
+            List.of(UUID.randomUUID()),
+            "PENDING",
+            "APPROVED",
+            Instant.now());
+
+    videoStatusChangedHandler.handleVideoStatusChanged(event);
+
+    verify(userStatsService).handleStatusChange(submittedBy, "PENDING", "APPROVED");
+  }
+
+  @Test
+  void handleVideoStatusChanged_serviceFailure_rethrowsException() {
+    UUID submittedBy = UUID.randomUUID();
+    VideoStatusChangedEvent event =
+        new VideoStatusChangedEvent(
+            UUID.randomUUID(), submittedBy, List.of(), "APPROVED", "REJECTED", Instant.now());
+
+    RuntimeException exception = new RuntimeException("DB error");
+    doThrow(exception)
+        .when(userStatsService)
+        .handleStatusChange(submittedBy, "APPROVED", "REJECTED");
+
+    assertThatThrownBy(() -> videoStatusChangedHandler.handleVideoStatusChanged(event))
+        .isSameAs(exception);
+  }
+}

src/test/java/com/accountabilityatlas/userservice/event/VideoSubmittedHandlerTest.java

@@ -0,0 +1,54 @@
+package com.accountabilityatlas.userservice.event;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.verify;
+
+import com.accountabilityatlas.userservice.service.UserStatsService;
+import java.time.Instant;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class VideoSubmittedHandlerTest {
+
+  @Mock private UserStatsService userStatsService;
+  @InjectMocks private VideoSubmittedHandler videoSubmittedHandler;
+
+  @Test
+  void handleVideoSubmitted_delegatesToUserStatsService() {
+    UUID submitterId = UUID.randomUUID();
+    VideoSubmittedEvent event =
+        new VideoSubmittedEvent(
+            UUID.randomUUID(),
+            submitterId,
+            "NEW",
+            "Test Video",
+            Set.of("FIRST"),
+            List.of(UUID.randomUUID()),
+            Instant.now());
+
+    videoSubmittedHandler.handleVideoSubmitted(event);
+
+    verify(userStatsService).incrementSubmissionCount(submitterId);
+  }
+
+  @Test
+  void handleVideoSubmitted_serviceFailure_rethrowsException() {
+    UUID submitterId = UUID.randomUUID();
+    VideoSubmittedEvent event =
+        new VideoSubmittedEvent(
+            UUID.randomUUID(), submitterId, "TRUSTED", "Test", Set.of(), List.of(), Instant.now());
+
+    RuntimeException exception = new RuntimeException("DB error");
+    doThrow(exception).when(userStatsService).incrementSubmissionCount(submitterId);
+
+    assertThatThrownBy(() -> videoSubmittedHandler.handleVideoSubmitted(event)).isSameAs(exception);
+  }
+}

src/test/java/com/accountabilityatlas/userservice/integration/AuthIntegrationTest.java

@@ -23,6 +23,7 @@
 import org.testcontainers.containers.PostgreSQLContainer;
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
+import software.amazon.awssdk.services.sqs.SqsAsyncClient;
 
 @SpringBootTest
 @AutoConfigureMockMvc
@@ -45,15 +46,18 @@ static void configureProperties(DynamicPropertyRegistry registry) {
     registry.add("spring.flyway.url", postgres::getJdbcUrl);
     registry.add("spring.flyway.user", postgres::getUsername);
     registry.add("spring.flyway.password", postgres::getPassword);
-    // Exclude Redis auto-configuration since no Redis container is provided
+    // Exclude Redis and SQS auto-configuration since no Redis/LocalStack containers are provided
     registry.add(
         "spring.autoconfigure.exclude",
-        () -> "org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration");
+        () ->
+            "org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration,"
+                + "io.awspring.cloud.autoconfigure.sqs.SqsAutoConfiguration");
   }
 
   @Autowired private MockMvc mockMvc;
 
   @MockitoBean private EventPublisher eventPublisher;
+  @MockitoBean private SqsAsyncClient sqsAsyncClient;
 
   @Test
   void registerThenLogin_fullFlow() throws Exception {