admin343/browser
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move game log processing into SCUM companion

Browse Source
This commit is contained in:
npc0-hue
2026-09-02 10:18:27 +08:00
parent 7d9c1b7e9e
commit 40ac46ba17
14 changed files with 328 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files
platform/api/log_events_handlers.go
+65 -23
View File
@@ -1,6 +1,7 @@
package api
import (
"context"
"encoding/json"
"fmt"
"net/http"
@@ -12,6 +13,8 @@ import (
"browser.local/platform/service"
)
type componentLogServerContextKey struct{}
const (
logEventHeartbeatInterval = 15 * time.Second
liveLogSourceClockSkew = 90 * time.Second
@@ -24,17 +27,13 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) {
writeMethodNotAllowed(w, http.MethodGet)
return
}
liveBoundary := time.Now().UTC().Add(-liveLogSourceClockSkew)
instance, streams, liveEligible, subscription, err := h.openLogEventSubscription(r)
if err != nil {
writeServiceError(w, err)
return
}
defer subscription.Close()
h.streamCurrentLogEvents(w, r, instance, streams, liveEligible, subscription)
}
func (h *coreHandlers) streamCurrentLogEvents(w http.ResponseWriter, r *http.Request, instance domain.ServerInstance, streams []domain.LogStream, liveEligible bool, subscription service.LogEventSubscription) {
liveBoundary := time.Now().UTC().Add(-liveLogSourceClockSkew)
flusher, ok := w.(http.Flusher)
if !ok {
writeServiceError(w, fmt.Errorf("streaming response unsupported"))
@@ -50,7 +49,11 @@ func (h *coreHandlers) streamCurrentLogEvents(w http.ResponseWriter, r *http.Req
active := supervisedLogSession{}
if liveEligible {
active = activeSupervisedLogSession(streams)
if isComponentLogRequest(r) {
active = activeComponentLogSession(streams)
} else {
active = activeSupervisedLogSession(streams)
}
}
emittedThrough, err := h.writeCurrentLogSession(w, instance.ID, active)
if err != nil {
@@ -74,6 +77,9 @@ func (h *coreHandlers) streamCurrentLogEvents(w http.ResponseWriter, r *http.Req
return
}
if subscriptionEvent.Kind == service.LogEventSubscriptionEventProcessState {
if isComponentLogRequest(r) {
continue
}
if subscriptionEvent.ServerInstanceID != instance.ID {
continue
}
@@ -90,7 +96,7 @@ func (h *coreHandlers) streamCurrentLogEvents(w http.ResponseWriter, r *http.Req
flusher.Flush()
continue
}
streams, liveEligible, err = h.loadLiveLogSnapshot(instance.ID)
streams, liveEligible, err = h.loadLiveLogSnapshot(instance.ID, isComponentLogRequest(r))
if err != nil {
return
}
@@ -114,8 +120,11 @@ func (h *coreHandlers) streamCurrentLogEvents(w http.ResponseWriter, r *http.Req
}
event := subscriptionEvent.LogEvent
candidate := activeSupervisedLogSession([]domain.LogStream{event.Stream})
if active.allStreams {
candidate = supervisedLogSession{}
}
if candidate.sessionID != "" && newerLogSession(candidate, active) {
streams, liveEligible, err = h.loadLiveLogSnapshot(instance.ID)
streams, liveEligible, err = h.loadLiveLogSnapshot(instance.ID, isComponentLogRequest(r))
if err != nil {
return
}
@@ -133,9 +142,15 @@ func (h *coreHandlers) streamCurrentLogEvents(w http.ResponseWriter, r *http.Req
}
}
if !active.contains(event.Stream) {
continue
if !active.allStreams || event.Stream.ServerInstanceID != instance.ID {
continue
}
active.streams = append(active.streams, event.Stream)
if err := writeSSEJSON(w, "stream", "", dto.LogStreamFromDomain(event.Stream)); err != nil {
return
}
}
if !subscriptionEvent.Live && !sourceLogEntryIsLive(event.Entry, liveBoundary) {
if !sourceLogEntryIsLive(event.Entry, liveBoundary) {
if event.Entry.Seq > emittedThrough[event.Stream.ID] {
emittedThrough[event.Stream.ID] = event.Entry.Seq
}
@@ -147,10 +162,7 @@ func (h *coreHandlers) streamCurrentLogEvents(w http.ResponseWriter, r *http.Req
return
}
}
// Live relay traffic is already a current best-effort observation.
// Do not sequence-gate it: a restarted Run intentionally starts with
// fresh in-memory sequence state and must still reach this subscriber.
if !subscriptionEvent.Live && event.Entry.Seq <= emittedThrough[event.Stream.ID] {
if event.Entry.Seq <= emittedThrough[event.Stream.ID] {
continue
}
if err := writeSSEJSON(w, "log", logEventID(event), dto.LogStreamEventFromDomain(event)); err != nil {
@@ -172,9 +184,14 @@ func sourceLogEntryIsLive(entry domain.LogEntry, liveBoundary time.Time) bool {
}
type supervisedLogSession struct {
sessionID string
startedAt time.Time
streams []domain.LogStream
sessionID string
startedAt time.Time
streams []domain.LogStream
allStreams bool
}
func activeComponentLogSession(streams []domain.LogStream) supervisedLogSession {
return supervisedLogSession{sessionID: "component", streams: append([]domain.LogStream(nil), streams...), allStreams: true}
}
func activeSupervisedLogSession(streams []domain.LogStream) supervisedLogSession {
@@ -217,6 +234,9 @@ func sameSupervisedLogSession(left supervisedLogSession, right supervisedLogSess
}
func (session supervisedLogSession) contains(stream domain.LogStream) bool {
if session.allStreams {
return session.hasStream(stream.ID)
}
return session.sessionID != "" && stream.Source == domain.LogStreamSourceProcess && stream.LogSessionID == session.sessionID && stream.SessionStartedAt.Equal(session.startedAt)
}
@@ -235,9 +255,7 @@ func (h *coreHandlers) writeCurrentLogSession(w http.ResponseWriter, serverInsta
return nil, err
}
for _, stream := range active.streams {
// Stream metadata identifies the current channel only. The platform
// never replays its old body when a live subscriber connects.
emittedThrough[stream.ID] = 0
emittedThrough[stream.ID] = stream.LatestSeq
if err := writeSSEJSON(w, "stream", "", dto.LogStreamFromDomain(stream)); err != nil {
return nil, err
}
@@ -251,7 +269,12 @@ func (h *coreHandlers) openLogEventSubscription(r *http.Request) (domain.ServerI
var liveEligible bool
var subscription service.LogEventSubscription
var err error
if h.enforceAuthorization {
if serverInstanceID, ok := r.Context().Value(componentLogServerContextKey{}).(string); ok && strings.TrimSpace(serverInstanceID) != "" {
instance, err = h.core.GetServerInstance(serverInstanceID)
if err == nil {
subscription, err = h.core.SubscribeLogEvents(instance.ID)
}
} else if h.enforceAuthorization {
sessionID := bearerToken(r)
instance, err = h.core.GetServerInstanceForSession(sessionID, r.PathValue("id"))
if err != nil {
@@ -272,7 +295,7 @@ func (h *coreHandlers) openLogEventSubscription(r *http.Request) (domain.ServerI
}
}
if err == nil {
streams, liveEligible, err = h.loadLiveLogSnapshot(instance.ID)
streams, liveEligible, err = h.loadLiveLogSnapshot(instance.ID, isComponentLogRequest(r))
}
if err != nil && subscription.Close != nil {
subscription.Close()
@@ -280,7 +303,16 @@ func (h *coreHandlers) openLogEventSubscription(r *http.Request) (domain.ServerI
return instance, streams, liveEligible, subscription, err
}
func (h *coreHandlers) loadLiveLogSnapshot(serverInstanceID string) ([]domain.LogStream, bool, error) {
func withComponentLogServer(r *http.Request, serverInstanceID string) *http.Request {
return r.WithContext(context.WithValue(r.Context(), componentLogServerContextKey{}, serverInstanceID))
}
func isComponentLogRequest(r *http.Request) bool {
_, ok := r.Context().Value(componentLogServerContextKey{}).(string)
return ok
}
func (h *coreHandlers) loadLiveLogSnapshot(serverInstanceID string, includeDeclaredStreams bool) ([]domain.LogStream, bool, error) {
instance, err := h.core.GetServerInstance(serverInstanceID)
if err != nil {
return nil, false, err
@@ -305,6 +337,16 @@ func (h *coreHandlers) loadLiveLogSnapshot(serverInstanceID string) ([]domain.Lo
}
current := make([]domain.LogStream, 0, len(streams))
for _, stream := range streams {
if includeDeclaredStreams {
if stream.LogSessionID != "" && stream.LogSessionID != logSessionID {
continue
}
if stream.Source != domain.LogStreamSourceProcess && stream.Source != domain.LogStreamSourceFile && stream.Source != domain.LogStreamSourceManagementProgram {
continue
}
current = append(current, stream)
continue
}
if stream.Source == domain.LogStreamSourceProcess && stream.LogSessionID == logSessionID {
current = append(current, stream)
}