Low-latency AT Protocol interaction indexer.
firehose: add cursor + resume logic; store: serialize writes, free up reads
Aly Raffauf
(Jul 3, 2026, 2:30 PM EDT)
a254492a
7b1932df
+77
-15
+4
internal/firehose/commit.go
+4
internal/firehose/commit.go
···
14
14
15
15
func (c *Consumer) HandleCommit(ctx context.Context, event *atproto.SyncSubscribeRepos_Commit) error {
16
16
// fmt.Println("repo:", event.Repo, "commit:", event.Rev)
17
+
//
18
+
if err := c.Store.SaveCursor(ctx, event.Seq); err != nil {
19
+
fmt.Println("could not save cursor:", err)
20
+
}
17
21
18
22
if event.TooBig {
19
23
go func() {
+11
-1
internal/firehose/stream.go
+11
-1
internal/firehose/stream.go
···
2
2
3
3
import (
4
4
"context"
5
+
"fmt"
5
6
"log/slog"
6
7
"net/http"
7
8
"time"
···
21
22
backoff := minBackoff
22
23
23
24
for {
24
-
conn, _, err := websocket.DefaultDialer.DialContext(ctx, relayURL, http.Header{})
25
+
dialURL := relayURL
26
+
27
+
if cursor, err := c.Store.GetCursor(ctx); err != nil {
28
+
logger.Warn("could not load cursor, starting from live tip", "err", err)
29
+
} else if cursor > 0 {
30
+
dialURL = fmt.Sprintf("%s?cursor=%d", relayURL, cursor)
31
+
}
32
+
33
+
34
+
conn, _, err := websocket.DefaultDialer.DialContext(ctx, dialURL, http.Header{})
25
35
if err != nil {
26
36
logger.Warn("dial failed", "err", err, "retry in", backoff)
27
37
} else {
+34
internal/store/cursor.go
+34
internal/store/cursor.go
···
1
+
package store
2
+
3
+
import (
4
+
"context"
5
+
"database/sql"
6
+
"errors"
7
+
"fmt"
8
+
)
9
+
10
+
func (s *Store) SaveCursor(ctx context.Context, seq int64) error {
11
+
_, err := s.writeDB.ExecContext(ctx,
12
+
`INSERT INTO cursor (id, seq) VALUES (0, ?) ON CONFLICT (id) DO UPDATE SET seq = excluded.seq`, seq,
13
+
)
14
+
15
+
if err != nil {
16
+
return fmt.Errorf("save cursor: %w", err)
17
+
}
18
+
return nil
19
+
}
20
+
21
+
22
+
func (s *Store) GetCursor(ctx context.Context) (int64, error) {
23
+
var seq int64
24
+
err := s.readDB.QueryRowContext(ctx, `SELECT seq FROM cursor WHERE id = 0`).Scan(&seq)
25
+
if errors.Is(err, sql.ErrNoRows) {
26
+
return 0, nil
27
+
}
28
+
29
+
if err != nil {
30
+
return 0, fmt.Errorf("get cursor: %w", err)
31
+
}
32
+
33
+
return seq, nil
34
+
}
+2
-2
internal/store/links.go
+2
-2
internal/store/links.go
···
21
21
}
22
22
23
23
func (s *Store) DeleteLinks(ctx context.Context, actorDid, collection, recordKey string) error {
24
-
tx, err := s.db.BeginTx(ctx, nil)
24
+
tx, err := s.writeDB.BeginTx(ctx, nil)
25
25
if err != nil {
26
26
return fmt.Errorf("begin transaction: %w", err)
27
27
}
···
35
35
}
36
36
37
37
func (s *Store) SaveLinks(ctx context.Context, actorDid, collection, recordKey string, links []backlink.Link) error {
38
-
tx, err := s.db.BeginTx(ctx, nil)
38
+
tx, err := s.writeDB.BeginTx(ctx, nil)
39
39
if err != nil {
40
40
return fmt.Errorf("begin transaction: %w", err)
41
41
}
+5
-5
internal/store/query.go
+5
-5
internal/store/query.go
···
16
16
func (s *Store) CountBacklinks(ctx context.Context, target, collection, fieldPath string) (uint64, error) {
17
17
var total uint64
18
18
19
-
err := s.db.QueryRowContext(ctx,
19
+
err := s.readDB.QueryRowContext(ctx,
20
20
`SELECT COUNT(*) FROM links WHERE target = ? AND collection = ? AND field_path = ?`,
21
21
target, collection, fieldPath,
22
22
).Scan(&total)
···
28
28
}
29
29
30
30
func (s *Store) DistinctBacklinkDids(ctx context.Context, target, collection, fieldPath string, after string, limit uint64) (total uint64, dids []string, err error) {
31
-
err = s.db.QueryRowContext(ctx,
31
+
err = s.readDB.QueryRowContext(ctx,
32
32
`SELECT COUNT(DISTINCT actor_did) FROM links WHERE target = ? AND collection = ? AND field_path = ?`,
33
33
target, collection, fieldPath,
34
34
).Scan(&total)
···
36
36
return 0, nil, fmt.Errorf("count distinct dids: %w", err)
37
37
}
38
38
39
-
rows, err := s.db.QueryContext(ctx,
39
+
rows, err := s.readDB.QueryContext(ctx,
40
40
`SELECT DISTINCT actor_did FROM links
41
41
WHERE target = ? AND collection = ? AND field_path = ? AND actor_did > ?
42
42
ORDER BY actor_did LIMIT ?`,
···
73
73
}
74
74
where += `AND actor_did IN (` + strings.Join(placeholders, ", ") + `)`
75
75
76
-
err = s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM links WHERE `+where, args...).Scan(&total)
76
+
err = s.readDB.QueryRowContext(ctx, `SELECT COUNT(*) FROM links WHERE `+where, args...).Scan(&total)
77
77
if err != nil {
78
78
return 0, nil, fmt.Errorf("count backlinks: %w", err)
79
79
}
···
103
103
}
104
104
listArgs = append(listArgs, limit)
105
105
106
-
rows, err := s.db.QueryContext(ctx, query, listArgs...)
106
+
rows, err := s.readDB.QueryContext(ctx, query, listArgs...)
107
107
if err != nil {
108
108
return 0, nil, fmt.Errorf("query backlinks: %w", err)
109
109
}
+21
-7
internal/store/store.go
+21
-7
internal/store/store.go
···
8
8
)
9
9
10
10
type Store struct {
11
-
db *sql.DB
11
+
writeDB *sql.DB
12
+
readDB *sql.DB
12
13
}
13
14
14
15
const schema = `
···
25
26
);
26
27
CREATE INDEX IF NOT EXISTS idx_links_target ON links(target);
27
28
CREATE INDEX IF NOT EXISTS idx_links_source ON links(actor_did, collection, record_key);
29
+
CREATE TABLE IF NOT EXISTS cursor (
30
+
id INTEGER PRIMARY KEY CHECK (id = 0),
31
+
seq INTEGER NOT NULL
32
+
);
28
33
`
29
34
30
35
func Open(path string) (*Store, error) {
31
-
db, err := sql.Open("sqlite", path)
36
+
writeDB, err := sql.Open("sqlite", path)
32
37
if err != nil {
33
-
return nil, fmt.Errorf("open database: %w", err)
38
+
return nil, fmt.Errorf("open write database: %w", err)
34
39
}
35
40
36
-
if _, err := db.Exec("PRAGMA journal_mode=WAL;"); err != nil {
41
+
if _, err := writeDB.Exec("PRAGMA journal_mode=WAL;"); err != nil {
37
42
return nil, fmt.Errorf("set journal mode: %w", err)
38
43
}
39
-
if _, err := db.Exec("PRAGMA busy_timeout=5000;"); err != nil {
44
+
if _, err := writeDB.Exec("PRAGMA busy_timeout=5000;"); err != nil {
40
45
return nil, fmt.Errorf("set busy timeout: %w", err)
41
46
}
47
+
writeDB.SetMaxOpenConns(1)
42
48
43
-
if _, err := db.Exec(schema); err != nil {
49
+
if _, err := writeDB.Exec(schema); err != nil {
44
50
return nil, fmt.Errorf("create schema: %w", err)
45
51
}
46
52
47
-
return &Store{db: db}, nil
53
+
readDB, err := sql.Open("sqlite", path)
54
+
if err != nil {
55
+
return nil, fmt.Errorf("open read database: %w", err)
56
+
}
57
+
if _, err := readDB.Exec("PRAGMA busy_timeout=5000;"); err != nil {
58
+
return nil, fmt.Errorf("set read busy timeout: %w", err)
59
+
}
60
+
61
+
return &Store{writeDB: writeDB, readDB: readDB}, nil
48
62
}