Files
meta-core/internal/storage/client.go
T

573 lines
14 KiB
Go

package storage
import (
"context"
"fmt"
"log"
"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
}
// buildHashKey constructs the Redis Hash key for a file
func (c *Client) buildHashKey(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 Redis Hash: HGETALL file:{hashId}
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()
hashKey := c.buildHashKey(hashID)
result, err := c.client.HGetAll(ctx, hashKey).Result()
if err != nil {
return nil, fmt.Errorf("hgetall failed: %w", err)
}
if len(result) == 0 {
return nil, nil
}
return result, nil
}
// SetMetadataFlat stores metadata for a file using Redis Hash
// Uses Redis Hash: HMSET file:{hashId} prop1 val1 prop2 val2...
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()
hashKey := c.buildHashKey(hashID)
// Use HMSET to set all fields at once
if err := c.client.HMSet(ctx, hashKey, metadata).Err(); err != nil {
return fmt.Errorf("hmset 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 Redis Hash: HGET 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()
hashKey := c.buildHashKey(hashID)
result, err := c.client.HGet(ctx, hashKey, property).Result()
if err == redis.Nil {
return "", nil
}
return result, err
}
// SetProperty sets a single property value
// Uses Redis Hash: HSET 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()
hashKey := c.buildHashKey(hashID)
// Use HSET to set the field
if err := c.client.HSet(ctx, hashKey, property, value).Err(); err != nil {
return fmt.Errorf("hset 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 Redis Hash: DEL 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()
hashKey := c.buildHashKey(hashID)
// Get field count before deletion
fieldCount, err := c.client.HLen(ctx, hashKey).Result()
if err != nil {
return 0, fmt.Errorf("hlen failed: %w", err)
}
// Delete the hash
if err := c.client.Del(ctx, hashKey).Err(); err != nil {
return 0, fmt.Errorf("del failed: %w", err)
}
// Remove from index set
if err := c.client.SRem(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
return fieldCount, fmt.Errorf("srem index failed: %w", err)
}
return fieldCount, 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 Hash operations
for _, hashID := range hashIDs {
hashKey := c.buildHashKey(hashID)
// Check poster
posterCID, err := c.client.HGet(ctx, hashKey, "poster").Result()
if err == nil && posterCID == cid {
posterPath, err := c.client.HGet(ctx, hashKey, "posterPath").Result()
if err == nil && posterPath != "" {
return posterPath, nil
}
}
// Check backdrop
backdropCID, err := c.client.HGet(ctx, hashKey, "backdrop").Result()
if err == nil && backdropCID == cid {
backdropPath, err := c.client.HGet(ctx, hashKey, "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 Redis Hash: HMSET file:{hashId}
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()
hashKey := c.buildHashKey(hashID)
// Use HMSET to merge fields
if err := c.client.HMSet(ctx, hashKey, metadata).Err(); err != nil {
return 0, fmt.Errorf("hmset 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 Redis Hash: HDEL 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()
hashKey := c.buildHashKey(hashID)
return c.client.HDel(ctx, hashKey, property).Err()
}
// AddToSet adds a value to a set-type field (stored as pipe-delimited string in Hash field)
// 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()
hashKey := c.buildHashKey(hashID)
// Get current value
current, err := c.client.HGet(ctx, hashKey, property).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 HSET
if err := c.client.HSet(ctx, hashKey, property, newValue).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 Hash)
// 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()
hashKey := c.buildHashKey(hashID)
// Get current value
current, err := c.client.HGet(ctx, hashKey, property).Result()
if err == redis.Nil {
return false, nil // Field 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 field if empty)
if len(newValues) == 0 {
return true, c.client.HDel(ctx, hashKey, property).Err()
}
newValue := strings.Join(newValues, "|")
return true, c.client.HSet(ctx, hashKey, property, newValue).Err()
}
// 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 hash
for _, hashID := range hashIDs {
hashKey := c.buildHashKey(hashID)
if err := c.client.Del(ctx, hashKey).Err(); err != nil {
log.Printf("[Storage] Warning: failed to delete %s: %v", hashKey, err)
continue
}
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
}
// 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
}