Compare commits
1 Commits
background
...
cache_impr
Author | SHA1 | Date | |
---|---|---|---|
3a9d4ce645 |
45
relay/cache/cache.go
vendored
45
relay/cache/cache.go
vendored
@ -8,11 +8,13 @@ import (
|
||||
type item struct {
|
||||
value interface{}
|
||||
expiration int64
|
||||
staleTime int64
|
||||
}
|
||||
|
||||
type Cache struct {
|
||||
items map[string]item
|
||||
mu sync.RWMutex
|
||||
refreshing sync.Map
|
||||
}
|
||||
|
||||
func New() *Cache {
|
||||
@ -27,35 +29,68 @@ func (c *Cache) Set(key string, value interface{}, duration time.Duration) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
now := time.Now().UnixNano()
|
||||
c.items[key] = item{
|
||||
value: value,
|
||||
expiration: time.Now().Add(duration).UnixNano(),
|
||||
expiration: now + duration.Nanoseconds(),
|
||||
staleTime: now + (duration * 2).Nanoseconds(), // Stale time is double the normal duration
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) Get(key string) (interface{}, bool) {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
|
||||
item, exists := c.items[key]
|
||||
c.mu.RUnlock()
|
||||
|
||||
if !exists {
|
||||
return nil, false
|
||||
}
|
||||
|
||||
if time.Now().UnixNano() > item.expiration {
|
||||
now := time.Now().UnixNano()
|
||||
if now > item.staleTime {
|
||||
return nil, false
|
||||
}
|
||||
|
||||
return item.value, true
|
||||
}
|
||||
|
||||
func (c *Cache) GetOrRefresh(key string, refreshFn func() (interface{}, error), duration time.Duration) (interface{}, bool) {
|
||||
// Try to get from cache first
|
||||
if value, exists := c.Get(key); exists {
|
||||
now := time.Now().UnixNano()
|
||||
item := c.items[key]
|
||||
|
||||
// If the value is expired but not stale, trigger refresh in background
|
||||
if now > item.expiration && now <= item.staleTime {
|
||||
if _, loading := c.refreshing.LoadOrStore(key, true); !loading {
|
||||
go func() {
|
||||
defer c.refreshing.Delete(key)
|
||||
if newValue, err := refreshFn(); err == nil {
|
||||
c.Set(key, newValue, duration)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
return value, true
|
||||
}
|
||||
|
||||
// If no value exists, do a blocking refresh
|
||||
value, err := refreshFn()
|
||||
if err != nil {
|
||||
return nil, false
|
||||
}
|
||||
|
||||
c.Set(key, value, duration)
|
||||
return value, true
|
||||
}
|
||||
|
||||
func (c *Cache) startCleanup() {
|
||||
ticker := time.NewTicker(time.Minute)
|
||||
for range ticker.C {
|
||||
c.mu.Lock()
|
||||
now := time.Now().UnixNano()
|
||||
for k, v := range c.items {
|
||||
if now > v.expiration {
|
||||
if now > v.staleTime {
|
||||
delete(c.items, k)
|
||||
}
|
||||
}
|
||||
|
@ -48,10 +48,6 @@ func main() {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
// Initialize trending system to start background calculations
|
||||
fmt.Println("Initializing trending system...")
|
||||
trending.Initialize(db.DB.DB)
|
||||
|
||||
relay.StoreEvent = append(relay.StoreEvent, db.SaveEvent)
|
||||
relay.QueryEvents = append(relay.QueryEvents, db.QueryEvents)
|
||||
relay.DeleteEvent = append(relay.DeleteEvent, db.DeleteEvent)
|
||||
|
@ -3,7 +3,6 @@ package trending
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"git.highperfocused.tech/highperfocused/lumina-relay/relay/cache"
|
||||
@ -21,47 +20,10 @@ type Post struct {
|
||||
|
||||
var (
|
||||
trendingCache = cache.New()
|
||||
cacheDuration = 35 * time.Minute // Slightly longer than update interval
|
||||
updateInterval = 30 * time.Minute
|
||||
cacheDuration = 5 * time.Minute
|
||||
)
|
||||
|
||||
// Initialize sets up the background updating of trending calculations
|
||||
// This should be called once at application startup
|
||||
func Initialize(db *sql.DB) {
|
||||
// Perform initial calculation
|
||||
if _, err := calculateTrendingKind20(db); err != nil {
|
||||
fmt.Printf("Error in initial trending calculation: %v\n", err)
|
||||
}
|
||||
|
||||
// Set up periodic updates
|
||||
go func() {
|
||||
ticker := time.NewTicker(updateInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for range ticker.C {
|
||||
if _, err := calculateTrendingKind20(db); err != nil {
|
||||
fmt.Printf("Error updating trending data: %v\n", err)
|
||||
} else {
|
||||
fmt.Printf("Successfully updated trending data at %s\n", time.Now().Format(time.RFC3339))
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// GetTrendingKind20 returns the top 20 trending posts of kind 20 from the last 24 hours
|
||||
// It returns cached results that are updated periodically in the background
|
||||
func GetTrendingKind20(db *sql.DB) ([]Post, error) {
|
||||
if cached, ok := trendingCache.Get("trending_kind_20"); ok {
|
||||
return cached.([]Post), nil
|
||||
}
|
||||
|
||||
// If cache is empty (which shouldn't happen with background updates),
|
||||
// calculate on demand
|
||||
return calculateTrendingKind20(db)
|
||||
}
|
||||
|
||||
// calculateTrendingKind20 performs the actual calculation and updates the cache
|
||||
func calculateTrendingKind20(db *sql.DB) ([]Post, error) {
|
||||
func fetchTrendingKind20(db *sql.DB) ([]Post, error) {
|
||||
query := `
|
||||
WITH reactions AS (
|
||||
SELECT
|
||||
@ -109,6 +71,17 @@ func calculateTrendingKind20(db *sql.DB) ([]Post, error) {
|
||||
trendingPosts = append(trendingPosts, post)
|
||||
}
|
||||
|
||||
trendingCache.Set("trending_kind_20", trendingPosts, cacheDuration)
|
||||
return trendingPosts, nil
|
||||
}
|
||||
|
||||
func GetTrendingKind20(db *sql.DB) ([]Post, error) {
|
||||
posts, exists := trendingCache.GetOrRefresh("trending_kind_20", func() (interface{}, error) {
|
||||
return fetchTrendingKind20(db)
|
||||
}, cacheDuration)
|
||||
|
||||
if !exists {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
return posts.([]Post), nil
|
||||
}
|
||||
|
Reference in New Issue
Block a user