review: move CaptureSession to standalone capture package

Author: archandatta

server/cmd/api/api/api.go

@@ -10,9 +10,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/kernel/kernel-images/server/lib/capture"
 	"github.com/kernel/kernel-images/server/lib/cdpmonitor"
 	"github.com/kernel/kernel-images/server/lib/devtoolsproxy"
-	"github.com/kernel/kernel-images/server/lib/events"
 	"github.com/kernel/kernel-images/server/lib/logger"
 	"github.com/kernel/kernel-images/server/lib/nekoclient"
 	oapi "github.com/kernel/kernel-images/server/lib/oapi"
@@ -81,7 +81,7 @@ type ApiService struct {
 	xvfbResizeMu sync.Mutex
 
 	// CDP event pipeline and cdpMonitor.
-	captureSession  *events.CaptureSession
+	captureSession  *capture.CaptureSession
 	cdpMonitor      cdpMonitorController
 	monitorMu       sync.Mutex
 	lifecycleCtx    context.Context
@@ -96,7 +96,7 @@ func New(
 	upstreamMgr *devtoolsproxy.UpstreamManager,
 	stz scaletozero.Controller,
 	nekoAuthClient *nekoclient.AuthClient,
-	captureSession *events.CaptureSession,
+	captureSession *capture.CaptureSession,
 	displayNum int,
 ) (*ApiService, error) {
 	switch {

server/cmd/api/api/api_test.go

@@ -11,6 +11,7 @@ import (
 
 	"log/slog"
 
+	"github.com/kernel/kernel-images/server/lib/capture"
 	"github.com/kernel/kernel-images/server/lib/devtoolsproxy"
 	"github.com/kernel/kernel-images/server/lib/events"
 	"github.com/kernel/kernel-images/server/lib/nekoclient"
@@ -304,13 +305,13 @@ func newMockNekoClient(t *testing.T) *nekoclient.AuthClient {
 	return client
 }
 
-func newCaptureSession(t *testing.T) *events.CaptureSession {
+func newCaptureSession(t *testing.T) *capture.CaptureSession {
 	t.Helper()
 	es, err := events.NewEventStream(events.EventStreamConfig{RingCapacity: 64})
 	if err != nil {
 		t.Fatal(err)
 	}
-	return events.NewCaptureSession(es)
+	return capture.NewCaptureSession(es)
 }
 
 func TestApiService_PatchChromiumFlags(t *testing.T) {

server/cmd/api/api/capture_session.go

@@ -6,6 +6,7 @@ import (
 	"sort"
 
 	"github.com/nrednav/cuid2"
+	"github.com/kernel/kernel-images/server/lib/capture"
 	oapi "github.com/kernel/kernel-images/server/lib/oapi"
 
 	"github.com/kernel/kernel-images/server/lib/events"
@@ -119,26 +120,26 @@ func (s *ApiService) buildSessionResponse() oapi.CaptureSession {
 }
 
 // captureConfigFrom converts the optional StartCaptureSessionRequest body
-// into an events.CaptureConfig.
-func captureConfigFrom(body *oapi.StartCaptureSessionRequest) (events.CaptureConfig, error) {
+// into a capture.CaptureConfig.
+func captureConfigFrom(body *oapi.StartCaptureSessionRequest) (capture.CaptureConfig, error) {
 	if body == nil {
-		return events.CaptureConfig{}, nil
+		return capture.CaptureConfig{}, nil
 	}
 	return captureConfigFromOAPI(body.Config)
 }
 
-// captureConfigFromOAPI converts an oapi.CaptureConfig to events.CaptureConfig.
-func captureConfigFromOAPI(cfg *oapi.CaptureConfig) (events.CaptureConfig, error) {
+// captureConfigFromOAPI converts an oapi.CaptureConfig to capture.CaptureConfig.
+func captureConfigFromOAPI(cfg *oapi.CaptureConfig) (capture.CaptureConfig, error) {
 	if cfg == nil || cfg.Categories == nil {
-		return events.CaptureConfig{}, nil
+		return capture.CaptureConfig{}, nil
 	}
-	out := events.CaptureConfig{
+	out := capture.CaptureConfig{
 		Categories: make([]events.EventCategory, 0, len(*cfg.Categories)),
 	}
 	for _, c := range *cfg.Categories {
 		cat := events.EventCategory(c)
 		if !events.ValidCategory(cat) {
-			return events.CaptureConfig{}, fmt.Errorf("unknown category: %q", c)
+			return capture.CaptureConfig{}, fmt.Errorf("unknown category: %q", c)
 		}
 		out.Categories = append(out.Categories, cat)
 	}

server/cmd/api/main.go

@@ -23,6 +23,7 @@ import (
 	"github.com/kernel/kernel-images/server/cmd/api/api"
 	"github.com/kernel/kernel-images/server/cmd/config"
 	"github.com/kernel/kernel-images/server/lib/chromedriverproxy"
+	"github.com/kernel/kernel-images/server/lib/capture"
 	"github.com/kernel/kernel-images/server/lib/devtoolsproxy"
 	"github.com/kernel/kernel-images/server/lib/events"
 	"github.com/kernel/kernel-images/server/lib/logger"
@@ -99,7 +100,7 @@ func main() {
 		slogger.Error("failed to create event stream", "err", err)
 		os.Exit(1)
 	}
-	captureSession := events.NewCaptureSession(eventStream)
+	captureSession := capture.NewCaptureSession(eventStream)
 
 	apiService, err := api.New(
 		recorder.NewFFmpegManager(),

server/lib/events/capturesession.go server/lib/capture/capturesession.goserver/lib/events/capturesession.go renamed to server/lib/capture/capturesession.go

@@ -1,33 +1,35 @@
-package events
+package capture
 
 import (
 	"sync"
 	"time"
+
+	"github.com/kernel/kernel-images/server/lib/events"
 )
 
 // CaptureConfig holds caller-supplied capture preferences. All fields are
 // optional; zero values mean "use server defaults" (all categories).
 type CaptureConfig struct {
 	// Categories limits which event categories are captured.
 	// nil or empty includes all categories.
-	Categories []EventCategory
+	Categories []events.EventCategory
 }
 
 // CaptureSession manages a capture session against a shared EventStream.
 // It is responsible for (a) category-filtering Publish calls and (b) tracking
 // session-scoped metadata (ID, config, timestamps).
 type CaptureSession struct {
-	es               *EventStream
+	es               *events.EventStream
 	mu               sync.Mutex
 	captureSessionID string
 	sessionStartSeq  uint64
-	categories       map[EventCategory]struct{}
+	categories       map[events.EventCategory]struct{}
 	createdAt        time.Time
 }
 
-func NewCaptureSession(es *EventStream) *CaptureSession {
-	cats := make(map[EventCategory]struct{}, len(allCategories))
-	for _, c := range allCategories {
+func NewCaptureSession(es *events.EventStream) *CaptureSession {
+	cats := make(map[events.EventCategory]struct{}, len(events.AllCategories()))
+	for _, c := range events.AllCategories() {
 		cats[c] = struct{}{}
 	}
 	return &CaptureSession{es: es, categories: cats}
@@ -36,8 +38,8 @@ func NewCaptureSession(es *EventStream) *CaptureSession {
 // Start begins a new capture session with the given ID and config. Sequence
 // numbers are process-monotonic and do not reset between sessions; a
 // Last-Event-ID from any previous session is valid for resuming the SSE stream.
-// Events from different sessions are interleaved in the same per-category JSONL
-// files and distinguished by their envelope's captureSessionID.
+// Events from different sessions are distinguished by their envelope's
+// CaptureSessionID.
 func (s *CaptureSession) Start(captureSessionID string, cfg CaptureConfig) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -46,28 +48,28 @@ func (s *CaptureSession) Start(captureSessionID string, cfg CaptureConfig) {
 	s.createdAt = time.Now()
 	cats := cfg.Categories
 	if len(cats) == 0 {
-		cats = allCategories
+		cats = events.AllCategories()
 	}
-	s.categories = make(map[EventCategory]struct{}, len(cats))
+	s.categories = make(map[events.EventCategory]struct{}, len(cats))
 	for _, c := range cats {
 		s.categories[c] = struct{}{}
 	}
 }
 
 // publishLocked builds an envelope and forwards it to the EventStream.
 // Requires s.mu to be held.
-func (s *CaptureSession) publishLocked(ev Event) Envelope {
+func (s *CaptureSession) publishLocked(ev events.Event) events.Envelope {
 	if ev.Ts == 0 {
 		ev.Ts = time.Now().UnixMicro()
 	}
-	return s.es.publish(Envelope{
+	return s.es.Publish(events.Envelope{
 		CaptureSessionID: s.captureSessionID,
 		Event:            ev,
 	})
 }
 
 // Publish applies the category filter then forwards ev to the EventStream.
-func (s *CaptureSession) Publish(ev Event) {
+func (s *CaptureSession) Publish(ev events.Event) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if s.captureSessionID == "" {
@@ -80,18 +82,18 @@ func (s *CaptureSession) Publish(ev Event) {
 }
 
 // PublishUnfiltered forwards ev to the EventStream without applying the category
-// filter. Returns the assigned Envelope, or a zero Envelope if no session is active.
-func (s *CaptureSession) PublishUnfiltered(ev Event) Envelope {
+// filter. Returns the assigned Envelope.
+func (s *CaptureSession) PublishUnfiltered(ev events.Event) events.Envelope {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if s.captureSessionID == "" {
-		return Envelope{}
+		return events.Envelope{}
 	}
 	return s.publishLocked(ev)
 }
 
 // NewReader returns a Reader from the EventStream positioned after afterSeq.
-func (s *CaptureSession) NewReader(afterSeq uint64) *Reader {
+func (s *CaptureSession) NewReader(afterSeq uint64) *events.Reader {
 	return s.es.NewReader(afterSeq)
 }
 
@@ -102,7 +104,6 @@ func (s *CaptureSession) ID() string {
 	return s.captureSessionID
 }
 
-// Seq returns the sequence number of the last published event.
 func (s *CaptureSession) Seq() uint64 {
 	return s.es.Seq()
 }
@@ -115,39 +116,35 @@ func (s *CaptureSession) SessionStartSeq() uint64 {
 	return s.sessionStartSeq
 }
 
-// Config returns the current capture configuration.
 func (s *CaptureSession) Config() CaptureConfig {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	cats := make([]EventCategory, 0, len(s.categories))
+	cats := make([]events.EventCategory, 0, len(s.categories))
 	for c := range s.categories {
 		cats = append(cats, c)
 	}
 	return CaptureConfig{Categories: cats}
 }
 
-// CreatedAt returns when the current session was started.
 func (s *CaptureSession) CreatedAt() time.Time {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	return s.createdAt
 }
 
-// UpdateConfig applies a new CaptureConfig to the running session.
 func (s *CaptureSession) UpdateConfig(cfg CaptureConfig) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	cats := cfg.Categories
 	if len(cats) == 0 {
-		cats = allCategories
+		cats = events.AllCategories()
 	}
-	s.categories = make(map[EventCategory]struct{}, len(cats))
+	s.categories = make(map[events.EventCategory]struct{}, len(cats))
 	for _, c := range cats {
 		s.categories[c] = struct{}{}
 	}
 }
 
-// Active reports whether a capture session is currently running.
 func (s *CaptureSession) Active() bool {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -163,11 +160,10 @@ func (s *CaptureSession) Stop() {
 	if s.captureSessionID == "" {
 		return
 	}
-	s.publishLocked(Event{
-		Type:     TypeSessionEnded,
-		Category: CategorySystem,
-		Source:   Source{Kind: KindKernelAPI},
+	s.publishLocked(events.Event{
+		Type:     events.TypeSessionEnded,
+		Category: events.CategorySystem,
+		Source:   events.Source{Kind: events.KindKernelAPI},
 	})
 	s.captureSessionID = ""
 }
-