feat: cache workbench sessions summaries in redis

2026-06-22 13:27:27 +08:00
parent ad2883fa2a
commit cf72e7038e
2 changed files with 319 additions and 6 deletions
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"strings"
 	"testing"
+	"time"

 	"github.com/redis/go-redis/v9"
 )
@@ -93,6 +94,70 @@ func TestDisabledCacheDoesNotFailReadPath(t *testing.T) {
 	}
 }

+func TestSessionsSummaryCacheTTLUsesShortTTLForUnstablePages(t *testing.T) {
+	config := cacheConfig{SessionsSummaryTTL: 30 * time.Second, RunningPageTTL: 5 * time.Second}
+	stable := map[string]any{"sessions": []any{map[string]any{"terminal": true, "running": false, "projectionStatus": "caught-up"}}}
+	ttl, unstable, reason := sessionsSummaryCacheTTL(stable, config)
+	if ttl != 30*time.Second || unstable || reason != "stable_page" {
+		t.Fatalf("stable ttl=%s unstable=%v reason=%s", ttl, unstable, reason)
+	}
+
+	running := map[string]any{"sessions": []any{map[string]any{"terminal": false, "running": true, "projectionStatus": "projecting"}}}
+	ttl, unstable, reason = sessionsSummaryCacheTTL(running, config)
+	if ttl != 5*time.Second || !unstable || reason != "running_or_projecting_page" {
+		t.Fatalf("running ttl=%s unstable=%v reason=%s", ttl, unstable, reason)
+	}
+
+	config.RunningPageTTL = 0
+	ttl, unstable, reason = sessionsSummaryCacheTTL(running, config)
+	if ttl != 0 || !unstable || reason != "running_page_ttl_disabled" {
+		t.Fatalf("disabled running ttl=%s unstable=%v reason=%s", ttl, unstable, reason)
+	}
+}
+
+func TestSessionsCacheStorablePayloadRemovesRuntimeOnlySecretKeys(t *testing.T) {
+	payload := map[string]any{
+		"ok":                   true,
+		"contractVersion":      "workbench-sessions-v1",
+		"secretMaterialStored": false,
+		"persistence": map[string]any{
+			"cache": map[string]any{"secretMaterialStored": false, "valuesRedacted": true},
+		},
+		"sessions": []any{map[string]any{"sessionId": "ses_1", "terminal": true, "projectionStatus": "caught-up", "valuesRedacted": true}},
+	}
+	stored := sessionsCacheStorablePayload(payload)
+	if _, ok := stored["secretMaterialStored"]; ok {
+		t.Fatalf("storable payload must not keep secretMaterialStored: %#v", stored)
+	}
+	if _, ok := stored["persistence"]; ok {
+		t.Fatalf("storable payload must not keep runtime persistence summary: %#v", stored)
+	}
+	entry := sessionsCacheEntry{CachedAt: time.Now().UTC().Format(time.RFC3339Nano), Payload: stored, PayloadBytes: 128, TTLMillis: 30000, ValuesRedacted: true}
+	if _, err := cachePayloadBytes(entry, 4096); err != nil {
+		t.Fatalf("storable sessions payload rejected: %v", err)
+	}
+}
+
+func TestSessionsSummaryCacheKeyActorIsolation(t *testing.T) {
+	config := cacheConfig{Enabled: true, RedisURL: "redis://localhost:6379/0", MaxKeyBytes: 512}
+	server := &Server{config: Config{Cache: config}, cache: newDerivedCache(config)}
+	query := sessionListQuery{Limit: 20, Actor: sessionActor{ID: "actor-a", Role: "user"}}
+	keyA, meta := server.sessionsSummaryCacheKey(query, 20, 0)
+	if keyA == "" || meta["status"] != "miss" || meta["enabled"] != true {
+		t.Fatalf("unexpected key/meta: key=%q meta=%#v", keyA, meta)
+	}
+	query.Actor.ID = "actor-b"
+	keyB, _ := server.sessionsSummaryCacheKey(query, 20, 0)
+	if keyA == keyB {
+		t.Fatalf("expected actor-isolated session cache keys, got %q", keyA)
+	}
+	query.Actor.ID = ""
+	keyMissingActor, meta := server.sessionsSummaryCacheKey(query, 20, 0)
+	if keyMissingActor != "" || meta["status"] != "skipped" || meta["skipReason"] != "missing_actor" {
+		t.Fatalf("expected missing actor skip, key=%q meta=%#v", keyMissingActor, meta)
+	}
+}
+
 func cacheErrKind(err error) string {
 	var cacheErr cacheError
 	if errors.As(err, &cacheErr) {
@@ -24,8 +24,17 @@ const (
 	sessionSummaryQueryMaxAttempts    = 2
 	sessionSummaryQueryAttemptTimeout = 2 * time.Second
 	sessionSummaryQueryRetryBackoff   = 150 * time.Millisecond
+	sessionsSummaryCacheClass         = "sessions.summary"
 )

+type sessionsCacheEntry struct {
+	CachedAt       string         `json:"cachedAt"`
+	Payload        map[string]any `json:"payload"`
+	PayloadBytes   int            `json:"payloadBytes"`
+	TTLMillis      int64          `json:"ttlMs"`
+	ValuesRedacted bool           `json:"valuesRedacted"`
+}
+
 type Config struct {
 	Addr           string
 	DatabaseURL    string
@@ -315,14 +324,80 @@ func (s *Server) handleSessions(w http.ResponseWriter, r *http.Request) {
 }

 func (s *Server) listSessions(ctx context.Context, query sessionListQuery) (map[string]any, error) {
-	limit := boundedLimit(query.Limit, 20, 100)
-	offset := query.Offset
-	if offset <= 0 {
-		offset = cursorOffset(query.Cursor)
+	limit, offset := sessionListPage(query)
+	cacheKey, cacheMeta := s.sessionsSummaryCacheKey(query, limit, offset)
+	if cacheKey != "" {
+		var entry sessionsCacheEntry
+		hit, err := s.cache.Get(ctx, cacheKey, &entry)
+		if err != nil {
+			setSessionsCacheStatus(cacheMeta, "unavailable")
+			cacheMeta["errorKind"] = classifyCacheError(err)
+		} else if hit && entry.Payload != nil {
+			ageMs, stale := sessionsCacheEntryAgeMs(entry, time.Now().UTC())
+			if !stale {
+				payload := sessionsCacheResponsePayload(entry.Payload, s.persistenceSummary())
+				setSessionsCacheStatus(cacheMeta, "hit")
+				cacheMeta["cacheAgeMs"] = ageMs
+				cacheMeta["dbQueryAvoided"] = true
+				cacheMeta["dbQueryDurationMs"] = int64(0)
+				cacheMeta["payloadBytes"] = entry.PayloadBytes
+				cacheMeta["ttlMs"] = entry.TTLMillis
+				attachSessionsCacheMetadata(payload, cacheMeta)
+				return payload, nil
+			}
+			setSessionsCacheStatus(cacheMeta, "stale")
+			cacheMeta["cacheAgeMs"] = ageMs
+		}
 	}
-	if offset < 0 {
-		offset = 0
+
+	dbStart := time.Now()
+	payload, err := s.listSessionsFromDB(ctx, query, limit, offset)
+	dbDuration := time.Since(dbStart)
+	if err != nil {
+		return nil, err
 	}
+	cacheMeta["dbQueryAvoided"] = false
+	cacheMeta["dbQueryDurationMs"] = dbDuration.Milliseconds()
+	if cacheKey == "" {
+		attachSessionsCacheMetadata(payload, cacheMeta)
+		return payload, nil
+	}
+
+	ttl, unstable, ttlReason := sessionsSummaryCacheTTL(payload, s.config.Cache)
+	cacheMeta["unstable"] = unstable
+	cacheMeta["ttlMs"] = ttl.Milliseconds()
+	cacheMeta["ttlReason"] = ttlReason
+	storedPayload := sessionsCacheStorablePayload(payload)
+	payloadBytes, payloadErr := cachePayloadBytes(storedPayload, s.config.Cache.MaxPayloadBytes)
+	if payloadErr != nil {
+		setSessionsCacheStatus(cacheMeta, "skipped")
+		cacheMeta["skipReason"] = classifyCacheError(payloadErr)
+		attachSessionsCacheMetadata(payload, cacheMeta)
+		return payload, nil
+	}
+	cacheMeta["payloadBytes"] = len(payloadBytes)
+	if ttl <= 0 {
+		setSessionsCacheStatus(cacheMeta, "skipped")
+		cacheMeta["skipReason"] = ttlReason
+		attachSessionsCacheMetadata(payload, cacheMeta)
+		return payload, nil
+	}
+	entry := sessionsCacheEntry{CachedAt: time.Now().UTC().Format(time.RFC3339Nano), Payload: storedPayload, PayloadBytes: len(payloadBytes), TTLMillis: ttl.Milliseconds(), ValuesRedacted: true}
+	if err := s.cache.Set(ctx, cacheKey, entry, ttl); err != nil {
+		setSessionsCacheStatus(cacheMeta, "set_failed")
+		cacheMeta["errorKind"] = classifyCacheError(err)
+		attachSessionsCacheMetadata(payload, cacheMeta)
+		return payload, nil
+	}
+	if cacheMeta["status"] != "stale" {
+		setSessionsCacheStatus(cacheMeta, "miss")
+	}
+	cacheMeta["stored"] = true
+	attachSessionsCacheMetadata(payload, cacheMeta)
+	return payload, nil
+}
+
+func (s *Server) listSessionsFromDB(ctx context.Context, query sessionListQuery, limit int, offset int) (map[string]any, error) {
 	ownerUserID := strings.TrimSpace(query.Actor.ID)
 	pageFacts, err := s.queryFacts(ctx, factQuery{OwnerUserID: ownerUserID, Limit: limit + offset + 1, Families: []string{"sessions"}, SessionsOrder: "updated_desc", SessionProjection: "summary"})
 	if err != nil {
@@ -370,6 +445,179 @@ func (s *Server) listSessions(ctx context.Context, query sessionListQuery) (map[
 	return map[string]any{"ok": true, "status": "succeeded", "contractVersion": "workbench-sessions-v1", "sessions": summaries, "count": len(summaries), "cursor": cursorOrNil(offset), "hasMore": hasMore, "nextCursor": nextCursor(offset, limit, hasMore), "persistence": s.persistenceSummary(), "servedBy": serviceID, "valuesRedacted": true, "secretMaterialStored": false}, nil
 }

+func sessionListPage(query sessionListQuery) (int, int) {
+	limit := boundedLimit(query.Limit, 20, 100)
+	offset := query.Offset
+	if offset <= 0 {
+		offset = cursorOffset(query.Cursor)
+	}
+	if offset < 0 {
+		offset = 0
+	}
+	return limit, offset
+}
+
+func (s *Server) sessionsSummaryCacheKey(query sessionListQuery, limit int, offset int) (string, map[string]any) {
+	meta := sessionsCacheMetadata("disabled")
+	if s == nil || s.cache == nil || !s.config.Cache.Enabled {
+		return "", meta
+	}
+	meta["enabled"] = true
+	actorID := strings.TrimSpace(query.Actor.ID)
+	if actorID == "" {
+		setSessionsCacheStatus(meta, "skipped")
+		meta["skipReason"] = "missing_actor"
+		return "", meta
+	}
+	includeID := strings.TrimSpace(query.IncludeSessionID)
+	actorRole := strings.TrimSpace(query.Actor.Role)
+	cursor := fmt.Sprintf("idx:%d|limit:%d|include:%s|role:%s", offset, limit, includeID, actorRole)
+	authority := fmt.Sprintf("contract=workbench-sessions-v1|limit=%d|offset=%d|include=%t|role=%s", limit, offset, includeID != "", actorRole)
+	key, err := s.cache.BuildKey(cacheKeyParts{Class: sessionsSummaryCacheClass, ActorID: actorID, SessionID: includeID, Cursor: cursor, ProjectionRevision: "workbench-sessions-v1", Authority: authority})
+	if err != nil {
+		setSessionsCacheStatus(meta, "skipped")
+		meta["skipReason"] = classifyCacheError(err)
+		return "", meta
+	}
+	setSessionsCacheStatus(meta, "miss")
+	return key, meta
+}
+
+func sessionsCacheMetadata(status string) map[string]any {
+	meta := map[string]any{
+		"class":                sessionsSummaryCacheClass,
+		"schemaVersion":        cacheKeySchemaVersion,
+		"enabled":              false,
+		"status":               status,
+		"hit":                  false,
+		"miss":                 false,
+		"stale":                false,
+		"unavailable":          false,
+		"dbQueryAvoided":       false,
+		"cacheAgeMs":           nil,
+		"dbQueryDurationMs":    nil,
+		"payloadBytes":         nil,
+		"ttlMs":                nil,
+		"valuesRedacted":       true,
+		"secretMaterialStored": false,
+	}
+	setSessionsCacheStatus(meta, status)
+	return meta
+}
+
+func setSessionsCacheStatus(meta map[string]any, status string) {
+	if meta == nil {
+		return
+	}
+	meta["status"] = status
+	meta["hit"] = status == "hit"
+	meta["miss"] = status == "miss"
+	meta["stale"] = status == "stale"
+	meta["unavailable"] = status == "unavailable" || status == "set_failed"
+}
+
+func attachSessionsCacheMetadata(payload map[string]any, meta map[string]any) {
+	if payload == nil || meta == nil {
+		return
+	}
+	payload["cache"] = meta
+}
+
+func sessionsSummaryCacheTTL(payload map[string]any, config cacheConfig) (time.Duration, bool, string) {
+	if sessionsSummaryPayloadUnstable(payload) {
+		if config.RunningPageTTL <= 0 {
+			return 0, true, "running_page_ttl_disabled"
+		}
+		return config.RunningPageTTL, true, "running_or_projecting_page"
+	}
+	if config.SessionsSummaryTTL <= 0 {
+		return 0, false, "sessions_summary_ttl_disabled"
+	}
+	return config.SessionsSummaryTTL, false, "stable_page"
+}
+
+func sessionsSummaryPayloadUnstable(payload map[string]any) bool {
+	for _, row := range sessionsPayloadRows(payload["sessions"]) {
+		item := objectMap(row)
+		if len(item) == 0 {
+			continue
+		}
+		if boolValue(item["running"]) || !boolValue(item["terminal"]) {
+			return true
+		}
+		if status := normalizeProjectionStatus(text(item["projectionStatus"])); status != "" && status != "caught-up" {
+			return true
+		}
+		turnSummary := objectMap(item["turnSummary"])
+		if len(turnSummary) > 0 && (boolValue(turnSummary["running"]) || !boolValue(turnSummary["terminal"])) {
+			return true
+		}
+	}
+	return false
+}
+
+func sessionsPayloadRows(value any) []any {
+	switch rows := value.(type) {
+	case []any:
+		return rows
+	case []map[string]any:
+		out := make([]any, 0, len(rows))
+		for _, row := range rows {
+			out = append(out, row)
+		}
+		return out
+	default:
+		return []any{}
+	}
+}
+
+func sessionsCacheStorablePayload(payload map[string]any) map[string]any {
+	clone := cloneJSONMap(payload)
+	delete(clone, "cache")
+	delete(clone, "persistence")
+	delete(clone, "secretMaterialStored")
+	return clone
+}
+
+func sessionsCacheResponsePayload(payload map[string]any, persistence map[string]any) map[string]any {
+	clone := cloneJSONMap(payload)
+	clone["persistence"] = persistence
+	clone["secretMaterialStored"] = false
+	clone["valuesRedacted"] = true
+	clone["servedBy"] = serviceID
+	return clone
+}
+
+func sessionsCacheEntryAgeMs(entry sessionsCacheEntry, now time.Time) (int64, bool) {
+	cachedAt, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(entry.CachedAt))
+	if err != nil {
+		return 0, true
+	}
+	age := now.Sub(cachedAt)
+	if age < 0 {
+		age = 0
+	}
+	if entry.TTLMillis > 0 && age.Milliseconds() > entry.TTLMillis {
+		return age.Milliseconds(), true
+	}
+	return age.Milliseconds(), false
+}
+
+func cloneJSONMap(value map[string]any) map[string]any {
+	if value == nil {
+		return map[string]any{}
+	}
+	payload, err := json.Marshal(value)
+	if err != nil {
+		return map[string]any{}
+	}
+	var clone map[string]any
+	if err := json.Unmarshal(payload, &clone); err != nil || clone == nil {
+		return map[string]any{}
+	}
+	return clone
+}
+
 func (s *Server) queryIncludedSession(ctx context.Context, routeID string) (map[string][]any, error) {
 	if strings.HasPrefix(routeID, "ses_") {
 		return s.queryFacts(ctx, factQuery{SessionID: routeID, Limit: 1, Families: []string{"sessions"}, SessionProjection: "summary"})