From 05c48e2e97703e697517e2fe9e692b9a0a20bda6 Mon Sep 17 00:00:00 2001 From: Aly Raffauf Date: Fri, 3 Jul 2026 18:58:45 -0400 Subject: [PATCH] handle account deletion events --- README.md | 77 ++++++++++++++++++------------------ internal/firehose/account.go | 20 ++++++++++ internal/firehose/stream.go | 4 +- internal/store/cursor.go | 1 - internal/store/links.go | 8 ++++ internal/store/query.go | 1 - 6 files changed, 70 insertions(+), 41 deletions(-) create mode 100644 internal/firehose/account.go diff --git a/README.md b/README.md index b9a95a4..5783e40 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ Constellation is vital community infrastructure, and many ATProto apps have been 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. -> **Early stage.** Functional but *very* incomplete. APIs may change, backfill is rudimentary, and several features are not yet implemented. See [Roadmap](#roadmap). +> **Early stage.** Functional but _very_ incomplete. APIs may change, backfill is rudimentary, and several features are not yet implemented. See [Roadmap](#roadmap). ## What it does @@ -51,13 +51,13 @@ go run ./cmd/asterism/ Every flag can also be set with an environment variable. -| Flag | Environment variable | Default | Description | -|---|---|---|---| -| `--collections` | `ASTERISM_COLLECTIONS` | empty | Comma-separated collection NSIDs to index. Empty means all collections. | -| `--backfill` | `ASTERISM_BACKFILL` | false | Backfill existing repos for configured collections on startup. | -| `--database` | `ASTERISM_DATABASE` | `asterism.db` | SQLite database path. | -| `--listen` | `ASTERISM_LISTEN` | `:8081` | HTTP API listen address. | -| `--relay` | `ASTERISM_RELAY` | `relay1.us-east.bsky.network` | Relay host. Asterism derives the Firehose websocket and relay HTTP API URLs from this host. | +| Flag | Environment variable | Default | Description | +| --------------- | ---------------------- | ----------------------------- | ------------------------------------------------------------------------------------------- | +| `--collections` | `ASTERISM_COLLECTIONS` | empty | Comma-separated collection NSIDs to index. Empty means all collections. | +| `--backfill` | `ASTERISM_BACKFILL` | false | Backfill existing repos for configured collections on startup. | +| `--database` | `ASTERISM_DATABASE` | `asterism.db` | SQLite database path. | +| `--listen` | `ASTERISM_LISTEN` | `:8081` | HTTP API listen address. | +| `--relay` | `ASTERISM_RELAY` | `relay1.us-east.bsky.network` | Relay host. Asterism derives the Firehose websocket and relay HTTP API URLs from this host. | For example: @@ -89,12 +89,12 @@ Response: `{"total": 42}` List distinct DIDs that have records linking to a subject. Paginated. -| Parameter | Description | -|---|---| -| `subject` | Target AT-URI, DID, or URL (required) | -| `source` | Collection and field path, e.g. `app.bsky.feed.like:subject.uri` (required) | -| `limit` | Page size, 1–1000 (default 100) | -| `cursor` | Pagination cursor from previous response | +| Parameter | Description | +| --------- | --------------------------------------------------------------------------- | +| `subject` | Target AT-URI, DID, or URL (required) | +| `source` | Collection and field path, e.g. `app.bsky.feed.like:subject.uri` (required) | +| `limit` | Page size, 1–1000 (default 100) | +| `cursor` | Pagination cursor from previous response | Response: `{"total": 42, "linking_dids": ["did:plc:..."], "cursor": "..."}` @@ -102,14 +102,14 @@ Response: `{"total": 42, "linking_dids": ["did:plc:..."], "cursor": "..."}` List source records linking to a subject. Paginated. -| Parameter | Description | -|---|---| -| `subject` | Target AT-URI, DID, or URL (required) | -| `source` | Collection and field path (required) | -| `did` | Filter to specific actor DIDs (repeatable) | -| `limit` | Page size, 1–1000 (default 100) | +| Parameter | Description | +| --------- | ----------------------------------------------- | +| `subject` | Target AT-URI, DID, or URL (required) | +| `source` | Collection and field path (required) | +| `did` | Filter to specific actor DIDs (repeatable) | +| `limit` | Page size, 1–1000 (default 100) | | `reverse` | Return links in ascending order (default false) | -| `cursor` | Pagination cursor from previous response | +| `cursor` | Pagination cursor from previous response | Response: `{"total": 42, "records": [{"did": "...", "collection": "...", "rkey": "..."}], "cursor": "..."}` @@ -119,15 +119,15 @@ Records identify the linking record by DID, collection, and rkey. Clients must h 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. -| Parameter | Description | -|---|---| -| `subject` | Target AT-URI, DID, or URL (required) | -| `source` | Collection and field path (required) | -| `pathToOther` | Second field path on the same source record (required) | -| `linkDid` | Filter to specific linking-record DIDs (repeatable) | +| Parameter | Description | +| -------------- | ------------------------------------------------------ | +| `subject` | Target AT-URI, DID, or URL (required) | +| `source` | Collection and field path (required) | +| `pathToOther` | Second field path on the same source record (required) | +| `linkDid` | Filter to specific linking-record DIDs (repeatable) | | `otherSubject` | Filter to specific secondary link targets (repeatable) | -| `limit` | Page size, 1–1000 (default 100) | -| `cursor` | Pagination cursor from previous response | +| `limit` | Page size, 1–1000 (default 100) | +| `cursor` | Pagination cursor from previous response | Response: `{"total": 42, "items": [{"linkRecord": {"did": "...", "collection": "...", "rkey": "..."}, "otherSubject": "..."}], "cursor": "..."}` @@ -135,15 +135,15 @@ Response: `{"total": 42, "items": [{"linkRecord": {"did": "...", "collection": " 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. -| Parameter | Description | -|---|---| -| `subject` | Target AT-URI, DID, or URL (required) | -| `source` | Collection and field path (required) | -| `pathToOther` | Second field path on the same source record (required) | -| `did` | Filter to specific linking-record DIDs (repeatable) | +| Parameter | Description | +| -------------- | ------------------------------------------------------ | +| `subject` | Target AT-URI, DID, or URL (required) | +| `source` | Collection and field path (required) | +| `pathToOther` | Second field path on the same source record (required) | +| `did` | Filter to specific linking-record DIDs (repeatable) | | `otherSubject` | Filter to specific secondary link targets (repeatable) | -| `limit` | Page size, 1–1000 (default 100) | -| `cursor` | Pagination cursor from previous response | +| `limit` | Page size, 1–1000 (default 100) | +| `cursor` | Pagination cursor from previous response | Response: `{"counts_by_other_subject": [{"subject": "...", "total": 42, "distinct": 12}], "cursor": "..."}` @@ -155,7 +155,8 @@ Note the DID filter parameter is `did` here, not `linkDid` like `getManyToMany` - [x] Full Constellation API parity (`getBacklinksCount`, `getBacklinkDids`, `getBacklinks`, `getManyToMany`, `getManyToManyCounts`) - [x] Configurable listen address, database path, relay host, and startup backfill -- [ ] Account deletion and deactivation handling +- [x] Account deletion handling +- [ ] Account deactivation handling - [x] Graceful shutdown and Firehose reconnect **Medium term** diff --git a/internal/firehose/account.go b/internal/firehose/account.go new file mode 100644 index 0000000..f40f30f --- /dev/null +++ b/internal/firehose/account.go @@ -0,0 +1,20 @@ +package firehose + +import ( + "context" + "fmt" + + "github.com/bluesky-social/indigo/api/atproto" +) + +func (c *Consumer) HandleAccount(ctx context.Context, event *atproto.SyncSubscribeRepos_Account) error { + if err := c.Store.SaveCursor(ctx, event.Seq); err != nil { + fmt.Println("could not save cursor:", err) + } + + if event.Status != nil && *event.Status == "deleted" { + return c.Store.DeleteAllLinks(ctx, event.Did) + } + + return nil +} diff --git a/internal/firehose/stream.go b/internal/firehose/stream.go index 8a4cfb4..b879082 100644 --- a/internal/firehose/stream.go +++ b/internal/firehose/stream.go @@ -30,7 +30,6 @@ func (c *Consumer) Run(ctx context.Context, relayURL string, logger *slog.Logger dialURL = fmt.Sprintf("%s?cursor=%d", relayURL, cursor) } - conn, _, err := websocket.DefaultDialer.DialContext(ctx, dialURL, http.Header{}) if err != nil { logger.Warn("dial failed", "err", err, "retry in", backoff) @@ -56,6 +55,9 @@ func (c *Consumer) stream(ctx context.Context, conn *websocket.Conn, logger *slo RepoCommit: func(event *atproto.SyncSubscribeRepos_Commit) error { return c.HandleCommit(ctx, event) }, + RepoAccount: func(event *atproto.SyncSubscribeRepos_Account) error { + return c.HandleAccount(ctx, event) + }, } scheduler := sequential.NewScheduler("asterism", callbacks.EventHandler) diff --git a/internal/store/cursor.go b/internal/store/cursor.go index 1dc6a7d..7406aa8 100644 --- a/internal/store/cursor.go +++ b/internal/store/cursor.go @@ -18,7 +18,6 @@ func (s *Store) SaveCursor(ctx context.Context, seq int64) error { return nil } - func (s *Store) GetCursor(ctx context.Context) (int64, error) { var seq int64 err := s.readDB.QueryRowContext(ctx, `SELECT seq FROM cursor WHERE id = 0`).Scan(&seq) diff --git a/internal/store/links.go b/internal/store/links.go index f1f41f3..4f92d0f 100644 --- a/internal/store/links.go +++ b/internal/store/links.go @@ -34,6 +34,14 @@ func (s *Store) DeleteLinks(ctx context.Context, actorDid, collection, recordKey return tx.Commit() } +func (s *Store) DeleteAllLinks(ctx context.Context, actorDid string) error { + _, err := s.writeDB.ExecContext(ctx, `DELETE FROM links WHERE actor_did = ?`, actorDid) + if err != nil { + return fmt.Errorf("delete all links: %w", err) + } + return nil +} + func (s *Store) SaveLinks(ctx context.Context, actorDid, collection, recordKey string, links []backlink.Link) error { tx, err := s.writeDB.BeginTx(ctx, nil) if err != nil { diff --git a/internal/store/query.go b/internal/store/query.go index f99fda9..0905970 100644 --- a/internal/store/query.go +++ b/internal/store/query.go @@ -18,7 +18,6 @@ type ManyToManyItem struct { OtherSubject string `json:"otherSubject"` } - type OtherSubjectCount struct { Subject string `json:"subject"` Total uint64 `json:"total"` -- 2.51.2