Bridge broker events onto SSE streams and replay history on reconnect

src/main/java/edu/stanford/protege/webprotege/gateway/SecurityConfig.java

@@ -7,6 +7,7 @@
 import org.springframework.security.config.annotation.web.builders.HttpSecurity;
 import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
 import org.springframework.security.config.annotation.web.configuration.WebSecurityCustomizer;
+import org.springframework.http.HttpMethod;
 import org.springframework.security.config.annotation.web.configurers.AbstractHttpConfigurer;
 import org.springframework.security.core.authority.SimpleGrantedAuthority;
 import org.springframework.security.core.authority.mapping.GrantedAuthoritiesMapper;
@@ -63,6 +64,8 @@ public SecurityFilterChain resourceServerFilterChain(HttpSecurity http) throws E
         http.csrf(AbstractHttpConfigurer::disable)
                 .authorizeHttpRequests(auth -> auth
                         .requestMatchers("/wsapps").permitAll()
+                        // The SSE controller enforces its own interim VIEW_PROJECT check (hardened by #305).
+                        .requestMatchers(HttpMethod.GET, "/data/projects/*/events").permitAll()
                         .anyRequest()
                         .authenticated()
                 );

src/main/java/edu/stanford/protege/webprotege/gateway/sse/HistoryReplaySseCatchUpService.java

@@ -0,0 +1,99 @@
+package edu.stanford.protege.webprotege.gateway.sse;
+
+import edu.stanford.protege.webprotege.common.ProjectId;
+import edu.stanford.protege.webprotege.event.EventTag;
+import edu.stanford.protege.webprotege.gateway.websocket.dto.ProjectEventsQueryRequest;
+import edu.stanford.protege.webprotege.gateway.websocket.dto.ProjectEventsQueryResponse;
+import edu.stanford.protege.webprotege.ipc.CommandExecutor;
+import edu.stanford.protege.webprotege.ipc.ExecutionContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import javax.annotation.Nullable;
+
+/**
+ * Replays the events a reconnecting client missed by querying the durable event-history service.
+ *
+ * <p>By the time this runs the registry is already buffering live events for the stream (the
+ * controller subscribes first). This service queries history for everything after {@code lastEventId},
+ * hands the whole batch back to the registry as one catch-up frame stamped with the batch's end
+ * sequence, and lets the registry flush the buffered live events — dropping any the replay already
+ * covered — before resuming live delivery. History flattening drops per-event ordinals, so a single
+ * batch replayed with a strictly-greater-than query and monotonic ids is what satisfies "nothing
+ * missed, nothing duplicated".
+ *
+ * <p>Failure modes never kill the stream, so the client's gap detection can recover the hole:
+ * a fresh connection ({@code null} id) has nothing to replay; a non-numeric id is treated as fresh
+ * but still releases the buffer; a failed or empty history query resumes live-only.
+ */
+@Component
+public class HistoryReplaySseCatchUpService implements SseCatchUpService {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(HistoryReplaySseCatchUpService.class);
+
+    /** Dedupe threshold that keeps every buffered event when there is nothing to replay against. */
+    private static final long REPLAY_NOTHING = -1L;
+
+    private final SseStreamRegistry registry;
+
+    private final CommandExecutor<ProjectEventsQueryRequest, ProjectEventsQueryResponse> eventsQueryExecutor;
+
+    public HistoryReplaySseCatchUpService(SseStreamRegistry registry,
+                                          CommandExecutor<ProjectEventsQueryRequest, ProjectEventsQueryResponse> eventsQueryExecutor) {
+        this.registry = registry;
+        this.eventsQueryExecutor = eventsQueryExecutor;
+    }
+
+    @Override
+    public void catchUp(ProjectId projectId,
+                        @Nullable String lastEventId,
+                        SseEmitter emitter,
+                        ExecutionContext executionContext) {
+        if (lastEventId == null) {
+            // Fresh connection: the registry never began buffering, live delivery is already active.
+            return;
+        }
+        Integer since = parseSequence(lastEventId);
+        if (since == null) {
+            // The stream is buffering (lastEventId was present) but the id is unusable: release the
+            // buffer with no replay so the stream resumes live instead of stalling.
+            LOGGER.warn("Ignoring non-numeric Last-Event-ID '{}' for project {}; resuming live-only", lastEventId, projectId.id());
+            registry.completeCatchUp(projectId, emitter, REPLAY_NOTHING, null);
+            return;
+        }
+        try {
+            ProjectEventsQueryResponse response = queryHistory(projectId, since, executionContext);
+            EventTag endTag = (response != null && response.events != null) ? response.events.endTag() : null;
+            if (endTag == null) {
+                LOGGER.warn("Empty history response for project {} since {}; resuming live-only", projectId.id(), since);
+                registry.completeCatchUp(projectId, emitter, since, null);
+                return;
+            }
+            registry.completeCatchUp(projectId, emitter, endTag.getOrdinal(), response);
+        } catch (Exception e) {
+            // Keep the stream alive; the client's gap detection recovers the missed window.
+            LOGGER.error("History replay failed for project {} since {}; resuming live-only", projectId.id(), since, e);
+            registry.completeCatchUp(projectId, emitter, since, null);
+        }
+    }
+
+    private ProjectEventsQueryResponse queryHistory(ProjectId projectId, int since, ExecutionContext executionContext) throws Exception {
+        ProjectEventsQueryRequest request = new ProjectEventsQueryRequest(projectId, EventTag.get(since));
+        return eventsQueryExecutor.execute(request, executionContext).get();
+    }
+
+    @Nullable
+    private static Integer parseSequence(String lastEventId) {
+        String trimmed = lastEventId.trim();
+        if (trimmed.isEmpty()) {
+            return null;
+        }
+        try {
+            return Integer.valueOf(trimmed);
+        } catch (NumberFormatException e) {
+            return null;
+        }
+    }
+}

src/main/java/edu/stanford/protege/webprotege/gateway/sse/ProjectEventsSseController.java

@@ -0,0 +1,97 @@
+package edu.stanford.protege.webprotege.gateway.sse;
+
+import edu.stanford.protege.webprotege.authorization.ProjectResource;
+import edu.stanford.protege.webprotege.authorization.Subject;
+import edu.stanford.protege.webprotege.common.ProjectId;
+import edu.stanford.protege.webprotege.common.UserId;
+import edu.stanford.protege.webprotege.gateway.websocket.AccessManager;
+import edu.stanford.protege.webprotege.gateway.websocket.dto.BuiltInCapability;
+import edu.stanford.protege.webprotege.ipc.ExecutionContext;
+import jakarta.servlet.http.HttpServletResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestHeader;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.server.ResponseStatusException;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import javax.annotation.Nullable;
+import java.util.UUID;
+
+/**
+ * Streams project-change events to a single viewer over a long-lived HTTP connection
+ * (server-sent events).
+ *
+ * <p>Interim authorization mirrors the STOMP {@code ProjectEventsInterceptor}: the caller passes
+ * {@code userId} and {@code token} query params and must hold {@code VIEW_PROJECT}. Identity
+ * resolution is isolated in {@link #resolveExecutionContext} so #305 can swap it for a short-lived
+ * redeemable ticket.
+ */
+@RestController
+public class ProjectEventsSseController {
+
+    static final String EVENTS_PATH = "/data/projects/{projectId}/events";
+
+    /** nginx-specific header that disables response buffering so events flush immediately. */
+    static final String X_ACCEL_BUFFERING = "X-Accel-Buffering";
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(ProjectEventsSseController.class);
+
+    private static final String PROJECT_ID = "projectId";
+
+    private final SseStreamRegistry registry;
+
+    private final SseCatchUpService catchUpService;
+
+    private final AccessManager accessManager;
+
+    public ProjectEventsSseController(SseStreamRegistry registry,
+                                      SseCatchUpService catchUpService,
+                                      AccessManager accessManager) {
+        this.registry = registry;
+        this.catchUpService = catchUpService;
+        this.accessManager = accessManager;
+    }
+
+    @GetMapping(path = EVENTS_PATH, produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    public SseEmitter streamEvents(@PathVariable(PROJECT_ID) ProjectId projectId,
+                                   @RequestParam("userId") String userId,
+                                   @RequestParam("token") String token,
+                                   @RequestParam(value = "lastEventId", required = false) String lastEventIdParam,
+                                   @RequestHeader(value = "Last-Event-ID", required = false) String lastEventIdHeader,
+                                   HttpServletResponse response) {
+        ExecutionContext executionContext = resolveExecutionContext(userId, token);
+        if (!accessManager.hasPermission(Subject.forUser(userId),
+                                         ProjectResource.forProject(projectId),
+                                         BuiltInCapability.VIEW_PROJECT,
+                                         executionContext)) {
+            LOGGER.info("Denied SSE stream: user {} lacks VIEW_PROJECT on project {}", userId, projectId.id());
+            throw new ResponseStatusException(HttpStatus.FORBIDDEN,
+                                              "User " + userId + " does not have access to project " + projectId.id());
+        }
+
+        // Browsers send Last-Event-ID only on automatic reconnects; fresh loads use the query param.
+        String lastEventId = (lastEventIdHeader != null) ? lastEventIdHeader : lastEventIdParam;
+
+        response.setHeader(HttpHeaders.CACHE_CONTROL, "no-cache");
+        response.setHeader(X_ACCEL_BUFFERING, "no");
+
+        SseEmitter emitter = registry.subscribe(projectId, lastEventId, executionContext);
+        catchUpService.catchUp(projectId, lastEventId, emitter, executionContext);
+        return emitter;
+    }
+
+    /**
+     * Resolve the identity/authorization context from the connection's credentials. #305 replaces
+     * this interim query-param scheme with a redeemable, project-scoped ticket.
+     */
+    private ExecutionContext resolveExecutionContext(String userId, @Nullable String token) {
+        return new ExecutionContext(UserId.valueOf(userId), token, UUID.randomUUID().toString());
+    }
+}

src/main/java/edu/stanford/protege/webprotege/gateway/sse/SseCatchUpService.java

@@ -0,0 +1,29 @@
+package edu.stanford.protege.webprotege.gateway.sse;
+
+import edu.stanford.protege.webprotege.common.ProjectId;
+import edu.stanford.protege.webprotege.ipc.ExecutionContext;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+/**
+ * Seam for replaying events a reconnecting client missed while it was disconnected.
+ *
+ * <p>The controller subscribes the emitter to live events first, then calls this service so the
+ * events missed since {@code lastEventId} can be replayed from the durable history ahead of the
+ * live events buffered in the meantime. See {@link HistoryReplaySseCatchUpService}.
+ */
+public interface SseCatchUpService {
+
+    /**
+     * Replay events that occurred after {@code lastEventId} onto {@code emitter}.
+     *
+     * @param projectId        the project whose events are streamed.
+     * @param lastEventId      the last event id the client already received, or {@code null} for a
+     *                         fresh connection with nothing to replay.
+     * @param emitter          the already-subscribed emitter to replay onto.
+     * @param executionContext the identity/authorization context resolved for this connection.
+     */
+    void catchUp(ProjectId projectId,
+                 String lastEventId,
+                 SseEmitter emitter,
+                 ExecutionContext executionContext);
+}

src/main/java/edu/stanford/protege/webprotege/gateway/sse/SseCommandExecutorConfiguration.java

@@ -0,0 +1,23 @@
+package edu.stanford.protege.webprotege.gateway.sse;
+
+import edu.stanford.protege.webprotege.gateway.websocket.dto.ProjectEventsQueryRequest;
+import edu.stanford.protege.webprotege.gateway.websocket.dto.ProjectEventsQueryResponse;
+import edu.stanford.protege.webprotege.ipc.CommandExecutor;
+import edu.stanford.protege.webprotege.ipc.impl.CommandExecutorImpl;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * Wires the command executor the SSE catch-up path uses to pull missed events from the durable
+ * event-history service on {@code webprotege.hierarchies.GetProjectEvents}. Declared the same way the
+ * rest of the platform declares {@link CommandExecutor} beans (see the ipc application's
+ * authorization-status executor).
+ */
+@Configuration
+public class SseCommandExecutorConfiguration {
+
+    @Bean
+    CommandExecutor<ProjectEventsQueryRequest, ProjectEventsQueryResponse> projectEventsQueryExecutor() {
+        return new CommandExecutorImpl<>(ProjectEventsQueryResponse.class);
+    }
+}

src/main/java/edu/stanford/protege/webprotege/gateway/sse/SseProperties.java

@@ -0,0 +1,51 @@
+package edu.stanford.protege.webprotege.gateway.sse;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+import java.time.Duration;
+
+/**
+ * Tunables for the server-sent events streaming endpoint.
+ *
+ * <p>The heartbeat interval must stay comfortably below the nginx {@code proxy_read_timeout}
+ * (60s by default) so idle streams are not dropped. The stream timeout closes long-lived
+ * connections so each reconnect becomes a fresh authorization checkpoint.
+ */
+@ConfigurationProperties(prefix = "webprotege.sse")
+public class SseProperties {
+
+    private Duration heartbeatInterval = Duration.ofSeconds(20);
+
+    private Duration streamTimeout = Duration.ofMinutes(30);
+
+    /**
+     * Cap on the live events a reconnecting stream buffers while its history replay is fetched.
+     * Overflow drops the excess and lets the client's gap detection recover; it only guards against
+     * unbounded growth if a history query stalls.
+     */
+    private int catchUpBufferLimit = 1000;
+
+    public Duration getHeartbeatInterval() {
+        return heartbeatInterval;
+    }
+
+    public void setHeartbeatInterval(Duration heartbeatInterval) {
+        this.heartbeatInterval = heartbeatInterval;
+    }
+
+    public Duration getStreamTimeout() {
+        return streamTimeout;
+    }
+
+    public void setStreamTimeout(Duration streamTimeout) {
+        this.streamTimeout = streamTimeout;
+    }
+
+    public int getCatchUpBufferLimit() {
+        return catchUpBufferLimit;
+    }
+
+    public void setCatchUpBufferLimit(int catchUpBufferLimit) {
+        this.catchUpBufferLimit = catchUpBufferLimit;
+    }
+}