Low-latency AT Protocol interaction indexer.
handle TooBig events
Aly Raffauf
(Jul 3, 2026, 1:30 AM EDT)
18ceacb6
be9c010a
+27
-19
+7
-5
cmd/asterism/main.go
+7
-5
cmd/asterism/main.go
···
69
69
collections = append(collections, collection)
70
70
}
71
71
72
+
bf := &backfill.Backfill{
73
+
Client: &xrpc.Client{Host: "https://relay1.us-east.bsky.network"},
74
+
Directory: identity.DefaultDirectory(),
75
+
Store: linkStore,
76
+
}
77
+
72
78
if len(collections) > 0 {
73
-
bf := &backfill.Backfill{
74
-
Client: &xrpc.Client{Host: "https://relay1.us-east.bsky.network"},
75
-
Directory: identity.DefaultDirectory(),
76
-
Store: linkStore,
77
-
}
78
79
go func() {
79
80
if err := bf.Run(ctx, collections); err != nil {
80
81
fmt.Println("backfill error:", err)
···
85
86
consumer := &firehose.Consumer{
86
87
WantedCollections: wantedCollections,
87
88
Store: linkStore,
89
+
Backfill: bf,
88
90
}
89
91
90
92
if err := consumer.Run(ctx, conn, logger); err != nil {
+2
-3
internal/api/dids.go
+2
-3
internal/api/dids.go
···
9
9
)
10
10
11
11
type backlinkDidsResponse struct {
12
-
Total uint64 `json:"total"`
12
+
Total uint64 `json:"total"`
13
13
LinkingDids []string `json:"linking_dids"`
14
-
Cursor *string `json:"cursor"`
14
+
Cursor *string `json:"cursor"`
15
15
}
16
16
17
17
func (s *Server) GetBacklinkDids(w http.ResponseWriter, r *http.Request) {
···
25
25
http.Error(w, err.Error(), http.StatusBadRequest)
26
26
return
27
27
}
28
-
29
28
30
29
limit := uint64(100)
31
30
if raw := query.Get("limit"); raw != "" {
+6
-4
internal/backfill/backfill.go
+6
-4
internal/backfill/backfill.go
···
59
59
}
60
60
61
61
for did, wantedCollections := range dids {
62
-
if err := b.repo(ctx, did, wantedCollections); err != nil {
62
+
if err := b.Repo(ctx, did, wantedCollections); err != nil {
63
63
fmt.Println("could not backfill repo:", did, err)
64
64
continue
65
65
}
···
90
90
}
91
91
}
92
92
93
-
func (b *Backfill) repo(ctx context.Context, did string, wantedCollections map[string]struct{}) error {
93
+
func (b *Backfill) Repo(ctx context.Context, did string, wantedCollections map[string]struct{}) error {
94
94
resolveCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
95
95
identity, err := b.Directory.LookupDID(resolveCtx, syntax.DID(did))
96
96
cancel()
···
126
126
return fmt.Errorf("bad path: %s", k)
127
127
}
128
128
129
-
if _, wanted := wantedCollections[collection]; !wanted {
130
-
return nil
129
+
if len(wantedCollections) > 0 {
130
+
if _, wanted := wantedCollections[collection]; !wanted {
131
+
return nil
132
+
}
131
133
}
132
134
133
135
recordCid, recordBytes, err := repo.GetRecordBytes(ctx, k)
-3
internal/backlink/extract.go
-3
internal/backlink/extract.go
···
17
17
return false
18
18
}
19
19
20
-
21
20
func tryStrongRef(obj map[string]any) (target string, targetCid string, ok bool) {
22
21
if len(obj) != 2 {
23
22
return "", "", false
···
50
49
return "[]"
51
50
}
52
51
53
-
54
52
func walk(path string, value any, base Link) []Link {
55
53
switch v := value.(type) {
56
54
case map[string]any:
···
74
72
out = append(out, walk(path+arrayPathSuffix(val), val, base)...)
75
73
}
76
74
return out
77
-
78
75
79
76
case string:
80
77
if isLinkTarget(v) {
+5
-1
internal/firehose/commit.go
+5
-1
internal/firehose/commit.go
···
16
16
// fmt.Println("repo:", event.Repo, "commit:", event.Rev)
17
17
18
18
if event.TooBig {
19
-
fmt.Println("too big, queue for backfill later")
19
+
go func() {
20
+
if err := c.Backfill.Repo(ctx, event.Repo, c.WantedCollections); err != nil {
21
+
fmt.Println("could not backfill too-big repo:", event.Repo, err)
22
+
}
23
+
}()
20
24
return nil
21
25
}
22
26
+5
-1
internal/firehose/consumer.go
+5
-1
internal/firehose/consumer.go
···
1
1
package firehose
2
2
3
-
import "github.com/alyraffauf/asterism/internal/store"
3
+
import (
4
+
"github.com/alyraffauf/asterism/internal/backfill"
5
+
"github.com/alyraffauf/asterism/internal/store"
6
+
)
4
7
5
8
type Consumer struct {
6
9
WantedCollections map[string]struct{}
7
10
Store *store.Store
11
+
Backfill *backfill.Backfill
8
12
}
9
13
10
14
func (c *Consumer) wants(collection string) bool {
+2
-2
internal/store/query.go
+2
-2
internal/store/query.go
···
61
61
return total, dids, nil
62
62
}
63
63
64
-
func (s *Store) ListBacklinks(ctx context.Context, target, collection, fieldPath string, dids[]string, reverse bool, after int64, limit uint64) (total uint64, records []Record, err error) {
64
+
func (s *Store) ListBacklinks(ctx context.Context, target, collection, fieldPath string, dids []string, reverse bool, after int64, limit uint64) (total uint64, records []Record, err error) {
65
65
where := `target = ? AND collection = ? AND field_path = ?`
66
66
args := []any{target, collection, fieldPath}
67
67
···
74
74
where += `AND actor_did IN (` + strings.Join(placeholders, ", ") + `)`
75
75
76
76
err = s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM links WHERE `+where, args...).Scan(&total)
77
-
if err!= nil {
77
+
if err != nil {
78
78
return 0, nil, fmt.Errorf("count backlinks: %w", err)
79
79
}
80
80
} else {