why.bsky.team
A locally focused bluesky appview
missing post fetcher
+68
+1
handlers.go
+1
handlers.go
+3
main.go
+3
main.go
···
159
dir: dir,
160
161
missingProfiles: make(chan string, 1024),
162
}
163
164
pgb := &PostgresBackend{
···
193
}()
194
195
go s.missingProfileFetcher()
196
197
seqno, err := loadLastSeq("sequence.txt")
198
if err != nil {
···
219
220
mpLk sync.Mutex
221
missingProfiles chan string
222
}
223
224
func (s *Server) getXrpcClient() (*xrpc.Client, error) {
···
159
dir: dir,
160
161
missingProfiles: make(chan string, 1024),
162
+
missingPosts: make(chan string, 1024),
163
}
164
165
pgb := &PostgresBackend{
···
194
}()
195
196
go s.missingProfileFetcher()
197
+
go s.missingPostFetcher()
198
199
seqno, err := loadLastSeq("sequence.txt")
200
if err != nil {
···
221
222
mpLk sync.Mutex
223
missingProfiles chan string
224
+
missingPosts chan string
225
}
226
227
func (s *Server) getXrpcClient() (*xrpc.Client, error) {
+64
missing.go
+64
missing.go
···
4
"bytes"
5
"context"
6
"fmt"
7
+
"strings"
8
9
"github.com/bluesky-social/indigo/api/atproto"
10
"github.com/bluesky-social/indigo/api/bsky"
···
66
67
return s.backend.HandleUpdateProfile(ctx, repo, "self", "", buf.Bytes(), cc)
68
}
69
+
70
+
func (s *Server) addMissingPost(ctx context.Context, uri string) {
71
+
select {
72
+
case s.missingPosts <- uri:
73
+
case <-ctx.Done():
74
+
}
75
+
}
76
+
77
+
func (s *Server) missingPostFetcher() {
78
+
for uri := range s.missingPosts {
79
+
if err := s.fetchMissingPost(context.TODO(), uri); err != nil {
80
+
log.Warn("failed to fetch missing post", "uri", uri, "error", err)
81
+
}
82
+
}
83
+
}
84
+
85
+
func (s *Server) fetchMissingPost(ctx context.Context, uri string) error {
86
+
// Parse AT URI: at://did:plc:xxx/app.bsky.feed.post/rkey
87
+
parts := strings.Split(uri, "/")
88
+
if len(parts) < 5 || !strings.HasPrefix(parts[2], "did:") {
89
+
return fmt.Errorf("invalid AT URI: %s", uri)
90
+
}
91
+
92
+
did := parts[2]
93
+
collection := parts[3]
94
+
rkey := parts[4]
95
+
96
+
repo, err := s.backend.getOrCreateRepo(ctx, did)
97
+
if err != nil {
98
+
return err
99
+
}
100
+
101
+
resp, err := s.dir.LookupDID(ctx, syntax.DID(did))
102
+
if err != nil {
103
+
return err
104
+
}
105
+
106
+
c := &xrpc.Client{
107
+
Host: resp.PDSEndpoint(),
108
+
}
109
+
110
+
rec, err := atproto.RepoGetRecord(ctx, c, "", collection, did, rkey)
111
+
if err != nil {
112
+
return err
113
+
}
114
+
115
+
post, ok := rec.Value.Val.(*bsky.FeedPost)
116
+
if !ok {
117
+
return fmt.Errorf("record we got back wasn't a post somehow")
118
+
}
119
+
120
+
buf := new(bytes.Buffer)
121
+
if err := post.MarshalCBOR(buf); err != nil {
122
+
return err
123
+
}
124
+
125
+
cc, err := cid.Decode(*rec.Cid)
126
+
if err != nil {
127
+
return err
128
+
}
129
+
130
+
return s.backend.HandleCreatePost(ctx, repo, rkey, buf.Bytes(), cc)
131
+
}