feat: add atomic Workbench realtime sync
This commit is contained in:
root
@@ -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
|
||||
}
|
||||
@@ -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",
|
||||
}
|
||||
}
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user