72 lines
1.5 KiB
Go
72 lines
1.5 KiB
Go
package filesystem
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"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
|
|
cache valkey.Client
|
|
ttl time.Duration
|
|
hits atomic.Int64
|
|
misses atomic.Int64
|
|
readDirHits atomic.Int64
|
|
readHits atomic.Int64
|
|
}
|
|
|
|
// NewCachedClient creates a new CachedClient.
|
|
func NewCachedClient(impl FS, client valkey.Client, refreshtime time.Duration, lifetime time.Duration) *CachedClient {
|
|
c := &CachedClient{
|
|
impl: impl,
|
|
cache: client,
|
|
ttl: lifetime,
|
|
}
|
|
|
|
go func() {
|
|
t := time.NewTicker(refreshtime)
|
|
for {
|
|
<-t.C
|
|
c.revalidate(context.Background())
|
|
}
|
|
}()
|
|
|
|
return c
|
|
}
|
|
|
|
func (c *CachedClient) revalidate(ctx context.Context) {
|
|
slog.Info("starting revalidation")
|
|
go c.revalidateReadDir(ctx)
|
|
go c.revalidateRead(ctx)
|
|
go c.revalidateStat(ctx)
|
|
}
|
|
|
|
func (c *CachedClient) scanAndProcess(ctx context.Context, match string, process func(key string) error) {
|
|
cursor := uint64(0)
|
|
for {
|
|
res, err := c.cache.Do(ctx, c.cache.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)
|
|
}
|
|
}
|
|
|
|
cursor = res.Cursor
|
|
if cursor == 0 {
|
|
break
|
|
}
|
|
}
|
|
}
|