why.bsky.team
A locally focused bluesky appview
some parallization for the getThreadV2 api
+19
backend/backend.go
+19
backend/backend.go
···
43
repoCache *lru.TwoQueueCache[string, *Repo]
44
reposLk sync.Mutex
45
46
postInfoCache *lru.TwoQueueCache[string, cachedPostInfo]
47
}
48
···
56
rc, _ := lru.New2Q[string, *Repo](1_000_000)
57
pc, _ := lru.New2Q[string, cachedPostInfo](1_000_000)
58
revc, _ := lru.New2Q[uint, string](1_000_000)
59
60
b := &PostgresBackend{
61
client: client,
···
67
repoCache: rc,
68
postInfoCache: pc,
69
revCache: revc,
70
}
71
72
r, err := b.GetOrCreateRepo(context.TODO(), mydid)
···
363
}
364
365
return &r, nil
366
}
367
368
func (b *PostgresBackend) checkPostExists(ctx context.Context, repo *Repo, rkey string) (bool, error) {
···
43
repoCache *lru.TwoQueueCache[string, *Repo]
44
reposLk sync.Mutex
45
46
+
didByIDCache *lru.TwoQueueCache[uint, string]
47
+
48
postInfoCache *lru.TwoQueueCache[string, cachedPostInfo]
49
}
50
···
58
rc, _ := lru.New2Q[string, *Repo](1_000_000)
59
pc, _ := lru.New2Q[string, cachedPostInfo](1_000_000)
60
revc, _ := lru.New2Q[uint, string](1_000_000)
61
+
dbic, _ := lru.New2Q[uint, string](1_000_000)
62
63
b := &PostgresBackend{
64
client: client,
···
70
repoCache: rc,
71
postInfoCache: pc,
72
revCache: revc,
73
+
didByIDCache: dbic,
74
}
75
76
r, err := b.GetOrCreateRepo(context.TODO(), mydid)
···
367
}
368
369
return &r, nil
370
+
}
371
+
372
+
func (b *PostgresBackend) DidFromID(ctx context.Context, uid uint) (string, error) {
373
+
val, ok := b.didByIDCache.Get(uid)
374
+
if ok {
375
+
return val, nil
376
+
}
377
+
378
+
r, err := b.GetRepoByID(ctx, uid)
379
+
if err != nil {
380
+
return "", err
381
+
}
382
+
383
+
b.didByIDCache.Add(uid, r.Did)
384
+
return r.Did, nil
385
}
386
387
func (b *PostgresBackend) checkPostExists(ctx context.Context, repo *Repo, rkey string) (bool, error) {
+2
-2
hydration/utils.go
+2
-2
hydration/utils.go
+1
main.go
+1
main.go
···
135
db.AutoMigrate(SequenceTracker{})
136
db.Exec("CREATE INDEX IF NOT EXISTS reposts_subject_idx ON reposts (subject)")
137
db.Exec("CREATE INDEX IF NOT EXISTS posts_reply_to_idx ON posts (reply_to)")
138
+
db.Exec("CREATE INDEX IF NOT EXISTS posts_in_thread_idx ON posts (in_thread)")
139
140
ctx := context.TODO()
141
+38
-12
xrpc/unspecced/getPostThreadV2.go
+38
-12
xrpc/unspecced/getPostThreadV2.go
···
7
"log/slog"
8
"net/http"
9
"strconv"
10
11
"github.com/bluesky-social/indigo/api/bsky"
12
"github.com/labstack/echo/v4"
13
"github.com/whyrusleeping/konbini/hydration"
14
"github.com/whyrusleeping/konbini/views"
15
"github.com/whyrusleeping/market/models"
16
"gorm.io/gorm"
17
)
18
19
// HandleGetPostThreadV2 implements app.bsky.unspecced.getPostThreadV2
20
func HandleGetPostThreadV2(c echo.Context, db *gorm.DB, hydrator *hydration.Hydrator) error {
21
-
ctx := c.Request().Context()
22
ctx = context.WithValue(ctx, "auto-fetch", true)
23
24
// Parse parameters
···
152
return nil, nil
153
}
154
155
-
var out []*bsky.UnspeccedGetPostThreadV2_ThreadItem
156
-
for _, child := range curnode.children {
157
-
out = append(out, buildThreadItem(ctx, hydrator, child, depth+1, viewer))
158
-
if child.missing {
159
-
continue
160
-
}
161
162
-
sub, err := collectReplies(ctx, hydrator, child, depth+1, below-1, branchingFactor, sort, viewer)
163
-
if err != nil {
164
-
return nil, err
165
-
}
166
167
-
out = append(out, sub...)
168
}
169
170
return out, nil
···
7
"log/slog"
8
"net/http"
9
"strconv"
10
+
"sync"
11
12
"github.com/bluesky-social/indigo/api/bsky"
13
"github.com/labstack/echo/v4"
14
"github.com/whyrusleeping/konbini/hydration"
15
"github.com/whyrusleeping/konbini/views"
16
"github.com/whyrusleeping/market/models"
17
+
"go.opentelemetry.io/otel"
18
"gorm.io/gorm"
19
)
20
+
21
+
var tracer = otel.Tracer("xrpc/unspecced")
22
23
// HandleGetPostThreadV2 implements app.bsky.unspecced.getPostThreadV2
24
func HandleGetPostThreadV2(c echo.Context, db *gorm.DB, hydrator *hydration.Hydrator) error {
25
+
ctx, span := tracer.Start(c.Request().Context(), "getPostThreadV2")
26
+
defer span.End()
27
ctx = context.WithValue(ctx, "auto-fetch", true)
28
29
// Parse parameters
···
157
return nil, nil
158
}
159
160
+
type parThreadResults struct {
161
+
node *bsky.UnspeccedGetPostThreadV2_ThreadItem
162
+
children []*bsky.UnspeccedGetPostThreadV2_ThreadItem
163
+
}
164
+
165
+
results := make([]parThreadResults, len(curnode.children))
166
+
167
+
var wg sync.WaitGroup
168
+
for i := range curnode.children {
169
+
ix := i
170
+
wg.Go(func() {
171
+
child := curnode.children[ix]
172
+
173
+
results[ix].node = buildThreadItem(ctx, hydrator, child, depth+1, viewer)
174
+
if child.missing {
175
+
return
176
+
}
177
+
178
+
sub, err := collectReplies(ctx, hydrator, child, depth+1, below-1, branchingFactor, sort, viewer)
179
+
if err != nil {
180
+
slog.Error("failed to collect replies", "node", child.uri, "error", err)
181
+
return
182
+
}
183
+
184
+
results[ix].children = sub
185
+
})
186
+
}
187
188
+
wg.Wait()
189
190
+
var out []*bsky.UnspeccedGetPostThreadV2_ThreadItem
191
+
for _, res := range results {
192
+
out = append(out, res.node)
193
+
out = append(out, res.children...)
194
}
195
196
return out, nil