+3

events.go

··· 18 18 "github.com/jackc/pgx/v5/pgconn" 19 19 "github.com/jackc/pgx/v5/pgxpool" 20 20 "gorm.io/gorm" 21 21 + 22 22 + . "github.com/whyrusleeping/konbini/models" 21 23 ) 22 24 23 25 type PostgresBackend struct { ··· 645 647 Author: repo.ID, 646 648 Rkey: rkey, 647 649 Did: rec.Did, 650 650 + Raw: recb, 648 651 }).Error; err != nil { 649 652 return err 650 653 }

+2

handlers.go

··· 16 16 "github.com/labstack/echo/v4/middleware" 17 17 "github.com/labstack/gommon/log" 18 18 "github.com/whyrusleeping/market/models" 19 19 + 20 20 + . "github.com/whyrusleeping/konbini/models" 19 21 ) 20 22 21 23 func (s *Server) runApiServer() error {

+13 -2

hydration/hydrator.go

··· 10 10 db *gorm.DB 11 11 dir identity.Directory 12 12 13 13 - missingActorCallback func(string) 14 14 - missingPostCallback func(string) 13 13 + missingActorCallback func(string) 14 14 + missingPostCallback func(string) 15 15 + missingFeedGeneratorCallback func(string) 15 16 } 16 17 17 18 // NewHydrator creates a new Hydrator ··· 29 30 func (h *Hydrator) addMissingActor(did string) { 30 31 if h.missingActorCallback != nil { 31 32 h.missingActorCallback(did) 33 33 + } 34 34 + } 35 35 + 36 36 + func (h *Hydrator) SetMissingFeedGeneratorCallback(fn func(string)) { 37 37 + h.missingFeedGeneratorCallback = fn 38 38 + } 39 39 + 40 40 + func (h *Hydrator) AddMissingFeedGenerator(uri string) { 41 41 + if h.missingFeedGeneratorCallback != nil { 42 42 + h.missingFeedGeneratorCallback(uri) 32 43 } 33 44 } 34 45

+29

hydration/utils.go

··· 1 1 + package hydration 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "fmt" 6 6 + 7 7 + "github.com/bluesky-social/indigo/atproto/syntax" 8 8 + ) 9 9 + 10 10 + func (h *Hydrator) NormalizeUri(ctx context.Context, uri string) (string, error) { 11 11 + puri, err := syntax.ParseATURI(uri) 12 12 + if err != nil { 13 13 + return "", fmt.Errorf("invalid uri: %w", err) 14 14 + } 15 15 + 16 16 + var did string 17 17 + if !puri.Authority().IsDID() { 18 18 + resp, err := h.dir.LookupHandle(ctx, syntax.Handle(puri.Authority().String())) 19 19 + if err != nil { 20 20 + return "", err 21 21 + } 22 22 + 23 23 + did = resp.DID.String() 24 24 + } else { 25 25 + did = puri.Authority().String() 26 26 + } 27 27 + 28 28 + return fmt.Sprintf("at://%s/%s/%s", did, puri.Collection().String(), puri.RecordKey().String()), nil 29 29 + }

+10 -5

main.go

··· 30 30 "github.com/urfave/cli/v2" 31 31 "github.com/whyrusleeping/konbini/xrpc" 32 32 "gorm.io/gorm/logger" 33 33 + 34 34 + . "github.com/whyrusleeping/konbini/models" 33 35 ) 34 36 35 37 var handleOpHist = promauto.NewHistogramVec(prometheus.HistogramOpts{ ··· 151 153 client: cc, 152 154 dir: dir, 153 155 154 154 - missingProfiles: make(chan string, 1024), 155 155 - missingPosts: make(chan string, 1024), 156 156 + missingProfiles: make(chan string, 1024), 157 157 + missingPosts: make(chan string, 1024), 158 158 + missingFeedGenerators: make(chan string, 1024), 156 159 } 157 160 fmt.Println("MY DID: ", s.mydid) 158 161 ··· 199 202 200 203 go s.missingProfileFetcher() 201 204 go s.missingPostFetcher() 205 205 + go s.missingFeedGeneratorFetcher() 202 206 203 207 seqno, err := loadLastSeq(db, "firehose_seq") 204 208 if err != nil { ··· 223 227 seqLk sync.Mutex 224 228 lastSeq int64 225 229 226 226 - mpLk sync.Mutex 227 227 - missingProfiles chan string 228 228 - missingPosts chan string 230 230 + mpLk sync.Mutex 231 231 + missingProfiles chan string 232 232 + missingPosts chan string 233 233 + missingFeedGenerators chan string 229 234 } 230 235 231 236 func (s *Server) getXrpcClient() (*xrpclib.Client, error) {

+64

missing.go

··· 131 131 132 132 return s.backend.HandleCreatePost(ctx, repo, rkey, buf.Bytes(), cc) 133 133 } 134 134 + 135 135 + func (s *Server) addMissingFeedGenerator(ctx context.Context, uri string) { 136 136 + slog.Info("adding missing feed generator to fetch queue", "uri", uri) 137 137 + select { 138 138 + case s.missingFeedGenerators <- uri: 139 139 + case <-ctx.Done(): 140 140 + } 141 141 + } 142 142 + 143 143 + func (s *Server) missingFeedGeneratorFetcher() { 144 144 + for uri := range s.missingFeedGenerators { 145 145 + if err := s.fetchMissingFeedGenerator(context.TODO(), uri); err != nil { 146 146 + log.Warn("failed to fetch missing feed generator", "uri", uri, "error", err) 147 147 + } 148 148 + } 149 149 + } 150 150 + 151 151 + func (s *Server) fetchMissingFeedGenerator(ctx context.Context, uri string) error { 152 152 + // Parse AT URI: at://did:plc:xxx/app.bsky.feed.generator/rkey 153 153 + puri, err := syntax.ParseATURI(uri) 154 154 + if err != nil { 155 155 + return fmt.Errorf("invalid AT URI: %s", uri) 156 156 + } 157 157 + 158 158 + did := puri.Authority().String() 159 159 + collection := puri.Collection().String() 160 160 + rkey := puri.RecordKey().String() 161 161 + 162 162 + repo, err := s.backend.getOrCreateRepo(ctx, did) 163 163 + if err != nil { 164 164 + return err 165 165 + } 166 166 + 167 167 + resp, err := s.dir.LookupDID(ctx, syntax.DID(did)) 168 168 + if err != nil { 169 169 + return err 170 170 + } 171 171 + 172 172 + c := &xrpclib.Client{ 173 173 + Host: resp.PDSEndpoint(), 174 174 + } 175 175 + 176 176 + rec, err := atproto.RepoGetRecord(ctx, c, "", collection, did, rkey) 177 177 + if err != nil { 178 178 + return err 179 179 + } 180 180 + 181 181 + feedGen, ok := rec.Value.Val.(*bsky.FeedGenerator) 182 182 + if !ok { 183 183 + return fmt.Errorf("record we got back wasn't a feed generator somehow") 184 184 + } 185 185 + 186 186 + buf := new(bytes.Buffer) 187 187 + if err := feedGen.MarshalCBOR(buf); err != nil { 188 188 + return err 189 189 + } 190 190 + 191 191 + cc, err := cid.Decode(*rec.Cid) 192 192 + if err != nil { 193 193 + return err 194 194 + } 195 195 + 196 196 + return s.backend.HandleCreateFeedGenerator(ctx, repo, rkey, buf.Bytes(), cc) 197 197 + }

+1 -1

models.go models/models.go

··· 1 1 - package main 1 1 + package models 2 2 3 3 import ( 4 4 "time"

+6

pgbackend.go

··· 17 17 "gorm.io/gorm" 18 18 "gorm.io/gorm/clause" 19 19 "gorm.io/gorm/logger" 20 20 + 21 21 + . "github.com/whyrusleeping/konbini/models" 20 22 ) 21 23 22 24 func (b *PostgresBackend) getOrCreateRepo(ctx context.Context, did string) (*Repo, error) { ··· 404 406 func (b *PostgresBackend) TrackMissingActor(did string) { 405 407 b.s.addMissingProfile(context.TODO(), did) 406 408 } 409 409 + 410 410 + func (b *PostgresBackend) TrackMissingFeedGenerator(uri string) { 411 411 + b.s.addMissingFeedGenerator(context.TODO(), uri) 412 412 + }

+2

seqno.go

··· 3 3 import ( 4 4 "gorm.io/gorm" 5 5 "gorm.io/gorm/clause" 6 6 + 7 7 + . "github.com/whyrusleeping/konbini/models" 6 8 ) 7 9 8 10 func storeLastSeq(db *gorm.DB, key string, seq int64) error {

+136 -5

views/feed.go

··· 1 1 package views 2 2 3 3 import ( 4 4 + "fmt" 5 5 + 4 6 "github.com/bluesky-social/indigo/api/bsky" 5 7 "github.com/bluesky-social/indigo/lex/util" 6 8 "github.com/whyrusleeping/konbini/hydration" ··· 40 42 } 41 43 } 42 44 43 43 - // TODO: Add embed handling - need to convert embed types to proper views 44 44 - // if post.Post.Embed != nil { 45 45 - // view.Embed = formatEmbed(post.Post.Embed) 46 46 - // } 45 45 + // Add embed handling 46 46 + if post.Post.Embed != nil { 47 47 + view.Embed = formatEmbed(post.Post.Embed, post.Author) 48 48 + } 47 49 48 50 return view 49 51 } ··· 56 58 } 57 59 58 60 // ThreadViewPost builds a thread view post (app.bsky.feed.defs#threadViewPost) 59 59 - func ThreadViewPost(post *hydration.PostInfo, author *hydration.ActorInfo, parent, replies interface{}) *bsky.FeedDefs_ThreadViewPost { 61 61 + func ThreadViewPost(post *hydration.PostInfo, author *hydration.ActorInfo, parent, replies any) *bsky.FeedDefs_ThreadViewPost { 60 62 view := &bsky.FeedDefs_ThreadViewPost{ 61 63 LexiconTypeID: "app.bsky.feed.defs#threadViewPost", 62 64 Post: PostView(post, author), ··· 67 69 68 70 return view 69 71 } 72 72 + 73 73 + func formatEmbed(embed *bsky.FeedPost_Embed, authorDID string) *bsky.FeedDefs_PostView_Embed { 74 74 + if embed == nil { 75 75 + return nil 76 76 + } 77 77 + 78 78 + result := &bsky.FeedDefs_PostView_Embed{} 79 79 + 80 80 + // Handle images 81 81 + if embed.EmbedImages != nil { 82 82 + viewImages := make([]*bsky.EmbedImages_ViewImage, len(embed.EmbedImages.Images)) 83 83 + for i, img := range embed.EmbedImages.Images { 84 84 + // Convert blob to CDN URLs 85 85 + fullsize := "" 86 86 + thumb := "" 87 87 + if img.Image != nil { 88 88 + // CDN URL format for feed images 89 89 + cid := img.Image.Ref.String() 90 90 + fullsize = fmt.Sprintf("https://cdn.bsky.app/img/feed_fullsize/plain/%s/%s@jpeg", authorDID, cid) 91 91 + thumb = fmt.Sprintf("https://cdn.bsky.app/img/feed_thumbnail/plain/%s/%s@jpeg", authorDID, cid) 92 92 + } 93 93 + 94 94 + viewImages[i] = &bsky.EmbedImages_ViewImage{ 95 95 + Alt: img.Alt, 96 96 + AspectRatio: img.AspectRatio, 97 97 + Fullsize: fullsize, 98 98 + Thumb: thumb, 99 99 + } 100 100 + } 101 101 + result.EmbedImages_View = &bsky.EmbedImages_View{ 102 102 + LexiconTypeID: "app.bsky.embed.images#view", 103 103 + Images: viewImages, 104 104 + } 105 105 + return result 106 106 + } 107 107 + 108 108 + // Handle external links 109 109 + if embed.EmbedExternal != nil && embed.EmbedExternal.External != nil { 110 110 + // Convert blob thumb to CDN URL if present 111 111 + var thumbURL *string 112 112 + if embed.EmbedExternal.External.Thumb != nil { 113 113 + // CDN URL for external link thumbnails 114 114 + cid := embed.EmbedExternal.External.Thumb.Ref.String() 115 115 + url := fmt.Sprintf("https://cdn.bsky.app/img/feed_thumbnail/plain/%s/%s@jpeg", authorDID, cid) 116 116 + thumbURL = &url 117 117 + } 118 118 + 119 119 + result.EmbedExternal_View = &bsky.EmbedExternal_View{ 120 120 + LexiconTypeID: "app.bsky.embed.external#view", 121 121 + External: &bsky.EmbedExternal_ViewExternal{ 122 122 + Uri: embed.EmbedExternal.External.Uri, 123 123 + Title: embed.EmbedExternal.External.Title, 124 124 + Description: embed.EmbedExternal.External.Description, 125 125 + Thumb: thumbURL, 126 126 + }, 127 127 + } 128 128 + return result 129 129 + } 130 130 + 131 131 + // Handle video 132 132 + if embed.EmbedVideo != nil { 133 133 + // TODO: Implement video embed view 134 134 + // This would require converting video blob to CDN URLs and playlist URLs 135 135 + return nil 136 136 + } 137 137 + 138 138 + // Handle record (quote posts, etc.) 139 139 + if embed.EmbedRecord != nil { 140 140 + // TODO: Implement record embed view 141 141 + // This requires hydrating the embedded record, which is complex 142 142 + // For now, return nil to skip these embeds 143 143 + return nil 144 144 + } 145 145 + 146 146 + // Handle record with media (quote post with images/external) 147 147 + if embed.EmbedRecordWithMedia != nil { 148 148 + // TODO: Implement record with media embed view 149 149 + // This combines record hydration with media conversion 150 150 + return nil 151 151 + } 152 152 + 153 153 + return nil 154 154 + } 155 155 + 156 156 + // GeneratorView builds a feed generator view (app.bsky.feed.defs#generatorView) 157 157 + func GeneratorView(uri, cid string, record *bsky.FeedGenerator, creator *hydration.ActorInfo, likeCount int64, viewerLike string, indexedAt string) *bsky.FeedDefs_GeneratorView { 158 158 + view := &bsky.FeedDefs_GeneratorView{ 159 159 + LexiconTypeID: "app.bsky.feed.defs#generatorView", 160 160 + Uri: uri, 161 161 + Cid: cid, 162 162 + Did: record.Did, 163 163 + Creator: ProfileView(creator), 164 164 + DisplayName: record.DisplayName, 165 165 + Description: record.Description, 166 166 + IndexedAt: indexedAt, 167 167 + } 168 168 + 169 169 + // Add optional fields 170 170 + if record.Avatar != nil { 171 171 + avatarURL := fmt.Sprintf("https://cdn.bsky.app/img/avatar/plain/%s/%s@jpeg", creator.DID, record.Avatar.Ref.String()) 172 172 + view.Avatar = &avatarURL 173 173 + } 174 174 + 175 175 + if record.DescriptionFacets != nil && len(record.DescriptionFacets) > 0 { 176 176 + view.DescriptionFacets = record.DescriptionFacets 177 177 + } 178 178 + 179 179 + if record.AcceptsInteractions != nil { 180 180 + view.AcceptsInteractions = record.AcceptsInteractions 181 181 + } 182 182 + 183 183 + if record.ContentMode != nil { 184 184 + view.ContentMode = record.ContentMode 185 185 + } 186 186 + 187 187 + // Add like count if present 188 188 + if likeCount > 0 { 189 189 + view.LikeCount = &likeCount 190 190 + } 191 191 + 192 192 + // Add viewer state if viewer has liked 193 193 + if viewerLike != "" { 194 194 + view.Viewer = &bsky.FeedDefs_GeneratorViewerState{ 195 195 + Like: &viewerLike, 196 196 + } 197 197 + } 198 198 + 199 199 + return view 200 200 + }

+2 -15

xrpc/actor/getProfiles.go

··· 3 3 import ( 4 4 "net/http" 5 5 6 6 + "github.com/bluesky-social/indigo/api/bsky" 6 7 "github.com/labstack/echo/v4" 7 8 "github.com/whyrusleeping/konbini/hydration" 8 9 "github.com/whyrusleeping/konbini/views" ··· 28 29 ctx := c.Request().Context() 29 30 30 31 // Resolve all actors to DIDs and hydrate profiles 31 31 - profiles := make([]interface{}, 0) 32 32 + profiles := make([]*bsky.ActorDefs_ProfileViewDetailed, 0, len(actors)) 32 33 for _, actor := range actors { 33 34 // Resolve actor to DID 34 35 did, err := hydrator.ResolveDID(ctx, actor) ··· 43 44 // Skip actors that can't be hydrated 44 45 continue 45 46 } 46 46 - 47 47 - // Get counts for the profile 48 48 - type counts struct { 49 49 - Followers int 50 50 - Follows int 51 51 - Posts int 52 52 - } 53 53 - var c counts 54 54 - db.Raw(` 55 55 - SELECT 56 56 - (SELECT COUNT(*) FROM follows WHERE subject = (SELECT id FROM repos WHERE did = ?)) as followers, 57 57 - (SELECT COUNT(*) FROM follows WHERE author = (SELECT id FROM repos WHERE did = ?)) as follows, 58 58 - (SELECT COUNT(*) FROM posts WHERE author = (SELECT id FROM repos WHERE did = ?)) as posts 59 59 - `, did, did, did).Scan(&c) 60 47 61 48 profiles = append(profiles, views.ProfileViewDetailed(actorInfo)) 62 49 }

+59 -25

xrpc/feed/getAuthorFeed.go

··· 1 1 package feed 2 2 3 3 import ( 4 4 + "context" 5 5 + "log/slog" 4 6 "net/http" 5 7 "strconv" 8 8 + "sync" 6 9 "time" 7 10 11 11 + "github.com/bluesky-social/indigo/api/bsky" 8 12 "github.com/labstack/echo/v4" 9 13 "github.com/whyrusleeping/konbini/hydration" 10 14 "github.com/whyrusleeping/konbini/views" 11 15 "gorm.io/gorm" 12 16 ) 13 17 18 18 + type postRow struct { 19 19 + URI string 20 20 + AuthorID uint 21 21 + } 22 22 + 14 23 // HandleGetAuthorFeed implements app.bsky.feed.getAuthorFeed 15 24 func HandleGetAuthorFeed(c echo.Context, db *gorm.DB, hydrator *hydration.Hydrator) error { 16 25 actorParam := c.QueryParam("actor") 17 26 if actorParam == "" { 18 18 - return c.JSON(http.StatusBadRequest, map[string]interface{}{ 27 27 + return c.JSON(http.StatusBadRequest, map[string]any{ 19 28 "error": "InvalidRequest", 20 29 "message": "actor parameter is required", 21 30 }) ··· 49 58 // Resolve actor to DID 50 59 did, err := hydrator.ResolveDID(ctx, actorParam) 51 60 if err != nil { 52 52 - return c.JSON(http.StatusBadRequest, map[string]interface{}{ 61 61 + return c.JSON(http.StatusBadRequest, map[string]any{ 53 62 "error": "ActorNotFound", 54 63 "message": "actor not found", 55 64 }) ··· 87 96 ` 88 97 } 89 98 90 90 - type postRow struct { 91 91 - URI string 92 92 - AuthorID uint 93 93 - } 94 99 var rows []postRow 95 100 if err := db.Raw(query, did, cursor, limit).Scan(&rows).Error; err != nil { 96 96 - return c.JSON(http.StatusInternalServerError, map[string]interface{}{ 101 101 + return c.JSON(http.StatusInternalServerError, map[string]any{ 97 102 "error": "InternalError", 98 103 "message": "failed to query author feed", 99 104 }) 100 105 } 101 106 102 102 - // Hydrate posts 103 103 - feed := make([]interface{}, 0) 104 104 - for _, row := range rows { 105 105 - postInfo, err := hydrator.HydratePost(ctx, row.URI, viewer) 106 106 - if err != nil { 107 107 - continue 108 108 - } 109 109 - 110 110 - // Hydrate author 111 111 - authorInfo, err := hydrator.HydrateActor(ctx, postInfo.Author) 112 112 - if err != nil { 113 113 - continue 114 114 - } 115 115 - 116 116 - feedItem := views.FeedViewPost(postInfo, authorInfo) 117 117 - feed = append(feed, feedItem) 118 118 - } 107 107 + feed := hydratePostRows(ctx, hydrator, viewer, rows) 119 108 120 109 // Generate next cursor 121 110 var nextCursor string ··· 130 119 } 131 120 } 132 121 133 133 - return c.JSON(http.StatusOK, map[string]interface{}{ 122 122 + return c.JSON(http.StatusOK, map[string]any{ 134 123 "feed": feed, 135 124 "cursor": nextCursor, 136 125 }) 137 126 } 127 127 + 128 128 + func hydratePostRows(ctx context.Context, hydrator *hydration.Hydrator, viewer string, rows []postRow) []*bsky.FeedDefs_FeedViewPost { 129 129 + // Hydrate posts 130 130 + var wg sync.WaitGroup 131 131 + 132 132 + var outLk sync.Mutex 133 133 + feed := make([]*bsky.FeedDefs_FeedViewPost, len(rows)) 134 134 + for i, row := range rows { 135 135 + wg.Add(1) 136 136 + go func(i int, row postRow) { 137 137 + defer wg.Done() 138 138 + 139 139 + postInfo, err := hydrator.HydratePost(ctx, row.URI, viewer) 140 140 + if err != nil { 141 141 + slog.Error("failed to hydrate post", "uri", row.URI, "error", err) 142 142 + return 143 143 + } 144 144 + 145 145 + // Hydrate author 146 146 + authorInfo, err := hydrator.HydrateActor(ctx, postInfo.Author) 147 147 + if err != nil { 148 148 + slog.Error("failed to hydrate actor", "actor", postInfo.Author, "error", err) 149 149 + return 150 150 + } 151 151 + 152 152 + feedItem := views.FeedViewPost(postInfo, authorInfo) 153 153 + outLk.Lock() 154 154 + feed[i] = feedItem 155 155 + outLk.Unlock() 156 156 + }(i, row) 157 157 + } 158 158 + wg.Wait() 159 159 + 160 160 + x := 0 161 161 + for i := 0; i < len(feed); i++ { 162 162 + if feed[i] != nil { 163 163 + feed[x] = feed[i] 164 164 + x++ 165 165 + continue 166 166 + } 167 167 + } 168 168 + feed = feed[:x] 169 169 + 170 170 + return feed 171 171 + }

+158

xrpc/feed/getFeed.go

··· 1 1 + package feed 2 2 + 3 3 + import ( 4 4 + "bytes" 5 5 + "log/slog" 6 6 + "net/http" 7 7 + "strconv" 8 8 + 9 9 + "github.com/bluesky-social/indigo/api/bsky" 10 10 + "github.com/bluesky-social/indigo/atproto/identity" 11 11 + "github.com/bluesky-social/indigo/atproto/syntax" 12 12 + "github.com/bluesky-social/indigo/xrpc" 13 13 + "github.com/labstack/echo/v4" 14 14 + "github.com/whyrusleeping/konbini/hydration" 15 15 + "github.com/whyrusleeping/konbini/views" 16 16 + "github.com/whyrusleeping/market/models" 17 17 + "gorm.io/gorm" 18 18 + ) 19 19 + 20 20 + // HandleGetFeed implements app.bsky.feed.getFeed 21 21 + // Gets posts from a custom feed generator 22 22 + func HandleGetFeed(c echo.Context, db *gorm.DB, hydrator *hydration.Hydrator, dir identity.Directory) error { 23 23 + // Parse parameters 24 24 + feedURI := c.QueryParam("feed") 25 25 + if feedURI == "" { 26 26 + return c.JSON(http.StatusBadRequest, map[string]interface{}{ 27 27 + "error": "InvalidRequest", 28 28 + "message": "feed parameter is required", 29 29 + }) 30 30 + } 31 31 + 32 32 + // Parse limit 33 33 + limit := int64(50) 34 34 + if limitParam := c.QueryParam("limit"); limitParam != "" { 35 35 + if l, err := strconv.ParseInt(limitParam, 10, 64); err == nil && l > 0 && l <= 100 { 36 36 + limit = l 37 37 + } 38 38 + } 39 39 + 40 40 + // Parse cursor 41 41 + cursor := c.QueryParam("cursor") 42 42 + 43 43 + ctx := c.Request().Context() 44 44 + viewer := getUserDID(c) 45 45 + 46 46 + // Extract feed generator DID and rkey from URI 47 47 + // URI format: at://did:plc:xxx/app.bsky.feed.generator/rkey 48 48 + did := extractDIDFromURI(feedURI) 49 49 + rkey := extractRkeyFromURI(feedURI) 50 50 + 51 51 + if did == "" || rkey == "" { 52 52 + return c.JSON(http.StatusBadRequest, map[string]interface{}{ 53 53 + "error": "InvalidRequest", 54 54 + "message": "invalid feed URI format", 55 55 + }) 56 56 + } 57 57 + 58 58 + // Check if feed generator exists in database 59 59 + var feedGen models.FeedGenerator 60 60 + if err := db.Raw(` 61 61 + SELECT * FROM feed_generators fg WHERE fg.author = (select id from repos where did = ?) AND fg.rkey = ? 62 62 + `, did, rkey).Scan(&feedGen).Error; err != nil { 63 63 + return err 64 64 + } 65 65 + 66 66 + if feedGen.ID == 0 { 67 67 + hydrator.AddMissingFeedGenerator(feedURI) 68 68 + return c.JSON(http.StatusNotFound, map[string]interface{}{ 69 69 + "error": "NotFound", 70 70 + "message": "feed generator not found", 71 71 + }) 72 72 + } 73 73 + 74 74 + // Decode the feed generator record to get the service DID 75 75 + var feedGenRecord bsky.FeedGenerator 76 76 + if err := feedGenRecord.UnmarshalCBOR(bytes.NewReader(feedGen.Raw)); err != nil { 77 77 + slog.Error("failed to decode feed generator record", "error", err) 78 78 + return c.JSON(http.StatusInternalServerError, map[string]interface{}{ 79 79 + "error": "InternalError", 80 80 + "message": "failed to decode feed generator record", 81 81 + }) 82 82 + } 83 83 + 84 84 + // Parse the service DID 85 85 + serviceDID, err := syntax.ParseDID(feedGenRecord.Did) 86 86 + if err != nil { 87 87 + slog.Error("invalid service DID in feed generator", "error", err, "did", feedGenRecord.Did) 88 88 + return c.JSON(http.StatusInternalServerError, map[string]interface{}{ 89 89 + "error": "InternalError", 90 90 + "message": "invalid service DID", 91 91 + }) 92 92 + } 93 93 + 94 94 + // Resolve the service DID to get its endpoint 95 95 + serviceIdent, err := dir.LookupDID(ctx, serviceDID) 96 96 + if err != nil { 97 97 + slog.Error("failed to resolve service DID", "error", err, "did", serviceDID) 98 98 + return c.JSON(http.StatusInternalServerError, map[string]interface{}{ 99 99 + "error": "InternalError", 100 100 + "message": "failed to resolve service endpoint", 101 101 + }) 102 102 + } 103 103 + 104 104 + serviceEndpoint := serviceIdent.GetServiceEndpoint("bsky_fg") 105 105 + if serviceEndpoint == "" { 106 106 + slog.Error("service has no bsky_fg endpoint", "did", serviceDID) 107 107 + return c.JSON(http.StatusInternalServerError, map[string]interface{}{ 108 108 + "error": "InternalError", 109 109 + "message": "service has no endpoint", 110 110 + }) 111 111 + } 112 112 + 113 113 + // Create XRPC client for the feed generator service 114 114 + client := &xrpc.Client{ 115 115 + Host: serviceEndpoint, 116 116 + } 117 117 + 118 118 + // Call getFeedSkeleton on the service 119 119 + skeleton, err := bsky.FeedGetFeedSkeleton(ctx, client, cursor, feedURI, limit) 120 120 + if err != nil { 121 121 + slog.Error("failed to call getFeedSkeleton", "error", err, "service", serviceEndpoint) 122 122 + // Return empty feed on error rather than failing completely 123 123 + return c.JSON(http.StatusOK, &bsky.FeedGetFeed_Output{ 124 124 + Feed: make([]*bsky.FeedDefs_FeedViewPost, 0), 125 125 + }) 126 126 + } 127 127 + 128 128 + // Hydrate the posts from the skeleton 129 129 + posts := make([]*bsky.FeedDefs_FeedViewPost, 0, len(skeleton.Feed)) 130 130 + for _, skeletonPost := range skeleton.Feed { 131 131 + postURI, err := syntax.ParseATURI(skeletonPost.Post) 132 132 + if err != nil { 133 133 + slog.Warn("invalid post URI in skeleton", "uri", skeletonPost.Post, "error", err) 134 134 + continue 135 135 + } 136 136 + 137 137 + postInfo, err := hydrator.HydratePost(ctx, string(postURI), viewer) 138 138 + if err != nil { 139 139 + slog.Warn("failed to hydrate post", "uri", postURI, "error", err) 140 140 + continue 141 141 + } 142 142 + 143 143 + authorInfo, err := hydrator.HydrateActor(ctx, postInfo.Author) 144 144 + if err != nil { 145 145 + slog.Warn("failed to hydrate author", "did", postInfo.Author, "error", err) 146 146 + continue 147 147 + } 148 148 + 149 149 + posts = append(posts, views.FeedViewPost(postInfo, authorInfo)) 150 150 + } 151 151 + 152 152 + output := &bsky.FeedGetFeed_Output{ 153 153 + Feed: posts, 154 154 + Cursor: skeleton.Cursor, 155 155 + } 156 156 + 157 157 + return c.JSON(http.StatusOK, output) 158 158 + }

+161

xrpc/feed/getFeedGenerator.go

··· 1 1 + package feed 2 2 + 3 3 + import ( 4 4 + "bytes" 5 5 + "log/slog" 6 6 + "net/http" 7 7 + "time" 8 8 + 9 9 + "github.com/bluesky-social/indigo/api/bsky" 10 10 + "github.com/bluesky-social/indigo/atproto/identity" 11 11 + "github.com/bluesky-social/indigo/atproto/syntax" 12 12 + cid "github.com/ipfs/go-cid" 13 13 + "github.com/labstack/echo/v4" 14 14 + mh "github.com/multiformats/go-multihash" 15 15 + "github.com/whyrusleeping/konbini/hydration" 16 16 + "github.com/whyrusleeping/konbini/views" 17 17 + "gorm.io/gorm" 18 18 + ) 19 19 + 20 20 + // HandleGetFeedGenerator implements app.bsky.feed.getFeedGenerator 21 21 + func HandleGetFeedGenerator(c echo.Context, db *gorm.DB, hydrator *hydration.Hydrator, dir identity.Directory) error { 22 22 + ctx := c.Request().Context() 23 23 + 24 24 + // Parse parameters 25 25 + feedURI := c.QueryParam("feed") 26 26 + if feedURI == "" { 27 27 + return c.JSON(http.StatusBadRequest, map[string]any{ 28 28 + "error": "InvalidRequest", 29 29 + "message": "feed parameter is required", 30 30 + }) 31 31 + } 32 32 + 33 33 + nu, err := hydrator.NormalizeUri(ctx, feedURI) 34 34 + if err != nil { 35 35 + return err 36 36 + } 37 37 + feedURI = nu 38 38 + 39 39 + viewer := getUserDID(c) 40 40 + _ = viewer 41 41 + 42 42 + // Extract feed generator DID and rkey from URI 43 43 + did := extractDIDFromURI(feedURI) 44 44 + rkey := extractRkeyFromURI(feedURI) 45 45 + 46 46 + if did == "" || rkey == "" { 47 47 + return c.JSON(http.StatusBadRequest, map[string]any{ 48 48 + "error": "InvalidRequest", 49 49 + "message": "invalid feed URI format", 50 50 + }) 51 51 + } 52 52 + 53 53 + // Query feed generator from database 54 54 + type feedGenRow struct { 55 55 + ID uint 56 56 + Did string 57 57 + Raw []byte 58 58 + AuthorDid string 59 59 + Indexed time.Time 60 60 + } 61 61 + var feedGen feedGenRow 62 62 + err = db.Raw(` 63 63 + SELECT fg.id, fg.did, fg.raw, r.did as author_did, indexed 64 64 + FROM feed_generators fg 65 65 + JOIN repos r ON r.id = fg.author 66 66 + WHERE r.did = ? AND fg.rkey = ? 67 67 + `, did, rkey).Scan(&feedGen).Error 68 68 + 69 69 + if err != nil || feedGen.ID == 0 { 70 70 + // Track this missing feed generator for fetching 71 71 + hydrator.AddMissingFeedGenerator(feedURI) 72 72 + 73 73 + return c.JSON(http.StatusNotFound, map[string]any{ 74 74 + "error": "NotFound", 75 75 + "message": "feed generator not found", 76 76 + }) 77 77 + } 78 78 + 79 79 + // Decode the feed generator record 80 80 + var feedGenRecord bsky.FeedGenerator 81 81 + if err := feedGenRecord.UnmarshalCBOR(bytes.NewReader(feedGen.Raw)); err != nil { 82 82 + slog.Error("failed to decode feed generator record", "error", err) 83 83 + return c.JSON(http.StatusInternalServerError, map[string]any{ 84 84 + "error": "InternalError", 85 85 + "message": "failed to decode feed generator record", 86 86 + }) 87 87 + } 88 88 + 89 89 + // Compute CID from raw bytes 90 90 + hash, err := mh.Sum(feedGen.Raw, mh.SHA2_256, -1) 91 91 + if err != nil { 92 92 + slog.Error("failed to hash record", "error", err) 93 93 + return c.JSON(http.StatusInternalServerError, map[string]any{ 94 94 + "error": "InternalError", 95 95 + "message": "failed to compute CID", 96 96 + }) 97 97 + } 98 98 + recordCid := cid.NewCidV1(cid.DagCBOR, hash).String() 99 99 + 100 100 + // Hydrate the creator 101 101 + creatorInfo, err := hydrator.HydrateActor(ctx, feedGen.AuthorDid) 102 102 + if err != nil { 103 103 + slog.Error("failed to hydrate creator", "error", err, "did", feedGen.AuthorDid) 104 104 + return c.JSON(http.StatusInternalServerError, map[string]any{ 105 105 + "error": "InternalError", 106 106 + "message": "failed to hydrate creator", 107 107 + }) 108 108 + } 109 109 + 110 110 + // Count likes for this feed generator 111 111 + var likeCount int64 112 112 + 113 113 + // Check if viewer has liked this feed generator 114 114 + viewerLike := "" 115 115 + 116 116 + // Validate the service DID (check if it's resolvable) 117 117 + serviceDID, err := syntax.ParseDID(feedGenRecord.Did) 118 118 + if err != nil { 119 119 + slog.Error("invalid service DID in feed generator", "error", err, "did", feedGenRecord.Did) 120 120 + return c.JSON(http.StatusInternalServerError, map[string]any{ 121 121 + "error": "InternalError", 122 122 + "message": "invalid service DID", 123 123 + }) 124 124 + } 125 125 + 126 126 + // Try to resolve the service DID to check if it's online/valid 127 127 + isOnline := true 128 128 + isValid := true 129 129 + serviceIdent, err := dir.LookupDID(ctx, serviceDID) 130 130 + if err != nil { 131 131 + slog.Warn("failed to resolve service DID", "error", err, "did", serviceDID) 132 132 + isOnline = false 133 133 + isValid = false 134 134 + } else { 135 135 + // Check if service has an endpoint 136 136 + serviceEndpoint := serviceIdent.PDSEndpoint() 137 137 + if serviceEndpoint == "" { 138 138 + slog.Warn("service has no PDS endpoint", "did", serviceDID) 139 139 + isValid = false 140 140 + } 141 141 + } 142 142 + 143 143 + // Build the generator view 144 144 + generatorView := views.GeneratorView( 145 145 + feedURI, 146 146 + recordCid, 147 147 + &feedGenRecord, 148 148 + creatorInfo, 149 149 + likeCount, 150 150 + viewerLike, 151 151 + feedGen.Indexed.Format(time.RFC3339), 152 152 + ) 153 153 + 154 154 + output := &bsky.FeedGetFeedGenerator_Output{ 155 155 + View: generatorView, 156 156 + IsOnline: isOnline, 157 157 + IsValid: isValid, 158 158 + } 159 159 + 160 160 + return c.JSON(http.StatusOK, output) 161 161 + }

+5 -27

xrpc/feed/getTimeline.go

··· 1 1 package feed 2 2 3 3 import ( 4 4 - "log/slog" 5 4 "net/http" 6 5 "strconv" 7 6 "time" 8 7 9 8 "github.com/labstack/echo/v4" 10 9 "github.com/whyrusleeping/konbini/hydration" 11 11 - "github.com/whyrusleeping/konbini/views" 12 10 "gorm.io/gorm" 13 11 ) 14 12 ··· 16 14 func HandleGetTimeline(c echo.Context, db *gorm.DB, hydrator *hydration.Hydrator) error { 17 15 viewer := getUserDID(c) 18 16 if viewer == "" { 19 19 - return c.JSON(http.StatusUnauthorized, map[string]interface{}{ 17 17 + return c.JSON(http.StatusUnauthorized, map[string]any{ 20 18 "error": "AuthenticationRequired", 21 19 "message": "authentication required", 22 20 }) ··· 43 41 // Get viewer's repo ID 44 42 var viewerRepoID uint 45 43 if err := db.Raw("SELECT id FROM repos WHERE did = ?", viewer).Scan(&viewerRepoID).Error; err != nil { 46 46 - return c.JSON(http.StatusInternalServerError, map[string]interface{}{ 44 44 + return c.JSON(http.StatusInternalServerError, map[string]any{ 47 45 "error": "InternalError", 48 46 "message": "failed to load viewer", 49 47 }) 50 48 } 51 49 52 50 // Query posts from followed users 53 53 - type postRow struct { 54 54 - URI string 55 55 - AuthorID uint 56 56 - } 57 51 var rows []postRow 58 52 err := db.Raw(` 59 53 SELECT ··· 70 64 `, viewerRepoID, cursor, limit).Scan(&rows).Error 71 65 72 66 if err != nil { 73 73 - return c.JSON(http.StatusInternalServerError, map[string]interface{}{ 67 67 + return c.JSON(http.StatusInternalServerError, map[string]any{ 74 68 "error": "InternalError", 75 69 "message": "failed to query timeline", 76 70 }) 77 71 } 78 72 79 73 // Hydrate posts 80 80 - feed := make([]interface{}, 0) 81 81 - for _, row := range rows { 82 82 - postInfo, err := hydrator.HydratePost(ctx, row.URI, viewer) 83 83 - if err != nil { 84 84 - continue 85 85 - } 86 86 - 87 87 - // Hydrate author 88 88 - authorInfo, err := hydrator.HydrateActor(ctx, postInfo.Author) 89 89 - if err != nil { 90 90 - slog.Error("failed to hydrate actor", "author", postInfo.Author, "error", err) 91 91 - continue 92 92 - } 93 93 - 94 94 - feedItem := views.FeedViewPost(postInfo, authorInfo) 95 95 - feed = append(feed, feedItem) 96 96 - } 74 74 + feed := hydratePostRows(ctx, hydrator, viewer, rows) 97 75 98 76 // Generate next cursor 99 77 var nextCursor string ··· 111 89 } 112 90 } 113 91 114 114 - return c.JSON(http.StatusOK, map[string]interface{}{ 92 92 + return c.JSON(http.StatusOK, map[string]any{ 115 93 "feed": feed, 116 94 "cursor": nextCursor, 117 95 })

+150 -20

xrpc/notification/listNotifications.go

··· 1 1 package notification 2 2 3 3 import ( 4 4 + "bytes" 5 5 + "fmt" 4 6 "net/http" 5 7 "strconv" 8 8 + "time" 6 9 10 10 + "github.com/bluesky-social/indigo/api/atproto" 11 11 + "github.com/bluesky-social/indigo/api/bsky" 12 12 + "github.com/bluesky-social/indigo/lex/util" 13 13 + lexutil "github.com/bluesky-social/indigo/lex/util" 7 14 "github.com/labstack/echo/v4" 8 15 "github.com/whyrusleeping/konbini/hydration" 9 16 "github.com/whyrusleeping/konbini/views" 17 17 + "github.com/whyrusleeping/market/models" 10 18 "gorm.io/gorm" 11 19 ) 12 20 ··· 84 92 } 85 93 86 94 // Hydrate notifications 87 87 - notifications := make([]interface{}, 0) 95 95 + notifications := make([]*bsky.NotificationListNotifications_Notification, 0) 88 96 for _, row := range rows { 89 97 authorInfo, err := hydrator.HydrateActor(ctx, row.AuthorDid) 90 98 if err != nil { 91 99 continue 92 100 } 93 101 94 94 - notif := map[string]interface{}{ 95 95 - "uri": row.Source, 96 96 - "author": views.ProfileView(authorInfo), 97 97 - "reason": mapNotifKind(row.Kind), 98 98 - "record": nil, // Could hydrate the source record here 99 99 - "isRead": false, 100 100 - "indexedAt": row.CreatedAt, 101 101 - "labels": []interface{}{}, 102 102 + // Skip notifications without CIDs as they're invalid 103 103 + if row.SourceCid == "" { 104 104 + continue 102 105 } 103 106 104 104 - // Only include CID if we have one (required field) 105 105 - if row.SourceCid != "" { 106 106 - notif["cid"] = row.SourceCid 107 107 - } else { 108 108 - // Skip notifications without CIDs as they're invalid 107 107 + // Fetch and decode the raw record 108 108 + recordDecoder, err := fetchNotificationRecord(db, row.Source, row.Kind) 109 109 + if err != nil { 109 110 continue 110 111 } 111 112 113 113 + notif := &bsky.NotificationListNotifications_Notification{ 114 114 + Uri: row.Source, 115 115 + Cid: row.SourceCid, 116 116 + Author: views.ProfileView(authorInfo), 117 117 + Reason: mapNotifKind(row.Kind), 118 118 + Record: recordDecoder, 119 119 + IsRead: false, 120 120 + IndexedAt: row.CreatedAt, 121 121 + } 122 122 + 112 123 notifications = append(notifications, notif) 113 124 } 114 125 115 126 // Generate next cursor 116 116 - var nextCursor string 127 127 + var cursorPtr *string 117 128 if len(rows) > 0 { 118 118 - nextCursor = strconv.FormatUint(uint64(rows[len(rows)-1].ID), 10) 129 129 + cursor := strconv.FormatUint(uint64(rows[len(rows)-1].ID), 10) 130 130 + cursorPtr = &cursor 131 131 + } 132 132 + 133 133 + output := &bsky.NotificationListNotifications_Output{ 134 134 + Notifications: notifications, 135 135 + Cursor: cursorPtr, 119 136 } 120 137 121 121 - return c.JSON(http.StatusOK, map[string]interface{}{ 122 122 - "notifications": notifications, 123 123 - "cursor": nextCursor, 124 124 - }) 138 138 + return c.JSON(http.StatusOK, output) 125 139 } 126 140 127 141 // HandleGetUnreadCount implements app.bsky.notification.getUnreadCount ··· 175 189 return "repost" 176 190 case "mention": 177 191 return "mention" 192 192 + case "follow": 193 193 + return "follow" 178 194 default: 179 195 return kind 180 196 } 181 197 } 198 198 + 199 199 + // fetchNotificationRecord fetches and decodes the raw record for a notification 200 200 + func fetchNotificationRecord(db *gorm.DB, sourceURI string, kind string) (*util.LexiconTypeDecoder, error) { 201 201 + // Parse the source URI to extract DID and rkey 202 202 + // URI format: at://did:plc:xxx/collection/rkey 203 203 + did := extractDIDFromURI(sourceURI) 204 204 + rkey := extractRkeyFromURI(sourceURI) 205 205 + 206 206 + if did == "" || rkey == "" { 207 207 + return nil, fmt.Errorf("invalid source URI") 208 208 + } 209 209 + 210 210 + var raw []byte 211 211 + var err error 212 212 + 213 213 + // Fetch raw data based on notification kind 214 214 + switch kind { 215 215 + case "reply", "mention", "quote": 216 216 + // These reference posts 217 217 + err = db.Raw(` 218 218 + SELECT p.raw 219 219 + FROM posts p 220 220 + JOIN repos r ON r.id = p.author 221 221 + WHERE r.did = ? AND p.rkey = ? 222 222 + `, did, rkey).Scan(&raw).Error 223 223 + 224 224 + case "like": 225 225 + // we don't store the raw like objects, so we just reconstruct it here... 226 226 + // These reference like records 227 227 + var like models.Like 228 228 + err = db.Raw(` 229 229 + SELECT * 230 230 + FROM likes l 231 231 + JOIN repos r ON r.id = l.author 232 232 + WHERE r.did = ? AND l.rkey = ? 233 233 + `, did, rkey).Scan(&like).Error 234 234 + 235 235 + lk := bsky.FeedLike{ 236 236 + CreatedAt: like.Created.Format(time.RFC3339), 237 237 + Subject: &atproto.RepoStrongRef{ 238 238 + Cid: "", 239 239 + Uri: "", 240 240 + }, 241 241 + } 242 242 + buf := new(bytes.Buffer) 243 243 + if err := lk.MarshalCBOR(buf); err != nil { 244 244 + return nil, fmt.Errorf("failed to marshal reconstructed like: %w", err) 245 245 + } 246 246 + raw = buf.Bytes() 247 247 + 248 248 + case "repost": 249 249 + // These reference repost records 250 250 + err = db.Raw(` 251 251 + SELECT r.raw 252 252 + FROM reposts r 253 253 + JOIN repos repo ON repo.id = r.author 254 254 + WHERE repo.did = ? AND r.rkey = ? 255 255 + `, did, rkey).Scan(&raw).Error 256 256 + 257 257 + case "follow": 258 258 + // These reference follow records 259 259 + err = db.Raw(` 260 260 + SELECT f.raw 261 261 + FROM follows f 262 262 + JOIN repos r ON r.id = f.author 263 263 + WHERE r.did = ? AND f.rkey = ? 264 264 + `, did, rkey).Scan(&raw).Error 265 265 + 266 266 + default: 267 267 + return nil, fmt.Errorf("unknown notification kind: %s", kind) 268 268 + } 269 269 + 270 270 + if err != nil || len(raw) == 0 { 271 271 + return nil, fmt.Errorf("failed to fetch record: %w", err) 272 272 + } 273 273 + 274 274 + // Decode the CBOR data 275 275 + decoded, err := lexutil.CborDecodeValue(raw) 276 276 + if err != nil { 277 277 + return nil, fmt.Errorf("failed to decode CBOR: %w", err) 278 278 + } 279 279 + 280 280 + return &util.LexiconTypeDecoder{ 281 281 + Val: decoded, 282 282 + }, nil 283 283 + } 284 284 + 285 285 + func extractDIDFromURI(uri string) string { 286 286 + // URI format: at://did:plc:xxx/collection/rkey 287 287 + if len(uri) < 5 || uri[:5] != "at://" { 288 288 + return "" 289 289 + } 290 290 + parts := []rune(uri[5:]) 291 291 + for i, r := range parts { 292 292 + if r == '/' { 293 293 + return string(parts[:i]) 294 294 + } 295 295 + } 296 296 + return string(parts) 297 297 + } 298 298 + 299 299 + func extractRkeyFromURI(uri string) string { 300 300 + // URI format: at://did:plc:xxx/collection/rkey 301 301 + if len(uri) < 5 || uri[:5] != "at://" { 302 302 + return "" 303 303 + } 304 304 + // Find last slash 305 305 + for i := len(uri) - 1; i >= 5; i-- { 306 306 + if uri[i] == '/' { 307 307 + return uri[i+1:] 308 308 + } 309 309 + } 310 310 + return "" 311 311 + }

+11

xrpc/server.go

··· 32 32 // Add methods as needed for data access 33 33 34 34 TrackMissingActor(did string) 35 35 + TrackMissingFeedGenerator(uri string) 35 36 } 36 37 37 38 // NewServer creates a new XRPC server ··· 60 61 } 61 62 62 63 s.hydrator.SetMissingActorCallback(backend.TrackMissingActor) 64 64 + s.hydrator.SetMissingFeedGeneratorCallback(backend.TrackMissingFeedGenerator) 63 65 64 66 // Register XRPC endpoints 65 67 s.registerEndpoints() ··· 124 126 xrpcGroup.GET("/app.bsky.feed.getActorLikes", func(c echo.Context) error { 125 127 return feed.HandleGetActorLikes(c, s.db, s.hydrator) 126 128 }, s.requireAuth) 129 129 + xrpcGroup.GET("/app.bsky.feed.getFeed", func(c echo.Context) error { 130 130 + return feed.HandleGetFeed(c, s.db, s.hydrator, s.dir) 131 131 + }) 132 132 + xrpcGroup.GET("/app.bsky.feed.getFeedGenerator", func(c echo.Context) error { 133 133 + return feed.HandleGetFeedGenerator(c, s.db, s.hydrator, s.dir) 134 134 + }) 127 135 128 136 // app.bsky.graph.* 129 137 xrpcGroup.GET("/app.bsky.graph.getFollows", func(c echo.Context) error { ··· 166 174 }) 167 175 xrpcGroup.GET("/app.bsky.unspecced.getTrendingTopics", func(c echo.Context) error { 168 176 return unspecced.HandleGetTrendingTopics(c) 177 177 + }) 178 178 + xrpcGroup.GET("/app.bsky.unspecced.getPostThreadV2", func(c echo.Context) error { 179 179 + return unspecced.HandleGetPostThreadV2(c, s.db, s.hydrator) 169 180 }) 170 181 } 171 182

+333

xrpc/unspecced/getPostThreadV2.go

··· 1 1 + package unspecced 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "fmt" 6 6 + "log/slog" 7 7 + "net/http" 8 8 + "strconv" 9 9 + 10 10 + "github.com/bluesky-social/indigo/api/bsky" 11 11 + "github.com/labstack/echo/v4" 12 12 + "github.com/whyrusleeping/konbini/hydration" 13 13 + "github.com/whyrusleeping/konbini/views" 14 14 + "gorm.io/gorm" 15 15 + ) 16 16 + 17 17 + // HandleGetPostThreadV2 implements app.bsky.unspecced.getPostThreadV2 18 18 + func HandleGetPostThreadV2(c echo.Context, db *gorm.DB, hydrator *hydration.Hydrator) error { 19 19 + ctx := c.Request().Context() 20 20 + 21 21 + // Parse parameters 22 22 + anchorRaw := c.QueryParam("anchor") 23 23 + if anchorRaw == "" { 24 24 + return c.JSON(http.StatusBadRequest, map[string]interface{}{ 25 25 + "error": "InvalidRequest", 26 26 + "message": "anchor parameter is required", 27 27 + }) 28 28 + } 29 29 + 30 30 + anchorUri, err := hydrator.NormalizeUri(ctx, anchorRaw) 31 31 + if err != nil { 32 32 + return err 33 33 + } 34 34 + 35 35 + // Parse optional parameters with defaults 36 36 + above := c.QueryParam("above") != "false" // default true 37 37 + 38 38 + below := int64(6) // default 39 39 + if belowParam := c.QueryParam("below"); belowParam != "" { 40 40 + if b, err := strconv.ParseInt(belowParam, 10, 64); err == nil && b >= 0 && b <= 20 { 41 41 + below = b 42 42 + } 43 43 + } 44 44 + 45 45 + branchingFactor := int64(10) // default 46 46 + if bfParam := c.QueryParam("branchingFactor"); bfParam != "" { 47 47 + if bf, err := strconv.ParseInt(bfParam, 10, 64); err == nil && bf > 0 { 48 48 + branchingFactor = bf 49 49 + } 50 50 + } 51 51 + 52 52 + _ = c.QueryParam("prioritizeFollowedUsers") == "true" // TODO: implement prioritization 53 53 + 54 54 + sort := c.QueryParam("sort") 55 55 + if sort == "" { 56 56 + sort = "newest" 57 57 + } 58 58 + 59 59 + viewer := getUserDID(c) 60 60 + 61 61 + // Hydrate the anchor post 62 62 + anchorPostInfo, err := hydrator.HydratePost(ctx, anchorUri, viewer) 63 63 + if err != nil { 64 64 + slog.Error("failed to hydrate post", "error", err, "anchor", anchorUri) 65 65 + return c.JSON(http.StatusNotFound, map[string]interface{}{ 66 66 + "error": "NotFound", 67 67 + "message": "anchor post not found", 68 68 + }) 69 69 + } 70 70 + 71 71 + // Determine the root post ID for the thread 72 72 + rootPostID := anchorPostInfo.InThread 73 73 + if rootPostID == 0 { 74 74 + // This post is the root - get its ID 75 75 + var postID uint 76 76 + db.Raw(` 77 77 + SELECT id FROM posts 78 78 + WHERE author = (SELECT id FROM repos WHERE did = ?) 79 79 + AND rkey = ? 80 80 + `, extractDIDFromURI(anchorUri), extractRkeyFromURI(anchorUri)).Scan(&postID) 81 81 + rootPostID = postID 82 82 + } 83 83 + 84 84 + // Query all posts in this thread 85 85 + type threadPostRow struct { 86 86 + ID uint 87 87 + Rkey string 88 88 + ReplyTo uint 89 89 + InThread uint 90 90 + AuthorDid string 91 91 + } 92 92 + var threadPosts []threadPostRow 93 93 + db.Raw(` 94 94 + SELECT p.id, p.rkey, p.reply_to, p.in_thread, r.did as author_did 95 95 + FROM posts p 96 96 + JOIN repos r ON r.id = p.author 97 97 + WHERE (p.id = ? OR p.in_thread = ?) 98 98 + AND p.not_found = false 99 99 + ORDER BY p.created ASC 100 100 + `, rootPostID, rootPostID).Scan(&threadPosts) 101 101 + 102 102 + // Build a map of posts by ID 103 103 + postsByID := make(map[uint]*threadNode) 104 104 + for _, tp := range threadPosts { 105 105 + uri := fmt.Sprintf("at://%s/app.bsky.feed.post/%s", tp.AuthorDid, tp.Rkey) 106 106 + postsByID[tp.ID] = &threadNode{ 107 107 + id: tp.ID, 108 108 + uri: uri, 109 109 + replyTo: tp.ReplyTo, 110 110 + inThread: tp.InThread, 111 111 + children: []*threadNode{}, 112 112 + } 113 113 + } 114 114 + 115 115 + // Build parent-child relationships 116 116 + for _, node := range postsByID { 117 117 + if node.replyTo != 0 { 118 118 + parent := postsByID[node.replyTo] 119 119 + if parent != nil { 120 120 + parent.children = append(parent.children, node) 121 121 + } 122 122 + } 123 123 + } 124 124 + 125 125 + // Find the anchor node 126 126 + anchorID := uint(0) 127 127 + for id, node := range postsByID { 128 128 + if node.uri == anchorUri { 129 129 + anchorID = id 130 130 + break 131 131 + } 132 132 + } 133 133 + 134 134 + if anchorID == 0 { 135 135 + return c.JSON(http.StatusNotFound, map[string]interface{}{ 136 136 + "error": "NotFound", 137 137 + "message": "anchor post not found in thread", 138 138 + }) 139 139 + } 140 140 + 141 141 + anchorNode := postsByID[anchorID] 142 142 + 143 143 + // Build flat thread items list 144 144 + var threadItems []*bsky.UnspeccedGetPostThreadV2_ThreadItem 145 145 + hasOtherReplies := false 146 146 + 147 147 + // Add parents if requested 148 148 + if above { 149 149 + parents := collectParents(anchorNode, postsByID) 150 150 + for i := len(parents) - 1; i >= 0; i-- { 151 151 + depth := int64(-(len(parents) - i)) 152 152 + item := buildThreadItem(ctx, hydrator, parents[i], depth, viewer) 153 153 + if item != nil { 154 154 + threadItems = append(threadItems, item) 155 155 + } 156 156 + } 157 157 + } 158 158 + 159 159 + // Add anchor post (depth 0) 160 160 + anchorItem := buildThreadItem(ctx, hydrator, anchorNode, 0, viewer) 161 161 + if anchorItem != nil { 162 162 + threadItems = append(threadItems, anchorItem) 163 163 + } 164 164 + 165 165 + // Add replies below anchor 166 166 + if below > 0 { 167 167 + replies, hasMore := collectReplies(ctx, hydrator, anchorNode, 1, below, branchingFactor, sort, viewer) 168 168 + threadItems = append(threadItems, replies...) 169 169 + hasOtherReplies = hasMore 170 170 + } 171 171 + 172 172 + return c.JSON(http.StatusOK, &bsky.UnspeccedGetPostThreadV2_Output{ 173 173 + Thread: threadItems, 174 174 + HasOtherReplies: hasOtherReplies, 175 175 + }) 176 176 + } 177 177 + 178 178 + type threadNode struct { 179 179 + id uint 180 180 + uri string 181 181 + replyTo uint 182 182 + inThread uint 183 183 + children []*threadNode 184 184 + } 185 185 + 186 186 + func collectParents(node *threadNode, allNodes map[uint]*threadNode) []*threadNode { 187 187 + var parents []*threadNode 188 188 + current := node 189 189 + for current.replyTo != 0 { 190 190 + parent := allNodes[current.replyTo] 191 191 + if parent == nil { 192 192 + break 193 193 + } 194 194 + parents = append(parents, parent) 195 195 + current = parent 196 196 + } 197 197 + return parents 198 198 + } 199 199 + 200 200 + func collectReplies(ctx context.Context, hydrator *hydration.Hydrator, node *threadNode, currentDepth, maxDepth, branchingFactor int64, sort string, viewer string) ([]*bsky.UnspeccedGetPostThreadV2_ThreadItem, bool) { 201 201 + var items []*bsky.UnspeccedGetPostThreadV2_ThreadItem 202 202 + hasMore := false 203 203 + 204 204 + if currentDepth > maxDepth { 205 205 + return items, false 206 206 + } 207 207 + 208 208 + // Sort children based on sort parameter 209 209 + children := node.children 210 210 + // TODO: Actually sort based on the sort parameter (newest/oldest/top) 211 211 + // For now, just use the order we have 212 212 + 213 213 + // Limit to branchingFactor 214 214 + limit := int(branchingFactor) 215 215 + if len(children) > limit { 216 216 + hasMore = true 217 217 + children = children[:limit] 218 218 + } 219 219 + 220 220 + for _, child := range children { 221 221 + item := buildThreadItem(ctx, hydrator, child, currentDepth, viewer) 222 222 + if item != nil { 223 223 + items = append(items, item) 224 224 + 225 225 + // Recursively collect replies 226 226 + if currentDepth < maxDepth { 227 227 + childReplies, childHasMore := collectReplies(ctx, hydrator, child, currentDepth+1, maxDepth, branchingFactor, sort, viewer) 228 228 + items = append(items, childReplies...) 229 229 + if childHasMore { 230 230 + hasMore = true 231 231 + } 232 232 + } 233 233 + } 234 234 + } 235 235 + 236 236 + return items, hasMore 237 237 + } 238 238 + 239 239 + func buildThreadItem(ctx context.Context, hydrator *hydration.Hydrator, node *threadNode, depth int64, viewer string) *bsky.UnspeccedGetPostThreadV2_ThreadItem { 240 240 + // Hydrate the post 241 241 + postInfo, err := hydrator.HydratePost(ctx, node.uri, viewer) 242 242 + if err != nil { 243 243 + // Return not found item 244 244 + return &bsky.UnspeccedGetPostThreadV2_ThreadItem{ 245 245 + Depth: depth, 246 246 + Uri: node.uri, 247 247 + Value: &bsky.UnspeccedGetPostThreadV2_ThreadItem_Value{ 248 248 + UnspeccedDefs_ThreadItemNotFound: &bsky.UnspeccedDefs_ThreadItemNotFound{ 249 249 + LexiconTypeID: "app.bsky.unspecced.defs#threadItemNotFound", 250 250 + }, 251 251 + }, 252 252 + } 253 253 + } 254 254 + 255 255 + // Hydrate author 256 256 + authorInfo, err := hydrator.HydrateActor(ctx, postInfo.Author) 257 257 + if err != nil { 258 258 + return &bsky.UnspeccedGetPostThreadV2_ThreadItem{ 259 259 + Depth: depth, 260 260 + Uri: node.uri, 261 261 + Value: &bsky.UnspeccedGetPostThreadV2_ThreadItem_Value{ 262 262 + UnspeccedDefs_ThreadItemNotFound: &bsky.UnspeccedDefs_ThreadItemNotFound{ 263 263 + LexiconTypeID: "app.bsky.unspecced.defs#threadItemNotFound", 264 264 + }, 265 265 + }, 266 266 + } 267 267 + } 268 268 + 269 269 + // Build post view 270 270 + postView := views.PostView(postInfo, authorInfo) 271 271 + 272 272 + // Calculate moreReplies count 273 273 + moreReplies := int64(0) 274 274 + if len(node.children) > 0 { 275 275 + // This is a simplified calculation - actual count would need more complex logic 276 276 + moreReplies = int64(len(node.children)) 277 277 + } 278 278 + 279 279 + return &bsky.UnspeccedGetPostThreadV2_ThreadItem{ 280 280 + Depth: depth, 281 281 + Uri: node.uri, 282 282 + Value: &bsky.UnspeccedGetPostThreadV2_ThreadItem_Value{ 283 283 + UnspeccedDefs_ThreadItemPost: &bsky.UnspeccedDefs_ThreadItemPost{ 284 284 + LexiconTypeID: "app.bsky.unspecced.defs#threadItemPost", 285 285 + Post: postView, 286 286 + HiddenByThreadgate: false, 287 287 + MoreParents: false, 288 288 + MoreReplies: moreReplies, 289 289 + MutedByViewer: false, 290 290 + OpThread: false, // TODO: Calculate this properly 291 291 + }, 292 292 + }, 293 293 + } 294 294 + } 295 295 + 296 296 + func getUserDID(c echo.Context) string { 297 297 + did := c.Get("viewer") 298 298 + if did == nil { 299 299 + return "" 300 300 + } 301 301 + if s, ok := did.(string); ok { 302 302 + return s 303 303 + } 304 304 + return "" 305 305 + } 306 306 + 307 307 + func extractDIDFromURI(uri string) string { 308 308 + // URI format: at://did:plc:xxx/collection/rkey 309 309 + if len(uri) < 5 || uri[:5] != "at://" { 310 310 + return "" 311 311 + } 312 312 + parts := []rune(uri[5:]) 313 313 + for i, r := range parts { 314 314 + if r == '/' { 315 315 + return string(parts[:i]) 316 316 + } 317 317 + } 318 318 + return string(parts) 319 319 + } 320 320 + 321 321 + func extractRkeyFromURI(uri string) string { 322 322 + // URI format: at://did:plc:xxx/collection/rkey 323 323 + if len(uri) < 5 || uri[:5] != "at://" { 324 324 + return "" 325 325 + } 326 326 + // Find last slash 327 327 + for i := len(uri) - 1; i >= 5; i-- { 328 328 + if uri[i] == '/' { 329 329 + return uri[i+1:] 330 330 + } 331 331 + } 332 332 + return "" 333 333 + }