+36 -7

backend/backend.go

··· 10 10 11 11 "github.com/bluesky-social/indigo/api/atproto" 12 12 "github.com/bluesky-social/indigo/api/bsky" 13 13 + "github.com/bluesky-social/indigo/atproto/identity" 13 14 "github.com/bluesky-social/indigo/atproto/syntax" 14 15 "github.com/bluesky-social/indigo/util" 15 16 "github.com/bluesky-social/indigo/xrpc" ··· 26 27 27 28 // PostgresBackend handles database operations 28 29 type PostgresBackend struct { 29 29 - db *gorm.DB 30 30 - pgx *pgxpool.Pool 31 31 - tracker RecordTracker 30 30 + db *gorm.DB 31 31 + pgx *pgxpool.Pool 32 32 + 33 33 + dir identity.Directory 32 34 33 35 client *xrpc.Client 34 36 ··· 46 48 didByIDCache *lru.TwoQueueCache[uint, string] 47 49 48 50 postInfoCache *lru.TwoQueueCache[string, cachedPostInfo] 51 51 + 52 52 + missingRecords chan MissingRecord 49 53 } 50 54 51 55 type cachedPostInfo struct { ··· 54 58 } 55 59 56 60 // NewPostgresBackend creates a new PostgresBackend 57 57 - func NewPostgresBackend(mydid string, db *gorm.DB, pgx *pgxpool.Pool, client *xrpc.Client, tracker RecordTracker) (*PostgresBackend, error) { 61 61 + func NewPostgresBackend(mydid string, db *gorm.DB, pgx *pgxpool.Pool, client *xrpc.Client, dir identity.Directory) (*PostgresBackend, error) { 58 62 rc, _ := lru.New2Q[string, *Repo](1_000_000) 59 63 pc, _ := lru.New2Q[string, cachedPostInfo](1_000_000) 60 64 revc, _ := lru.New2Q[uint, string](1_000_000) ··· 65 69 mydid: mydid, 66 70 db: db, 67 71 pgx: pgx, 68 68 - tracker: tracker, 69 72 relevantDids: make(map[string]bool), 70 73 repoCache: rc, 71 74 postInfoCache: pc, 72 75 revCache: revc, 73 76 didByIDCache: dbic, 77 77 + dir: dir, 78 78 + 79 79 + missingRecords: make(chan MissingRecord, 1000), 74 80 } 75 81 76 82 r, err := b.GetOrCreateRepo(context.TODO(), mydid) ··· 79 85 } 80 86 81 87 b.myrepo = r 88 88 + 89 89 + go b.missingRecordFetcher() 82 90 return b, nil 83 91 } 84 92 85 93 // TrackMissingRecord implements the RecordTracker interface 86 94 func (b *PostgresBackend) TrackMissingRecord(identifier string, wait bool) { 87 87 - if b.tracker != nil { 88 88 - b.tracker.TrackMissingRecord(identifier, wait) 95 95 + mr := MissingRecord{ 96 96 + Type: mrTypeFromIdent(identifier), 97 97 + Identifier: identifier, 98 98 + Wait: wait, 99 99 + } 100 100 + 101 101 + b.addMissingRecord(context.TODO(), mr) 102 102 + } 103 103 + 104 104 + func mrTypeFromIdent(ident string) MissingRecordType { 105 105 + if strings.HasPrefix(ident, "did:") { 106 106 + return MissingRecordTypeProfile 107 107 + } 108 108 + 109 109 + puri, _ := syntax.ParseATURI(ident) 110 110 + switch puri.Collection().String() { 111 111 + case "app.bsky.feed.post": 112 112 + return MissingRecordTypePost 113 113 + case "app.bsky.feed.generator": 114 114 + return MissingRecordTypeFeedGenerator 115 115 + default: 116 116 + return MissingRecordTypeUnknown 89 117 } 118 118 + 90 119 } 91 120 92 121 // DidToID converts a DID to a database ID

+8 -8

handlers.go

··· 146 146 } 147 147 148 148 if profile.Raw == nil || len(profile.Raw) == 0 { 149 149 - s.addMissingProfile(ctx, accdid) 149 149 + s.backend.TrackMissingRecord(accdid, false) 150 150 return e.JSON(404, map[string]any{ 151 151 "error": "missing profile info for user", 152 152 }) ··· 307 307 } 308 308 309 309 if profile.Raw == nil || len(profile.Raw) == 0 { 310 310 - s.addMissingProfile(ctx, r.Did) 310 310 + s.backend.TrackMissingRecord(r.Did, false) 311 311 return &authorInfo{ 312 312 Handle: resp.Handle.String(), 313 313 Did: r.Did, ··· 379 379 380 380 uri := fmt.Sprintf("at://%s/app.bsky.feed.post/%s", r.Did, p.Rkey) 381 381 if len(p.Raw) == 0 || p.NotFound { 382 382 - s.addMissingPost(ctx, uri) 382 382 + s.backend.TrackMissingRecord(uri, false) 383 383 posts[ix] = postResponse{ 384 384 Uri: uri, 385 385 Missing: true, ··· 515 515 quotedPost, err := s.backend.GetPostByUri(ctx, quotedURI, "*") 516 516 if err != nil { 517 517 slog.Warn("failed to get quoted post", "uri", quotedURI, "error", err) 518 518 - s.addMissingPost(ctx, quotedURI) 518 518 + s.backend.TrackMissingRecord(quotedURI, false) 519 519 return s.buildQuoteFallback(quotedURI, quotedCid) 520 520 } 521 521 522 522 if quotedPost == nil || quotedPost.Raw == nil || len(quotedPost.Raw) == 0 || quotedPost.NotFound { 523 523 - s.addMissingPost(ctx, quotedURI) 523 523 + s.backend.TrackMissingRecord(quotedURI, false) 524 524 return s.buildQuoteFallback(quotedURI, quotedCid) 525 525 } 526 526 ··· 707 707 prof = &p 708 708 } 709 709 } else { 710 710 - s.addMissingProfile(ctx, r.Did) 710 710 + s.backend.TrackMissingRecord(r.Did, false) 711 711 } 712 712 713 713 users = append(users, engagementUser{ ··· 767 767 prof = &p 768 768 } 769 769 } else { 770 770 - s.addMissingProfile(ctx, r.Did) 770 770 + s.backend.TrackMissingRecord(r.Did, false) 771 771 } 772 772 773 773 users = append(users, engagementUser{ ··· 835 835 prof = &p 836 836 } 837 837 } else { 838 838 - s.addMissingProfile(ctx, r.Did) 838 838 + s.backend.TrackMissingRecord(r.Did, false) 839 839 } 840 840 841 841 users = append(users, engagementUser{

+3

hydration/post.go

··· 388 388 389 389 // hydrateEmbeddedRecord hydrates an embedded record (for quote posts, etc.) 390 390 func (h *Hydrator) hydrateEmbeddedRecord(ctx context.Context, uri string, viewerDID string) *bsky.EmbedRecord_View_Record { 391 391 + ctx, span := tracer.Start(ctx, "hydrateEmbeddedRecord") 392 392 + defer span.End() 393 393 + 391 394 // Check if it's a post URI 392 395 if !isPostURI(uri) { 393 396 // Could be a feed generator, list, labeler, or starter pack

+3 -7

main.go

··· 200 200 client: cc, 201 201 dir: dir, 202 202 203 203 - missingRecords: make(chan MissingRecord, 1024), 204 204 - db: db, 203 203 + db: db, 205 204 } 206 205 fmt.Println("MY DID: ", s.mydid) 207 206 208 208 - pgb, err := backend.NewPostgresBackend(mydid, db, pool, cc, nil) 207 207 + pgb, err := backend.NewPostgresBackend(mydid, db, pool, cc, dir) 209 208 if err != nil { 210 209 return err 211 210 } ··· 242 241 http.ListenAndServe(":4445", nil) 243 242 }() 244 243 245 245 - go s.missingRecordFetcher() 246 246 - 247 244 seqno, err := loadLastSeq(db, "firehose_seq") 248 245 if err != nil { 249 246 fmt.Println("failed to load sequence number, starting over", err) ··· 267 264 seqLk sync.Mutex 268 265 lastSeq int64 269 266 270 270 - mpLk sync.Mutex 271 271 - missingRecords chan MissingRecord 267 267 + mpLk sync.Mutex 272 268 273 269 db *gorm.DB 274 270 }

+24 -47

missing.go backend/missing.go

··· 1 1 - package main 1 1 + package backend 2 2 3 3 import ( 4 4 "bytes" ··· 19 19 MissingRecordTypeProfile MissingRecordType = "profile" 20 20 MissingRecordTypePost MissingRecordType = "post" 21 21 MissingRecordTypeFeedGenerator MissingRecordType = "feedgenerator" 22 22 + MissingRecordTypeUnknown MissingRecordType = "unknown" 22 23 ) 23 24 24 25 type MissingRecord struct { ··· 29 30 waitch chan struct{} 30 31 } 31 32 32 32 - func (s *Server) addMissingRecord(ctx context.Context, rec MissingRecord) { 33 33 + func (b *PostgresBackend) addMissingRecord(ctx context.Context, rec MissingRecord) { 33 34 if rec.Wait { 34 35 rec.waitch = make(chan struct{}) 35 36 } 36 37 37 38 select { 38 38 - case s.missingRecords <- rec: 39 39 + case b.missingRecords <- rec: 39 40 case <-ctx.Done(): 40 41 } 41 42 ··· 47 48 } 48 49 } 49 50 50 50 - // Legacy methods for backward compatibility 51 51 - func (s *Server) addMissingProfile(ctx context.Context, did string) { 52 52 - s.addMissingRecord(ctx, MissingRecord{ 53 53 - Type: MissingRecordTypeProfile, 54 54 - Identifier: did, 55 55 - }) 56 56 - } 57 57 - 58 58 - func (s *Server) addMissingPost(ctx context.Context, uri string) { 59 59 - slog.Info("adding missing post to fetch queue", "uri", uri) 60 60 - s.addMissingRecord(ctx, MissingRecord{ 61 61 - Type: MissingRecordTypePost, 62 62 - Identifier: uri, 63 63 - }) 64 64 - } 65 65 - 66 66 - func (s *Server) addMissingFeedGenerator(ctx context.Context, uri string) { 67 67 - slog.Info("adding missing feed generator to fetch queue", "uri", uri) 68 68 - s.addMissingRecord(ctx, MissingRecord{ 69 69 - Type: MissingRecordTypeFeedGenerator, 70 70 - Identifier: uri, 71 71 - }) 72 72 - } 73 73 - 74 74 - func (s *Server) missingRecordFetcher() { 75 75 - for rec := range s.missingRecords { 51 51 + func (b *PostgresBackend) missingRecordFetcher() { 52 52 + for rec := range b.missingRecords { 76 53 var err error 77 54 switch rec.Type { 78 55 case MissingRecordTypeProfile: 79 79 - err = s.fetchMissingProfile(context.TODO(), rec.Identifier) 56 56 + err = b.fetchMissingProfile(context.TODO(), rec.Identifier) 80 57 case MissingRecordTypePost: 81 81 - err = s.fetchMissingPost(context.TODO(), rec.Identifier) 58 58 + err = b.fetchMissingPost(context.TODO(), rec.Identifier) 82 59 case MissingRecordTypeFeedGenerator: 83 83 - err = s.fetchMissingFeedGenerator(context.TODO(), rec.Identifier) 60 60 + err = b.fetchMissingFeedGenerator(context.TODO(), rec.Identifier) 84 61 default: 85 62 slog.Error("unknown missing record type", "type", rec.Type) 86 63 continue ··· 96 73 } 97 74 } 98 75 99 99 - func (s *Server) fetchMissingProfile(ctx context.Context, did string) error { 100 100 - s.backend.AddRelevantDid(did) 76 76 + func (b *PostgresBackend) fetchMissingProfile(ctx context.Context, did string) error { 77 77 + b.AddRelevantDid(did) 101 78 102 102 - repo, err := s.backend.GetOrCreateRepo(ctx, did) 79 79 + repo, err := b.GetOrCreateRepo(ctx, did) 103 80 if err != nil { 104 81 return err 105 82 } 106 83 107 107 - resp, err := s.dir.LookupDID(ctx, syntax.DID(did)) 84 84 + resp, err := b.dir.LookupDID(ctx, syntax.DID(did)) 108 85 if err != nil { 109 86 return err 110 87 } ··· 133 110 return err 134 111 } 135 112 136 136 - return s.backend.HandleUpdateProfile(ctx, repo, "self", "", buf.Bytes(), cc) 113 113 + return b.HandleUpdateProfile(ctx, repo, "self", "", buf.Bytes(), cc) 137 114 } 138 115 139 139 - func (s *Server) fetchMissingPost(ctx context.Context, uri string) error { 116 116 + func (b *PostgresBackend) fetchMissingPost(ctx context.Context, uri string) error { 140 117 puri, err := syntax.ParseATURI(uri) 141 118 if err != nil { 142 119 return fmt.Errorf("invalid AT URI: %s", uri) ··· 146 123 collection := puri.Collection().String() 147 124 rkey := puri.RecordKey().String() 148 125 149 149 - s.backend.AddRelevantDid(did) 126 126 + b.AddRelevantDid(did) 150 127 151 151 - repo, err := s.backend.GetOrCreateRepo(ctx, did) 128 128 + repo, err := b.GetOrCreateRepo(ctx, did) 152 129 if err != nil { 153 130 return err 154 131 } 155 132 156 156 - resp, err := s.dir.LookupDID(ctx, syntax.DID(did)) 133 133 + resp, err := b.dir.LookupDID(ctx, syntax.DID(did)) 157 134 if err != nil { 158 135 return err 159 136 } ··· 182 159 return err 183 160 } 184 161 185 185 - return s.backend.HandleCreatePost(ctx, repo, rkey, buf.Bytes(), cc) 162 162 + return b.HandleCreatePost(ctx, repo, rkey, buf.Bytes(), cc) 186 163 } 187 164 188 188 - func (s *Server) fetchMissingFeedGenerator(ctx context.Context, uri string) error { 165 165 + func (b *PostgresBackend) fetchMissingFeedGenerator(ctx context.Context, uri string) error { 189 166 puri, err := syntax.ParseATURI(uri) 190 167 if err != nil { 191 168 return fmt.Errorf("invalid AT URI: %s", uri) ··· 194 171 did := puri.Authority().String() 195 172 collection := puri.Collection().String() 196 173 rkey := puri.RecordKey().String() 197 197 - s.backend.AddRelevantDid(did) 174 174 + b.AddRelevantDid(did) 198 175 199 199 - repo, err := s.backend.GetOrCreateRepo(ctx, did) 176 176 + repo, err := b.GetOrCreateRepo(ctx, did) 200 177 if err != nil { 201 178 return err 202 179 } 203 180 204 204 - resp, err := s.dir.LookupDID(ctx, syntax.DID(did)) 181 181 + resp, err := b.dir.LookupDID(ctx, syntax.DID(did)) 205 182 if err != nil { 206 183 return err 207 184 } ··· 230 207 return err 231 208 } 232 209 233 233 - return s.backend.HandleCreateFeedGenerator(ctx, repo, rkey, buf.Bytes(), cc) 210 210 + return b.HandleCreateFeedGenerator(ctx, repo, rkey, buf.Bytes(), cc) 234 211 }

+12

xrpc/notification/listNotifications.go

··· 131 131 cursorPtr = &cursor 132 132 } 133 133 134 134 + var lastSeen time.Time 135 135 + if err := db.Raw("SELECT seen_at FROM notification_seens WHERE repo = (select id from repos where did = ?)", viewer).Scan(&lastSeen).Error; err != nil { 136 136 + return err 137 137 + } 138 138 + 139 139 + var lastSeenStr *string 140 140 + if !lastSeen.IsZero() { 141 141 + s := lastSeen.Format(time.RFC3339) 142 142 + lastSeenStr = &s 143 143 + } 144 144 + 134 145 output := &bsky.NotificationListNotifications_Output{ 135 146 Notifications: notifications, 136 147 Cursor: cursorPtr, 148 148 + SeenAt: lastSeenStr, 137 149 } 138 150 139 151 return c.JSON(http.StatusOK, output)