[kernel-1116] browser events: add external events

server/Makefile

@@ -19,7 +19,7 @@ oapi-generate:
 	openapi-down-convert --input openapi.yaml --output openapi-3.0.yaml
 	go tool oapi-codegen -config ./oapi-codegen.yaml ./openapi-3.0.yaml
 	@echo "Fixing oapi-codegen issue https://github.com/oapi-codegen/oapi-codegen/issues/1764..."
-	go run ./scripts/oapi/patch_sse_methods -file ./lib/oapi/oapi.go -expected-replacements 3
+	go run ./scripts/oapi/patch_sse_methods -file ./lib/oapi/oapi.go -expected-replacements 4
 	go fmt ./lib/oapi/oapi.go
 	go mod tidy
 

server/cmd/api/api/capture_session_test.go

@@ -19,7 +19,7 @@ func TestCaptureConfigFrom(t *testing.T) {
 	})
 
 	t.Run("valid categories", func(t *testing.T) {
-		cats := []oapi.CaptureConfigCategories{oapi.Console, oapi.Network}
+		cats := []oapi.CaptureConfigCategories{oapi.CaptureConfigCategoriesConsole, oapi.CaptureConfigCategoriesNetwork}
 		body := &oapi.StartCaptureSessionRequest{
 			Config: &oapi.CaptureConfig{Categories: &cats},
 		}
@@ -64,7 +64,7 @@ func TestStartCaptureSession(t *testing.T) {
 
 	t.Run("success with config", func(t *testing.T) {
 		svc := newTestService(t, newMockRecordManager())
-		cats := []oapi.CaptureConfigCategories{oapi.Console}
+		cats := []oapi.CaptureConfigCategories{oapi.CaptureConfigCategoriesConsole}
 		resp, err := svc.StartCaptureSession(ctx, oapi.StartCaptureSessionRequestObject{
 			Body: &oapi.StartCaptureSessionRequest{
 				Config: &oapi.CaptureConfig{Categories: &cats},
@@ -141,7 +141,7 @@ func TestUpdateCaptureSession(t *testing.T) {
 		_, err := svc.StartCaptureSession(ctx, oapi.StartCaptureSessionRequestObject{})
 		require.NoError(t, err)
 
-		cats := []oapi.CaptureConfigCategories{oapi.Console}
+		cats := []oapi.CaptureConfigCategories{oapi.CaptureConfigCategoriesConsole}
 		resp, err := svc.UpdateCaptureSession(ctx, oapi.UpdateCaptureSessionRequestObject{
 			Body: &oapi.UpdateCaptureSessionRequest{
 				Config: &oapi.CaptureConfig{Categories: &cats},
@@ -152,7 +152,7 @@ func TestUpdateCaptureSession(t *testing.T) {
 		require.True(t, ok)
 		require.NotNil(t, r200.Config.Categories)
 		assert.Len(t, *r200.Config.Categories, 1)
-		assert.Equal(t, oapi.Console, (*r200.Config.Categories)[0])
+		assert.Equal(t, oapi.CaptureConfigCategoriesConsole, (*r200.Config.Categories)[0])
 	})
 
 	t.Run("empty body is no-op", func(t *testing.T) {

server/cmd/api/api/events.go

@@ -0,0 +1,166 @@
+package api
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"strconv"
+	"time"
+
+	"github.com/kernel/kernel-images/server/lib/events"
+	oapi "github.com/kernel/kernel-images/server/lib/oapi"
+)
+
+// PublishEvent handles POST /events/publish.
+// Injects a caller-supplied event into the active capture session. Returns 400
+// if no session is active or the event fails validation.
+func (s *ApiService) PublishEvent(_ context.Context, req oapi.PublishEventRequestObject) (oapi.PublishEventResponseObject, error) {
+	if !s.captureSession.Active() {
+		return oapi.PublishEvent400JSONResponse{BadRequestErrorJSONResponse: oapi.BadRequestErrorJSONResponse{Message: "no active capture session"}}, nil
+	}
+
+	body := req.Body
+	if body == nil || body.Type == "" {
+		return oapi.PublishEvent400JSONResponse{BadRequestErrorJSONResponse: oapi.BadRequestErrorJSONResponse{Message: "type is required"}}, nil
+	}
+	if body.Type == events.TypeSessionEnded || body.Type == events.TypeEventsDropped {
+		return oapi.PublishEvent400JSONResponse{BadRequestErrorJSONResponse: oapi.BadRequestErrorJSONResponse{Message: "type is reserved"}}, nil
+	}
+
+	ev := events.Event{Type: body.Type}
+
+	if body.Ts != nil {
+		ev.Ts = *body.Ts
+	}
+	if body.Category != nil {
+		cat := events.EventCategory(*body.Category)
+		if !events.ValidCategory(cat) {
+			return oapi.PublishEvent400JSONResponse{BadRequestErrorJSONResponse: oapi.BadRequestErrorJSONResponse{Message: "invalid category"}}, nil
+		}
+		ev.Category = cat
+	} else {
+		ev.Category = events.CategorySystem
+	}
+
+	// Enforce source.kind = KindKernelAPI so callers can't spoof the origin.
+	ev.Source.Kind = events.KindKernelAPI
+	if body.Source != nil {
+		if body.Source.Event != nil {
+			ev.Source.Event = *body.Source.Event
+		}
+		if body.Source.Metadata != nil {
+			ev.Source.Metadata = *body.Source.Metadata
+		}
+	}
+
+	if body.Data != nil {
+		// re-marshal body.Data to normalize it into a canonical RawMessage byte slice.
+		data, err := json.Marshal(body.Data)
+		if err != nil {
+			return oapi.PublishEvent400JSONResponse{BadRequestErrorJSONResponse: oapi.BadRequestErrorJSONResponse{Message: "invalid data"}}, nil
+		}
+		ev.Data = json.RawMessage(data)
+	}
+
+	s.captureSession.PublishUnfiltered(ev)
+	return oapi.PublishEvent200Response{}, nil
+}
+
+// StreamEvents handles GET /events/stream.
+// Opens an SSE stream of envelopes from the active capture session's ring buffer.
+// Supports reconnection via the Last-Event-ID header. Emits a keepalive comment
+// frame every 15 s when no event arrives, and exits cleanly on session_ended.
+func (s *ApiService) StreamEvents(ctx context.Context, req oapi.StreamEventsRequestObject) (oapi.StreamEventsResponseObject, error) {
+	if !s.captureSession.Active() {
+		return oapi.StreamEvents400JSONResponse{BadRequestErrorJSONResponse: oapi.BadRequestErrorJSONResponse{Message: "no active capture session"}}, nil
+	}
+
+	afterSeq := uint64(0)
+	if id := req.Params.LastEventID; id != nil && *id != "" {
+		// Invalid/non-numeric values fall back to 0, replaying all events from the start.
+		// Note: seq is per capture session and resets on each Start(). A Last-Event-ID
+		// from a previous session may silently overlap with the current session's seqs.
+		if n, err := strconv.ParseUint(*id, 10, 64); err == nil {
+			afterSeq = n
+		}
+	}
+
+	sessionID := s.captureSession.ID()
+	reader := s.captureSession.NewReader(afterSeq)
+
+	pr, pw := io.Pipe()
+	go func() {
+		defer pw.Close()
+		for {
+			readCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
+			result, err := reader.Read(readCtx)
+			cancel()
+			if err != nil {
+				if errors.Is(err, context.DeadlineExceeded) {
+					select {
+					case <-ctx.Done():
+						return
+					default:
+						// No event in 15 s and client still connected, send keepalive.
+						if _, err := pw.Write([]byte(":\n\n")); err != nil {
+							return
+						}
+						continue
+					}
+				}
+				return
+			}
+
+			if result.Dropped > 0 {
+				env := events.Envelope{
+					CaptureSessionID: sessionID,
+					Seq:              0,
+					Event: events.Event{
+						Ts:       time.Now().UnixMicro(),
+						Type:     events.TypeEventsDropped,
+						Category: events.CategorySystem,
+						Source:   events.Source{Kind: events.KindKernelAPI},
+						Data:     json.RawMessage(fmt.Sprintf(`{"dropped":%d}`, result.Dropped)),
+					},
+				}
+				// Omit the id: field so the client's Last-Event-ID is not overwritten.
+				if err := writeEnvelopeFrame(pw, nil, env); err != nil {
+					return
+				}
+				continue
+			}
+
+			env := result.Envelope
+			if err := writeEnvelopeFrame(pw, &env.Seq, *env); err != nil {
+				return
+			}
+			if env.Event.Type == events.TypeSessionEnded {
+				return
+			}
+		}
+	}()
+
+	headers := oapi.StreamEvents200ResponseHeaders{XSSEContentType: "application/json"}
+	return oapi.StreamEvents200TexteventStreamResponse{Body: pr, Headers: headers}, nil
+}
+
+// writeEnvelopeFrame writes a single SSE frame. If seq is non-nil it is
+// emitted as the id: field, updating the client's Last-Event-ID.
+func writeEnvelopeFrame(w io.Writer, seq *uint64, env events.Envelope) error {
+	data, err := json.Marshal(env)
+	if err != nil {
+		return err
+	}
+	var buf bytes.Buffer
+	if seq != nil {
+		fmt.Fprintf(&buf, "id: %d\n", *seq)
+	}
+	buf.WriteString("data: ")
+	buf.Write(data)
+	buf.WriteString("\n\n")
+	_, err = w.Write(buf.Bytes())
+	return err
+}

server/cmd/api/api/events_test.go

@@ -0,0 +1,92 @@
+package api
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/kernel/kernel-images/server/lib/events"
+	oapi "github.com/kernel/kernel-images/server/lib/oapi"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEventLifecycle(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	svc := newTestService(t, newMockRecordManager())
+
+	// Start a capture session.
+	startResp, err := svc.StartCaptureSession(ctx, oapi.StartCaptureSessionRequestObject{})
+	require.NoError(t, err)
+	require.IsType(t, oapi.StartCaptureSession201JSONResponse{}, startResp)
+
+	// Open an SSE stream (5s budget covers the three 2s selects below).
+	streamCtx, streamCancel := context.WithTimeout(ctx, 5*time.Second)
+	defer streamCancel()
+	streamResp, err := svc.StreamEvents(streamCtx, oapi.StreamEventsRequestObject{})
+	require.NoError(t, err)
+	r200, ok := streamResp.(oapi.StreamEvents200TexteventStreamResponse)
+	require.True(t, ok)
+
+	// Drain SSE frames into a channel.
+	received := make(chan events.Envelope, 4)
+	go func() {
+		defer close(received)
+		rd := bufio.NewReader(r200.Body)
+		for {
+			line, err := rd.ReadString('\n')
+			if err != nil {
+				return
+			}
+			if !strings.HasPrefix(line, "data: ") {
+				continue
+			}
+			payload := strings.TrimSpace(strings.TrimPrefix(line, "data: "))
+			var env events.Envelope
+			if err := json.Unmarshal([]byte(payload), &env); err != nil {
+				continue
+			}
+			received <- env
+		}
+	}()
+
+	// Publish an event.
+	resp, err := svc.PublishEvent(ctx, oapi.PublishEventRequestObject{
+		Body: &oapi.Event{Type: "test.event"},
+	})
+	require.NoError(t, err)
+	assert.IsType(t, oapi.PublishEvent200Response{}, resp)
+
+	// Verify the published event arrives on the stream.
+	select {
+	case env := <-received:
+		assert.Equal(t, "test.event", env.Event.Type)
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for test.event")
+	}
+
+	// Stop the session.
+	stopResp, err := svc.StopCaptureSession(ctx, oapi.StopCaptureSessionRequestObject{})
+	require.NoError(t, err)
+	assert.IsType(t, oapi.StopCaptureSession200JSONResponse{}, stopResp)
+
+	// Verify session_ended arrives on the stream.
+	select {
+	case env := <-received:
+		assert.Equal(t, events.TypeSessionEnded, env.Event.Type)
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for session_ended")
+	}
+
+	// Verify the stream closes after session_ended.
+	select {
+	case _, open := <-received:
+		assert.False(t, open, "stream should be closed after session_ended")
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for stream to close")
+	}
+}