aboutsummaryrefslogtreecommitdiff
path: root/backend/fetchers
diff options
context:
space:
mode:
Diffstat (limited to 'backend/fetchers')
-rw-r--r--backend/fetchers/app.bsky.actor.go12
-rw-r--r--backend/fetchers/app.bsky.feed.go48
-rw-r--r--backend/fetchers/cache.go184
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()
}
find me elsewhere · generated with cgit