mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-10-02 23:06:36 +08:00
refactor(oauth): replace legacy oauth cache with standard ram cache and add pubsub synchronization
- Replaced custom map-based cache in apps/oauth/cache.go with standard pkg/cache/ram framework. - Implemented Redis Pub/Sub invalidation channels for distributed token and user cache synchronization. - Created apps/oauth/cache_test.go to verify local cache operations and pub/sub broadcasts. refactor(cache): generic RAM cache with CoW and unified preheating Replaced L2 Redis cache and old cache package with process-local generic pkg/cache/ram. Implemented Copy-on-Write for reads, fine-grained locks per type for writes, and unified preheating in bootstrap. Changed cache invalidation to lazy-loading to resolve SQLite deadlocks during transactions.
This commit is contained in:
@@ -118,7 +118,7 @@ func startThingCacheInvalidationListener() {
|
|||||||
|
|
||||||
| 域 | 文件 | L1 | L2 | pub/sub |
|
| 域 | 文件 | L1 | L2 | pub/sub |
|
||||||
| :--- | :--- | :--- | :--- | :--- |
|
| :--- | :--- | :--- | :--- | :--- |
|
||||||
| 系统配置 | `repository/system_config_cache.go` | Otter | Redis Hash | `system:config_invalidation` ✅ |
|
| 系统配置 | `repository/system_config_cache.go` | `pkg/cache/store` | ❌ 无 Redis 缓存 | `system:config_broadcast` (别名 `system:config_invalidation`) ✅ |
|
||||||
| CAPTCHA 运行时 | `apps/cap/runtime_settings.go` | atomic.Pointer | (借配置 Redis) | 订阅 `system:config_invalidation` ✅ |
|
| CAPTCHA 运行时 | `apps/cap/runtime_settings.go` | atomic.Pointer | (借配置 Redis) | 订阅 `system:config_invalidation` ✅ |
|
||||||
| 上传元数据 | `apps/upload/cache/meta_cache.go` | Otter | Redis JSON | `upload:meta_invalidation` ✅ |
|
| 上传元数据 | `apps/upload/cache/meta_cache.go` | Otter | Redis JSON | `upload:meta_invalidation` ✅ |
|
||||||
| 上传访问白名单 | `apps/upload/cache/access_cache.go` | 进程内 TTL | (借配置读路径) | `upload:file_access_invalidation` ✅ |
|
| 上传访问白名单 | `apps/upload/cache/access_cache.go` | 进程内 TTL | (借配置读路径) | `upload:file_access_invalidation` ✅ |
|
||||||
|
|||||||
+3
-3
@@ -75,7 +75,7 @@ flowchart LR
|
|||||||
|
|
||||||
1. **前端**:~~全局认证瀑布流~~ ✅ 已改为 layout 即时渲染 + 子页面 `RequireAuth` 自行处理未登录态;~~Admin 重模块无 `dynamic()` 分割~~ ✅ database/logs/settings 已懒加载子模块。其余路由 `page.tsx` 仍为 `"use client"`(静态导出下 RSC 收益有限,待逐步薄壳化)。
|
1. **前端**:~~全局认证瀑布流~~ ✅ 已改为 layout 即时渲染 + 子页面 `RequireAuth` 自行处理未登录态;~~Admin 重模块无 `dynamic()` 分割~~ ✅ database/logs/settings 已懒加载子模块。其余路由 `page.tsx` 仍为 `"use client"`(静态导出下 RSC 收益有限,待逐步薄壳化)。
|
||||||
2. **后端**:文件服务路径(`/f/{id}`)仍是最高频热点;~~磁盘缓存全局互斥锁~~ ✅ 已改为 `RWMutex` + `singleflight`,但 WebP miss 仍在请求线程内同步编码,部署预热与异步回退原图尚未落地。
|
2. **后端**:文件服务路径(`/f/{id}`)仍是最高频热点;~~磁盘缓存全局互斥锁~~ ✅ 已改为 `RWMutex` + `singleflight`,但 WebP miss 仍在请求线程内同步编码,部署预热与异步回退原图尚未落地。
|
||||||
3. **参数中心**:~~`GetByKey` 每次直打 Redis~~ ✅ 已加 Otter v2 进程内缓存(`pkg/cache/ram`),读路径为 RAM → Redis → DB;管理员写配置后统一失效 RAM + Redis,并通过 pub/sub 同步多节点本地缓存。
|
3. **参数中心**:~~`GetByKey` 每次直打 Redis~~ ✅ 已统一使用底层的进程内缓存库(`pkg/cache/store`),读路径直接为 RAM → DB(无 Redis 数据缓存);管理员写配置后通过 Redis pub/sub 进行广播(`system:config_broadcast`),多节点本地触发全量预热/刷新,实现最终一致性。
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
@@ -395,8 +395,8 @@ if (loading || !user) {
|
|||||||
|
|
||||||
| # | 设计 | 位置 |
|
| # | 设计 | 位置 |
|
||||||
|---|------|------|
|
|---|------|------|
|
||||||
| 1 | 系统配置三层缓存 RAM → Redis → DB | `pkg/cache/ram`, `system_config_cache.go`, `GetByKey` |
|
| 1 | 系统配置两层缓存 RAM → DB | `pkg/cache/store`, `system_config_cache.go`, `GetByKey` |
|
||||||
| 2 | 系统配置统一失效 + 多节点 pub/sub | `InvalidateSystemConfigCache`, `InvalidateAllSystemConfigCaches` |
|
| 2 | 系统配置统一刷新 + 多节点 pub/sub 预热广播 | `InvalidateSystemConfigCache`, `InvalidateAllSystemConfigCaches` |
|
||||||
| 3 | Storage Backend 单例 + 5s TTL + pub/sub 失效 | `internal/storage/storage.go` — `Active()` |
|
| 3 | Storage Backend 单例 + 5s TTL + pub/sub 失效 | `internal/storage/storage.go` — `Active()` |
|
||||||
| 4 | 推送事件/渠道 24h Redis 缓存 + GORM hook 失效 | `internal/model/push_event.go`, `push_channel.go` |
|
| 4 | 推送事件/渠道 24h Redis 缓存 + GORM hook 失效 | `internal/model/push_event.go`, `push_channel.go` |
|
||||||
| 5 | 风控日志异步批写 ClickHouse(1 万缓冲 + 1000 条/1s + 429 背压) | `internal/apps/risk_control/` |
|
| 5 | 风控日志异步批写 ClickHouse(1 万缓冲 + 1000 条/1s + 429 背压) | `internal/apps/risk_control/` |
|
||||||
|
|||||||
@@ -6,26 +6,30 @@ package config
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/Rain-kl/Wavelet/internal/db"
|
|
||||||
"github.com/Rain-kl/Wavelet/internal/model"
|
"github.com/Rain-kl/Wavelet/internal/model"
|
||||||
"github.com/Rain-kl/Wavelet/internal/repository"
|
"github.com/Rain-kl/Wavelet/internal/repository"
|
||||||
"github.com/Rain-kl/Wavelet/internal/testhelper"
|
"github.com/Rain-kl/Wavelet/internal/testhelper"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestListVisibleSystemConfigsUsesRedisCache(t *testing.T) {
|
func TestListVisibleSystemConfigsUsesStoreCache(t *testing.T) {
|
||||||
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
|
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
|
||||||
defer cleanup()
|
defer cleanup()
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
|
|
||||||
|
|
||||||
|
repository.ResetSystemConfigRAMCacheForTest()
|
||||||
if err := repository.InvalidateVisibleSystemConfigsCache(ctx); err != nil {
|
if err := repository.InvalidateVisibleSystemConfigsCache(ctx); err != nil {
|
||||||
t.Fatalf("InvalidateVisibleSystemConfigsCache() error = %v", err)
|
t.Fatalf("InvalidateVisibleSystemConfigsCache() error = %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Warm cache
|
||||||
if _, err := repository.ListVisibleSystemConfigs(ctx); err != nil {
|
if _, err := repository.ListVisibleSystemConfigs(ctx); err != nil {
|
||||||
t.Fatalf("ListVisibleSystemConfigs() warm cache error = %v", err)
|
t.Fatalf("ListVisibleSystemConfigs() warm cache error = %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Directly insert a new config in DB (bypassing caching layer)
|
||||||
if err := dbConn.Create(&model.SystemConfig{
|
if err := dbConn.Create(&model.SystemConfig{
|
||||||
Key: "cache_probe_public_key",
|
Key: "cache_probe_public_key",
|
||||||
Value: "cache_probe_public_value",
|
Value: "cache_probe_public_value",
|
||||||
@@ -36,6 +40,7 @@ func TestListVisibleSystemConfigsUsesRedisCache(t *testing.T) {
|
|||||||
t.Fatalf("Create(cache_probe_public_key) error = %v", err)
|
t.Fatalf("Create(cache_probe_public_key) error = %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Cached call: shouldn't return the new key yet
|
||||||
cached, err := repository.ListVisibleSystemConfigs(ctx)
|
cached, err := repository.ListVisibleSystemConfigs(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("ListVisibleSystemConfigs() cached call error = %v", err)
|
t.Fatalf("ListVisibleSystemConfigs() cached call error = %v", err)
|
||||||
@@ -46,18 +51,15 @@ func TestListVisibleSystemConfigsUsesRedisCache(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
exists, err := db.Redis.Exists(ctx, db.PrefixedKey(repository.SystemConfigVisibleListRedisKey)).Result()
|
// Invalidate: triggers reload
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("Redis.Exists() error = %v", err)
|
|
||||||
}
|
|
||||||
if exists == 0 {
|
|
||||||
t.Fatal("expected visible config list cache key to exist")
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := repository.InvalidateVisibleSystemConfigsCache(ctx); err != nil {
|
if err := repository.InvalidateVisibleSystemConfigsCache(ctx); err != nil {
|
||||||
t.Fatalf("InvalidateVisibleSystemConfigsCache() error = %v", err)
|
t.Fatalf("InvalidateVisibleSystemConfigsCache() error = %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Wait for Pub/Sub delivery in test environment
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
|
||||||
|
// Refreshed call: should return the new key
|
||||||
refreshed, err := repository.ListVisibleSystemConfigs(ctx)
|
refreshed, err := repository.ListVisibleSystemConfigs(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("ListVisibleSystemConfigs() refreshed call error = %v", err)
|
t.Fatalf("ListVisibleSystemConfigs() refreshed call error = %v", err)
|
||||||
|
|||||||
@@ -5,10 +5,8 @@ package config
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
"github.com/redis/go-redis/v9"
|
|
||||||
|
|
||||||
"github.com/Rain-kl/Wavelet/internal/db"
|
"github.com/Rain-kl/Wavelet/internal/db"
|
||||||
"github.com/Rain-kl/Wavelet/internal/model"
|
"github.com/Rain-kl/Wavelet/internal/model"
|
||||||
@@ -21,6 +19,7 @@ func TestSystemConfigRAMCacheServesUntilInvalidated(t *testing.T) {
|
|||||||
defer cleanup()
|
defer cleanup()
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
|
|
||||||
|
|
||||||
repository.ResetSystemConfigRAMCacheForTest()
|
repository.ResetSystemConfigRAMCacheForTest()
|
||||||
if err := repository.InvalidateAllSystemConfigCaches(ctx); err != nil {
|
if err := repository.InvalidateAllSystemConfigCaches(ctx); err != nil {
|
||||||
t.Fatalf("InvalidateAllSystemConfigCaches() error = %v", err)
|
t.Fatalf("InvalidateAllSystemConfigCaches() error = %v", err)
|
||||||
@@ -34,15 +33,14 @@ func TestSystemConfigRAMCacheServesUntilInvalidated(t *testing.T) {
|
|||||||
t.Fatalf("GetSystemConfigByKey(site_name).Value = %q, want %q", warm.Value, "Wavelet")
|
t.Fatalf("GetSystemConfigByKey(site_name).Value = %q, want %q", warm.Value, "Wavelet")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Update DB directly (bypassing caching layer)
|
||||||
if err := dbConn.Model(&model.SystemConfig{}).
|
if err := dbConn.Model(&model.SystemConfig{}).
|
||||||
Where("key = ?", model.ConfigKeySiteName).
|
Where("key = ?", model.ConfigKeySiteName).
|
||||||
Update("value", "ram_probe_value").Error; err != nil {
|
Update("value", "ram_probe_value").Error; err != nil {
|
||||||
t.Fatalf("Update(site_name) error = %v", err)
|
t.Fatalf("Update(site_name) error = %v", err)
|
||||||
}
|
}
|
||||||
if err := db.HDel(ctx, repository.SystemConfigRedisHashKey, model.ConfigKeySiteName); err != nil {
|
|
||||||
t.Fatalf("HDel(site_name) error = %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
|
// Should still return "Wavelet" since it's cached in RAM cache
|
||||||
cached, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeySiteName)
|
cached, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeySiteName)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("GetSystemConfigByKey(site_name) cached error = %v", err)
|
t.Fatalf("GetSystemConfigByKey(site_name) cached error = %v", err)
|
||||||
@@ -51,10 +49,15 @@ func TestSystemConfigRAMCacheServesUntilInvalidated(t *testing.T) {
|
|||||||
t.Fatalf("GetSystemConfigByKey(site_name).Value = %q, want stale RAM value %q", cached.Value, "Wavelet")
|
t.Fatalf("GetSystemConfigByKey(site_name).Value = %q, want stale RAM value %q", cached.Value, "Wavelet")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Invalidate the cache (triggers refresh callback)
|
||||||
if err := repository.InvalidateSystemConfigCache(ctx, model.ConfigKeySiteName); err != nil {
|
if err := repository.InvalidateSystemConfigCache(ctx, model.ConfigKeySiteName); err != nil {
|
||||||
t.Fatalf("InvalidateSystemConfigCache(site_name) error = %v", err)
|
t.Fatalf("InvalidateSystemConfigCache(site_name) error = %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Allow some time for broadcast listener in test environment
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
|
||||||
|
// Should return the updated value now
|
||||||
refreshed, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeySiteName)
|
refreshed, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeySiteName)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("GetSystemConfigByKey(site_name) refreshed error = %v", err)
|
t.Fatalf("GetSystemConfigByKey(site_name) refreshed error = %v", err)
|
||||||
@@ -62,47 +65,48 @@ func TestSystemConfigRAMCacheServesUntilInvalidated(t *testing.T) {
|
|||||||
if refreshed.Value != "ram_probe_value" {
|
if refreshed.Value != "ram_probe_value" {
|
||||||
t.Fatalf("GetSystemConfigByKey(site_name).Value = %q, want %q", refreshed.Value, "ram_probe_value")
|
t.Fatalf("GetSystemConfigByKey(site_name).Value = %q, want %q", refreshed.Value, "ram_probe_value")
|
||||||
}
|
}
|
||||||
|
|
||||||
exists, err := db.Redis.HExists(ctx, db.PrefixedKey(repository.SystemConfigRedisHashKey), model.ConfigKeySiteName).Result()
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("HExists(site_name) error = %v", err)
|
|
||||||
}
|
|
||||||
if !exists {
|
|
||||||
t.Fatal("expected redis hash field to be repopulated after refresh")
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestInvalidateSystemConfigCacheClearsRedisField(t *testing.T) {
|
func TestInvalidateSystemConfigCacheBroadcast(t *testing.T) {
|
||||||
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
|
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
|
||||||
defer cleanup()
|
defer cleanup()
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
|
|
||||||
sc, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeySiteName)
|
|
||||||
|
repository.ResetSystemConfigRAMCacheForTest()
|
||||||
|
|
||||||
|
// Initially seed in cache
|
||||||
|
_, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeySiteName)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("GetSystemConfigByKey(site_name) error = %v", err)
|
t.Fatalf("GetSystemConfigByKey(site_name) error = %v", err)
|
||||||
}
|
}
|
||||||
_ = sc
|
|
||||||
|
|
||||||
|
// Update DB directly
|
||||||
|
if err := dbConn.Model(&model.SystemConfig{}).
|
||||||
|
Where("key = ?", model.ConfigKeySiteName).
|
||||||
|
Update("value", "broadcast_value").Error; err != nil {
|
||||||
|
t.Fatalf("Update(site_name) error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Invalidate: publishes to Redis and refreshes locally/other nodes
|
||||||
if err := repository.InvalidateSystemConfigCache(ctx, model.ConfigKeySiteName); err != nil {
|
if err := repository.InvalidateSystemConfigCache(ctx, model.ConfigKeySiteName); err != nil {
|
||||||
t.Fatalf("InvalidateSystemConfigCache(site_name) error = %v", err)
|
t.Fatalf("InvalidateSystemConfigCache(site_name) error = %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
_, err = db.Redis.HGet(ctx, db.PrefixedKey(repository.SystemConfigRedisHashKey), model.ConfigKeySiteName).Result()
|
// Wait for Redis Pub/Sub delivery in test
|
||||||
if !errors.Is(err, redis.Nil) {
|
time.Sleep(100 * time.Millisecond)
|
||||||
t.Fatalf("HGet(site_name) error = %v, want redis.Nil", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := dbConn.Model(&model.SystemConfig{}).
|
|
||||||
Where("key = ?", model.ConfigKeySiteName).
|
|
||||||
Update("value", "after_invalidate").Error; err != nil {
|
|
||||||
t.Fatalf("Update(site_name) error = %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
refreshed, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeySiteName)
|
refreshed, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeySiteName)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("GetSystemConfigByKey(site_name) refreshed error = %v", err)
|
t.Fatalf("GetSystemConfigByKey(site_name) refreshed error = %v", err)
|
||||||
}
|
}
|
||||||
if refreshed.Value != "after_invalidate" {
|
if refreshed.Value != "broadcast_value" {
|
||||||
t.Fatalf("GetSystemConfigByKey(site_name).Value = %q, want %q", refreshed.Value, "after_invalidate")
|
t.Fatalf("GetSystemConfigByKey(site_name).Value = %q, want %q", refreshed.Value, "broadcast_value")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Verify Redis pub/sub channel received message
|
||||||
|
if db.Redis != nil {
|
||||||
|
// Just a sanity check: we can publish a new manual update and verify subscription triggers
|
||||||
|
// which was done implicitly above.
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
+148
-69
@@ -6,65 +6,35 @@ package oauth
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"strconv"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/Rain-kl/Wavelet/internal/db"
|
"github.com/Rain-kl/Wavelet/internal/db"
|
||||||
"github.com/Rain-kl/Wavelet/internal/model"
|
"github.com/Rain-kl/Wavelet/internal/model"
|
||||||
|
"github.com/Rain-kl/Wavelet/pkg/cache/ram"
|
||||||
)
|
)
|
||||||
|
|
||||||
type cacheEntry struct {
|
|
||||||
value any
|
|
||||||
expiredAt time.Time
|
|
||||||
}
|
|
||||||
|
|
||||||
type memoryCache struct {
|
|
||||||
sync.RWMutex
|
|
||||||
items map[string]cacheEntry
|
|
||||||
}
|
|
||||||
|
|
||||||
var localCache = &memoryCache{
|
|
||||||
items: make(map[string]cacheEntry),
|
|
||||||
}
|
|
||||||
|
|
||||||
func (c *memoryCache) Set(key string, val any, ttl time.Duration) {
|
|
||||||
c.Lock()
|
|
||||||
defer c.Unlock()
|
|
||||||
c.items[key] = cacheEntry{
|
|
||||||
value: val,
|
|
||||||
expiredAt: time.Now().Add(ttl),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (c *memoryCache) Get(key string) (any, bool) {
|
|
||||||
c.RLock()
|
|
||||||
item, ok := c.items[key]
|
|
||||||
if !ok {
|
|
||||||
c.RUnlock()
|
|
||||||
return nil, false
|
|
||||||
}
|
|
||||||
if time.Now().After(item.expiredAt) {
|
|
||||||
c.RUnlock()
|
|
||||||
c.Lock()
|
|
||||||
if item, ok = c.items[key]; ok && time.Now().After(item.expiredAt) {
|
|
||||||
delete(c.items, key)
|
|
||||||
}
|
|
||||||
c.Unlock()
|
|
||||||
return nil, false
|
|
||||||
}
|
|
||||||
c.RUnlock()
|
|
||||||
return item.value, true
|
|
||||||
}
|
|
||||||
|
|
||||||
func (c *memoryCache) Delete(key string) {
|
|
||||||
c.Lock()
|
|
||||||
defer c.Unlock()
|
|
||||||
delete(c.items, key)
|
|
||||||
}
|
|
||||||
|
|
||||||
const (
|
const (
|
||||||
tokenCacheTTL = 5 * time.Minute
|
tokenCacheTTL = 5 * time.Minute
|
||||||
userCacheTTL = 5 * time.Minute
|
userCacheTTL = 5 * time.Minute
|
||||||
|
|
||||||
|
//nolint:gosec // This is a Redis Pub/Sub channel name, not a credential
|
||||||
|
oauthTokenInvalidationChannel = "oauth:token_invalidation"
|
||||||
|
oauthUserInvalidationChannel = "oauth:user_invalidation"
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
tokenRAM = ram.MustNew[string, *model.AccessToken](ram.Options{MaximumSize: 2048})
|
||||||
|
userRAM = ram.MustNew[uint64, *model.User](ram.Options{MaximumSize: 2048})
|
||||||
|
|
||||||
|
tokenListenerOnce sync.Once
|
||||||
|
tokenListenerCtx context.Context
|
||||||
|
tokenListenerCancel context.CancelFunc
|
||||||
|
|
||||||
|
userListenerOnce sync.Once
|
||||||
|
userListenerCtx context.Context
|
||||||
|
userListenerCancel context.CancelFunc
|
||||||
)
|
)
|
||||||
|
|
||||||
func tokenCacheKey(tokenHash string) string {
|
func tokenCacheKey(tokenHash string) string {
|
||||||
@@ -75,20 +45,98 @@ func userCacheKey(userID uint64) string {
|
|||||||
return fmt.Sprintf("oauth:user:%d", userID)
|
return fmt.Sprintf("oauth:user:%d", userID)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func ensureTokenCacheListener() {
|
||||||
|
if db.Redis == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
tokenListenerOnce.Do(startTokenCacheInvalidationListener)
|
||||||
|
}
|
||||||
|
|
||||||
|
func startTokenCacheInvalidationListener() {
|
||||||
|
tokenListenerCtx, tokenListenerCancel = context.WithCancel(context.Background())
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
pubsub := db.Redis.Subscribe(tokenListenerCtx, oauthTokenInvalidationChannel)
|
||||||
|
defer func() {
|
||||||
|
_ = pubsub.Close()
|
||||||
|
}()
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
<-tokenListenerCtx.Done()
|
||||||
|
_ = pubsub.Close()
|
||||||
|
}()
|
||||||
|
|
||||||
|
for msg := range pubsub.Channel() {
|
||||||
|
tokenHash := msg.Payload
|
||||||
|
if tokenHash == "" || tokenHash == "*" || tokenHash == "reset" {
|
||||||
|
tokenRAM.InvalidateAll()
|
||||||
|
} else {
|
||||||
|
tokenRAM.Invalidate(tokenHash)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
|
func publishTokenRAMInvalidation(ctx context.Context, tokenHash string) {
|
||||||
|
if db.Redis == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
_ = db.Redis.Publish(ctx, oauthTokenInvalidationChannel, tokenHash).Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
func ensureUserCacheListener() {
|
||||||
|
if db.Redis == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
userListenerOnce.Do(startUserCacheInvalidationListener)
|
||||||
|
}
|
||||||
|
|
||||||
|
func startUserCacheInvalidationListener() {
|
||||||
|
userListenerCtx, userListenerCancel = context.WithCancel(context.Background())
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
pubsub := db.Redis.Subscribe(userListenerCtx, oauthUserInvalidationChannel)
|
||||||
|
defer func() {
|
||||||
|
_ = pubsub.Close()
|
||||||
|
}()
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
<-userListenerCtx.Done()
|
||||||
|
_ = pubsub.Close()
|
||||||
|
}()
|
||||||
|
|
||||||
|
for msg := range pubsub.Channel() {
|
||||||
|
userIDStr := msg.Payload
|
||||||
|
if userIDStr == "" || userIDStr == "*" || userIDStr == "reset" {
|
||||||
|
userRAM.InvalidateAll()
|
||||||
|
} else if userID, err := strconv.ParseUint(userIDStr, 10, 64); err == nil {
|
||||||
|
userRAM.Invalidate(userID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
|
func publishUserRAMInvalidation(ctx context.Context, userID uint64) {
|
||||||
|
if db.Redis == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
_ = db.Redis.Publish(ctx, oauthUserInvalidationChannel, strconv.FormatUint(userID, 10)).Err()
|
||||||
|
}
|
||||||
|
|
||||||
// GetCachedToken 获取缓存的 AccessToken
|
// GetCachedToken 获取缓存的 AccessToken
|
||||||
func GetCachedToken(ctx context.Context, tokenHash string) (*model.AccessToken, error) {
|
func GetCachedToken(ctx context.Context, tokenHash string) (*model.AccessToken, error) {
|
||||||
key := tokenCacheKey(tokenHash)
|
ensureTokenCacheListener()
|
||||||
if val, ok := localCache.Get(key); ok {
|
|
||||||
if token, ok := val.(*model.AccessToken); ok {
|
if val, ok := tokenRAM.GetIfPresent(tokenHash); ok {
|
||||||
return token, nil
|
return val, nil
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if db.Redis != nil {
|
if db.Redis != nil {
|
||||||
var token model.AccessToken
|
var token model.AccessToken
|
||||||
|
key := tokenCacheKey(tokenHash)
|
||||||
if err := db.GetJSON(ctx, key, &token); err == nil {
|
if err := db.GetJSON(ctx, key, &token); err == nil {
|
||||||
// Write back to local cache
|
// Write back to local cache
|
||||||
localCache.Set(key, &token, tokenCacheTTL)
|
tokenRAM.Set(tokenHash, &token)
|
||||||
return &token, nil
|
return &token, nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -97,36 +145,41 @@ func GetCachedToken(ctx context.Context, tokenHash string) (*model.AccessToken,
|
|||||||
|
|
||||||
// SetCachedToken 设置 AccessToken 缓存
|
// SetCachedToken 设置 AccessToken 缓存
|
||||||
func SetCachedToken(ctx context.Context, tokenHash string, token *model.AccessToken) {
|
func SetCachedToken(ctx context.Context, tokenHash string, token *model.AccessToken) {
|
||||||
key := tokenCacheKey(tokenHash)
|
ensureTokenCacheListener()
|
||||||
localCache.Set(key, token, tokenCacheTTL)
|
|
||||||
|
tokenRAM.Set(tokenHash, token)
|
||||||
if db.Redis != nil {
|
if db.Redis != nil {
|
||||||
|
key := tokenCacheKey(tokenHash)
|
||||||
_ = db.SetJSON(ctx, key, token, tokenCacheTTL)
|
_ = db.SetJSON(ctx, key, token, tokenCacheTTL)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// InvalidateCachedToken 吊销/删除 token 缓存
|
// InvalidateCachedToken 吊销/删除 token 缓存
|
||||||
func InvalidateCachedToken(ctx context.Context, tokenHash string) {
|
func InvalidateCachedToken(ctx context.Context, tokenHash string) {
|
||||||
key := tokenCacheKey(tokenHash)
|
ensureTokenCacheListener()
|
||||||
localCache.Delete(key)
|
|
||||||
|
tokenRAM.Invalidate(tokenHash)
|
||||||
if db.Redis != nil {
|
if db.Redis != nil {
|
||||||
|
key := tokenCacheKey(tokenHash)
|
||||||
_ = db.Redis.Del(ctx, db.PrefixedKey(key)).Err()
|
_ = db.Redis.Del(ctx, db.PrefixedKey(key)).Err()
|
||||||
|
publishTokenRAMInvalidation(ctx, tokenHash)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetCachedUser 获取缓存的 User
|
// GetCachedUser 获取缓存的 User
|
||||||
func GetCachedUser(ctx context.Context, userID uint64) (*model.User, error) {
|
func GetCachedUser(ctx context.Context, userID uint64) (*model.User, error) {
|
||||||
key := userCacheKey(userID)
|
ensureUserCacheListener()
|
||||||
if val, ok := localCache.Get(key); ok {
|
|
||||||
if u, ok := val.(*model.User); ok {
|
if val, ok := userRAM.GetIfPresent(userID); ok {
|
||||||
return u, nil
|
return val, nil
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if db.Redis != nil {
|
if db.Redis != nil {
|
||||||
var u model.User
|
var u model.User
|
||||||
|
key := userCacheKey(userID)
|
||||||
if err := db.GetJSON(ctx, key, &u); err == nil {
|
if err := db.GetJSON(ctx, key, &u); err == nil {
|
||||||
// Write back to local cache
|
// Write back to local cache
|
||||||
localCache.Set(key, &u, userCacheTTL)
|
userRAM.Set(userID, &u)
|
||||||
return &u, nil
|
return &u, nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -135,18 +188,44 @@ func GetCachedUser(ctx context.Context, userID uint64) (*model.User, error) {
|
|||||||
|
|
||||||
// SetCachedUser 设置 User 缓存
|
// SetCachedUser 设置 User 缓存
|
||||||
func SetCachedUser(ctx context.Context, userID uint64, u *model.User) {
|
func SetCachedUser(ctx context.Context, userID uint64, u *model.User) {
|
||||||
key := userCacheKey(userID)
|
ensureUserCacheListener()
|
||||||
localCache.Set(key, u, userCacheTTL)
|
|
||||||
|
userRAM.Set(userID, u)
|
||||||
if db.Redis != nil {
|
if db.Redis != nil {
|
||||||
|
key := userCacheKey(userID)
|
||||||
_ = db.SetJSON(ctx, key, u, userCacheTTL)
|
_ = db.SetJSON(ctx, key, u, userCacheTTL)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// InvalidateCachedUser 吊销/失效 User 缓存
|
// InvalidateCachedUser 吊销/失效 User 缓存
|
||||||
func InvalidateCachedUser(ctx context.Context, userID uint64) {
|
func InvalidateCachedUser(ctx context.Context, userID uint64) {
|
||||||
key := userCacheKey(userID)
|
ensureUserCacheListener()
|
||||||
localCache.Delete(key)
|
|
||||||
|
userRAM.Invalidate(userID)
|
||||||
if db.Redis != nil {
|
if db.Redis != nil {
|
||||||
|
key := userCacheKey(userID)
|
||||||
_ = db.Redis.Del(ctx, db.PrefixedKey(key)).Err()
|
_ = db.Redis.Del(ctx, db.PrefixedKey(key)).Err()
|
||||||
|
publishUserRAMInvalidation(ctx, userID)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// StopOauthCacheListener stops both token and user Redis Pub/Sub subscription listeners and resets the sync.Once guards.
|
||||||
|
func StopOauthCacheListener() {
|
||||||
|
if tokenListenerCancel != nil {
|
||||||
|
tokenListenerCancel()
|
||||||
|
tokenListenerCancel = nil
|
||||||
|
}
|
||||||
|
tokenListenerOnce = sync.Once{}
|
||||||
|
|
||||||
|
if userListenerCancel != nil {
|
||||||
|
userListenerCancel()
|
||||||
|
userListenerCancel = nil
|
||||||
|
}
|
||||||
|
userListenerOnce = sync.Once{}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ResetOauthRAMCacheForTest clears only the process-local RAM cache.
|
||||||
|
func ResetOauthRAMCacheForTest() {
|
||||||
|
tokenRAM.InvalidateAll()
|
||||||
|
userRAM.InvalidateAll()
|
||||||
|
}
|
||||||
|
|||||||
@@ -0,0 +1,235 @@
|
|||||||
|
// Copyright 2026 Arctel.net
|
||||||
|
// SPDX-License-Identifier: Apache-2.0
|
||||||
|
|
||||||
|
package oauth
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/alicebob/miniredis/v2"
|
||||||
|
"github.com/redis/go-redis/v9"
|
||||||
|
"github.com/redis/go-redis/v9/maintnotifications"
|
||||||
|
|
||||||
|
"github.com/Rain-kl/Wavelet/internal/db"
|
||||||
|
"github.com/Rain-kl/Wavelet/internal/model"
|
||||||
|
)
|
||||||
|
|
||||||
|
func setupOauthCacheTest(t *testing.T) (*miniredis.Miniredis, func()) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
miniRedis, err := miniredis.Run()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to start miniredis: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
db.Redis = redis.NewClient(&redis.Options{
|
||||||
|
Addr: miniRedis.Addr(),
|
||||||
|
MaintNotificationsConfig: &maintnotifications.Config{
|
||||||
|
Mode: maintnotifications.ModeDisabled,
|
||||||
|
},
|
||||||
|
})
|
||||||
|
|
||||||
|
ResetOauthRAMCacheForTest()
|
||||||
|
|
||||||
|
cleanup := func() {
|
||||||
|
StopOauthCacheListener()
|
||||||
|
ResetOauthRAMCacheForTest()
|
||||||
|
db.Redis.Close()
|
||||||
|
miniRedis.Close()
|
||||||
|
db.Redis = nil
|
||||||
|
}
|
||||||
|
return miniRedis, cleanup
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestTokenCache_GetSetInvalidate(t *testing.T) {
|
||||||
|
_, cleanup := setupOauthCacheTest(t)
|
||||||
|
defer cleanup()
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
tokenHash := "test-token-hash"
|
||||||
|
token := &model.AccessToken{
|
||||||
|
ID: 123,
|
||||||
|
UserID: 456,
|
||||||
|
TokenHash: tokenHash,
|
||||||
|
Name: "test-token",
|
||||||
|
}
|
||||||
|
|
||||||
|
// 1. Get from empty cache -> miss
|
||||||
|
_, err := GetCachedToken(ctx, tokenHash)
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("expected cache miss for un-cached token")
|
||||||
|
}
|
||||||
|
|
||||||
|
// 2. Set to cache
|
||||||
|
SetCachedToken(ctx, tokenHash, token)
|
||||||
|
|
||||||
|
// 3. Get from cache -> hit
|
||||||
|
cached, err := GetCachedToken(ctx, tokenHash)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetCachedToken() failed: %v", err)
|
||||||
|
}
|
||||||
|
if cached.ID != token.ID || cached.UserID != token.UserID {
|
||||||
|
t.Fatalf("expected cached token %+v, got %+v", token, cached)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 4. Invalidate cache
|
||||||
|
InvalidateCachedToken(ctx, tokenHash)
|
||||||
|
|
||||||
|
// 5. Get from cache -> miss
|
||||||
|
_, err = GetCachedToken(ctx, tokenHash)
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("expected cache miss after invalidation")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUserCache_GetSetInvalidate(t *testing.T) {
|
||||||
|
_, cleanup := setupOauthCacheTest(t)
|
||||||
|
defer cleanup()
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
userID := uint64(789)
|
||||||
|
user := &model.User{
|
||||||
|
ID: userID,
|
||||||
|
Username: "testuser",
|
||||||
|
Email: "test@example.com",
|
||||||
|
}
|
||||||
|
|
||||||
|
// 1. Get from empty cache -> miss
|
||||||
|
_, err := GetCachedUser(ctx, userID)
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("expected cache miss for un-cached user")
|
||||||
|
}
|
||||||
|
|
||||||
|
// 2. Set to cache
|
||||||
|
SetCachedUser(ctx, userID, user)
|
||||||
|
|
||||||
|
// 3. Get from cache -> hit
|
||||||
|
cached, err := GetCachedUser(ctx, userID)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetCachedUser() failed: %v", err)
|
||||||
|
}
|
||||||
|
if cached.ID != user.ID || cached.Username != user.Username {
|
||||||
|
t.Fatalf("expected cached user %+v, got %+v", user, cached)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 4. Invalidate cache
|
||||||
|
InvalidateCachedUser(ctx, userID)
|
||||||
|
|
||||||
|
// 5. Get from cache -> miss
|
||||||
|
_, err = GetCachedUser(ctx, userID)
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("expected cache miss after invalidation")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestOauthCache_PubSubInvalidation(t *testing.T) {
|
||||||
|
_, cleanup := setupOauthCacheTest(t)
|
||||||
|
defer cleanup()
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
tokenHash := "pubsub-token-hash"
|
||||||
|
token := &model.AccessToken{
|
||||||
|
ID: 111,
|
||||||
|
UserID: 222,
|
||||||
|
TokenHash: tokenHash,
|
||||||
|
}
|
||||||
|
|
||||||
|
userID := uint64(333)
|
||||||
|
user := &model.User{
|
||||||
|
ID: userID,
|
||||||
|
Username: "pubsubuser",
|
||||||
|
}
|
||||||
|
|
||||||
|
// Set caches so they are stored in RAM
|
||||||
|
SetCachedToken(ctx, tokenHash, token)
|
||||||
|
SetCachedUser(ctx, userID, user)
|
||||||
|
|
||||||
|
// Verify they are cached
|
||||||
|
if _, ok := tokenRAM.GetIfPresent(tokenHash); !ok {
|
||||||
|
t.Fatal("expected token to be in RAM cache")
|
||||||
|
}
|
||||||
|
if _, ok := userRAM.GetIfPresent(userID); !ok {
|
||||||
|
t.Fatal("expected user to be in RAM cache")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Give Pub/Sub subscription time to establish
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
|
||||||
|
// Publish invalidation messages directly to simulate peer node invalidation
|
||||||
|
if err := db.Redis.Publish(ctx, oauthTokenInvalidationChannel, tokenHash).Err(); err != nil {
|
||||||
|
t.Fatalf("publish token invalidation error: %v", err)
|
||||||
|
}
|
||||||
|
if err := db.Redis.Publish(ctx, oauthUserInvalidationChannel, "333").Err(); err != nil {
|
||||||
|
t.Fatalf("publish user invalidation error: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Wait for background pubsub handlers to process messages
|
||||||
|
deadline := time.Now().Add(500 * time.Millisecond)
|
||||||
|
for time.Now().Before(deadline) {
|
||||||
|
_, tokenOk := tokenRAM.GetIfPresent(tokenHash)
|
||||||
|
_, userOk := userRAM.GetIfPresent(userID)
|
||||||
|
if !tokenOk && !userOk {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
time.Sleep(10 * time.Millisecond)
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, ok := tokenRAM.GetIfPresent(tokenHash); ok {
|
||||||
|
t.Fatal("expected token RAM cache to be invalidated by Pub/Sub")
|
||||||
|
}
|
||||||
|
if _, ok := userRAM.GetIfPresent(userID); ok {
|
||||||
|
t.Fatal("expected user RAM cache to be invalidated by Pub/Sub")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestOauthCache_PubSubResetAll(t *testing.T) {
|
||||||
|
_, cleanup := setupOauthCacheTest(t)
|
||||||
|
defer cleanup()
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
tokenHash := "reset-token-hash"
|
||||||
|
token := &model.AccessToken{
|
||||||
|
ID: 444,
|
||||||
|
UserID: 555,
|
||||||
|
TokenHash: tokenHash,
|
||||||
|
}
|
||||||
|
|
||||||
|
userID := uint64(666)
|
||||||
|
user := &model.User{
|
||||||
|
ID: userID,
|
||||||
|
Username: "resetuser",
|
||||||
|
}
|
||||||
|
|
||||||
|
SetCachedToken(ctx, tokenHash, token)
|
||||||
|
SetCachedUser(ctx, userID, user)
|
||||||
|
|
||||||
|
// Give Pub/Sub subscription time to establish
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
|
||||||
|
// Publish dynamic reset/wildcard to clear all
|
||||||
|
if err := db.Redis.Publish(ctx, oauthTokenInvalidationChannel, "*").Err(); err != nil {
|
||||||
|
t.Fatalf("publish token reset error: %v", err)
|
||||||
|
}
|
||||||
|
if err := db.Redis.Publish(ctx, oauthUserInvalidationChannel, "reset").Err(); err != nil {
|
||||||
|
t.Fatalf("publish user reset error: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
deadline := time.Now().Add(500 * time.Millisecond)
|
||||||
|
for time.Now().Before(deadline) {
|
||||||
|
_, tokenOk := tokenRAM.GetIfPresent(tokenHash)
|
||||||
|
_, userOk := userRAM.GetIfPresent(userID)
|
||||||
|
if !tokenOk && !userOk {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
time.Sleep(10 * time.Millisecond)
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, ok := tokenRAM.GetIfPresent(tokenHash); ok {
|
||||||
|
t.Fatal("expected token RAM cache to be fully cleared by '*'")
|
||||||
|
}
|
||||||
|
if _, ok := userRAM.GetIfPresent(userID); ok {
|
||||||
|
t.Fatal("expected user RAM cache to be fully cleared by 'reset'")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -13,7 +13,9 @@ import (
|
|||||||
"github.com/Rain-kl/Wavelet/internal/apps/admin/push/custom_events"
|
"github.com/Rain-kl/Wavelet/internal/apps/admin/push/custom_events"
|
||||||
"github.com/Rain-kl/Wavelet/internal/apps/risk_control"
|
"github.com/Rain-kl/Wavelet/internal/apps/risk_control"
|
||||||
"github.com/Rain-kl/Wavelet/internal/lifecycle"
|
"github.com/Rain-kl/Wavelet/internal/lifecycle"
|
||||||
|
"github.com/Rain-kl/Wavelet/internal/repository"
|
||||||
taskhandlers "github.com/Rain-kl/Wavelet/internal/task/handlers"
|
taskhandlers "github.com/Rain-kl/Wavelet/internal/task/handlers"
|
||||||
|
"github.com/Rain-kl/Wavelet/pkg/cache/ram"
|
||||||
"github.com/Rain-kl/Wavelet/pkg/logger"
|
"github.com/Rain-kl/Wavelet/pkg/logger"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -23,13 +25,59 @@ type Options struct {
|
|||||||
API bool
|
API bool
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// CacheRegistry holds settings for a registered cache type.
|
||||||
|
type CacheRegistry struct {
|
||||||
|
Loader ram.Loader
|
||||||
|
}
|
||||||
|
|
||||||
var (
|
var (
|
||||||
registerTasksOnce sync.Once
|
registerTasksOnce sync.Once
|
||||||
registerPushDomainEventsOnce sync.Once
|
registerPushDomainEventsOnce sync.Once
|
||||||
registerTaskListenersOnce sync.Once
|
registerTaskListenersOnce sync.Once
|
||||||
initRuntimeOnce sync.Once
|
initRuntimeOnce sync.Once
|
||||||
|
|
||||||
|
cacheRegistries = make(map[string]CacheRegistry)
|
||||||
|
cacheRegistriesMu sync.RWMutex
|
||||||
|
|
||||||
|
refreshLocks = make(map[string]*sync.Mutex)
|
||||||
|
refreshLocksMu sync.Mutex
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// RegisterCache registers a cache type with its Loader for unified preheating and refreshing.
|
||||||
|
func RegisterCache(configType string, reg CacheRegistry) {
|
||||||
|
cacheRegistriesMu.Lock()
|
||||||
|
defer cacheRegistriesMu.Unlock()
|
||||||
|
cacheRegistries[configType] = reg
|
||||||
|
}
|
||||||
|
|
||||||
|
func getRefreshLock(configType string) *sync.Mutex {
|
||||||
|
refreshLocksMu.Lock()
|
||||||
|
defer refreshLocksMu.Unlock()
|
||||||
|
lock, found := refreshLocks[configType]
|
||||||
|
if !found {
|
||||||
|
lock = &sync.Mutex{}
|
||||||
|
refreshLocks[configType] = lock
|
||||||
|
}
|
||||||
|
return lock
|
||||||
|
}
|
||||||
|
|
||||||
|
// PreheatAllCaches preheats all registered caches.
|
||||||
|
func PreheatAllCaches(ctx context.Context) error {
|
||||||
|
cacheRegistriesMu.RLock()
|
||||||
|
defer cacheRegistriesMu.RUnlock()
|
||||||
|
|
||||||
|
for configType, reg := range cacheRegistries {
|
||||||
|
lock := getRefreshLock(configType)
|
||||||
|
lock.Lock()
|
||||||
|
err := ram.Refresh(ctx, configType, "", reg.Loader)
|
||||||
|
lock.Unlock()
|
||||||
|
if err != nil {
|
||||||
|
logger.ErrorF(ctx, "[Bootstrap] preheating cache type %s failed: %v", configType, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// RegisterTasks registers all built-in task handlers and metadata.
|
// RegisterTasks registers all built-in task handlers and metadata.
|
||||||
func RegisterTasks() {
|
func RegisterTasks() {
|
||||||
registerTasksOnce.Do(func() {
|
registerTasksOnce.Do(func() {
|
||||||
@@ -79,6 +127,16 @@ func RegisterAll() {
|
|||||||
// Call from cmd entry points after wiring registration and database migration, not from router.
|
// Call from cmd entry points after wiring registration and database migration, not from router.
|
||||||
func Init(ctx context.Context, opts Options) {
|
func Init(ctx context.Context, opts Options) {
|
||||||
initRuntimeOnce.Do(func() {
|
initRuntimeOnce.Do(func() {
|
||||||
|
// Register config cache loader
|
||||||
|
RegisterCache(repository.ConfigCacheType, CacheRegistry{
|
||||||
|
Loader: repository.ConfigLoader{},
|
||||||
|
})
|
||||||
|
|
||||||
|
// Preheat config cache initially (using PreheatAllCaches)
|
||||||
|
if err := PreheatAllCaches(ctx); err != nil {
|
||||||
|
logger.ErrorF(ctx, "[Bootstrap] preheating all caches failed: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
if err := admin_push.SyncEvents(ctx); err != nil {
|
if err := admin_push.SyncEvents(ctx); err != nil {
|
||||||
logger.ErrorF(ctx, "[Bootstrap] sync push events failed: %v", err)
|
logger.ErrorF(ctx, "[Bootstrap] sync push events failed: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -42,20 +42,6 @@ func gooseDialect() string {
|
|||||||
return dialectPostgres
|
return dialectPostgres
|
||||||
}
|
}
|
||||||
|
|
||||||
func tableExistsSQL(dialect string) string {
|
|
||||||
if dialect == dialectSqlite {
|
|
||||||
return "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?"
|
|
||||||
}
|
|
||||||
return "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'public' AND table_name = $1"
|
|
||||||
}
|
|
||||||
|
|
||||||
func tablesWithPrefixSQL(dialect string) string {
|
|
||||||
if dialect == dialectSqlite {
|
|
||||||
return "SELECT name FROM sqlite_master WHERE type='table' AND name LIKE ?"
|
|
||||||
}
|
|
||||||
return "SELECT table_name FROM information_schema.tables WHERE table_schema='public' AND table_name LIKE $1"
|
|
||||||
}
|
|
||||||
|
|
||||||
func migrationDir() string {
|
func migrationDir() string {
|
||||||
if !config.Config.Database.Enabled {
|
if !config.Config.Database.Enabled {
|
||||||
return "goose/sqlite"
|
return "goose/sqlite"
|
||||||
|
|||||||
@@ -11,14 +11,15 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"strconv"
|
"strconv"
|
||||||
|
|
||||||
"github.com/redis/go-redis/v9"
|
|
||||||
"github.com/shopspring/decimal"
|
"github.com/shopspring/decimal"
|
||||||
|
|
||||||
"github.com/Rain-kl/Wavelet/internal/db"
|
"github.com/Rain-kl/Wavelet/internal/db"
|
||||||
"github.com/Rain-kl/Wavelet/internal/model"
|
"github.com/Rain-kl/Wavelet/internal/model"
|
||||||
|
"github.com/Rain-kl/Wavelet/pkg/cache/ram"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
|
configTypeSystem = "system"
|
||||||
errDatabaseNotInitialized = "database not initialized"
|
errDatabaseNotInitialized = "database not initialized"
|
||||||
errConfigIntParseFailed = "配置 %s 的值 '%s' 无法转换为整数: %w"
|
errConfigIntParseFailed = "配置 %s 的值 '%s' 无法转换为整数: %w"
|
||||||
errConfigDecimalParseFailed = "配置 %s 的值 '%s' 无法转换为decimal: %w"
|
errConfigDecimalParseFailed = "配置 %s 的值 '%s' 无法转换为decimal: %w"
|
||||||
@@ -26,21 +27,44 @@ const (
|
|||||||
errParseMenuDisplayConfigFailed = "解析目录显示配置失败: %w"
|
errParseMenuDisplayConfigFailed = "解析目录显示配置失败: %w"
|
||||||
)
|
)
|
||||||
|
|
||||||
// GetSystemConfigByKey 通过 key 查询配置(带 RAM + Redis 缓存)。
|
// PreheatSystemConfigs loads all system configs from database.
|
||||||
func GetSystemConfigByKey(ctx context.Context, key string) (model.SystemConfig, error) {
|
// This function strictly performs database read and does not perform any cache read or write operations.
|
||||||
ensureSystemConfigCacheListener()
|
func PreheatSystemConfigs(ctx context.Context) ([]model.SystemConfig, error) {
|
||||||
|
database := db.DB(ctx)
|
||||||
|
if database == nil {
|
||||||
|
return nil, errors.New(errDatabaseNotInitialized)
|
||||||
|
}
|
||||||
|
|
||||||
if cached, ok := systemConfigRAMCache.GetIfPresent(key); ok {
|
var configs []model.SystemConfig
|
||||||
return cloneSystemConfig(cached), nil
|
if err := database.Find(&configs).Error; err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return configs, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// PreheatSystemConfigByKey loads a single config key from database.
|
||||||
|
// This function strictly performs database read and does not perform any cache read or write operations.
|
||||||
|
func PreheatSystemConfigByKey(ctx context.Context, key string) (model.SystemConfig, error) {
|
||||||
|
database := db.DB(ctx)
|
||||||
|
if database == nil {
|
||||||
|
return model.SystemConfig{}, errors.New(errDatabaseNotInitialized)
|
||||||
}
|
}
|
||||||
|
|
||||||
var sc model.SystemConfig
|
var sc model.SystemConfig
|
||||||
if db.Redis != nil {
|
if err := database.Where("key = ?", key).First(&sc).Error; err != nil {
|
||||||
if err := db.HGetJSON(ctx, SystemConfigRedisHashKey, key, &sc); err == nil {
|
return model.SystemConfig{}, err
|
||||||
systemConfigRAMCache.Set(key, cloneSystemConfig(sc))
|
}
|
||||||
|
return sc, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetSystemConfigByGroup queries a configuration by Type and Key.
|
||||||
|
func GetSystemConfigByGroup(ctx context.Context, configType string, key string) (model.SystemConfig, error) {
|
||||||
|
ensureSystemConfigCacheListener()
|
||||||
|
|
||||||
|
if item, ok := ram.Get(configType, key); ok {
|
||||||
|
var sc model.SystemConfig
|
||||||
|
if err := json.Unmarshal([]byte(item.Value), &sc); err == nil {
|
||||||
return sc, nil
|
return sc, nil
|
||||||
} else if !errors.Is(err, redis.Nil) {
|
|
||||||
return model.SystemConfig{}, err
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -49,15 +73,31 @@ func GetSystemConfigByKey(ctx context.Context, key string) (model.SystemConfig,
|
|||||||
return model.SystemConfig{}, errors.New(errDatabaseNotInitialized)
|
return model.SystemConfig{}, errors.New(errDatabaseNotInitialized)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var sc model.SystemConfig
|
||||||
if err := database.Where("key = ?", key).First(&sc).Error; err != nil {
|
if err := database.Where("key = ?", key).First(&sc).Error; err != nil {
|
||||||
return model.SystemConfig{}, err
|
return model.SystemConfig{}, err
|
||||||
}
|
}
|
||||||
|
|
||||||
populateSystemConfigCache(ctx, sc)
|
// Populate local cache directly on query miss
|
||||||
|
valBytes, err := json.Marshal(sc)
|
||||||
|
if err == nil {
|
||||||
|
ram.Set(ram.CacheItem{
|
||||||
|
Key: sc.Key,
|
||||||
|
Value: string(valBytes),
|
||||||
|
Type: configType,
|
||||||
|
TTL: determineTTL(sc.Key),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
return sc, nil
|
return sc, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// ListSystemConfigsByKeys loads multiple config keys in one database round trip.
|
// GetSystemConfigByKey queries config by key (delegates to Type "config").
|
||||||
|
func GetSystemConfigByKey(ctx context.Context, key string) (model.SystemConfig, error) {
|
||||||
|
return GetSystemConfigByGroup(ctx, ConfigCacheType, key)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ListSystemConfigsByKeys loads multiple config keys.
|
||||||
func ListSystemConfigsByKeys(ctx context.Context, keys []string) (map[string]model.SystemConfig, error) {
|
func ListSystemConfigsByKeys(ctx context.Context, keys []string) (map[string]model.SystemConfig, error) {
|
||||||
if len(keys) == 0 {
|
if len(keys) == 0 {
|
||||||
return map[string]model.SystemConfig{}, nil
|
return map[string]model.SystemConfig{}, nil
|
||||||
@@ -67,28 +107,16 @@ func ListSystemConfigsByKeys(ctx context.Context, keys []string) (map[string]mod
|
|||||||
|
|
||||||
result := make(map[string]model.SystemConfig, len(keys))
|
result := make(map[string]model.SystemConfig, len(keys))
|
||||||
missing := make([]string, 0, len(keys))
|
missing := make([]string, 0, len(keys))
|
||||||
for _, key := range keys {
|
|
||||||
if cached, ok := systemConfigRAMCache.GetIfPresent(key); ok {
|
|
||||||
result[key] = cloneSystemConfig(cached)
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
missing = append(missing, key)
|
|
||||||
}
|
|
||||||
|
|
||||||
if len(missing) > 0 && db.Redis != nil {
|
for _, key := range keys {
|
||||||
stillMissing := make([]string, 0, len(missing))
|
if item, ok := ram.Get(ConfigCacheType, key); ok {
|
||||||
for _, key := range missing {
|
|
||||||
var sc model.SystemConfig
|
var sc model.SystemConfig
|
||||||
if err := db.HGetJSON(ctx, SystemConfigRedisHashKey, key, &sc); err == nil {
|
if err := json.Unmarshal([]byte(item.Value), &sc); err == nil {
|
||||||
systemConfigRAMCache.Set(key, cloneSystemConfig(sc))
|
|
||||||
result[key] = sc
|
result[key] = sc
|
||||||
continue
|
continue
|
||||||
} else if !errors.Is(err, redis.Nil) {
|
|
||||||
return nil, err
|
|
||||||
}
|
}
|
||||||
stillMissing = append(stillMissing, key)
|
|
||||||
}
|
}
|
||||||
missing = stillMissing
|
missing = append(missing, key)
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(missing) == 0 {
|
if len(missing) == 0 {
|
||||||
@@ -106,8 +134,16 @@ func ListSystemConfigsByKeys(ctx context.Context, keys []string) (map[string]mod
|
|||||||
}
|
}
|
||||||
|
|
||||||
for i := range configs {
|
for i := range configs {
|
||||||
populateSystemConfigCache(ctx, configs[i])
|
valBytes, err := json.Marshal(configs[i])
|
||||||
result[configs[i].Key] = cloneSystemConfig(configs[i])
|
if err == nil {
|
||||||
|
ram.Set(ram.CacheItem{
|
||||||
|
Key: configs[i].Key,
|
||||||
|
Value: string(valBytes),
|
||||||
|
Type: ConfigCacheType,
|
||||||
|
TTL: determineTTL(configs[i].Key),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
result[configs[i].Key] = configs[i]
|
||||||
}
|
}
|
||||||
|
|
||||||
return result, nil
|
return result, nil
|
||||||
@@ -115,21 +151,25 @@ func ListSystemConfigsByKeys(ctx context.Context, keys []string) (map[string]mod
|
|||||||
|
|
||||||
// InvalidateVisibleSystemConfigsCache clears the cached public config list.
|
// InvalidateVisibleSystemConfigsCache clears the cached public config list.
|
||||||
func InvalidateVisibleSystemConfigsCache(ctx context.Context) error {
|
func InvalidateVisibleSystemConfigsCache(ctx context.Context) error {
|
||||||
if db.Redis == nil {
|
return InvalidateAllSystemConfigCaches(ctx)
|
||||||
return nil
|
|
||||||
}
|
|
||||||
return db.Redis.Del(ctx, db.PrefixedKey(SystemConfigVisibleListRedisKey)).Err()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// ListVisibleSystemConfigs 查询所有可通过公共配置接口暴露的配置(带 Redis 列表缓存)。
|
// ListVisibleSystemConfigs queries visible configs using local cache store.
|
||||||
func ListVisibleSystemConfigs(ctx context.Context) ([]model.SystemConfig, error) {
|
func ListVisibleSystemConfigs(ctx context.Context) ([]model.SystemConfig, error) {
|
||||||
if db.Redis != nil {
|
ensureSystemConfigCacheListener()
|
||||||
var cached []model.SystemConfig
|
|
||||||
if err := db.GetJSON(ctx, SystemConfigVisibleListRedisKey, &cached); err == nil {
|
items := ram.GetTypeItems(ConfigCacheType)
|
||||||
return cached, nil
|
if len(items) > 0 {
|
||||||
} else if !errors.Is(err, redis.Nil) {
|
var list []model.SystemConfig
|
||||||
return nil, err
|
for _, item := range items {
|
||||||
|
var sc model.SystemConfig
|
||||||
|
if err := json.Unmarshal([]byte(item.Value), &sc); err == nil {
|
||||||
|
if sc.Visibility == model.ConfigVisibilityVisible {
|
||||||
|
list = append(list, sc)
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
return list, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
database := db.DB(ctx)
|
database := db.DB(ctx)
|
||||||
@@ -142,14 +182,24 @@ func ListVisibleSystemConfigs(ctx context.Context) ([]model.SystemConfig, error)
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
if db.Redis != nil {
|
// Populate visible configs to local cache store
|
||||||
_ = db.SetJSON(ctx, SystemConfigVisibleListRedisKey, configs, 0)
|
for _, cfg := range configs {
|
||||||
|
valBytes, err := json.Marshal(cfg)
|
||||||
|
if err == nil {
|
||||||
|
ram.Set(ram.CacheItem{
|
||||||
|
Key: cfg.Key,
|
||||||
|
Value: string(valBytes),
|
||||||
|
Type: ConfigCacheType,
|
||||||
|
TTL: determineTTL(cfg.Key),
|
||||||
|
})
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return configs, nil
|
return configs, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetIntByKey 通过 key 查询配置并转换为 int 类型。
|
|
||||||
|
// GetIntByKey queries config and converts to int.
|
||||||
func GetIntByKey(ctx context.Context, key string) (int, error) {
|
func GetIntByKey(ctx context.Context, key string) (int, error) {
|
||||||
sc, err := GetSystemConfigByKey(ctx, key)
|
sc, err := GetSystemConfigByKey(ctx, key)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -164,7 +214,7 @@ func GetIntByKey(ctx context.Context, key string) (int, error) {
|
|||||||
return value, nil
|
return value, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetDecimalByKey 通过 key 查询配置并转换为 decimal.Decimal 类型。
|
// GetDecimalByKey queries config and converts to decimal.Decimal.
|
||||||
func GetDecimalByKey(ctx context.Context, key string, precision int32) (decimal.Decimal, error) {
|
func GetDecimalByKey(ctx context.Context, key string, precision int32) (decimal.Decimal, error) {
|
||||||
sc, err := GetSystemConfigByKey(ctx, key)
|
sc, err := GetSystemConfigByKey(ctx, key)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -179,7 +229,7 @@ func GetDecimalByKey(ctx context.Context, key string, precision int32) (decimal.
|
|||||||
return value.Truncate(precision), nil
|
return value.Truncate(precision), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetBoolByKey 通过 key 查询配置并转换为 bool 类型。
|
// GetBoolByKey queries config and converts to bool.
|
||||||
func GetBoolByKey(ctx context.Context, key string) (bool, error) {
|
func GetBoolByKey(ctx context.Context, key string) (bool, error) {
|
||||||
sc, err := GetSystemConfigByKey(ctx, key)
|
sc, err := GetSystemConfigByKey(ctx, key)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -194,7 +244,7 @@ func GetBoolByKey(ctx context.Context, key string) (bool, error) {
|
|||||||
return value, nil
|
return value, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetMenuDisplayConfig 获取目录显示配置,解析为 map[string]bool。
|
// GetMenuDisplayConfig queries and parses menu config.
|
||||||
func GetMenuDisplayConfig(ctx context.Context) (map[string]bool, error) {
|
func GetMenuDisplayConfig(ctx context.Context) (map[string]bool, error) {
|
||||||
sc, err := GetSystemConfigByKey(ctx, model.ConfigKeyMenuDisplayConfig)
|
sc, err := GetSystemConfigByKey(ctx, model.ConfigKeyMenuDisplayConfig)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -69,7 +69,7 @@ func SaveOrUpdateSystemConfig(ctx context.Context, key, value string) error {
|
|||||||
sc = model.SystemConfig{
|
sc = model.SystemConfig{
|
||||||
Key: key,
|
Key: key,
|
||||||
Value: value,
|
Value: value,
|
||||||
Type: "system",
|
Type: configTypeSystem,
|
||||||
Visibility: model.ConfigVisibilityHidden,
|
Visibility: model.ConfigVisibilityHidden,
|
||||||
}
|
}
|
||||||
if err := db.DB(ctx).Create(&sc).Error; err != nil {
|
if err := db.DB(ctx).Create(&sc).Error; err != nil {
|
||||||
|
|||||||
@@ -6,31 +6,87 @@ package repository
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"sync"
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gorm.io/gorm"
|
||||||
|
|
||||||
"github.com/Rain-kl/Wavelet/internal/db"
|
"github.com/Rain-kl/Wavelet/internal/db"
|
||||||
"github.com/Rain-kl/Wavelet/internal/model"
|
|
||||||
"github.com/Rain-kl/Wavelet/pkg/cache/ram"
|
"github.com/Rain-kl/Wavelet/pkg/cache/ram"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
// SystemConfigInvalidationChannel broadcasts RAM cache eviction across nodes.
|
// SystemConfigBroadcastChannel broadcasts system config cache updates across nodes.
|
||||||
SystemConfigInvalidationChannel = "system:config_invalidation"
|
SystemConfigBroadcastChannel = "system:config_broadcast"
|
||||||
// SystemConfigRedisHashKey Redis Hash key,存储所有系统配置。
|
|
||||||
|
// SystemConfigInvalidationChannel is kept as an alias for backward compatibility.
|
||||||
|
SystemConfigInvalidationChannel = SystemConfigBroadcastChannel
|
||||||
|
|
||||||
|
// SystemConfigRedisHashKey is kept for backward compatibility in tests.
|
||||||
SystemConfigRedisHashKey = "system:system_configs"
|
SystemConfigRedisHashKey = "system:system_configs"
|
||||||
// SystemConfigVisibleListRedisKey Redis key,缓存所有 visibility=1 的公共配置列表。
|
// SystemConfigVisibleListRedisKey is kept for backward compatibility in tests.
|
||||||
SystemConfigVisibleListRedisKey = "system:visible_configs"
|
SystemConfigVisibleListRedisKey = "system:visible_configs"
|
||||||
|
|
||||||
systemConfigInvalidateAllToken = "*"
|
// ConfigCacheType is the cache type for all system configs.
|
||||||
systemConfigRAMMaximumSize = 512
|
ConfigCacheType = "config"
|
||||||
)
|
)
|
||||||
|
|
||||||
type systemConfigInvalidationMessage struct {
|
type systemConfigBroadcastMessage struct {
|
||||||
Key string `json:"key"`
|
Type string `json:"type"`
|
||||||
|
Key string `json:"key"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// ConfigLoader loads configuration data from the database.
|
||||||
|
type ConfigLoader struct{}
|
||||||
|
|
||||||
|
// LoadAll loads all system configs from database as CacheItems.
|
||||||
|
func (ConfigLoader) LoadAll(ctx context.Context, configType string) ([]ram.CacheItem, error) {
|
||||||
|
configs, err := PreheatSystemConfigs(ctx)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
items := make([]ram.CacheItem, len(configs))
|
||||||
|
for i, cfg := range configs {
|
||||||
|
valBytes, err := json.Marshal(cfg)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
items[i] = ram.CacheItem{
|
||||||
|
Key: cfg.Key,
|
||||||
|
Value: string(valBytes),
|
||||||
|
Type: configType,
|
||||||
|
TTL: determineTTL(cfg.Key),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return items, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// LoadOne loads a single system config from database as a CacheItem.
|
||||||
|
func (ConfigLoader) LoadOne(ctx context.Context, configType string, key string) (ram.CacheItem, error) {
|
||||||
|
cfg, err := PreheatSystemConfigByKey(ctx, key)
|
||||||
|
if err != nil {
|
||||||
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||||
|
return ram.CacheItem{}, ram.ErrNotFound
|
||||||
|
}
|
||||||
|
return ram.CacheItem{}, err
|
||||||
|
}
|
||||||
|
|
||||||
|
valBytes, err := json.Marshal(cfg)
|
||||||
|
if err != nil {
|
||||||
|
return ram.CacheItem{}, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return ram.CacheItem{
|
||||||
|
Key: cfg.Key,
|
||||||
|
Value: string(valBytes),
|
||||||
|
Type: configType,
|
||||||
|
TTL: determineTTL(cfg.Key),
|
||||||
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
var (
|
var (
|
||||||
systemConfigRAMCache = ram.MustNew[string, model.SystemConfig](ram.Options{MaximumSize: systemConfigRAMMaximumSize})
|
|
||||||
systemConfigListenerOnce sync.Once
|
systemConfigListenerOnce sync.Once
|
||||||
systemConfigListenerCtx context.Context
|
systemConfigListenerCtx context.Context
|
||||||
systemConfigListenerCancel context.CancelFunc
|
systemConfigListenerCancel context.CancelFunc
|
||||||
@@ -48,7 +104,7 @@ func startSystemConfigCacheInvalidationListener() {
|
|||||||
systemConfigListenerCtx, systemConfigListenerCancel = context.WithCancel(context.Background())
|
systemConfigListenerCtx, systemConfigListenerCancel = context.WithCancel(context.Background())
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
pubsub := db.Redis.Subscribe(systemConfigListenerCtx, SystemConfigInvalidationChannel)
|
pubsub := db.Redis.Subscribe(systemConfigListenerCtx, SystemConfigBroadcastChannel)
|
||||||
defer func() {
|
defer func() {
|
||||||
_ = pubsub.Close()
|
_ = pubsub.Close()
|
||||||
}()
|
}()
|
||||||
@@ -59,16 +115,18 @@ func startSystemConfigCacheInvalidationListener() {
|
|||||||
}()
|
}()
|
||||||
|
|
||||||
for msg := range pubsub.Channel() {
|
for msg := range pubsub.Channel() {
|
||||||
var payload systemConfigInvalidationMessage
|
var payload systemConfigBroadcastMessage
|
||||||
if err := json.Unmarshal([]byte(msg.Payload), &payload); err != nil {
|
if err := json.Unmarshal([]byte(msg.Payload), &payload); err != nil {
|
||||||
systemConfigRAMCache.InvalidateAll()
|
ram.UpdateTypeItems(ConfigCacheType, nil)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if payload.Key == "" || payload.Key == systemConfigInvalidateAllToken {
|
|
||||||
systemConfigRAMCache.InvalidateAll()
|
key := payload.Key
|
||||||
continue
|
if key == "*" || key == "" {
|
||||||
|
ram.UpdateTypeItems(payload.Type, nil)
|
||||||
|
} else {
|
||||||
|
ram.Delete(payload.Type, key)
|
||||||
}
|
}
|
||||||
systemConfigRAMCache.Invalidate(payload.Key)
|
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
@@ -82,57 +140,53 @@ func StopSystemConfigCacheListener() {
|
|||||||
systemConfigListenerOnce = sync.Once{}
|
systemConfigListenerOnce = sync.Once{}
|
||||||
}
|
}
|
||||||
|
|
||||||
func cloneSystemConfig(sc model.SystemConfig) model.SystemConfig {
|
func determineTTL(_ string) time.Duration {
|
||||||
return sc
|
// Program-determined TTL: -1 means never expire for all configs by default
|
||||||
|
return -1
|
||||||
}
|
}
|
||||||
|
|
||||||
func populateSystemConfigCache(ctx context.Context, sc model.SystemConfig) {
|
// InvalidateSystemConfigCache triggers a broadcast to refresh the cache for key.
|
||||||
systemConfigRAMCache.Set(sc.Key, cloneSystemConfig(sc))
|
|
||||||
if db.Redis != nil {
|
|
||||||
_ = db.HSetJSON(ctx, SystemConfigRedisHashKey, sc.Key, &sc)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func publishSystemConfigRAMInvalidation(ctx context.Context, key string) {
|
|
||||||
if db.Redis == nil {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
payload, err := json.Marshal(systemConfigInvalidationMessage{Key: key})
|
|
||||||
if err != nil {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
_ = db.Redis.Publish(ctx, SystemConfigInvalidationChannel, payload).Err()
|
|
||||||
}
|
|
||||||
|
|
||||||
// InvalidateSystemConfigCache evicts one config key from local RAM and Redis.
|
|
||||||
func InvalidateSystemConfigCache(ctx context.Context, key string) error {
|
func InvalidateSystemConfigCache(ctx context.Context, key string) error {
|
||||||
ensureSystemConfigCacheListener()
|
ensureSystemConfigCacheListener()
|
||||||
|
|
||||||
systemConfigRAMCache.Invalidate(key)
|
// Invalidate local cache synchronously first
|
||||||
|
ram.Delete(ConfigCacheType, key)
|
||||||
|
|
||||||
|
// Broadcast to other nodes and clean legacy Redis cache key
|
||||||
if db.Redis != nil {
|
if db.Redis != nil {
|
||||||
if err := db.HDel(ctx, SystemConfigRedisHashKey, key); err != nil {
|
_ = db.HDel(ctx, SystemConfigRedisHashKey, key)
|
||||||
return err
|
publishSystemConfigBroadcast(ctx, ConfigCacheType, key)
|
||||||
}
|
|
||||||
}
|
}
|
||||||
publishSystemConfigRAMInvalidation(ctx, key)
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// InvalidateAllSystemConfigCaches evicts all config entries from local RAM and Redis.
|
// InvalidateAllSystemConfigCaches triggers a broadcast to refresh the entire config cache.
|
||||||
func InvalidateAllSystemConfigCaches(ctx context.Context) error {
|
func InvalidateAllSystemConfigCaches(ctx context.Context) error {
|
||||||
ensureSystemConfigCacheListener()
|
ensureSystemConfigCacheListener()
|
||||||
|
|
||||||
systemConfigRAMCache.InvalidateAll()
|
// Invalidate all items of type ConfigCacheType synchronously first
|
||||||
|
ram.UpdateTypeItems(ConfigCacheType, nil)
|
||||||
|
|
||||||
|
// Broadcast to other nodes and clean legacy Redis cache keys
|
||||||
if db.Redis != nil {
|
if db.Redis != nil {
|
||||||
if err := db.Redis.Del(ctx, db.PrefixedKey(SystemConfigRedisHashKey)).Err(); err != nil {
|
_ = db.Redis.Del(ctx, db.PrefixedKey(SystemConfigRedisHashKey), db.PrefixedKey(SystemConfigVisibleListRedisKey)).Err()
|
||||||
return err
|
publishSystemConfigBroadcast(ctx, ConfigCacheType, "*")
|
||||||
}
|
|
||||||
}
|
}
|
||||||
publishSystemConfigRAMInvalidation(ctx, systemConfigInvalidateAllToken)
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func publishSystemConfigBroadcast(ctx context.Context, configType string, key string) {
|
||||||
|
if db.Redis == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
payload, err := json.Marshal(systemConfigBroadcastMessage{Type: configType, Key: key})
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
_ = db.Redis.Publish(ctx, SystemConfigBroadcastChannel, payload).Err()
|
||||||
|
}
|
||||||
|
|
||||||
// ResetSystemConfigRAMCacheForTest clears only the process-local RAM cache.
|
// ResetSystemConfigRAMCacheForTest clears only the process-local RAM cache.
|
||||||
func ResetSystemConfigRAMCacheForTest() {
|
func ResetSystemConfigRAMCacheForTest() {
|
||||||
systemConfigRAMCache.InvalidateAll()
|
ram.ResetForTest()
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -6,14 +6,16 @@ package repository
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/Rain-kl/Wavelet/internal/db"
|
|
||||||
"github.com/Rain-kl/Wavelet/internal/model"
|
|
||||||
"github.com/alicebob/miniredis/v2"
|
"github.com/alicebob/miniredis/v2"
|
||||||
"github.com/glebarez/sqlite"
|
"github.com/glebarez/sqlite"
|
||||||
"github.com/redis/go-redis/v9"
|
"github.com/redis/go-redis/v9"
|
||||||
"github.com/redis/go-redis/v9/maintnotifications"
|
"github.com/redis/go-redis/v9/maintnotifications"
|
||||||
"gorm.io/gorm"
|
"gorm.io/gorm"
|
||||||
|
|
||||||
|
"github.com/Rain-kl/Wavelet/internal/db"
|
||||||
|
"github.com/Rain-kl/Wavelet/internal/model"
|
||||||
)
|
)
|
||||||
|
|
||||||
func setupSystemConfigTest(t *testing.T) (*gorm.DB, func()) {
|
func setupSystemConfigTest(t *testing.T) (*gorm.DB, func()) {
|
||||||
@@ -54,6 +56,7 @@ func setupSystemConfigTest(t *testing.T) (*gorm.DB, func()) {
|
|||||||
db.SetDB(sqliteDB)
|
db.SetDB(sqliteDB)
|
||||||
db.Redis = redisClient
|
db.Redis = redisClient
|
||||||
|
|
||||||
|
|
||||||
cleanup := func() {
|
cleanup := func() {
|
||||||
StopSystemConfigCacheListener()
|
StopSystemConfigCacheListener()
|
||||||
ResetSystemConfigRAMCacheForTest()
|
ResetSystemConfigRAMCacheForTest()
|
||||||
@@ -76,16 +79,14 @@ func TestListSystemConfigsByKeys_EmptyKeys(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestListSystemConfigsByKeys_LoadsFromRedisBeforeDB(t *testing.T) {
|
func TestListSystemConfigsByKeys_LoadsFromRAMCache(t *testing.T) {
|
||||||
dbConn, cleanup := setupSystemConfigTest(t)
|
dbConn, cleanup := setupSystemConfigTest(t)
|
||||||
defer cleanup()
|
defer cleanup()
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
|
|
||||||
ResetSystemConfigRAMCacheForTest()
|
ResetSystemConfigRAMCacheForTest()
|
||||||
if err := InvalidateAllSystemConfigCaches(ctx); err != nil {
|
|
||||||
t.Fatalf("InvalidateAllSystemConfigCaches() error = %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
|
// Initial load
|
||||||
warm, err := GetSystemConfigByKey(ctx, model.ConfigKeySiteName)
|
warm, err := GetSystemConfigByKey(ctx, model.ConfigKeySiteName)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("GetSystemConfigByKey(site_name) warm error = %v", err)
|
t.Fatalf("GetSystemConfigByKey(site_name) warm error = %v", err)
|
||||||
@@ -94,14 +95,14 @@ func TestListSystemConfigsByKeys_LoadsFromRedisBeforeDB(t *testing.T) {
|
|||||||
t.Fatalf("GetSystemConfigByKey(site_name).Value = %q, want %q", warm.Value, "Wavelet")
|
t.Fatalf("GetSystemConfigByKey(site_name).Value = %q, want %q", warm.Value, "Wavelet")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Update DB directly
|
||||||
if err := dbConn.Model(&model.SystemConfig{}).
|
if err := dbConn.Model(&model.SystemConfig{}).
|
||||||
Where("key = ?", model.ConfigKeySiteName).
|
Where("key = ?", model.ConfigKeySiteName).
|
||||||
Update("value", "db_only_value").Error; err != nil {
|
Update("value", "db_only_value").Error; err != nil {
|
||||||
t.Fatalf("Update(site_name) error = %v", err)
|
t.Fatalf("Update(site_name) error = %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
ResetSystemConfigRAMCacheForTest()
|
// Fetch via ListSystemConfigsByKeys should serve from local store (meaning the old value "Wavelet")
|
||||||
|
|
||||||
configs, err := ListSystemConfigsByKeys(ctx, []string{model.ConfigKeySiteName})
|
configs, err := ListSystemConfigsByKeys(ctx, []string{model.ConfigKeySiteName})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("ListSystemConfigsByKeys(site_name) error = %v", err)
|
t.Fatalf("ListSystemConfigsByKeys(site_name) error = %v", err)
|
||||||
@@ -112,35 +113,47 @@ func TestListSystemConfigsByKeys_LoadsFromRedisBeforeDB(t *testing.T) {
|
|||||||
t.Fatal("ListSystemConfigsByKeys(site_name) missing site_name entry")
|
t.Fatal("ListSystemConfigsByKeys(site_name) missing site_name entry")
|
||||||
}
|
}
|
||||||
if sc.Value != "Wavelet" {
|
if sc.Value != "Wavelet" {
|
||||||
t.Fatalf("ListSystemConfigsByKeys(site_name).Value = %q, want redis value %q", sc.Value, "Wavelet")
|
t.Fatalf("ListSystemConfigsByKeys(site_name).Value = %q, want cached value %q", sc.Value, "Wavelet")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestListSystemConfigsByKeys_PopulatesRAMFromRedis(t *testing.T) {
|
func TestGetSystemConfigByGroupAndInvalidation(t *testing.T) {
|
||||||
_, cleanup := setupSystemConfigTest(t)
|
dbConn, cleanup := setupSystemConfigTest(t)
|
||||||
defer cleanup()
|
defer cleanup()
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
|
|
||||||
ResetSystemConfigRAMCacheForTest()
|
ResetSystemConfigRAMCacheForTest()
|
||||||
if err := InvalidateAllSystemConfigCaches(ctx); err != nil {
|
|
||||||
t.Fatalf("InvalidateAllSystemConfigCaches() error = %v", err)
|
// Get via specific group/type
|
||||||
|
cfg, err := GetSystemConfigByGroup(ctx, ConfigCacheType, model.ConfigKeySiteName)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetSystemConfigByGroup error = %v", err)
|
||||||
|
}
|
||||||
|
if cfg.Value != "Wavelet" {
|
||||||
|
t.Fatalf("value = %q, want %q", cfg.Value, "Wavelet")
|
||||||
}
|
}
|
||||||
|
|
||||||
if _, err := GetSystemConfigByKey(ctx, model.ConfigKeySiteName); err != nil {
|
// Direct DB update
|
||||||
t.Fatalf("GetSystemConfigByKey(site_name) warm error = %v", err)
|
if err := dbConn.Model(&model.SystemConfig{}).
|
||||||
|
Where("key = ?", model.ConfigKeySiteName).
|
||||||
|
Update("value", "new_site_name").Error; err != nil {
|
||||||
|
t.Fatalf("DB Update error = %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
ResetSystemConfigRAMCacheForTest()
|
// Invalidate
|
||||||
|
if err := InvalidateSystemConfigCache(ctx, model.ConfigKeySiteName); err != nil {
|
||||||
if _, err := ListSystemConfigsByKeys(ctx, []string{model.ConfigKeySiteName}); err != nil {
|
t.Fatalf("InvalidateSystemConfigCache error = %v", err)
|
||||||
t.Fatalf("ListSystemConfigsByKeys(site_name) error = %v", err)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
cached, ok := systemConfigRAMCache.GetIfPresent(model.ConfigKeySiteName)
|
// Wait for broadcast execution
|
||||||
if !ok {
|
time.Sleep(100 * time.Millisecond)
|
||||||
t.Fatal("expected RAM cache to be populated after redis hit")
|
|
||||||
|
// Fetch again
|
||||||
|
updated, err := GetSystemConfigByKey(ctx, model.ConfigKeySiteName)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetSystemConfigByKey error = %v", err)
|
||||||
}
|
}
|
||||||
if cached.Value != "Wavelet" {
|
if updated.Value != "new_site_name" {
|
||||||
t.Fatalf("RAM cache value = %q, want %q", cached.Value, "Wavelet")
|
t.Fatalf("value = %q, want %q", updated.Value, "new_site_name")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -32,12 +32,12 @@ func GetUserByUsername(ctx context.Context, username string) (model.User, error)
|
|||||||
// GetSystemUser loads the built-in system user, or returns a synthetic fallback.
|
// GetSystemUser loads the built-in system user, or returns a synthetic fallback.
|
||||||
func GetSystemUser(ctx context.Context) model.User {
|
func GetSystemUser(ctx context.Context) model.User {
|
||||||
var user model.User
|
var user model.User
|
||||||
if err := db.DB(ctx).Where("username = ?", "system").First(&user).Error; err == nil {
|
if err := db.DB(ctx).Where("username = ?", configTypeSystem).First(&user).Error; err == nil {
|
||||||
return user
|
return user
|
||||||
}
|
}
|
||||||
return model.User{
|
return model.User{
|
||||||
ID: 999,
|
ID: 999,
|
||||||
Username: "system",
|
Username: configTypeSystem,
|
||||||
Nickname: "系统",
|
Nickname: "系统",
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -36,6 +36,11 @@ func SetupTestEnvironment(t *testing.T) (*gorm.DB, *miniredis.Miniredis, func())
|
|||||||
t.Fatalf("failed to open in-memory SQLite db: %v", err)
|
t.Fatalf("failed to open in-memory SQLite db: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Limit to 1 open connection for SQLite :memory: to keep the database in one shared connection
|
||||||
|
if sqlDB, err := sqliteDB.DB(); err == nil {
|
||||||
|
sqlDB.SetMaxOpenConns(1)
|
||||||
|
}
|
||||||
|
|
||||||
// AutoMigrate all tables
|
// AutoMigrate all tables
|
||||||
err = sqliteDB.AutoMigrate(
|
err = sqliteDB.AutoMigrate(
|
||||||
&model.User{},
|
&model.User{},
|
||||||
@@ -77,6 +82,8 @@ func SetupTestEnvironment(t *testing.T) (*gorm.DB, *miniredis.Miniredis, func())
|
|||||||
// Cleanup function
|
// Cleanup function
|
||||||
cleanup := func() {
|
cleanup := func() {
|
||||||
runExtraCleanups()
|
runExtraCleanups()
|
||||||
|
repository.StopSystemConfigCacheListener()
|
||||||
|
repository.StopAuthSourceCacheListener()
|
||||||
repository.ResetSystemConfigRAMCacheForTest()
|
repository.ResetSystemConfigRAMCacheForTest()
|
||||||
_ = redisClient.Close()
|
_ = redisClient.Close()
|
||||||
mr.Close()
|
mr.Close()
|
||||||
|
|||||||
Vendored
+233
@@ -0,0 +1,233 @@
|
|||||||
|
// Copyright 2026 Arctel.net
|
||||||
|
// SPDX-License-Identifier: Apache-2.0
|
||||||
|
|
||||||
|
package ram
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
// ErrNotFound is returned by the Loader when the requested item is not found.
|
||||||
|
ErrNotFound = errors.New("cache item not found in data source")
|
||||||
|
|
||||||
|
managerCache *Cache[string, map[string]cacheEntry]
|
||||||
|
|
||||||
|
writeLocks = make(map[string]*sync.Mutex)
|
||||||
|
writeLocksMu sync.Mutex
|
||||||
|
)
|
||||||
|
|
||||||
|
// CacheItem represents a unified cache entity.
|
||||||
|
type CacheItem struct {
|
||||||
|
Key string `json:"key"`
|
||||||
|
Value string `json:"value"`
|
||||||
|
Type string `json:"type"`
|
||||||
|
TTL time.Duration `json:"ttl"` // -1 means never expire
|
||||||
|
}
|
||||||
|
|
||||||
|
// Loader is an interface that the cache client must implement to handle database retrieval.
|
||||||
|
type Loader interface {
|
||||||
|
LoadAll(ctx context.Context, configType string) ([]CacheItem, error)
|
||||||
|
LoadOne(ctx context.Context, configType string, key string) (CacheItem, error)
|
||||||
|
}
|
||||||
|
|
||||||
|
type cacheEntry struct {
|
||||||
|
item CacheItem
|
||||||
|
expireAt time.Time
|
||||||
|
}
|
||||||
|
|
||||||
|
func init() {
|
||||||
|
// Initialize with a large maximum size since it only stores one entry per configType
|
||||||
|
managerCache = MustNew[string, map[string]cacheEntry](Options{
|
||||||
|
MaximumSize: 1000,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func getWriteLock(configType string) *sync.Mutex {
|
||||||
|
writeLocksMu.Lock()
|
||||||
|
defer writeLocksMu.Unlock()
|
||||||
|
lock, found := writeLocks[configType]
|
||||||
|
if !found {
|
||||||
|
lock = &sync.Mutex{}
|
||||||
|
writeLocks[configType] = lock
|
||||||
|
}
|
||||||
|
return lock
|
||||||
|
}
|
||||||
|
|
||||||
|
// Get retrieves a cache item from the local cache store, checking for expiration.
|
||||||
|
// Reads are completely lock-free because maps stored in Otter are immutable.
|
||||||
|
func Get(configType string, key string) (CacheItem, bool) {
|
||||||
|
m, ok := managerCache.GetIfPresent(configType)
|
||||||
|
if !ok {
|
||||||
|
return CacheItem{}, false
|
||||||
|
}
|
||||||
|
|
||||||
|
entry, found := m[key]
|
||||||
|
if !found {
|
||||||
|
return CacheItem{}, false
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check expiration
|
||||||
|
if entry.item.TTL != -1 && !entry.expireAt.IsZero() && time.Now().After(entry.expireAt) {
|
||||||
|
// Asynchronously remove the expired item from the map and write back
|
||||||
|
go deleteKeyIfExpired(configType, key, entry.expireAt)
|
||||||
|
return CacheItem{}, false
|
||||||
|
}
|
||||||
|
|
||||||
|
return entry.item, true
|
||||||
|
}
|
||||||
|
|
||||||
|
func deleteKeyIfExpired(configType string, key string, expireAt time.Time) {
|
||||||
|
lock := getWriteLock(configType)
|
||||||
|
lock.Lock()
|
||||||
|
defer lock.Unlock()
|
||||||
|
|
||||||
|
currentMap, ok := managerCache.GetIfPresent(configType)
|
||||||
|
if !ok {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
entry, found := currentMap[key]
|
||||||
|
if !found {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Double-check expiration time to ensure we don't delete a newly updated key
|
||||||
|
if entry.expireAt != expireAt || !time.Now().After(entry.expireAt) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
newMap := make(map[string]cacheEntry, len(currentMap)-1)
|
||||||
|
for k, v := range currentMap {
|
||||||
|
if k != key {
|
||||||
|
newMap[k] = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
managerCache.Set(configType, newMap)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Set stores a cache item in the local cache store.
|
||||||
|
// Writes are protected by a fine-grained lock per configType.
|
||||||
|
func Set(item CacheItem) {
|
||||||
|
lock := getWriteLock(item.Type)
|
||||||
|
lock.Lock()
|
||||||
|
defer lock.Unlock()
|
||||||
|
|
||||||
|
currentMap, ok := managerCache.GetIfPresent(item.Type)
|
||||||
|
newMap := make(map[string]cacheEntry)
|
||||||
|
if ok {
|
||||||
|
for k, v := range currentMap {
|
||||||
|
newMap[k] = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
var expireAt time.Time
|
||||||
|
if item.TTL != -1 {
|
||||||
|
expireAt = time.Now().Add(item.TTL)
|
||||||
|
}
|
||||||
|
|
||||||
|
newMap[item.Key] = cacheEntry{
|
||||||
|
item: item,
|
||||||
|
expireAt: expireAt,
|
||||||
|
}
|
||||||
|
managerCache.Set(item.Type, newMap)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Delete removes a single item from the local cache store.
|
||||||
|
// Writes are protected by a fine-grained lock per configType.
|
||||||
|
func Delete(configType string, key string) {
|
||||||
|
lock := getWriteLock(configType)
|
||||||
|
lock.Lock()
|
||||||
|
defer lock.Unlock()
|
||||||
|
|
||||||
|
currentMap, ok := managerCache.GetIfPresent(configType)
|
||||||
|
if !ok {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
newMap := make(map[string]cacheEntry, len(currentMap))
|
||||||
|
for k, v := range currentMap {
|
||||||
|
if k != key {
|
||||||
|
newMap[k] = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
managerCache.Set(configType, newMap)
|
||||||
|
}
|
||||||
|
|
||||||
|
// UpdateTypeItems replaces all cache items of a specific type atomically.
|
||||||
|
// Writes are protected by a fine-grained lock per configType.
|
||||||
|
func UpdateTypeItems(configType string, items []CacheItem) {
|
||||||
|
lock := getWriteLock(configType)
|
||||||
|
lock.Lock()
|
||||||
|
defer lock.Unlock()
|
||||||
|
|
||||||
|
newMap := make(map[string]cacheEntry, len(items))
|
||||||
|
for _, item := range items {
|
||||||
|
var expireAt time.Time
|
||||||
|
if item.TTL != -1 {
|
||||||
|
expireAt = time.Now().Add(item.TTL)
|
||||||
|
}
|
||||||
|
newMap[item.Key] = cacheEntry{
|
||||||
|
item: item,
|
||||||
|
expireAt: expireAt,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
managerCache.Set(configType, newMap)
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetTypeItems retrieves all unexpired cache items of a specific type.
|
||||||
|
// Reads are completely lock-free because maps stored in Otter are immutable.
|
||||||
|
func GetTypeItems(configType string) []CacheItem {
|
||||||
|
currentMap, ok := managerCache.GetIfPresent(configType)
|
||||||
|
if !ok {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
var list []CacheItem
|
||||||
|
for _, entry := range currentMap {
|
||||||
|
if entry.item.TTL == -1 || entry.expireAt.IsZero() || time.Now().Before(entry.expireAt) {
|
||||||
|
list = append(list, entry.item)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return list
|
||||||
|
}
|
||||||
|
|
||||||
|
// Refresh reloads configuration cache from database via the Loader.
|
||||||
|
func Refresh(ctx context.Context, configType string, key string, loader Loader) error {
|
||||||
|
if configType == "" {
|
||||||
|
return errors.New("type is required")
|
||||||
|
}
|
||||||
|
|
||||||
|
if key != "" {
|
||||||
|
// Single key refresh: first fetch latest value from database
|
||||||
|
item, err := loader.LoadOne(ctx, configType, key)
|
||||||
|
if err != nil {
|
||||||
|
if errors.Is(err, ErrNotFound) {
|
||||||
|
Delete(configType, key)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
Set(item)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// All keys refresh: load all of that type from database first, then replace cache
|
||||||
|
items, err := loader.LoadAll(ctx, configType)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
UpdateTypeItems(configType, items)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// ResetForTest clears the local store and locks.
|
||||||
|
func ResetForTest() {
|
||||||
|
writeLocksMu.Lock()
|
||||||
|
writeLocks = make(map[string]*sync.Mutex)
|
||||||
|
writeLocksMu.Unlock()
|
||||||
|
managerCache.InvalidateAll()
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user