From a685be0b785244a6c8aaa4864d74f1cf5609cd78 Mon Sep 17 00:00:00 2001 From: root Date: Fri, 10 Jul 2026 02:59:27 +0200 Subject: [PATCH] feat: add atomic Workbench realtime sync --- internal/workbenchruntime/realtime_sync.go | 471 ++++++++++++++++++ .../workbenchruntime/realtime_sync_test.go | 358 +++++++++++++ internal/workbenchruntime/service.go | 1 + 3 files changed, 830 insertions(+) create mode 100644 internal/workbenchruntime/realtime_sync.go create mode 100644 internal/workbenchruntime/realtime_sync_test.go diff --git a/internal/workbenchruntime/realtime_sync.go b/internal/workbenchruntime/realtime_sync.go new file mode 100644 index 00000000..13a22086 --- /dev/null +++ b/internal/workbenchruntime/realtime_sync.go @@ -0,0 +1,471 @@ +package workbenchruntime + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" +) + +const ( + realtimeSyncContractVersion = "workbench-runtime-sync-v1" + realtimeSyncDefaultLimit = 100 + realtimeSyncMaxLimit = 500 + realtimeSyncMaxIdentityLen = 512 +) + +type realtimeSyncQuery struct { + SessionID string `json:"sessionId"` + TraceID string `json:"traceId"` + AfterOutboxSeq int64 `json:"afterOutboxSeq"` + Limit int `json:"limit"` + Actor sessionActor `json:"actor"` +} + +type realtimeSyncSnapshot struct { + Facts map[string][]any `json:"facts"` + Events []realtimeSyncEvent `json:"events"` + CutoffOutboxSeq int64 `json:"cutoffOutboxSeq"` + CursorOutboxSeq int64 `json:"cursorOutboxSeq"` + HasMore bool `json:"hasMore"` +} + +type realtimeSyncEvent struct { + OutboxSeq int64 `json:"outboxSeq"` + OutboxEventID string `json:"outboxEventId"` + EntityFamily string `json:"entityFamily"` + EntityID string `json:"entityId"` + EventSeq any `json:"eventSeq"` + AggregateID any `json:"aggregateId"` + AggregateSeq int64 `json:"aggregateSeq"` + ProjectionRevision int64 `json:"projectionRevision"` + TraceID string `json:"traceId"` + SessionID any `json:"sessionId"` + TurnID any `json:"turnId"` + MessageID any `json:"messageId"` + ProjectedSeq int64 `json:"projectedSeq"` + SourceSeq int64 `json:"sourceSeq"` + SourceEventID any `json:"sourceEventId"` + CommitType string `json:"commitType"` + Terminal bool `json:"terminal"` + Sealed bool `json:"sealed"` + Payload any `json:"payload"` + CreatedAt string `json:"createdAt"` + ValuesRedacted bool `json:"valuesRedacted"` +} + +type realtimeSyncFactSpec struct { + Family string + Table string + JSONColumn string + OrderBy string +} + +var realtimeSyncFactSpecs = []realtimeSyncFactSpec{ + {Family: "sessions", Table: "workbench_sessions", JSONColumn: "session_json", OrderBy: "updated_at ASC, session_id ASC"}, + {Family: "messages", Table: "workbench_messages", JSONColumn: "message_json", OrderBy: "updated_at ASC, message_id ASC"}, + {Family: "parts", Table: "workbench_parts", JSONColumn: "part_json", OrderBy: "updated_at ASC, part_index ASC, part_id ASC"}, + {Family: "turns", Table: "workbench_turns", JSONColumn: "turn_json", OrderBy: "updated_at ASC, turn_id ASC"}, + {Family: "traceEvents", Table: "workbench_trace_events", JSONColumn: "event_json", OrderBy: "projected_seq ASC, id ASC"}, + {Family: "checkpoints", Table: "workbench_projection_checkpoints", JSONColumn: "checkpoint_json", OrderBy: "updated_at ASC, trace_id ASC"}, +} + +const realtimeSyncResolveScopeSQL = ` +SELECT sessions.session_id +FROM workbench_sessions AS sessions +WHERE ($1 = '' OR sessions.session_id = $1) + AND ( + $2 = '' + OR sessions.last_trace_id = $2 + OR EXISTS ( + SELECT 1 + FROM workbench_projection_outbox AS scoped_outbox + WHERE scoped_outbox.session_id = sessions.session_id + AND scoped_outbox.trace_id = $2 + ) + ) + AND ($3 OR sessions.owner_user_id = $4) +ORDER BY + CASE WHEN sessions.session_id = NULLIF($1, '') THEN 0 ELSE 1 END, + sessions.updated_at DESC, + sessions.session_id ASC +LIMIT 1` + +const realtimeSyncCutoffSQL = ` +SELECT COALESCE(MAX(outbox_seq), 0)::bigint +FROM workbench_projection_outbox +WHERE session_id = $1 + AND ($2 = '' OR trace_id = $2)` + +const realtimeSyncEventsSQL = ` +SELECT + outbox_seq, + outbox_event_id, + entity_family, + entity_id, + event_seq, + aggregate_id, + aggregate_seq, + projection_revision, + trace_id, + session_id, + turn_id, + message_id, + projected_seq, + source_seq, + source_event_id, + commit_type, + terminal, + sealed, + payload_json, + created_at +FROM workbench_projection_outbox +WHERE session_id = $1 + AND ($2 = '' OR trace_id = $2) + AND outbox_seq > $3 + AND outbox_seq <= $4 +ORDER BY outbox_seq ASC +LIMIT $5` + +func (s *Server) handleRealtimeSync(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + methodNotAllowed(w, "POST") + return + } + query, ok := decodeRealtimeSyncQuery(w, r) + if !ok { + return + } + query, validation := normalizeRealtimeSyncQuery(query) + if validation != nil { + writeAPIError(w, validation.Status, validation.Code, validation.Message) + return + } + ctx, cancel := context.WithTimeout(r.Context(), s.config.QueryTimeout) + defer cancel() + snapshot, err := s.readRealtimeSync(ctx, query) + if errors.Is(err, sql.ErrNoRows) { + writeAPIError(w, http.StatusNotFound, "workbench_sync_scope_not_found", "workbench sync scope was not found") + return + } + if err != nil { + writeQueryError(w, err) + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "ok": true, + "status": "succeeded", + "contractVersion": realtimeSyncContractVersion, + "facts": snapshot.Facts, + "events": snapshot.Events, + "cutoffOutboxSeq": snapshot.CutoffOutboxSeq, + "cursorOutboxSeq": snapshot.CursorOutboxSeq, + "hasMore": snapshot.HasMore, + "persistence": s.persistenceSummary(), + "servedBy": serviceID, + "valuesRedacted": true, + "secretMaterialStored": false, + }) +} + +type realtimeSyncValidationError struct { + Status int + Code string + Message string +} + +func decodeRealtimeSyncQuery(w http.ResponseWriter, r *http.Request) (realtimeSyncQuery, bool) { + defer r.Body.Close() + decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)) + decoder.DisallowUnknownFields() + var query realtimeSyncQuery + if err := decoder.Decode(&query); err != nil { + writeAPIError(w, http.StatusBadRequest, "invalid_json", "request body must be a closed workbench sync JSON object") + return realtimeSyncQuery{}, false + } + var trailing any + if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) { + writeAPIError(w, http.StatusBadRequest, "invalid_json", "request body must contain exactly one JSON object") + return realtimeSyncQuery{}, false + } + return query, true +} + +func normalizeRealtimeSyncQuery(query realtimeSyncQuery) (realtimeSyncQuery, *realtimeSyncValidationError) { + query.SessionID = strings.TrimSpace(query.SessionID) + query.TraceID = strings.TrimSpace(query.TraceID) + query.Actor.ID = strings.TrimSpace(query.Actor.ID) + query.Actor.Role = strings.ToLower(strings.TrimSpace(query.Actor.Role)) + if query.SessionID == "" && query.TraceID == "" { + return query, &realtimeSyncValidationError{Status: http.StatusBadRequest, Code: "workbench_sync_scope_required", Message: "sessionId or traceId is required"} + } + if len(query.SessionID) > realtimeSyncMaxIdentityLen || len(query.TraceID) > realtimeSyncMaxIdentityLen || len(query.Actor.ID) > realtimeSyncMaxIdentityLen { + return query, &realtimeSyncValidationError{Status: http.StatusBadRequest, Code: "workbench_sync_identity_invalid", Message: "workbench sync identifiers are too long"} + } + if query.AfterOutboxSeq < 0 { + return query, &realtimeSyncValidationError{Status: http.StatusBadRequest, Code: "workbench_sync_cursor_invalid", Message: "afterOutboxSeq must be non-negative"} + } + if query.Limit < 0 { + return query, &realtimeSyncValidationError{Status: http.StatusBadRequest, Code: "workbench_sync_limit_invalid", Message: "limit must be non-negative"} + } + 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"} + } + if query.Actor.Role != "admin" && query.Actor.ID == "" { + return query, &realtimeSyncValidationError{Status: http.StatusForbidden, Code: "workbench_sync_actor_required", Message: "ordinary users require an actor id"} + } + return query, nil +} + +func realtimeSyncTxOptions() *sql.TxOptions { + return &sql.TxOptions{Isolation: sql.LevelRepeatableRead, ReadOnly: true} +} + +func (s *Server) readRealtimeSync(ctx context.Context, query realtimeSyncQuery) (realtimeSyncSnapshot, error) { + tx, err := s.db.BeginTx(ctx, realtimeSyncTxOptions()) + if err != nil { + return realtimeSyncSnapshot{}, wrapQueryStage("workbench_runtime.db.sync.begin", err) + } + committed := false + defer func() { + if !committed { + _ = tx.Rollback() + } + }() + snapshot, err := readRealtimeSyncSnapshot(ctx, sqlRealtimeSyncQueryer{tx: tx}, query) + if err != nil { + return realtimeSyncSnapshot{}, err + } + if err := tx.Commit(); err != nil { + return realtimeSyncSnapshot{}, wrapQueryStage("workbench_runtime.db.sync.commit", err) + } + committed = true + return snapshot, nil +} + +type realtimeSyncRow interface { + Scan(...any) error +} + +type realtimeSyncRows interface { + Close() error + Err() error + Next() bool + Scan(...any) error +} + +type realtimeSyncQueryer interface { + QueryContext(context.Context, string, ...any) (realtimeSyncRows, error) + QueryRowContext(context.Context, string, ...any) realtimeSyncRow +} + +type sqlRealtimeSyncQueryer struct { + tx *sql.Tx +} + +func (queryer sqlRealtimeSyncQueryer) QueryContext(ctx context.Context, query string, args ...any) (realtimeSyncRows, error) { + return queryer.tx.QueryContext(ctx, query, args...) +} + +func (queryer sqlRealtimeSyncQueryer) QueryRowContext(ctx context.Context, query string, args ...any) realtimeSyncRow { + return queryer.tx.QueryRowContext(ctx, query, args...) +} + +func readRealtimeSyncSnapshot(ctx context.Context, tx realtimeSyncQueryer, query realtimeSyncQuery) (realtimeSyncSnapshot, error) { + sessionID, err := resolveRealtimeSyncSession(ctx, tx, query) + if err != nil { + return realtimeSyncSnapshot{}, err + } + cutoff, err := readRealtimeSyncCutoff(ctx, tx, sessionID, query.TraceID) + 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 + } + 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 + } + cursor := query.AfterOutboxSeq + if cursor > cutoff { + cursor = cutoff + } + if len(events) > 0 { + cursor = events[len(events)-1].OutboxSeq + } + return realtimeSyncSnapshot{Facts: facts, Events: events, CutoffOutboxSeq: cutoff, CursorOutboxSeq: cursor, HasMore: hasMore}, nil +} + +func resolveRealtimeSyncSession(ctx context.Context, tx realtimeSyncQueryer, query realtimeSyncQuery) (string, error) { + var sessionID string + err := tx.QueryRowContext(ctx, realtimeSyncResolveScopeSQL, query.SessionID, query.TraceID, query.Actor.Role == "admin", query.Actor.ID).Scan(&sessionID) + if err != nil { + return "", wrapQueryStage("workbench_runtime.db.sync.scope", err) + } + return sessionID, nil +} + +func readRealtimeSyncCutoff(ctx context.Context, tx realtimeSyncQueryer, sessionID string, traceID string) (int64, error) { + var cutoff int64 + if err := tx.QueryRowContext(ctx, realtimeSyncCutoffSQL, sessionID, traceID).Scan(&cutoff); err != nil { + return 0, wrapQueryStage("workbench_runtime.db.sync.cutoff", err) + } + return cutoff, nil +} + +func readRealtimeSyncEvents(ctx context.Context, tx realtimeSyncQueryer, sessionID string, traceID string, after int64, cutoff int64, limit int) ([]realtimeSyncEvent, error) { + rows, err := tx.QueryContext(ctx, realtimeSyncEventsSQL, sessionID, traceID, after, cutoff, limit) + if err != nil { + return nil, wrapQueryStage("workbench_runtime.db.sync.events", err) + } + defer rows.Close() + events := make([]realtimeSyncEvent, 0) + for rows.Next() { + event, err := scanRealtimeSyncEvent(rows) + if err != nil { + return nil, wrapQueryStage("workbench_runtime.db.sync.events", err) + } + events = append(events, event) + } + if err := rows.Err(); err != nil { + return nil, wrapQueryStage("workbench_runtime.db.sync.events", err) + } + return events, nil +} + +type realtimeSyncScanner interface { + Scan(...any) error +} + +func scanRealtimeSyncEvent(scanner realtimeSyncScanner) (realtimeSyncEvent, error) { + var event realtimeSyncEvent + var outboxEventID, entityFamily, entityID, traceID, commitType, payloadJSON, createdAt sql.NullString + var eventSeq, aggregateSeq, projectionRevision, projectedSeq, sourceSeq sql.NullInt64 + var aggregateID, sessionID, turnID, messageID, sourceEventID sql.NullString + var terminal, sealed sql.NullBool + if err := scanner.Scan( + &event.OutboxSeq, + &outboxEventID, + &entityFamily, + &entityID, + &eventSeq, + &aggregateID, + &aggregateSeq, + &projectionRevision, + &traceID, + &sessionID, + &turnID, + &messageID, + &projectedSeq, + &sourceSeq, + &sourceEventID, + &commitType, + &terminal, + &sealed, + &payloadJSON, + &createdAt, + ); err != nil { + return realtimeSyncEvent{}, err + } + payload := any(map[string]any{}) + if payloadJSON.Valid && strings.TrimSpace(payloadJSON.String) != "" { + if err := json.Unmarshal([]byte(payloadJSON.String), &payload); err != nil { + return realtimeSyncEvent{}, fmt.Errorf("decode projection outbox payload: %w", err) + } + } + event.OutboxEventID = nullableStringValue(outboxEventID) + event.EntityFamily = nullableStringValue(entityFamily) + event.EntityID = nullableStringValue(entityID) + event.EventSeq = realtimeSyncNullableInt(eventSeq) + event.AggregateID = nullableString(aggregateID) + event.AggregateSeq = realtimeSyncInt(aggregateSeq) + event.ProjectionRevision = realtimeSyncInt(projectionRevision) + event.TraceID = nullableStringValue(traceID) + event.SessionID = nullableString(sessionID) + event.TurnID = nullableString(turnID) + event.MessageID = nullableString(messageID) + event.ProjectedSeq = realtimeSyncInt(projectedSeq) + event.SourceSeq = realtimeSyncInt(sourceSeq) + event.SourceEventID = nullableString(sourceEventID) + event.CommitType = nullableStringValue(commitType) + event.Terminal = nullableBool(terminal) + event.Sealed = nullableBool(sealed) + event.Payload = payload + event.CreatedAt = nullableTimeText(createdAt) + event.ValuesRedacted = true + 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 { + args := []any{sessionID} + where := "session_id = $1" + if spec.Family != "sessions" { + args = append(args, traceID) + where += " AND ($2 = '' OR trace_id = $2)" + } + sqlText := fmt.Sprintf("SELECT %s FROM %s WHERE %s ORDER BY %s", spec.JSONColumn, spec.Table, where, spec.OrderBy) + rows, err := tx.QueryContext(ctx, sqlText, args...) + if err != nil { + return nil, wrapQueryStage("workbench_runtime.db.sync.facts."+spec.Family, err) + } + items, err := scanRealtimeSyncFacts(rows) + if err != nil { + return nil, wrapQueryStage("workbench_runtime.db.sync.facts."+spec.Family, err) + } + facts[spec.Family] = items + } + return facts, nil +} + +func scanRealtimeSyncFacts(rows realtimeSyncRows) ([]any, error) { + defer rows.Close() + items := make([]any, 0) + for rows.Next() { + var raw sql.NullString + if err := rows.Scan(&raw); err != nil { + return nil, err + } + if !raw.Valid || strings.TrimSpace(raw.String) == "" { + continue + } + var item any + if err := json.Unmarshal([]byte(raw.String), &item); err != nil { + return nil, fmt.Errorf("decode workbench fact: %w", err) + } + if item != nil { + items = append(items, item) + } + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +func realtimeSyncNullableInt(value sql.NullInt64) any { + if !value.Valid { + return nil + } + return value.Int64 +} + +func realtimeSyncInt(value sql.NullInt64) int64 { + if !value.Valid { + return 0 + } + return value.Int64 +} diff --git a/internal/workbenchruntime/realtime_sync_test.go b/internal/workbenchruntime/realtime_sync_test.go new file mode 100644 index 00000000..49497c68 --- /dev/null +++ b/internal/workbenchruntime/realtime_sync_test.go @@ -0,0 +1,358 @@ +package workbenchruntime + +import ( + "context" + "database/sql" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestRealtimeSyncTxOptionsAreReadOnlyRepeatableRead(t *testing.T) { + options := realtimeSyncTxOptions() + if options.Isolation != sql.LevelRepeatableRead { + t.Fatalf("isolation=%v, want repeatable read", options.Isolation) + } + if !options.ReadOnly { + t.Fatal("realtime sync transaction must be read only") + } +} + +func TestNormalizeRealtimeSyncQueryRequiresScopedActorAndBoundsLimit(t *testing.T) { + query, validation := normalizeRealtimeSyncQuery(realtimeSyncQuery{ + SessionID: " ses_sync ", + Limit: 900, + Actor: sessionActor{ID: " usr_owner ", Role: " USER "}, + }) + if validation != nil { + t.Fatalf("unexpected validation: %#v", validation) + } + if query.SessionID != "ses_sync" || query.Actor.ID != "usr_owner" || query.Actor.Role != "user" { + t.Fatalf("query was not normalized: %#v", query) + } + if query.Limit != realtimeSyncMaxLimit { + t.Fatalf("limit=%d, want %d", query.Limit, realtimeSyncMaxLimit) + } + + _, validation = normalizeRealtimeSyncQuery(realtimeSyncQuery{TraceID: "trc_sync", Actor: sessionActor{Role: "user"}}) + if validation == nil || validation.Status != http.StatusForbidden || validation.Code != "workbench_sync_actor_required" { + t.Fatalf("ordinary user without owner id validation=%#v", validation) + } + + admin, validation := normalizeRealtimeSyncQuery(realtimeSyncQuery{TraceID: "trc_sync", Actor: sessionActor{Role: "admin"}}) + if validation != nil || admin.Actor.ID != "" { + t.Fatalf("admin should not require owner id: query=%#v validation=%#v", admin, validation) + } + + _, validation = normalizeRealtimeSyncQuery(realtimeSyncQuery{Actor: sessionActor{ID: "usr_owner", Role: "user"}}) + if validation == nil || validation.Code != "workbench_sync_scope_required" { + t.Fatalf("missing scope validation=%#v", validation) + } + + _, validation = normalizeRealtimeSyncQuery(realtimeSyncQuery{SessionID: "ses_sync", AfterOutboxSeq: -1, Actor: sessionActor{ID: "usr_owner", Role: "user"}}) + if validation == nil || validation.Code != "workbench_sync_cursor_invalid" { + t.Fatalf("negative cursor validation=%#v", validation) + } +} + +func TestRealtimeSyncRouteIsPostOnlyAndRejectsUnknownFields(t *testing.T) { + server := &Server{config: Config{}, mux: http.NewServeMux()} + server.routes() + + methodRecorder := httptest.NewRecorder() + server.mux.ServeHTTP(methodRecorder, httptest.NewRequest(http.MethodGet, "/v1/workbench-runtime/sync", nil)) + if methodRecorder.Code != http.StatusMethodNotAllowed { + t.Fatalf("GET status=%d, want %d", methodRecorder.Code, http.StatusMethodNotAllowed) + } + if methodRecorder.Header().Get("Allow") != "POST" { + t.Fatalf("Allow=%q, want POST", methodRecorder.Header().Get("Allow")) + } + + unknownRecorder := httptest.NewRecorder() + unknownRequest := httptest.NewRequest(http.MethodPost, "/v1/workbench-runtime/sync", strings.NewReader(`{"sessionId":"ses_sync","actor":{"id":"usr_owner","role":"user"},"unknown":true}`)) + server.mux.ServeHTTP(unknownRecorder, unknownRequest) + if unknownRecorder.Code != http.StatusBadRequest || !strings.Contains(unknownRecorder.Body.String(), "invalid_json") { + t.Fatalf("unknown field response=%d %s", unknownRecorder.Code, unknownRecorder.Body.String()) + } + + trailingRecorder := httptest.NewRecorder() + trailingRequest := httptest.NewRequest(http.MethodPost, "/v1/workbench-runtime/sync", strings.NewReader(`{"sessionId":"ses_sync","actor":{"id":"usr_owner","role":"user"}} {}`)) + server.mux.ServeHTTP(trailingRecorder, trailingRequest) + if trailingRecorder.Code != http.StatusBadRequest || !strings.Contains(trailingRecorder.Body.String(), "invalid_json") { + t.Fatalf("trailing JSON response=%d %s", trailingRecorder.Code, trailingRecorder.Body.String()) + } +} + +func TestReadRealtimeSyncSnapshotUsesOwnerScopeAndSingleCutoff(t *testing.T) { + queryer := newRealtimeSyncFakeQueryer("usr_owner") + query := realtimeSyncQuery{ + SessionID: "ses_sync", + TraceID: "trc_sync", + AfterOutboxSeq: 0, + Limit: 2, + Actor: sessionActor{ID: "usr_owner", Role: "user"}, + } + snapshot, err := readRealtimeSyncSnapshot(context.Background(), queryer, query) + if err != nil { + t.Fatalf("read sync snapshot: %v", err) + } + if snapshot.CutoffOutboxSeq != 3 || snapshot.CursorOutboxSeq != 2 || !snapshot.HasMore { + t.Fatalf("unexpected cursor boundary: %#v", snapshot) + } + if len(snapshot.Events) != 2 { + t.Fatalf("events=%d, want 2", len(snapshot.Events)) + } + first := snapshot.Events[0] + if first.OutboxEventID != "obe_1" || first.EntityFamily != "traceEvents" || first.EntityID != "evt_1" { + t.Fatalf("event identity fields were not returned: %#v", first) + } + if first.EventSeq != int64(101) || first.AggregateID != "trc_sync" || first.AggregateSeq != 1 || first.ProjectionRevision != 11 { + t.Fatalf("event aggregate fields were not returned: %#v", first) + } + if len(snapshot.Facts) != len(realtimeSyncFactSpecs) { + t.Fatalf("fact families=%d, want %d: %#v", len(snapshot.Facts), len(realtimeSyncFactSpecs), snapshot.Facts) + } + for _, spec := range realtimeSyncFactSpecs { + if len(snapshot.Facts[spec.Family]) != 1 { + t.Fatalf("facts[%s]=%#v, want one fact", spec.Family, snapshot.Facts[spec.Family]) + } + } + if len(queryer.calls) != 3+len(realtimeSyncFactSpecs) { + t.Fatalf("query calls=%v", queryer.calls) + } + if queryer.calls[0].kind != "scope" || queryer.calls[1].kind != "cutoff" || queryer.calls[2].kind != "events" { + t.Fatalf("snapshot boundary query order=%v", queryer.calls) + } + if got := queryer.calls[0].args; len(got) != 4 || got[2] != false || got[3] != "usr_owner" { + t.Fatalf("owner scope args=%#v", got) + } + if got := queryer.calls[2].args; len(got) != 5 || got[3] != int64(3) || got[4] != 3 { + t.Fatalf("events must use cutoff and limit+1: %#v", got) + } +} + +func TestReadRealtimeSyncSnapshotRejectsOtherOwnerBeforeReadingFacts(t *testing.T) { + if !strings.Contains(realtimeSyncResolveScopeSQL, "sessions.owner_user_id = $4") { + t.Fatalf("ordinary user scope must be constrained by workbench_sessions.owner_user_id: %s", realtimeSyncResolveScopeSQL) + } + queryer := newRealtimeSyncFakeQueryer("usr_owner") + _, err := readRealtimeSyncSnapshot(context.Background(), queryer, realtimeSyncQuery{ + SessionID: "ses_sync", + Limit: 100, + Actor: sessionActor{ID: "usr_other", Role: "user"}, + }) + if !errors.Is(err, sql.ErrNoRows) { + t.Fatalf("error=%v, want sql.ErrNoRows", err) + } + if len(queryer.calls) != 1 || queryer.calls[0].kind != "scope" { + t.Fatalf("unauthorized scope must stop before outbox/facts reads: %v", queryer.calls) + } +} + +func TestReadRealtimeSyncSnapshotClampsForeignCursorToScopedCutoff(t *testing.T) { + queryer := newRealtimeSyncFakeQueryer("usr_owner") + queryer.events = [][]any{} + snapshot, err := readRealtimeSyncSnapshot(context.Background(), queryer, realtimeSyncQuery{ + SessionID: "ses_sync", + AfterOutboxSeq: 999, + Limit: 100, + Actor: sessionActor{ID: "usr_owner", Role: "user"}, + }) + if err != nil { + t.Fatalf("read sync snapshot: %v", err) + } + if snapshot.CutoffOutboxSeq != 3 || snapshot.CursorOutboxSeq != 3 || snapshot.HasMore || len(snapshot.Events) != 0 { + t.Fatalf("foreign cursor was not clamped to scoped cutoff: %#v", snapshot) + } +} + +func TestReadRealtimeSyncSnapshotAllowsAdminWithoutOwnerFilter(t *testing.T) { + queryer := newRealtimeSyncFakeQueryer("usr_owner") + snapshot, err := readRealtimeSyncSnapshot(context.Background(), queryer, realtimeSyncQuery{ + TraceID: "trc_sync", + Limit: 3, + Actor: sessionActor{Role: "admin"}, + }) + if err != nil { + t.Fatalf("admin sync: %v", err) + } + if len(queryer.calls) == 0 || queryer.calls[0].args[2] != true || queryer.calls[0].args[3] != "" { + t.Fatalf("admin scope args=%#v", queryer.calls) + } + if snapshot.CursorOutboxSeq != snapshot.CutoffOutboxSeq || snapshot.HasMore { + t.Fatalf("admin complete page boundary=%#v", snapshot) + } +} + +type realtimeSyncFakeCall struct { + kind string + query string + args []any +} + +type realtimeSyncFakeQueryer struct { + ownerID string + sessionID string + cutoff int64 + events [][]any + facts map[string][][]any + calls []realtimeSyncFakeCall +} + +func newRealtimeSyncFakeQueryer(ownerID string) *realtimeSyncFakeQueryer { + facts := map[string][][]any{} + for _, spec := range realtimeSyncFactSpecs { + facts[spec.Table] = [][]any{{fmt.Sprintf(`{"family":%q,"sessionId":"ses_sync"}`, spec.Family)}} + } + return &realtimeSyncFakeQueryer{ + ownerID: ownerID, + sessionID: "ses_sync", + cutoff: 3, + events: [][]any{realtimeSyncTestEventRow(1), realtimeSyncTestEventRow(2), realtimeSyncTestEventRow(3)}, + facts: facts, + } +} + +func (queryer *realtimeSyncFakeQueryer) QueryRowContext(_ context.Context, query string, args ...any) realtimeSyncRow { + if query == realtimeSyncResolveScopeSQL { + queryer.record("scope", query, args) + isAdmin, _ := args[2].(bool) + actorID, _ := args[3].(string) + if !isAdmin && actorID != queryer.ownerID { + return realtimeSyncFakeRow{err: sql.ErrNoRows} + } + return realtimeSyncFakeRow{values: []any{queryer.sessionID}} + } + if query == realtimeSyncCutoffSQL { + queryer.record("cutoff", query, args) + return realtimeSyncFakeRow{values: []any{queryer.cutoff}} + } + return realtimeSyncFakeRow{err: fmt.Errorf("unexpected QueryRowContext: %s", query)} +} + +func (queryer *realtimeSyncFakeQueryer) QueryContext(_ context.Context, query string, args ...any) (realtimeSyncRows, error) { + if query == realtimeSyncEventsSQL { + queryer.record("events", query, args) + return &realtimeSyncFakeRows{values: queryer.events}, nil + } + for _, spec := range realtimeSyncFactSpecs { + if strings.Contains(query, "FROM "+spec.Table+" ") { + queryer.record("facts."+spec.Family, query, args) + return &realtimeSyncFakeRows{values: queryer.facts[spec.Table]}, nil + } + } + return nil, fmt.Errorf("unexpected QueryContext: %s", query) +} + +func (queryer *realtimeSyncFakeQueryer) record(kind string, query string, args []any) { + cloned := append([]any(nil), args...) + queryer.calls = append(queryer.calls, realtimeSyncFakeCall{kind: kind, query: query, args: cloned}) +} + +type realtimeSyncFakeRow struct { + values []any + err error +} + +func (row realtimeSyncFakeRow) Scan(destinations ...any) error { + if row.err != nil { + return row.err + } + return assignRealtimeSyncFakeValues(destinations, row.values) +} + +type realtimeSyncFakeRows struct { + values [][]any + index int + err error +} + +func (rows *realtimeSyncFakeRows) Close() error { return nil } + +func (rows *realtimeSyncFakeRows) Err() error { return rows.err } + +func (rows *realtimeSyncFakeRows) Next() bool { return rows.index < len(rows.values) } + +func (rows *realtimeSyncFakeRows) Scan(destinations ...any) error { + if rows.index >= len(rows.values) { + return errors.New("realtime sync fake rows exhausted") + } + values := rows.values[rows.index] + rows.index++ + return assignRealtimeSyncFakeValues(destinations, values) +} + +func assignRealtimeSyncFakeValues(destinations []any, values []any) error { + if len(destinations) != len(values) { + return fmt.Errorf("scan destinations=%d values=%d", len(destinations), len(values)) + } + for index, destination := range destinations { + value := values[index] + switch target := destination.(type) { + case *string: + if value == nil { + return fmt.Errorf("cannot scan nil into string") + } + *target = fmt.Sprint(value) + case *int64: + integer, ok := value.(int64) + if !ok { + return fmt.Errorf("cannot scan %T into int64", value) + } + *target = integer + case *sql.NullString: + if value == nil { + *target = sql.NullString{} + } else { + *target = sql.NullString{String: fmt.Sprint(value), Valid: true} + } + case *sql.NullInt64: + if value == nil { + *target = sql.NullInt64{} + } else if integer, ok := value.(int64); ok { + *target = sql.NullInt64{Int64: integer, Valid: true} + } else { + return fmt.Errorf("cannot scan %T into NullInt64", value) + } + case *sql.NullBool: + if value == nil { + *target = sql.NullBool{} + } else if boolean, ok := value.(bool); ok { + *target = sql.NullBool{Bool: boolean, Valid: true} + } else { + return fmt.Errorf("cannot scan %T into NullBool", value) + } + default: + return fmt.Errorf("unsupported scan destination %T", destination) + } + } + return nil +} + +func realtimeSyncTestEventRow(sequence int64) []any { + return []any{ + sequence, + fmt.Sprintf("obe_%d", sequence), + "traceEvents", + fmt.Sprintf("evt_%d", sequence), + int64(100) + sequence, + "trc_sync", + sequence, + int64(10) + sequence, + "trc_sync", + "ses_sync", + "turn_sync", + nil, + sequence, + sequence, + fmt.Sprintf("source_%d", sequence), + "event", + false, + false, + fmt.Sprintf(`{"sequence":%d}`, sequence), + "2026-07-10T12:00:00Z", + } +} diff --git a/internal/workbenchruntime/service.go b/internal/workbenchruntime/service.go index b81912a8..6db67401 100644 --- a/internal/workbenchruntime/service.go +++ b/internal/workbenchruntime/service.go @@ -258,6 +258,7 @@ func (s *Server) routes() { s.mux.HandleFunc("/v1/workbench-runtime/trace-events", s.handleTraceEvents) s.mux.HandleFunc("/v1/workbench-runtime/sessions", s.handleSessions) s.mux.HandleFunc("/v1/workbench-runtime/projection-outbox", s.handleProjectionOutbox) + s.mux.HandleFunc("/v1/workbench-runtime/sync", s.handleRealtimeSync) } func (s *Server) handleMetrics(w http.ResponseWriter, r *http.Request) {