scriptoriumadmin/skillhub
1
0
Fork 0
mirror of https://github.com/iflytek/skillhub.git synced 2026-10-06 02:48:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Merge pull request #807 from iflytek/codex/refactor/issue-622-notification-polling

Browse source 
refactor(notification): replace SSE with HTTP polling
This commit is contained in:
XiaoSeS 2026-09-03 13:48:57 +08:00 • committed by GitHub
commit d6afc43364
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
36 changed files with 398 additions and 1343 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
docs/skillhub/en/guide/social.md
View file

@ -74,6 +74,11 @@ View skill packages with the most stars and highest ratings to discover best pra
3. Click a notification to navigate to the relevant page
4. Mark as read or mark all as read
The notification list and unread count refresh through ordinary HTTP requests every 10 seconds and
immediately when the window regains focus. The legacy `GET /api/v1/notifications/sse` endpoint has
been removed. Custom clients should poll `GET /api/v1/notifications` and
`GET /api/v1/notifications/unread-count` instead.
**Viewing My Stars**:
1. Navigate to `/dashboard/stars`

4
docs/skillhub/guide/social.md
View file

@ -74,6 +74,10 @@ SkillHub 提供了丰富的社交功能，让团队成员可以互动、分享
3. 点击通知跳转到相关页面
4. 标记为已读或全部已读
通知列表和未读数每 10 秒通过普通 HTTP 请求刷新，并在窗口重新获得焦点时立即刷新。旧版
`GET /api/v1/notifications/sse` 接口已移除；自建客户端应改为轮询
`GET /api/v1/notifications` 和 `GET /api/v1/notifications/unread-count`。
**查看我的收藏**：
1. 访问 `/dashboard/stars`

48
docs/superpowers/specs/2026-03-19-notification-system-design.md
View file

@ -2,11 +2,11 @@
## Goal
Build an independent in-app notification subsystem for SkillHub that delivers real-time notifications for skill lifecycle events (publish, review, promotion, report), with SSE push, user preference control, and extensibility for future third-party channels.
Build an independent in-app notification subsystem for SkillHub that delivers near-real-time notifications for skill lifecycle events (publish, review, promotion, report), with HTTP polling, user preference control, and extensibility for future third-party channels.
## Scope
- **In scope**: In-app notifications, SSE real-time push, notification preferences (category × channel), bell icon + dropdown + notification page, data cleanup
- **In scope**: In-app notifications, 10-second HTTP polling, notification preferences (category × channel), bell icon + dropdown + notification page, data cleanup
- **Out of scope**: External channels (email, Feishu, DingTalk), migration of existing governance notifications, external webhook delivery
## Architecture
@ -22,13 +22,11 @@ Domain Events (existing + new)
└── NotificationModule (NEW)
├── NotificationEventListener
├── RecipientResolver
├── NotificationPreferenceService (filter)
├── NotificationDispatcher (channel routing)
├── NotificationService (persist)
└── SseEmitterManager (push)
├── NotificationPreferenceService (preference CRUD)
└── NotificationService (preference filter + persist + HTTP reads)
```
The notification module consumes domain events via `@TransactionalEventListener(phase = AFTER_COMMIT)` + `@Async("skillhubEventExecutor")`, following the same pattern as existing listeners. The async executor pool (max 4 threads) is sufficient for the added load since notification processing is lightweight (DB insert + SSE push).
The notification module consumes domain events via `@TransactionalEventListener(phase = AFTER_COMMIT)` + `@Async("skillhubEventExecutor")`, following the same pattern as existing listeners. The async executor pool (max 4 threads) is sufficient for the added load since notification processing is a lightweight database insert. Clients discover persisted changes through HTTP polling.
## Data Model
@ -163,30 +161,24 @@ skillhub-notification/ -- new module (depends on: skillhub-do
│ ├── NotificationPreference.java
│ ├── NotificationRepository.java
│ └── NotificationPreferenceRepository.java
├── service/
│ ├── NotificationService.java -- CRUD: create, list, mark read, batch read, unread count
│ ├── NotificationPreferenceService.java -- preference CRUD + default fallback
│ └── NotificationDispatcher.java -- route by channel (currently IN_APP only)
└── sse/
└── SseEmitterManager.java -- manage SSE connections: register, push, heartbeat, cleanup
└── service/
├── NotificationService.java -- apply preference, create, list, mark read, batch read, unread count
└── NotificationPreferenceService.java -- preference CRUD + default fallback
skillhub-app/ -- existing module
└── listener/
├── NotificationEventListener.java -- consume domain events, call RecipientResolver + Dispatcher
├── NotificationEventListener.java -- consume domain events, call RecipientResolver + NotificationService
└── RecipientResolver.java -- resolve recipient list per event type (needs auth + domain repos)
```
## SSE Real-Time Push
## HTTP Polling
- Endpoint: `GET /api/notifications/sse`
- `SseEmitterManager` uses `ConcurrentHashMap<String, CopyOnWriteArrayList<SseEmitter>>` (thread-safe for concurrent tab open/close)
- Per-user connection limit: max 5 emitters (reject new connections beyond limit)
- Global connection limit: max 1000 emitters (configurable, reject with 503 when exceeded)
- SseEmitter timeout: 60s, browser `EventSource` auto-reconnects
- Heartbeat: `:ping` every 30s to prevent proxy/LB disconnection
- On emitter complete/timeout/error: auto-remove from map
- Push failure: silent ignore (notification already persisted, visible on refresh)
- On `EventSource` reconnect: frontend fetches unread count to sync badge
- The global bell polls `GET /api/notifications/unread-count` every 10 seconds while a user is authenticated.
- An active dropdown or notification page polls its paginated `GET /api/notifications` query every 10 seconds.
- Polling pauses while the browser tab is in the background.
- Window focus and network reconnect trigger a fresh request.
- Poll responses are authoritative; the unread badge uses the server count instead of incrementing a client-side event counter.
- Closing the dropdown unmounts its list query, so the full notification list is not polled when it is not visible.
## API Design
@ -195,8 +187,6 @@ GET /api/notifications -- List (paginated + category filter)
GET /api/notifications/unread-count -- Unread count (for bell badge)
PUT /api/notifications/{id}/read -- Mark single as read
PUT /api/notifications/read-all -- Mark all as read
GET /api/notifications/sse -- SSE connection
GET /api/notification-preferences -- Get current user preferences
PUT /api/notification-preferences -- Batch update preferences
```
@ -208,10 +198,12 @@ Response format follows existing SkillHub API conventions (code + data wrapper).
### Bell Component (global nav bar)
- Bell icon in nav bar, left of user avatar
- Red badge with unread count (> 99 shows "99+")
- Polls the unread count over HTTP every 10 seconds while the tab is visible
- Click to expand dropdown
### Dropdown List
- Shows latest 5 notifications
- Polls the visible list over HTTP every 10 seconds
- Each item: title + relative time ("3 minutes ago")
- Click item → navigate to entity page + mark as read
- Footer: "View all notifications" link
@ -219,6 +211,7 @@ Response format follows existing SkillHub API conventions (code + data wrapper).
### Notification Page (`/dashboard/notifications`)
- Full notification list with pagination
- Polls the visible page over HTTP every 10 seconds
- Tab filter by category: All / Publish / Review / Promotion / Report
- Batch mark all as read
- Click to navigate
@ -242,8 +235,6 @@ Response format follows existing SkillHub API conventions (code + data wrapper).
```yaml
skillhub:
notification:
sse-timeout: 60s
sse-heartbeat: 30s
cleanup:
read-retention-days: 30
unread-retention-days: 90
@ -260,6 +251,5 @@ skillhub:
## Extensibility
- New event types: add domain event record + mapping in `NotificationEventListener`
- New channels: add enum value to `NotificationChannel` + implement channel-specific dispatcher
- Third-party integrations: add new `@TransactionalEventListener` beans that consume the same domain events
- Preference table already supports category × channel granularity, no schema change needed

9
server/skillhub-app/src/main/java/com/iflytek/skillhub/config/WebMvcRateLimitConfig.java
View file

@ -1,9 +1,7 @@
package com.iflytek.skillhub.config;
import com.iflytek.skillhub.notification.sse.SseEmitterManager;
import com.iflytek.skillhub.ratelimit.RateLimitInterceptor;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.servlet.config.annotation.AsyncSupportConfigurer;
import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
@ -24,11 +22,4 @@ public class WebMvcRateLimitConfig implements WebMvcConfigurer {
registry.addInterceptor(rateLimitInterceptor)
.addPathPatterns("/api/**");
}
@Override
public void configureAsyncSupport(AsyncSupportConfigurer configurer) {
// Keep MVC async timeouts above the SSE emitter timeout so EventSource
// connections are not forcibly torn down every few seconds.
configurer.setDefaultTimeout(SseEmitterManager.defaultTimeoutMillis());
}
}

11
server/skillhub-app/src/main/java/com/iflytek/skillhub/controller/portal/NotificationController.java
View file

@ -8,7 +8,6 @@ import com.iflytek.skillhub.dto.*;
import com.iflytek.skillhub.notification.domain.Notification;
import com.iflytek.skillhub.notification.domain.NotificationCategory;
import com.iflytek.skillhub.notification.service.NotificationService;
import com.iflytek.skillhub.notification.sse.SseEmitterManager;
import java.util.Collections;
import java.util.Map;
import jakarta.validation.constraints.Max;
@ -16,10 +15,8 @@ import jakarta.validation.constraints.Min;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.http.MediaType;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
@RestController
@Validated
@ -27,16 +24,13 @@ import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
public class NotificationController extends BaseApiController {
private final NotificationService notificationService;
private final SseEmitterManager sseEmitterManager;
private final ObjectMapper objectMapper;
public NotificationController(NotificationService notificationService,
SseEmitterManager sseEmitterManager,
ObjectMapper objectMapper,
ApiResponseFactory responseFactory) {
super(responseFactory);
this.notificationService = notificationService;
this.sseEmitterManager = sseEmitterManager;
this.objectMapper = objectMapper;
}
@ -79,11 +73,6 @@ public class NotificationController extends BaseApiController {
return ok("response.success.deleted", null);
}
@GetMapping(value = "/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter sse(@RequestAttribute("userId") String userId) {
return sseEmitterManager.register(userId);
}
private NotificationResponse toResponse(Notification n) {
NotificationTarget target = resolveTarget(n);
return new NotificationResponse(

6
server/skillhub-app/src/main/java/com/iflytek/skillhub/exception/GlobalExceptionHandler.java
View file

@ -187,12 +187,6 @@ public class GlobalExceptionHandler {
@ExceptionHandler(AsyncRequestTimeoutException.class)
public ResponseEntity<?> handleAsyncRequestTimeout(AsyncRequestTimeoutException ex, HttpServletRequest request) {
String path = request.getRequestURI();
if (path != null && path.endsWith("/sse")) {
logger.debug("SSE timeout [requestId={}, path={}]", requestIdAccessor.current(), path);
return ResponseEntity.noContent().build();
}
logHandledException(HttpStatus.REQUEST_TIMEOUT, "error.request.timeout", request);
return ResponseEntity.status(HttpStatus.REQUEST_TIMEOUT).body(
apiResponseFactory.error(408, "error.request.timeout"));

26
server/skillhub-app/src/main/java/com/iflytek/skillhub/filter/RequestLoggingFilter.java
View file

@ -8,8 +8,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.filter.OncePerRequestFilter;
import org.springframework.web.util.ContentCachingRequestWrapper;
@ -29,20 +27,11 @@ public class RequestLoggingFilter extends OncePerRequestFilter {
private static final Set<String> SKIP_PREFIXES = Set.of(
"/actuator", "/favicon.ico", "/assets/"
);
private static final Set<String> SKIP_SUFFIXES = Set.of(
"/sse"
);
@Override
protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain)
throws ServletException, IOException {
String uri = request.getRequestURI();
if (isNotificationSse(uri)) {
prepareSseResponse(response);
filterChain.doFilter(request, response);
return;
}
if (shouldSkip(uri)) {
filterChain.doFilter(request, response);
return;
@ -91,24 +80,9 @@ public class RequestLoggingFilter extends OncePerRequestFilter {
return true;
}
}
for (String suffix : SKIP_SUFFIXES) {
if (uri.endsWith(suffix)) {
return true;
}
}
return false;
}
private boolean isNotificationSse(String uri) {
return uri != null && uri.endsWith("/notifications/sse");
}
private void prepareSseResponse(HttpServletResponse response) {
response.setContentType(MediaType.TEXT_EVENT_STREAM_VALUE);
response.setHeader(HttpHeaders.CACHE_CONTROL, "no-cache, no-transform");
response.setHeader("X-Accel-Buffering", "no");
}
private String truncate(String value, int maxLength) {
if (value == null || value.length() <= maxLength) {
return value;

32
server/skillhub-app/src/main/java/com/iflytek/skillhub/listener/NotificationEventListener.java
View file

@ -10,7 +10,7 @@ import com.iflytek.skillhub.domain.skill.SkillVersionRepository;
import com.iflytek.skillhub.domain.social.SkillSubscriptionService;
import com.iflytek.skillhub.domain.social.SubscriptionRecipientEligibility;
import com.iflytek.skillhub.notification.domain.NotificationCategory;
import com.iflytek.skillhub.notification.service.NotificationDispatcher;
import com.iflytek.skillhub.notification.service.NotificationService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
@ -31,7 +31,7 @@ public class NotificationEventListener {
private final SkillVersionRepository skillVersionRepository;
private final NamespaceRepository namespaceRepository;
private final RecipientResolver recipientResolver;
private final NotificationDispatcher dispatcher;
private final NotificationService notificationService;
private final SkillSubscriptionService skillSubscriptionService;
private final ObjectMapper objectMapper;
private final SubscriptionRecipientEligibility subscriptionEligibility;
@ -40,7 +40,7 @@ public class NotificationEventListener {
SkillVersionRepository skillVersionRepository,
NamespaceRepository namespaceRepository,
RecipientResolver recipientResolver,
NotificationDispatcher dispatcher,
NotificationService notificationService,
SkillSubscriptionService skillSubscriptionService,
ObjectMapper objectMapper,
SubscriptionRecipientEligibility subscriptionEligibility) {
@ -48,7 +48,7 @@ public class NotificationEventListener {
this.skillVersionRepository = skillVersionRepository;
this.namespaceRepository = namespaceRepository;
this.recipientResolver = recipientResolver;
this.dispatcher = dispatcher;
this.notificationService = notificationService;
this.skillSubscriptionService = skillSubscriptionService;
this.objectMapper = objectMapper;
this.subscriptionEligibility = subscriptionEligibility;
@ -65,7 +65,7 @@ public class NotificationEventListener {
Map<String, Object> body = bodyWithSkill(skill);
versionLabel(event.versionId(), body);
String json = toJson(body);
dispatcher.dispatch(event.publisherId(), NotificationCategory.PUBLISH,
notificationService.create(event.publisherId(), NotificationCategory.PUBLISH,
"SKILL_PUBLISHED", title, json, "SKILL", event.skillId());
});
}
@ -88,7 +88,7 @@ public class NotificationEventListener {
if (subscriberId.equals(event.publisherId())) {
continue; // skip the publisher
}
dispatcher.dispatch(subscriberId, NotificationCategory.PUBLISH,
notificationService.create(subscriberId, NotificationCategory.PUBLISH,
"SUBSCRIPTION_NEW_VERSION", title, json, "SKILL", event.skillId());
}
});
@ -112,7 +112,7 @@ public class NotificationEventListener {
if (subscriberId.equals(event.actorUserId())) {
continue; // skip the actor
}
dispatcher.dispatch(subscriberId, NotificationCategory.PUBLISH,
notificationService.create(subscriberId, NotificationCategory.PUBLISH,
"SUBSCRIPTION_VERSION_YANKED", title, json, "SKILL", event.skillId());
}
});
@ -130,7 +130,7 @@ public class NotificationEventListener {
String json = toJson(body);
List<String> admins = recipientResolver.resolveNamespaceAdmins(event.namespaceId());
for (String admin : admins.stream().distinct().toList()) {
dispatcher.dispatch(admin, NotificationCategory.REVIEW,
notificationService.create(admin, NotificationCategory.REVIEW,
"REVIEW_SUBMITTED", title, json, "REVIEW", event.reviewId());
}
});
@ -147,7 +147,7 @@ public class NotificationEventListener {
String json = toJson(body);
List<String> admins = recipientResolver.resolvePlatformUserAdmins();
for (String admin : admins.stream().distinct().toList()) {
dispatcher.dispatch(admin, NotificationCategory.REVIEW,
notificationService.create(admin, NotificationCategory.REVIEW,
"PROFILE_REVIEW_SUBMITTED", title, json, "PROFILE_REVIEW", event.profileReviewId());
}
}
@ -162,7 +162,7 @@ public class NotificationEventListener {
body.put("reviewerId", event.reviewerId());
versionLabel(event.versionId(), body);
String json = toJson(body);
dispatcher.dispatch(event.submitterId(), NotificationCategory.REVIEW,
notificationService.create(event.submitterId(), NotificationCategory.REVIEW,
"REVIEW_APPROVED", title, json, "SKILL", event.skillId());
});
}
@ -178,7 +178,7 @@ public class NotificationEventListener {
body.put("reason", event.reason());
versionLabel(event.versionId(), body);
String json = toJson(body);
dispatcher.dispatch(event.submitterId(), NotificationCategory.REVIEW,
notificationService.create(event.submitterId(), NotificationCategory.REVIEW,
"REVIEW_REJECTED", title, json, "SKILL", event.skillId());
});
}
@ -195,7 +195,7 @@ public class NotificationEventListener {
String json = toJson(body);
List<String> admins = recipientResolver.resolvePlatformSkillAdmins();
for (String admin : admins.stream().distinct().toList()) {
dispatcher.dispatch(admin, NotificationCategory.PROMOTION,
notificationService.create(admin, NotificationCategory.PROMOTION,
"PROMOTION_SUBMITTED", title, json, "PROMOTION", event.promotionId());
}
});
@ -210,7 +210,7 @@ public class NotificationEventListener {
body.put("promotionId", event.promotionId());
body.put("reviewerId", event.reviewerId());
String json = toJson(body);
dispatcher.dispatch(event.submitterId(), NotificationCategory.PROMOTION,
notificationService.create(event.submitterId(), NotificationCategory.PROMOTION,
"PROMOTION_APPROVED", title, json, "SKILL", event.skillId());
});
}
@ -225,7 +225,7 @@ public class NotificationEventListener {
body.put("reviewerId", event.reviewerId());
body.put("reason", event.reason());
String json = toJson(body);
dispatcher.dispatch(event.submitterId(), NotificationCategory.PROMOTION,
notificationService.create(event.submitterId(), NotificationCategory.PROMOTION,
"PROMOTION_REJECTED", title, json, "SKILL", event.skillId());
});
}
@ -241,7 +241,7 @@ public class NotificationEventListener {
String json = toJson(body);
List<String> admins = recipientResolver.resolvePlatformSkillAdmins();
for (String admin : admins.stream().distinct().toList()) {
dispatcher.dispatch(admin, NotificationCategory.REPORT,
notificationService.create(admin, NotificationCategory.REPORT,
"REPORT_SUBMITTED", title, json, "REPORT", event.reportId());
}
});
@ -257,7 +257,7 @@ public class NotificationEventListener {
body.put("handlerId", event.handlerId());
body.put("action", event.action());
String json = toJson(body);
dispatcher.dispatch(event.reporterId(), NotificationCategory.REPORT,
notificationService.create(event.reporterId(), NotificationCategory.REPORT,
"REPORT_RESOLVED", title, json, "SKILL", event.skillId());
});
}

28
server/skillhub-app/src/test/java/com/iflytek/skillhub/config/WebMvcRateLimitConfigTest.java
View file

@ -1,28 +0,0 @@
package com.iflytek.skillhub.config;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import com.iflytek.skillhub.notification.sse.SseEmitterManager;
import com.iflytek.skillhub.ratelimit.RateLimitInterceptor;
import org.junit.jupiter.api.Test;
import org.springframework.web.servlet.config.annotation.AsyncSupportConfigurer;
class WebMvcRateLimitConfigTest {
@Test
void configureAsyncSupport_shouldSetTimeoutToMatchSseTimeout() {
WebMvcRateLimitConfig config = new WebMvcRateLimitConfig(mock(RateLimitInterceptor.class));
TestAsyncSupportConfigurer asyncSupportConfigurer = new TestAsyncSupportConfigurer();
config.configureAsyncSupport(asyncSupportConfigurer);
assertThat(asyncSupportConfigurer.timeout()).isEqualTo(SseEmitterManager.defaultTimeoutMillis());
}
private static final class TestAsyncSupportConfigurer extends AsyncSupportConfigurer {
private Long timeout() {
return getTimeout();
}
}
}

4
server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/admin/SkillReviewModerationFlowIntegrationTest.java
View file

@ -29,7 +29,7 @@ import com.iflytek.skillhub.infra.jpa.JpaSkillRatingRepository;
import com.iflytek.skillhub.infra.jpa.NamespaceJpaRepository;
import com.iflytek.skillhub.infra.jpa.SkillJpaRepository;
import com.iflytek.skillhub.infra.jpa.UserAccountJpaRepository;
import com.iflytek.skillhub.notification.service.NotificationDispatcher;
import com.iflytek.skillhub.notification.service.NotificationService;
import java.util.List;
import java.util.Set;
import java.util.UUID;
@ -67,7 +67,7 @@ class SkillReviewModerationFlowIntegrationTest {
@MockBean private DeviceAuthService deviceAuthService;
@MockBean private RbacService rbacService;
@MockBean private GovernanceNotificationService governanceNotificationService;
@MockBean private NotificationDispatcher notificationDispatcher;
@MockBean private NotificationService notificationService;
@Test
void hideAndRestorePersistModerationStateAndAuditRows() throws Exception {

6
server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/portal/NotificationControllerTest.java
View file

@ -11,7 +11,6 @@ import com.iflytek.skillhub.dto.PageResponse;
import com.iflytek.skillhub.notification.domain.Notification;
import com.iflytek.skillhub.notification.domain.NotificationCategory;
import com.iflytek.skillhub.notification.service.NotificationService;
import com.iflytek.skillhub.notification.sse.SseEmitterManager;
import com.iflytek.skillhub.observability.RequestIdAccessor;
import java.time.Clock;
import java.time.Instant;
@ -32,9 +31,6 @@ class NotificationControllerTest {
@Mock
private NotificationService notificationService;
@Mock
private SseEmitterManager sseEmitterManager;
private NotificationController controller;
@BeforeEach
@ -46,7 +42,7 @@ class NotificationControllerTest {
Clock.fixed(Instant.parse("2026-03-20T00:00:00Z"), ZoneOffset.UTC),
new RequestIdAccessor()
);
controller = new NotificationController(notificationService, sseEmitterManager, new ObjectMapper(), responseFactory);
controller = new NotificationController(notificationService, new ObjectMapper(), responseFactory);
}
@Test

4
server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/portal/PromotionApprovalFlowIntegrationTest.java
View file

@ -21,7 +21,7 @@ import com.iflytek.skillhub.infra.jpa.PromotionRequestJpaRepository;
import com.iflytek.skillhub.infra.jpa.SkillJpaRepository;
import com.iflytek.skillhub.infra.jpa.SkillVersionJpaRepository;
import com.iflytek.skillhub.infra.jpa.UserAccountJpaRepository;
import com.iflytek.skillhub.notification.service.NotificationDispatcher;
import com.iflytek.skillhub.notification.service.NotificationService;
import java.time.Instant;
import java.util.List;
import java.util.Set;
@ -91,7 +91,7 @@ class PromotionApprovalFlowIntegrationTest {
private GovernanceNotificationService governanceNotificationService;
@MockBean
private NotificationDispatcher notificationDispatcher;
private NotificationService notificationService;
@MockBean
private AuditLogRepository auditLogRepository;

13
server/skillhub-app/src/test/java/com/iflytek/skillhub/exception/GlobalExceptionHandlerTest.java
View file

@ -99,19 +99,8 @@ class GlobalExceptionHandlerTest {
}
@Test
void handleAsyncRequestTimeout_shouldReturnNoContentForSseRequests() {
when(request.getRequestURI()).thenReturn("/api/v1/notifications/sse");
ResponseEntity<?> response = handler.handleAsyncRequestTimeout(new AsyncRequestTimeoutException(), request);
assertThat(response.getStatusCode()).isEqualTo(HttpStatus.NO_CONTENT);
assertThat(response.getBody()).isNull();
}
@Test
void handleAsyncRequestTimeout_shouldReturnApiEnvelopeForNonSseRequests() {
void handleAsyncRequestTimeout_shouldReturnApiEnvelope() {
attachAppender();
when(request.getRequestURI()).thenReturn("/api/v1/publish");
when(request.getMethod()).thenReturn("POST");
when(sensitiveLogSanitizer.sanitizeRequestTarget(request)).thenReturn("/api/v1/publish");

48
server/skillhub-app/src/test/java/com/iflytek/skillhub/filter/RequestLoggingFilterTest.java
View file

@ -16,8 +16,6 @@ import java.util.concurrent.atomic.AtomicReference;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.mock.web.MockHttpServletRequest;
import org.springframework.mock.web.MockHttpServletResponse;
import org.springframework.web.util.ContentCachingResponseWrapper;
@ -83,31 +81,6 @@ class RequestLoggingFilterTest {
assertThat(loggedMessages()).noneMatch(message -> message.contains("/actuator/health"));
}
@Test
void doFilterInternal_skipsOtherSseEndpointsWithoutWrappingResponse()
throws ServletException, IOException {
RequestLoggingFilter filter = new RequestLoggingFilter();
attachAppender();
MockHttpServletRequest request = new MockHttpServletRequest("GET", "/api/web/scan/sse");
MockHttpServletResponse response = new MockHttpServletResponse();
FilterChain filterChain = (req, res) -> {
assertThat(res).isSameAs(response);
res.setContentType("text/event-stream");
res.getWriter().write("event:connected\n");
res.getWriter().flush();
};
filter.doFilter(request, response, filterChain);
assertThat(response.getHeader("Content-Length")).isNull();
assertThat(response.getHeader("X-Accel-Buffering")).isNull();
assertThat(response.getHeader(HttpHeaders.CACHE_CONTROL)).isNull();
assertThat(response.getContentAsString()).isEqualTo("event:connected\n");
assertThat(loggedMessages()).noneMatch(message -> message.contains("/api/web/scan/sse"));
}
@Test
void doFilterInternal_logsCoreSummaryFields()
throws ServletException, IOException {
@ -131,27 +104,6 @@ class RequestLoggingFilterTest {
assertThat(loggedMessages()).noneMatch(message -> message.contains("Headers: {"));
}
@Test
void doFilterInternal_shouldBypassCachingWrapperForNotificationSse() throws Exception {
RequestLoggingFilter filter = new RequestLoggingFilter();
MockHttpServletRequest request = new MockHttpServletRequest("GET", "/api/web/notifications/sse");
MockHttpServletResponse response = new MockHttpServletResponse();
AtomicReference<ServletResponse> responseSeenByChain = new AtomicReference<>();
FilterChain chain = (servletRequest, servletResponse) -> {
responseSeenByChain.set(servletResponse);
servletResponse.getWriter().write("event: connected\n");
servletResponse.flushBuffer();
};
filter.doFilter(request, response, chain);
assertThat(responseSeenByChain.get()).isSameAs(response);
assertThat(response.getHeader("X-Accel-Buffering")).isEqualTo("no");
assertThat(response.getHeader(HttpHeaders.CACHE_CONTROL)).isEqualTo("no-cache, no-transform");
assertThat(response.getContentType()).isEqualTo(MediaType.TEXT_EVENT_STREAM_VALUE);
assertThat(response.getContentAsString()).contains("event: connected");
}
@Test
void doFilterInternal_shouldKeepCachingWrapperForRegularApiResponses() throws Exception {
RequestLoggingFilter filter = new RequestLoggingFilter();

64
server/skillhub-app/src/test/java/com/iflytek/skillhub/listener/NotificationEventListenerTest.java
View file

@ -19,7 +19,7 @@ import com.iflytek.skillhub.domain.user.UserAccount;
import com.iflytek.skillhub.domain.user.UserAccountRepository;
import com.iflytek.skillhub.domain.user.UserStatus;
import com.iflytek.skillhub.notification.domain.NotificationCategory;
import com.iflytek.skillhub.notification.service.NotificationDispatcher;
import com.iflytek.skillhub.notification.service.NotificationService;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@ -39,7 +39,7 @@ class NotificationEventListenerTest {
@Mock SkillVersionRepository skillVersionRepository;
@Mock NamespaceRepository namespaceRepository;
@Mock RecipientResolver recipientResolver;
@Mock NotificationDispatcher dispatcher;
@Mock NotificationService notificationService;
@Mock ObjectMapper objectMapper;
@Mock SkillSubscriptionService skillSubscriptionService;
@Mock UserAccountRepository userAccountRepository;
@ -51,7 +51,7 @@ class NotificationEventListenerTest {
@org.junit.jupiter.api.BeforeEach
void setUpListener() {
listener = new NotificationEventListener(skillRepository, skillVersionRepository, namespaceRepository,
recipientResolver, dispatcher, skillSubscriptionService, objectMapper,
recipientResolver, notificationService, skillSubscriptionService, objectMapper,
new SubscriptionRecipientEligibility(userAccountRepository, namespaceMemberRepository,
new SubscriptionMetadataAccessPolicy()));
}
@ -98,7 +98,7 @@ class NotificationEventListenerTest {
listener.onSkillPublished(new SkillPublishedEvent(1L, 10L, "publisher-1"));
verify(dispatcher).dispatch(eq("publisher-1"), eq(NotificationCategory.PUBLISH),
verify(notificationService).create(eq("publisher-1"), eq(NotificationCategory.PUBLISH),
eq("SKILL_PUBLISHED"), anyString(), anyString(), eq("SKILL"), eq(1L));
}
@ -109,7 +109,7 @@ class NotificationEventListenerTest {
listener.onSkillPublished(new SkillPublishedEvent(1L, 10L, "reviewer-1"));
verifyNoInteractions(dispatcher);
verifyNoInteractions(notificationService);
}
@Test
@ -120,7 +120,7 @@ class NotificationEventListenerTest {
listener.onSkillPublished(new SkillPublishedEvent(1L, 10L, "reviewer-1"));
verifyNoInteractions(dispatcher);
verifyNoInteractions(notificationService);
}
@Test
@ -129,7 +129,7 @@ class NotificationEventListenerTest {
listener.onSkillPublished(new SkillPublishedEvent(99L, 10L, "publisher-1"));
verifyNoInteractions(dispatcher);
verifyNoInteractions(notificationService);
}
@Test
@ -142,10 +142,10 @@ class NotificationEventListenerTest {
listener.onReviewSubmitted(new ReviewSubmittedEvent(100L, 1L, 10L, "submitter-1", 5L));
verify(dispatcher, times(2)).dispatch(anyString(), eq(NotificationCategory.REVIEW),
verify(notificationService, times(2)).create(anyString(), eq(NotificationCategory.REVIEW),
eq("REVIEW_SUBMITTED"), anyString(), anyString(), eq("REVIEW"), eq(100L));
verify(dispatcher).dispatch(eq("admin-1"), any(), any(), any(), any(), any(), any());
verify(dispatcher).dispatch(eq("admin-2"), any(), any(), any(), any(), any(), any());
verify(notificationService).create(eq("admin-1"), any(), any(), any(), any(), any(), any());
verify(notificationService).create(eq("admin-2"), any(), any(), any(), any(), any(), any());
}
@Test
@ -157,10 +157,10 @@ class NotificationEventListenerTest {
listener.onProfileReviewSubmitted(
new ProfileReviewSubmittedEvent(77L, "submitter-1", List.of("displayName")));
verify(dispatcher, times(2)).dispatch(anyString(), eq(NotificationCategory.REVIEW),
verify(notificationService, times(2)).create(anyString(), eq(NotificationCategory.REVIEW),
eq("PROFILE_REVIEW_SUBMITTED"), anyString(), anyString(), eq("PROFILE_REVIEW"), eq(77L));
verify(dispatcher).dispatch(eq("user-admin-1"), any(), any(), any(), any(), any(), any());
verify(dispatcher).dispatch(eq("super-admin-1"), any(), any(), any(), any(), any(), any());
verify(notificationService).create(eq("user-admin-1"), any(), any(), any(), any(), any(), any());
verify(notificationService).create(eq("super-admin-1"), any(), any(), any(), any(), any(), any());
}
@Test
@ -172,7 +172,7 @@ class NotificationEventListenerTest {
listener.onReviewApproved(new ReviewApprovedEvent(100L, 1L, 10L, "reviewer-1", "submitter-1"));
verify(dispatcher).dispatch(eq("submitter-1"), eq(NotificationCategory.REVIEW),
verify(notificationService).create(eq("submitter-1"), eq(NotificationCategory.REVIEW),
eq("REVIEW_APPROVED"), anyString(), anyString(), eq("SKILL"), eq(1L));
}
@ -187,10 +187,10 @@ class NotificationEventListenerTest {
listener.onPromotionSubmitted(new PromotionSubmittedEvent(200L, 1L, 10L, "submitter-1"));
verify(dispatcher, times(2)).dispatch(anyString(), eq(NotificationCategory.PROMOTION),
verify(notificationService, times(2)).create(anyString(), eq(NotificationCategory.PROMOTION),
eq("PROMOTION_SUBMITTED"), anyString(), anyString(), eq("PROMOTION"), eq(200L));
verify(dispatcher).dispatch(eq("platform-admin-1"), any(), any(), any(), any(), any(), any());
verify(dispatcher).dispatch(eq("super-admin-1"), any(), any(), any(), any(), any(), any());
verify(notificationService).create(eq("platform-admin-1"), any(), any(), any(), any(), any(), any());
verify(notificationService).create(eq("super-admin-1"), any(), any(), any(), any(), any(), any());
}
@Test
@ -204,7 +204,7 @@ class NotificationEventListenerTest {
listener.onPromotionSubmitted(new PromotionSubmittedEvent(200L, 1L, 10L, "submitter-1"));
verify(dispatcher, times(1)).dispatch(eq("platform-admin-1"), eq(NotificationCategory.PROMOTION),
verify(notificationService, times(1)).create(eq("platform-admin-1"), eq(NotificationCategory.PROMOTION),
eq("PROMOTION_SUBMITTED"), anyString(), anyString(), eq("PROMOTION"), eq(200L));
}
@ -217,7 +217,7 @@ class NotificationEventListenerTest {
listener.onPromotionApproved(new PromotionApprovedEvent(200L, 1L, "self-admin", "self-admin"));
verify(dispatcher).dispatch(eq("self-admin"), eq(NotificationCategory.PROMOTION),
verify(notificationService).create(eq("self-admin"), eq(NotificationCategory.PROMOTION),
eq("PROMOTION_APPROVED"), anyString(), anyString(), eq("SKILL"), eq(1L));
}
@ -230,7 +230,7 @@ class NotificationEventListenerTest {
listener.onPromotionRejected(new PromotionRejectedEvent(200L, 1L, "self-admin", "self-admin", "not ready"));
verify(dispatcher).dispatch(eq("self-admin"), eq(NotificationCategory.PROMOTION),
verify(notificationService).create(eq("self-admin"), eq(NotificationCategory.PROMOTION),
eq("PROMOTION_REJECTED"), anyString(), anyString(), eq("SKILL"), eq(1L));
}
@ -243,7 +243,7 @@ class NotificationEventListenerTest {
listener.onReportResolved(new ReportResolvedEvent(300L, 1L, "handler-1", "reporter-1", "DISMISSED"));
verify(dispatcher).dispatch(eq("reporter-1"), eq(NotificationCategory.REPORT),
verify(notificationService).create(eq("reporter-1"), eq(NotificationCategory.REPORT),
eq("REPORT_RESOLVED"), anyString(), anyString(), eq("SKILL"), eq(1L));
}
@ -260,7 +260,7 @@ class NotificationEventListenerTest {
listener.onSkillPublishedForSubscribers(new SkillPublishedEvent(1L, 10L, "owner"));
verifyNoInteractions(dispatcher);
verifyNoInteractions(notificationService);
}
@Test
@ -275,7 +275,7 @@ class NotificationEventListenerTest {
listener.onSkillPublishedForSubscribers(new SkillPublishedEvent(1L, 10L, "owner")))
.isInstanceOf(IllegalStateException.class);
verifyNoInteractions(dispatcher);
verifyNoInteractions(notificationService);
}
@Test
@ -300,9 +300,9 @@ class NotificationEventListenerTest {
listener.onSkillPublishedForSubscribers(new SkillPublishedEvent(1L, 10L, "publisher"));
verify(dispatcher).dispatch("admin", NotificationCategory.PUBLISH, "SUBSCRIPTION_NEW_VERSION",
verify(notificationService).create("admin", NotificationCategory.PUBLISH, "SUBSCRIPTION_NEW_VERSION",
"Skill updated: Test Skill", "{\"skillId\":1,\"version\":\"1.0.0\"}", "SKILL", 1L);
verifyNoMoreInteractions(dispatcher);
verifyNoMoreInteractions(notificationService);
}
@Test
@ -318,9 +318,9 @@ class NotificationEventListenerTest {
listener.onSkillVersionYankedForSubscribers(new SkillVersionYankedEvent(1L, 10L, "actor", true));
verify(dispatcher).dispatch("subscriber", NotificationCategory.PUBLISH, "SUBSCRIPTION_VERSION_YANKED",
verify(notificationService).create("subscriber", NotificationCategory.PUBLISH, "SUBSCRIPTION_VERSION_YANKED",
"Skill version yanked: Test Skill", "{\"skillId\":1,\"versionId\":10}", "SKILL", 1L);
verifyNoMoreInteractions(dispatcher);
verifyNoMoreInteractions(notificationService);
}
@Test
@ -335,7 +335,7 @@ class NotificationEventListenerTest {
listener.onSkillVersionYankedForSubscribers(new SkillVersionYankedEvent(1L, 10L, "actor", false));
verifyNoInteractions(dispatcher);
verifyNoInteractions(notificationService);
}
@Test
@ -350,7 +350,7 @@ class NotificationEventListenerTest {
listener.onSkillPublishedForSubscribers(new SkillPublishedEvent(1L, 10L, "owner")))
.isInstanceOf(IllegalStateException.class);
verifyNoInteractions(dispatcher);
verifyNoInteractions(notificationService);
}
@Test
@ -370,7 +370,7 @@ class NotificationEventListenerTest {
listener.onSkillVersionYankedForSubscribers(new SkillVersionYankedEvent(1L, 10L, "actor", true)))
.isInstanceOf(IllegalStateException.class);
verifyNoInteractions(dispatcher);
verifyNoInteractions(notificationService);
}
@Test
@ -391,8 +391,8 @@ class NotificationEventListenerTest {
listener.onSkillVersionYankedForSubscribers(new SkillVersionYankedEvent(1L, 10L, "actor", true));
verify(dispatcher).dispatch(eq("current"), eq(NotificationCategory.PUBLISH),
verify(notificationService).create(eq("current"), eq(NotificationCategory.PUBLISH),
eq("SUBSCRIPTION_VERSION_YANKED"), anyString(), eq("{}"), eq("SKILL"), eq(1L));
verifyNoMoreInteractions(dispatcher);
verifyNoMoreInteractions(notificationService);
}
}

62
server/skillhub-app/src/test/java/com/iflytek/skillhub/listener/SubscriberNotificationSinkTest.java
View file

@ -19,26 +19,18 @@ import com.iflytek.skillhub.domain.social.SubscriptionRecipientEligibility;
import com.iflytek.skillhub.domain.user.UserAccount;
import com.iflytek.skillhub.domain.user.UserAccountRepository;
import com.iflytek.skillhub.domain.user.UserStatus;
import com.iflytek.skillhub.notification.domain.Notification;
import com.iflytek.skillhub.notification.domain.NotificationCategory;
import com.iflytek.skillhub.notification.domain.NotificationChannel;
import com.iflytek.skillhub.notification.service.NotificationDispatcher;
import com.iflytek.skillhub.notification.service.NotificationPreferenceService;
import com.iflytek.skillhub.notification.service.NotificationService;
import com.iflytek.skillhub.notification.sse.SseEmitterManager;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
@ -59,7 +51,6 @@ class SubscriberNotificationSinkTest {
private static final Long SKILL_ID = 1L;
private static final Long NAMESPACE_ID = 5L;
private static final Long VERSION_ID = 10L;
private static final Instant CREATED_AT = Instant.parse("2026-08-19T20:30:00Z");
@Mock SkillRepository skillRepository;
@Mock SkillVersionRepository skillVersionRepository;
@ -69,8 +60,6 @@ class SubscriberNotificationSinkTest {
@Mock UserAccountRepository accountRepository;
@Mock NamespaceMemberRepository memberRepository;
@Mock NotificationService notificationService;
@Mock NotificationPreferenceService preferenceService;
@Mock SseEmitterManager sseEmitterManager;
private NotificationEventListener listener;
@ -78,14 +67,12 @@ class SubscriberNotificationSinkTest {
void setUp() {
SubscriptionRecipientEligibility eligibility = new SubscriptionRecipientEligibility(
accountRepository, memberRepository, new SubscriptionMetadataAccessPolicy());
NotificationDispatcher dispatcher = new NotificationDispatcher(
notificationService, preferenceService, sseEmitterManager);
listener = new NotificationEventListener(skillRepository, skillVersionRepository, namespaceRepository,
recipientResolver, dispatcher, subscriptionService, new ObjectMapper(), eligibility);
recipientResolver, notificationService, subscriptionService, new ObjectMapper(), eligibility);
}
@Test
void publishPersistsAndPushesOnlyCurrentEligibleNonPublisherAcrossAuthorizationMatrix() {
void publishNotifiesOnlyCurrentEligibleNonPublisherAcrossAuthorizationMatrix() {
Skill skill = skill(SkillVisibility.PRIVATE, false, VERSION_ID);
Namespace namespace = namespace(NamespaceStatus.ACTIVE);
List<String> candidates = List.of("publisher", "current-admin", "stale-removed", "inactive",
@ -98,20 +85,18 @@ class SubscriberNotificationSinkTest {
when(memberRepository.findByNamespaceIdAndUserIdIn(NAMESPACE_ID, candidates)).thenReturn(List.of(
member("current-admin", NamespaceRole.ADMIN),
member("private-member", NamespaceRole.MEMBER)));
enablePersistenceFor("current-admin", "SUBSCRIPTION_NEW_VERSION");
listener.onSkillPublishedForSubscribers(new SkillPublishedEvent(SKILL_ID, VERSION_ID, "publisher"));
String body = "{\"skillId\":1,\"skillName\":\"Test Skill\",\"slug\":\"test-skill\",\"namespace\":\"demo\"}";
verify(notificationService).create("current-admin", NotificationCategory.PUBLISH,
"SUBSCRIPTION_NEW_VERSION", "Skill updated: Test Skill", body, "SKILL", SKILL_ID);
assertSingleSse("current-admin", "SUBSCRIPTION_NEW_VERSION", body);
verify(accountRepository).findByIdIn(candidates);
verify(memberRepository).findByNamespaceIdAndUserIdIn(NAMESPACE_ID, candidates);
}
@Test
void hiddenPublishPersistsAndPushesOnlyManagerWhileOrdinaryAndPlatformOnlyCandidatesStayAtZero() {
void hiddenPublishNotifiesOnlyManagerWhileOrdinaryAndPlatformOnlyCandidatesStayAtZero() {
Skill skill = skill(SkillVisibility.PUBLIC, true, VERSION_ID);
Namespace namespace = namespace(NamespaceStatus.ACTIVE);
List<String> candidates = List.of("manager", "ordinary-member", "platform-super-admin");
@ -120,21 +105,19 @@ class SubscriberNotificationSinkTest {
account("manager"), account("ordinary-member"), account("platform-super-admin")));
when(memberRepository.findByNamespaceIdAndUserIdIn(NAMESPACE_ID, candidates)).thenReturn(List.of(
member("manager", NamespaceRole.ADMIN), member("ordinary-member", NamespaceRole.MEMBER)));
enablePersistenceFor("manager", "SUBSCRIPTION_NEW_VERSION");
listener.onSkillPublishedForSubscribers(new SkillPublishedEvent(SKILL_ID, VERSION_ID, "publisher"));
String body = "{\"skillId\":1,\"skillName\":\"Test Skill\",\"slug\":\"test-skill\",\"namespace\":\"demo\"}";
verify(notificationService).create("manager", NotificationCategory.PUBLISH,
"SUBSCRIPTION_NEW_VERSION", "Skill updated: Test Skill", body, "SKILL", SKILL_ID);
assertSingleSse("manager", "SUBSCRIPTION_NEW_VERSION", body);
verify(accountRepository).findByIdIn(candidates);
verify(memberRepository).findByNamespaceIdAndUserIdIn(NAMESPACE_ID, candidates);
}
@ParameterizedTest(name = "yank wasPublished with fallback={0} reaches only current archived-namespace member")
@ValueSource(booleans = {true, false})
void yankPersistsAndPushesOnlyCurrentMemberForFallbackAndNoFallback(boolean hasFallback) {
void yankNotifiesOnlyCurrentMemberForFallbackAndNoFallback(boolean hasFallback) {
Skill skill = skill(SkillVisibility.PUBLIC, false, hasFallback ? 9L : null);
Namespace namespace = namespace(NamespaceStatus.ARCHIVED);
List<String> candidates = List.of("actor", "current", "removed", "inactive", "missing");
@ -143,7 +126,6 @@ class SubscriberNotificationSinkTest {
account("actor"), account("current"), account("removed"), inactiveAccount("inactive")));
when(memberRepository.findByNamespaceIdAndUserIdIn(NAMESPACE_ID, candidates)).thenReturn(List.of(
member("actor", NamespaceRole.ADMIN), member("current", NamespaceRole.MEMBER)));
enablePersistenceFor("current", "SUBSCRIPTION_VERSION_YANKED");
listener.onSkillVersionYankedForSubscribers(
new SkillVersionYankedEvent(SKILL_ID, VERSION_ID, "actor", true));
@ -151,13 +133,12 @@ class SubscriberNotificationSinkTest {
String body = "{\"skillId\":1,\"skillName\":\"Test Skill\",\"slug\":\"test-skill\",\"namespace\":\"demo\"}";
verify(notificationService).create("current", NotificationCategory.PUBLISH,
"SUBSCRIPTION_VERSION_YANKED", "Skill version yanked: Test Skill", body, "SKILL", SKILL_ID);
assertSingleSse("current", "SUBSCRIPTION_VERSION_YANKED", body);
verify(accountRepository).findByIdIn(candidates);
verify(memberRepository).findByNamespaceIdAndUserIdIn(NAMESPACE_ID, candidates);
}
@Test
void yankWithoutVerifiedPublishedPreStateProducesNoPersistenceOrSse() {
void yankWithoutVerifiedPublishedPreStateProducesNoNotification() {
Skill skill = skill(SkillVisibility.PUBLIC, false, null);
Namespace namespace = namespace(NamespaceStatus.ACTIVE);
List<String> candidates = List.of("current");
@ -168,14 +149,14 @@ class SubscriberNotificationSinkTest {
listener.onSkillVersionYankedForSubscribers(
new SkillVersionYankedEvent(SKILL_ID, VERSION_ID, "actor", false));
verifyNoInteractions(notificationService, preferenceService, sseEmitterManager);
verifyNoInteractions(notificationService);
verify(accountRepository).findByIdIn(candidates);
verify(memberRepository).findByNamespaceIdAndUserIdIn(NAMESPACE_ID, candidates);
}
@ParameterizedTest(name = "{0} batch failure happens before every final sink")
@EnumSource(BatchFailure.class)
void authoritativeBatchFailureProducesNoPartialPersistenceOrSse(BatchFailure failure) {
void authoritativeBatchFailureProducesNoPartialNotification(BatchFailure failure) {
Skill skill = skill(SkillVisibility.PUBLIC, false, VERSION_ID);
Namespace namespace = namespace(NamespaceStatus.ACTIVE);
List<String> candidates = List.of("first", "second");
@ -199,7 +180,7 @@ class SubscriberNotificationSinkTest {
.isInstanceOf(IllegalStateException.class)
.hasMessageContaining(failure.name().toLowerCase());
verifyNoInteractions(notificationService, preferenceService, sseEmitterManager);
verifyNoInteractions(notificationService);
verify(namespaceRepository, times(1)).findById(NAMESPACE_ID);
if (failure == BatchFailure.NAMESPACE) {
verify(accountRepository, never()).findByIdIn(anyList());
@ -217,27 +198,6 @@ class SubscriberNotificationSinkTest {
when(namespaceRepository.findById(NAMESPACE_ID)).thenReturn(Optional.of(namespace));
}
private void enablePersistenceFor(String recipient, String eventType) {
when(preferenceService.isEnabled(recipient, NotificationCategory.PUBLISH, NotificationChannel.IN_APP))
.thenReturn(true);
when(notificationService.create(eq(recipient), eq(NotificationCategory.PUBLISH), eq(eventType),
any(String.class), any(String.class), eq("SKILL"), eq(SKILL_ID)))
.thenAnswer(invocation -> notification(recipient, eventType,
invocation.getArgument(3), invocation.getArgument(4)));
}
private void assertSingleSse(String recipient, String eventType, String body) {
@SuppressWarnings("unchecked")
ArgumentCaptor<Map<String, Object>> payload = ArgumentCaptor.forClass(Map.class);
verify(sseEmitterManager).push(eq(recipient), payload.capture());
assertThat(payload.getValue())
.containsEntry("id", 42L)
.containsEntry("category", "PUBLISH")
.containsEntry("eventType", eventType)
.containsEntry("bodyJson", body)
.containsEntry("entityType", "SKILL")
.containsEntry("entityId", SKILL_ID);
}
private Skill skill(SkillVisibility visibility, boolean hidden, Long latestVersionId) {
Skill skill = new Skill(NAMESPACE_ID, "test-skill", "publisher", visibility);
@ -268,12 +228,6 @@ class SubscriberNotificationSinkTest {
return new NamespaceMember(NAMESPACE_ID, userId, role);
}
private Notification notification(String recipient, String eventType, String title, String body) {
Notification notification = new Notification(recipient, NotificationCategory.PUBLISH, eventType,
title, body, "SKILL", SKILL_ID, CREATED_AT);
setId(notification, 42L);
return notification;
}
private void setId(Object entity, Long id) {
try {

4
server/skillhub-notification/pom.xml
View file

@ -14,10 +14,6 @@
<groupId>com.iflytek.skillhub</groupId>
<artifactId>skillhub-domain</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>

60
server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/service/NotificationDispatcher.java
View file

@ -1,60 +0,0 @@
package com.iflytek.skillhub.notification.service;
import com.iflytek.skillhub.notification.domain.NotificationCategory;
import com.iflytek.skillhub.notification.domain.NotificationChannel;
import com.iflytek.skillhub.notification.domain.Notification;
import com.iflytek.skillhub.notification.sse.SseEmitterManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import java.util.Map;
@Service
public class NotificationDispatcher {
private static final Logger log = LoggerFactory.getLogger(NotificationDispatcher.class);
private final NotificationService notificationService;
private final NotificationPreferenceService preferenceService;
private final SseEmitterManager sseEmitterManager;
public NotificationDispatcher(NotificationService notificationService,
NotificationPreferenceService preferenceService,
SseEmitterManager sseEmitterManager) {
this.notificationService = notificationService;
this.preferenceService = preferenceService;
this.sseEmitterManager = sseEmitterManager;
}
public void dispatch(String recipientId, NotificationCategory category,
String eventType, String title, String bodyJson,
String entityType, Long entityId) {
// Check user preference
if (!preferenceService.isEnabled(recipientId, category, NotificationChannel.IN_APP)) {
log.debug("Notification {} suppressed for user {} (preference disabled)", eventType, recipientId);
return;
}
// Persist notification
Notification notification = notificationService.create(
recipientId, category, eventType, title, bodyJson, entityType, entityId);
// Push via SSE
try {
sseEmitterManager.push(recipientId, Map.of(
"id", notification.getId(),
"category", notification.getCategory().name(),
"eventType", notification.getEventType(),
"title", notification.getTitle(),
"bodyJson", notification.getBodyJson() != null ? notification.getBodyJson() : "",
"entityType", notification.getEntityType() != null ? notification.getEntityType() : "",
"entityId", notification.getEntityId() != null ? notification.getEntityId() : 0,
"createdAt", notification.getCreatedAt().toString()
));
} catch (Exception e) {
log.warn("Failed to push SSE notification to user {}", recipientId, e);
// Notification is already persisted, SSE push failure is non-critical
}
}
}

17
server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/service/NotificationService.java
View file

@ -16,20 +16,27 @@ import java.time.Instant;
public class NotificationService {
private final NotificationRepository notificationRepository;
private final NotificationPreferenceService preferenceService;
private final Clock clock;
public NotificationService(NotificationRepository notificationRepository, Clock clock) {
public NotificationService(NotificationRepository notificationRepository,
NotificationPreferenceService preferenceService,
Clock clock) {
this.notificationRepository = notificationRepository;
this.preferenceService = preferenceService;
this.clock = clock;
}
@Transactional
public Notification create(String recipientId, NotificationCategory category,
String eventType, String title, String bodyJson,
String entityType, Long entityId) {
public void create(String recipientId, NotificationCategory category,
String eventType, String title, String bodyJson,
String entityType, Long entityId) {
if (!preferenceService.isEnabled(recipientId, category, NotificationChannel.IN_APP)) {
return;
}
Notification notification = new Notification(recipientId, category, eventType,
title, bodyJson, entityType, entityId, Instant.now(clock));
return notificationRepository.save(notification);
notificationRepository.save(notification);
}
@Transactional(readOnly = true)

136
server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/sse/SseEmitterManager.java
View file

@ -1,136 +0,0 @@
package com.iflytek.skillhub.notification.sse;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
@Component
public class SseEmitterManager {
private static final Logger log = LoggerFactory.getLogger(SseEmitterManager.class);
private static final long SSE_TIMEOUT = 10 * 60_000L;
private static final long HEARTBEAT_INTERVAL = 30_000L;
private static final int MAX_EMITTERS_PER_USER = 5;
private static final int MAX_TOTAL_EMITTERS = 1000;
private final ConcurrentHashMap<String, CopyOnWriteArrayList<TrackedEmitter>> emitters = new ConcurrentHashMap<>();
private final AtomicInteger totalCount = new AtomicInteger(0);
private final Function<String, SseEmitter> emitterFactory;
public SseEmitterManager() {
this(userId -> new SseEmitter(SSE_TIMEOUT));
}
SseEmitterManager(Function<String, SseEmitter> emitterFactory) {
this.emitterFactory = emitterFactory;
}
public SseEmitter register(String userId) {
if (totalCount.get() >= MAX_TOTAL_EMITTERS) {
throw new IllegalStateException("SSE connection limit reached");
}
CopyOnWriteArrayList<TrackedEmitter> userEmitters = emitters.computeIfAbsent(userId, k -> new CopyOnWriteArrayList<>());
if (userEmitters.size() >= MAX_EMITTERS_PER_USER) {
TrackedEmitter oldest = userEmitters.get(0);
cleanup(userId, userEmitters, oldest);
try {
oldest.emitter().complete();
} catch (IllegalStateException ex) {
log.debug("Emitter already completed during eviction for user {}", userId);
}
}
TrackedEmitter trackedEmitter = new TrackedEmitter(emitterFactory.apply(userId));
userEmitters.add(trackedEmitter);
totalCount.incrementAndGet();
Runnable cleanup = () -> cleanup(userId, userEmitters, trackedEmitter);
trackedEmitter.emitter().onCompletion(cleanup);
trackedEmitter.emitter().onTimeout(cleanup);
trackedEmitter.emitter().onError(e -> cleanup.run());
try {
trackedEmitter.emitter().send(SseEmitter.event().name("connected").data("ok"));
} catch (IOException e) {
cleanup.run();
}
return trackedEmitter.emitter();
}
public void push(String userId, Object data) {
CopyOnWriteArrayList<TrackedEmitter> userEmitters = emitters.get(userId);
if (userEmitters == null) return;
for (TrackedEmitter trackedEmitter : userEmitters) {
try {
trackedEmitter.emitter().send(SseEmitter.event().name("notification").data(data));
} catch (IOException e) {
log.debug("Failed to push to user {}, removing emitter", userId);
cleanup(userId, userEmitters, trackedEmitter);
}
}
}
@Scheduled(fixedRate = HEARTBEAT_INTERVAL)
public void heartbeat() {
emitters.forEach((userId, userEmitters) -> {
for (TrackedEmitter trackedEmitter : userEmitters) {
try {
trackedEmitter.emitter().send(SseEmitter.event().comment("ping"));
} catch (IOException e) {
log.debug("Heartbeat failed for user {}", userId);
cleanup(userId, userEmitters, trackedEmitter);
}
}
});
}
int totalEmitters() {
return totalCount.get();
}
int emittersForUser(String userId) {
return emitters.getOrDefault(userId, new CopyOnWriteArrayList<>()).size();
}
public static long defaultTimeoutMillis() {
return SSE_TIMEOUT;
}
public static long heartbeatIntervalMillis() {
return HEARTBEAT_INTERVAL;
}
private void cleanup(String userId,
CopyOnWriteArrayList<TrackedEmitter> userEmitters,
TrackedEmitter trackedEmitter) {
if (!trackedEmitter.markCleaned()) {
return;
}
userEmitters.remove(trackedEmitter);
totalCount.decrementAndGet();
if (userEmitters.isEmpty()) {
emitters.remove(userId, userEmitters);
}
}
private record TrackedEmitter(SseEmitter emitter, AtomicBoolean cleaned) {
private TrackedEmitter(SseEmitter emitter) {
this(emitter, new AtomicBoolean(false));
}
boolean markCleaned() {
return cleaned.compareAndSet(false, true);
}
}
}

128
server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/service/NotificationDispatcherTest.java
View file

@ -1,128 +0,0 @@
package com.iflytek.skillhub.notification.service;
import com.iflytek.skillhub.notification.domain.Notification;
import com.iflytek.skillhub.notification.domain.NotificationCategory;
import com.iflytek.skillhub.notification.domain.NotificationChannel;
import com.iflytek.skillhub.notification.sse.SseEmitterManager;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import java.time.Instant;
import java.util.Map;
import org.mockito.ArgumentCaptor;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
@ExtendWith(MockitoExtension.class)
class NotificationDispatcherTest {
@Mock private NotificationService notificationService;
@Mock private NotificationPreferenceService preferenceService;
@Mock private SseEmitterManager sseEmitterManager;
private NotificationDispatcher dispatcher;
@BeforeEach
void setUp() {
dispatcher = new NotificationDispatcher(notificationService, preferenceService, sseEmitterManager);
}
private Notification buildNotificationMock() {
Notification n = mock(Notification.class);
lenient().when(n.getId()).thenReturn(1L);
lenient().when(n.getCategory()).thenReturn(NotificationCategory.REVIEW);
lenient().when(n.getEventType()).thenReturn("review.approved");
lenient().when(n.getTitle()).thenReturn("Title");
lenient().when(n.getBodyJson()).thenReturn("{}");
lenient().when(n.getEntityType()).thenReturn("skill");
lenient().when(n.getEntityId()).thenReturn(1L);
lenient().when(n.getCreatedAt()).thenReturn(Instant.parse("2026-03-19T10:00:00Z"));
return n;
}
@Test
void dispatch_shouldPersistAndPushWhenEnabled() {
Notification notification = buildNotificationMock();
when(preferenceService.isEnabled("user-1", NotificationCategory.REVIEW, NotificationChannel.IN_APP))
.thenReturn(true);
when(notificationService.create(any(), any(), any(), any(), any(), any(), any()))
.thenReturn(notification);
dispatcher.dispatch("user-1", NotificationCategory.REVIEW,
"review.approved", "Title", "{}", "skill", 1L);
verify(notificationService).create("user-1", NotificationCategory.REVIEW,
"review.approved", "Title", "{}", "skill", 1L);
verify(sseEmitterManager).push(eq("user-1"), any());
}
@Test
void dispatch_persistsExactSubscriberNotificationAndPushesSameRecipientVisiblePayload() {
Notification notification = new Notification("subscriber-1", NotificationCategory.PUBLISH,
"SUBSCRIPTION_NEW_VERSION", "Skill updated: Demo",
"{\"skillId\":1,\"versionId\":10}", "SKILL", 1L,
Instant.parse("2026-08-19T20:30:00Z"));
try {
var id = Notification.class.getDeclaredField("id");
id.setAccessible(true);
id.set(notification, 42L);
} catch (ReflectiveOperationException e) {
throw new IllegalStateException(e);
}
when(preferenceService.isEnabled("subscriber-1", NotificationCategory.PUBLISH,
NotificationChannel.IN_APP)).thenReturn(true);
when(notificationService.create(any(), any(), any(), any(), any(), any(), any()))
.thenReturn(notification);
dispatcher.dispatch("subscriber-1", NotificationCategory.PUBLISH, "SUBSCRIPTION_NEW_VERSION",
"Skill updated: Demo", "{\"skillId\":1,\"versionId\":10}", "SKILL", 1L);
verify(notificationService).create("subscriber-1", NotificationCategory.PUBLISH,
"SUBSCRIPTION_NEW_VERSION", "Skill updated: Demo",
"{\"skillId\":1,\"versionId\":10}", "SKILL", 1L);
@SuppressWarnings("unchecked")
ArgumentCaptor<Map<String, Object>> payload = ArgumentCaptor.forClass(Map.class);
verify(sseEmitterManager).push(eq("subscriber-1"), payload.capture());
assertThat(payload.getValue()).containsEntry("id", 42L)
.containsEntry("category", "PUBLISH")
.containsEntry("eventType", "SUBSCRIPTION_NEW_VERSION")
.containsEntry("bodyJson", "{\"skillId\":1,\"versionId\":10}")
.containsEntry("entityType", "SKILL")
.containsEntry("entityId", 1L);
}
@Test
void dispatch_shouldSkipWhenPreferenceDisabled() {
when(preferenceService.isEnabled("user-1", NotificationCategory.REVIEW, NotificationChannel.IN_APP))
.thenReturn(false);
dispatcher.dispatch("user-1", NotificationCategory.REVIEW,
"review.approved", "Title", "{}", "skill", 1L);
verify(notificationService, never()).create(any(), any(), any(), any(), any(), any(), any());
verify(sseEmitterManager, never()).push(any(), any());
}
@Test
void dispatch_shouldStillPersistWhenSsePushFails() {
Notification notification = buildNotificationMock();
when(preferenceService.isEnabled("user-1", NotificationCategory.REVIEW, NotificationChannel.IN_APP))
.thenReturn(true);
when(notificationService.create(any(), any(), any(), any(), any(), any(), any()))
.thenReturn(notification);
doThrow(new RuntimeException("SSE failure")).when(sseEmitterManager).push(any(), any());
dispatcher.dispatch("user-1", NotificationCategory.REVIEW,
"review.approved", "Title", "{}", "skill", 1L);
verify(notificationService).create("user-1", NotificationCategory.REVIEW,
"review.approved", "Title", "{}", "skill", 1L);
}
}

26
server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/service/NotificationServiceTest.java
View file

@ -27,30 +27,40 @@ import static org.mockito.Mockito.*;
class NotificationServiceTest {
@Mock private NotificationRepository notificationRepository;
@Mock private NotificationPreferenceService preferenceService;
private Clock clock;
private NotificationService service;
@BeforeEach
void setUp() {
clock = Clock.fixed(Instant.parse("2026-03-19T10:00:00Z"), ZoneOffset.UTC);
service = new NotificationService(notificationRepository, clock);
service = new NotificationService(notificationRepository, preferenceService, clock);
}
@Test
void createNotification_shouldSaveAndReturn() {
Notification notification = new Notification("user-1", NotificationCategory.REVIEW,
"review.approved", "notification.review.approved",
"{\"skillName\":\"test\"}", "skill", 1L, Instant.now(clock));
when(notificationRepository.save(any())).thenReturn(notification);
void createNotification_shouldSaveWhenEnabled() {
when(preferenceService.isEnabled("user-1", NotificationCategory.REVIEW, NotificationChannel.IN_APP))
.thenReturn(true);
Notification result = service.create("user-1", NotificationCategory.REVIEW,
service.create("user-1", NotificationCategory.REVIEW,
"review.approved", "notification.review.approved",
"{\"skillName\":\"test\"}", "skill", 1L);
assertNotNull(result);
verify(notificationRepository).save(any(Notification.class));
}
@Test
void createNotification_shouldSkipWhenPreferenceDisabled() {
when(preferenceService.isEnabled("user-1", NotificationCategory.REVIEW, NotificationChannel.IN_APP))
.thenReturn(false);
service.create("user-1", NotificationCategory.REVIEW,
"review.approved", "notification.review.approved",
"{\"skillName\":\"test\"}", "skill", 1L);
verifyNoInteractions(notificationRepository);
}
@Test
void getUnreadCount_shouldReturnCount() {
when(notificationRepository.countByRecipientIdAndStatus("user-1", NotificationStatus.UNREAD))

256
server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/sse/SseEmitterManagerTest.java
View file

@ -1,256 +0,0 @@
package com.iflytek.skillhub.notification.sse;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import java.io.IOException;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
class SseEmitterManagerTest {
private Queue<TestEmitter> emitters;
private SseEmitterManager manager;
@BeforeEach
void setUp() {
emitters = new ArrayDeque<>();
manager = new SseEmitterManager(userId -> {
TestEmitter emitter = emitters.remove();
emitter.registerUser(userId);
return emitter;
});
}
@Test
void register_shouldReturnEmitter() {
TestEmitter testEmitter = new TestEmitter();
emitters.add(testEmitter);
SseEmitter emitter = manager.register("user-1");
assertNotNull(emitter);
assertEquals(1, manager.totalEmitters());
assertEquals(1, manager.emittersForUser("user-1"));
assertEquals(1, testEmitter.sentEventCount());
assertTrue(testEmitter.sentEventData(0).stream()
.anyMatch(value -> value.toString().contains("event:connected")));
assertTrue(testEmitter.sentEventData(0).contains("ok"));
assertTrue(testEmitter.isOpen());
}
@Test
void defaultTimeout_shouldOutliveManyHeartbeats() {
assertTrue(SseEmitterManager.defaultTimeoutMillis() >= 10 * 60_000L);
assertTrue(SseEmitterManager.defaultTimeoutMillis() > SseEmitterManager.heartbeatIntervalMillis() * 2);
}
@Test
void register_shouldKeepAccurateCountWhenEvictingOldestEmitter() {
for (int i = 0; i < 6; i++) {
emitters.add(new TestEmitter());
}
for (int i = 0; i < 6; i++) {
manager.register("user-evict");
}
assertEquals(5, manager.totalEmitters());
assertEquals(5, manager.emittersForUser("user-evict"));
}
@Test
void register_shouldTolerateEmitterThatThrowsDuringEvictionCompletion() {
TestEmitter oldest = new TestEmitter();
oldest.throwOnComplete();
emitters.add(oldest);
for (int i = 0; i < 5; i++) {
emitters.add(new TestEmitter());
}
for (int i = 0; i < 5; i++) {
manager.register("user-evict");
}
assertDoesNotThrow(() -> manager.register("user-evict"));
assertEquals(5, manager.totalEmitters());
assertEquals(5, manager.emittersForUser("user-evict"));
}
@Test
void push_shouldRemoveEmitterWhenSendFails() {
TestEmitter healthy = new TestEmitter();
TestEmitter broken = new TestEmitter();
broken.failAfterConnected();
emitters.add(healthy);
emitters.add(broken);
manager.register("user-1");
manager.register("user-1");
manager.push("user-1", "payload");
assertEquals(1, manager.totalEmitters());
assertEquals(1, manager.emittersForUser("user-1"));
}
@Test
void push_shouldSendNotificationEventToRegisteredOpenEmitter() {
TestEmitter emitter = new TestEmitter();
emitters.add(emitter);
manager.register("user-1");
Map<String, Object> payload = Map.of(
"id", 42L,
"eventType", "PROFILE_REVIEW_SUBMITTED"
);
manager.push("user-1", payload);
assertEquals(2, emitter.sentEventCount());
assertTrue(emitter.sentEventData(1).stream()
.anyMatch(value -> value.toString().contains("event:notification")));
assertTrue(emitter.sentEventData(1).contains(payload));
assertTrue(emitter.isOpen());
assertEquals(1, manager.totalEmitters());
assertEquals(1, manager.emittersForUser("user-1"));
}
@Test
void heartbeat_shouldRemoveEmitterWhenSendFails() {
TestEmitter healthy = new TestEmitter();
TestEmitter broken = new TestEmitter();
broken.failAfterConnected();
emitters.add(healthy);
emitters.add(broken);
manager.register("user-1");
manager.register("user-1");
manager.heartbeat();
assertEquals(1, manager.totalEmitters());
assertEquals(1, manager.emittersForUser("user-1"));
}
@Test
void cleanup_shouldBeIdempotent() {
TestEmitter emitter = new TestEmitter();
emitters.add(emitter);
manager.register("user-1");
emitter.fireError();
emitter.fireError();
assertEquals(0, manager.totalEmitters());
assertEquals(0, manager.emittersForUser("user-1"));
}
@Test
void push_shouldDoNothingForUnregisteredUser() {
assertDoesNotThrow(() -> manager.push("unknown-user", "some-data"));
assertEquals(0, manager.totalEmitters());
}
@Test
void register_multipleUsers_shouldTrackSeparately() {
emitters.add(new TestEmitter());
emitters.add(new TestEmitter());
SseEmitter emitter1 = manager.register("user-1");
SseEmitter emitter2 = manager.register("user-2");
assertNotNull(emitter1);
assertNotNull(emitter2);
assertEquals(2, manager.totalEmitters());
assertEquals(1, manager.emittersForUser("user-1"));
assertEquals(1, manager.emittersForUser("user-2"));
}
private static final class TestEmitter extends SseEmitter {
private final AtomicInteger errorCallbacks = new AtomicInteger(0);
private Runnable completionCallback = () -> {};
private Runnable timeoutCallback = () -> {};
private java.util.function.Consumer<Throwable> errorCallback = error -> {};
private String userId;
private boolean failAfterConnected;
private boolean throwOnComplete;
private int sendCount;
private boolean completed;
private final List<List<Object>> sentEvents = new ArrayList<>();
private TestEmitter() {
super(60_000L);
}
void registerUser(String userId) {
this.userId = userId;
}
void failAfterConnected() {
this.failAfterConnected = true;
}
void throwOnComplete() {
this.throwOnComplete = true;
}
void fireError() {
errorCallback.accept(new IOException("boom-" + userId + "-" + errorCallbacks.incrementAndGet()));
}
boolean isOpen() {
return !completed;
}
int sentEventCount() {
return sentEvents.size();
}
List<Object> sentEventData(int index) {
return sentEvents.get(index);
}
@Override
public synchronized void onCompletion(Runnable callback) {
this.completionCallback = callback;
}
@Override
public synchronized void onTimeout(Runnable callback) {
this.timeoutCallback = callback;
}
@Override
public synchronized void onError(java.util.function.Consumer<Throwable> callback) {
this.errorCallback = callback;
}
@Override
public void complete() {
if (throwOnComplete) {
throw new IllegalStateException("already complete");
}
completed = true;
completionCallback.run();
}
@Override
public void send(SseEventBuilder builder) throws IOException {
sendCount++;
if (failAfterConnected && sendCount > 1) {
throw new IOException("send failed");
}
sentEvents.add(builder.build().stream()
.map(ResponseBodyEmitter.DataWithMediaType::getData)
.toList());
}
}
}

7
web/e2e/promotions-review.spec.ts
View file

@ -87,13 +87,6 @@ test.describe('Promotion review dashboard', () => {
}),
})
})
await page.route('**/api/web/notifications/sse', async (route) => {
await route.fulfill({
status: 200,
contentType: 'text/event-stream',
body: '',
})
})
})
async function installPromotionRouteMock(page: Page, expectedSignatures: string[]) {

3
web/e2e/theme-toggle.spec.ts
View file

@ -51,9 +51,6 @@ test.describe('Light and dark theme', () => {
}),
})
})
await page.route('**/api/web/notifications/sse', async (route) => {
await route.fulfill({ status: 204 })
})
await page.route('**/api/web/me/stars?*', async (route) => {
await route.fulfill({
status: 200,

76
web/src/api/generated/schema.d.ts vendored
View file

@ -2788,38 +2788,6 @@ export interface paths {
patch?: never;
trace?: never;
};
"/api/v1/notifications/sse": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get: operations["sse"];
put?: never;
post?: never;
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/api/web/notifications/sse": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get: operations["sse_1"];
put?: never;
post?: never;
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/api/web/notifications": {
parameters: {
query?: never;
@ -4997,10 +4965,6 @@ export interface components {
timestamp?: string;
requestId?: string;
};
SseEmitter: {
/** Format: int64 */
timeout?: number;
};
ApiResponsePageResponseNotificationResponse: {
/** Format: int32 */
code?: number;
@ -10994,46 +10958,6 @@ export interface operations {
};
};
};
sse: {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
requestBody?: never;
responses: {
/** @description OK */
200: {
headers: {
[name: string]: unknown;
};
content: {
"text/event-stream": components["schemas"]["SseEmitter"];
};
};
};
};
sse_1: {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
requestBody?: never;
responses: {
/** @description OK */
200: {
headers: {
[name: string]: unknown;
};
content: {
"text/event-stream": components["schemas"]["SseEmitter"];
};
};
};
};
list_2: {
parameters: {
query?: {

5
web/src/features/notification/notification-bell.tsx
View file

@ -2,7 +2,6 @@ import { useState, useRef, useEffect } from 'react'
import { useTranslation } from 'react-i18next'
import { useAuth } from '@/features/auth/use-auth'
import { useUnreadCount } from './use-notifications'
import { useNotificationSse } from './use-notification-sse'
import { NotificationDropdown } from './notification-dropdown'
export function resolveNotificationUserId(user?: { userId?: string } | null) {
@ -11,7 +10,7 @@ export function resolveNotificationUserId(user?: { userId?: string } | null) {
/**
* Bell icon with unread badge. Toggles the notification dropdown on click.
* SSE connection is established here at the authenticated user level.
* The unread count is refreshed through the notification HTTP polling module.
*/
export function NotificationBell() {
const { t } = useTranslation()
@ -23,8 +22,6 @@ export function NotificationBell() {
const { data: unreadData } = useUnreadCount(notificationUserId)
const unreadCount = unreadData?.count ?? 0
useNotificationSse(notificationUserId)
// Close dropdown when clicking outside
useEffect(() => {
if (!open) return

4
web/src/features/notification/notification-dropdown.tsx
View file

@ -4,7 +4,7 @@ import type { NotificationItem } from '@/api/types'
import { getNotificationItems } from './notification-page'
import { resolveNotificationDisplay } from './notification-content'
import { useAuth } from '@/features/auth/use-auth'
import { useNotifications, useMarkAllRead, useMarkRead } from './use-notifications'
import { useNotificationList, useMarkAllRead, useMarkRead } from './use-notifications'
import { resolveNotificationTarget } from './notification-target'
import { formatRelativeTime } from '@/shared/lib/format-relative-time'
@ -18,7 +18,7 @@ interface Props {
export function NotificationDropdown({ onClose }: Props) {
const { t, i18n } = useTranslation()
const { user } = useAuth()
const { data, isLoading } = useNotifications(user?.userId, 0, 5)
const { data, isLoading } = useNotificationList(user?.userId, 0, 5)
const markAllRead = useMarkAllRead(user?.userId)
const markRead = useMarkRead(user?.userId)

85
web/src/features/notification/notification-sse-coordinator.test.ts
View file

@ -1,85 +0,0 @@
import { describe, expect, it, vi } from 'vitest'
import { createNotificationSseConnection } from './notification-sse-coordinator'
class FakeEventSource {
listeners = new Map<string, Array<(event: MessageEvent) => void>>()
closed = false
addEventListener(type: string, listener: (event: MessageEvent) => void) {
const current = this.listeners.get(type) ?? []
current.push(listener)
this.listeners.set(type, current)
}
close() {
this.closed = true
}
emit(type: string) {
for (const listener of this.listeners.get(type) ?? []) {
listener(new MessageEvent(type))
}
}
}
describe('createNotificationSseConnection', () => {
it('backs off reconnect attempts after repeated errors', () => {
vi.useFakeTimers()
const sources: FakeEventSource[] = []
const connection = createNotificationSseConnection(
'/api/web/notifications/sse',
() => {
const source = new FakeEventSource()
sources.push(source)
return source
},
{ setTimeout, clearTimeout },
)
expect(sources).toHaveLength(1)
sources[0].emit('error')
expect(sources[0].closed).toBe(true)
expect(sources).toHaveLength(1)
vi.advanceTimersByTime(999)
expect(sources).toHaveLength(1)
vi.advanceTimersByTime(1)
expect(sources).toHaveLength(2)
sources[1].emit('error')
vi.advanceTimersByTime(1_999)
expect(sources).toHaveLength(2)
vi.advanceTimersByTime(1)
expect(sources).toHaveLength(3)
connection.close()
vi.useRealTimers()
})
it('resets reconnect delay after a successful open event', () => {
vi.useFakeTimers()
const sources: FakeEventSource[] = []
createNotificationSseConnection(
'/api/web/notifications/sse',
() => {
const source = new FakeEventSource()
sources.push(source)
return source
},
{ setTimeout, clearTimeout },
)
sources[0].emit('error')
vi.advanceTimersByTime(1_000)
expect(sources).toHaveLength(2)
sources[1].emit('open')
sources[1].emit('error')
vi.advanceTimersByTime(1_000)
expect(sources).toHaveLength(3)
vi.useRealTimers()
})
})

102
web/src/features/notification/notification-sse-coordinator.ts
View file

@ -1,102 +0,0 @@
const SHARED_BROWSER_SSE_ENABLED = false
const INITIAL_RECONNECT_DELAY_MS = 1_000
const MAX_RECONNECT_DELAY_MS = 30_000
type NotificationListener = (event: MessageEvent) => void
type SourceEventListener = (event: Event) => void
type NotificationEventSource = {
addEventListener: (type: string, listener: SourceEventListener) => void
close: () => void
}
type EventSourceFactory = (url: string) => NotificationEventSource
type TimerApi = {
setTimeout: typeof setTimeout
clearTimeout: typeof clearTimeout
}
export type NotificationSseConnection = {
addEventListener: (type: string, listener: NotificationListener) => void
close: () => void
}
export function isSharedBrowserSseEnabled() {
return SHARED_BROWSER_SSE_ENABLED
}
export function createNotificationSseConnection(
url: string,
eventSourceFactory: EventSourceFactory = (targetUrl) =>
new EventSource(targetUrl, { withCredentials: true }),
timerApi: TimerApi = {
setTimeout: (...args) => globalThis.setTimeout(...args),
clearTimeout: (timer) => globalThis.clearTimeout(timer),
},
): NotificationSseConnection {
return new ManagedNotificationSseConnection(url, eventSourceFactory, timerApi)
}
class ManagedNotificationSseConnection implements NotificationSseConnection {
private readonly listeners = new Map<string, NotificationListener[]>()
private currentSource: NotificationEventSource | null = null
private reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS
private reconnectTimer: ReturnType<typeof setTimeout> | null = null
private closed = false
constructor(
private readonly url: string,
private readonly eventSourceFactory: EventSourceFactory,
private readonly timerApi: TimerApi,
) {
this.connect()
}
addEventListener(type: string, listener: NotificationListener) {
const current = this.listeners.get(type) ?? []
current.push(listener)
this.listeners.set(type, current)
}
close() {
this.closed = true
if (this.reconnectTimer) {
this.timerApi.clearTimeout(this.reconnectTimer)
this.reconnectTimer = null
}
this.currentSource?.close()
this.currentSource = null
}
private connect() {
if (this.closed) {
return
}
const source = this.eventSourceFactory(this.url)
this.currentSource = source
source.addEventListener('open', (event) => {
this.reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS
this.emit('open', event as MessageEvent)
})
source.addEventListener('notification', (event) => {
this.emit('notification', event as MessageEvent)
})
source.addEventListener('error', () => {
source.close()
if (this.closed || this.reconnectTimer) {
return
}
const delay = this.reconnectDelayMs
this.reconnectDelayMs = Math.min(this.reconnectDelayMs * 2, MAX_RECONNECT_DELAY_MS)
this.reconnectTimer = this.timerApi.setTimeout(() => {
this.reconnectTimer = null
this.connect()
}, delay)
})
}
private emit(type: string, event: MessageEvent) {
for (const listener of this.listeners.get(type) ?? []) {
listener(event)
}
}
}

19
web/src/features/notification/notification-unread-cache.test.ts
View file

@ -1,26 +1,9 @@
import { QueryClient } from '@tanstack/react-query'
import { describe, expect, it } from 'vitest'
import { decrementUnreadCount, incrementUnreadCount, resetUnreadCount } from './notification-unread-cache'
import { decrementUnreadCount, resetUnreadCount } from './notification-unread-cache'
import { NOTIFICATION_QUERY_KEYS } from './use-notifications'
describe('notification unread cache helpers', () => {
it('increments unread count from the existing cached value', () => {
const queryClient = new QueryClient()
queryClient.setQueryData(NOTIFICATION_QUERY_KEYS.unreadCount('user-a'), { count: 2 })
incrementUnreadCount(queryClient, 'user-a')
expect(queryClient.getQueryData(NOTIFICATION_QUERY_KEYS.unreadCount('user-a'))).toEqual({ count: 3 })
})
it('initializes unread count cache when incrementing without existing data', () => {
const queryClient = new QueryClient()
incrementUnreadCount(queryClient, 'user-a')
expect(queryClient.getQueryData(NOTIFICATION_QUERY_KEYS.unreadCount('user-a'))).toEqual({ count: 1 })
})
it('decrements unread count without going below zero', () => {
const queryClient = new QueryClient()
queryClient.setQueryData(NOTIFICATION_QUERY_KEYS.unreadCount('user-a'), { count: 1 })

7
web/src/features/notification/notification-unread-cache.ts
View file

@ -6,13 +6,6 @@ function normalizeUnreadCount(data: NotificationUnreadCount | undefined) {
return Math.max(data?.count ?? 0, 0)
}
export function incrementUnreadCount(queryClient: QueryClient, userId?: string | null) {
queryClient.setQueryData<NotificationUnreadCount>(
NOTIFICATION_QUERY_KEYS.unreadCount(userId),
(current) => ({ count: normalizeUnreadCount(current) + 1 })
)
}
export function decrementUnreadCount(queryClient: QueryClient, userId?: string | null) {
queryClient.setQueryData<NotificationUnreadCount>(
NOTIFICATION_QUERY_KEYS.unreadCount(userId),

77
web/src/features/notification/use-notification-sse.test.ts
View file

@ -1,77 +0,0 @@
import { QueryClient } from '@tanstack/react-query'
import { describe, expect, it, vi } from 'vitest'
import { attachNotificationSseListeners } from './use-notification-sse'
function createFakeConnection() {
const listeners = new Map<string, Array<(event: MessageEvent) => void>>()
return {
addEventListener(type: string, listener: (event: MessageEvent) => void) {
const current = listeners.get(type) ?? []
current.push(listener)
listeners.set(type, current)
},
close() {
// no-op for tests
},
emit(type: string) {
for (const listener of listeners.get(type) ?? []) {
listener(new MessageEvent(type))
}
},
}
}
describe('attachNotificationSseListeners', () => {
it('does not refetch unread count when the sse connection opens or reconnects', () => {
const queryClient = new QueryClient()
const invalidateQueries = vi.spyOn(queryClient, 'invalidateQueries')
const connection = createFakeConnection()
attachNotificationSseListeners(connection, queryClient, 'user-a')
connection.emit('open')
connection.emit('open')
expect(invalidateQueries).not.toHaveBeenCalledWith({
queryKey: ['notifications', 'user-a', 'unread-count'],
})
})
it('does not mutate the unread badge when the connection opens or reconnects', () => {
const queryClient = new QueryClient()
queryClient.setQueryData(['notifications', 'user-a', 'unread-count'], { count: 4 })
const connection = createFakeConnection()
attachNotificationSseListeners(connection, queryClient, 'user-a')
connection.emit('open')
connection.emit('open')
expect(queryClient.getQueryData(['notifications', 'user-a', 'unread-count'])).toEqual({ count: 4 })
})
it('increments unread count and invalidates notification list on new notification events', () => {
const queryClient = new QueryClient()
queryClient.setQueryData(['notifications', 'user-a', 'unread-count'], { count: 1 })
const invalidateQueries = vi.spyOn(queryClient, 'invalidateQueries')
const connection = createFakeConnection()
attachNotificationSseListeners(connection, queryClient, 'user-a')
connection.emit('notification')
expect(queryClient.getQueryData(['notifications', 'user-a', 'unread-count'])).toEqual({ count: 2 })
expect(invalidateQueries).toHaveBeenCalledWith({
queryKey: ['notifications', 'user-a', 'list'],
})
})
it('starts the unread badge from one when no cache exists yet', () => {
const queryClient = new QueryClient()
const connection = createFakeConnection()
attachNotificationSseListeners(connection, queryClient, 'user-a')
connection.emit('notification')
expect(queryClient.getQueryData(['notifications', 'user-a', 'unread-count'])).toEqual({ count: 1 })
})
})

50
web/src/features/notification/use-notification-sse.ts
View file

@ -1,50 +0,0 @@
import { useEffect, useRef } from 'react'
import type { QueryClient } from '@tanstack/react-query'
import { useQueryClient } from '@tanstack/react-query'
import { buildApiUrl, WEB_API_PREFIX } from '@/api/client'
import { incrementUnreadCount } from './notification-unread-cache'
import { createNotificationSseConnection } from './notification-sse-coordinator'
const SSE_URL = buildApiUrl(`${WEB_API_PREFIX}/notifications/sse`)
type NotificationSseConnectionLike = ReturnType<typeof createNotificationSseConnection>
export function attachNotificationSseListeners(
connection: NotificationSseConnectionLike,
queryClient: QueryClient,
userId: string,
) {
connection.addEventListener('open', () => {
// No unread-count sync here. The badge is hydrated once on page load and then
// updated locally from SSE events to avoid reconnect-driven request loops.
})
connection.addEventListener('notification', () => {
incrementUnreadCount(queryClient, userId)
void queryClient.invalidateQueries({ queryKey: ['notifications', userId, 'list'] })
})
}
/**
* Opens an SSE connection to the notification stream.
* On receiving a "notification" event, updates the local unread badge and invalidates
* the notification list. Reconnects no longer refetch unread-count to avoid turning
* SSE churn into near-polling traffic.
*/
export function useNotificationSse(userId?: string | null) {
const queryClient = useQueryClient()
const esRef = useRef<ReturnType<typeof createNotificationSseConnection> | null>(null)
useEffect(() => {
if (!userId) return
const es = createNotificationSseConnection(SSE_URL)
esRef.current = es
attachNotificationSseListeners(es, queryClient, userId)
return () => {
es.close()
esRef.current = null
}
}, [userId, queryClient])
}

177
web/src/features/notification/use-notifications.test.ts Normal file
View file

@ -0,0 +1,177 @@
// @vitest-environment jsdom
import { createElement, type ReactNode } from 'react'
import { focusManager, QueryClient, QueryClientProvider } from '@tanstack/react-query'
import { act, renderHook } from '@testing-library/react'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import type { NotificationItem, PagedResponse } from '@/api/types'
const { getUnreadCount } = vi.hoisted(() => ({
getUnreadCount: vi.fn().mockResolvedValue({ count: 1 }),
}))
vi.mock('@/api/client', () => ({
notificationApi: {
getUnreadCount,
},
}))
import {
getNotificationListQueryOptions,
getUnreadCountQueryOptions,
markAllCachedNotificationsRead,
markCachedNotificationRead,
removeCachedNotification,
useUnreadCount,
} from './use-notifications'
function createWrapper(queryClient: QueryClient) {
return function Wrapper({ children }: { children: ReactNode }) {
return createElement(QueryClientProvider, { client: queryClient }, children)
}
}
describe('useUnreadCount polling lifecycle', () => {
beforeEach(() => {
vi.useFakeTimers()
getUnreadCount.mockClear()
})
afterEach(() => {
focusManager.setFocused(undefined)
vi.useRealTimers()
})
it('polls every ten seconds and stops after logout', async () => {
const queryClient = new QueryClient({
defaultOptions: { queries: { retry: false, gcTime: Infinity } },
})
const { rerender, unmount } = renderHook(
({ userId }: { userId?: string }) => useUnreadCount(userId),
{ initialProps: { userId: 'user-a' as string | undefined }, wrapper: createWrapper(queryClient) },
)
await vi.waitFor(() => expect(getUnreadCount).toHaveBeenCalledTimes(1))
await act(() => vi.advanceTimersByTimeAsync(10_000))
await vi.waitFor(() => expect(getUnreadCount).toHaveBeenCalledTimes(2))
rerender({ userId: undefined })
await act(() => vi.advanceTimersByTimeAsync(20_000))
expect(getUnreadCount).toHaveBeenCalledTimes(2)
unmount()
queryClient.clear()
})
it('uses a separate cache key after the authenticated user changes', async () => {
const queryClient = new QueryClient({
defaultOptions: { queries: { retry: false, gcTime: Infinity } },
})
const { rerender, unmount } = renderHook(
({ userId }: { userId?: string }) => useUnreadCount(userId),
{ initialProps: { userId: 'user-a' as string | undefined }, wrapper: createWrapper(queryClient) },
)
await vi.waitFor(() => expect(getUnreadCount).toHaveBeenCalledTimes(1))
rerender({ userId: 'user-b' })
await vi.waitFor(() => expect(
queryClient.getQueryData(['notifications', 'user-b', 'unread-count']),
).toEqual({ count: 1 }))
expect(queryClient.getQueryData(['notifications', 'user-a', 'unread-count'])).toEqual({ count: 1 })
unmount()
queryClient.clear()
})
it('refreshes on focus and clears its timer after unmount', async () => {
const queryClient = new QueryClient({
defaultOptions: { queries: { retry: false, gcTime: Infinity } },
})
focusManager.setFocused(false)
const { unmount } = renderHook(() => useUnreadCount('user-a'), {
wrapper: createWrapper(queryClient),
})
await vi.waitFor(() => expect(getUnreadCount).toHaveBeenCalledTimes(1))
focusManager.setFocused(true)
await vi.waitFor(() => expect(getUnreadCount).toHaveBeenCalledTimes(2))
unmount()
await act(() => vi.advanceTimersByTimeAsync(20_000))
expect(getUnreadCount).toHaveBeenCalledTimes(2)
queryClient.clear()
})
})
describe('notification mutation cache updates', () => {
const userAKey = ['notifications', 'user-a', 'list', 0, 20] as const
const userBKey = ['notifications', 'user-b', 'list', 0, 20] as const
const page: PagedResponse<NotificationItem> = {
items: [
{ id: 1, category: 'REVIEW' as const, eventType: 'A', title: 'A', status: 'UNREAD' as const, createdAt: '2026-09-03T00:00:00Z' },
{ id: 2, category: 'REVIEW' as const, eventType: 'B', title: 'B', status: 'UNREAD' as const, createdAt: '2026-09-03T00:00:01Z' },
],
total: 2,
page: 0,
size: 20,
}
it('updates only the active user list immediately after mark-read operations', () => {
const queryClient = new QueryClient()
queryClient.setQueryData(userAKey, page)
queryClient.setQueryData(userBKey, page)
markCachedNotificationRead(queryClient, 'user-a', 1, '2026-09-03T01:00:00Z')
expect(queryClient.getQueryData<typeof page>(userAKey)?.items[0]).toMatchObject({
status: 'READ',
readAt: '2026-09-03T01:00:00Z',
})
expect(queryClient.getQueryData<typeof page>(userBKey)?.items[0].status).toBe('UNREAD')
markAllCachedNotificationsRead(queryClient, 'user-a', '2026-09-03T01:01:00Z')
expect(queryClient.getQueryData<typeof page>(userAKey)?.items.every((item) => item.status === 'READ')).toBe(true)
})
it('removes a deleted notification and adjusts the cached total', () => {
const queryClient = new QueryClient()
queryClient.setQueryData(userAKey, page)
removeCachedNotification(queryClient, 'user-a', 1)
expect(queryClient.getQueryData<typeof page>(userAKey)).toMatchObject({
items: [{ id: 2 }],
total: 1,
})
})
})
describe('getUnreadCountQueryOptions', () => {
it('polls the unread count over HTTP every ten seconds while the user is signed in', () => {
const options = getUnreadCountQueryOptions('user-a')
expect(options.queryKey).toEqual(['notifications', 'user-a', 'unread-count'])
expect(options.enabled).toBe(true)
expect(options.staleTime).toBe(0)
expect(options.refetchInterval).toBe(10_000)
expect(options.refetchOnWindowFocus).toBe(true)
})
it('does not poll before an authenticated user is available', () => {
const options = getUnreadCountQueryOptions(undefined)
expect(options.enabled).toBe(false)
})
})
describe('getNotificationListQueryOptions', () => {
it('polls an active notification list every ten seconds', () => {
const options = getNotificationListQueryOptions('user-a', 0, 20, 'REVIEW')
expect(options.queryKey).toEqual(['notifications', 'user-a', 'list', 0, 20, 'REVIEW'])
expect(options.enabled).toBe(true)
expect(options.staleTime).toBe(0)
expect(options.refetchInterval).toBe(10_000)
expect(options.refetchOnWindowFocus).toBe(true)
})
})

132
web/src/features/notification/use-notifications.ts
View file

@ -1,56 +1,115 @@
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
import { useMutation, useQuery, useQueryClient, type QueryClient } from '@tanstack/react-query'
import { notificationApi } from '@/api/client'
import type { NotificationItem, PagedResponse } from '@/api/types'
import { decrementUnreadCount, resetUnreadCount } from './notification-unread-cache'
import { getNotificationQueryKeyScope } from './notification-session'
export const NOTIFICATION_QUERY_KEYS = {
list: (userId?: string | null, page?: number, size?: number) => [...getNotificationQueryKeyScope(userId), 'list', page, size] as const,
list: (userId?: string | null, page?: number, size?: number, category?: string) => [
...getNotificationQueryKeyScope(userId),
'list',
page,
size,
...(category ? [category] : []),
] as const,
unreadCount: (userId?: string | null) => [...getNotificationQueryKeyScope(userId), 'unread-count'] as const,
listByCategory: (userId?: string | null, page?: number, size?: number, category?: string) =>
[...getNotificationQueryKeyScope(userId), 'list', page, size, category] as const,
}
/**
* Fetches paginated notification list.
*/
export function useNotifications(userId?: string | null, page = 0, size = 5) {
return useQuery({
queryKey: NOTIFICATION_QUERY_KEYS.list(userId, page, size),
queryFn: () => notificationApi.list({ page, size }) as Promise<PagedResponse<NotificationItem>>,
enabled: !!userId,
staleTime: Infinity,
refetchOnWindowFocus: false,
refetchOnReconnect: false,
const NOTIFICATION_POLL_INTERVAL_MS = 10_000
type NotificationPage = PagedResponse<NotificationItem>
function updateCachedLists(
queryClient: QueryClient,
userId: string | null | undefined,
update: (page: NotificationPage) => NotificationPage,
) {
queryClient.setQueriesData<NotificationPage>(
{ queryKey: [...getNotificationQueryKeyScope(userId), 'list'] },
(current) => current ? update(current) : current,
)
}
export function markCachedNotificationRead(
queryClient: QueryClient,
userId: string | null | undefined,
notificationId: number,
readAt = new Date().toISOString(),
) {
updateCachedLists(queryClient, userId, (page) => ({
...page,
items: page.items.map((item) => item.id === notificationId
? { ...item, status: 'READ', readAt: item.readAt ?? readAt }
: item),
}))
}
export function markAllCachedNotificationsRead(
queryClient: QueryClient,
userId: string | null | undefined,
readAt = new Date().toISOString(),
) {
updateCachedLists(queryClient, userId, (page) => ({
...page,
items: page.items.map((item) => item.status === 'READ'
? item
: { ...item, status: 'READ', readAt }),
}))
}
export function removeCachedNotification(
queryClient: QueryClient,
userId: string | null | undefined,
notificationId: number,
) {
updateCachedLists(queryClient, userId, (page) => {
const items = page.items.filter((item) => item.id !== notificationId)
return {
...page,
items,
total: page.total - (page.items.length - items.length),
}
})
}
export function getNotificationListQueryOptions(
userId?: string | null,
page = 0,
size = 20,
category?: string,
) {
return {
queryKey: NOTIFICATION_QUERY_KEYS.list(userId, page, size, category),
queryFn: () => notificationApi.list({ page, size, category }),
enabled: !!userId,
staleTime: 0,
refetchInterval: NOTIFICATION_POLL_INTERVAL_MS,
refetchOnWindowFocus: true,
}
}
export function getUnreadCountQueryOptions(userId?: string | null) {
return {
queryKey: NOTIFICATION_QUERY_KEYS.unreadCount(userId),
queryFn: () => notificationApi.getUnreadCount(),
enabled: !!userId,
staleTime: 0,
refetchInterval: NOTIFICATION_POLL_INTERVAL_MS,
refetchOnWindowFocus: true,
}
}
/**
* Fetches the current unread notification count for the badge.
*/
export function useUnreadCount(userId?: string | null) {
return useQuery({
queryKey: NOTIFICATION_QUERY_KEYS.unreadCount(userId),
queryFn: () => notificationApi.getUnreadCount(),
enabled: !!userId,
staleTime: Infinity,
refetchOnWindowFocus: false,
refetchOnReconnect: false,
})
return useQuery(getUnreadCountQueryOptions(userId))
}
/**
* Fetches paginated notification list with optional category filter.
*/
export function useNotificationList(userId?: string | null, page = 0, size = 20, category?: string) {
return useQuery({
queryKey: NOTIFICATION_QUERY_KEYS.listByCategory(userId, page, size, category),
queryFn: () => notificationApi.list({ page, size, category }) as Promise<PagedResponse<NotificationItem>>,
enabled: !!userId,
staleTime: Infinity,
refetchOnWindowFocus: false,
refetchOnReconnect: false,
})
return useQuery(getNotificationListQueryOptions(userId, page, size, category))
}
/**
@ -62,7 +121,8 @@ export function useMarkAllRead(userId?: string | null) {
mutationFn: () => notificationApi.markAllRead(),
onSuccess: () => {
resetUnreadCount(queryClient, userId)
void queryClient.invalidateQueries({ queryKey: ['notifications'] })
markAllCachedNotificationsRead(queryClient, userId)
void queryClient.invalidateQueries({ queryKey: getNotificationQueryKeyScope(userId) })
},
})
}
@ -74,9 +134,10 @@ export function useMarkRead(userId?: string | null) {
const queryClient = useQueryClient()
return useMutation({
mutationFn: (id: number) => notificationApi.markRead(id),
onSuccess: () => {
onSuccess: (_result, id) => {
decrementUnreadCount(queryClient, userId)
void queryClient.invalidateQueries({ queryKey: ['notifications'] })
markCachedNotificationRead(queryClient, userId, id)
void queryClient.invalidateQueries({ queryKey: getNotificationQueryKeyScope(userId) })
},
})
}
@ -85,7 +146,8 @@ export function useDeleteReadNotification(userId?: string | null) {
const queryClient = useQueryClient()
return useMutation({
mutationFn: (id: number) => notificationApi.deleteRead(id),
onSuccess: () => {
onSuccess: (_result, id) => {
removeCachedNotification(queryClient, userId, id)
void queryClient.invalidateQueries({ queryKey: getNotificationQueryKeyScope(userId) })
},
})