From c2472b20ae056bee40cec95fe11bbbef04651d52 Mon Sep 17 00:00:00 2001 From: Guillermo Serrahima Date: Fri, 20 Mar 2026 00:55:06 +0100 Subject: recursive fetching of reply parents + thread deduplication --- backend/fetchers/app.bsky.feed.go | 57 ++++++++++++++++++++++++++++++++++++--- backend/notes/note.go | 40 +++++++++++++++++++++++++++ 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 +} -- cgit v1.2.3