fix KV API to unified format

This commit is contained in:
Pierre
2026-01-24 22:19:42 +01:00
parent 8276fbb874
commit 768942d28b
2 changed files with 135 additions and 163 deletions
+33 -4
View File
@@ -507,8 +507,14 @@ func (s *Server) handleGetProperty(w http.ResponseWriter, r *http.Request) {
w.Write([]byte(value)) w.Write([]byte(value))
} }
// PropertyBody is the request body for PUT /meta/{hash}/{key}
type PropertyBody struct {
Value string `json:"value"`
}
// handlePutProperty handles PUT /meta/{hash}/{key} // handlePutProperty handles PUT /meta/{hash}/{key}
// Sets a single property value // Sets a single property value
// Accepts JSON body with {"value": "..."} or plain text
func (s *Server) handlePutProperty(w http.ResponseWriter, r *http.Request) { func (s *Server) handlePutProperty(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r) vars := mux.Vars(r)
hashID := vars["hash"] hashID := vars["hash"]
@@ -529,14 +535,25 @@ func (s *Server) handlePutProperty(w http.ResponseWriter, r *http.Request) {
return return
} }
// Read body as plain text // Read body
buf := make([]byte, 1024*1024) // 1MB max buf := make([]byte, 1024*1024) // 1MB max
n, err := r.Body.Read(buf) n, err := r.Body.Read(buf)
if err != nil && err.Error() != "EOF" { if err != nil && err.Error() != "EOF" {
writeError(w, http.StatusBadRequest, "failed to read body") writeError(w, http.StatusBadRequest, "failed to read body")
return return
} }
value := string(buf[:n]) body := buf[:n]
// Try to parse as JSON {"value": "..."} first
var propBody PropertyBody
var value string
if err := json.Unmarshal(body, &propBody); err == nil && propBody.Value != "" {
// JSON body with value field
value = propBody.Value
} else {
// Plain text body
value = string(body)
}
if err := s.storage.SetProperty(hashID, key, value); err != nil { if err := s.storage.SetProperty(hashID, key, value); err != nil {
writeError(w, http.StatusInternalServerError, err.Error()) writeError(w, http.StatusInternalServerError, err.Error())
@@ -586,6 +603,7 @@ func (s *Server) handleDeleteProperty(w http.ResponseWriter, r *http.Request) {
// handleAddToSet handles POST /meta/{hash}/_add/{key} // handleAddToSet handles POST /meta/{hash}/_add/{key}
// Adds a value to a set-type field // Adds a value to a set-type field
// Accepts JSON body with {"value": "..."} or plain text
func (s *Server) handleAddToSet(w http.ResponseWriter, r *http.Request) { func (s *Server) handleAddToSet(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r) vars := mux.Vars(r)
hashID := vars["hash"] hashID := vars["hash"]
@@ -606,14 +624,25 @@ func (s *Server) handleAddToSet(w http.ResponseWriter, r *http.Request) {
return return
} }
// Read body as plain text // Read body
buf := make([]byte, 1024*1024) // 1MB max buf := make([]byte, 1024*1024) // 1MB max
n, err := r.Body.Read(buf) n, err := r.Body.Read(buf)
if err != nil && err.Error() != "EOF" { if err != nil && err.Error() != "EOF" {
writeError(w, http.StatusBadRequest, "failed to read body") writeError(w, http.StatusBadRequest, "failed to read body")
return return
} }
value := string(buf[:n]) body := buf[:n]
// Try to parse as JSON {"value": "..."} first
var propBody PropertyBody
var value string
if err := json.Unmarshal(body, &propBody); err == nil && propBody.Value != "" {
// JSON body with value field
value = propBody.Value
} else {
// Plain text body
value = string(body)
}
if value == "" { if value == "" {
writeError(w, http.StatusBadRequest, "value is required") writeError(w, http.StatusBadRequest, "value is required")
+102 -159
View File
@@ -97,13 +97,18 @@ func (c *Client) buildKey(key string) string {
return key return key
} }
// buildFilePrefix constructs the prefix for a file's metadata keys // buildHashKey constructs the Redis Hash key for a file
func (c *Client) buildFilePrefix(hashID string) string { func (c *Client) buildHashKey(hashID string) string {
return fmt.Sprintf("/file/%s", hashID) 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 // GetMetadataFlat retrieves all metadata for a file as a flat map
// Key format: /file/{hashId}/{property} -> value // Uses Redis Hash: HGETALL file:{hashId}
func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) { func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) {
c.mu.RLock() c.mu.RLock()
defer c.mu.RUnlock() defer c.mu.RUnlock()
@@ -115,37 +120,10 @@ func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel() defer cancel()
prefix := c.buildKey(c.buildFilePrefix(hashID)) hashKey := c.buildHashKey(hashID)
result := make(map[string]string) result, err := c.client.HGetAll(ctx, hashKey).Result()
if err != nil {
// Use SCAN to find all keys with this prefix return nil, fmt.Errorf("hgetall failed: %w", err)
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 {
// Get values for all keys
values, err := c.client.MGet(ctx, keys...).Result()
if err != nil {
return nil, fmt.Errorf("mget failed: %w", err)
}
for i, key := range keys {
if values[i] != nil {
// Strip prefix to get property path
propPath := strings.TrimPrefix(key, prefix+"/")
result[propPath] = values[i].(string)
}
}
}
cursor = nextCursor
if cursor == 0 {
break
}
} }
if len(result) == 0 { if len(result) == 0 {
@@ -155,7 +133,8 @@ func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) {
return result, nil return result, nil
} }
// SetMetadataFlat stores metadata for a file as flat key-value pairs // 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 { func (c *Client) SetMetadataFlat(hashID string, metadata map[string]string) error {
c.mu.RLock() c.mu.RLock()
defer c.mu.RUnlock() defer c.mu.RUnlock()
@@ -171,20 +150,23 @@ func (c *Client) SetMetadataFlat(hashID string, metadata map[string]string) erro
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel() defer cancel()
prefix := c.buildKey(c.buildFilePrefix(hashID)) hashKey := c.buildHashKey(hashID)
// Use pipeline for batch operations // Use HMSET to set all fields at once
pipe := c.client.Pipeline() if err := c.client.HMSet(ctx, hashKey, metadata).Err(); err != nil {
for prop, value := range metadata { return fmt.Errorf("hmset failed: %w", err)
key := prefix + "/" + prop
pipe.Set(ctx, key, value, 0)
} }
_, err := pipe.Exec(ctx) // Add to index set
return err 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 // GetAllHashIDs returns all unique file hash IDs stored
// Uses Redis Set: SMEMBERS file:__index__
func (c *Client) GetAllHashIDs() ([]string, error) { func (c *Client) GetAllHashIDs() ([]string, error) {
c.mu.RLock() c.mu.RLock()
defer c.mu.RUnlock() defer c.mu.RUnlock()
@@ -196,42 +178,17 @@ func (c *Client) GetAllHashIDs() ([]string, error) {
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel() defer cancel()
prefix := c.buildKey("/file/") indexKey := c.buildIndexKey()
hashSet := make(map[string]bool) result, err := c.client.SMembers(ctx, indexKey).Result()
if err != nil {
// Scan for all file keys return nil, fmt.Errorf("smembers failed: %w", err)
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)
}
for _, key := range keys {
// Extract hashID from key /file/{hashID}/property
stripped := strings.TrimPrefix(key, prefix)
parts := strings.SplitN(stripped, "/", 2)
if len(parts) > 0 && parts[0] != "" {
hashSet[parts[0]] = true
}
}
cursor = nextCursor
if cursor == 0 {
break
}
}
// Convert set to slice
result := make([]string, 0, len(hashSet))
for hashID := range hashSet {
result = append(result, hashID)
} }
return result, nil return result, nil
} }
// GetProperty retrieves a single property value // GetProperty retrieves a single property value
// Uses Redis Hash: HGET file:{hashId} {property}
func (c *Client) GetProperty(hashID, property string) (string, error) { func (c *Client) GetProperty(hashID, property string) (string, error) {
c.mu.RLock() c.mu.RLock()
defer c.mu.RUnlock() defer c.mu.RUnlock()
@@ -243,8 +200,8 @@ func (c *Client) GetProperty(hashID, property string) (string, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel() defer cancel()
key := c.buildKey(c.buildFilePrefix(hashID) + "/" + property) hashKey := c.buildHashKey(hashID)
result, err := c.client.Get(ctx, key).Result() result, err := c.client.HGet(ctx, hashKey, property).Result()
if err == redis.Nil { if err == redis.Nil {
return "", nil return "", nil
} }
@@ -252,6 +209,7 @@ func (c *Client) GetProperty(hashID, property string) (string, error) {
} }
// SetProperty sets a single property value // SetProperty sets a single property value
// Uses Redis Hash: HSET file:{hashId} {property} {value}
func (c *Client) SetProperty(hashID, property, value string) error { func (c *Client) SetProperty(hashID, property, value string) error {
c.mu.RLock() c.mu.RLock()
defer c.mu.RUnlock() defer c.mu.RUnlock()
@@ -263,11 +221,23 @@ func (c *Client) SetProperty(hashID, property, value string) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel() defer cancel()
key := c.buildKey(c.buildFilePrefix(hashID) + "/" + property) hashKey := c.buildHashKey(hashID)
return c.client.Set(ctx, key, value, 0).Err()
// 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 // DeleteMetadata deletes all metadata for a file
// Uses Redis Hash: DEL file:{hashId}
func (c *Client) DeleteMetadata(hashID string) (int64, error) { func (c *Client) DeleteMetadata(hashID string) (int64, error) {
c.mu.RLock() c.mu.RLock()
defer c.mu.RUnlock() defer c.mu.RUnlock()
@@ -279,32 +249,25 @@ func (c *Client) DeleteMetadata(hashID string) (int64, error) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel() defer cancel()
prefix := c.buildKey(c.buildFilePrefix(hashID)) hashKey := c.buildHashKey(hashID)
var deleted int64
// Scan and delete all keys with this prefix // Get field count before deletion
var cursor uint64 fieldCount, err := c.client.HLen(ctx, hashKey).Result()
for { if err != nil {
keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"/*", 1000).Result() return 0, fmt.Errorf("hlen failed: %w", err)
if err != nil {
return deleted, fmt.Errorf("scan failed: %w", err)
}
if len(keys) > 0 {
count, err := c.client.Del(ctx, keys...).Result()
if err != nil {
return deleted, fmt.Errorf("del failed: %w", err)
}
deleted += count
}
cursor = nextCursor
if cursor == 0 {
break
}
} }
return deleted, nil // 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 // CountFiles returns the number of unique files stored
@@ -339,23 +302,23 @@ func (c *Client) LookupPathByCID(cid string) (string, error) {
return "", err return "", err
} }
// For each hash, check poster and backdrop CIDs // For each hash, check poster and backdrop CIDs using Hash operations
for _, hashID := range hashIDs { for _, hashID := range hashIDs {
prefix := c.buildKey(c.buildFilePrefix(hashID)) hashKey := c.buildHashKey(hashID)
// Check poster // Check poster
posterCID, err := c.client.Get(ctx, prefix+"/poster").Result() posterCID, err := c.client.HGet(ctx, hashKey, "poster").Result()
if err == nil && posterCID == cid { if err == nil && posterCID == cid {
posterPath, err := c.client.Get(ctx, prefix+"/posterPath").Result() posterPath, err := c.client.HGet(ctx, hashKey, "posterPath").Result()
if err == nil && posterPath != "" { if err == nil && posterPath != "" {
return posterPath, nil return posterPath, nil
} }
} }
// Check backdrop // Check backdrop
backdropCID, err := c.client.Get(ctx, prefix+"/backdrop").Result() backdropCID, err := c.client.HGet(ctx, hashKey, "backdrop").Result()
if err == nil && backdropCID == cid { if err == nil && backdropCID == cid {
backdropPath, err := c.client.Get(ctx, prefix+"/backdropPath").Result() backdropPath, err := c.client.HGet(ctx, hashKey, "backdropPath").Result()
if err == nil && backdropPath != "" { if err == nil && backdropPath != "" {
return backdropPath, nil return backdropPath, nil
} }
@@ -368,40 +331,17 @@ func (c *Client) LookupPathByCID(cid string) (string, error) {
// getAllHashIDsInternal is an internal version that doesn't acquire locks // getAllHashIDsInternal is an internal version that doesn't acquire locks
// (caller must hold the lock) // (caller must hold the lock)
func (c *Client) getAllHashIDsInternal(ctx context.Context) ([]string, error) { func (c *Client) getAllHashIDsInternal(ctx context.Context) ([]string, error) {
prefix := c.buildKey("/file/") indexKey := c.buildIndexKey()
hashSet := make(map[string]bool) result, err := c.client.SMembers(ctx, indexKey).Result()
if err != nil {
var cursor uint64 return nil, fmt.Errorf("smembers failed: %w", err)
for {
keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
if err != nil {
return nil, fmt.Errorf("scan failed: %w", err)
}
for _, key := range keys {
stripped := strings.TrimPrefix(key, prefix)
parts := strings.SplitN(stripped, "/", 2)
if len(parts) > 0 && parts[0] != "" {
hashSet[parts[0]] = true
}
}
cursor = nextCursor
if cursor == 0 {
break
}
} }
result := make([]string, 0, len(hashSet))
for hashID := range hashSet {
result = append(result, hashID)
}
return result, nil return result, nil
} }
// MergeMetadataFlat merges new metadata into existing (PATCH semantics) // MergeMetadataFlat merges new metadata into existing (PATCH semantics)
// New keys are added, existing keys are updated, missing keys are NOT deleted // 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) { func (c *Client) MergeMetadataFlat(hashID string, metadata map[string]string) (int, error) {
c.mu.RLock() c.mu.RLock()
defer c.mu.RUnlock() defer c.mu.RUnlock()
@@ -417,24 +357,23 @@ func (c *Client) MergeMetadataFlat(hashID string, metadata map[string]string) (i
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel() defer cancel()
prefix := c.buildKey(c.buildFilePrefix(hashID)) hashKey := c.buildHashKey(hashID)
// Use pipeline for batch operations // Use HMSET to merge fields
pipe := c.client.Pipeline() if err := c.client.HMSet(ctx, hashKey, metadata).Err(); err != nil {
for prop, value := range metadata { return 0, fmt.Errorf("hmset failed: %w", err)
key := prefix + "/" + prop
pipe.Set(ctx, key, value, 0)
} }
_, err := pipe.Exec(ctx) // Add to index set
if err != nil { if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
return 0, err return len(metadata), fmt.Errorf("sadd index failed: %w", err)
} }
return len(metadata), nil return len(metadata), nil
} }
// DeleteProperty deletes a single property // DeleteProperty deletes a single property
// Uses Redis Hash: HDEL file:{hashId} {property}
func (c *Client) DeleteProperty(hashID, property string) error { func (c *Client) DeleteProperty(hashID, property string) error {
c.mu.RLock() c.mu.RLock()
defer c.mu.RUnlock() defer c.mu.RUnlock()
@@ -446,11 +385,11 @@ func (c *Client) DeleteProperty(hashID, property string) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel() defer cancel()
key := c.buildKey(c.buildFilePrefix(hashID) + "/" + property) hashKey := c.buildHashKey(hashID)
return c.client.Del(ctx, key).Err() return c.client.HDel(ctx, hashKey, property).Err()
} }
// AddToSet adds a value to a set-type field (stored as pipe-delimited string) // 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 // Returns true if value was added, false if it already existed
func (c *Client) AddToSet(hashID, property, value string) (bool, error) { func (c *Client) AddToSet(hashID, property, value string) (bool, error) {
c.mu.Lock() c.mu.Lock()
@@ -463,10 +402,10 @@ func (c *Client) AddToSet(hashID, property, value string) (bool, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel() defer cancel()
key := c.buildKey(c.buildFilePrefix(hashID) + "/" + property) hashKey := c.buildHashKey(hashID)
// Get current value // Get current value
current, err := c.client.Get(ctx, key).Result() current, err := c.client.HGet(ctx, hashKey, property).Result()
if err != nil && err != redis.Nil { if err != nil && err != redis.Nil {
return false, err return false, err
} }
@@ -488,16 +427,20 @@ func (c *Client) AddToSet(hashID, property, value string) (bool, error) {
values = append(values, value) values = append(values, value)
newValue := strings.Join(values, "|") newValue := strings.Join(values, "|")
// Save back // Save back using HSET
err = c.client.Set(ctx, key, newValue, 0).Err() if err := c.client.HSet(ctx, hashKey, property, newValue).Err(); err != nil {
if err != nil {
return false, err 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 return true, nil
} }
// RemoveFromSet removes a value from a set-type field // RemoveFromSet removes a value from a set-type field (stored in Hash)
// Returns true if value was removed, false if it didn't exist // Returns true if value was removed, false if it didn't exist
func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) { func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) {
c.mu.Lock() c.mu.Lock()
@@ -510,12 +453,12 @@ func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel() defer cancel()
key := c.buildKey(c.buildFilePrefix(hashID) + "/" + property) hashKey := c.buildHashKey(hashID)
// Get current value // Get current value
current, err := c.client.Get(ctx, key).Result() current, err := c.client.HGet(ctx, hashKey, property).Result()
if err == redis.Nil { if err == redis.Nil {
return false, nil // Key doesn't exist return false, nil // Field doesn't exist
} }
if err != nil { if err != nil {
return false, err return false, err
@@ -539,11 +482,11 @@ func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) {
return false, nil return false, nil
} }
// Save back (or delete if empty) // Save back (or delete field if empty)
if len(newValues) == 0 { if len(newValues) == 0 {
return true, c.client.Del(ctx, key).Err() return true, c.client.HDel(ctx, hashKey, property).Err()
} }
newValue := strings.Join(newValues, "|") newValue := strings.Join(newValues, "|")
return true, c.client.Set(ctx, key, newValue, 0).Err() return true, c.client.HSet(ctx, hashKey, property, newValue).Err()
} }