recursive fetching of reply parents + thread deduplication

2 files changed, 93 insertions, 4 deletions

diff --git a/backend/fetchers/app.bsky.feed.go b/backend/fetchers/app.bsky.feed.go
index 2a38d5a..1b9acc1 100644
--- a/backend/fetchers/app.bsky.feed.go
+++ b/backend/fetchers/app.bsky.feed.go
@@ -26,7 +26,7 @@ func FetchAppBskyFeedPosts(logger *slog.Logger, did, pdsURL, slingshotURL string
 		if err != nil {
 			logger.Error("Fetching from slingshot failed")
 		} else {
-			return parseAppBskyFeedPostCollection(logger, records, author, cache)
+			return parseAppBskyFeedPostCollection(logger, slingshotURL, session, records, author, cache)
 		}
 	}
 
@@ -35,7 +35,7 @@ func FetchAppBskyFeedPosts(logger *slog.Logger, did, pdsURL, slingshotURL string
 		return nil, err
 	}
 
-	return parseAppBskyFeedPostCollection(logger, records, author, cache)
+	return parseAppBskyFeedPostCollection(logger, slingshotURL, session, records, author, cache)
 }
 
 func GetAppBskyFeedPostByURI(logger *slog.Logger, uri, slingshotURL string, session *auth.OAuthState, cache *Cache) (*notes.Note, error) {
@@ -114,7 +114,7 @@ func parseAppBskyFeedPostRecord(rawRecord *general.RawRecord, author *notes.Auth
 	return note, nil
 }
 
-func parseAppBskyFeedPostCollection(logger *slog.Logger, records []general.RawRecord, author *notes.Author, cache *Cache) ([]notes.Note, error) {
+func parseAppBskyFeedPostCollection(logger *slog.Logger, slingshotURL string, session *auth.OAuthState, records []general.RawRecord, author *notes.Author, cache *Cache) ([]notes.Note, error) {
 	posts := make([]notes.Note, 0, len(records))
 	for _, record := range records {
 		note, err := parseAppBskyFeedPostRecord(&record, author, cache)
@@ -122,9 +122,58 @@ func parseAppBskyFeedPostCollection(logger *slog.Logger, records []general.RawRe
 			logger.Error("failed to parse post record", "error", err)
 			continue
 		}
+		if note.ReplyTo != nil {
+			parents, err := FetchParents(logger, *note.ReplyTo, slingshotURL, session, cache, 10)
+			if err != nil {
+				logger.Warn("failed to fetch parent chain", "uri", note.URI, "error", err)
+			}
+			note.Parents = parents
+		}
 		cache.Set(note.URI, note)
 		posts = append(posts, *note)
 	}
 
-	return posts, nil
+	return notes.DeduplicateThreads(posts), nil
+}
+
+func FetchParents(logger *slog.Logger, replyToURI string, slingshotURL string, session *auth.OAuthState, cache *Cache, maxDepth int) ([]string, error) {
+	parents := []string{}
+	seen := make(map[string]bool)
+	currentURI := replyToURI
+
+	for currentURI != "" {
+		// check for circular reference
+		if seen[currentURI] {
+			logger.Warn("circular reference detected in thread", "uri", currentURI)
+			break
+		}
+		seen[currentURI] = true
+
+		note, err := GetAppBskyFeedPostByURI(logger, currentURI, slingshotURL, session, cache)
+		if err != nil {
+			logger.Warn("failed to fetch parent", "uri", currentURI, "error", err)
+			break
+		}
+
+		// preemptively fetch posts from author to avoid hammering servers if there's a back and forth
+		if _, found := cache.GetProfile(note.AuthorDID); !found {
+			logger.Debug("caching posts from thread participant", "did", note.AuthorDID)
+			authorPDS, err := FetchPDS(note.AuthorDID)
+			if err != nil {
+				logger.Warn("failed to fetch parent", "uri", currentURI, "error", err)
+				break
+			}
+			go FetchAppBskyFeedPosts(logger, note.AuthorDID, authorPDS, slingshotURL, session, cache)
+		}
+
+		parents = append(parents, currentURI)
+
+		if note.ReplyTo == nil {
+			break
+		}
+
+		currentURI = *note.ReplyTo
+	}
+
+	return parents, nil
 }
diff --git a/backend/notes/note.go b/backend/notes/note.go
index 3b32aa1..591caff 100644
--- a/backend/notes/note.go
+++ b/backend/notes/note.go
@@ -1,6 +1,7 @@
 package notes
 
 import (
+	"sort"
 	"time"
 )
 
@@ -24,3 +25,42 @@ type Facet struct {
 	Type      string `json:"type"`
 	Value     string `json:"value"` // the link destination, whether url, profile or tag
 }
+
+func isChainContained(smallChain, bigChain []string) bool {
+	if len(smallChain) > len(bigChain) {
+		return false
+	}
+
+	// check if smallChain is a suffix of bigChain
+	offset := len(bigChain) - len(smallChain)
+	for i, uri := range smallChain {
+		if bigChain[offset+i] != uri {
+			return false
+		}
+	}
+	return true
+}
+
+func DeduplicateThreads(posts []Note) []Note {
+	// sort by parent chain length (longest first)
+	sort.Slice(posts, func(i, j int) bool {
+		return len(posts[i].Parents) > len(posts[j].Parents)
+	})
+
+	filtered := []Note{}
+	for _, post := range posts {
+		isRedundant := false
+		for _, kept := range filtered {
+			// check if this post's chain is contained in any kept post
+			if isChainContained(post.Parents, kept.Parents) && post.URI == kept.URI {
+				isRedundant = true
+				break
+			}
+		}
+		if !isRedundant {
+			filtered = append(filtered, post)
+		}
+	}
+
+	return filtered
+}