feat(notifications): Phase A foundation — module + schema 0021

2026-05-10 15:41:34 +00:00
parent c675c5e310
commit 3b7b1bffe5
24 changed files with 1802 additions and 0 deletions
@@ -0,0 +1,161 @@
+package cloud.nstart.terravault.ordinis.domain.notification;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.EnumType;
+import jakarta.persistence.Enumerated;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+import jakarta.persistence.Index;
+import jakarta.persistence.Table;
+import jakarta.persistence.UniqueConstraint;
+
+import java.time.OffsetDateTime;
+import java.util.UUID;
+
+/**
+ * Per-recipient notification delivery audit row.
+ *
+ * <p>UNIQUE {@code (event_id, channel, recipient)} — primary idempotency guard
+ * (см. migration 0021 + design doc D2). Insert pattern в
+ * {@code IdempotencyGuard.shouldSend()}:
+ *
+ * <pre>{@code
+ * INSERT INTO notification_log (event_id, channel, recipient, status, sent_at)
+ * VALUES (?, ?, ?, 'PENDING', NOW())
+ * ON CONFLICT (event_id, channel, recipient) DO NOTHING
+ * RETURNING id;
+ * }</pre>
+ *
+ * <p>Если RETURNING пустой — другой consumer уже claim'ил отправку (или sweep
+ * job не освободил). Else — мы owner, dispatch + UPDATE status final.
+ *
+ * <p>Status enum UPPERCASE (Hibernate {@code @Enumerated(EnumType.STRING)}
+ * lesson из v1.1.0 P3 — migration 0020 fix). DB CHECK constraint matches
+ * Java enum names exactly.
+ */
+@Entity
+@Table(
+    name = "notification_log",
+    uniqueConstraints = {
+        // Mirrors notification_log_idempotency UNIQUE constraint в migration 0021.
+        @UniqueConstraint(
+            name = "notification_log_idempotency",
+            columnNames = {"event_id", "channel", "recipient"}
+        )
+    },
+    indexes = {
+        @Index(name = "idx_notif_log_draft", columnList = "draft_id"),
+        @Index(name = "idx_notif_log_recipient_time", columnList = "recipient,sent_at DESC"),
+        @Index(name = "idx_notif_log_event", columnList = "event_id")
+    }
+)
+public class NotificationLog {
+
+  /** Lifecycle status. UPPERCASE — matches CHECK constraint в 0021. */
+  public enum Status {
+    /** Inserted but not yet dispatched (или crashed mid-dispatch). Pending sweep job
+     *  cleans up stuck rows старше {@code notifications.pendingSweep.olderThanMs}. */
+    PENDING,
+    /** Successfully sent через channel. Idempotency lock — не replay'ится. */
+    SENT,
+    /** Channel returned non-retryable error (auth, malformed, etc). */
+    FAILED,
+    /** Rate limiter denied — recipient hit per-hour cap. */
+    RATE_LIMITED,
+    /** User opted out этого channel в user_notification_prefs. */
+    SKIPPED_PREF,
+    /** Event старше maxEventAgeHours — backlog burst guard fired. */
+    SKIPPED_TOO_OLD,
+    /** One-click unsubscribe — токен HMAC validated, prefs.enabled=false. */
+    UNSUBSCRIBED
+  }
+
+  /** Wire format channel — lowercase strings ('email', 'express', 'telegram')
+   *  matches CHECK constraint и Helm config keys. NOT EnumType.STRING на channel
+   *  потому что хотим lowercase (брендовые токены), не Java UPPERCASE conv. */
+  @Id
+  @GeneratedValue(strategy = GenerationType.IDENTITY)
+  @Column(name = "id")
+  private Long id;
+
+  @Column(name = "event_id", nullable = false)
+  private UUID eventId;
+
+  @Column(name = "event_type", nullable = false, length = 64)
+  private String eventType;
+
+  /** Nullable: некоторые events без draft контекста (future expansion). */
+  @Column(name = "draft_id")
+  private UUID draftId;
+
+  /** Lower-case wire format. См. {@code ChannelKind.wireName()} в notifications module. */
+  @Column(name = "channel", nullable = false, length = 16)
+  private String channel;
+
+  /** Email / express chat_id (UUID string) / telegram chat_id (numeric string). */
+  @Column(name = "recipient", nullable = false, length = 256)
+  private String recipient;
+
+  @Enumerated(EnumType.STRING)
+  @Column(name = "status", nullable = false, length = 20)
+  private Status status;
+
+  @Column(name = "sent_at", nullable = false)
+  private OffsetDateTime sentAt;
+
+  @Column(name = "error_msg", columnDefinition = "TEXT")
+  private String errorMsg;
+
+  protected NotificationLog() {}
+
+  /** Constructs row in PENDING state — typical entry point для IdempotencyGuard. */
+  public NotificationLog(UUID eventId, String eventType, UUID draftId,
+                         String channel, String recipient) {
+    this.eventId = eventId;
+    this.eventType = eventType;
+    this.draftId = draftId;
+    this.channel = channel;
+    this.recipient = recipient;
+    this.status = Status.PENDING;
+    this.sentAt = OffsetDateTime.now();
+  }
+
+  public void markSent() {
+    this.status = Status.SENT;
+    this.errorMsg = null;
+  }
+
+  public void markFailed(String error) {
+    this.status = Status.FAILED;
+    this.errorMsg = error;
+  }
+
+  public void markRateLimited() {
+    this.status = Status.RATE_LIMITED;
+  }
+
+  public void markSkippedPref() {
+    this.status = Status.SKIPPED_PREF;
+  }
+
+  public void markSkippedTooOld() {
+    this.status = Status.SKIPPED_TOO_OLD;
+  }
+
+  public void markUnsubscribed() {
+    this.status = Status.UNSUBSCRIBED;
+  }
+
+  // Getters (no setters — entity мутирует через named transitions)
+  public Long getId() { return id; }
+  public UUID getEventId() { return eventId; }
+  public String getEventType() { return eventType; }
+  public UUID getDraftId() { return draftId; }
+  public String getChannel() { return channel; }
+  public String getRecipient() { return recipient; }
+  public Status getStatus() { return status; }
+  public OffsetDateTime getSentAt() { return sentAt; }
+  public String getErrorMsg() { return errorMsg; }
+}
@@ -0,0 +1,83 @@
+package cloud.nstart.terravault.ordinis.domain.notification;
+
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+
+import java.time.OffsetDateTime;
+import java.util.Optional;
+import java.util.UUID;
+
+public interface NotificationLogRepository extends JpaRepository<NotificationLog, Long> {
+
+  /**
+   * Idempotency claim. Native query потому что Spring Data JPA не поддерживает
+   * {@code INSERT ... ON CONFLICT DO NOTHING RETURNING} напрямую (PostgreSQL specific).
+   *
+   * <p>Returns inserted row's id если claim succeeded, или empty если конфликт
+   * (другой consumer уже claim'ил, или sweep job не освободил stuck row).
+   *
+   * <p>Используется в {@code IdempotencyGuard.shouldSend()}.
+   */
+  @Modifying
+  @Query(value = """
+      INSERT INTO notification_log (event_id, event_type, draft_id, channel, recipient, status, sent_at)
+      VALUES (:eventId, :eventType, :draftId, :channel, :recipient, 'PENDING', NOW())
+      ON CONFLICT (event_id, channel, recipient) DO NOTHING
+      RETURNING id
+      """, nativeQuery = true)
+  Optional<Long> tryClaim(
+      @Param("eventId") UUID eventId,
+      @Param("eventType") String eventType,
+      @Param("draftId") UUID draftId,
+      @Param("channel") String channel,
+      @Param("recipient") String recipient);
+
+  /**
+   * Pending sweep — освобождает stuck PENDING rows старше {@code threshold}.
+   * Returns deleted count для metrics. Использует partial index
+   * {@code idx_notif_log_pending_sweep}.
+   */
+  @Modifying
+  @Query(value = """
+      DELETE FROM notification_log
+      WHERE status = 'PENDING' AND sent_at < :threshold
+      """, nativeQuery = true)
+  int deletePendingOlderThan(@Param("threshold") OffsetDateTime threshold);
+
+  /**
+   * Retention sweep — удаляет terminal status rows старше {@code threshold}.
+   * FAILED rows preserved bessrochno (для post-mortem). Partial index
+   * {@code idx_notif_log_retention}.
+   */
+  @Modifying
+  @Query(value = """
+      DELETE FROM notification_log
+      WHERE status IN ('SENT','RATE_LIMITED','SKIPPED_PREF','SKIPPED_TOO_OLD','UNSUBSCRIBED')
+        AND sent_at < :threshold
+      """, nativeQuery = true)
+  int deleteRetainedOlderThan(@Param("threshold") OffsetDateTime threshold);
+
+  /**
+   * UI "Last sent" lookup для {@code GET /api/v1/me/notifications}.
+   * Cache 60s в service layer.
+   */
+  @Query("""
+      SELECT MAX(n.sentAt) FROM NotificationLog n
+      WHERE n.recipient = :recipient
+        AND n.channel = :channel
+        AND n.status = cloud.nstart.terravault.ordinis.domain.notification.NotificationLog.Status.SENT
+      """)
+  Optional<OffsetDateTime> findLastSentAt(
+      @Param("recipient") String recipient,
+      @Param("channel") String channel);
+
+  /** Admin endpoint pagination. */
+  Page<NotificationLog> findByRecipientOrderBySentAtDesc(String recipient, Pageable pageable);
+
+  /** Admin endpoint — full audit lookup by event. */
+  Page<NotificationLog> findByEventIdOrderBySentAtDesc(UUID eventId, Pageable pageable);
+}
@@ -0,0 +1,78 @@
+package cloud.nstart.terravault.ordinis.domain.notification;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Embeddable;
+import jakarta.persistence.EmbeddedId;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Table;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * Per-user channel opt-in/out (см. migration 0021 + design doc D3).
+ *
+ * <p>Composite PK {@code (user_id, channel)}. Default {@code enabled=false} для
+ * новых rows; bootstrap loadData может включить email для existing reviewer/maker
+ * ролей при первом deploy.
+ *
+ * <p>Channel — lowercase wire format ('email' / 'express' / 'telegram').
+ */
+@Entity
+@Table(name = "user_notification_prefs")
+public class UserNotificationPref {
+
+  @Embeddable
+  public static class Pk implements Serializable {
+
+    @Column(name = "user_id", length = 128, nullable = false)
+    private String userId;
+
+    @Column(name = "channel", length = 16, nullable = false)
+    private String channel;
+
+    protected Pk() {}
+
+    public Pk(String userId, String channel) {
+      this.userId = userId;
+      this.channel = channel;
+    }
+
+    public String getUserId() { return userId; }
+    public String getChannel() { return channel; }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (!(o instanceof Pk pk)) return false;
+      return Objects.equals(userId, pk.userId) && Objects.equals(channel, pk.channel);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(userId, channel);
+    }
+  }
+
+  @EmbeddedId
+  private Pk id;
+
+  @Column(name = "enabled", nullable = false)
+  private boolean enabled;
+
+  protected UserNotificationPref() {}
+
+  public UserNotificationPref(String userId, String channel, boolean enabled) {
+    this.id = new Pk(userId, channel);
+    this.enabled = enabled;
+  }
+
+  public Pk getId() { return id; }
+  public String getUserId() { return id.getUserId(); }
+  public String getChannel() { return id.getChannel(); }
+  public boolean isEnabled() { return enabled; }
+
+  public void setEnabled(boolean enabled) {
+    this.enabled = enabled;
+  }
+}
@@ -0,0 +1,16 @@
+package cloud.nstart.terravault.ordinis.domain.notification;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+
+import java.util.List;
+import java.util.Optional;
+
+public interface UserNotificationPrefRepository
+    extends JpaRepository<UserNotificationPref, UserNotificationPref.Pk> {
+
+  /** Все каналы пользователя — для GET /api/v1/me/notifications page. */
+  List<UserNotificationPref> findByIdUserId(String userId);
+
+  /** Single channel lookup для dispatcher (skip if disabled). */
+  Optional<UserNotificationPref> findByIdUserIdAndIdChannel(String userId, String channel);
+}