feat(notifications): TODO 7 — draft decision toast + email + bell badge

2026-05-14 14:40:35 +00:00
parent 50d263745a
commit 9050e2427e
21 changed files with 1992 additions and 82 deletions
@@ -0,0 +1,375 @@
+package cloud.nstart.terravault.ordinis.notifications.dispatcher;
+
+import cloud.nstart.terravault.ordinis.domain.notification.NotificationLog;
+import cloud.nstart.terravault.ordinis.domain.notification.NotificationLogRepository;
+import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEvent;
+import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEventRepository;
+import cloud.nstart.terravault.ordinis.notifications.channel.ChannelKind;
+import cloud.nstart.terravault.ordinis.notifications.channel.NotificationChannel;
+import cloud.nstart.terravault.ordinis.notifications.channel.NotificationChannel.DispatchResult;
+import cloud.nstart.terravault.ordinis.notifications.channel.RenderedMessage;
+import cloud.nstart.terravault.ordinis.notifications.channel.UserRef;
+import cloud.nstart.terravault.ordinis.notifications.event.NotificationEvent;
+import cloud.nstart.terravault.ordinis.notifications.template.MessageRenderer;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
+import jakarta.annotation.PostConstruct;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.OffsetDateTime;
+import java.util.List;
+import java.util.Locale;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+
+/**
+ * Notifications dispatcher — polls outbox_events for draft lifecycle events and
+ * fans out email notifications to the relevant recipients.
+ *
+ * <p>Mirrors the pattern of {@code WebhookDispatcher} (two @Scheduled ticks):
+ * <ol>
+ *   <li><b>matchTick</b>: scans recent outbox_events for draft decision event
+ *       types not yet processed by this dispatcher.</li>
+ *   <li><b>sendTick</b>: for each matched event, resolves recipient(s), renders
+ *       via {@code MessageRenderer}, claims via {@code IdempotencyGuard}, dispatches
+ *       via enabled channels.</li>
+ * </ol>
+ *
+ * <p>Recipient resolution:
+ * <ul>
+ *   <li>{@code RecordDraftApproved / RecordDraftRejected / RecordDraftWithdrawn}:
+ *       notify the maker (maker_id from payload → look up email in UserDisplayService).</li>
+ *   <li>{@code RecordDraftSubmitted}: notify a configurable reviewer pool address
+ *       ({@code ordinis.notifications.reviewer-pool-email}). Skip silently if not set.</li>
+ * </ul>
+ *
+ * <p>Idempotency: per-recipient claim via {@code IdempotencyGuard} (INSERT ON CONFLICT DO NOTHING).
+ * The dispatcher reads events regardless of their {@code published_at} state — it does not require
+ * Kafka publication to have completed (same pattern as WebhookDispatcher).
+ *
+ * <p>Activated only when {@code ordinis.notifications.enabled=true}.
+ */
+@Component
+@ConditionalOnProperty(name = "ordinis.notifications.enabled", havingValue = "true", matchIfMissing = false)
+public class NotificationsDispatcher {
+
+  private static final Logger log = LoggerFactory.getLogger(NotificationsDispatcher.class);
+
+  private static final Set<String> DRAFT_EVENT_TYPES = Set.of(
+      "RecordDraftApproved",
+      "RecordDraftRejected",
+      "RecordDraftWithdrawn",
+      "RecordDraftSubmitted"
+  );
+
+  private final OutboxEventRepository outboxRepository;
+  private final NotificationLogRepository notificationLogRepository;
+  private final IdempotencyGuard idempotencyGuard;
+  private final MessageRenderer messageRenderer;
+  private final List<NotificationChannel> channels;
+  private final RecipientResolver recipientResolver;
+
+  private final Counter sentCounter;
+  private final Counter failedCounter;
+  private final Counter skippedCounter;
+
+  @Value("${ordinis.notifications.poll-interval-ms:30000}")
+  private long pollIntervalMs;
+
+  @Value("${ordinis.notifications.batch-size:50}")
+  private int batchSize;
+
+  @Value("${ordinis.notifications.reviewer-pool-email:}")
+  private String reviewerPoolEmail;
+
+  @Value("${ordinis.notifications.base-url:${ordinis.notifications.email.base-url:}}")
+  private String baseUrl;
+
+  public NotificationsDispatcher(
+      OutboxEventRepository outboxRepository,
+      NotificationLogRepository notificationLogRepository,
+      IdempotencyGuard idempotencyGuard,
+      MessageRenderer messageRenderer,
+      List<NotificationChannel> channels,
+      RecipientResolver recipientResolver,
+      MeterRegistry meterRegistry) {
+    this.outboxRepository = outboxRepository;
+    this.notificationLogRepository = notificationLogRepository;
+    this.idempotencyGuard = idempotencyGuard;
+    this.messageRenderer = messageRenderer;
+    this.channels = channels;
+    this.recipientResolver = recipientResolver;
+
+    this.sentCounter = Counter.builder("nsi_notification_sent_total")
+        .description("Total notifications successfully sent")
+        .register(meterRegistry);
+    this.failedCounter = Counter.builder("nsi_notification_failed_total")
+        .description("Total notification send failures")
+        .register(meterRegistry);
+    this.skippedCounter = Counter.builder("nsi_notification_skipped_total")
+        .description("Total notifications skipped (idempotency hit or no recipient)")
+        .register(meterRegistry);
+  }
+
+  @PostConstruct
+  void init() {
+    log.info("NotificationsDispatcher initialized. Channels: {}, reviewerPoolEmail configured: {}",
+        channels.stream().map(c -> c.kind().wireName()).toList(),
+        !reviewerPoolEmail.isBlank());
+  }
+
+  /**
+   * Scan recent outbox events for draft lifecycle types and process each.
+   * Runs every {@code ordinis.notifications.poll-interval-ms} ms.
+   */
+  @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();
+
+    if (drafts.isEmpty()) return;
+    log.debug("NotificationsDispatcher sendTick: {} draft events to process", drafts.size());
+
+    for (OutboxEvent event : drafts) {
+      processEvent(event);
+    }
+  }
+
+  /**
+   * Process a single outbox event: resolve recipient(s), render, claim, dispatch.
+   * Errors per event are caught so one bad event does not block others.
+   */
+  void processEvent(OutboxEvent event) {
+    try {
+      JsonNode payload = event.getPayload();
+      String eventType = event.getEventType();
+      UUID eventUuid = extractEventUuid(payload, event);
+
+      NotificationEvent notifEvent = buildNotificationEvent(event, eventUuid, payload);
+      List<UserRef> recipients = resolveRecipients(eventType, payload);
+
+      if (recipients.isEmpty()) {
+        log.debug("No recipients for event {} ({}). Skipping.", event.getId(), eventType);
+        skippedCounter.increment();
+        return;
+      }
+
+      for (UserRef recipient : recipients) {
+        dispatchToRecipient(notifEvent, recipient, eventType);
+      }
+    } catch (Exception ex) {
+      log.warn("Failed to process outbox event id={} type={}: {}",
+          event.getId(), event.getEventType(), ex.getMessage(), ex);
+      failedCounter.increment();
+    }
+  }
+
+  private void dispatchToRecipient(NotificationEvent notifEvent, UserRef recipient, String eventType) {
+    for (NotificationChannel channel : channels) {
+      if (!channel.isEnabled()) continue;
+
+      String recipientId = recipient.recipientFor(channel.kind());
+      if (recipientId == null || recipientId.isBlank()) {
+        skippedCounter.increment();
+        continue;
+      }
+
+      Optional<NotificationLog> claimed = idempotencyGuard.tryClaim(
+          notifEvent, channel.kind(), recipientId);
+
+      if (claimed.isEmpty()) {
+        skippedCounter.increment();
+        continue;
+      }
+
+      Long rowId = claimed.get().getId();
+      try {
+        String eventKey = toEventKey(eventType);
+        RenderedMessage msg = messageRenderer.render(
+            eventKey, channel.kind(), recipient.locale(),
+            buildParams(notifEvent, channel.kind()));
+
+        DispatchResult result = channel.dispatch(notifEvent, recipient, msg);
+
+        if (result.ok()) {
+          idempotencyGuard.markSent(rowId);
+          sentCounter.increment();
+          log.debug("Notification sent: event={} channel={} recipient={}",
+              notifEvent.eventId(), channel.kind().wireName(), recipientId);
+        } else {
+          idempotencyGuard.markFailed(rowId, result.errorMsg());
+          failedCounter.increment();
+          log.warn("Notification failed: event={} channel={} recipient={} error={}",
+              notifEvent.eventId(), channel.kind().wireName(), recipientId, result.errorMsg());
+        }
+      } catch (Exception ex) {
+        idempotencyGuard.markFailed(rowId, ex.getClass().getSimpleName() + ": " + ex.getMessage());
+        failedCounter.increment();
+        log.warn("Notification dispatch error: event={} channel={} recipient={}",
+            notifEvent.eventId(), channel.kind().wireName(), recipientId, ex);
+      }
+    }
+  }
+
+  /**
+   * Resolve notification recipients based on event type.
+   * <ul>
+   *   <li>Approved / Rejected: notify maker (draft decision sent back to maker).</li>
+   *   <li>Withdrawn: notify reviewer pool (queue cleanup ping).</li>
+   *   <li>Submitted: notify reviewer pool (new item in queue).</li>
+   * </ul>
+   */
+  private List<UserRef> resolveRecipients(String eventType, JsonNode payload) {
+    return switch (eventType) {
+      case "RecordDraftApproved", "RecordDraftRejected" -> resolveMaker(payload);
+      case "RecordDraftWithdrawn", "RecordDraftSubmitted" -> resolveReviewerPool();
+      default -> List.of();
+    };
+  }
+
+  private List<UserRef> resolveMaker(JsonNode payload) {
+    String makerId = stringField(payload, "makerId");
+    if (makerId == null) return List.of();
+    return recipientResolver.resolveById(makerId)
+        .map(List::of)
+        .orElse(List.of());
+  }
+
+  private List<UserRef> resolveReviewerPool() {
+    if (reviewerPoolEmail.isBlank()) return List.of();
+    UserRef poolRef = new UserRef("reviewer-pool", reviewerPoolEmail, null, null, Locale.forLanguageTag("ru"));
+    return List.of(poolRef);
+  }
+
+  /**
+   * Map outbox event type string to notification template event key.
+   */
+  static String toEventKey(String eventType) {
+    return switch (eventType) {
+      case "RecordDraftApproved", "RecordDraftRejected" -> "draft_decision";
+      case "RecordDraftWithdrawn" -> "draft_withdrawn";
+      case "RecordDraftSubmitted" -> "draft_submitted";
+      default -> throw new IllegalArgumentException("Unknown draft event type: " + eventType);
+    };
+  }
+
+  /**
+   * Build positional params array for {@code MessageRenderer.render()}.
+   * Order per notifications_ru.properties:
+   * {0}=dictDisplayName, {1}=businessKey, {2}=makerId, {3}=operation,
+   * {4}=baseUrl, {5}=draftId, {6}=decisionOutcome(0/1), {7}=reviewerId,
+   * {8}=commentPresent(0/1), {9}=comment
+   */
+  private Object[] buildParams(NotificationEvent event, ChannelKind channel) {
+    NotificationEvent.DraftRef draft = event.draft();
+    NotificationEvent.Decision decision = event.decision();
+
+    String dictDisplayName = draft != null ? draft.dictionary() : "";
+    String businessKey = draft != null ? draft.businessKey() : "";
+    String maker = draft != null ? draft.maker() : "";
+    String operation = draft != null ? draft.operation() : "";
+    long draftIdLong = extractDraftIdLong(draft);
+
+    int decisionOutcome = 0;
+    String reviewerId = "";
+    int commentPresent = 0;
+    String comment = "";
+
+    if (decision != null) {
+      decisionOutcome = "APPROVE".equals(decision.outcome()) ? 1 : 0;
+      reviewerId = decision.decidedBy() != null ? decision.decidedBy() : "";
+      comment = decision.comment() != null ? decision.comment() : "";
+      commentPresent = comment.isBlank() ? 0 : 1;
+    }
+
+    return new Object[]{
+        dictDisplayName,    // {0}
+        businessKey,        // {1}
+        maker,              // {2}
+        operation,          // {3}
+        baseUrl,            // {4}
+        draftIdLong,        // {5}
+        decisionOutcome,    // {6}
+        reviewerId,         // {7}
+        commentPresent,     // {8}
+        comment             // {9}
+    };
+  }
+
+  private long extractDraftIdLong(NotificationEvent.DraftRef draft) {
+    if (draft == null || draft.id() == null) return 0L;
+    // Draft ID is UUID; templates show it as a link fragment — use hashCode as numeric proxy
+    return Math.abs(draft.id().getMostSignificantBits());
+  }
+
+  private NotificationEvent buildNotificationEvent(OutboxEvent event, UUID eventUuid, JsonNode payload) {
+    String eventType = event.getEventType();
+    String makerId = stringField(payload, "makerId");
+    String reviewerId = stringField(payload, "reviewerId");
+    String reviewComment = stringField(payload, "reviewComment");
+    String draftIdStr = stringField(payload, "draftId");
+    String businessKey = stringField(payload, "businessKey");
+    String dictionaryName = stringField(payload, "dictionaryName");
+    String operation = stringField(payload, "operation");
+
+    UUID draftUuid = draftIdStr != null ? parseUuidOrNull(draftIdStr) : null;
+
+    NotificationEvent.DraftRef draftRef = new NotificationEvent.DraftRef(
+        draftUuid, businessKey, dictionaryName, operation, makerId);
+
+    NotificationEvent.Decision decision = null;
+    if ("RecordDraftApproved".equals(eventType) || "RecordDraftRejected".equals(eventType)) {
+      String outcome = "RecordDraftApproved".equals(eventType) ? "APPROVE" : "REJECT";
+      decision = new NotificationEvent.Decision(outcome, reviewerId, reviewComment);
+    }
+
+    return new NotificationEvent(
+        eventUuid,
+        eventType,
+        event.getCreatedAt(),
+        draftRef,
+        decision,
+        null);
+  }
+
+  private UUID extractEventUuid(JsonNode payload, OutboxEvent event) {
+    String uuidStr = stringField(payload, "eventId");
+    if (uuidStr != null) {
+      UUID parsed = parseUuidOrNull(uuidStr);
+      if (parsed != null) return parsed;
+    }
+    // Derive deterministic UUID from outbox event's numeric id if not present.
+    // event.getId() may be null in tests (entity not persisted). Fall back to random.
+    Long numericId = event.getId();
+    return numericId != null ? new UUID(0L, numericId) : UUID.randomUUID();
+  }
+
+  private static String stringField(JsonNode node, String field) {
+    if (node == null || !node.has(field)) return null;
+    JsonNode val = node.get(field);
+    return val.isNull() ? null : val.asText();
+  }
+
+  private static UUID parseUuidOrNull(String s) {
+    if (s == null || s.isBlank()) return null;
+    try {
+      return UUID.fromString(s);
+    } catch (IllegalArgumentException e) {
+      return null;
+    }
+  }
+}
@@ -0,0 +1,27 @@
+package cloud.nstart.terravault.ordinis.notifications.dispatcher;
+
+import cloud.nstart.terravault.ordinis.notifications.channel.UserRef;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+import java.util.Optional;
+
+/**
+ * Resolves a Keycloak subject (user UUID) to a {@link UserRef} for notification dispatch.
+ *
+ * <p>Backed by {@code UserDisplayService} (ordinis-rest-api module) injected via interface
+ * to keep ordinis-notifications free of a direct compile-time dependency on rest-api.
+ * In production the {@link UserDisplayServiceBridge} implementation is wired by the app module;
+ * in unit tests a mock can be provided directly.
+ */
+public interface RecipientResolver {
+
+  /**
+   * Resolve user id → UserRef with email + locale. Returns empty when the user is unknown
+   * (not in user_display_cache and Keycloak admin is not configured / unreachable).
+   *
+   * @param userId Keycloak subject (UUID string)
+   * @return populated UserRef or empty
+   */
+  Optional<UserRef> resolveById(String userId);
+}
@@ -0,0 +1,17 @@
+package cloud.nstart.terravault.ordinis.notifications.dispatcher;
+
+import java.util.Optional;
+
+/**
+ * Narrow anti-corruption port over {@code UserDisplayService} (ordinis-rest-api).
+ *
+ * <p>Keeps ordinis-notifications independent of ordinis-rest-api at compile time.
+ * The production implementation in ordinis-app delegates to {@code UserDisplayService}.
+ * Tests can mock this port with a simple lambda.
+ */
+public interface UserDisplayPort {
+
+  Optional<DisplayInfo> findById(String userId);
+
+  record DisplayInfo(String sub, String email, String name) {}
+}
@@ -0,0 +1,50 @@
+package cloud.nstart.terravault.ordinis.notifications.dispatcher;
+
+import cloud.nstart.terravault.ordinis.notifications.channel.UserRef;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+import java.util.Locale;
+import java.util.Optional;
+
+/**
+ * Production {@link RecipientResolver} backed by a {@link UserDisplayPort} —
+ * a narrow interface over {@code UserDisplayService} to avoid an explicit
+ * compile-time dependency from ordinis-notifications on ordinis-rest-api.
+ *
+ * <p>The port is implemented in the app module (ordinis-app) which has
+ * visibility over both modules. Tests can provide a mock port directly.
+ *
+ * <p>Locale: the user_display_cache table does not store locale; we default
+ * to {@code ru} matching the existing {@code MessageRenderer.DEFAULT_LOCALE}
+ * convention. When per-user locale is added, update this resolver.
+ */
+@Component
+@ConditionalOnProperty(name = "ordinis.notifications.enabled", havingValue = "true", matchIfMissing = false)
+public class UserDisplayRecipientResolver implements RecipientResolver {
+
+  private static final Logger log = LoggerFactory.getLogger(UserDisplayRecipientResolver.class);
+
+  private final UserDisplayPort userDisplayPort;
+
+  public UserDisplayRecipientResolver(UserDisplayPort userDisplayPort) {
+    this.userDisplayPort = userDisplayPort;
+  }
+
+  @Override
+  public Optional<UserRef> resolveById(String userId) {
+    if (userId == null || userId.isBlank()) return Optional.empty();
+
+    Optional<UserDisplayPort.DisplayInfo> info = userDisplayPort.findById(userId);
+    if (info.isEmpty()) {
+      log.debug("RecipientResolver: user {} not found in display cache", userId);
+      return Optional.empty();
+    }
+
+    UserDisplayPort.DisplayInfo d = info.get();
+    // Phase A: email only; expressHuid/telegramChatId resolved later (Phase B/C).
+    return Optional.of(new UserRef(userId, d.email(), null, null, Locale.forLanguageTag("ru")));
+  }
+}