From 53ede6765a1583538b2abc3142d858464b3cf01c Mon Sep 17 00:00:00 2001 From: Guillermo Serrahima Date: Thu, 16 Jul 2026 02:38:46 +0200 Subject: Implement fetched posts count tracking for catching gaps in cached posts --- fetchers/app.bsky.feed.go | 13 ++++++++++++- fetchers/cache.go | 25 +++++++++++++++++++++++++ fetchers/jetstream.go | 6 ++++++ main.go | 20 +++++++++++++++++--- 4 files changed, 60 insertions(+), 4 deletions(-) diff --git a/fetchers/app.bsky.feed.go b/fetchers/app.bsky.feed.go index a0eaeb2..232d020 100644 --- a/fetchers/app.bsky.feed.go +++ b/fetchers/app.bsky.feed.go @@ -57,6 +57,7 @@ func FetchAppBskyFeedPostsWithCursor(conf *Config, did, pdsURL string, session * } else { notes, err := parseAppBskyFeedPostCollection(conf, session, records, author) conf.Cache.SetCursor(did, nextCursor) + updatePostCount(conf, did, cursor, len(notes)) return notes, err } } @@ -69,9 +70,18 @@ func FetchAppBskyFeedPostsWithCursor(conf *Config, did, pdsURL string, session * notes, err := parseAppBskyFeedPostCollection(conf, session, records, author) conf.Cache.SetCursor(did, nextCursor) + updatePostCount(conf, did, cursor, len(notes)) return notes, err } +func updatePostCount(conf *Config, did, cursor string, fetchedCount int) { + if cursor == "" { + conf.Cache.SetPostCount(did, fetchedCount) + } else { + conf.Cache.AddPostCount(did, fetchedCount) + } +} + func GetAppBskyFeedPostByURI(conf *Config, uri string, session *auth.OAuthState) (*notes.Note, error) { // check cache, return if it was cached without auth or if it was cached with auth and we have auth conf.Logger.Debug("Getting app.bsky.feed.post", "uri", uri) @@ -81,7 +91,8 @@ func GetAppBskyFeedPostByURI(conf *Config, uri string, session *auth.OAuthState) return cached, nil } - // author requires auth - if we have a session, serve it + // author requires auth + // if we have a session, serve it // if the token was invalid, trust that it would have been caught by the "refresh token" call in main if session != nil { return cached, nil diff --git a/fetchers/cache.go b/fetchers/cache.go index 3802823..b45856d 100644 --- a/fetchers/cache.go +++ b/fetchers/cache.go @@ -161,6 +161,31 @@ func (c *Cache) GetPostsByDID(did string) ([]notes.Note, error) { return posts, nil } +// returns how many posts we expect to have cached for an author +// for detecting gaps caused by uneven TTL expiry +func (c *Cache) GetPostCount(did string) (int, bool) { + key := fmt.Sprintf("post_count:%s", did) + count, err := c.client.Get(c.ctx, key).Int() + if err != nil { + return 0, false + } + return count, true +} + +// resets the expected count. Used after a fresh fetch +func (c *Cache) SetPostCount(did string, count int) { + key := fmt.Sprintf("post_count:%s", did) + c.client.Set(c.ctx, key, count, c.postExpiry) +} + +// increments the expected post count. Used for continuation fetches +// (fetching from a previous cursor) and jetstream events +func (c *Cache) AddPostCount(did string, incr int) { + key := fmt.Sprintf("post_count:%s", did) + c.client.IncrBy(c.ctx, key, int64(incr)) + c.client.Expire(c.ctx, key, c.postExpiry) +} + // track oldest sent post per profile func (c *Cache) GetOldestSent(did string) (time.Time, bool) { key := fmt.Sprintf("oldest_sent:%s", did) diff --git a/fetchers/jetstream.go b/fetchers/jetstream.go index 866533a..b73137e 100644 --- a/fetchers/jetstream.go +++ b/fetchers/jetstream.go @@ -98,6 +98,7 @@ func handleJetstreamPost(conf *Config, did, uri string, commit *jetstreamCommit) if commit.Operation == "delete" { if _, found := conf.Cache.Get(uri); found { conf.Cache.Delete(uri) + conf.Cache.AddPostCount(did, -1) } return } @@ -113,6 +114,11 @@ func handleJetstreamPost(conf *Config, did, uri string, commit *jetstreamCommit) // cache the result if _, err := parseAppBskyFeedPostRecord(rawRecord, author, conf); err != nil { conf.Logger.Warn("jetstream: failed to parse post", "uri", uri, "error", err) + return + } + + if commit.Operation == "create" { + conf.Cache.AddPostCount(did, 1) } } diff --git a/main.go b/main.go index 85c8712..6fb7bcf 100644 --- a/main.go +++ b/main.go @@ -373,14 +373,28 @@ func postsHandler(w http.ResponseWriter, r *http.Request) { } config.Logger.Debug("posts ready", "count", len(posts)) - // background refetch if cache is running low + // background refetch if cache is running low or has developed a gap oldestSent, hasOldest := config.Cache.GetOldestSent(session.DID) if !hasOldest || len(posts) > 0 { oldest := posts[len(posts)-1].CreatedAt count, _ := config.Cache.CountPostsOlderThan(session.DID, oldest) cursor, hasCursor := config.Cache.GetCursor(session.DID) - config.Logger.Debug("refetch check", "count", count, "pageSize", config.PageSize, "hasCursor", hasCursor, "oldest", oldest) - if count < config.PageSize && hasCursor { + + // skip right after a fresh fetch; expected count was just reset to match it + hasGap := false + if len(cached) > 0 { + if expectedCount, hasExpected := config.Cache.GetPostCount(session.DID); hasExpected { + hasGap = len(cached) < expectedCount + } + } + + config.Logger.Debug("refetch check", "count", count, "pageSize", config.PageSize, "hasCursor", hasCursor, "oldest", oldest, "hasGap", hasGap) + + switch { + case hasGap: + sessionCopy := session.Copy() + go fetchers.FetchAppBskyFeedPostsWithCursor(config, sessionCopy.DID, sessionCopy.PDS, sessionCopy, "") + case count < config.PageSize && hasCursor: sessionCopy := session.Copy() go fetchers.FetchAppBskyFeedPostsWithCursor(config, sessionCopy.DID, sessionCopy.PDS, sessionCopy, cursor) } -- cgit v1.2.3