From efa3c1ae65892e7f32731068b552d4c90e7fc2f1 Mon Sep 17 00:00:00 2001 From: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com> Date: Thu, 3 Sep 2026 10:12:51 +0800 Subject: [PATCH] refactor(notification): replace SSE with HTTP polling Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com> --- .../2026-03-19-notification-system-design.md | 48 ++-- .../config/WebMvcRateLimitConfig.java | 9 - .../portal/NotificationController.java | 11 - .../exception/GlobalExceptionHandler.java | 6 - .../skillhub/filter/RequestLoggingFilter.java | 17 -- .../listener/NotificationEventListener.java | 32 +-- .../config/WebMvcRateLimitConfigTest.java | 28 -- ...llReviewModerationFlowIntegrationTest.java | 4 +- .../portal/NotificationControllerTest.java | 6 +- .../PromotionApprovalFlowIntegrationTest.java | 4 +- .../exception/GlobalExceptionHandlerTest.java | 13 +- .../filter/RequestLoggingFilterTest.java | 22 -- .../NotificationEventListenerTest.java | 64 ++--- .../SubscriberNotificationSinkTest.java | 62 +---- server/skillhub-notification/pom.xml | 4 - .../service/NotificationDispatcher.java | 60 ---- .../service/NotificationService.java | 17 +- .../notification/sse/SseEmitterManager.java | 136 ---------- .../service/NotificationDispatcherTest.java | 128 --------- .../service/NotificationServiceTest.java | 26 +- .../sse/SseEmitterManagerTest.java | 256 ------------------ web/src/api/generated/schema.d.ts | 76 ------ .../notification/notification-bell.tsx | 5 +- .../notification/notification-dropdown.tsx | 4 +- .../notification-sse-coordinator.test.ts | 85 ------ .../notification-sse-coordinator.ts | 102 ------- .../notification-unread-cache.test.ts | 19 +- .../notification/notification-unread-cache.ts | 7 - .../notification/use-notification-sse.test.ts | 77 ------ .../notification/use-notification-sse.ts | 50 ---- .../notification/use-notifications.test.ts | 32 +++ .../notification/use-notifications.ts | 66 ++--- 32 files changed, 182 insertions(+), 1294 deletions(-) delete mode 100644 server/skillhub-app/src/test/java/com/iflytek/skillhub/config/WebMvcRateLimitConfigTest.java delete mode 100644 server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/service/NotificationDispatcher.java delete mode 100644 server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/sse/SseEmitterManager.java delete mode 100644 server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/service/NotificationDispatcherTest.java delete mode 100644 server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/sse/SseEmitterManagerTest.java delete mode 100644 web/src/features/notification/notification-sse-coordinator.test.ts delete mode 100644 web/src/features/notification/notification-sse-coordinator.ts delete mode 100644 web/src/features/notification/use-notification-sse.test.ts delete mode 100644 web/src/features/notification/use-notification-sse.ts create mode 100644 web/src/features/notification/use-notifications.test.ts diff --git a/docs/superpowers/specs/2026-03-19-notification-system-design.md b/docs/superpowers/specs/2026-03-19-notification-system-design.md index 33b1527a..d85ab977 100644 --- a/docs/superpowers/specs/2026-03-19-notification-system-design.md +++ b/docs/superpowers/specs/2026-03-19-notification-system-design.md @@ -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>` (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 diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/WebMvcRateLimitConfig.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/WebMvcRateLimitConfig.java index 431b46a8..0a7bfd93 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/WebMvcRateLimitConfig.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/WebMvcRateLimitConfig.java @@ -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()); - } } diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/controller/portal/NotificationController.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/controller/portal/NotificationController.java index ec62ebb9..cb8f7d4c 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/controller/portal/NotificationController.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/controller/portal/NotificationController.java @@ -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( diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/exception/GlobalExceptionHandler.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/exception/GlobalExceptionHandler.java index ba93bb1d..60a31773 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/exception/GlobalExceptionHandler.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/exception/GlobalExceptionHandler.java @@ -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")); diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/filter/RequestLoggingFilter.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/filter/RequestLoggingFilter.java index 1de3809b..3b78155d 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/filter/RequestLoggingFilter.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/filter/RequestLoggingFilter.java @@ -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; diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/listener/NotificationEventListener.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/listener/NotificationEventListener.java index 6ba40f6e..7e86fb6a 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/listener/NotificationEventListener.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/listener/NotificationEventListener.java @@ -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 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 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 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 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 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()); }); } diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/config/WebMvcRateLimitConfigTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/config/WebMvcRateLimitConfigTest.java deleted file mode 100644 index daf845c8..00000000 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/config/WebMvcRateLimitConfigTest.java +++ /dev/null @@ -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(); - } - } -} diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/admin/SkillReviewModerationFlowIntegrationTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/admin/SkillReviewModerationFlowIntegrationTest.java index c93e07a2..6663a8a8 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/admin/SkillReviewModerationFlowIntegrationTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/admin/SkillReviewModerationFlowIntegrationTest.java @@ -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 { diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/portal/NotificationControllerTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/portal/NotificationControllerTest.java index f44d2941..90e195bd 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/portal/NotificationControllerTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/portal/NotificationControllerTest.java @@ -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 diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/portal/PromotionApprovalFlowIntegrationTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/portal/PromotionApprovalFlowIntegrationTest.java index ca543a37..7d9249b1 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/portal/PromotionApprovalFlowIntegrationTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/controller/portal/PromotionApprovalFlowIntegrationTest.java @@ -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; diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/exception/GlobalExceptionHandlerTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/exception/GlobalExceptionHandlerTest.java index 82b2a3b6..f469638f 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/exception/GlobalExceptionHandlerTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/exception/GlobalExceptionHandlerTest.java @@ -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"); diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/filter/RequestLoggingFilterTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/filter/RequestLoggingFilterTest.java index 18e6f88a..871e0a45 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/filter/RequestLoggingFilterTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/filter/RequestLoggingFilterTest.java @@ -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 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(); diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/listener/NotificationEventListenerTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/listener/NotificationEventListenerTest.java index 80304261..67028efa 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/listener/NotificationEventListenerTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/listener/NotificationEventListenerTest.java @@ -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); } } diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/listener/SubscriberNotificationSinkTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/listener/SubscriberNotificationSinkTest.java index 93385750..015aad8b 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/listener/SubscriberNotificationSinkTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/listener/SubscriberNotificationSinkTest.java @@ -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 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 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 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 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 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> 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 { diff --git a/server/skillhub-notification/pom.xml b/server/skillhub-notification/pom.xml index 5c01d1ba..f62e3a8a 100644 --- a/server/skillhub-notification/pom.xml +++ b/server/skillhub-notification/pom.xml @@ -14,10 +14,6 @@ com.iflytek.skillhub skillhub-domain - - org.springframework.boot - spring-boot-starter-web - org.springframework.boot spring-boot-starter-data-jpa diff --git a/server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/service/NotificationDispatcher.java b/server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/service/NotificationDispatcher.java deleted file mode 100644 index e162dbd9..00000000 --- a/server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/service/NotificationDispatcher.java +++ /dev/null @@ -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 - } - } -} diff --git a/server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/service/NotificationService.java b/server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/service/NotificationService.java index e0a72837..62747778 100644 --- a/server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/service/NotificationService.java +++ b/server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/service/NotificationService.java @@ -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) diff --git a/server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/sse/SseEmitterManager.java b/server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/sse/SseEmitterManager.java deleted file mode 100644 index 341a4d50..00000000 --- a/server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/sse/SseEmitterManager.java +++ /dev/null @@ -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> emitters = new ConcurrentHashMap<>(); - private final AtomicInteger totalCount = new AtomicInteger(0); - private final Function emitterFactory; - - public SseEmitterManager() { - this(userId -> new SseEmitter(SSE_TIMEOUT)); - } - - SseEmitterManager(Function emitterFactory) { - this.emitterFactory = emitterFactory; - } - - public SseEmitter register(String userId) { - if (totalCount.get() >= MAX_TOTAL_EMITTERS) { - throw new IllegalStateException("SSE connection limit reached"); - } - - CopyOnWriteArrayList 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 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 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); - } - } -} diff --git a/server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/service/NotificationDispatcherTest.java b/server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/service/NotificationDispatcherTest.java deleted file mode 100644 index db6af43b..00000000 --- a/server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/service/NotificationDispatcherTest.java +++ /dev/null @@ -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> 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); - } -} diff --git a/server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/service/NotificationServiceTest.java b/server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/service/NotificationServiceTest.java index e92179cb..7f4eb10c 100644 --- a/server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/service/NotificationServiceTest.java +++ b/server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/service/NotificationServiceTest.java @@ -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)) diff --git a/server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/sse/SseEmitterManagerTest.java b/server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/sse/SseEmitterManagerTest.java deleted file mode 100644 index 43646817..00000000 --- a/server/skillhub-notification/src/test/java/com/iflytek/skillhub/notification/sse/SseEmitterManagerTest.java +++ /dev/null @@ -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 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 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 errorCallback = error -> {}; - private String userId; - private boolean failAfterConnected; - private boolean throwOnComplete; - private int sendCount; - private boolean completed; - private final List> 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 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 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()); - } - } -} diff --git a/web/src/api/generated/schema.d.ts b/web/src/api/generated/schema.d.ts index 8fbd4236..fc2e53e9 100644 --- a/web/src/api/generated/schema.d.ts +++ b/web/src/api/generated/schema.d.ts @@ -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?: { diff --git a/web/src/features/notification/notification-bell.tsx b/web/src/features/notification/notification-bell.tsx index 1299d0f0..b62b7a13 100644 --- a/web/src/features/notification/notification-bell.tsx +++ b/web/src/features/notification/notification-bell.tsx @@ -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 diff --git a/web/src/features/notification/notification-dropdown.tsx b/web/src/features/notification/notification-dropdown.tsx index aaf0780d..5276bbb4 100644 --- a/web/src/features/notification/notification-dropdown.tsx +++ b/web/src/features/notification/notification-dropdown.tsx @@ -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) diff --git a/web/src/features/notification/notification-sse-coordinator.test.ts b/web/src/features/notification/notification-sse-coordinator.test.ts deleted file mode 100644 index faa28062..00000000 --- a/web/src/features/notification/notification-sse-coordinator.test.ts +++ /dev/null @@ -1,85 +0,0 @@ -import { describe, expect, it, vi } from 'vitest' -import { createNotificationSseConnection } from './notification-sse-coordinator' - -class FakeEventSource { - listeners = new Map 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() - }) -}) diff --git a/web/src/features/notification/notification-sse-coordinator.ts b/web/src/features/notification/notification-sse-coordinator.ts deleted file mode 100644 index 52000a25..00000000 --- a/web/src/features/notification/notification-sse-coordinator.ts +++ /dev/null @@ -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() - private currentSource: NotificationEventSource | null = null - private reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS - private reconnectTimer: ReturnType | 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) - } - } -} diff --git a/web/src/features/notification/notification-unread-cache.test.ts b/web/src/features/notification/notification-unread-cache.test.ts index 093f7fac..2e38c03c 100644 --- a/web/src/features/notification/notification-unread-cache.test.ts +++ b/web/src/features/notification/notification-unread-cache.test.ts @@ -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 }) diff --git a/web/src/features/notification/notification-unread-cache.ts b/web/src/features/notification/notification-unread-cache.ts index 23ca2612..9a23784b 100644 --- a/web/src/features/notification/notification-unread-cache.ts +++ b/web/src/features/notification/notification-unread-cache.ts @@ -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( - NOTIFICATION_QUERY_KEYS.unreadCount(userId), - (current) => ({ count: normalizeUnreadCount(current) + 1 }) - ) -} - export function decrementUnreadCount(queryClient: QueryClient, userId?: string | null) { queryClient.setQueryData( NOTIFICATION_QUERY_KEYS.unreadCount(userId), diff --git a/web/src/features/notification/use-notification-sse.test.ts b/web/src/features/notification/use-notification-sse.test.ts deleted file mode 100644 index 3b3e8213..00000000 --- a/web/src/features/notification/use-notification-sse.test.ts +++ /dev/null @@ -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 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 }) - }) -}) diff --git a/web/src/features/notification/use-notification-sse.ts b/web/src/features/notification/use-notification-sse.ts deleted file mode 100644 index 5325651c..00000000 --- a/web/src/features/notification/use-notification-sse.ts +++ /dev/null @@ -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 - -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 | 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]) -} diff --git a/web/src/features/notification/use-notifications.test.ts b/web/src/features/notification/use-notifications.test.ts new file mode 100644 index 00000000..9acb1ea6 --- /dev/null +++ b/web/src/features/notification/use-notifications.test.ts @@ -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) + }) +}) diff --git a/web/src/features/notification/use-notifications.ts b/web/src/features/notification/use-notifications.ts index 491d4f7c..05c5fa74 100644 --- a/web/src/features/notification/use-notifications.ts +++ b/web/src/features/notification/use-notifications.ts @@ -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>, +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>, - enabled: !!userId, - staleTime: Infinity, - refetchOnWindowFocus: false, - refetchOnReconnect: false, - }) + return useQuery(getNotificationListQueryOptions(userId, page, size, category)) } /**