meta-core/internal/storage/client.go

package storage

import (
	"context"
	"fmt"
	"log"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

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

// Client wraps Redis operations for metadata storage
type Client struct {
	client    *redis.Client
	prefix    string
	connected bool
	mu        sync.RWMutex
}

// NewClient creates a new storage client
func NewClient(prefix string) *Client {
	return &Client{
		prefix: prefix,
	}
}

// Connect connects to Redis at the given URL
func (c *Client) Connect(url string) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Parse Redis URL (redis://host:port)
	addr := strings.TrimPrefix(url, "redis://")

	c.client = redis.NewClient(&redis.Options{
		Addr:         addr,
		DialTimeout:  5 * time.Second,
		ReadTimeout:  30 * time.Second,
		WriteTimeout: 30 * time.Second,
		PoolSize:     10,
	})

	// Test connection
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	if err := c.client.Ping(ctx).Err(); err != nil {
		return fmt.Errorf("failed to connect to Redis: %w", err)
	}

	c.connected = true
	log.Printf("[Storage] Connected to Redis at %s", addr)
	return nil
}

// Close closes the Redis connection
func (c *Client) Close() error {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.client != nil {
		c.connected = false
		return c.client.Close()
	}
	return nil
}

// IsConnected returns true if connected to Redis
func (c *Client) IsConnected() bool {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.connected
}

// Health checks if Redis is responding
func (c *Client) Health() bool {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil || !c.connected {
		return false
	}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	return c.client.Ping(ctx).Err() == nil
}

// buildKey constructs the full key with prefix
func (c *Client) buildKey(key string) string {
	if c.prefix != "" {
		return c.prefix + key
	}
	return key
}

// buildKeyPrefix constructs the prefix for a file's flat keys
func (c *Client) buildKeyPrefix(hashID string) string {
	return c.buildKey(fmt.Sprintf("file:%s/", hashID))
}

// buildIndexKey constructs the key for the file index set
func (c *Client) buildIndexKey() string {
	return c.buildKey("file:__index__")
}

// GetMetadataFlat retrieves all metadata for a file as a flat map
// Uses flat keys: SCAN file:{hashId}/* then MGET
func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return nil, fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	prefix := c.buildKeyPrefix(hashID)
	result := make(map[string]string)

	// Scan for all keys with this prefix
	var cursor uint64
	for {
		keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
		if err != nil {
			return nil, fmt.Errorf("scan failed: %w", err)
		}

		if len(keys) > 0 {
			values, err := c.client.MGet(ctx, keys...).Result()
			if err != nil {
				return nil, fmt.Errorf("mget failed: %w", err)
			}
			for i, key := range keys {
				field := strings.TrimPrefix(key, prefix)
				if values[i] != nil {
					result[field] = values[i].(string)
				}
			}
		}

		cursor = nextCursor
		if cursor == 0 {
			break
		}
	}

	if len(result) == 0 {
		return nil, nil
	}

	return result, nil
}

// SetMetadataFlat stores metadata for a file using flat keys
// Uses pipeline SET for each key: file:{hashId}/{property}
func (c *Client) SetMetadataFlat(hashID string, metadata map[string]string) error {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return fmt.Errorf("not connected")
	}

	if len(metadata) == 0 {
		return nil
	}

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	prefix := c.buildKeyPrefix(hashID)
	pipe := c.client.Pipeline()

	for field, value := range metadata {
		pipe.Set(ctx, prefix+field, value, 0)
	}

	if _, err := pipe.Exec(ctx); err != nil {
		return fmt.Errorf("pipeline exec failed: %w", err)
	}

	// Add to index set
	if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
		return fmt.Errorf("sadd index failed: %w", err)
	}

	return nil
}

// GetAllHashIDs returns all unique file hash IDs stored
// Uses Redis Set: SMEMBERS file:__index__
func (c *Client) GetAllHashIDs() ([]string, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return nil, fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	indexKey := c.buildIndexKey()
	result, err := c.client.SMembers(ctx, indexKey).Result()
	if err != nil {
		return nil, fmt.Errorf("smembers failed: %w", err)
	}

	return result, nil
}

// GetProperty retrieves a single property value
// Uses flat key: GET file:{hashId}/{property}
func (c *Client) GetProperty(hashID, property string) (string, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return "", fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	key := c.buildKeyPrefix(hashID) + property
	result, err := c.client.Get(ctx, key).Result()
	if err == redis.Nil {
		return "", nil
	}
	return result, err
}

// SetProperty sets a single property value
// Uses flat key: SET file:{hashId}/{property} {value}
func (c *Client) SetProperty(hashID, property, value string) error {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	key := c.buildKeyPrefix(hashID) + property

	// Use SET to set the key
	if err := c.client.Set(ctx, key, value, 0).Err(); err != nil {
		return fmt.Errorf("set failed: %w", err)
	}

	// Add to index set
	if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
		return fmt.Errorf("sadd index failed: %w", err)
	}

	return nil
}

// DeleteMetadata deletes all metadata for a file
// Uses SCAN + DEL for flat keys: file:{hashId}/*
func (c *Client) DeleteMetadata(hashID string) (int64, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return 0, fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	prefix := c.buildKeyPrefix(hashID)

	// Scan and delete all keys with this prefix
	var deletedCount int64
	var cursor uint64
	for {
		keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
		if err != nil {
			return deletedCount, fmt.Errorf("scan failed: %w", err)
		}

		if len(keys) > 0 {
			deleted, err := c.client.Del(ctx, keys...).Result()
			if err != nil {
				return deletedCount, fmt.Errorf("del failed: %w", err)
			}
			deletedCount += deleted
		}

		cursor = nextCursor
		if cursor == 0 {
			break
		}
	}

	// Remove from index set
	if err := c.client.SRem(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
		return deletedCount, fmt.Errorf("srem index failed: %w", err)
	}

	return deletedCount, nil
}

// CountFiles returns the number of unique files stored
func (c *Client) CountFiles() (int, error) {
	hashIDs, err := c.GetAllHashIDs()
	if err != nil {
		return 0, err
	}
	return len(hashIDs), nil
}

// LookupPathByCID searches all file metadata for a matching CID
// Returns the file path if found in poster/posterPath or backdrop/backdropPath
func (c *Client) LookupPathByCID(cid string) (string, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return "", fmt.Errorf("not connected")
	}

	if cid == "" {
		return "", nil
	}

	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	// Get all hash IDs first
	hashIDs, err := c.getAllHashIDsInternal(ctx)
	if err != nil {
		return "", err
	}

	// For each hash, check poster and backdrop CIDs using flat keys
	for _, hashID := range hashIDs {
		prefix := c.buildKeyPrefix(hashID)

		// Check poster
		posterCID, err := c.client.Get(ctx, prefix+"poster").Result()
		if err == nil && posterCID == cid {
			posterPath, err := c.client.Get(ctx, prefix+"posterPath").Result()
			if err == nil && posterPath != "" {
				return posterPath, nil
			}
		}

		// Check backdrop
		backdropCID, err := c.client.Get(ctx, prefix+"backdrop").Result()
		if err == nil && backdropCID == cid {
			backdropPath, err := c.client.Get(ctx, prefix+"backdropPath").Result()
			if err == nil && backdropPath != "" {
				return backdropPath, nil
			}
		}
	}

	return "", nil
}

// getAllHashIDsInternal is an internal version that doesn't acquire locks
// (caller must hold the lock)
func (c *Client) getAllHashIDsInternal(ctx context.Context) ([]string, error) {
	indexKey := c.buildIndexKey()
	result, err := c.client.SMembers(ctx, indexKey).Result()
	if err != nil {
		return nil, fmt.Errorf("smembers failed: %w", err)
	}
	return result, nil
}

// MergeMetadataFlat merges new metadata into existing (PATCH semantics)
// New keys are added, existing keys are updated, missing keys are NOT deleted
// Uses pipeline SET for flat keys: file:{hashId}/{property}
func (c *Client) MergeMetadataFlat(hashID string, metadata map[string]string) (int, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return 0, fmt.Errorf("not connected")
	}

	if len(metadata) == 0 {
		return 0, nil
	}

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	prefix := c.buildKeyPrefix(hashID)
	pipe := c.client.Pipeline()

	for field, value := range metadata {
		pipe.Set(ctx, prefix+field, value, 0)
	}

	if _, err := pipe.Exec(ctx); err != nil {
		return 0, fmt.Errorf("pipeline exec failed: %w", err)
	}

	// Add to index set
	if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
		return len(metadata), fmt.Errorf("sadd index failed: %w", err)
	}

	return len(metadata), nil
}

// DeleteProperty deletes a single property
// Uses flat key: DEL file:{hashId}/{property}
func (c *Client) DeleteProperty(hashID, property string) error {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	key := c.buildKeyPrefix(hashID) + property
	return c.client.Del(ctx, key).Err()
}

// AddToSet adds a value to a set-type field (stored as pipe-delimited string in flat key)
// Returns true if value was added, false if it already existed
func (c *Client) AddToSet(hashID, property, value string) (bool, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.client == nil {
		return false, fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	key := c.buildKeyPrefix(hashID) + property

	// Get current value
	current, err := c.client.Get(ctx, key).Result()
	if err != nil && err != redis.Nil {
		return false, err
	}

	// Parse existing values (pipe-delimited)
	var values []string
	if current != "" {
		values = strings.Split(current, "|")
	}

	// Check if value already exists
	for _, v := range values {
		if v == value {
			return false, nil // Already exists
		}
	}

	// Add new value
	values = append(values, value)
	newValue := strings.Join(values, "|")

	// Save back using SET
	if err := c.client.Set(ctx, key, newValue, 0).Err(); err != nil {
		return false, err
	}

	// Add to index set
	if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
		return true, fmt.Errorf("sadd index failed: %w", err)
	}

	return true, nil
}

// RemoveFromSet removes a value from a set-type field (stored in flat key)
// Returns true if value was removed, false if it didn't exist
func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.client == nil {
		return false, fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	key := c.buildKeyPrefix(hashID) + property

	// Get current value
	current, err := c.client.Get(ctx, key).Result()
	if err == redis.Nil {
		return false, nil // Key doesn't exist
	}
	if err != nil {
		return false, err
	}

	// Parse existing values
	values := strings.Split(current, "|")

	// Find and remove value
	found := false
	newValues := make([]string, 0, len(values))
	for _, v := range values {
		if v == value {
			found = true
		} else {
			newValues = append(newValues, v)
		}
	}

	if !found {
		return false, nil
	}

	// Save back (or delete key if empty)
	if len(newValues) == 0 {
		return true, c.client.Del(ctx, key).Err()
	}

	newValue := strings.Join(newValues, "|")
	return true, c.client.Set(ctx, key, newValue, 0).Err()
}

// FileTuple holds the (path, size, mtime) record for one file. Used by the
// watcher hydrator at startup to avoid re-mid-hashing files whose tuple
// matches what's already in Redis.
type FileTuple struct {
	HashID    string
	FilePath  string
	Size      int64
	MtimeNano int64
}

// GetTuplesForAllFiles returns one FileTuple per indexed CID, fetched in
// batched MGETs (no SCAN per CID). Skips files missing any of the three
// fields. Sub-second on local Redis for ~10K files.
func (c *Client) GetTuplesForAllFiles() ([]FileTuple, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return nil, fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	hashIDs, err := c.client.SMembers(ctx, c.buildIndexKey()).Result()
	if err != nil {
		return nil, fmt.Errorf("smembers index: %w", err)
	}
	if len(hashIDs) == 0 {
		return nil, nil
	}

	// Build the key list: 3 keys per CID, in a fixed order so we can decode
	// the MGET response by position. fields[i] for hashIDs[j] sits at
	// keys[j*3+i].
	const fieldsPerFile = 3
	fields := [fieldsPerFile]string{"filePath", "sizeByte", "mtimeNano"}
	keys := make([]string, 0, len(hashIDs)*fieldsPerFile)
	for _, h := range hashIDs {
		prefix := c.buildKeyPrefix(h)
		for _, f := range fields {
			keys = append(keys, prefix+f)
		}
	}

	// Chunked MGET — Redis caps args per command; 1000 is a safe batch.
	const batch = 1000
	values := make([]interface{}, 0, len(keys))
	for start := 0; start < len(keys); start += batch {
		end := start + batch
		if end > len(keys) {
			end = len(keys)
		}
		chunk, err := c.client.MGet(ctx, keys[start:end]...).Result()
		if err != nil {
			return nil, fmt.Errorf("mget chunk [%d:%d]: %w", start, end, err)
		}
		values = append(values, chunk...)
	}

	out := make([]FileTuple, 0, len(hashIDs))
	for i, h := range hashIDs {
		base := i * fieldsPerFile
		filePath, _ := values[base].(string)
		sizeStr, _ := values[base+1].(string)
		mtimeStr, _ := values[base+2].(string)
		if filePath == "" || sizeStr == "" || mtimeStr == "" {
			continue
		}
		size, err := strconv.ParseInt(sizeStr, 10, 64)
		if err != nil {
			continue
		}
		mtime, err := strconv.ParseInt(mtimeStr, 10, 64)
		if err != nil {
			continue
		}
		out = append(out, FileTuple{
			HashID:    h,
			FilePath:  filePath,
			Size:      size,
			MtimeNano: mtime,
		})
	}
	return out, nil
}

// GetMemoryInfo returns Redis memory usage information
func (c *Client) GetMemoryInfo() (string, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return "", fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	info, err := c.client.Info(ctx, "memory").Result()
	if err != nil {
		return "", fmt.Errorf("info memory failed: %w", err)
	}

	// Extract used_memory_human from info string
	for _, line := range strings.Split(info, "\r\n") {
		if strings.HasPrefix(line, "used_memory_human:") {
			return strings.TrimPrefix(line, "used_memory_human:"), nil
		}
	}

	return "N/A", nil
}

// ClearAllMetadata deletes all file metadata and index
// Returns the number of files deleted
func (c *Client) ClearAllMetadata() (int64, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.client == nil {
		return 0, fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
	defer cancel()

	// Get all hash IDs from index
	indexKey := c.buildIndexKey()
	hashIDs, err := c.client.SMembers(ctx, indexKey).Result()
	if err != nil {
		return 0, fmt.Errorf("smembers failed: %w", err)
	}

	var deletedCount int64

	// Delete each file's flat keys
	for _, hashID := range hashIDs {
		prefix := c.buildKeyPrefix(hashID)

		// Scan and delete all keys with this prefix
		var cursor uint64
		for {
			keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
			if err != nil {
				log.Printf("[Storage] Warning: failed to scan %s: %v", prefix, err)
				break
			}

			if len(keys) > 0 {
				if err := c.client.Del(ctx, keys...).Err(); err != nil {
					log.Printf("[Storage] Warning: failed to delete keys for %s: %v", hashID, err)
				}
			}

			cursor = nextCursor
			if cursor == 0 {
				break
			}
		}
		deletedCount++
	}

	// Delete the index set
	if err := c.client.Del(ctx, indexKey).Err(); err != nil {
		log.Printf("[Storage] Warning: failed to delete index: %v", err)
	}

	log.Printf("[Storage] Cleared %d file metadata entries", deletedCount)
	return deletedCount, nil
}

// ScanKeys lists the immediate children at the given prefix, S3-style.
//
// Walks the keyspace under prefix using Redis SCAN. When delimiter is set,
// keys with a delimiter occurrence after the prefix collapse into a branch
// path (the prefix up to and including that first delimiter); the rest are
// returned as leaves. When delimiter is empty all matching keys are leaves.
//
// Continues SCAN-ing until either the keyspace is exhausted (nextCursor == "0")
// or leaves+branches reaches maxResults — whichever comes first. The caller
// passes nextCursor back to continue. Results are deduped and sorted.
func (c *Client) ScanKeys(prefix, delimiter, cursorStr string, maxResults int) ([]string, []string, string, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return nil, nil, "", fmt.Errorf("not connected")
	}

	if maxResults <= 0 {
		maxResults = 500
	}

	var cursor uint64
	if cursorStr != "" {
		if v, err := strconv.ParseUint(cursorStr, 10, 64); err == nil {
			cursor = v
		}
	}

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	fullPrefix := c.buildKey(prefix)
	pattern := escapeGlob(fullPrefix) + "*"

	leafSet := make(map[string]struct{})
	branchSet := make(map[string]struct{})

	for {
		// SCAN COUNT is a hint; ask for a larger batch when grouping by
		// delimiter since many keys collapse into one branch.
		scanCount := int64(maxResults * 4)
		if scanCount > 5000 {
			scanCount = 5000
		}
		keys, next, err := c.client.Scan(ctx, cursor, pattern, scanCount).Result()
		if err != nil {
			return nil, nil, "", fmt.Errorf("scan failed: %w", err)
		}
		for _, k := range keys {
			userKey := strings.TrimPrefix(k, c.prefix)
			suffix := strings.TrimPrefix(userKey, prefix)
			if delimiter != "" {
				if idx := strings.Index(suffix, delimiter); idx >= 0 {
					branchSet[prefix+suffix[:idx+len(delimiter)]] = struct{}{}
					continue
				}
			}
			leafSet[userKey] = struct{}{}
		}
		cursor = next
		if cursor == 0 {
			break
		}
		if len(leafSet)+len(branchSet) >= maxResults {
			break
		}
	}

	leaves := make([]string, 0, len(leafSet))
	for k := range leafSet {
		leaves = append(leaves, k)
	}
	branches := make([]string, 0, len(branchSet))
	for k := range branchSet {
		branches = append(branches, k)
	}
	sort.Strings(leaves)
	sort.Strings(branches)
	return leaves, branches, strconv.FormatUint(cursor, 10), nil
}

// FindMatch is a single value-search hit.
type FindMatch struct {
	Key       string `json:"key"`
	Value     string `json:"value"`
	EntryPath string `json:"entryPath"`
	Field     string `json:"field"`
}

// FindByValue scans every key whose path ends with /<field> for each field
// in fields, MGETs them, and returns the keys whose value contains the
// substring (case-insensitive). Returns truncated when limit was hit before
// the scan exhausted.
//
// EntryPath is the parent prefix up to (and including) the slash before
// field — i.e. the file branch path. Useful for revealing the matched entry
// in the tree.
func (c *Client) FindByValue(contains string, fields []string, limit int) ([]FindMatch, bool, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return nil, false, fmt.Errorf("not connected")
	}
	if contains == "" || len(fields) == 0 {
		return nil, false, nil
	}
	if limit <= 0 {
		limit = 200
	}

	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	needle := strings.ToLower(contains)
	var matches []FindMatch
	truncated := false

	for _, field := range fields {
		field = strings.TrimSpace(field)
		if field == "" {
			continue
		}
		if len(matches) >= limit {
			truncated = true
			break
		}
		// Walk all keys ending with /field via SCAN MATCH "*/<field>".
		pattern := "*/" + escapeGlob(field)
		if c.prefix != "" {
			pattern = escapeGlob(c.prefix) + pattern
		}
		var keysToFetch []string
		var cursor uint64
		for {
			keys, next, err := c.client.Scan(ctx, cursor, pattern, 1000).Result()
			if err != nil {
				return nil, false, fmt.Errorf("scan failed: %w", err)
			}
			keysToFetch = append(keysToFetch, keys...)
			cursor = next
			if cursor == 0 {
				break
			}
			if len(keysToFetch) > 20000 {
				break
			}
		}
		// MGET in chunks (Redis limits the number of args per command).
		for chunkStart := 0; chunkStart < len(keysToFetch); chunkStart += 500 {
			chunkEnd := chunkStart + 500
			if chunkEnd > len(keysToFetch) {
				chunkEnd = len(keysToFetch)
			}
			chunk := keysToFetch[chunkStart:chunkEnd]
			values, err := c.client.MGet(ctx, chunk...).Result()
			if err != nil {
				return nil, false, fmt.Errorf("mget failed: %w", err)
			}
			for i, v := range values {
				if v == nil {
					continue
				}
				sv, _ := v.(string)
				if !strings.Contains(strings.ToLower(sv), needle) {
					continue
				}
				userKey := strings.TrimPrefix(chunk[i], c.prefix)
				entryPath := strings.TrimSuffix(userKey, field)
				matches = append(matches, FindMatch{
					Key:       userKey,
					Value:     sv,
					EntryPath: entryPath,
					Field:     field,
				})
				if len(matches) >= limit {
					truncated = true
					break
				}
			}
			if truncated {
				break
			}
		}
	}

	sort.Slice(matches, func(i, j int) bool { return matches[i].Key < matches[j].Key })
	return matches, truncated, nil
}

// SearchKeys returns up to limit keys whose full path contains the substring.
// truncated is true when the search stopped at limit before exhausting Redis.
func (c *Client) SearchKeys(contains string, limit int) ([]string, bool, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return nil, false, fmt.Errorf("not connected")
	}

	if contains == "" {
		return nil, false, nil
	}
	if limit <= 0 {
		limit = 500
	}

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	pattern := escapeGlob(c.prefix) + "*" + escapeGlob(contains) + "*"

	results := make([]string, 0, limit)
	truncated := false
	var cursor uint64
	for {
		keys, next, err := c.client.Scan(ctx, cursor, pattern, 1000).Result()
		if err != nil {
			return nil, false, fmt.Errorf("scan failed: %w", err)
		}
		for _, k := range keys {
			if len(results) >= limit {
				truncated = true
				break
			}
			results = append(results, strings.TrimPrefix(k, c.prefix))
		}
		cursor = next
		if truncated || cursor == 0 {
			break
		}
	}
	sort.Strings(results)
	return results, truncated, nil
}

// GetKeyInfo returns the Redis type ("string", "set", "hash", ...) and, for
// string-typed keys, the value. exists is false when the key is absent.
func (c *Client) GetKeyInfo(key string) (kind string, value string, exists bool, err error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return "", "", false, fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	full := c.buildKey(key)
	t, err := c.client.Type(ctx, full).Result()
	if err != nil {
		return "", "", false, fmt.Errorf("type failed: %w", err)
	}
	if t == "none" {
		return "", "", false, nil
	}
	if t != "string" {
		return t, "", true, nil
	}
	v, err := c.client.Get(ctx, full).Result()
	if err == redis.Nil {
		return "", "", false, nil
	}
	if err != nil {
		return "", "", false, fmt.Errorf("get failed: %w", err)
	}
	return "string", v, true, nil
}

// SetRawValue writes a string value at the exact key. Refuses to overwrite a
// key whose existing Redis type is not "string".
func (c *Client) SetRawValue(key, value string) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.client == nil {
		return fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	full := c.buildKey(key)
	t, err := c.client.Type(ctx, full).Result()
	if err != nil {
		return fmt.Errorf("type failed: %w", err)
	}
	if t != "none" && t != "string" {
		return fmt.Errorf("key has incompatible type %q (only string is editable)", t)
	}
	return c.client.Set(ctx, full, value, 0).Err()
}

// DeleteRawKey deletes the exact key. Returns true if the key existed.
func (c *Client) DeleteRawKey(key string) (bool, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.client == nil {
		return false, fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	n, err := c.client.Del(ctx, c.buildKey(key)).Result()
	if err != nil {
		return false, err
	}
	return n > 0, nil
}

// IndexAdd adds a hash ID to the file index set. Idempotent.
func (c *Client) IndexAdd(hashID string) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.client == nil {
		return fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	return c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err()
}

// IndexRemoveIfEmpty removes hashID from the index set when no
// file:<hashID>/* keys remain. Returns true if the index entry was removed.
func (c *Client) IndexRemoveIfEmpty(hashID string) (bool, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.client == nil {
		return false, fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	prefix := c.buildKeyPrefix(hashID)
	keys, _, err := c.client.Scan(ctx, 0, prefix+"*", 1).Result()
	if err != nil {
		return false, fmt.Errorf("scan failed: %w", err)
	}
	if len(keys) > 0 {
		return false, nil
	}
	removed, err := c.client.SRem(ctx, c.buildIndexKey(), hashID).Result()
	if err != nil {
		return false, fmt.Errorf("srem failed: %w", err)
	}
	return removed > 0, nil
}

// escapeGlob escapes Redis SCAN MATCH glob metacharacters so a literal string
// can be embedded inside a pattern. Redis uses *, ?, [, \ as metacharacters.
func escapeGlob(s string) string {
	var b strings.Builder
	b.Grow(len(s))
	for _, r := range s {
		switch r {
		case '*', '?', '[', ']', '\\':
			b.WriteByte('\\')
		}
		b.WriteRune(r)
	}
	return b.String()
}

// GetRedisClient returns the underlying redis client for advanced operations
func (c *Client) GetRedisClient() *redis.Client {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.client
}

// GetPrefix returns the key prefix used by this client
func (c *Client) GetPrefix() string {
	return c.prefix
}

// XAdd publishes an event to a Redis Stream
// If maxLen > 0, uses XADD with MAXLEN ~ for approximate trimming
// If maxLen <= 0, no trimming is applied (stream grows unbounded)
func (c *Client) XAdd(stream string, maxLen int64, fields map[string]interface{}) (string, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return "", fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// Build XADD arguments
	args := &redis.XAddArgs{
		Stream: stream,
		Values: fields,
	}

	// Only set MaxLen if positive (0 means no limit)
	if maxLen > 0 {
		args.MaxLen = maxLen
		args.Approx = true
	}

	id, err := c.client.XAdd(ctx, args).Result()
	if err != nil {
		return "", fmt.Errorf("xadd failed: %w", err)
	}

	return id, nil
}

// ClearStream deletes a Redis stream entirely
func (c *Client) ClearStream(stream string) error {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	if err := c.client.Del(ctx, stream).Err(); err != nil {
		return fmt.Errorf("del stream failed: %w", err)
	}

	return nil
}

// StreamEntry represents one entry from a Redis stream.
type StreamEntry struct {
	ID     string
	Values map[string]interface{}
}

// XRevRange reads stream entries from newest to oldest.
// start defaults to "+" (newest) and stop defaults to "-" (oldest) when empty.
// If count > 0, at most that many entries are returned.
func (c *Client) XRevRange(stream, start, stop string, count int64) ([]StreamEntry, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return nil, fmt.Errorf("not connected")
	}

	if start == "" {
		start = "+"
	}
	if stop == "" {
		stop = "-"
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	var msgs []redis.XMessage
	var err error
	if count > 0 {
		msgs, err = c.client.XRevRangeN(ctx, stream, start, stop, count).Result()
	} else {
		msgs, err = c.client.XRevRange(ctx, stream, start, stop).Result()
	}
	if err != nil {
		return nil, fmt.Errorf("xrevrange failed: %w", err)
	}

	entries := make([]StreamEntry, 0, len(msgs))
	for _, m := range msgs {
		entries = append(entries, StreamEntry{ID: m.ID, Values: m.Values})
	}
	return entries, nil
}

// GetStreamLength returns the length of a Redis stream
func (c *Client) GetStreamLength(stream string) (int64, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return 0, fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	length, err := c.client.XLen(ctx, stream).Result()
	if err != nil {
		return 0, fmt.Errorf("xlen failed: %w", err)
	}

	return length, nil
}