Low-latency AT Protocol interaction indexer.
add backfill
Aly Raffauf
(Jul 3, 2026, 1:09 AM EDT)
b9822e22
cb68724b
+169
-23
+25
-2
cmd/asterism/main.go
+25
-2
cmd/asterism/main.go
···
3
3
import (
4
4
"context"
5
5
"flag"
6
+
"fmt"
6
7
"log/slog"
7
8
"net/http"
8
9
"strings"
9
10
11
+
"github.com/bluesky-social/indigo/atproto/identity"
12
+
"github.com/bluesky-social/indigo/xrpc"
10
13
"github.com/gorilla/websocket"
11
14
12
15
"github.com/alyraffauf/asterism/internal/api"
16
+
"github.com/alyraffauf/asterism/internal/backfill"
13
17
"github.com/alyraffauf/asterism/internal/firehose"
14
18
"github.com/alyraffauf/asterism/internal/store"
15
19
)
···
57
61
panic(err)
58
62
}
59
63
}()
60
-
defer conn.Close()
64
+
65
+
wantedCollections := parseCollections(*collectionsFlag)
66
+
67
+
var collections []string
68
+
for collection := range wantedCollections {
69
+
collections = append(collections, collection)
70
+
}
71
+
72
+
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
+
go func() {
79
+
if err := bf.Run(ctx, collections); err != nil {
80
+
fmt.Println("backfill error:", err)
81
+
}
82
+
}()
83
+
}
61
84
62
85
consumer := &firehose.Consumer{
63
-
WantedCollections: parseCollections(*collectionsFlag),
86
+
WantedCollections: wantedCollections,
64
87
Store: linkStore,
65
88
}
66
89
+112
internal/backfill/backfill.go
+112
internal/backfill/backfill.go
···
1
+
package backfill
2
+
3
+
import (
4
+
"bytes"
5
+
"context"
6
+
"fmt"
7
+
"net/http"
8
+
"strings"
9
+
"time"
10
+
11
+
"github.com/alyraffauf/asterism/internal/index"
12
+
"github.com/alyraffauf/asterism/internal/store"
13
+
"github.com/bluesky-social/indigo/api/atproto"
14
+
"github.com/bluesky-social/indigo/atproto/identity"
15
+
"github.com/bluesky-social/indigo/atproto/syntax"
16
+
"github.com/bluesky-social/indigo/xrpc"
17
+
"github.com/ipfs/go-cid"
18
+
19
+
indigorepo "github.com/bluesky-social/indigo/repo"
20
+
)
21
+
22
+
type Backfill struct {
23
+
Client *xrpc.Client
24
+
Directory identity.Directory
25
+
Store *store.Store
26
+
}
27
+
28
+
func (b *Backfill) Run(ctx context.Context, collections []string) error {
29
+
dids := make(map[string]map[string]struct{})
30
+
31
+
for _, collection := range collections {
32
+
if err := b.listRepos(ctx, collection, dids); err != nil {
33
+
return fmt.Errorf("list repos for %s: %w", collection, err)
34
+
}
35
+
}
36
+
37
+
for did, wantedCollections := range dids {
38
+
if err := b.repo(ctx, did, wantedCollections); err != nil {
39
+
fmt.Println("could not backfill repo:", did, err)
40
+
continue
41
+
}
42
+
}
43
+
44
+
return nil
45
+
}
46
+
47
+
func (b *Backfill) listRepos(ctx context.Context, collection string, dids map[string]map[string]struct{}) error {
48
+
cursor := ""
49
+
for {
50
+
page, err := atproto.SyncListReposByCollection(ctx, b.Client, collection, cursor, 1000)
51
+
if err != nil {
52
+
return fmt.Errorf("list repos: %w", err)
53
+
}
54
+
55
+
for _, repo := range page.Repos {
56
+
if dids[repo.Did] == nil {
57
+
dids[repo.Did] = make(map[string]struct{})
58
+
}
59
+
dids[repo.Did][collection] = struct{}{}
60
+
}
61
+
62
+
if page.Cursor == nil || *page.Cursor == "" {
63
+
return nil
64
+
}
65
+
cursor = *page.Cursor
66
+
}
67
+
}
68
+
69
+
func (b *Backfill) repo(ctx context.Context, did string, wantedCollections map[string]struct{}) error {
70
+
resolveCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
71
+
identity, err := b.Directory.LookupDID(resolveCtx, syntax.DID(did))
72
+
cancel()
73
+
if err != nil {
74
+
return fmt.Errorf("resolve did: %w", err)
75
+
}
76
+
77
+
pdsClient := &xrpc.Client{
78
+
Host: identity.PDSEndpoint(),
79
+
Client: &http.Client{Timeout: 5 * time.Minute},
80
+
}
81
+
82
+
fetchCtx, cancel := context.WithTimeout(ctx, 5*time.Minute)
83
+
defer cancel()
84
+
85
+
carBytes, err := atproto.SyncGetRepo(fetchCtx, pdsClient, did, "")
86
+
if err != nil {
87
+
return fmt.Errorf("get repo: %w", err)
88
+
}
89
+
90
+
repo, err := indigorepo.ReadRepoFromCar(ctx, bytes.NewReader(carBytes))
91
+
if err != nil {
92
+
return fmt.Errorf("read repo car: %w", err)
93
+
}
94
+
95
+
return repo.ForEach(ctx, "", func(k string, v cid.Cid) error {
96
+
collection, recordKey, ok := strings.Cut(k, "/")
97
+
if !ok {
98
+
return fmt.Errorf("bad path: %s", k)
99
+
}
100
+
101
+
if _, wanted := wantedCollections[collection]; !wanted {
102
+
return nil
103
+
}
104
+
105
+
recordCid, recordBytes, err := repo.GetRecordBytes(ctx, k)
106
+
if err != nil {
107
+
return fmt.Errorf("read record: %w", err)
108
+
}
109
+
110
+
return index.Record(ctx, b.Store, did, collection, recordKey, recordCid.String(), "", *recordBytes)
111
+
})
112
+
}
+2
-21
internal/firehose/commit.go
+2
-21
internal/firehose/commit.go
···
6
6
"fmt"
7
7
"strings"
8
8
9
-
"github.com/alyraffauf/asterism/internal/backlink"
9
+
"github.com/alyraffauf/asterism/internal/index"
10
10
"github.com/bluesky-social/indigo/api/atproto"
11
-
"github.com/bluesky-social/indigo/atproto/atdata"
12
11
indigorepo "github.com/bluesky-social/indigo/repo"
13
12
"github.com/ipfs/go-cid"
14
13
)
···
38
37
39
38
func (c *Consumer) handleOperation(ctx context.Context, event *atproto.SyncSubscribeRepos_Commit, repo *indigorepo.Repo, operation *atproto.SyncSubscribeRepos_RepoOp) error {
40
39
collection, recordKey, ok := strings.Cut(operation.Path, "/")
41
-
42
40
if !ok {
43
41
return fmt.Errorf("bad path: %s", operation.Path)
44
42
}
···
46
44
if !c.wants(collection) {
47
45
return nil
48
46
}
49
-
50
-
fmt.Println("op:", operation.Action, "collection:", collection, "rkey:", recordKey)
51
47
52
48
if operation.Action == "delete" {
53
49
return c.Store.DeleteLinks(ctx, event.Repo, collection, recordKey)
···
67
63
return fmt.Errorf("cid mismatch: %s operation=%s record=%s", operation.Path, operationCid, recordCid)
68
64
}
69
65
70
-
record, err := atdata.UnmarshalCBOR(*recordBytes)
71
-
if err != nil {
72
-
return fmt.Errorf("decode record: %w", err)
73
-
}
74
-
75
-
base := backlink.Link{
76
-
ActorDid: event.Repo,
77
-
Collection: collection,
78
-
RecordKey: recordKey,
79
-
RecordCid: recordCid.String(),
80
-
Rev: event.Rev,
81
-
}
82
-
83
-
links := backlink.Extract(record, base)
84
-
85
-
return c.Store.SaveLinks(ctx, event.Repo, collection, recordKey, links)
66
+
return index.Record(ctx, c.Store, event.Repo, collection, recordKey, recordCid.String(), event.Rev, *recordBytes)
86
67
}
+30
internal/index/record.go
+30
internal/index/record.go
···
1
+
package index
2
+
3
+
import (
4
+
"context"
5
+
"fmt"
6
+
7
+
"github.com/bluesky-social/indigo/atproto/atdata"
8
+
9
+
"github.com/alyraffauf/asterism/internal/backlink"
10
+
"github.com/alyraffauf/asterism/internal/store"
11
+
)
12
+
13
+
func Record(ctx context.Context, s *store.Store, actorDid, collection, recordKey, recordCid, rev string, recordBytes []byte) error {
14
+
record, err := atdata.UnmarshalCBOR(recordBytes)
15
+
if err != nil {
16
+
return fmt.Errorf("decode record: %w", err)
17
+
}
18
+
19
+
base := backlink.Link{
20
+
ActorDid: actorDid,
21
+
Collection: collection,
22
+
RecordKey: recordKey,
23
+
RecordCid: recordCid,
24
+
Rev: rev,
25
+
}
26
+
27
+
links := backlink.Extract(record, base)
28
+
29
+
return s.SaveLinks(ctx, actorDid, collection, recordKey, links)
30
+
}