fix(notifications): dispatcher race с OutboxPoller — emails never sent

ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/outbox/OutboxEventRepository.java

@@ -5,6 +5,8 @@ import org.springframework.data.domain.Pageable;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.Query;
 
+import java.time.OffsetDateTime;
+import java.util.Collection;
 import java.util.List;
 
 public interface OutboxEventRepository extends JpaRepository<OutboxEvent, Long> {
@@ -20,6 +22,31 @@ public interface OutboxEventRepository extends JpaRepository<OutboxEvent, Long>
       """)
   List<OutboxEvent> findUnpublished(Pageable pageable);
 
+  /**
+   * Recent events filtered by event_type, regardless of {@code published_at}.
+   * Used by NotificationsDispatcher (and future side-channel processors) что
+   * RACE с OutboxPoller (Kafka publisher) — Poller polls every 1s и flips
+   * {@code published_at} быстрее чем 30s notifications tick → notifications
+   * никогда не находил event'ы с {@link #findUnpublished}.
+   *
+   * <p>Idempotency обеспечивается через own dedupe table (notification_log
+   * UNIQUE (event_id, channel, recipient)) — same event может быть processed
+   * многократно без duplicate sends.
+   *
+   * @param types event_type filter (e.g. RecordDraftSubmitted, ...)
+   * @param since createdAt cutoff — обычно last 24h чтобы не сканировать
+   *              историческую таблицу при cold start
+   */
+  @Query("""
+      SELECT o FROM OutboxEvent o
+      WHERE o.eventType IN :types
+        AND o.createdAt >= :since
+        AND o.dlqAt IS NULL
+      ORDER BY o.id ASC
+      """)
+  List<OutboxEvent> findRecentByEventTypes(
+      Collection<String> types, OffsetDateTime since, Pageable pageable);
+
   /**
    * Сколько ожидает публикации (без DLQ). Метрика
    * {@code ordinis_outbox_pending_events}.

ordinis-notifications/src/main/java/cloud/nstart/terravault/ordinis/notifications/dispatcher/NotificationsDispatcher.java

@@ -148,16 +148,22 @@ public class NotificationsDispatcher {
   /**
    * Scan recent outbox events for draft lifecycle types and process each.
    * Runs every {@code ordinis.notifications.poll-interval-ms} ms.
+   *
+   * <p>Uses {@link OutboxEventRepository#findRecentByEventTypes} (event_type
+   * + createdAt window, NOT published_at). Раньше использовался
+   * {@code findUnpublished} что race'илось с OutboxPoller (1s tick) — Poller
+   * setting {@code published_at} быстрее чем notifications 30s tick → events
+   * никогда не reached dispatcher. Now: dispatcher polls all recent draft
+   * events независимо от Kafka publish state, idempotency через
+   * notification_log UNIQUE (event_id, channel, recipient).
    */
   @Scheduled(fixedDelayString = "${ordinis.notifications.poll-interval-ms:30000}")
   @Transactional
   public void sendTick() {
-    List<OutboxEvent> events = outboxRepository.findUnpublished(PageRequest.of(0, batchSize));
-    if (events.isEmpty()) return;
-
-    List<OutboxEvent> drafts = events.stream()
-        .filter(e -> DRAFT_EVENT_TYPES.contains(e.getEventType()))
-        .toList();
+    // 24h window — covers max retention scenarios без full table scan.
+    OffsetDateTime since = OffsetDateTime.now().minusHours(24);
+    List<OutboxEvent> drafts = outboxRepository.findRecentByEventTypes(
+        DRAFT_EVENT_TYPES, since, PageRequest.of(0, batchSize));
 
     if (drafts.isEmpty()) return;
     log.debug("NotificationsDispatcher sendTick: {} draft events to process", drafts.size());

ordinis-notifications/src/test/java/cloud/nstart/terravault/ordinis/notifications/dispatcher/NotificationsDispatcherTest.java

@@ -108,7 +108,7 @@ class NotificationsDispatcherTest {
     String makerEmail = "maker@nstart.space";
 
     OutboxEvent event = buildDraftApprovedEvent(draftId, makerId);
-    when(outboxRepository.findUnpublished(any(Pageable.class))).thenReturn(List.of(event));
+    when(outboxRepository.findRecentByEventTypes(any(), any(), any(Pageable.class))).thenReturn(List.of(event));
 
     UserRef makerRef = new UserRef(makerId, makerEmail, null, null, Locale.forLanguageTag("ru"));
     when(recipientResolver.resolveById(makerId)).thenReturn(Optional.of(makerRef));
@@ -140,7 +140,7 @@ class NotificationsDispatcherTest {
         new ObjectMapper().createObjectNode(),
         "ordinis.cuod.events.public", "spacecraft:test");
 
-    when(outboxRepository.findUnpublished(any(Pageable.class))).thenReturn(List.of(event));
+    when(outboxRepository.findRecentByEventTypes(any(), any(), any(Pageable.class))).thenReturn(List.of(event));
 
     dispatcher.sendTick();
 
@@ -154,7 +154,7 @@ class NotificationsDispatcherTest {
     UUID draftId = UUID.randomUUID();
     String makerId = "maker-sub-456";
     OutboxEvent event = buildDraftApprovedEvent(draftId, makerId);
-    when(outboxRepository.findUnpublished(any(Pageable.class))).thenReturn(List.of(event));
+    when(outboxRepository.findRecentByEventTypes(any(), any(), any(Pageable.class))).thenReturn(List.of(event));
 
     UserRef makerRef = new UserRef(makerId, "maker@nstart.space", null, null, Locale.forLanguageTag("ru"));
     when(recipientResolver.resolveById(makerId)).thenReturn(Optional.of(makerRef));
@@ -182,7 +182,7 @@ class NotificationsDispatcherTest {
   void unknownMaker_noDispatch() {
     UUID draftId = UUID.randomUUID();
     OutboxEvent event = buildDraftApprovedEvent(draftId, "unknown-sub");
-    when(outboxRepository.findUnpublished(any(Pageable.class))).thenReturn(List.of(event));
+    when(outboxRepository.findRecentByEventTypes(any(), any(), any(Pageable.class))).thenReturn(List.of(event));
 
     when(recipientResolver.resolveById("unknown-sub")).thenReturn(Optional.empty());