feat: cache revalidation
This commit is contained in:
+40
-132
@@ -2,21 +2,20 @@ package filesystem
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/valkey-io/valkey-go"
|
||||
)
|
||||
|
||||
const keyStat = "webdav:stat:"
|
||||
|
||||
// CachedClient wraps a FileSystem implementation with Valkey caching.
|
||||
type CachedClient struct {
|
||||
impl FS
|
||||
client valkey.Client
|
||||
refreshTime time.Duration
|
||||
ttl time.Duration
|
||||
hits atomic.Int64
|
||||
misses atomic.Int64
|
||||
readDirHits atomic.Int64
|
||||
@@ -24,139 +23,48 @@ type CachedClient struct {
|
||||
}
|
||||
|
||||
// NewCachedClient creates a new CachedClient.
|
||||
func NewCachedClient(impl FS, client valkey.Client, ttl time.Duration) *CachedClient {
|
||||
return &CachedClient{
|
||||
impl: impl,
|
||||
client: client,
|
||||
refreshTime: ttl,
|
||||
func NewCachedClient(impl FS, client valkey.Client, refreshtime time.Duration) *CachedClient {
|
||||
c := &CachedClient{
|
||||
impl: impl,
|
||||
client: client,
|
||||
ttl: refreshtime * 5,
|
||||
}
|
||||
|
||||
go func() {
|
||||
t := time.NewTicker(refreshtime)
|
||||
for {
|
||||
<-t.C
|
||||
c.revalidate(context.Background())
|
||||
}
|
||||
}()
|
||||
|
||||
return c
|
||||
}
|
||||
|
||||
func (c *CachedClient) ReadDir(path string) ([]os.FileInfo, error) {
|
||||
key := fmt.Sprintf("webdav:readdir:%s", path)
|
||||
ctx := context.Background()
|
||||
func (c *CachedClient) revalidate(ctx context.Context) {
|
||||
go c.revalidateReadDir(ctx)
|
||||
go c.revalidateRead(ctx)
|
||||
go c.revalidateStat(ctx)
|
||||
}
|
||||
|
||||
// Try cache
|
||||
val, err := c.client.Do(ctx, c.client.B().Get().Key(key).Build()).ToString()
|
||||
if err == nil && val != "" {
|
||||
var cached []File
|
||||
if err := json.Unmarshal([]byte(val), &cached); err == nil {
|
||||
c.readDirHits.Add(1)
|
||||
c.hits.Add(1)
|
||||
slog.Debug("cache hit", "key", key)
|
||||
infos := make([]os.FileInfo, len(cached))
|
||||
for i, f := range cached {
|
||||
infos[i] = f
|
||||
func (c *CachedClient) scanAndProcess(ctx context.Context, match string, process func(key string) error) {
|
||||
cursor := uint64(0)
|
||||
for {
|
||||
res, err := c.client.Do(ctx, c.client.B().Scan().Cursor(cursor).Match(match).Build()).AsScanEntry()
|
||||
if err != nil {
|
||||
slog.Error("scan failed", "match", match, "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
for _, key := range res.Elements {
|
||||
if err := process(key); err != nil {
|
||||
slog.Debug("process failed", "key", key, "err", err)
|
||||
}
|
||||
return infos, nil
|
||||
}
|
||||
|
||||
cursor = res.Cursor
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// Cache miss
|
||||
c.misses.Add(1)
|
||||
slog.Debug("cache miss", "key", key)
|
||||
infos, err := c.impl.ReadDir(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Serialize and cache
|
||||
cached := make([]File, 0, len(infos))
|
||||
for _, info := range infos {
|
||||
// Type assert to ensure we capture ContentType and Path
|
||||
if fi, ok := info.(FileInfo); ok {
|
||||
cached = append(cached, File{
|
||||
name: fi.Name(),
|
||||
size: fi.Size(),
|
||||
mode: fi.Mode(),
|
||||
modTime: fi.ModTime(),
|
||||
isDir: fi.IsDir(),
|
||||
contentType: fi.ContentType(),
|
||||
path: fi.Path(),
|
||||
})
|
||||
} else {
|
||||
// Fallback if underlying impl returns generic os.FileInfo
|
||||
cached = append(cached, File{
|
||||
name: info.Name(),
|
||||
size: info.Size(),
|
||||
mode: info.Mode(),
|
||||
modTime: info.ModTime(),
|
||||
isDir: info.IsDir(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
bytes, err := json.Marshal(cached)
|
||||
if err == nil {
|
||||
c.client.Do(ctx, c.client.B().Set().Key(key).Value(string(bytes)).Ex(c.refreshTime).Build())
|
||||
}
|
||||
|
||||
return infos, nil
|
||||
}
|
||||
|
||||
func (c *CachedClient) Read(path string) ([]byte, error) {
|
||||
key := fmt.Sprintf("webdav:read:%s", path)
|
||||
ctx := context.Background()
|
||||
|
||||
// Try cache
|
||||
val, err := c.client.Do(ctx, c.client.B().Get().Key(key).Build()).AsBytes()
|
||||
if err == nil && len(val) > 0 {
|
||||
c.readHits.Add(1)
|
||||
c.hits.Add(1)
|
||||
slog.Debug("cache hit", "key", key)
|
||||
return val, nil
|
||||
}
|
||||
|
||||
// Cache miss
|
||||
c.misses.Add(1)
|
||||
slog.Debug("cache miss", "key", key)
|
||||
data, err := c.impl.Read(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Cache
|
||||
c.client.Do(ctx, c.client.B().Set().Key(key).Value(valkey.BinaryString(data)).Ex(c.refreshTime).Build())
|
||||
|
||||
return data, nil
|
||||
}
|
||||
|
||||
func (c *CachedClient) Stat(path string) (info os.FileInfo, err error) {
|
||||
key := fmt.Sprintf("webdav:stat:%s", path)
|
||||
ctx := context.Background()
|
||||
|
||||
// Try cache
|
||||
val, err := c.client.Do(ctx, c.client.B().Get().Key(key).Build()).ToString()
|
||||
if err == nil && val != "" {
|
||||
if err := json.Unmarshal([]byte(val), &info); err == nil {
|
||||
c.hits.Add(1)
|
||||
slog.Debug("cache hit", "key", key)
|
||||
|
||||
return info, nil
|
||||
}
|
||||
}
|
||||
|
||||
// Cache miss
|
||||
c.misses.Add(1)
|
||||
slog.Debug("cache miss", "key", key)
|
||||
fileInfo, err := c.impl.Stat(path)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Serialize and cache
|
||||
// Type assert to ensure we capture ContentType and Path
|
||||
info = ParseToFile(fileInfo)
|
||||
|
||||
bytes, err := json.Marshal(info)
|
||||
if err == nil {
|
||||
c.client.Do(ctx, c.client.B().Set().Key(key).Value(string(bytes)).Ex(c.refreshTime).Build())
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (c *CachedClient) Invalidate(path string) {
|
||||
ctx := context.Background()
|
||||
c.client.Do(ctx, c.client.B().Del().Key(fmt.Sprintf("webdav:readdir:%s", path)).Key(fmt.Sprintf("webdav:read:%s", path)).Build())
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user