-431

pds/feedgen.go

reviewed

··· 1 1 - package pds 2 2 - 3 3 - import ( 4 4 - "context" 5 5 - "fmt" 6 6 - "log/slog" 7 7 - "sort" 8 8 - "strings" 9 9 - "time" 10 10 - 11 11 - bsky "github.com/bluesky-social/indigo/api/bsky" 12 12 - "github.com/bluesky-social/indigo/indexer" 13 13 - lexutil "github.com/bluesky-social/indigo/lex/util" 14 14 - "github.com/bluesky-social/indigo/models" 15 15 - 16 16 - "github.com/ipfs/go-cid" 17 17 - "go.opentelemetry.io/otel" 18 18 - "gorm.io/gorm" 19 19 - ) 20 20 - 21 21 - type FeedGenerator struct { 22 22 - db *gorm.DB 23 23 - ix *indexer.Indexer 24 24 - 25 25 - readRecord ReadRecordFunc 26 26 - 27 27 - log *slog.Logger 28 28 - } 29 29 - 30 30 - func NewFeedGenerator(db *gorm.DB, ix *indexer.Indexer, readRecord ReadRecordFunc, log *slog.Logger) (*FeedGenerator, error) { 31 31 - return &FeedGenerator{ 32 32 - db: db, 33 33 - ix: ix, 34 34 - readRecord: readRecord, 35 35 - log: log, 36 36 - }, nil 37 37 - } 38 38 - 39 39 - type ReadRecordFunc func(context.Context, models.Uid, cid.Cid) (lexutil.CBOR, error) 40 40 - 41 41 - /* 42 42 - type HydratedFeedItem struct { 43 43 - Uri string 44 44 - RepostedBy *bsky.ActorDefs_ProfileViewBasic 45 45 - Record any 46 46 - ReplyCount int64 47 47 - RepostCount int64 48 48 - UpvoteCount int64 49 49 - DownvoteCount int64 50 50 - MyState *bsky.FeedGetAuthorFeed_MyState 51 51 - Cid string 52 52 - Author *bsky.ActorDefs_ProfileViewBasic 53 53 - TrendedBy *bsky.ActorDefs_ProfileViewBasic 54 54 - Embed *bsky.FeedEmbed 55 55 - IndexedAt string 56 56 - } 57 57 - */ 58 58 - 59 59 - func (fg *FeedGenerator) hydrateFeed(ctx context.Context, items []*models.FeedPost, reposts []*models.RepostRecord) ([]*bsky.FeedDefs_FeedViewPost, error) { 60 60 - out := make([]*bsky.FeedDefs_FeedViewPost, 0, len(items)) 61 61 - for _, it := range items { 62 62 - hit, err := fg.hydrateItem(ctx, it) 63 63 - if err != nil { 64 64 - return nil, err 65 65 - } 66 66 - 67 67 - out = append(out, hit) 68 68 - } 69 69 - 70 70 - if len(reposts) > 0 { 71 71 - for _, rp := range reposts { 72 72 - var fp models.FeedPost 73 73 - if err := fg.db.First(&fp, "id = ?", rp.Post).Error; err != nil { 74 74 - return nil, err 75 75 - } 76 76 - 77 77 - fvp, err := fg.hydrateItem(ctx, &fp) 78 78 - if err != nil { 79 79 - return nil, err 80 80 - } 81 81 - 82 82 - reposter, err := fg.getActorRefInfo(ctx, rp.Reposter) 83 83 - if err != nil { 84 84 - return nil, err 85 85 - } 86 86 - 87 87 - fvp.Reason = &bsky.FeedDefs_FeedViewPost_Reason{ 88 88 - FeedDefs_ReasonRepost: &bsky.FeedDefs_ReasonRepost{ 89 89 - By: reposter, 90 90 - IndexedAt: rp.CreatedAt.Format(time.RFC3339), 91 91 - }, 92 92 - } 93 93 - 94 94 - out = append(out, fvp) 95 95 - } 96 96 - } 97 97 - 98 98 - return out, nil 99 99 - } 100 100 - 101 101 - func (fg *FeedGenerator) didForUser(ctx context.Context, user models.Uid) (string, error) { 102 102 - // TODO: cache the shit out of this 103 103 - var ai models.ActorInfo 104 104 - if err := fg.db.First(&ai, "uid = ?", user).Error; err != nil { 105 105 - return "", err 106 106 - } 107 107 - 108 108 - return ai.Did, nil 109 109 - } 110 110 - 111 111 - func (fg *FeedGenerator) getActorRefInfo(ctx context.Context, user models.Uid) (*bsky.ActorDefs_ProfileViewBasic, error) { 112 112 - // TODO: cache the shit out of this too 113 113 - var ai models.ActorInfo 114 114 - if err := fg.db.First(&ai, "uid = ?", user).Error; err != nil { 115 115 - return nil, err 116 116 - } 117 117 - 118 118 - return ai.ActorRef(), nil 119 119 - } 120 120 - 121 121 - func (fg *FeedGenerator) hydrateItem(ctx context.Context, item *models.FeedPost) (*bsky.FeedDefs_FeedViewPost, error) { 122 122 - authorDid, err := fg.didForUser(ctx, item.Author) 123 123 - if err != nil { 124 124 - return nil, err 125 125 - } 126 126 - 127 127 - out := &bsky.FeedDefs_FeedViewPost{} 128 128 - 129 129 - out.Post = &bsky.FeedDefs_PostView{ 130 130 - Uri: "at://" + authorDid + "/app.bsky.feed.post/" + item.Rkey, 131 131 - ReplyCount: &item.ReplyCount, 132 132 - RepostCount: &item.RepostCount, 133 133 - LikeCount: &item.UpCount, 134 134 - Cid: item.Cid, 135 135 - IndexedAt: item.UpdatedAt.Format(time.RFC3339), 136 136 - } 137 137 - 138 138 - author, err := fg.getActorRefInfo(ctx, item.Author) 139 139 - if err != nil { 140 140 - return nil, err 141 141 - } 142 142 - 143 143 - out.Post.Author = author 144 144 - 145 145 - reccid, err := cid.Decode(item.Cid) 146 146 - if err != nil { 147 147 - return nil, err 148 148 - } 149 149 - 150 150 - rec, err := fg.readRecord(ctx, item.Author, reccid) 151 151 - if err != nil { 152 152 - return nil, err 153 153 - } 154 154 - 155 155 - out.Post.Record = &lexutil.LexiconTypeDecoder{Val: rec} 156 156 - 157 157 - return out, nil 158 158 - } 159 159 - 160 160 - func (fg *FeedGenerator) getPostViewerState(ctx context.Context, item uint, viewer models.Uid, viewerDid string) (*bsky.FeedDefs_ViewerState, error) { 161 161 - var out bsky.FeedDefs_ViewerState 162 162 - 163 163 - var vote models.VoteRecord 164 164 - if err := fg.db.Find(&vote, "post = ? AND voter = ?", item, viewer).Error; err != nil { 165 165 - return nil, err 166 166 - } 167 167 - 168 168 - if vote.ID != 0 { 169 169 - vuri := fmt.Sprintf("at://%s/app.bsky.feed.vote/%s", viewerDid, vote.Rkey) 170 170 - out.Like = &vuri 171 171 - } 172 172 - 173 173 - var rep models.RepostRecord 174 174 - if err := fg.db.Find(&rep, "post = ? AND reposter = ?", item, viewer).Error; err != nil { 175 175 - return nil, err 176 176 - } 177 177 - 178 178 - if rep.ID != 0 { 179 179 - rpuri := fmt.Sprintf("at://%s/app.bsky.feed.repost/%s", viewerDid, rep.Rkey) 180 180 - out.Repost = &rpuri 181 181 - } 182 182 - 183 183 - return &out, nil 184 184 - } 185 185 - 186 186 - func (fg *FeedGenerator) GetTimeline(ctx context.Context, user *User, algo string, before string, limit int) ([]*bsky.FeedDefs_FeedViewPost, error) { 187 187 - ctx, span := otel.Tracer("feedgen").Start(context.Background(), "GetTimeline") 188 188 - defer span.End() 189 189 - 190 190 - // TODO: this query is just a temporary hack... 191 191 - var feed []*models.FeedPost 192 192 - if err := fg.db.Debug().Find(&feed, "author in (?)", 193 193 - fg.db.Model(models.FollowRecord{}).Where("follower = ?", user.ID).Select("target"), 194 194 - ).Error; err != nil { 195 195 - return nil, err 196 196 - } 197 197 - 198 198 - var rps []*models.RepostRecord 199 199 - if err := fg.db.Debug().Find(&rps, "reposter in (?)", 200 200 - fg.db.Model(models.FollowRecord{}).Where("follower = ?", user.ID).Select("target"), 201 201 - ).Error; err != nil { 202 202 - return nil, err 203 203 - } 204 204 - 205 205 - fout, err := fg.hydrateFeed(ctx, feed, rps) 206 206 - if err != nil { 207 207 - return nil, fmt.Errorf("hydrating feed: %w", err) 208 208 - } 209 209 - 210 210 - sort.Slice(fout, func(i, j int) bool { 211 211 - ti, _ := time.Parse(time.RFC3339, fout[i].Post.IndexedAt) 212 212 - tj, _ := time.Parse(time.RFC3339, fout[j].Post.IndexedAt) 213 213 - 214 214 - return tj.Before(ti) 215 215 - }) 216 216 - 217 217 - return fg.personalizeFeed(ctx, fout, user) 218 218 - } 219 219 - 220 220 - func (fg *FeedGenerator) personalizeFeed(ctx context.Context, feed []*bsky.FeedDefs_FeedViewPost, viewer *User) ([]*bsky.FeedDefs_FeedViewPost, error) { 221 221 - for _, p := range feed { 222 222 - 223 223 - // TODO: its inefficient to have to call 'GetPost' again here when we could instead be doing that inside the 'hydrateFeed' call earlier. 224 224 - // However, if we introduce per-user information into hydrateFeed, then 225 225 - // it cannot be effectively cached, so we separate the 'cacheable' 226 226 - // portion of feed generation from the 'per user' portion. An 227 227 - // optimization could be to hide the internal post IDs in the Post 228 228 - // structs for internal use (stripped out before sending to client) 229 229 - item, err := fg.ix.GetPost(ctx, p.Post.Uri) 230 230 - if err != nil { 231 231 - return nil, err 232 232 - } 233 233 - 234 234 - vs, err := fg.getPostViewerState(ctx, item.ID, viewer.ID, viewer.Did) 235 235 - if err != nil { 236 236 - return nil, fmt.Errorf("getting viewer state: %w", err) 237 237 - } 238 238 - 239 239 - p.Post.Viewer = vs 240 240 - } 241 241 - 242 242 - return feed, nil 243 243 - } 244 244 - 245 245 - func (fg *FeedGenerator) GetAuthorFeed(ctx context.Context, user *User, before string, limit int) ([]*bsky.FeedDefs_FeedViewPost, error) { 246 246 - ctx, span := otel.Tracer("feedgen").Start(context.Background(), "GetAuthorFeed") 247 247 - defer span.End() 248 248 - 249 249 - // for memory efficiency, should probably return the actual type that goes out to the user... 250 250 - // bsky.FeedGetAuthorFeed_FeedItem 251 251 - 252 252 - var feed []*models.FeedPost 253 253 - if err := fg.db.Find(&feed, "author = ?", user.ID).Error; err != nil { 254 254 - return nil, err 255 255 - } 256 256 - 257 257 - var reposts []*models.RepostRecord 258 258 - if err := fg.db.Find(&reposts, "reposter = ?", user.ID).Error; err != nil { 259 259 - return nil, err 260 260 - } 261 261 - 262 262 - fout, err := fg.hydrateFeed(ctx, feed, reposts) 263 263 - if err != nil { 264 264 - return nil, fmt.Errorf("hydrating feed: %w", err) 265 265 - } 266 266 - 267 267 - return fg.personalizeFeed(ctx, fout, user) 268 268 - } 269 269 - 270 270 - func (fg *FeedGenerator) GetActorProfileByID(ctx context.Context, actor uint) (*models.ActorInfo, error) { 271 271 - var ai models.ActorInfo 272 272 - if err := fg.db.First(&ai, "id = ?", actor).Error; err != nil { 273 273 - return nil, fmt.Errorf("getActorProfileByID: %w", err) 274 274 - } 275 275 - 276 276 - return &ai, nil 277 277 - } 278 278 - func (fg *FeedGenerator) GetActorProfile(ctx context.Context, actor string) (*models.ActorInfo, error) { 279 279 - fmt.Println("get actor profile: ", actor) 280 280 - var ai models.ActorInfo 281 281 - if strings.HasPrefix(actor, "did:") { 282 282 - if err := fg.db.First(&ai, "did = ?", actor).Error; err != nil { 283 283 - return nil, err 284 284 - } 285 285 - } else { 286 286 - if err := fg.db.First(&ai, "handle = ?", actor).Error; err != nil { 287 287 - return nil, err 288 288 - } 289 289 - } 290 290 - 291 291 - return &ai, nil 292 292 - } 293 293 - 294 294 - type ThreadPost struct { 295 295 - Post *bsky.FeedDefs_FeedViewPost 296 296 - PostID uint 297 297 - 298 298 - ParentUri string 299 299 - Parent *ThreadPost 300 300 - } 301 301 - 302 302 - func (fg *FeedGenerator) GetPostThread(ctx context.Context, uri string, depth int) (*ThreadPost, error) { 303 303 - post, err := fg.ix.GetPost(ctx, uri) 304 304 - if err != nil { 305 305 - return nil, fmt.Errorf("getting post for thread: %w", err) 306 306 - } 307 307 - 308 308 - hi, err := fg.hydrateItem(ctx, post) 309 309 - if err != nil { 310 310 - return nil, err 311 311 - } 312 312 - 313 313 - p, ok := hi.Post.Record.Val.(*bsky.FeedPost) 314 314 - if !ok { 315 315 - return nil, fmt.Errorf("getPostThread can only operate on app.bsky.feed.post records") 316 316 - } 317 317 - 318 318 - out := &ThreadPost{ 319 319 - Post: hi, 320 320 - PostID: post.ID, 321 321 - } 322 322 - 323 323 - if p.Reply != nil { 324 324 - out.ParentUri = p.Reply.Parent.Uri 325 325 - if depth > 0 { 326 326 - 327 327 - parent, err := fg.GetPostThread(ctx, p.Reply.Parent.Uri, depth-1) 328 328 - if err != nil { 329 329 - // TODO: check for and handle 'not found' 330 330 - return nil, err 331 331 - } 332 332 - out.Parent = parent 333 333 - } 334 334 - } 335 335 - 336 336 - return out, nil 337 337 - } 338 338 - 339 339 - type HydratedVote struct { 340 340 - Actor *bsky.ActorDefs_ProfileViewBasic 341 341 - Direction string 342 342 - IndexedAt time.Time 343 343 - CreatedAt string 344 344 - } 345 345 - 346 346 - func (fg *FeedGenerator) hydrateVote(ctx context.Context, v *models.VoteRecord) (*HydratedVote, error) { 347 347 - aref, err := fg.getActorRefInfo(ctx, v.Voter) 348 348 - if err != nil { 349 349 - return nil, err 350 350 - } 351 351 - 352 352 - return &HydratedVote{ 353 353 - Actor: aref, 354 354 - Direction: v.Dir.String(), 355 355 - IndexedAt: v.UpdatedAt, 356 356 - CreatedAt: v.Created, 357 357 - }, nil 358 358 - } 359 359 - 360 360 - func (fg *FeedGenerator) GetVotes(ctx context.Context, uri string, pcid cid.Cid, limit int, before string) ([]*HydratedVote, error) { 361 361 - if before != "" { 362 362 - fg.log.Warn("not respecting 'before' yet") 363 363 - } 364 364 - 365 365 - p, err := fg.ix.GetPost(ctx, uri) 366 366 - if err != nil { 367 367 - return nil, err 368 368 - } 369 369 - 370 370 - if p.Cid != pcid.String() { 371 371 - return nil, fmt.Errorf("listing likes of old post versions not supported") 372 372 - } 373 373 - 374 374 - var voterecs []models.VoteRecord 375 375 - if err := fg.db.Limit(limit).Find(&voterecs, "post = ?", p.ID).Error; err != nil { 376 376 - return nil, err 377 377 - } 378 378 - 379 379 - var out []*HydratedVote 380 380 - for _, vr := range voterecs { 381 381 - hv, err := fg.hydrateVote(ctx, &vr) 382 382 - if err != nil { 383 383 - return nil, err 384 384 - } 385 385 - out = append(out, hv) 386 386 - } 387 387 - 388 388 - return out, nil 389 389 - } 390 390 - 391 391 - type FollowInfo struct { 392 392 - Follower *bsky.ActorDefs_ProfileViewBasic 393 393 - Subject *bsky.ActorDefs_ProfileViewBasic 394 394 - CreatedAt string 395 395 - IndexedAt string 396 396 - } 397 397 - 398 398 - func (fg *FeedGenerator) GetFollows(ctx context.Context, user string, limit int, before string) ([]*FollowInfo, error) { 399 399 - var follows []models.FollowRecord 400 400 - if err := fg.db.Limit(limit).Find(&follows, "follower = (?)", fg.db.Model(models.ActorInfo{}).Where("did = ? or handle = ?", user, user).Select("uid")).Error; err != nil { 401 401 - return nil, err 402 402 - } 403 403 - 404 404 - profile, err := fg.GetActorProfile(ctx, user) 405 405 - if err != nil { 406 406 - return nil, err 407 407 - } 408 408 - 409 409 - ai, err := fg.getActorRefInfo(ctx, profile.Uid) 410 410 - if err != nil { 411 411 - return nil, err 412 412 - } 413 413 - 414 414 - out := []*FollowInfo{} 415 415 - for _, f := range follows { 416 416 - fai, err := fg.getActorRefInfo(ctx, f.Target) 417 417 - if err != nil { 418 418 - return nil, err 419 419 - } 420 420 - 421 421 - out = append(out, &FollowInfo{ 422 422 - Follower: ai, 423 423 - Subject: fai, 424 424 - CreatedAt: f.CreatedAt.Format(time.RFC3339), 425 425 - IndexedAt: f.CreatedAt.Format(time.RFC3339), 426 426 - }) 427 427 - 428 428 - } 429 429 - 430 430 - return nil, nil 431 431 - }

-8

pds/server.go

reviewed

··· 40 40 db *gorm.DB 41 41 cs carstore.CarStore 42 42 repoman *repomgr.RepoManager 43 43 - feedgen *FeedGenerator 44 43 indexer *indexer.Indexer 45 44 events *events.EventManager 46 45 signingKey *did.PrivKey ··· 104 103 105 104 //ix.SendRemoteFollow = s.sendRemoteFollow 106 105 ix.CreateExternalUser = s.createExternalUser 107 107 - 108 108 - feedgen, err := NewFeedGenerator(db, ix, s.readRecordFunc, s.log) 109 109 - if err != nil { 110 110 - return nil, err 111 111 - } 112 112 - 113 113 - s.feedgen = feedgen 114 106 115 107 return s, nil 116 108 }