Implement fetched posts count tracking for catching gaps in cached postsHEADmain

3 files changed, 43 insertions, 1 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)
 	}
 }