feat: add TTL-based retention for notifications and read states

openframe-data-mongo-common/src/main/java/com/openframe/data/document/notification/Notification.java

@@ -6,6 +6,7 @@
 import lombok.NoArgsConstructor;
 import org.springframework.data.annotation.CreatedDate;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.index.Indexed;
 import org.springframework.data.mongodb.core.mapping.Document;
 
 import java.time.Instant;
@@ -30,5 +31,8 @@ public class Notification {
     @CreatedDate
     private Instant createdAt;
 
+    @Indexed(name = "notifications_ttl", expireAfterSeconds = 0)
+    private Instant expireAt;
+
     private NotificationContext context;
 }

openframe-data-mongo-common/src/main/java/com/openframe/data/document/notification/NotificationReadState.java

@@ -7,6 +7,7 @@
 import org.springframework.data.annotation.Id;
 import org.springframework.data.mongodb.core.index.CompoundIndex;
 import org.springframework.data.mongodb.core.index.CompoundIndexes;
+import org.springframework.data.mongodb.core.index.Indexed;
 import org.springframework.data.mongodb.core.mapping.Document;
 
 import java.time.Instant;
@@ -46,4 +47,7 @@ public class NotificationReadState {
     private String title;
 
     private Instant readAt;
+
+    @Indexed(name = "read_states_ttl", expireAfterSeconds = 0)
+    private Instant expireAt;
 }

openframe-data-mongo-sync/src/main/java/com/openframe/data/service/notification/NotificationReadStateService.java

@@ -11,9 +11,12 @@
 import jakarta.validation.constraints.NotNull;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.validation.annotation.Validated;
 
+import java.time.Duration;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.EnumMap;
@@ -28,11 +31,15 @@ public class NotificationReadStateService {
 
     private final NotificationReadStateRepository repository;
 
+    @Value("${openframe.features.notifications.retention-days:30}")
+    private long retentionDays;
+
     public void createForAudience(@NotBlank String notificationId,
                                   @NotNull NotificationCategory category,
                                   String title,
                                   @NotNull RecipientType recipientType,
                                   @NotEmpty Collection<String> recipientIds) {
+        Instant expireAt = Instant.now().plus(Duration.ofDays(retentionDays));
         List<NotificationReadState> rows = new ArrayList<>(recipientIds.size());
         for (String recipientId : recipientIds) {
             rows.add(NotificationReadState.builder()
@@ -42,6 +49,7 @@ public void createForAudience(@NotBlank String notificationId,
                     .status(ReadStatus.UNREAD)
                     .category(category)
                     .title(title)
+                    .expireAt(expireAt)
                     .build());
         }
         repository.bulkInsertUnordered(rows);

openframe-data-mongo-sync/src/test/java/com/openframe/data/integration/service/notification/NotificationTtlIndexIT.java

@@ -0,0 +1,96 @@
+package com.openframe.data.integration.service.notification;
+
+import com.openframe.data.document.notification.Notification;
+import com.openframe.data.document.notification.NotificationCategory;
+import com.openframe.data.document.notification.NotificationReadState;
+import com.openframe.data.document.notification.RecipientType;
+import com.openframe.data.integration.BaseMongoIntegrationTest;
+import com.openframe.data.integration.support.IntegrationTestApplication;
+import com.openframe.data.service.notification.NotificationReadStateService;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.index.IndexInfo;
+import org.springframework.data.mongodb.core.index.MongoPersistentEntityIndexResolver;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Query;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest(classes = IntegrationTestApplication.class)
+@Tag("integration")
+@EnabledIfSystemProperty(named = "integration.tests", matches = "true")
+class NotificationTtlIndexIT extends BaseMongoIntegrationTest {
+
+    private static final String TTL_NOTIFICATIONS = "notifications_ttl";
+    private static final String TTL_READ_STATES = "read_states_ttl";
+
+    @Autowired
+    private NotificationReadStateService service;
+
+    @Autowired
+    private MongoTemplate mongoTemplate;
+
+    @BeforeEach
+    void resetCollections() {
+        mongoTemplate.dropCollection(Notification.class);
+        mongoTemplate.dropCollection(NotificationReadState.class);
+        ensureIndexes(Notification.class);
+        ensureIndexes(NotificationReadState.class);
+    }
+
+    @Test
+    @DisplayName("Given the notifications collection, when indexes are resolved, then a TTL index on expireAt with expireAfterSeconds=0 exists")
+    void notifications_collection_has_ttl_index() {
+        Optional<IndexInfo> ttl = findIndex(Notification.class, TTL_NOTIFICATIONS);
+        assertThat(ttl).isPresent();
+        assertThat(ttl.get().getExpireAfter()).contains(Duration.ZERO);
+    }
+
+    @Test
+    @DisplayName("Given the notification_read_states collection, when indexes are resolved, then a TTL index on expireAt with expireAfterSeconds=0 exists")
+    void read_states_collection_has_ttl_index() {
+        Optional<IndexInfo> ttl = findIndex(NotificationReadState.class, TTL_READ_STATES);
+        assertThat(ttl).isPresent();
+        assertThat(ttl.get().getExpireAfter()).contains(Duration.ZERO);
+    }
+
+    @Test
+    @DisplayName("Given the default retention, when createForAudience persists a row, then expireAt is stamped roughly 30 days in the future")
+    void create_for_audience_stamps_expire_at() {
+        Instant before = Instant.now();
+        service.createForAudience("notif-ttl", NotificationCategory.TICKETS, "title",
+                RecipientType.USER, Set.of("user-ttl"));
+
+        NotificationReadState row = mongoTemplate.findOne(
+                new Query(Criteria.where("recipientId").is("user-ttl")), NotificationReadState.class);
+
+        assertThat(row).isNotNull();
+        assertThat(row.getExpireAt())
+                .isAfter(before.plus(Duration.ofDays(29)))
+                .isBefore(before.plus(Duration.ofDays(31)));
+    }
+
+    private Optional<IndexInfo> findIndex(Class<?> entityClass, String name) {
+        List<IndexInfo> indexes = mongoTemplate.indexOps(entityClass).getIndexInfo();
+        return indexes.stream().filter(idx -> name.equals(idx.getName())).findFirst();
+    }
+
+    private void ensureIndexes(Class<?> entityClass) {
+        var indexOps = mongoTemplate.indexOps(entityClass);
+        new MongoPersistentEntityIndexResolver(mongoTemplate.getConverter().getMappingContext())
+                .resolveIndexFor(entityClass)
+                .forEach(indexOps::ensureIndex);
+    }
+}

openframe-data-nats/src/main/java/com/openframe/data/nats/service/NotificationBroadcaster.java

@@ -12,6 +12,8 @@
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
+import java.time.Duration;
+import java.time.Instant;
 import java.util.Optional;
 import java.util.Set;
 
@@ -28,6 +30,9 @@ public class NotificationBroadcaster {
     @Value("${openframe.features.notifications.enabled:false}")
     private boolean notificationsEnabled;
 
+    @Value("${openframe.features.notifications.retention-days:30}")
+    private long retentionDays;
+
     public Notification broadcast(NotificationCommand command) {
         if (!notificationsEnabled) {
             log.debug("Notifications feature disabled — broadcast skipped (no persistence, no NATS publish)");
@@ -38,6 +43,7 @@ public Notification broadcast(NotificationCommand command) {
                 .severity(command.getSeverity())
                 .title(command.getTitle())
                 .description(command.getDescription())
+                .expireAt(Instant.now().plus(Duration.ofDays(retentionDays)))
                 .context(command.getContext())
                 .build();
         Notification saved = notificationRepository.save(notification);

openframe-data-nats/src/test/java/com/openframe/data/nats/service/NotificationBroadcasterTest.java

@@ -15,6 +15,8 @@
 import org.mockito.ArgumentCaptor;
 import org.springframework.test.util.ReflectionTestUtils;
 
+import java.time.Duration;
+import java.time.Instant;
 import java.util.Optional;
 import java.util.Set;
 
@@ -161,6 +163,26 @@ void persisted_notification_carries_command_fields() {
         assertThat(persisted.getContext().getType()).isEqualTo("BULK_APPROVAL");
     }
 
+    @Test
+    @DisplayName("Given the configured retention, when broadcast persists the Notification, then expireAt is stamped roughly retention-days into the future — the TTL index reaps the document after retention elapses")
+    void persisted_notification_is_stamped_with_ttl_expiry() {
+        NotificationCommand cmd = NotificationCommand.builder()
+                .title("Approval")
+                .severity(NotificationSeverity.INFO)
+                .context(genericContext("APPROVAL"))
+                .adminAudience(Set.of("admin-1"))
+                .build();
+
+        Instant before = Instant.now();
+        broadcaster.broadcast(cmd);
+
+        ArgumentCaptor<Notification> captor = ArgumentCaptor.forClass(Notification.class);
+        verify(notificationRepository).save(captor.capture());
+        assertThat(captor.getValue().getExpireAt())
+                .isAfter(before.plus(Duration.ofDays(29)))
+                .isBefore(before.plus(Duration.ofDays(31)));
+    }
+
     @Test
     @DisplayName("Given a command, when broadcast returns, then the returned Notification is the saved one (with id populated) — caller must rely on the persisted id, not on a builder-only object")
     void broadcast_returns_persisted_notification() {
@@ -254,6 +276,7 @@ private NotificationBroadcaster newBroadcaster(Optional<NotificationNatsPublishe
         NotificationBroadcaster bc = new NotificationBroadcaster(
                 notificationRepository, readStateService, descriptorRegistry, publisher);
         ReflectionTestUtils.setField(bc, "notificationsEnabled", notificationsEnabled);
+        ReflectionTestUtils.setField(bc, "retentionDays", 30L);
         return bc;
     }