1 Commits

3 changed files with 57 additions and 53 deletions

45
relay/cache/cache.go vendored
View File

@ -8,11 +8,13 @@ import (
type item struct { type item struct {
value interface{} value interface{}
expiration int64 expiration int64
staleTime int64
} }
type Cache struct { type Cache struct {
items map[string]item items map[string]item
mu sync.RWMutex mu sync.RWMutex
refreshing sync.Map
} }
func New() *Cache { func New() *Cache {
@ -27,35 +29,68 @@ func (c *Cache) Set(key string, value interface{}, duration time.Duration) {
c.mu.Lock() c.mu.Lock()
defer c.mu.Unlock() defer c.mu.Unlock()
now := time.Now().UnixNano()
c.items[key] = item{ c.items[key] = item{
value: value, 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) { func (c *Cache) Get(key string) (interface{}, bool) {
c.mu.RLock() c.mu.RLock()
defer c.mu.RUnlock()
item, exists := c.items[key] item, exists := c.items[key]
c.mu.RUnlock()
if !exists { if !exists {
return nil, false return nil, false
} }
if time.Now().UnixNano() > item.expiration { now := time.Now().UnixNano()
if now > item.staleTime {
return nil, false return nil, false
} }
return item.value, true 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() { func (c *Cache) startCleanup() {
ticker := time.NewTicker(time.Minute) ticker := time.NewTicker(time.Minute)
for range ticker.C { for range ticker.C {
c.mu.Lock() c.mu.Lock()
now := time.Now().UnixNano() now := time.Now().UnixNano()
for k, v := range c.items { for k, v := range c.items {
if now > v.expiration { if now > v.staleTime {
delete(c.items, k) delete(c.items, k)
} }
} }

View File

@ -48,10 +48,6 @@ func main() {
panic(err) 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.StoreEvent = append(relay.StoreEvent, db.SaveEvent)
relay.QueryEvents = append(relay.QueryEvents, db.QueryEvents) relay.QueryEvents = append(relay.QueryEvents, db.QueryEvents)
relay.DeleteEvent = append(relay.DeleteEvent, db.DeleteEvent) relay.DeleteEvent = append(relay.DeleteEvent, db.DeleteEvent)

View File

@ -3,7 +3,6 @@ package trending
import ( import (
"database/sql" "database/sql"
"encoding/json" "encoding/json"
"fmt"
"time" "time"
"git.highperfocused.tech/highperfocused/lumina-relay/relay/cache" "git.highperfocused.tech/highperfocused/lumina-relay/relay/cache"
@ -21,47 +20,10 @@ type Post struct {
var ( var (
trendingCache = cache.New() trendingCache = cache.New()
cacheDuration = 35 * time.Minute // Slightly longer than update interval cacheDuration = 5 * time.Minute
updateInterval = 30 * time.Minute
) )
// Initialize sets up the background updating of trending calculations func fetchTrendingKind20(db *sql.DB) ([]Post, error) {
// 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) {
query := ` query := `
WITH reactions AS ( WITH reactions AS (
SELECT SELECT
@ -109,6 +71,17 @@ func calculateTrendingKind20(db *sql.DB) ([]Post, error) {
trendingPosts = append(trendingPosts, post) trendingPosts = append(trendingPosts, post)
} }
trendingCache.Set("trending_kind_20", trendingPosts, cacheDuration)
return trendingPosts, nil 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
}