Low-latency AT Protocol interaction indexer.
handle account deletion events
Aly Raffauf
(Jul 3, 2026, 6:58 PM EDT)
05c48e2e
37343725
+70
-41
+39
-38
README.md
+39
-38
README.md
···
6
6
7
7
Asterism, meanwhile, consumes cryptographically verifiable events directly from the Firehose, and filters them by the collection of your choice. There's no Jetstream in the middle, meaning fewer moving parts. And while Asterism has significant bandwidth requirements, the filtered index is significantly smaller and scales with your application, not with the network.
8
8
9
-
> **Early stage.** Functional but *very* incomplete. APIs may change, backfill is rudimentary, and several features are not yet implemented. See [Roadmap](#roadmap).
9
+
> **Early stage.** Functional but _very_ incomplete. APIs may change, backfill is rudimentary, and several features are not yet implemented. See [Roadmap](#roadmap).
10
10
11
11
## What it does
12
12
···
51
51
52
52
Every flag can also be set with an environment variable.
53
53
54
-
| Flag | Environment variable | Default | Description |
55
-
|---|---|---|---|
56
-
| `--collections` | `ASTERISM_COLLECTIONS` | empty | Comma-separated collection NSIDs to index. Empty means all collections. |
57
-
| `--backfill` | `ASTERISM_BACKFILL` | false | Backfill existing repos for configured collections on startup. |
58
-
| `--database` | `ASTERISM_DATABASE` | `asterism.db` | SQLite database path. |
59
-
| `--listen` | `ASTERISM_LISTEN` | `:8081` | HTTP API listen address. |
60
-
| `--relay` | `ASTERISM_RELAY` | `relay1.us-east.bsky.network` | Relay host. Asterism derives the Firehose websocket and relay HTTP API URLs from this host. |
54
+
| Flag | Environment variable | Default | Description |
55
+
| --------------- | ---------------------- | ----------------------------- | ------------------------------------------------------------------------------------------- |
56
+
| `--collections` | `ASTERISM_COLLECTIONS` | empty | Comma-separated collection NSIDs to index. Empty means all collections. |
57
+
| `--backfill` | `ASTERISM_BACKFILL` | false | Backfill existing repos for configured collections on startup. |
58
+
| `--database` | `ASTERISM_DATABASE` | `asterism.db` | SQLite database path. |
59
+
| `--listen` | `ASTERISM_LISTEN` | `:8081` | HTTP API listen address. |
60
+
| `--relay` | `ASTERISM_RELAY` | `relay1.us-east.bsky.network` | Relay host. Asterism derives the Firehose websocket and relay HTTP API URLs from this host. |
61
61
62
62
For example:
63
63
···
89
89
90
90
List distinct DIDs that have records linking to a subject. Paginated.
91
91
92
-
| Parameter | Description |
93
-
|---|---|
94
-
| `subject` | Target AT-URI, DID, or URL (required) |
95
-
| `source` | Collection and field path, e.g. `app.bsky.feed.like:subject.uri` (required) |
96
-
| `limit` | Page size, 1–1000 (default 100) |
97
-
| `cursor` | Pagination cursor from previous response |
92
+
| Parameter | Description |
93
+
| --------- | --------------------------------------------------------------------------- |
94
+
| `subject` | Target AT-URI, DID, or URL (required) |
95
+
| `source` | Collection and field path, e.g. `app.bsky.feed.like:subject.uri` (required) |
96
+
| `limit` | Page size, 1–1000 (default 100) |
97
+
| `cursor` | Pagination cursor from previous response |
98
98
99
99
Response: `{"total": 42, "linking_dids": ["did:plc:..."], "cursor": "..."}`
100
100
···
102
102
103
103
List source records linking to a subject. Paginated.
104
104
105
-
| Parameter | Description |
106
-
|---|---|
107
-
| `subject` | Target AT-URI, DID, or URL (required) |
108
-
| `source` | Collection and field path (required) |
109
-
| `did` | Filter to specific actor DIDs (repeatable) |
110
-
| `limit` | Page size, 1–1000 (default 100) |
105
+
| Parameter | Description |
106
+
| --------- | ----------------------------------------------- |
107
+
| `subject` | Target AT-URI, DID, or URL (required) |
108
+
| `source` | Collection and field path (required) |
109
+
| `did` | Filter to specific actor DIDs (repeatable) |
110
+
| `limit` | Page size, 1–1000 (default 100) |
111
111
| `reverse` | Return links in ascending order (default false) |
112
-
| `cursor` | Pagination cursor from previous response |
112
+
| `cursor` | Pagination cursor from previous response |
113
113
114
114
Response: `{"total": 42, "records": [{"did": "...", "collection": "...", "rkey": "..."}], "cursor": "..."}`
115
115
···
119
119
120
120
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.
121
121
122
-
| Parameter | Description |
123
-
|---|---|
124
-
| `subject` | Target AT-URI, DID, or URL (required) |
125
-
| `source` | Collection and field path (required) |
126
-
| `pathToOther` | Second field path on the same source record (required) |
127
-
| `linkDid` | Filter to specific linking-record DIDs (repeatable) |
122
+
| Parameter | Description |
123
+
| -------------- | ------------------------------------------------------ |
124
+
| `subject` | Target AT-URI, DID, or URL (required) |
125
+
| `source` | Collection and field path (required) |
126
+
| `pathToOther` | Second field path on the same source record (required) |
127
+
| `linkDid` | Filter to specific linking-record DIDs (repeatable) |
128
128
| `otherSubject` | Filter to specific secondary link targets (repeatable) |
129
-
| `limit` | Page size, 1–1000 (default 100) |
130
-
| `cursor` | Pagination cursor from previous response |
129
+
| `limit` | Page size, 1–1000 (default 100) |
130
+
| `cursor` | Pagination cursor from previous response |
131
131
132
132
Response: `{"total": 42, "items": [{"linkRecord": {"did": "...", "collection": "...", "rkey": "..."}, "otherSubject": "..."}], "cursor": "..."}`
133
133
···
135
135
136
136
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.
137
137
138
-
| Parameter | Description |
139
-
|---|---|
140
-
| `subject` | Target AT-URI, DID, or URL (required) |
141
-
| `source` | Collection and field path (required) |
142
-
| `pathToOther` | Second field path on the same source record (required) |
143
-
| `did` | Filter to specific linking-record DIDs (repeatable) |
138
+
| Parameter | Description |
139
+
| -------------- | ------------------------------------------------------ |
140
+
| `subject` | Target AT-URI, DID, or URL (required) |
141
+
| `source` | Collection and field path (required) |
142
+
| `pathToOther` | Second field path on the same source record (required) |
143
+
| `did` | Filter to specific linking-record DIDs (repeatable) |
144
144
| `otherSubject` | Filter to specific secondary link targets (repeatable) |
145
-
| `limit` | Page size, 1–1000 (default 100) |
146
-
| `cursor` | Pagination cursor from previous response |
145
+
| `limit` | Page size, 1–1000 (default 100) |
146
+
| `cursor` | Pagination cursor from previous response |
147
147
148
148
Response: `{"counts_by_other_subject": [{"subject": "...", "total": 42, "distinct": 12}], "cursor": "..."}`
149
149
···
155
155
156
156
- [x] Full Constellation API parity (`getBacklinksCount`, `getBacklinkDids`, `getBacklinks`, `getManyToMany`, `getManyToManyCounts`)
157
157
- [x] Configurable listen address, database path, relay host, and startup backfill
158
-
- [ ] Account deletion and deactivation handling
158
+
- [x] Account deletion handling
159
+
- [ ] Account deactivation handling
159
160
- [x] Graceful shutdown and Firehose reconnect
160
161
161
162
**Medium term**
+20
internal/firehose/account.go
+20
internal/firehose/account.go
···
1
+
package firehose
2
+
3
+
import (
4
+
"context"
5
+
"fmt"
6
+
7
+
"github.com/bluesky-social/indigo/api/atproto"
8
+
)
9
+
10
+
func (c *Consumer) HandleAccount(ctx context.Context, event *atproto.SyncSubscribeRepos_Account) error {
11
+
if err := c.Store.SaveCursor(ctx, event.Seq); err != nil {
12
+
fmt.Println("could not save cursor:", err)
13
+
}
14
+
15
+
if event.Status != nil && *event.Status == "deleted" {
16
+
return c.Store.DeleteAllLinks(ctx, event.Did)
17
+
}
18
+
19
+
return nil
20
+
}
+3
-1
internal/firehose/stream.go
+3
-1
internal/firehose/stream.go
···
30
30
dialURL = fmt.Sprintf("%s?cursor=%d", relayURL, cursor)
31
31
}
32
32
33
-
34
33
conn, _, err := websocket.DefaultDialer.DialContext(ctx, dialURL, http.Header{})
35
34
if err != nil {
36
35
logger.Warn("dial failed", "err", err, "retry in", backoff)
···
55
54
callbacks := &events.RepoStreamCallbacks{
56
55
RepoCommit: func(event *atproto.SyncSubscribeRepos_Commit) error {
57
56
return c.HandleCommit(ctx, event)
57
+
},
58
+
RepoAccount: func(event *atproto.SyncSubscribeRepos_Account) error {
59
+
return c.HandleAccount(ctx, event)
58
60
},
59
61
}
60
62
-1
internal/store/cursor.go
-1
internal/store/cursor.go
+8
internal/store/links.go
+8
internal/store/links.go
···
34
34
return tx.Commit()
35
35
}
36
36
37
+
func (s *Store) DeleteAllLinks(ctx context.Context, actorDid string) error {
38
+
_, err := s.writeDB.ExecContext(ctx, `DELETE FROM links WHERE actor_did = ?`, actorDid)
39
+
if err != nil {
40
+
return fmt.Errorf("delete all links: %w", err)
41
+
}
42
+
return nil
43
+
}
44
+
37
45
func (s *Store) SaveLinks(ctx context.Context, actorDid, collection, recordKey string, links []backlink.Link) error {
38
46
tx, err := s.writeDB.BeginTx(ctx, nil)
39
47
if err != nil {