Low-latency AT Protocol interaction indexer.
init 2 electric boogaloo
Aly Raffauf
(Jul 2, 2026, 7:41 PM EDT)
d86274fa
65d77623
+244
-122
+25
-122
cmd/asterism/main.go
+25
-122
cmd/asterism/main.go
···
1
1
package main
2
2
3
3
import (
4
-
"bytes"
5
4
"context"
6
-
"fmt"
5
+
"flag"
7
6
"log/slog"
8
7
"net/http"
9
8
"strings"
10
9
11
-
"github.com/bluesky-social/indigo/api/atproto"
12
-
"github.com/bluesky-social/indigo/api/bsky"
13
-
"github.com/bluesky-social/indigo/events"
14
-
"github.com/bluesky-social/indigo/events/schedulers/sequential"
15
-
indigorepo "github.com/bluesky-social/indigo/repo"
16
10
"github.com/gorilla/websocket"
17
-
"github.com/ipfs/go-cid"
18
-
)
19
11
20
-
type Edge struct {
21
-
Kind string
22
-
ActorDid string
23
-
Collection string
24
-
RecordKey string
25
-
RecordPath string
26
-
RecordCid string
27
-
Target string
28
-
TargetCid string
29
-
Rev string
30
-
}
31
-
32
-
func main() {
33
-
ctx := context.Background()
34
-
35
-
conn, _, err := websocket.DefaultDialer.Dial(
36
-
"wss://relay1.us-east.bsky.network/xrpc/com.atproto.sync.subscribeRepos",
37
-
http.Header{},
38
-
)
39
-
if err != nil {
40
-
panic(err)
41
-
}
42
-
defer conn.Close()
43
-
44
-
callbacks := &events.RepoStreamCallbacks{
45
-
RepoCommit: func(event *atproto.SyncSubscribeRepos_Commit) error {
46
-
return handleRepoCommit(ctx, event)
47
-
},
48
-
}
12
+
"github.com/alyraffauf/asterism/internal/firehose"
13
+
)
49
14
50
-
scheduler := sequential.NewScheduler("asterism", callbacks.EventHandler)
51
-
logger := slog.Default()
15
+
const relayURL = "wss://relay1.us-east.bsky.network/xrpc/com.atproto.sync.subscribeRepos"
52
16
53
-
if err := events.HandleRepoStream(ctx, conn, scheduler, logger); err != nil {
54
-
panic(err)
55
-
}
56
-
}
57
-
58
-
func handleRepoCommit(ctx context.Context, event *atproto.SyncSubscribeRepos_Commit) error {
59
-
fmt.Println("repo:", event.Repo, "commit:", event.Rev)
60
-
61
-
if event.TooBig {
62
-
fmt.Println("too big, queue for backfill later")
17
+
func parseCollections(raw string) map[string]struct{} {
18
+
if raw == "" {
63
19
return nil
64
20
}
65
21
66
-
repo, err := indigorepo.ReadRepoFromCar(ctx, bytes.NewReader(event.Blocks))
67
-
if err != nil {
68
-
return err
69
-
}
70
-
71
-
for _, operation := range event.Ops {
72
-
if err := handleRepoOperation(ctx, event, repo, operation); err != nil {
73
-
fmt.Println("could not handle operation:", err)
22
+
wanted := make(map[string]struct{})
23
+
for collection := range strings.SplitSeq(raw, ",") {
24
+
collection = strings.TrimSpace(collection)
25
+
if collection == "" {
74
26
continue
75
27
}
28
+
wanted[collection] = struct{}{}
29
+
76
30
}
77
-
78
-
return nil
31
+
return wanted
79
32
}
80
33
81
-
func handleRepoOperation(ctx context.Context, event *atproto.SyncSubscribeRepos_Commit, repo *indigorepo.Repo, operation *atproto.SyncSubscribeRepos_RepoOp) error {
82
-
collection, recordKey, ok := strings.Cut(operation.Path, "/")
83
-
if !ok {
84
-
return fmt.Errorf("bad path: %s", operation.Path)
85
-
}
34
+
func main() {
35
+
collectionsFlag := flag.String("collections", "", "comma-separated list of collection NSIDs to filter on (empty means all)")
36
+
flag.Parse()
86
37
87
-
fmt.Println("op:", operation.Action, "collection:", collection, "rkey:", recordKey)
38
+
ctx := context.Background()
39
+
logger := slog.Default()
88
40
89
-
if operation.Action == "delete" {
90
-
return nil
91
-
}
92
-
93
-
recordCid, record, err := repo.GetRecord(ctx, operation.Path)
41
+
conn, _, err := websocket.DefaultDialer.Dial(relayURL, http.Header{})
94
42
if err != nil {
95
-
return fmt.Errorf("read record: %w", err)
43
+
panic(err)
96
44
}
45
+
defer conn.Close()
97
46
98
-
if operation.Cid == nil {
99
-
return fmt.Errorf("missing operation cid: %s", operation.Path)
100
-
}
101
-
102
-
operationCid := cid.Cid(*operation.Cid)
103
-
if !recordCid.Equals(operationCid) {
104
-
return fmt.Errorf("cid mismatch: %s operation=%s record=%s", operation.Path, operationCid, recordCid)
47
+
consumer := &firehose.Consumer{
48
+
WantedCollections: parseCollections(*collectionsFlag),
105
49
}
106
50
107
-
printRecord(event, operation, collection, recordKey, recordCid, record)
108
-
109
-
return nil
110
-
}
111
-
112
-
func printRecord(
113
-
event *atproto.SyncSubscribeRepos_Commit,
114
-
operation *atproto.SyncSubscribeRepos_RepoOp,
115
-
collection string,
116
-
recordKey string,
117
-
recordCid cid.Cid,
118
-
record any,
119
-
) {
120
-
switch record := record.(type) {
121
-
case *bsky.FeedRepost:
122
-
edge := Edge{
123
-
Kind: "repost",
124
-
ActorDid: event.Repo,
125
-
Collection: collection,
126
-
RecordKey: recordKey,
127
-
RecordPath: operation.Path,
128
-
RecordCid: recordCid.String(),
129
-
Target: record.Subject.Uri,
130
-
TargetCid: record.Subject.Cid,
131
-
Rev: event.Rev,
132
-
}
133
-
fmt.Printf("edge: %+v\n", edge)
134
-
135
-
case *bsky.GraphFollow:
136
-
edge := Edge{
137
-
Kind: "follow",
138
-
ActorDid: event.Repo,
139
-
Collection: collection,
140
-
RecordKey: recordKey,
141
-
RecordPath: operation.Path,
142
-
RecordCid: recordCid.String(),
143
-
Target: record.Subject,
144
-
Rev: event.Rev,
145
-
}
146
-
fmt.Printf("edge: %+v\n", edge)
147
-
148
-
default:
149
-
fmt.Printf("other record type: %T\n", record)
51
+
if err := consumer.Run(ctx, conn, logger); err != nil {
52
+
panic(err)
150
53
}
151
54
}
+82
internal/backlink/extract.go
+82
internal/backlink/extract.go
···
1
+
package backlink
2
+
3
+
import (
4
+
"fmt"
5
+
6
+
"github.com/bluesky-social/indigo/atproto/syntax"
7
+
)
8
+
9
+
func isLinkTarget(s string) bool {
10
+
if _, err := syntax.ParseATURI(s); err == nil {
11
+
return true
12
+
}
13
+
if _, err := syntax.ParseDID(s); err == nil {
14
+
return true
15
+
}
16
+
return false
17
+
}
18
+
19
+
func tryStrongRef(obj map[string]any) (target string, targetCid string, ok bool) {
20
+
if len(obj) != 2 {
21
+
return "", "", false
22
+
}
23
+
24
+
uri, isURI := obj["uri"].(string)
25
+
c, isCid := obj["cid"].(string)
26
+
27
+
if !isURI || !isCid {
28
+
return "", "", false
29
+
}
30
+
31
+
if !isLinkTarget(uri) {
32
+
return "", "", false
33
+
}
34
+
35
+
return uri, c, true
36
+
}
37
+
38
+
func joinPath(base, key string) string {
39
+
if base == "" {
40
+
return key
41
+
}
42
+
return base + "." + key
43
+
}
44
+
45
+
func walk(path string, value any, base Link) []Link {
46
+
switch v := value.(type) {
47
+
case map[string]any:
48
+
if target, targetCid, ok := tryStrongRef(v); ok {
49
+
link := base
50
+
link.FieldPath = path
51
+
link.Target = target
52
+
link.TargetCid = targetCid
53
+
return []Link{link}
54
+
}
55
+
56
+
var out []Link
57
+
for key, val := range v {
58
+
out = append(out, walk(joinPath(path, key), val, base)...)
59
+
}
60
+
return out
61
+
62
+
case []any:
63
+
var out []Link
64
+
for i, val := range v {
65
+
out = append(out, walk(fmt.Sprintf("%s[%d]", path, i), val, base)...)
66
+
}
67
+
return out
68
+
69
+
case string:
70
+
if isLinkTarget(v) {
71
+
link := base
72
+
link.FieldPath = path
73
+
link.Target = v
74
+
return []Link{link}
75
+
}
76
+
}
77
+
return nil
78
+
}
79
+
80
+
func Extract(record map[string]any, base Link) []Link {
81
+
return walk("", record, base)
82
+
}
+12
internal/backlink/link.go
+12
internal/backlink/link.go
+89
internal/firehose/commit.go
+89
internal/firehose/commit.go
···
1
+
package firehose
2
+
3
+
import (
4
+
"bytes"
5
+
"context"
6
+
"fmt"
7
+
"strings"
8
+
9
+
"github.com/alyraffauf/asterism/internal/backlink"
10
+
"github.com/bluesky-social/indigo/api/atproto"
11
+
"github.com/bluesky-social/indigo/atproto/atdata"
12
+
indigorepo "github.com/bluesky-social/indigo/repo"
13
+
"github.com/ipfs/go-cid"
14
+
)
15
+
16
+
func (c *Consumer) HandleCommit(ctx context.Context, event *atproto.SyncSubscribeRepos_Commit) error {
17
+
// fmt.Println("repo:", event.Repo, "commit:", event.Rev)
18
+
19
+
if event.TooBig {
20
+
fmt.Println("too big, queue for backfill later")
21
+
return nil
22
+
}
23
+
24
+
repo, err := indigorepo.ReadRepoFromCar(ctx, bytes.NewReader(event.Blocks))
25
+
if err != nil {
26
+
return err
27
+
}
28
+
29
+
for _, operation := range event.Ops {
30
+
if err := c.handleOperation(ctx, event, repo, operation); err != nil {
31
+
fmt.Println("could not handle operation:", err)
32
+
continue
33
+
}
34
+
}
35
+
36
+
return nil
37
+
}
38
+
39
+
func (c *Consumer) handleOperation(ctx context.Context, event *atproto.SyncSubscribeRepos_Commit, repo *indigorepo.Repo, operation *atproto.SyncSubscribeRepos_RepoOp) error {
40
+
collection, recordKey, ok := strings.Cut(operation.Path, "/")
41
+
42
+
if !ok {
43
+
return fmt.Errorf("bad path: %s", operation.Path)
44
+
}
45
+
46
+
if !c.wants(collection) {
47
+
return nil
48
+
}
49
+
50
+
fmt.Println("op:", operation.Action, "collection:", collection, "rkey:", recordKey)
51
+
52
+
if operation.Action == "delete" {
53
+
return nil
54
+
}
55
+
56
+
recordCid, recordBytes, err := repo.GetRecordBytes(ctx, operation.Path)
57
+
if err != nil {
58
+
return fmt.Errorf("read record: %w", err)
59
+
}
60
+
61
+
if operation.Cid == nil {
62
+
return fmt.Errorf("missing operation cid: %s", operation.Path)
63
+
}
64
+
65
+
operationCid := cid.Cid(*operation.Cid)
66
+
if !recordCid.Equals(operationCid) {
67
+
return fmt.Errorf("cid mismatch: %s operation=%s record=%s", operation.Path, operationCid, recordCid)
68
+
}
69
+
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
+
for _, link := range backlink.Extract(record, base) {
84
+
fmt.Printf("link: %+v\n", link)
85
+
}
86
+
87
+
return nil
88
+
89
+
}
+13
internal/firehose/consumer.go
+13
internal/firehose/consumer.go
+23
internal/firehose/stream.go
+23
internal/firehose/stream.go
···
1
+
package firehose
2
+
3
+
import (
4
+
"context"
5
+
"log/slog"
6
+
7
+
"github.com/bluesky-social/indigo/api/atproto"
8
+
"github.com/bluesky-social/indigo/events"
9
+
"github.com/bluesky-social/indigo/events/schedulers/sequential"
10
+
"github.com/gorilla/websocket"
11
+
)
12
+
13
+
func (c *Consumer) Run(ctx context.Context, conn *websocket.Conn, logger *slog.Logger) error {
14
+
callbacks := &events.RepoStreamCallbacks{
15
+
RepoCommit: func(event *atproto.SyncSubscribeRepos_Commit) error {
16
+
return c.HandleCommit(ctx, event)
17
+
},
18
+
}
19
+
20
+
scheduler := sequential.NewScheduler("asterism", callbacks.EventHandler)
21
+
22
+
return events.HandleRepoStream(ctx, conn, scheduler, logger)
23
+
}