scriptoriumadmin/skillhub
1
0
Fork 0
mirror of https://github.com/iflytek/skillhub.git synced 2026-09-06 08:15:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

refactor(notification): replace SSE with HTTP polling

Browse source 
Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>
This commit is contained in:
XiaoSeS 2026-09-03 10:12:51 +08:00
parent fc7c59534a
commit efa3c1ae65
32 changed files with 182 additions and 1294 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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"));

17
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;
@ -38,11 +36,6 @@ public class RequestLoggingFilter extends OncePerRequestFilter {
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;
@ -99,16 +92,6 @@ public class RequestLoggingFilter extends OncePerRequestFilter {
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");

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

@ -17,7 +17,6 @@ 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;
@ -131,27 +130,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());
}
}
}

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])
}

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

@ -0,0 +1,32 @@
import { describe, expect, it } from 'vitest'
import { getNotificationListQueryOptions, getUnreadCountQueryOptions } from './use-notifications'
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)
})
})

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

@ -1,56 +1,60 @@
import { useMutation, useQuery, useQueryClient } 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>>,
const NOTIFICATION_POLL_INTERVAL_MS = 10_000
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: Infinity,
refetchOnWindowFocus: false,
refetchOnReconnect: false,
})
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))
}
/**