Low-latency AT Protocol interaction indexer.
add getManyToManyCounts and getManyToMany
Aly Raffauf
(Jul 3, 2026, 4:10 PM EDT)
dfbf789a
59cb5abd
+330
-6
+36
-2
README.md
+36
-2
README.md
···
43
43
44
44
## API
45
45
46
-
Asterism implements three endpoints from the [microcosm links XRPC namespace](https://constellation.microcosm.blue/):
46
+
Asterism implements all five current endpoints from the [microcosm links XRPC namespace](https://constellation.microcosm.blue/) (the older `/links/*` REST endpoints are deprecated upstream in favor of these and aren't implemented here):
47
47
48
48
### `GET /xrpc/blue.microcosm.links.getBacklinksCount`
49
49
···
87
87
88
88
Records identify the linking record by DID, collection, and rkey. Clients must hydrate display data separately (via AppView, PDS, etc.).
89
89
90
+
### `GET /xrpc/blue.microcosm.links.getManyToMany`
91
+
92
+
Join records linking to a subject with a second field path on those same records — a one-hop join in a single query. For example, `app.bsky.graph.listitem` records have both a `list` field and a `subject` field; joining them resolves list membership directly instead of requiring a `getBacklinks` call followed by N individual record lookups.
93
+
94
+
| Parameter | Description |
95
+
|---|---|
96
+
| `subject` | Target AT-URI, DID, or URL (required) |
97
+
| `source` | Collection and field path (required) |
98
+
| `pathToOther` | Second field path on the same source record (required) |
99
+
| `linkDid` | Filter to specific linking-record DIDs (repeatable) |
100
+
| `otherSubject` | Filter to specific secondary link targets (repeatable) |
101
+
| `limit` | Page size, 1–1000 (default 100) |
102
+
| `cursor` | Pagination cursor from previous response |
103
+
104
+
Response: `{"total": 42, "items": [{"linkRecord": {"did": "...", "collection": "...", "rkey": "..."}, "otherSubject": "..."}], "cursor": "..."}`
105
+
106
+
### `GET /xrpc/blue.microcosm.links.getManyToManyCounts`
107
+
108
+
Like `getManyToMany`, but grouped: counts of linking records per distinct secondary target instead of the individual records themselves. Useful when you only need aggregate counts, e.g. "how many people on each of these lists also follow me" without paginating every membership record.
109
+
110
+
| Parameter | Description |
111
+
|---|---|
112
+
| `subject` | Target AT-URI, DID, or URL (required) |
113
+
| `source` | Collection and field path (required) |
114
+
| `pathToOther` | Second field path on the same source record (required) |
115
+
| `did` | Filter to specific linking-record DIDs (repeatable) |
116
+
| `otherSubject` | Filter to specific secondary link targets (repeatable) |
117
+
| `limit` | Page size, 1–1000 (default 100) |
118
+
| `cursor` | Pagination cursor from previous response |
119
+
120
+
Response: `{"counts_by_other_subject": [{"subject": "...", "total": 42, "distinct": 12}], "cursor": "..."}`
121
+
122
+
Note the DID filter parameter is `did` here, not `linkDid` like `getManyToMany` — a real inconsistency in the upstream Constellation API (their own source flags it as a known TODO), preserved here for compatibility rather than "fixed."
123
+
90
124
## Roadmap
91
125
92
126
**Near term**
93
127
94
-
- [ ] `blue.microcosm.links.getManyToMany` endpoint (Constellation parity)
128
+
- [x] Full Constellation API parity (`getBacklinksCount`, `getBacklinkDids`, `getBacklinks`, `getManyToMany`, `getManyToManyCounts`)
95
129
- [ ] Configurable listen address, database path, and relay URL
96
130
- [ ] Account deletion and deactivation handling
97
131
- [x] Graceful shutdown and Firehose reconnect
+79
internal/api/many_to_many.go
+79
internal/api/many_to_many.go
···
1
+
package api
2
+
3
+
import (
4
+
"encoding/base64"
5
+
"encoding/json"
6
+
"log"
7
+
"net/http"
8
+
"strconv"
9
+
10
+
"github.com/alyraffauf/asterism/internal/store"
11
+
)
12
+
13
+
type manyToManyResponse struct {
14
+
Total uint64 `json:"total"`
15
+
Items []store.ManyToManyItem `json:"items"`
16
+
Cursor *string `json:"cursor"`
17
+
}
18
+
19
+
func (s *Server) GetManyToMany(w http.ResponseWriter, r *http.Request) {
20
+
query := r.URL.Query()
21
+
22
+
subject := query.Get("subject")
23
+
source := query.Get("source")
24
+
linkDids := query["linkDid"]
25
+
otherSubjects := query["otherSubject"]
26
+
27
+
collection, path, err := parseSource(source)
28
+
if err != nil {
29
+
http.Error(w, err.Error(), http.StatusBadRequest)
30
+
return
31
+
}
32
+
33
+
pathToOther, err := normalizePath(query.Get("pathToOther"))
34
+
if err != nil {
35
+
http.Error(w, "path_to_other: "+err.Error(), http.StatusBadRequest)
36
+
return
37
+
}
38
+
39
+
limit := uint64(100)
40
+
if raw := query.Get("limit"); raw != "" {
41
+
parsed, err := strconv.ParseUint(raw, 10, 64)
42
+
if err != nil || parsed == 0 || parsed > 1000 {
43
+
http.Error(w, "limit must be a number between 1 and 1000", http.StatusBadRequest)
44
+
return
45
+
}
46
+
limit = parsed
47
+
}
48
+
49
+
var after int64
50
+
if raw := query.Get("cursor"); raw != "" {
51
+
decoded, err := base64.StdEncoding.DecodeString(raw)
52
+
if err != nil {
53
+
http.Error(w, "invalid cursor", http.StatusBadRequest)
54
+
return
55
+
}
56
+
after, err = strconv.ParseInt(string(decoded), 10, 64)
57
+
if err != nil {
58
+
http.Error(w, "invalid cursor", http.StatusBadRequest)
59
+
return
60
+
}
61
+
}
62
+
63
+
total, items, err := s.Store.ManyToMany(r.Context(), subject, collection, path, pathToOther, linkDids, otherSubjects, after, limit)
64
+
if err != nil {
65
+
log.Println("many to many:", err)
66
+
http.Error(w, "internal error", http.StatusInternalServerError)
67
+
return
68
+
}
69
+
70
+
var cursor *string
71
+
if uint64(len(items)) == limit {
72
+
last := items[len(items)-1]
73
+
encoded := base64.StdEncoding.EncodeToString([]byte(strconv.FormatInt(last.LinkRecord.ID, 10)))
74
+
cursor = &encoded
75
+
}
76
+
77
+
w.Header().Set("Content-Type", "application/json")
78
+
json.NewEncoder(w).Encode(manyToManyResponse{Total: total, Items: items, Cursor: cursor})
79
+
}
+73
internal/api/many_to_many_counts.go
+73
internal/api/many_to_many_counts.go
···
1
+
package api
2
+
3
+
import (
4
+
"encoding/base64"
5
+
"encoding/json"
6
+
"log"
7
+
"net/http"
8
+
"strconv"
9
+
10
+
"github.com/alyraffauf/asterism/internal/store"
11
+
)
12
+
13
+
type manyToManyCountsResponse struct {
14
+
CountsByOtherSubject []store.OtherSubjectCount `json:"counts_by_other_subject"`
15
+
Cursor *string `json:"cursor"`
16
+
}
17
+
18
+
func (s *Server) GetManyToManyCounts(w http.ResponseWriter, r *http.Request) {
19
+
query := r.URL.Query()
20
+
21
+
subject := query.Get("subject")
22
+
source := query.Get("source")
23
+
linkDids := query["did"]
24
+
otherSubjects := query["otherSubject"]
25
+
26
+
collection, path, err := parseSource(source)
27
+
if err != nil {
28
+
http.Error(w, err.Error(), http.StatusBadRequest)
29
+
return
30
+
}
31
+
32
+
pathToOther, err := normalizePath(query.Get("pathToOther"))
33
+
if err != nil {
34
+
http.Error(w, "pathToOther: "+err.Error(), http.StatusBadRequest)
35
+
return
36
+
}
37
+
38
+
limit := uint64(100)
39
+
if raw := query.Get("limit"); raw != "" {
40
+
parsed, err := strconv.ParseUint(raw, 10, 64)
41
+
if err != nil || parsed == 0 || parsed > 1000 {
42
+
http.Error(w, "limit must be a number between 1 and 1000", http.StatusBadRequest)
43
+
return
44
+
}
45
+
limit = parsed
46
+
}
47
+
48
+
after := ""
49
+
if raw := query.Get("cursor"); raw != "" {
50
+
decoded, err := base64.StdEncoding.DecodeString(raw)
51
+
if err != nil {
52
+
http.Error(w, "invalid cursor", http.StatusBadRequest)
53
+
return
54
+
}
55
+
after = string(decoded)
56
+
}
57
+
58
+
counts, err := s.Store.ManyToManyCounts(r.Context(), subject, collection, path, pathToOther, linkDids, otherSubjects, after, limit)
59
+
if err != nil {
60
+
log.Println("many to many counts:", err)
61
+
http.Error(w, "internal error", http.StatusInternalServerError)
62
+
return
63
+
}
64
+
65
+
var cursor *string
66
+
if uint64(len(counts)) == limit {
67
+
encoded := base64.StdEncoding.EncodeToString([]byte(counts[len(counts)-1].Subject))
68
+
cursor = &encoded
69
+
}
70
+
71
+
w.Header().Set("Content-Type", "application/json")
72
+
json.NewEncoder(w).Encode(manyToManyCountsResponse{CountsByOtherSubject: counts, Cursor: cursor})
73
+
}
+2
internal/api/server.go
+2
internal/api/server.go
···
15
15
mux.HandleFunc("GET /xrpc/blue.microcosm.links.getBacklinksCount", s.GetBacklinksCount)
16
16
mux.HandleFunc("GET /xrpc/blue.microcosm.links.getBacklinkDids", s.GetBacklinkDids)
17
17
mux.HandleFunc("GET /xrpc/blue.microcosm.links.getBacklinks", s.GetBacklinks)
18
+
mux.HandleFunc("GET /xrpc/blue.microcosm.links.getManyToMany", s.GetManyToMany)
19
+
mux.HandleFunc("GET /xrpc/blue.microcosm.links.getManyToManyCounts", s.GetManyToManyCounts)
18
20
19
21
return http.ListenAndServe(addr, mux)
20
22
}
+13
-4
internal/api/source.go
+13
-4
internal/api/source.go
···
14
14
return "", "", fmt.Errorf("source is missing a collection before ':'")
15
15
}
16
16
17
+
path, err = normalizePath(rawPath)
18
+
if err != nil {
19
+
return "", "", fmt.Errorf("source path: %w", err)
20
+
}
21
+
22
+
return collection, path, nil
23
+
}
24
+
25
+
func normalizePath(rawPath string) (string, error) {
17
26
switch {
18
27
case rawPath == "":
19
-
return "", "", fmt.Errorf("source is missing a path after ':'")
28
+
return "", fmt.Errorf("path is empty")
20
29
case rawPath == ".":
21
-
return collection, ".", nil
30
+
return ".", nil
22
31
case strings.HasPrefix(rawPath, "."):
23
-
return "", "", fmt.Errorf("source path must not start with '.'")
32
+
return "", fmt.Errorf("path must not start with '.'")
24
33
default:
25
-
return collection, "." + rawPath, nil
34
+
return "." + rawPath, nil
26
35
}
27
36
}
+127
internal/store/query.go
+127
internal/store/query.go
···
13
13
RecordKey string `json:"rkey"`
14
14
}
15
15
16
+
type ManyToManyItem struct {
17
+
LinkRecord Record `json:"linkRecord"`
18
+
OtherSubject string `json:"otherSubject"`
19
+
}
20
+
21
+
22
+
type OtherSubjectCount struct {
23
+
Subject string `json:"subject"`
24
+
Total uint64 `json:"total"`
25
+
Distinct uint64 `json:"distinct"`
26
+
}
27
+
16
28
func (s *Store) CountBacklinks(ctx context.Context, target, collection, fieldPath string) (uint64, error) {
17
29
var total uint64
18
30
···
122
134
123
135
return total, records, nil
124
136
}
137
+
138
+
func (s *Store) ManyToMany(ctx context.Context, target, collection, fieldPath, pathToOther string, linkDids, otherSubjects []string, after int64, limit uint64) (total uint64, items []ManyToManyItem, err error) {
139
+
where := `a.target = ? AND a.collection = ? AND a.field_path = ? AND b.field_path = ?`
140
+
args := []any{target, collection, fieldPath, pathToOther}
141
+
142
+
if len(linkDids) > 0 {
143
+
placeholders := make([]string, len(linkDids))
144
+
for i, did := range linkDids {
145
+
placeholders[i] = "?"
146
+
args = append(args, did)
147
+
}
148
+
where += ` AND a.actor_did IN (` + strings.Join(placeholders, ", ") + `)`
149
+
}
150
+
151
+
if len(otherSubjects) > 0 {
152
+
placeholders := make([]string, len(otherSubjects))
153
+
for i, subj := range otherSubjects {
154
+
placeholders[i] = "?"
155
+
args = append(args, subj)
156
+
}
157
+
where += ` AND b.target IN (` + strings.Join(placeholders, ", ") + `)`
158
+
}
159
+
160
+
joinClause := ` FROM links a JOIN links b
161
+
ON a.actor_did = b.actor_did AND a.collection = b.collection AND a.record_key = b.record_key
162
+
WHERE ` + where
163
+
164
+
if err := s.readDB.QueryRowContext(ctx, `SELECT COUNT(*)`+joinClause, args...).Scan(&total); err != nil {
165
+
return 0, nil, fmt.Errorf("count many to many: %w", err)
166
+
}
167
+
168
+
listQuery := `SELECT a.id, a.actor_did, a.collection, a.record_key, b.target` + joinClause
169
+
listArgs := append([]any{}, args...)
170
+
171
+
if after != 0 {
172
+
listQuery += ` AND a.id > ?`
173
+
listArgs = append(listArgs, after)
174
+
}
175
+
176
+
listQuery += ` ORDER BY a.id ASC LIMIT ?`
177
+
listArgs = append(listArgs, limit)
178
+
179
+
rows, err := s.readDB.QueryContext(ctx, listQuery, listArgs...)
180
+
if err != nil {
181
+
return 0, nil, fmt.Errorf("query many to many: %w", err)
182
+
}
183
+
defer rows.Close()
184
+
185
+
for rows.Next() {
186
+
var item ManyToManyItem
187
+
if err := rows.Scan(&item.LinkRecord.ID, &item.LinkRecord.ActorDid, &item.LinkRecord.Collection, &item.LinkRecord.RecordKey, &item.OtherSubject); err != nil {
188
+
return 0, nil, fmt.Errorf("scan many to many item: %w", err)
189
+
}
190
+
items = append(items, item)
191
+
}
192
+
if err := rows.Err(); err != nil {
193
+
return 0, nil, fmt.Errorf("iterate many to many: %w", err)
194
+
}
195
+
196
+
return total, items, nil
197
+
}
198
+
199
+
func (s *Store) ManyToManyCounts(ctx context.Context, target, collection, fieldPath, pathToOther string, linkDids, otherSubjects []string, after string, limit uint64) (counts []OtherSubjectCount, err error) {
200
+
where := `a.target = ? AND a.collection = ? AND a.field_path = ? AND b.field_path = ?`
201
+
args := []any{target, collection, fieldPath, pathToOther}
202
+
203
+
if len(linkDids) > 0 {
204
+
placeholders := make([]string, len(linkDids))
205
+
for i, did := range linkDids {
206
+
placeholders[i] = "?"
207
+
args = append(args, did)
208
+
}
209
+
where += ` AND a.actor_did IN (` + strings.Join(placeholders, ", ") + `)`
210
+
}
211
+
212
+
if len(otherSubjects) > 0 {
213
+
placeholders := make([]string, len(otherSubjects))
214
+
for i, subj := range otherSubjects {
215
+
placeholders[i] = "?"
216
+
args = append(args, subj)
217
+
}
218
+
where += ` AND b.target IN (` + strings.Join(placeholders, ", ") + `)`
219
+
}
220
+
221
+
where += ` AND b.target > ?`
222
+
args = append(args, after)
223
+
224
+
query := `SELECT b.target, COUNT(*), COUNT(DISTINCT a.actor_did)
225
+
FROM links a JOIN links b
226
+
ON a.actor_did = b.actor_did AND a.collection = b.collection AND a.record_key = b.record_key
227
+
WHERE ` + where + `
228
+
GROUP BY b.target
229
+
ORDER BY b.target
230
+
LIMIT ?`
231
+
args = append(args, limit)
232
+
233
+
rows, err := s.readDB.QueryContext(ctx, query, args...)
234
+
if err != nil {
235
+
return nil, fmt.Errorf("query many to many counts: %w", err)
236
+
}
237
+
defer rows.Close()
238
+
239
+
for rows.Next() {
240
+
var c OtherSubjectCount
241
+
if err := rows.Scan(&c.Subject, &c.Total, &c.Distinct); err != nil {
242
+
return nil, fmt.Errorf("scan many to many count: %w", err)
243
+
}
244
+
counts = append(counts, c)
245
+
}
246
+
if err := rows.Err(); err != nil {
247
+
return nil, fmt.Errorf("iterate many to many counts: %w", err)
248
+
}
249
+
250
+
return counts, nil
251
+
}