FranklinWH-Toolbox/service/cache/cache_aside.go

package cache

import (
	"context"
	"errors"
	"time"

	"fwh-toolbox-server/cache/redisx"

	"go.uber.org/zap"
)

// LoaderFunc 定义
type LoaderFunc[T any] func(ctx context.Context) (*T, error)

// GetOrSet: cache-aside 模式，带分布式锁防击穿与异步写缓存
func GetOrSet[T any](
	ctx context.Context,
	key string,
	loader LoaderFunc[T],
	opts ...Option,
) (*T, error) {
	o := defaultOptions()
	for _, op := range opts {
		op(o)
	}
	full := fullKey(o.prefix, key)

	// 1. 尝试缓存读取（JSON）
	var out T
	if err := redisx.GetJSON(ctx, full, &out); err == nil {
		onHit(full)
		return &out, nil
	} else {
		onMiss(full)
	}

	// 2. 未命中：尝试加分布式锁（如果启用）
	if o.lockEnabled {
		lockKey := full + ":lock"
		lock, _, err := redisx.TryLock(ctx, lockKey, o.lockTTL)
		if err == nil && lock != nil {
			// acquired lock
			defer func() {
				if err := lock.Release(ctx); err != nil {
					// log but don't break business
					zap.L().Error("cachex: failed to release lock", zap.String("key", lockKey), zap.String("err", err.Error()))
				}
			}()

			// double check cache (防止其它线程已写入)
			if err := redisx.GetJSON(ctx, full, &out); err == nil {
				onHit(full)
				return &out, nil
			}

			// 调用 loader 加载真实数据
			data, err := loader(ctx)
			if err != nil {
				return nil, err
			}
			if data == nil {
				// 业务决定：我们不写空值，直接返回 not found
				return nil, errors.New("cachex: loader returned nil")
			}

			// 写入缓存（可异步）
			if o.async {
				go func() {
					if err := redisx.SetJSON(context.Background(), full, data, o.ttl); err != nil {
						zap.L().Error("cachex: async set failed", zap.String("full", full), zap.String("err", err.Error()))
					}
				}()
			} else {
				if err := redisx.SetJSON(ctx, full, data, o.ttl); err != nil {
					zap.L().Error("cachex: set failed", zap.String("full", full), zap.String("err", err.Error()))
				}
			}
			return data, nil
		}
		// 未能获得锁：改为短轮询等待缓存（其他 worker 可能正在写）
		start := time.Now()
		for {
			// 尝试从缓存读取
			var r T
			if err := redisx.GetJSON(ctx, full, &r); err == nil {
				onHit(full)
				return &r, nil
			}
			if time.Since(start) >= o.waitForCache {
				// 超时则 fallback 直接调用 loader（避免长时间阻塞）
				break
			}
			time.Sleep(50 * time.Millisecond)
		}
	}

	// 最后兜底：直接调用 loader（不持锁）
	data, err := loader(ctx)
	if err != nil {
		return nil, err
	}
	if data == nil {
		return nil, errors.New("cachex: loader returned nil")
	}
	// 写入缓存（异步）
	if o.async {
		go func() {
			if err := redisx.SetJSON(context.Background(), full, data, o.ttl); err != nil {
				zap.L().Error("cachex: async set fallback failed", zap.String("full", full), zap.String("err", err.Error()))
			}
		}()
	} else {
		_ = redisx.SetJSON(ctx, full, data, o.ttl)
	}
	return data, nil
}

// GetOrSetMany: 批量模式，keys 与 loaderFunc 的约定：loader 需要知道哪些 keys 要加载
// 这里提供一个通用版本：loader 接收 keys 并返回 map[key]*T
type BatchLoaderFunc[T any] func(ctx context.Context, keys []string) (map[string]*T, error)

// GetOrSetMany: 先 MGET 缓存，找出缺失 keys，调用 loader 批量加载并回写缓存
func GetOrSetMany[T any](
	ctx context.Context,
	keys []string,
	loader BatchLoaderFunc[T],
	opts ...Option,
) (map[string]*T, error) {
	o := defaultOptions()
	for _, op := range opts {
		op(o)
	}
	// prefix + full keys
	fullKeys := make([]string, len(keys))
	for i, k := range keys {
		fullKeys[i] = fullKey(o.prefix, k)
	}

	// MGET via redisx.getClient().MGet
	out, missing, err := redisx.MGetJSON[T](ctx, fullKeys...)
	if err != nil {
		return nil, err
	}

	if len(missing) == 0 {
		// 全命中
		return out, nil
	}

	// 批量 loader
	loaded, err := loader(ctx, missing)
	if err != nil {
		return nil, err
	}

	fullKeysLoaded := make(map[string]*T)
	for k, v := range loaded {
		// prefix + full keys
		fullKeysLoaded[fullKey(o.prefix, k)] = v
		out[k] = v
	}

	// 回写缓存（批量 MSET）
	if err := redisx.MSetJSON[T](ctx, fullKeysLoaded, o.ttl); err != nil {
		zap.L().Error("cachex: batch mset exec failed: %v", zap.String("err", err.Error()))
	}
	return out, nil
}