cache

Cache

A flexible and type-safe caching library for Go with support for both in-memory and Redis backends.

Features

• Multiple Backend Support: In-memory cache and Redis cache implementations
• Type-Safe API: Generic-based bucket operations for type safety
• Context Support: Full context.Context integration for cancellation and timeouts
• Production-Safe Error Handling: All operations return errors instead of panicking
• Label-Based Filtering: Organize and query cached items using labels
• Time-Based Updates: Conditional updates based on timestamps
• Expiration Support: Built-in TTL and expiration callbacks
• Collections: Manage sets of members associated with keys
• Timeline: Time-indexed state storage with sparse field updates, out-of-order insertion support, and config-driven retention policies
• Configurable Timeouts: Per-client timeout configuration for Redis operations

Installation

go get github.com/leonkaihao/cache/v2@v2

Quick Start

In-Memory Cache

package main

import (
    "context"
    "log"
    "time"
    cache "github.com/leonkaihao/cache/v2/pkg/client/mem"
)

type User struct {
    ID   int    `json:"id"`
    Name string `json:"name"`
    Age  int    `json:"age"`
}

func main() {
    ctx := context.Background()
    
    // Create client
    cli := cache.NewClient()
    
    // Create bucket with error handling
    userBkt, err := cache.NewBucket[User](cli, "users")
    if err != nil {
    log.Fatal(err)
}

Timeline for Time-Series State Management

ctx := context.Background()

// Create timeline
timeline := cli.Timeline("device_states")

// Set retention policy (config-driven, in-memory)
timeline.WithRetention(model.RetentionPolicy{
    MaxCount:    100,
    MaxDuration: 2 * time.Hour,
    Strategy:    model.RetentionMax,
})

// Record device state
if err := timeline.Append(ctx, "device_A", time.Now(), map[string]string{
    "zones":   "Z1,Z3",
    "beacons": "B5",
    "battery": "85",
}, false); err != nil {
    log.Fatal(err)
}

// Sparse update (only zones changed)
if err := timeline.Append(ctx, "device_A", time.Now().Add(5*time.Minute), map[string]string{
    "zones": "Z1,Z3,Z5",
}, false); err != nil {
    log.Fatal(err)
}

// Query current state (merged from all updates)
states, err := timeline.GetLatest(ctx, []string{"device_A"})
if err != nil {
    log.Fatal(err)
}
state := states[0]
log.Printf("Current state: zones=%s, beacons=%s, battery=%s\n",
    state["zones"], state["beacons"], state["battery"])
// Output: zones=Z1,Z3,Z5, beacons=B5, battery=85

// Query historical state
historicalStates, err := timeline.GetAt(ctx, []string{"device_A"}, time.Now().Add(-10*time.Minute))
if err != nil {
    log.Fatal(err)
}
historicalState := historicalStates[0]

// Insert out-of-order event
lateEvent := time.Now().Add(-1 * time.Hour)
if err := timeline.Insert(ctx, "device_A", lateEvent, map[string]string{
    "zones": "Z1,Z2",
}, false); err != nil {
    log.Fatal(err)
}

// Find affected states for recomputation
affected, err := timeline.GetAffectedRange(ctx, "device_A", lateEvent)
if err != nil {
    log.Fatal(err)
}
log.Printf("States needing recomputation: %d\n", len(affected))

Redis Cache

package main

import (
    "context"
    "log"
    "time"
    cache "github.com/leonkaihao/cache/v2/pkg/client/redis"
    "github.com/leonkaihao/cache/v2/pkg/coding"
)

type Product struct {
    ID    string  `json:"id"`
    Name  string  `json:"name"`
    Price float64 `json:"price"`
}

func main() {
    ctx := context.Background()
    
    // Create Redis client with custom timeout
    cli := cache.NewClient(
        "localhost:6379", 
        "password", 
        0,
        cache.WithTimeout(5*time.Second), // optional: custom timeout
    )
    
    // Create bucket with error handling
    productBkt, err := cache.NewBucket[Product](
        cli, 
        "products",
        coding.NewJsonCoder(),
    )
    if err != nil {
        log.Fatal(err)
    }
    cli.WithBucket(productBkt)
    
    // Update document with context
    doc, err := productBkt.Update(ctx, "prod1", &Product{
        ID:    "p001",
        Name:  "Laptop",
        Price: 999.99,
    })
    if err != nil {
        log.Fatal(err)
    }
    
    log.Printf("Created product: %s\n", doc.Key())
}

Context and Timeout Management

All operations accept a context.Context for cancellation and timeouts:

ctx := context.Background()

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

doc, err := bucket.Update(ctx, "key1", data)
if err != nil {
    // Handle timeout or cancellation
    log.Printf("Operation failed: %v", err)
}

// Redis client has default timeout (1 second)
// You can customize it:
cli := cache.NewClient("localhost:6379", "pass", 0, 
    cache.WithTimeout(5*time.Second))

// Context deadline takes precedence over default timeout
// If context has shorter deadline, it will be used
shortCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
_, err = bucket.Update(shortCtx, "key", data) // Uses 100ms timeout

API Overview

CacheClient

The main client interface for managing buckets and collections:

type CacheClient interface {
    WithBucket(CacheBucket) CacheBucket
    Bucket(name string) CacheBucket
    Buckets() []CacheBucket
    RemoveBucket(bktName string)
    
    Collection(name string) CacheCollection
    Collections() []CacheCollection
    RemoveCollection(name string)
    
    Timeline(name string) CacheTimeline
    Timelines() []CacheTimeline
    RemoveTimeline(name string) error
}

CacheBucket

Type-safe storage for cached objects (all methods accept context and return errors):

type CacheBucket interface {
    Name() string
    Docs(ctx context.Context, keys []string) ([]CacheDoc, error)
    Values(ctx context.Context, keys []string) ([]any, error)
    Update(ctx context.Context, key string, data any) (CacheDoc, error)
    UpdateWithTs(ctx context.Context, key string, data any, ts time.Time) (CacheDoc, bool, error)
    Filter(ctx context.Context, labelFilters ...[]string) ([]string, error)
    Scan(ctx context.Context, match string) ([]string, error)
    Remove(ctx context.Context, keys []string) error
    Clear(ctx context.Context) error
    Delete(ctx context.Context) error
}

CacheDoc

Individual cached document with metadata:

type CacheDoc interface {
    Key() string
    Val(ctx context.Context) (any, error)
    SetValue(ctx context.Context, val any) error
    Labels(ctx context.Context) (LabelSet, error)
    AddLabels(ctx context.Context, labels []string) error
    RemoveLabels(ctx context.Context, labels []string) error
    Delete(ctx context.Context) error
    WithTime(ctx context.Context, ts time.Time) error
    SetValueWithTs(ctx context.Context, val any, ts time.Time) (bool, error)
    Time(ctx context.Context) (time.Time, error)
    Expire(d time.Duration, onExpire func(CacheDoc)) error
    CancelExpire() error
}

CacheCollection

Manage sets of members:

type CacheCollection interface {
    Name() string
    Keys(ctx context.Context) ([]string, error)
    Add(ctx context.Context, key string, members []string) error
    Remove(ctx context.Context, key string, members []string) error
    MembersMap(ctx context.Context, key string) (MembersMap, error)
    MembersMaps(ctx context.Context, keys []string) ([]MembersMap, error)
    Clear(ctx context.Context, key string) error
    ClearAll(ctx context.Context) error
    Delete(ctx context.Context) error
}

CacheTimeline

Time-indexed state storage for managing historical data:

type CacheTimeline interface {
    Name() string
    
    // Write operations
    Append(ctx context.Context, key string, ts time.Time, data map[string]string, force bool) error
    Insert(ctx context.Context, key string, ts time.Time, data map[string]string, force bool) error
    
    // Batch query operations (results parallel to keys, nil = no data, BatchError = partial failure)
    GetAt(ctx context.Context, keys []string, ts time.Time) ([]map[string]string, error)
    GetExact(ctx context.Context, keys []string, ts time.Time) ([]map[string]string, error)
    GetRange(ctx context.Context, keys []string, start, end time.Time) ([][]*TimeValue, error)
    GetLatest(ctx context.Context, keys []string) ([]map[string]string, error)
    Timeline(ctx context.Context, key string) ([]*TimeValue, error)
    GetAffectedRange(ctx context.Context, key string, insertedAt time.Time) ([]*TimeValue, error)
    
    // Retention management (config-driven, in-memory)
    WithRetention(policy RetentionPolicy) CacheTimeline
    GetRetention() RetentionPolicy
    
    // Label management
    AddKeyLabels(ctx context.Context, key string, labels []string) error
    RemoveKeyLabels(ctx context.Context, key string, labels []string) error
    KeyLabels(ctx context.Context, key string) (LabelSet, error)
    
    // Management (Keys supports label-based filtering: OR within array, AND between arrays)
    Keys(ctx context.Context, labelFilters ...[]string) ([]string, error)
    GetUpdatedKeys(ctx context.Context, after time.Time) ([]string, error)
    Remove(ctx context.Context, keys []string) error
    Clear(ctx context.Context) error
    Delete(ctx context.Context) error
}

Advanced Features

Time-Based Updates

Only update cache if the new data is newer (equal timestamps are rejected):

ctx := context.Background()
ts := time.Now()

// First update
doc, updated, err := bucket.UpdateWithTs(ctx, "key1", data, ts)
if err != nil {
    log.Fatal(err)
}
log.Printf("Updated: %v", updated) // true

// Try to update with same timestamp (rejected)
_, updated, err = bucket.UpdateWithTs(ctx, "key1", newData, ts)
if err != nil {
    log.Fatal(err)
}
log.Printf("Updated: %v", updated) // false - equal timestamp rejected

// Update with newer timestamp (succeeds)
newerTs := ts.Add(time.Second)
_, updated, err = bucket.UpdateWithTs(ctx, "key1", newerData, newerTs)
if err != nil {
    log.Fatal(err)
}
log.Printf("Updated: %v", updated) // true

Expiration with Callbacks

ctx := context.Background()

doc, err := bucket.Update(ctx, "session", sessionData)
if err != nil {
    log.Fatal(err)
}

err = doc.Expire(time.Hour, func(d model.CacheDoc) {
    log.Printf("Session expired: %s", d.Key())
    _ = d.Delete(context.Background())
})
if err != nil {
    log.Fatal(err)
}

// Cancel expiration if needed
err = doc.CancelExpire()
if err != nil {
    log.Fatal(err)
}

Label-Based Filtering

ctx := context.Background()

// Add labels
if err := doc1.AddLabels(ctx, []string{"active", "premium"}); err != nil {
    log.Fatal(err)
}
if err := doc2.AddLabels(ctx, []string{"active", "free"}); err != nil {
    log.Fatal(err)
}

// Filter by single label
activeKeys, err := bucket.Filter(ctx, []string{"active"})
if err != nil {
    log.Fatal(err)
}
// Returns both doc1 and doc2

// Filter by multiple labels (OR within array, AND between arrays)
premiumKeys, err := bucket.Filter(ctx, []string{"premium", "free"})
if err != nil {
    log.Fatal(err)
}
// Returns doc1 and doc2

// Check labels
labels, err := doc1.Labels(ctx)
if err != nil {
    log.Fatal(err)
}
hasActive := labels.CheckAnd([]string{"active", "premium"}) // true
hasTrial := labels.CheckOr([]string{"active", "trial"})    // true

Collections for Set Operations

ctx := context.Background()
clt := cli.Collection("user_groups")

// Add members to sets (empty members rejected)
if err := clt.Add(ctx, "admins", []string{"user1", "user2"}); err != nil {
    log.Fatal(err)
}
// Merges with existing
if err := clt.Add(ctx, "admins", []string{"user2", "user3"}); err != nil {
    log.Fatal(err)
}

// Check membership
members, err := clt.MembersMap(ctx, "admins")
if err != nil {
    log.Fatal(err)
}
if members != nil {
    exists := members.Exists("user1") // true
    list := members.List()            // ["user1", "user2", "user3"]
}

// Remove members
if err := clt.Remove(ctx, "admins", []string{"user2"}); err != nil {
    log.Fatal(err)
}

Timeline Batch Queries and Labels

Timeline supports batch queries (multiple keys in one call) and label-based filtering for efficient time-series operations.

Batch Queries

Query multiple keys at once for better performance:

ctx := context.Background()
timeline := cli.Timeline("device_states")

// Query multiple devices at once
states, err := timeline.GetLatest(ctx, []string{"device_A", "device_B", "device_C"})
if err != nil {
    log.Fatal(err)
}

// Results are parallel to input keys
// nil at position i means key i has no data (not an error)
for i, key := range []string{"device_A", "device_B", "device_C"} {
    if states[i] == nil {
        log.Printf("%s: no data", key)
    } else {
        log.Printf("%s: battery=%s zones=%s", key, states[i]["battery"], states[i]["zones"])
    }
}

// Batch historical queries
ts := time.Now().Add(-1 * time.Hour)
historicalStates, err := timeline.GetAt(ctx, []string{"device_A", "device_B"}, ts)
if err != nil {
    log.Fatal(err)
}

// Batch range queries return [][]*TimeValue
// Outer nil = key has no data in range
// Inner slice elements are always non-nil pointers
start := time.Now().Add(-24 * time.Hour)
end := time.Now()
ranges, err := timeline.GetRange(ctx, []string{"device_A", "device_B"}, start, end)
if err != nil {
    log.Fatal(err)
}
for i, key := range []string{"device_A", "device_B"} {
    if ranges[i] == nil {
        log.Printf("%s: no data in range", key)
    } else {
        log.Printf("%s: %d time points in range", key, len(ranges[i]))
        for _, tv := range ranges[i] {
            log.Printf("  %v: battery=%s", tv.Time, tv.Value["battery"])
        }
    }
}

Label-Based Key Filtering

Organize timeline keys with labels for semantic grouping and filtering:

ctx := context.Background()
timeline := cli.Timeline("device_states")

// Add labels to categorize devices
if err := timeline.AddKeyLabels(ctx, "device_A", []string{"sensor", "outdoor", "region-west"}); err != nil {
    log.Fatal(err)
}
if err := timeline.AddKeyLabels(ctx, "device_B", []string{"sensor", "indoor", "region-west"}); err != nil {
    log.Fatal(err)
}
if err := timeline.AddKeyLabels(ctx, "device_C", []string{"actuator", "outdoor", "region-east"}); err != nil {
    log.Fatal(err)
}

// Query labels for a key
labels, err := timeline.KeyLabels(ctx, "device_A")
if err != nil {
    log.Fatal(err)
}
hasSensor := labels.CheckAnd([]string{"sensor", "outdoor"}) // true
hasIndoor := labels.CheckOr([]string{"indoor", "outdoor"})  // true

// Filter keys by labels (OR within array, AND between arrays)
// All keys with no filter
allKeys, err := timeline.Keys(ctx)

// Keys with "outdoor" label
outdoorKeys, err := timeline.Keys(ctx, []string{"outdoor"})
// Returns: ["device_A", "device_C"]

// Keys with "sensor" OR "actuator" label
deviceKeys, err := timeline.Keys(ctx, []string{"sensor", "actuator"})
// Returns: ["device_A", "device_B", "device_C"]

// Keys matching (outdoor OR indoor) AND region-west AND sensor
westSensors, err := timeline.Keys(ctx, 
    []string{"outdoor", "indoor"},  // OR: any of these
    []string{"region-west"},         // AND this
    []string{"sensor"},              // AND this
)
// Returns: ["device_A", "device_B"]

// Combine label filtering with batch queries
states, err := timeline.GetLatest(ctx, westSensors)
if err != nil {
    log.Fatal(err)
}
for i, key := range westSensors {
    if states[i] != nil {
        log.Printf("%s: %v", key, states[i])
    }
}

// Remove labels
if err := timeline.RemoveKeyLabels(ctx, "device_A", []string{"outdoor"}); err != nil {
    log.Fatal(err)
}

Partial Failure Handling with BatchError

Batch operations return *BatchError for partial failures, allowing you to use successful results even when some keys fail:

import "errors"

ctx := context.Background()
timeline := cli.Timeline("device_states")

deviceIDs := []string{"device_A", "device_B", "device_C", "device_D"}
states, err := timeline.GetLatest(ctx, deviceIDs)

if err != nil {
    var batchErr *model.BatchError
    if errors.As(err, &batchErr) {
        // Partial failure - some devices succeeded
        log.Printf("Retrieved %d/%d devices successfully", 
            batchErr.Total-batchErr.Failed, batchErr.Total)
        
        // Process successful results (nil = no data, which is not an error)
        for i, state := range states {
            if state != nil {
                log.Printf("%s: %v", deviceIDs[i], state)
            } else if _, failed := batchErr.KeyErrors[deviceIDs[i]]; !failed {
                log.Printf("%s: no data available", deviceIDs[i])
            }
        }
        
        // Handle failed devices
        for key, keyErr := range batchErr.KeyErrors {
            log.Printf("Device %s failed: %v", key, keyErr)
        }
    } else {
        // Total failure - no results available
        log.Fatalf("Timeline query failed: %v", err)
    }
} else {
    // Complete success - process all results
    for i, state := range states {
        if state != nil {
            log.Printf("%s: %v", deviceIDs[i], state)
        } else {
            log.Printf("%s: no data", deviceIDs[i])
        }
    }
}

Understanding Nil in Batch Results

// GetLatest/GetAt/GetExact: []map[string]string
states, err := timeline.GetLatest(ctx, []string{"key1", "key2"})
// states[i] == nil → "key i has no data" (NOT an error, just no time points exist)
// err != nil && errors.As(err, &BatchError{}) → partial failure (check KeyErrors for which keys failed)

// GetRange: [][]*TimeValue
ranges, err := timeline.GetRange(ctx, []string{"key1", "key2"}, start, end)
// ranges[i] == nil → "key i has no time points in the range"
// ranges[i][j] → Always non-nil if ranges[i] != nil (pointers avoid copying)

Query Keys by Update Time

Find keys that were updated after a specific timestamp:

ctx := context.Background()
timeline := cli.Timeline("device_states")

// Get all keys updated in the last hour
lastHour := time.Now().Add(-1 * time.Hour)
recentKeys, err := timeline.GetUpdatedKeys(ctx, lastHour)
if err != nil {
    log.Fatal(err)
}
log.Printf("Keys updated since %v: %v", lastHour, recentKeys)

// Query only recently updated devices
if len(recentKeys) > 0 {
    states, err := timeline.GetLatest(ctx, recentKeys)
    if err != nil {
        log.Fatal(err)
    }
    // Process recent device states
}

Error Handling Philosophy

v2.0.0 is production-safe: All operations return errors instead of panicking or using Logger.Fatal(). This allows your application to:

1. Gracefully handle failures - No unexpected crashes
2. Implement retry logic - Wrap operations in your own retry mechanism
3. Log errors appropriately - Use your application's logging system
4. Test error paths - Write tests that verify error handling

ctx := context.Background()

// Always check errors
doc, err := bucket.Update(ctx, "key1", data)
if err != nil {
    // Handle error appropriately
    log.Printf("Failed to update cache: %v", err)
    // Optionally retry, use fallback, or propagate error
    return fmt.Errorf("cache update failed: %w", err)
}

// Context cancellation is detected
ctx, cancel := context.WithCancel(context.Background())
cancel() // Cancel immediately

err = doc.AddLabels(ctx, []string{"label1"})
if errors.Is(err, context.Canceled) {
    log.Println("Operation was cancelled")
}

Migration Guide (v1.x → v2.0.0)

Breaking Changes

1. All operations now accept context.Context as first parameter
2. All operations now return error
3. SetValueWithTs returns (bool, error) instead of (CacheDoc, bool)
4. UpdateWithTs returns (CacheDoc, bool, error) instead of (CacheDoc, bool)
5. Remove returns error instead of []CacheDoc
6. GetLastErrors() removed - errors are returned directly
7. Empty members in Collection.Add() now rejected
8. Redis client accepts WithTimeout() option
9. NewBucket returns (CacheBucket, error) for both backends

Migration Examples

Before (v1.x):

cli := mem.NewClient()
bucket := mem.NewBucket[User](cli, "users")
doc := bucket.Update("key1", data)
doc.AddLabels([]string{"active"})
keys := bucket.Filter([]string{"active"})
bucket.Clear()

After (v2.0.0):

ctx := context.Background()
cli := mem.NewClient()
bucket, err := mem.NewBucket[User](cli, "users")
if err != nil {
    return err
}
doc, err := bucket.Update(ctx, "key1", data)
if err != nil {
    return err
}
if err := doc.AddLabels(ctx, []string{"active"}); err != nil {
    return err
}
keys, err := bucket.Filter(ctx, []string{"active"})
if err != nil {
    return err
}
if err := bucket.Clear(ctx); err != nil {
    return err
}

Redis Timeout Configuration:

// Before (v1.x): No timeout configuration
cli := redis.NewClient("localhost:6379", "pass", 0)

// After (v2.0.0): Optional timeout configuration
cli := redis.NewClient("localhost:6379", "pass", 0, 
    redis.WithTimeout(5*time.Second)) // default is 1s

Testing

# Run unit tests
make test

# Run integration tests (requires Redis)
make test/integration

# Run benchmarks
make test/bench

# Run specific test
go test ./pkg/client/mem/... -run TestBucket -v

Project Structure

cache/
├── cmd/
│   ├── sample-mem/      # In-memory cache example
│   └── sample-redis/    # Redis cache example
├── pkg/
│   ├── client/
│   │   ├── mem/         # In-memory implementation
│   │   ├── redis/       # Redis implementation
│   │   └── test/        # Shared test suite
│   ├── model/           # Core interfaces
│   ├── coding/          # Encoding/decoding utilities
│   ├── consts/          # Constants
│   └── logger/          # Logging interfaces
└── Makefile

Requirements

• Go 1.23 or higher
• Redis server (for Redis backend)

Dependencies

• go-redis/v9 - Redis client for Go
• protobuf - Protocol buffer support
• testify - Testing toolkit

Contributing

Contributions are welcome! Please feel free to submit issues or pull requests.

License

This project is licensed under the MIT License - see the LICENSE file for details.

Examples

Full examples can be found in the cmd/ directory:

• cmd/sample-mem/main.go - In-memory cache usage with context and error handling
• cmd/sample-redis/main.go - Redis cache usage with timeout configuration