feat: make Kafka projector the Workbench realtime authority

internal/workbenchruntime/realtime_sync.go

@@ -23,6 +23,8 @@ type realtimeSyncQuery struct {
 	TraceID        string       `json:"traceId"`
 	AfterOutboxSeq int64        `json:"afterOutboxSeq"`
 	Limit          int          `json:"limit"`
+	SnapshotOnly   bool         `json:"snapshotOnly,omitempty"`
+	DeltaOnly      bool         `json:"deltaOnly,omitempty"`
 	Actor          sessionActor `json:"actor"`
 }
 
@@ -74,6 +76,12 @@ var realtimeSyncFactSpecs = []realtimeSyncFactSpec{
 	{Family: "checkpoints", Table: "workbench_projection_checkpoints", JSONColumn: "checkpoint_json", OrderBy: "updated_at ASC, trace_id ASC"},
 }
 
+var realtimeSyncSnapshotFactSpecs = []realtimeSyncFactSpec{
+	realtimeSyncFactSpecs[0],
+	realtimeSyncFactSpecs[1],
+	realtimeSyncFactSpecs[3],
+}
+
 const realtimeSyncResolveScopeSQL = `
 SELECT sessions.session_id
 FROM workbench_sessions AS sessions
@@ -212,6 +220,9 @@ func normalizeRealtimeSyncQuery(query realtimeSyncQuery) (realtimeSyncQuery, *re
 	if query.Limit < 0 {
 		return query, &realtimeSyncValidationError{Status: http.StatusBadRequest, Code: "workbench_sync_limit_invalid", Message: "limit must be non-negative"}
 	}
+	if query.SnapshotOnly && query.DeltaOnly {
+		return query, &realtimeSyncValidationError{Status: http.StatusBadRequest, Code: "workbench_sync_mode_invalid", Message: "snapshotOnly and deltaOnly are mutually exclusive"}
+	}
 	query.Limit = boundedLimit(query.Limit, realtimeSyncDefaultLimit, realtimeSyncMaxLimit)
 	if query.Actor.Role != "admin" && query.Actor.Role != "user" {
 		return query, &realtimeSyncValidationError{Status: http.StatusBadRequest, Code: "workbench_sync_actor_invalid", Message: "actor.role must be user or admin"}
@@ -285,21 +296,32 @@ func readRealtimeSyncSnapshot(ctx context.Context, tx realtimeSyncQueryer, query
 	if err != nil {
 		return realtimeSyncSnapshot{}, err
 	}
-	allEvents, err := readRealtimeSyncEvents(ctx, tx, sessionID, query.TraceID, query.AfterOutboxSeq, cutoff, query.Limit+1)
-	if err != nil {
-		return realtimeSyncSnapshot{}, err
+	events := make([]realtimeSyncEvent, 0)
+	hasMore := false
+	if !query.SnapshotOnly {
+		allEvents, readErr := readRealtimeSyncEvents(ctx, tx, sessionID, query.TraceID, query.AfterOutboxSeq, cutoff, query.Limit+1)
+		if readErr != nil {
+			return realtimeSyncSnapshot{}, readErr
+		}
+		hasMore = len(allEvents) > query.Limit
+		events = allEvents
+		if hasMore {
+			events = allEvents[:query.Limit]
+		}
 	}
-	hasMore := len(allEvents) > query.Limit
-	events := allEvents
-	if hasMore {
-		events = allEvents[:query.Limit]
-	}
-	facts, err := readRealtimeSyncFacts(ctx, tx, sessionID, query.TraceID)
-	if err != nil {
-		return realtimeSyncSnapshot{}, err
+	facts := make(map[string][]any)
+	if !query.DeltaOnly {
+		specs := realtimeSyncFactSpecs
+		if query.SnapshotOnly {
+			specs = realtimeSyncSnapshotFactSpecs
+		}
+		facts, err = readRealtimeSyncFacts(ctx, tx, sessionID, query.TraceID, specs)
+		if err != nil {
+			return realtimeSyncSnapshot{}, err
+		}
 	}
 	cursor := query.AfterOutboxSeq
-	if cursor > cutoff {
+	if query.SnapshotOnly || cursor > cutoff {
 		cursor = cutoff
 	}
 	if len(events) > 0 {
@@ -408,9 +430,9 @@ func scanRealtimeSyncEvent(scanner realtimeSyncScanner) (realtimeSyncEvent, erro
 	return event, nil
 }
 
-func readRealtimeSyncFacts(ctx context.Context, tx realtimeSyncQueryer, sessionID string, traceID string) (map[string][]any, error) {
-	facts := make(map[string][]any, len(realtimeSyncFactSpecs))
-	for _, spec := range realtimeSyncFactSpecs {
+func readRealtimeSyncFacts(ctx context.Context, tx realtimeSyncQueryer, sessionID string, traceID string, specs []realtimeSyncFactSpec) (map[string][]any, error) {
+	facts := make(map[string][]any, len(specs))
+	for _, spec := range specs {
 		args := []any{sessionID}
 		where := "session_id = $1"
 		if spec.Family != "sessions" {

internal/workbenchruntime/realtime_sync_test.go

@@ -56,6 +56,11 @@ func TestNormalizeRealtimeSyncQueryRequiresScopedActorAndBoundsLimit(t *testing.
 	if validation == nil || validation.Code != "workbench_sync_cursor_invalid" {
 		t.Fatalf("negative cursor validation=%#v", validation)
 	}
+
+	_, validation = normalizeRealtimeSyncQuery(realtimeSyncQuery{SessionID: "ses_sync", SnapshotOnly: true, DeltaOnly: true, Actor: sessionActor{ID: "usr_owner", Role: "user"}})
+	if validation == nil || validation.Code != "workbench_sync_mode_invalid" {
+		t.Fatalf("conflicting sync mode validation=%#v", validation)
+	}
 }
 
 func TestRealtimeSyncRouteIsPostOnlyAndRejectsUnknownFields(t *testing.T) {
@@ -152,6 +157,52 @@ func TestReadRealtimeSyncSnapshotRejectsOtherOwnerBeforeReadingFacts(t *testing.
 	}
 }
 
+func TestReadRealtimeSyncSnapshotOnlySkipsHistoryAndReadsBoundedCurrentFacts(t *testing.T) {
+	queryer := newRealtimeSyncFakeQueryer("usr_owner")
+	snapshot, err := readRealtimeSyncSnapshot(context.Background(), queryer, realtimeSyncQuery{
+		SessionID:    "ses_sync",
+		Limit:        100,
+		SnapshotOnly: true,
+		Actor:        sessionActor{ID: "usr_owner", Role: "user"},
+	})
+	if err != nil {
+		t.Fatalf("read snapshot-only sync: %v", err)
+	}
+	if len(snapshot.Events) != 0 || snapshot.HasMore || snapshot.CursorOutboxSeq != snapshot.CutoffOutboxSeq {
+		t.Fatalf("snapshot-only boundary=%#v", snapshot)
+	}
+	if len(snapshot.Facts) != len(realtimeSyncSnapshotFactSpecs) {
+		t.Fatalf("snapshot-only facts=%#v", snapshot.Facts)
+	}
+	for _, call := range queryer.calls {
+		if call.kind == "events" || call.kind == "facts.parts" || call.kind == "facts.traceEvents" || call.kind == "facts.checkpoints" {
+			t.Fatalf("snapshot-only performed unbounded read: %v", queryer.calls)
+		}
+	}
+}
+
+func TestReadRealtimeSyncDeltaOnlyReadsOutboxWithoutFacts(t *testing.T) {
+	queryer := newRealtimeSyncFakeQueryer("usr_owner")
+	snapshot, err := readRealtimeSyncSnapshot(context.Background(), queryer, realtimeSyncQuery{
+		SessionID:      "ses_sync",
+		AfterOutboxSeq: 1,
+		Limit:          100,
+		DeltaOnly:      true,
+		Actor:          sessionActor{ID: "usr_owner", Role: "user"},
+	})
+	if err != nil {
+		t.Fatalf("read delta-only sync: %v", err)
+	}
+	if len(snapshot.Events) == 0 || len(snapshot.Facts) != 0 {
+		t.Fatalf("delta-only result=%#v", snapshot)
+	}
+	for _, call := range queryer.calls {
+		if strings.HasPrefix(call.kind, "facts.") {
+			t.Fatalf("delta-only read facts: %v", queryer.calls)
+		}
+	}
+}
+
 func TestReadRealtimeSyncSnapshotClampsForeignCursorToScopedCutoff(t *testing.T) {
 	queryer := newRealtimeSyncFakeQueryer("usr_owner")
 	queryer.events = [][]any{}