Skip to content

Commit 05a96aa

Browse files
johardijohardi
authored
Serve project events over SSE with sequence tags, replay, and stream tickets (#50)
2 parents 95fafcc + f0e9240 commit 05a96aa

24 files changed

Lines changed: 2000 additions & 33 deletions

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import org.springframework.security.config.annotation.web.builders.HttpSecurity;
88
import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
99
import org.springframework.security.config.annotation.web.configuration.WebSecurityCustomizer;
10+
import org.springframework.http.HttpMethod;
1011
import org.springframework.security.config.annotation.web.configurers.AbstractHttpConfigurer;
1112
import org.springframework.security.core.authority.SimpleGrantedAuthority;
1213
import org.springframework.security.core.authority.mapping.GrantedAuthoritiesMapper;
@@ -63,6 +64,11 @@ public SecurityFilterChain resourceServerFilterChain(HttpSecurity http) throws E
6364
http.csrf(AbstractHttpConfigurer::disable)
6465
.authorizeHttpRequests(auth -> auth
6566
.requestMatchers("/wsapps").permitAll()
67+
// The SSE stream cannot be bearer-authenticated (EventSource sends no Authorization
68+
// header); it is guarded instead by the short-lived stream ticket that the controller
69+
// redeems and re-checks for VIEW_PROJECT on every connect. The ticket-issuing endpoint
70+
// (/data/events/ticket) is deliberately left to anyRequest().authenticated() below.
71+
.requestMatchers(HttpMethod.GET, "/data/projects/*/events").permitAll()
6672
.anyRequest()
6773
.authenticated()
6874
);

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

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
package edu.stanford.protege.webprotege.gateway.sse;
2+
3+
import edu.stanford.protege.webprotege.common.ProjectId;
4+
import edu.stanford.protege.webprotege.event.EventTag;
5+
import edu.stanford.protege.webprotege.gateway.websocket.dto.ProjectEventsQueryRequest;
6+
import edu.stanford.protege.webprotege.gateway.websocket.dto.ProjectEventsQueryResponse;
7+
import edu.stanford.protege.webprotege.ipc.CommandExecutor;
8+
import edu.stanford.protege.webprotege.ipc.ExecutionContext;
9+
import org.slf4j.Logger;
10+
import org.slf4j.LoggerFactory;
11+
import org.springframework.stereotype.Component;
12+
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
13+
14+
import javax.annotation.Nullable;
15+
16+
/**
17+
* Replays the events a reconnecting client missed by querying the durable event-history service.
18+
*
19+
* <p>By the time this runs the registry is already buffering live events for the stream (the
20+
* controller subscribes first). This service queries history for everything after {@code lastEventId},
21+
* hands the whole batch back to the registry as one catch-up frame stamped with the batch's end
22+
* sequence, and lets the registry flush the buffered live events — dropping any the replay already
23+
* covered — before resuming live delivery. History flattening drops per-event ordinals, so a single
24+
* batch replayed with a strictly-greater-than query and monotonic ids is what satisfies "nothing
25+
* missed, nothing duplicated".
26+
*
27+
* <p>Failure modes never kill the stream, so the client's gap detection can recover the hole:
28+
* a fresh connection ({@code null} id) has nothing to replay; a non-numeric id is treated as fresh
29+
* but still releases the buffer; a failed or empty history query resumes live-only.
30+
*/
31+
@Component
32+
public class HistoryReplaySseCatchUpService implements SseCatchUpService {
33+
34+
private static final Logger LOGGER = LoggerFactory.getLogger(HistoryReplaySseCatchUpService.class);
35+
36+
/** Dedupe threshold that keeps every buffered event when there is nothing to replay against. */
37+
private static final long REPLAY_NOTHING = -1L;
38+
39+
private final SseStreamRegistry registry;
40+
41+
private final CommandExecutor<ProjectEventsQueryRequest, ProjectEventsQueryResponse> eventsQueryExecutor;
42+
43+
public HistoryReplaySseCatchUpService(SseStreamRegistry registry,
44+
CommandExecutor<ProjectEventsQueryRequest, ProjectEventsQueryResponse> eventsQueryExecutor) {
45+
this.registry = registry;
46+
this.eventsQueryExecutor = eventsQueryExecutor;
47+
}
48+
49+
@Override
50+
public void catchUp(ProjectId projectId,
51+
@Nullable String lastEventId,
52+
SseEmitter emitter,
53+
ExecutionContext executionContext) {
54+
if (lastEventId == null) {
55+
// Fresh connection: the registry never began buffering, live delivery is already active.
56+
return;
57+
}
58+
Integer since = parseSequence(lastEventId);
59+
if (since == null) {
60+
// The stream is buffering (lastEventId was present) but the id is unusable: release the
61+
// buffer with no replay so the stream resumes live instead of stalling.
62+
LOGGER.warn("Ignoring non-numeric Last-Event-ID '{}' for project {}; resuming live-only", lastEventId, projectId.id());
63+
registry.completeCatchUp(projectId, emitter, REPLAY_NOTHING, null);
64+
return;
65+
}
66+
try {
67+
ProjectEventsQueryResponse response = queryHistory(projectId, since, executionContext);
68+
EventTag endTag = (response != null && response.events != null) ? response.events.endTag() : null;
69+
if (endTag == null) {
70+
LOGGER.warn("Empty history response for project {} since {}; resuming live-only", projectId.id(), since);
71+
registry.completeCatchUp(projectId, emitter, since, null);
72+
return;
73+
}
74+
registry.completeCatchUp(projectId, emitter, endTag.getOrdinal(), response);
75+
} catch (Exception e) {
76+
// Keep the stream alive; the client's gap detection recovers the missed window.
77+
LOGGER.error("History replay failed for project {} since {}; resuming live-only", projectId.id(), since, e);
78+
registry.completeCatchUp(projectId, emitter, since, null);
79+
}
80+
}
81+
82+
private ProjectEventsQueryResponse queryHistory(ProjectId projectId, int since, ExecutionContext executionContext) throws Exception {
83+
ProjectEventsQueryRequest request = new ProjectEventsQueryRequest(projectId, EventTag.get(since));
84+
return eventsQueryExecutor.execute(request, executionContext).get();
85+
}
86+
87+
@Nullable
88+
private static Integer parseSequence(String lastEventId) {
89+
String trimmed = lastEventId.trim();
90+
if (trimmed.isEmpty()) {
91+
return null;
92+
}
93+
try {
94+
return Integer.valueOf(trimmed);
95+
} catch (NumberFormatException e) {
96+
return null;
97+
}
98+
}
99+
}

‎src/main/java/edu/stanford/protege/webprotege/gateway/sse/InMemoryStreamTicketStore.java‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
package edu.stanford.protege.webprotege.gateway.sse;
2+
3+
import com.google.common.base.Ticker;
4+
import com.google.common.cache.Cache;
5+
import com.google.common.cache.CacheBuilder;
6+
import edu.stanford.protege.webprotege.common.ProjectId;
7+
import edu.stanford.protege.webprotege.common.UserId;
8+
import org.springframework.beans.factory.annotation.Autowired;
9+
import org.springframework.stereotype.Component;
10+
11+
import java.security.SecureRandom;
12+
import java.time.Duration;
13+
import java.time.Instant;
14+
import java.util.Base64;
15+
import java.util.Optional;
16+
import java.util.concurrent.TimeUnit;
17+
18+
/**
19+
* In-memory {@link StreamTicketStore} backed by a Guava cache that evicts each ticket a fixed duration
20+
* after it is written ({@code expireAfterWrite}). Ticket values are 128 bits of {@link SecureRandom}
21+
* rendered URL-safe, so they carry safely in a query string and are infeasible to guess.
22+
*
23+
* <p>Single-instance only — a ticket lives only in the heap of the gateway that issued it. See
24+
* {@link StreamTicketStore} and #307.
25+
*/
26+
@Component
27+
public class InMemoryStreamTicketStore implements StreamTicketStore {
28+
29+
/** 16 bytes = 128 bits of entropy, per the ticket brief. */
30+
private static final int TICKET_BYTES = 16;
31+
32+
private final SecureRandom secureRandom = new SecureRandom();
33+
34+
private final Base64.Encoder encoder = Base64.getUrlEncoder().withoutPadding();
35+
36+
private final Duration ttl;
37+
38+
private final Cache<String, StreamTicket> tickets;
39+
40+
@Autowired
41+
public InMemoryStreamTicketStore(SseProperties properties) {
42+
this(properties.getTicketTtl(), Ticker.systemTicker());
43+
}
44+
45+
/** Package-visible so tests can drive expiry with a fake ticker instead of sleeping out the TTL. */
46+
InMemoryStreamTicketStore(Duration ttl, Ticker ticker) {
47+
this.ttl = ttl;
48+
this.tickets = CacheBuilder.newBuilder()
49+
.expireAfterWrite(ttl.toMillis(), TimeUnit.MILLISECONDS)
50+
.ticker(ticker)
51+
.build();
52+
}
53+
54+
@Override
55+
public String issue(UserId userId, ProjectId projectId, String jwt) {
56+
String ticket = newTicketValue();
57+
tickets.put(ticket, new StreamTicket(userId, projectId, jwt, Instant.now().plus(ttl)));
58+
return ticket;
59+
}
60+
61+
@Override
62+
public Optional<StreamTicket> redeem(String ticket) {
63+
if (ticket == null || ticket.isEmpty()) {
64+
return Optional.empty();
65+
}
66+
return Optional.ofNullable(tickets.getIfPresent(ticket));
67+
}
68+
69+
private String newTicketValue() {
70+
byte[] bytes = new byte[TICKET_BYTES];
71+
secureRandom.nextBytes(bytes);
72+
return encoder.encodeToString(bytes);
73+
}
74+
}

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

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
package edu.stanford.protege.webprotege.gateway.sse;
2+
3+
import edu.stanford.protege.webprotege.common.ProjectId;
4+
import edu.stanford.protege.webprotege.ipc.ExecutionContext;
5+
import jakarta.servlet.http.HttpServletResponse;
6+
import org.springframework.http.HttpHeaders;
7+
import org.springframework.http.MediaType;
8+
import org.springframework.web.bind.annotation.GetMapping;
9+
import org.springframework.web.bind.annotation.PathVariable;
10+
import org.springframework.web.bind.annotation.RequestHeader;
11+
import org.springframework.web.bind.annotation.RequestParam;
12+
import org.springframework.web.bind.annotation.RestController;
13+
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
14+
15+
/**
16+
* Streams project-change events to a single viewer over a long-lived HTTP connection (server-sent events).
17+
*
18+
* <p>The connection is authorized by a short-lived, project-scoped {@code ticket} query parameter (issued by
19+
* {@link StreamTicketController}) rather than a bearer header, because {@code EventSource} cannot set custom
20+
* headers. Every connection — including each native auto-reconnect, which re-GETs the same URL — redeems the
21+
* ticket afresh via {@link StreamTicketService}, so {@code VIEW_PROJECT} is re-checked on every (re)connect.
22+
* Identity comes solely from the server-side ticket record; nothing the request carries is trusted for
23+
* authorization. Redemption yields 401 for a missing/unknown/expired/wrong-project ticket and 403 once the
24+
* ticket's identity has lost view access.
25+
*/
26+
@RestController
27+
public class ProjectEventsSseController {
28+
29+
static final String EVENTS_PATH = "/data/projects/{projectId}/events";
30+
31+
/** nginx-specific header that disables response buffering so events flush immediately. */
32+
static final String X_ACCEL_BUFFERING = "X-Accel-Buffering";
33+
34+
private static final String PROJECT_ID = "projectId";
35+
36+
private final SseStreamRegistry registry;
37+
38+
private final SseCatchUpService catchUpService;
39+
40+
private final StreamTicketService ticketService;
41+
42+
public ProjectEventsSseController(SseStreamRegistry registry,
43+
SseCatchUpService catchUpService,
44+
StreamTicketService ticketService) {
45+
this.registry = registry;
46+
this.catchUpService = catchUpService;
47+
this.ticketService = ticketService;
48+
}
49+
50+
@GetMapping(path = EVENTS_PATH, produces = MediaType.TEXT_EVENT_STREAM_VALUE)
51+
public SseEmitter streamEvents(@PathVariable(PROJECT_ID) ProjectId projectId,
52+
@RequestParam(value = "ticket", required = false) String ticket,
53+
@RequestParam(value = "lastEventId", required = false) String lastEventIdParam,
54+
@RequestHeader(value = "Last-Event-ID", required = false) String lastEventIdHeader,
55+
HttpServletResponse response) {
56+
// Redeem before touching the stream: throws 401/403 if the ticket is not a valid, still-authorized
57+
// pass for this exact project. The identity/authorization context comes only from the ticket record.
58+
ExecutionContext executionContext = ticketService.redeem(ticket, projectId);
59+
60+
// Browsers send Last-Event-ID only on automatic reconnects; fresh loads use the query param.
61+
String lastEventId = (lastEventIdHeader != null) ? lastEventIdHeader : lastEventIdParam;
62+
63+
response.setHeader(HttpHeaders.CACHE_CONTROL, "no-cache");
64+
response.setHeader(X_ACCEL_BUFFERING, "no");
65+
66+
SseEmitter emitter = registry.subscribe(projectId, lastEventId, executionContext);
67+
catchUpService.catchUp(projectId, lastEventId, emitter, executionContext);
68+
return emitter;
69+
}
70+
}

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

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
package edu.stanford.protege.webprotege.gateway.sse;
2+
3+
import edu.stanford.protege.webprotege.common.ProjectId;
4+
import edu.stanford.protege.webprotege.ipc.ExecutionContext;
5+
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
6+
7+
/**
8+
* Seam for replaying events a reconnecting client missed while it was disconnected.
9+
*
10+
* <p>The controller subscribes the emitter to live events first, then calls this service so the
11+
* events missed since {@code lastEventId} can be replayed from the durable history ahead of the
12+
* live events buffered in the meantime. See {@link HistoryReplaySseCatchUpService}.
13+
*/
14+
public interface SseCatchUpService {
15+
16+
/**
17+
* Replay events that occurred after {@code lastEventId} onto {@code emitter}.
18+
*
19+
* @param projectId the project whose events are streamed.
20+
* @param lastEventId the last event id the client already received, or {@code null} for a
21+
* fresh connection with nothing to replay.
22+
* @param emitter the already-subscribed emitter to replay onto.
23+
* @param executionContext the identity/authorization context resolved for this connection.
24+
*/
25+
void catchUp(ProjectId projectId,
26+
String lastEventId,
27+
SseEmitter emitter,
28+
ExecutionContext executionContext);
29+
}

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

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
package edu.stanford.protege.webprotege.gateway.sse;
2+
3+
import edu.stanford.protege.webprotege.gateway.websocket.dto.ProjectEventsQueryRequest;
4+
import edu.stanford.protege.webprotege.gateway.websocket.dto.ProjectEventsQueryResponse;
5+
import edu.stanford.protege.webprotege.ipc.CommandExecutor;
6+
import edu.stanford.protege.webprotege.ipc.impl.CommandExecutorImpl;
7+
import org.springframework.context.annotation.Bean;
8+
import org.springframework.context.annotation.Configuration;
9+
10+
/**
11+
* Wires the command executor the SSE catch-up path uses to pull missed events from the durable
12+
* event-history service on {@code webprotege.hierarchies.GetProjectEvents}. Declared the same way the
13+
* rest of the platform declares {@link CommandExecutor} beans (see the ipc application's
14+
* authorization-status executor).
15+
*/
16+
@Configuration
17+
public class SseCommandExecutorConfiguration {
18+
19+
@Bean
20+
CommandExecutor<ProjectEventsQueryRequest, ProjectEventsQueryResponse> projectEventsQueryExecutor() {
21+
return new CommandExecutorImpl<>(ProjectEventsQueryResponse.class);
22+
}
23+
}

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

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
package edu.stanford.protege.webprotege.gateway.sse;
2+
3+
import org.springframework.boot.context.properties.ConfigurationProperties;
4+
5+
import java.time.Duration;
6+
7+
/**
8+
* Tunables for the server-sent events streaming endpoint.
9+
*
10+
* <p>The heartbeat interval must stay comfortably below the nginx {@code proxy_read_timeout}
11+
* (60s by default) so idle streams are not dropped. The stream timeout closes long-lived
12+
* connections so each reconnect becomes a fresh authorization checkpoint.
13+
*/
14+
@ConfigurationProperties(prefix = "webprotege.sse")
15+
public class SseProperties {
16+
17+
private Duration heartbeatInterval = Duration.ofSeconds(20);
18+
19+
private Duration streamTimeout = Duration.ofMinutes(30);
20+
21+
/**
22+
* How long a stream ticket stays valid after it is issued. Short by design: a ticket travels in the
23+
* stream URL's query string (and so into nginx access logs and browser history), and the window bounds
24+
* that exposure. It must be long enough to cover a browser's native reconnect cadence, since the client
25+
* reuses the same ticket for the whole TTL. Configurable via {@code webprotege.sse.ticket-ttl} so tests
26+
* need not wait out the default.
27+
*/
28+
private Duration ticketTtl = Duration.ofSeconds(120);
29+
30+
/**
31+
* Cap on the live events a reconnecting stream buffers while its history replay is fetched.
32+
* Overflow drops the excess and lets the client's gap detection recover; it only guards against
33+
* unbounded growth if a history query stalls.
34+
*/
35+
private int catchUpBufferLimit = 1000;
36+
37+
public Duration getHeartbeatInterval() {
38+
return heartbeatInterval;
39+
}
40+
41+
public void setHeartbeatInterval(Duration heartbeatInterval) {
42+
this.heartbeatInterval = heartbeatInterval;
43+
}
44+
45+
public Duration getStreamTimeout() {
46+
return streamTimeout;
47+
}
48+
49+
public void setStreamTimeout(Duration streamTimeout) {
50+
this.streamTimeout = streamTimeout;
51+
}
52+
53+
public int getCatchUpBufferLimit() {
54+
return catchUpBufferLimit;
55+
}
56+
57+
public void setCatchUpBufferLimit(int catchUpBufferLimit) {
58+
this.catchUpBufferLimit = catchUpBufferLimit;
59+
}
60+
61+
public Duration getTicketTtl() {
62+
return ticketTtl;
63+
}
64+
65+
public void setTicketTtl(Duration ticketTtl) {
66+
this.ticketTtl = ticketTtl;
67+
}
68+
}

0 commit comments

Comments
 (0)