pikasTech-HWLAB/internal/workbenchruntime/cache.go

// Redis derived cache boundary follows PJ2026-01060505, PJ2026-0104010803,
// PJ2026-010403 and implementation version draft-2026-06-22-p1-workbench-redis-derived-cache.
package workbenchruntime

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/redis/go-redis/v9"
)

const cacheKeySchemaVersion = "wb-derived-v1"

var errCacheDisabled = errors.New("workbench redis cache disabled")

type cacheConfig struct {
	Enabled                        bool
	RedisURL                       string
	ConnectTimeout                 time.Duration
	OperationTimeout               time.Duration
	MaxKeyBytes                    int
	MaxPayloadBytes                int
	SessionsSummaryTTL             time.Duration
	TerminalTurnTTL                time.Duration
	TerminalTracePageTTL           time.Duration
	RunningPageTTL                 time.Duration
	UnavailableLivenessFailure     bool
	UnavailableReadMode            string
	UnavailableFallbackStateSource bool
}

type cacheKeyParts struct {
	Class              string
	ActorID            string
	SessionID          string
	TraceID            string
	Cursor             string
	ProjectionRevision string
	ProjectionSeq      int64
	Authority          string
}

type derivedCache struct {
	config      cacheConfig
	client      *redis.Client
	configured  bool
	initErr     error
	metrics     cacheMetrics
	createdTime time.Time
}

type cacheMetrics struct {
	mu            sync.Mutex
	Hits          uint64
	Misses        uint64
	Sets          uint64
	Deletes       uint64
	Errors        uint64
	LastErrorKind string
	LastErrorOp   string
	LastErrorAt   time.Time
}

type cacheError struct {
	Op   string
	Kind string
	Err  error
}

func (e cacheError) Error() string {
	if e.Err == nil {
		return e.Kind
	}
	return e.Kind + ": " + e.Err.Error()
}

func (e cacheError) Unwrap() error { return e.Err }

func newDerivedCache(config cacheConfig) *derivedCache {
	cache := &derivedCache{config: config, configured: strings.TrimSpace(config.RedisURL) != "", createdTime: time.Now().UTC()}
	if !config.Enabled {
		return cache
	}
	if strings.TrimSpace(config.RedisURL) == "" {
		cache.initErr = cacheError{Op: "init", Kind: "cache_config_missing_url", Err: errors.New("redis url is empty")}
		return cache
	}
	options, err := redis.ParseURL(config.RedisURL)
	if err != nil {
		cache.initErr = cacheError{Op: "init", Kind: "cache_config_invalid_url", Err: err}
		return cache
	}
	if config.ConnectTimeout > 0 {
		options.DialTimeout = config.ConnectTimeout
	}
	if config.OperationTimeout > 0 {
		options.ReadTimeout = config.OperationTimeout
		options.WriteTimeout = config.OperationTimeout
	}
	cache.client = redis.NewClient(options)
	return cache
}

func cacheConfigFromEnv() cacheConfig {
	return cacheConfig{
		Enabled:                        envBool("HWLAB_WORKBENCH_CACHE_REDIS_ENABLED", false),
		RedisURL:                       first(os.Getenv("HWLAB_WORKBENCH_CACHE_REDIS_URL")),
		ConnectTimeout:                 time.Duration(envInt("HWLAB_WORKBENCH_CACHE_REDIS_CONNECT_TIMEOUT_MS", 200)) * time.Millisecond,
		OperationTimeout:               time.Duration(envInt("HWLAB_WORKBENCH_CACHE_REDIS_OPERATION_TIMEOUT_MS", 100)) * time.Millisecond,
		MaxKeyBytes:                    envInt("HWLAB_WORKBENCH_CACHE_REDIS_MAX_KEY_BYTES", 512),
		MaxPayloadBytes:                envInt("HWLAB_WORKBENCH_CACHE_REDIS_MAX_PAYLOAD_BYTES", 262144),
		SessionsSummaryTTL:             time.Duration(envInt("HWLAB_WORKBENCH_CACHE_REDIS_TTL_SESSIONS_SUMMARY_MS", 30000)) * time.Millisecond,
		TerminalTurnTTL:                time.Duration(envInt("HWLAB_WORKBENCH_CACHE_REDIS_TTL_TERMINAL_TURN_MS", 300000)) * time.Millisecond,
		TerminalTracePageTTL:           time.Duration(envInt("HWLAB_WORKBENCH_CACHE_REDIS_TTL_TERMINAL_TRACE_PAGE_MS", 300000)) * time.Millisecond,
		RunningPageTTL:                 time.Duration(envInt("HWLAB_WORKBENCH_CACHE_REDIS_TTL_RUNNING_PAGE_MS", 5000)) * time.Millisecond,
		UnavailableLivenessFailure:     envBool("HWLAB_WORKBENCH_CACHE_REDIS_UNAVAILABLE_LIVENESS_FAILURE", false),
		UnavailableReadMode:            first(os.Getenv("HWLAB_WORKBENCH_CACHE_REDIS_UNAVAILABLE_READ_MODE"), "db-or-degraded"),
		UnavailableFallbackStateSource: envBool("HWLAB_WORKBENCH_CACHE_REDIS_UNAVAILABLE_FALLBACK_STATE_SOURCE", false),
	}
}

func (c *derivedCache) Close() error {
	if c == nil || c.client == nil {
		return nil
	}
	return c.client.Close()
}

func (c *derivedCache) BuildKey(parts cacheKeyParts) (string, error) {
	if c == nil {
		return "", errCacheDisabled
	}
	return buildCacheKey(c.config.MaxKeyBytes, parts)
}

func (c *derivedCache) Get(ctx context.Context, key string, target any) (bool, error) {
	if !c.usable() {
		return false, nil
	}
	if err := validateCacheKey(c.config.MaxKeyBytes, key); err != nil {
		wrapped := cacheError{Op: "get", Kind: classifyCacheError(err), Err: err}
		c.metrics.recordError(wrapped)
		return false, wrapped
	}
	var payload []byte
	err := c.withTimeout(ctx, func(opCtx context.Context) error {
		var err error
		payload, err = c.client.Get(opCtx, key).Bytes()
		return err
	})
	if errors.Is(err, redis.Nil) {
		c.metrics.recordMiss()
		return false, nil
	}
	if err != nil {
		wrapped := cacheError{Op: "get", Kind: classifyCacheError(err), Err: err}
		c.metrics.recordError(wrapped)
		return false, wrapped
	}
	if len(payload) > c.config.MaxPayloadBytes {
		wrapped := cacheError{Op: "get", Kind: "cache_payload_too_large", Err: fmt.Errorf("payload bytes %d exceeds %d", len(payload), c.config.MaxPayloadBytes)}
		c.metrics.recordError(wrapped)
		return false, wrapped
	}
	if target != nil {
		if err := json.Unmarshal(payload, target); err != nil {
			wrapped := cacheError{Op: "get", Kind: "cache_payload_decode_failed", Err: err}
			c.metrics.recordError(wrapped)
			return false, wrapped
		}
	}
	c.metrics.recordHit()
	return true, nil
}

func (c *derivedCache) Set(ctx context.Context, key string, value any, ttl time.Duration) error {
	if !c.usable() {
		return nil
	}
	if err := validateCacheKey(c.config.MaxKeyBytes, key); err != nil {
		wrapped := cacheError{Op: "set", Kind: classifyCacheError(err), Err: err}
		c.metrics.recordError(wrapped)
		return wrapped
	}
	payload, err := cachePayloadBytes(value, c.config.MaxPayloadBytes)
	if err != nil {
		wrapped := cacheError{Op: "set", Kind: classifyCacheError(err), Err: err}
		c.metrics.recordError(wrapped)
		return wrapped
	}
	err = c.withTimeout(ctx, func(opCtx context.Context) error {
		return c.client.Set(opCtx, key, payload, ttl).Err()
	})
	if err != nil {
		wrapped := cacheError{Op: "set", Kind: classifyCacheError(err), Err: err}
		c.metrics.recordError(wrapped)
		return wrapped
	}
	c.metrics.recordSet()
	return nil
}

func (c *derivedCache) Delete(ctx context.Context, keys ...string) error {
	if !c.usable() || len(keys) == 0 {
		return nil
	}
	for _, key := range keys {
		if err := validateCacheKey(c.config.MaxKeyBytes, key); err != nil {
			wrapped := cacheError{Op: "delete", Kind: classifyCacheError(err), Err: err}
			c.metrics.recordError(wrapped)
			return wrapped
		}
	}
	err := c.withTimeout(ctx, func(opCtx context.Context) error {
		return c.client.Del(opCtx, keys...).Err()
	})
	if err != nil {
		wrapped := cacheError{Op: "delete", Kind: classifyCacheError(err), Err: err}
		c.metrics.recordError(wrapped)
		return wrapped
	}
	c.metrics.recordDelete(uint64(len(keys)))
	return nil
}

func (c *derivedCache) TTL(ctx context.Context, key string) (time.Duration, error) {
	if !c.usable() {
		return 0, nil
	}
	if err := validateCacheKey(c.config.MaxKeyBytes, key); err != nil {
		wrapped := cacheError{Op: "ttl", Kind: classifyCacheError(err), Err: err}
		c.metrics.recordError(wrapped)
		return 0, wrapped
	}
	var ttl time.Duration
	err := c.withTimeout(ctx, func(opCtx context.Context) error {
		var err error
		ttl, err = c.client.TTL(opCtx, key).Result()
		return err
	})
	if err != nil {
		wrapped := cacheError{Op: "ttl", Kind: classifyCacheError(err), Err: err}
		c.metrics.recordError(wrapped)
		return 0, wrapped
	}
	return ttl, nil
}

func (c *derivedCache) Summary() map[string]any {
	if c == nil {
		return map[string]any{"enabled": false, "configured": false, "role": "derived-read-cache", "valuesRedacted": true}
	}
	result := map[string]any{
		"enabled":                        c.config.Enabled,
		"configured":                     c.config.RedisURL != "",
		"role":                           "derived-read-cache",
		"adapter":                        "redis",
		"schemaVersion":                  cacheKeySchemaVersion,
		"maxKeyBytes":                    c.config.MaxKeyBytes,
		"maxPayloadBytes":                c.config.MaxPayloadBytes,
		"unavailableReadMode":            c.config.UnavailableReadMode,
		"unavailableFallbackStateSource": c.config.UnavailableFallbackStateSource,
		"secretMaterialStored":           false,
		"valuesRedacted":                 true,
		"metrics":                        c.metrics.snapshot(),
	}
	if c.initErr != nil {
		result["degraded"] = true
		result["errorKind"] = classifyCacheError(c.initErr)
	}
	return result
}

func (c *derivedCache) Diagnostic(ctx context.Context) map[string]any {
	result := c.Summary()
	if c == nil || !c.config.Enabled {
		result["available"] = false
		return result
	}
	if c.initErr != nil || c.client == nil {
		result["available"] = false
		result["degraded"] = true
		return result
	}
	err := c.withTimeout(ctx, func(opCtx context.Context) error {
		return c.client.Ping(opCtx).Err()
	})
	if err != nil {
		wrapped := cacheError{Op: "ping", Kind: classifyCacheError(err), Err: err}
		c.metrics.recordError(wrapped)
		result["available"] = false
		result["degraded"] = true
		result["errorKind"] = wrapped.Kind
		result["metrics"] = c.metrics.snapshot()
		return result
	}
	result["available"] = true
	result["degraded"] = false
	return result
}

func (c *derivedCache) usable() bool {
	return c != nil && c.config.Enabled && c.client != nil && c.initErr == nil
}

func (c *derivedCache) withTimeout(ctx context.Context, fn func(context.Context) error) error {
	if ctx == nil {
		ctx = context.Background()
	}
	if c == nil || c.config.OperationTimeout <= 0 {
		return fn(ctx)
	}
	opCtx, cancel := context.WithTimeout(ctx, c.config.OperationTimeout)
	defer cancel()
	return fn(opCtx)
}

func buildCacheKey(maxBytes int, parts cacheKeyParts) (string, error) {
	if strings.TrimSpace(parts.Class) == "" {
		return "", cacheError{Op: "key", Kind: "cache_key_invalid", Err: errors.New("class is required")}
	}
	if strings.TrimSpace(parts.ActorID) == "" {
		return "", cacheError{Op: "key", Kind: "cache_key_invalid", Err: errors.New("actor id is required")}
	}
	if strings.TrimSpace(parts.Authority) == "" && strings.TrimSpace(parts.ProjectionRevision) == "" && parts.ProjectionSeq <= 0 {
		return "", cacheError{Op: "key", Kind: "cache_key_invalid", Err: errors.New("projection authority is required")}
	}
	components := []string{
		"hwlab",
		"workbench",
		cacheKeySchemaVersion,
		"class=" + safeCacheToken(parts.Class),
		"actor=" + cacheHash(parts.ActorID),
		"session=" + optionalCacheHash(parts.SessionID),
		"trace=" + optionalCacheHash(parts.TraceID),
		"cursor=" + optionalCacheHash(parts.Cursor),
	}
	if parts.ProjectionSeq > 0 {
		components = append(components, "seq="+strconv.FormatInt(parts.ProjectionSeq, 10))
	}
	if strings.TrimSpace(parts.ProjectionRevision) != "" {
		components = append(components, "rev="+cacheHash(parts.ProjectionRevision))
	}
	if strings.TrimSpace(parts.Authority) != "" {
		components = append(components, "auth="+cacheHash(parts.Authority))
	}
	key := strings.Join(components, ":")
	if maxBytes <= 0 {
		maxBytes = 512
	}
	if len([]byte(key)) <= maxBytes {
		return key, nil
	}
	compact := strings.Join(components[:4], ":") + ":hash=" + cacheHash(key)
	if len([]byte(compact)) > maxBytes {
		return "", cacheError{Op: "key", Kind: "cache_key_too_large", Err: fmt.Errorf("key bytes exceed %d", maxBytes)}
	}
	return compact, nil
}

func validateCacheKey(maxBytes int, key string) error {
	if strings.TrimSpace(key) == "" {
		return cacheError{Op: "key", Kind: "cache_key_invalid", Err: errors.New("cache key is empty")}
	}
	if maxBytes <= 0 {
		maxBytes = 512
	}
	if len([]byte(key)) > maxBytes {
		return cacheError{Op: "key", Kind: "cache_key_too_large", Err: fmt.Errorf("key bytes exceed %d", maxBytes)}
	}
	return nil
}

func cachePayloadBytes(value any, maxBytes int) ([]byte, error) {
	payload, err := json.Marshal(value)
	if err != nil {
		return nil, cacheError{Op: "payload", Kind: "cache_payload_encode_failed", Err: err}
	}
	var decoded any
	if err := json.Unmarshal(payload, &decoded); err != nil {
		return nil, cacheError{Op: "payload", Kind: "cache_payload_decode_failed", Err: err}
	}
	if key, ok := cachePayloadSensitiveKey(decoded); ok {
		return nil, cacheError{Op: "payload", Kind: "cache_payload_rejected", Err: fmt.Errorf("sensitive payload key %q is not cacheable", key)}
	}
	if maxBytes <= 0 {
		maxBytes = 262144
	}
	if len(payload) > maxBytes {
		return nil, cacheError{Op: "payload", Kind: "cache_payload_too_large", Err: fmt.Errorf("payload bytes %d exceeds %d", len(payload), maxBytes)}
	}
	return payload, nil
}

func cachePayloadSensitiveKey(value any) (string, bool) {
	switch item := value.(type) {
	case map[string]any:
		for key, child := range item {
			if cachePayloadKeyRejected(key) {
				return key, true
			}
			if nested, ok := cachePayloadSensitiveKey(child); ok {
				return nested, true
			}
		}
	case []any:
		for _, child := range item {
			if nested, ok := cachePayloadSensitiveKey(child); ok {
				return nested, true
			}
		}
	case []map[string]any:
		for _, child := range item {
			if nested, ok := cachePayloadSensitiveKey(child); ok {
				return nested, true
			}
		}
	}
	return "", false
}

func cachePayloadKeyRejected(key string) bool {
	normalized := normalizedCachePayloadKey(key)
	for _, fragment := range []string{"authorization", "cookie", "secret", "password", "token", "apikey", "databaseurl", "dbdsn", "dsn", "prompt", "providerpayload", "stdout", "stderr"} {
		if strings.Contains(normalized, fragment) {
			return true
		}
	}
	return false
}

func normalizedCachePayloadKey(key string) string {
	lower := strings.ToLower(strings.TrimSpace(key))
	var builder strings.Builder
	for _, ch := range lower {
		if ch >= 'a' && ch <= 'z' || ch >= '0' && ch <= '9' {
			builder.WriteRune(ch)
		}
	}
	return builder.String()
}

func classifyCacheError(err error) string {
	if err == nil {
		return ""
	}
	var cacheErr cacheError
	if errors.As(err, &cacheErr) && cacheErr.Kind != "" {
		return cacheErr.Kind
	}
	if errors.Is(err, redis.Nil) {
		return "cache_miss"
	}
	if errors.Is(err, context.Canceled) {
		return "cache_context_canceled"
	}
	if errors.Is(err, context.DeadlineExceeded) {
		return "cache_timeout"
	}
	var netErr net.Error
	if errors.As(err, &netErr) && netErr.Timeout() {
		return "cache_timeout"
	}
	message := strings.ToLower(err.Error())
	switch {
	case strings.Contains(message, "connection refused") || strings.Contains(message, "no such host") || strings.Contains(message, "network is unreachable") || strings.Contains(message, "i/o timeout"):
		return "cache_unavailable"
	case strings.Contains(message, "payload") && strings.Contains(message, "large"):
		return "cache_payload_too_large"
	case strings.Contains(message, "sensitive payload"):
		return "cache_payload_rejected"
	case strings.Contains(message, "key") && strings.Contains(message, "exceed"):
		return "cache_key_too_large"
	case strings.Contains(message, "key"):
		return "cache_key_invalid"
	default:
		return "cache_error"
	}
}

func safeCacheToken(value string) string {
	text := strings.ToLower(strings.TrimSpace(value))
	var builder strings.Builder
	for _, ch := range text {
		switch {
		case ch >= 'a' && ch <= 'z' || ch >= '0' && ch <= '9':
			builder.WriteRune(ch)
		case ch == '-' || ch == '_' || ch == '.':
			builder.WriteRune(ch)
		default:
			builder.WriteRune('-')
		}
	}
	result := strings.Trim(builder.String(), "-._")
	if result == "" {
		return "unknown"
	}
	return result
}

func optionalCacheHash(value string) string {
	if strings.TrimSpace(value) == "" {
		return "none"
	}
	return cacheHash(value)
}

func cacheHash(value string) string {
	digest := sha256.Sum256([]byte(strings.TrimSpace(value)))
	return hex.EncodeToString(digest[:])[:24]
}

func (m *cacheMetrics) recordHit() {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.Hits++
}

func (m *cacheMetrics) recordMiss() {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.Misses++
}

func (m *cacheMetrics) recordSet() {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.Sets++
}

func (m *cacheMetrics) recordDelete(count uint64) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.Deletes += count
}

func (m *cacheMetrics) recordError(err error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.Errors++
	m.LastErrorKind = classifyCacheError(err)
	var cacheErr cacheError
	if errors.As(err, &cacheErr) {
		m.LastErrorOp = cacheErr.Op
	}
	m.LastErrorAt = time.Now().UTC()
}

func (m *cacheMetrics) snapshot() map[string]any {
	m.mu.Lock()
	defer m.mu.Unlock()
	result := map[string]any{
		"hits":           int64(m.Hits),
		"misses":         int64(m.Misses),
		"sets":           int64(m.Sets),
		"deletes":        int64(m.Deletes),
		"errors":         int64(m.Errors),
		"valuesRedacted": true,
	}
	if m.LastErrorKind != "" {
		result["lastErrorKind"] = m.LastErrorKind
		result["lastErrorOp"] = m.LastErrorOp
		result["lastErrorAt"] = m.LastErrorAt.Format(time.RFC3339Nano)
	}
	return result
}

func envBool(name string, fallback bool) bool {
	value := strings.ToLower(strings.TrimSpace(os.Getenv(name)))
	if value == "" {
		return fallback
	}
	switch value {
	case "1", "true", "yes", "y", "on", "enabled":
		return true
	case "0", "false", "no", "n", "off", "disabled":
		return false
	default:
		return fallback
	}
}