diff --git a/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/outbox/OutboxEventRepository.java b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/outbox/OutboxEventRepository.java index 7211f74..18b46cd 100644 --- a/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/outbox/OutboxEventRepository.java +++ b/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 { @@ -20,6 +22,31 @@ public interface OutboxEventRepository extends JpaRepository """) List 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}. + * + *

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 findRecentByEventTypes( + Collection types, OffsetDateTime since, Pageable pageable); + /** * Сколько ожидает публикации (без DLQ). Метрика * {@code ordinis_outbox_pending_events}. diff --git a/ordinis-notifications/src/main/java/cloud/nstart/terravault/ordinis/notifications/dispatcher/NotificationsDispatcher.java b/ordinis-notifications/src/main/java/cloud/nstart/terravault/ordinis/notifications/dispatcher/NotificationsDispatcher.java index 0af927c..6784017 100644 --- a/ordinis-notifications/src/main/java/cloud/nstart/terravault/ordinis/notifications/dispatcher/NotificationsDispatcher.java +++ b/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. + * + *

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 events = outboxRepository.findUnpublished(PageRequest.of(0, batchSize)); - if (events.isEmpty()) return; - - List 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 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()); diff --git a/ordinis-notifications/src/test/java/cloud/nstart/terravault/ordinis/notifications/dispatcher/NotificationsDispatcherTest.java b/ordinis-notifications/src/test/java/cloud/nstart/terravault/ordinis/notifications/dispatcher/NotificationsDispatcherTest.java index 3786cb7..dfc8ee8 100644 --- a/ordinis-notifications/src/test/java/cloud/nstart/terravault/ordinis/notifications/dispatcher/NotificationsDispatcherTest.java +++ b/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());