browser/platform/api/log_events_handlers.go

package api

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strings"
	"time"

	"browser.local/platform/domain"
	"browser.local/platform/dto"
	"browser.local/platform/service"
)

type componentLogServerContextKey struct{}

const (
	logEventHeartbeatInterval = 15 * time.Second
	liveLogSourceClockSkew    = 90 * time.Second
	managedLogSessionIDPrefix = "log-session:"
)

// serverLogEvents streams platform-accepted live append events for the terminal drawer.
func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		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()
	flusher, ok := w.(http.Flusher)
	if !ok {
		writeServiceError(w, fmt.Errorf("streaming response unsupported"))
		return
	}

	header := w.Header()
	header.Set("Content-Type", "text/event-stream")
	header.Set("Cache-Control", "no-cache, no-transform")
	header.Set("Connection", "keep-alive")
	header.Set("X-Accel-Buffering", "no")
	w.WriteHeader(http.StatusOK)

	active := supervisedLogSession{}
	if liveEligible {
		if isComponentLogRequest(r) {
			active = activeComponentLogSession(streams)
		} else {
			active = activeSupervisedLogSession(streams)
		}
	}
	emittedThrough, err := h.writeCurrentLogSession(w, instance.ID, active)
	if err != nil {
		_ = writeSSEJSON(w, "error", "", map[string]string{"message": err.Error()})
		flusher.Flush()
		return
	}
	if err := writeSSEJSON(w, "ready", "", dto.LogStreamEventsReadyResponse{ServerInstanceID: instance.ID, StreamCount: len(active.streams), ServerTime: time.Now().UTC()}); err != nil {
		return
	}
	flusher.Flush()

	heartbeat := time.NewTicker(logEventHeartbeatInterval)
	defer heartbeat.Stop()
	for {
		select {
		case <-r.Context().Done():
			return
		case subscriptionEvent, ok := <-subscription.Events:
			if !ok {
				return
			}
			if subscriptionEvent.Kind == service.LogEventSubscriptionEventProcessState {
				if isComponentLogRequest(r) {
					continue
				}
				if subscriptionEvent.ServerInstanceID != instance.ID {
					continue
				}
				if subscriptionEvent.ProcessState != domain.ServerInstanceStateRunning {
					liveEligible = false
					if active.sessionID == "" {
						continue
					}
					active = supervisedLogSession{}
					emittedThrough, err = h.writeCurrentLogSession(w, instance.ID, active)
					if err != nil {
						return
					}
					flusher.Flush()
					continue
				}
				streams, liveEligible, err = h.loadLiveLogSnapshot(instance.ID, isComponentLogRequest(r))
				if err != nil {
					return
				}
				next := supervisedLogSession{}
				if liveEligible {
					next = activeSupervisedLogSession(streams)
				}
				if sameSupervisedLogSession(active, next) {
					continue
				}
				active = next
				emittedThrough, err = h.writeCurrentLogSession(w, instance.ID, active)
				if err != nil {
					return
				}
				flusher.Flush()
				continue
			}
			if subscriptionEvent.Kind != service.LogEventSubscriptionEventLog || !liveEligible {
				continue
			}
			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, isComponentLogRequest(r))
				if err != nil {
					return
				}
				next := supervisedLogSession{}
				if liveEligible {
					next = activeSupervisedLogSession(streams)
				}
				if !sameSupervisedLogSession(active, next) {
					active = next
					emittedThrough, err = h.writeCurrentLogSession(w, instance.ID, active)
					if err != nil {
						return
					}
					flusher.Flush()
				}
			}
			if !active.contains(event.Stream) {
				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 !sourceLogEntryIsLive(event.Entry, liveBoundary) {
				if event.Entry.Seq > emittedThrough[event.Stream.ID] {
					emittedThrough[event.Stream.ID] = event.Entry.Seq
				}
				continue
			}
			if !active.hasStream(event.Stream.ID) {
				active.streams = append(active.streams, event.Stream)
				if err := writeSSEJSON(w, "stream", "", dto.LogStreamFromDomain(event.Stream)); err != nil {
					return
				}
			}
			if event.Entry.Seq <= emittedThrough[event.Stream.ID] {
				continue
			}
			if err := writeSSEJSON(w, "log", logEventID(event), dto.LogStreamEventFromDomain(event)); err != nil {
				return
			}
			emittedThrough[event.Stream.ID] = event.Entry.Seq
			flusher.Flush()
		case <-heartbeat.C:
			if _, err := fmt.Fprintf(w, ": heartbeat %s\n\n", time.Now().UTC().Format(time.RFC3339)); err != nil {
				return
			}
			flusher.Flush()
		}
	}
}

func sourceLogEntryIsLive(entry domain.LogEntry, liveBoundary time.Time) bool {
	return !entry.Timestamp.IsZero() && !entry.Timestamp.Before(liveBoundary)
}

type supervisedLogSession struct {
	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 {
	active := supervisedLogSession{}
	for _, stream := range streams {
		if stream.Source != domain.LogStreamSourceProcess || strings.TrimSpace(stream.LogSessionID) == "" || stream.SessionStartedAt.IsZero() {
			continue
		}
		candidate := supervisedLogSession{sessionID: stream.LogSessionID, startedAt: stream.SessionStartedAt}
		if active.sessionID == "" || newerLogSession(candidate, active) {
			active = candidate
		}
	}
	if active.sessionID == "" {
		return active
	}
	for _, stream := range streams {
		if stream.Source == domain.LogStreamSourceProcess && stream.LogSessionID == active.sessionID && stream.SessionStartedAt.Equal(active.startedAt) {
			active.streams = append(active.streams, stream)
		}
	}
	return active
}

func newerLogSession(candidate supervisedLogSession, current supervisedLogSession) bool {
	if candidate.sessionID == "" || candidate.sessionID == current.sessionID {
		return false
	}
	if current.sessionID == "" {
		return true
	}
	if !candidate.startedAt.Equal(current.startedAt) {
		return candidate.startedAt.After(current.startedAt)
	}
	return candidate.sessionID > current.sessionID
}

func sameSupervisedLogSession(left supervisedLogSession, right supervisedLogSession) bool {
	return left.sessionID == right.sessionID && left.startedAt.Equal(right.startedAt)
}

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

func (session supervisedLogSession) hasStream(streamID string) bool {
	for _, stream := range session.streams {
		if stream.ID == streamID {
			return true
		}
	}
	return false
}

func (h *coreHandlers) writeCurrentLogSession(w http.ResponseWriter, serverInstanceID string, active supervisedLogSession) (map[string]uint64, error) {
	emittedThrough := make(map[string]uint64, len(active.streams))
	if err := writeSSEJSON(w, "session", "", dto.LogStreamEventsSessionResponse{ServerInstanceID: serverInstanceID, LogSessionID: active.sessionID, SessionStartedAt: active.startedAt, StreamCount: len(active.streams), ServerTime: time.Now().UTC()}); err != nil {
		return nil, err
	}
	for _, stream := range active.streams {
		emittedThrough[stream.ID] = stream.LatestSeq
		if err := writeSSEJSON(w, "stream", "", dto.LogStreamFromDomain(stream)); err != nil {
			return nil, err
		}
	}
	return emittedThrough, nil
}

func (h *coreHandlers) openLogEventSubscription(r *http.Request) (domain.ServerInstance, []domain.LogStream, bool, service.LogEventSubscription, error) {
	var instance domain.ServerInstance
	var streams []domain.LogStream
	var liveEligible bool
	var subscription service.LogEventSubscription
	var err error
	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 {
			return domain.ServerInstance{}, nil, false, subscription, err
		}
		subscription, err = h.core.SubscribeLogEventsForSession(sessionID, instance.ID)
		if err != nil {
			return domain.ServerInstance{}, nil, false, subscription, err
		}
	} else {
		instance, err = h.core.GetServerInstance(r.PathValue("id"))
		if err != nil {
			return domain.ServerInstance{}, nil, false, subscription, err
		}
		subscription, err = h.core.SubscribeLogEvents(instance.ID)
		if err != nil {
			return domain.ServerInstance{}, nil, false, subscription, err
		}
	}
	if err == nil {
		streams, liveEligible, err = h.loadLiveLogSnapshot(instance.ID, isComponentLogRequest(r))
	}
	if err != nil && subscription.Close != nil {
		subscription.Close()
	}
	return instance, streams, liveEligible, subscription, err
}

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
	}
	if instance.State != domain.ServerInstanceStateRunning || strings.TrimSpace(instance.RunEndpointID) == "" {
		return nil, false, nil
	}
	endpoint, err := h.core.GetRunEndpoint(instance.RunEndpointID)
	if err != nil {
		return nil, false, err
	}
	if endpoint.Status != domain.RunEndpointStatusOnline {
		return nil, false, nil
	}
	logSessionID := strings.TrimPrefix(instance.LifecycleProcessID, managedLogSessionIDPrefix)
	if logSessionID == instance.LifecycleProcessID || strings.TrimSpace(logSessionID) == "" {
		return nil, false, nil
	}
	streams, err := h.core.ListLogStreams(domain.LogStreamFilter{ServerInstanceID: instance.ID})
	if err != nil {
		return nil, false, err
	}
	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)
		}
	}
	return current, true, nil
}

func writeSSEJSON(w http.ResponseWriter, eventName string, id string, value any) error {
	payload, err := json.Marshal(value)
	if err != nil {
		return err
	}
	if id != "" {
		if _, err := fmt.Fprintf(w, "id: %s\n", sanitizeSSEField(id)); err != nil {
			return err
		}
	}
	if _, err := fmt.Fprintf(w, "event: %s\n", sanitizeSSEField(eventName)); err != nil {
		return err
	}
	_, err = fmt.Fprintf(w, "data: %s\n\n", payload)
	return err
}

func sanitizeSSEField(value string) string {
	value = strings.ReplaceAll(value, "\r", "")
	value = strings.ReplaceAll(value, "\n", "")
	return value
}

func logEventID(event domain.LogStreamEvent) string {
	return fmt.Sprintf("%s:%d", sanitizeSSEField(event.Stream.ID), event.Entry.Seq)
}