reworked cache with redis + improved fetching

3 files changed, 170 insertions, 74 deletions

diff --git a/backend/fetchers/app.bsky.actor.go b/backend/fetchers/app.bsky.actor.go
index 0752223..a56fe0b 100644
--- a/backend/fetchers/app.bsky.actor.go
+++ b/backend/fetchers/app.bsky.actor.go
@@ -11,6 +11,7 @@ import (
 	"net"
 	"net/http"
 	"strings"
+	"time"
 )
 
 type DIDDocument struct {
@@ -136,12 +137,13 @@ func FetchAppBskyActorProfile(logger *slog.Logger, did, pdsURL, slingshotURL str
 	uri := fmt.Sprintf("at://%s/app.bsky.actor.profile/self", did)
 
 	// try slingshot first
-	if slingshotURL != "" {
+	if slingshotURL != "" && general.SlingshotAvailable() {
 		rawRecord, err := general.FetchRawRecordByURI(logger, uri, slingshotURL, session)
 		if err != nil {
 			logger.Warn("slingshot profile fetch failed", "did", did, "error", err)
+			general.SetSlingshotBackoff(5 * time.Minute)
 		} else {
-			return parseAppBskyActorProfile(did, rawRecord, cache)
+			return parseAppBskyActorProfile(did, pdsURL, rawRecord, cache)
 		}
 	}
 
@@ -151,7 +153,7 @@ func FetchAppBskyActorProfile(logger *slog.Logger, did, pdsURL, slingshotURL str
 		return nil, err
 	}
 
-	return parseAppBskyActorProfile(did, rawRecord, cache)
+	return parseAppBskyActorProfile(did, pdsURL, rawRecord, cache)
 }
 
 func FetchAppBskyActorProfiles(logger *slog.Logger, dids []string, pdsURL, slingshotURL string, session *auth.OAuthState, cache *Cache) map[string]notes.Author {
@@ -165,7 +167,7 @@ func FetchAppBskyActorProfiles(logger *slog.Logger, dids []string, pdsURL, sling
 	return authors
 }
 
-func parseAppBskyActorProfile(did string, rawRecord *general.RawRecord, cache *Cache) (*notes.Author, error) {
+func parseAppBskyActorProfile(did, pds string, rawRecord *general.RawRecord, cache *Cache) (*notes.Author, error) {
 	var profile notes.AppBskyActorProfileRecord
 	if err := json.Unmarshal(rawRecord.Value, &profile); err != nil {
 		return nil, err
@@ -176,7 +178,7 @@ func parseAppBskyActorProfile(did string, rawRecord *general.RawRecord, cache *C
 		handle = "handle.invalid"
 	}
 
-	author, err := notes.AuthorFromAppBskyActorProfileRecord(profile, did, handle)
+	author, err := notes.AuthorFromAppBskyActorProfileRecord(profile, did, handle, pds)
 
 	cache.SetAuthor(did, author)
 	return author, nil
diff --git a/backend/fetchers/app.bsky.feed.go b/backend/fetchers/app.bsky.feed.go
index 16e93c5..f490efb 100644
--- a/backend/fetchers/app.bsky.feed.go
+++ b/backend/fetchers/app.bsky.feed.go
@@ -7,11 +7,12 @@ import (
 	"llostre/backend/general"
 	"llostre/backend/notes"
 	"log/slog"
+	"time"
 )
 
 func FetchAppBskyFeedPosts(logger *slog.Logger, did, pdsURL, slingshotURL string, session *auth.OAuthState, cache *Cache) ([]notes.Note, error) {
 	// get author of the note
-	author, found := cache.GetProfile(did)
+	author, found := cache.GetAuthor(did)
 	var err error
 	if !found {
 		author, err = FetchAppBskyActorProfile(logger, did, pdsURL, slingshotURL, session, cache)
@@ -21,10 +22,11 @@ func FetchAppBskyFeedPosts(logger *slog.Logger, did, pdsURL, slingshotURL string
 	}
 
 	// try slingshot first if given
-	if slingshotURL != "" {
+	if slingshotURL != "" && general.SlingshotAvailable() {
 		records, err := general.FetchRawRecordCollection(logger, did, "app.bsky.feed.post", slingshotURL, session)
 		if err != nil {
-			logger.Error("Fetching from slingshot failed")
+			logger.Error("fetching from slingshot failed, falling back to PDS")
+			general.SetSlingshotBackoff(5 * time.Minute)
 		} else {
 			return parseAppBskyFeedPostCollection(logger, slingshotURL, session, records, author, cache)
 		}
@@ -38,6 +40,39 @@ func FetchAppBskyFeedPosts(logger *slog.Logger, did, pdsURL, slingshotURL string
 	return parseAppBskyFeedPostCollection(logger, slingshotURL, session, records, author, cache)
 }
 
+func FetchAppBskyFeedPostsWithCursor(logger *slog.Logger, did, pdsURL, slingshotURL string, session *auth.OAuthState, cache *Cache, cursor string, limit int) ([]notes.Note, error) {
+	author, found := cache.GetAuthor(did)
+	var err error
+	if !found {
+		author, err = FetchAppBskyActorProfile(logger, did, pdsURL, slingshotURL, session, cache)
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	if slingshotURL != "" && general.SlingshotAvailable() {
+		records, nextCursor, err := general.FetchRawRecordCollectionWithCursor(logger, did, "app.bsky.feed.post", slingshotURL, session, cursor, limit)
+		if err != nil {
+			logger.Error("fetching from slingshot failed, falling back to PDS")
+			general.SetSlingshotBackoff(5 * time.Minute)
+		} else {
+			notes, err := parseAppBskyFeedPostCollection(logger, slingshotURL, session, records, author, cache)
+			cache.SetCursor(did, nextCursor)
+			return notes, err
+		}
+	}
+
+	records, nextCursor, err := general.FetchRawRecordCollectionWithCursor(logger, did, "app.bsky.feed.post", pdsURL, session, cursor, limit)
+	if err != nil {
+		return nil, err
+	}
+
+	notes, err := parseAppBskyFeedPostCollection(logger, pdsURL, session, records, author, cache)
+
+	cache.SetCursor(did, nextCursor)
+	return notes, err
+}
+
 func GetAppBskyFeedPostByURI(logger *slog.Logger, uri, slingshotURL string, session *auth.OAuthState, cache *Cache) (*notes.Note, error) {
 	// check cache, return if it was cached without auth or if it was cached with auth and we have auth
 	logger.Debug("Getting app.bsky.feed.post", "uri", uri)
@@ -68,7 +103,7 @@ func FetchAppBskyFeedPostByURI(logger *slog.Logger, uri, slingshotURL string, se
 	logger.Debug("fetched origin PDS", "did", repo, "pdsURL", pdsURL)
 
 	// get author of the note
-	author, found := cache.GetProfile(repo)
+	author, found := cache.GetAuthor(repo)
 	if !found {
 		author, err = FetchAppBskyActorProfile(logger, repo, pdsURL, slingshotURL, session, cache)
 	}
@@ -77,10 +112,11 @@ func FetchAppBskyFeedPostByURI(logger *slog.Logger, uri, slingshotURL string, se
 	}
 
 	// try slingshot first if given
-	if slingshotURL != "" {
+	if slingshotURL != "" && general.SlingshotAvailable() {
 		rawRecord, err := general.FetchRawRecordByURI(logger, uri, slingshotURL, session)
 		if err != nil {
 			logger.Error("fetching from slingshot failed", "error", err)
+			general.SetSlingshotBackoff(5 * time.Minute)
 		} else {
 			return parseAppBskyFeedPostRecord(rawRecord, author, cache)
 		}
@@ -157,7 +193,7 @@ func FetchParents(logger *slog.Logger, replyToURI string, slingshotURL string, s
 		}
 
 		// preemptively fetch posts from author to avoid hammering servers if there's a back and forth
-		if _, found := cache.GetProfile(note.AuthorDID); !found {
+		if _, found := cache.GetAuthor(note.AuthorDID); !found {
 			logger.Debug("caching posts from thread participant", "did", note.AuthorDID)
 			authorPDS, err := FetchPDS(note.AuthorDID)
 			if err != nil {
diff --git a/backend/fetchers/cache.go b/backend/fetchers/cache.go
index 01c37fe..26be070 100644
--- a/backend/fetchers/cache.go
+++ b/backend/fetchers/cache.go
@@ -1,109 +1,167 @@
 package fetchers
 
 import (
-	"llostre/backend/notes"
-	"sync"
+	"context"
+	"encoding/json"
+	"fmt"
+	"sort"
 	"time"
-)
 
-type CacheEntry struct {
-	Note      *notes.Note
-	ExpiresAt time.Time
-}
+	"llostre/backend/notes"
+
+	"github.com/redis/go-redis/v9"
+)
 
 type Cache struct {
-	posts    map[string]*CacheEntry
-	profiles map[string]*AuthorCacheEntry
-	mu       sync.RWMutex
+	client     *redis.Client
+	ctx        context.Context
+	expiryTime time.Duration
 }
 
-func NewCache() *Cache {
-	return &Cache{
-		posts:    make(map[string]*CacheEntry),
-		profiles: make(map[string]*AuthorCacheEntry),
+func NewCache(addr, password string, expiryMinutes int) (*Cache, error) {
+	client := redis.NewClient(&redis.Options{
+		Addr:     addr,
+		Password: password,
+		DB:       0,
+	})
+
+	ctx := context.Background()
+	if err := client.Ping(ctx).Err(); err != nil {
+		return nil, fmt.Errorf("redis connection failed: %w", err)
 	}
+
+	return &Cache{
+		client:     client,
+		ctx:        ctx,
+		expiryTime: time.Duration(expiryMinutes) * time.Minute,
+	}, nil
 }
 
 func (c *Cache) Get(uri string) (*notes.Note, bool) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
+	key := fmt.Sprintf("post:%s", uri)
+	data, err := c.client.Get(c.ctx, key).Bytes()
+	if err != nil {
+		return nil, false
+	}
+
+	var note notes.Note
+	if err := json.Unmarshal(data, &note); err != nil {
+		return nil, false
+	}
+
+	return &note, true
+}
 
-	entry, exists := c.posts[uri]
+func (c *Cache) Set(uri string, note *notes.Note) {
+	key := fmt.Sprintf("post:%s", uri)
+	data, err := json.Marshal(note)
+	if err != nil {
+		return
+	}
+
+	c.client.Set(c.ctx, key, data, c.expiryTime)
+}
 
-	if !exists {
+func (c *Cache) GetAuthor(did string) (*notes.Author, bool) {
+	key := fmt.Sprintf("profile:%s", did)
+	data, err := c.client.Get(c.ctx, key).Bytes()
+	if err != nil {
 		return nil, false
 	}
 
-	if time.Now().After(entry.ExpiresAt) {
-		delete(c.posts, uri)
+	var author notes.Author
+	if err := json.Unmarshal(data, &author); err != nil {
 		return nil, false
 	}
 
-	return entry.Note, true
+	return &author, true
 }
 
-func (c *Cache) Set(uri string, note *notes.Note) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
+func (c *Cache) SetAuthor(did string, author *notes.Author) {
+	key := fmt.Sprintf("profile:%s", did)
+	data, err := json.Marshal(author)
+	if err != nil {
+		return
+	}
+
+	c.client.Set(c.ctx, key, data, c.expiryTime)
+}
 
-	c.posts[uri] = &CacheEntry{
-		Note:      note,
-		ExpiresAt: time.Now().Add(15 * time.Minute),
+func (c *Cache) GetCursor(did string) (string, bool) {
+	key := fmt.Sprintf("cursor:%s", did)
+	cursor, err := c.client.Get(c.ctx, key).Result()
+	if err != nil {
+		return "", false
 	}
+	return cursor, true
 }
 
-func (c *Cache) GetMany(uris []string) map[string]*notes.Note {
-	c.mu.Lock()
-	defer c.mu.Unlock()
+func (c *Cache) SetCursor(did, cursor string) {
+	key := fmt.Sprintf("cursor:%s", did)
+	if cursor == "" {
+		c.client.Del(c.ctx, key)
+		return
+	}
+	c.client.Set(c.ctx, key, cursor, 24*time.Hour)
+}
 
-	result := make(map[string]*notes.Note)
-	now := time.Now()
+func (c *Cache) GetPostsByDID(did string) ([]notes.Note, error) {
+	pattern := fmt.Sprintf("post:at://%s/*", did)
+	var posts []notes.Note
 
-	for _, uri := range uris {
-		entry, exists := c.posts[uri]
-		if !exists {
+	iter := c.client.Scan(c.ctx, 0, pattern, 0).Iterator()
+	for iter.Next(c.ctx) {
+		data, err := c.client.Get(c.ctx, iter.Val()).Bytes()
+		if err != nil {
 			continue
 		}
-
-		if now.After(entry.ExpiresAt) {
-			delete(c.posts, uri)
+		var note notes.Note
+		if err := json.Unmarshal(data, &note); err != nil {
 			continue
 		}
+		posts = append(posts, note)
+	}
 
-		result[uri] = entry.Note
+	if err := iter.Err(); err != nil {
+		return nil, err
 	}
 
-	return result
+	sort.Slice(posts, func(i, j int) bool {
+		return posts[i].CreatedAt.After(posts[j].CreatedAt)
+	})
+
+	return posts, nil
 }
 
-type AuthorCacheEntry struct {
-	Author    *notes.Author
-	ExpiresAt time.Time
+// track oldest sent post per profile
+func (c *Cache) GetOldestSent(did string) (time.Time, bool) {
+	key := fmt.Sprintf("oldest_sent:%s", did)
+	ts, err := c.client.Get(c.ctx, key).Int64()
+	if err != nil {
+		return time.Time{}, false
+	}
+	return time.Unix(ts, 0), true
 }
 
-func (c *Cache) GetProfile(did string) (*notes.Author, bool) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
+func (c *Cache) SetOldestSent(did string, timestamp time.Time) {
+	key := fmt.Sprintf("oldest_sent:%s", did)
+	c.client.Set(c.ctx, key, timestamp.Unix(), 24*time.Hour)
+}
 
-	entry, exists := c.profiles[did]
-	if !exists {
-		return nil, false
+// count posts older than timestamp for a profile
+func (c *Cache) CountPostsOlderThan(did string, timestamp time.Time) (int, error) {
+	posts, err := c.GetPostsByDID(did)
+	if err != nil {
+		return 0, err
 	}
-
-	if time.Now().After(entry.ExpiresAt) {
-		delete(c.profiles, did)
-		return nil, false
+	for i, p := range posts {
+		if p.CreatedAt.Before(timestamp) {
+			return len(posts) - i, nil
+		}
 	}
-
-	return entry.Author, true
+	return 0, nil
 }
 
-func (c *Cache) SetAuthor(did string, author *notes.Author) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.profiles[did] = &AuthorCacheEntry{
-		Author:    author,
-		ExpiresAt: time.Now().Add(15 * time.Minute),
-	}
+func (c *Cache) Close() error {
+	return c.client.Close()
 }