Low-latency AT Protocol interaction indexer.
tree-wide: switch fully to slog
Aly Raffauf
(Jul 4, 2026, 10:09 PM EDT)
acd6cdf8
5a0c7dcd
+32
-35
+5
-4
cmd/asterism/main.go
+5
-4
cmd/asterism/main.go
···
2
2
3
3
import (
4
4
"context"
5
-
"fmt"
6
5
"log/slog"
7
6
"strings"
8
7
···
70
69
71
70
directory := identity.DefaultDirectory()
72
71
73
-
server := &api.Server{Store: linkStore, Directory: directory}
72
+
server := &api.Server{Store: linkStore, Directory: directory, Logger: logger}
74
73
go func() {
75
74
if err := server.Run(cli.Listen); err != nil {
76
75
panic(err)
···
90
89
Client: &xrpc.Client{Host: relayHTTPHost(cli.Relay)},
91
90
Directory: directory,
92
91
Store: linkStore,
92
+
Logger: logger,
93
93
}
94
94
95
95
if cli.Backfill {
96
96
if len(collections) > 0 {
97
97
go func() {
98
98
if err := bf.Run(ctx, collections); err != nil {
99
-
fmt.Println("backfill error:", err)
99
+
logger.Error("backfill error", "err", err)
100
100
}
101
101
}()
102
102
}
···
107
107
Store: linkStore,
108
108
Directory: directory,
109
109
Backfill: bf,
110
+
Logger: logger,
110
111
}
111
112
112
-
if err := consumer.Run(ctx, relayURL, logger); err != nil {
113
+
if err := consumer.Run(ctx, relayURL); err != nil {
113
114
panic(err)
114
115
}
115
116
+1
-2
internal/api/backlinks.go
+1
-2
internal/api/backlinks.go
···
3
3
import (
4
4
"encoding/base64"
5
5
"encoding/json"
6
-
"log"
7
6
"net/http"
8
7
"strconv"
9
8
···
57
56
58
57
total, records, err := s.Store.ListBacklinks(r.Context(), subject, collection, path, dids, reverse, after, limit)
59
58
if err != nil {
60
-
log.Println("list backlinks:", err)
59
+
s.Logger.Error("list backlinks", "err", err)
61
60
http.Error(w, "internal error", http.StatusInternalServerError)
62
61
return
63
62
}
+1
-2
internal/api/backlinks_count.go
+1
-2
internal/api/backlinks_count.go
···
2
2
3
3
import (
4
4
"encoding/json"
5
-
"log"
6
5
"net/http"
7
6
)
8
7
···
24
23
25
24
total, err := s.Store.CountBacklinks(r.Context(), subject, collection, path)
26
25
if err != nil {
27
-
log.Println("count backlinks:", err)
26
+
s.Logger.Error("count backlinks", "err", err)
28
27
http.Error(w, "internal error", http.StatusInternalServerError)
29
28
return
30
29
}
+1
-2
internal/api/dids.go
+1
-2
internal/api/dids.go
···
3
3
import (
4
4
"encoding/base64"
5
5
"encoding/json"
6
-
"log"
7
6
"net/http"
8
7
"strconv"
9
8
)
···
48
47
49
48
total, dids, err := s.Store.DistinctBacklinkDids(r.Context(), subject, collection, path, after, limit)
50
49
if err != nil {
51
-
log.Println("distinct backlink dids:", err)
50
+
s.Logger.Error("distinct backlink dids", "err", err)
52
51
http.Error(w, "internal error", http.StatusInternalServerError)
53
52
return
54
53
}
+1
-2
internal/api/many_to_many.go
+1
-2
internal/api/many_to_many.go
···
3
3
import (
4
4
"encoding/base64"
5
5
"encoding/json"
6
-
"log"
7
6
"net/http"
8
7
"strconv"
9
8
···
62
61
63
62
total, items, err := s.Store.ManyToMany(r.Context(), subject, collection, path, pathToOther, linkDids, otherSubjects, after, limit)
64
63
if err != nil {
65
-
log.Println("many to many:", err)
64
+
s.Logger.Error("many to many", "err", err)
66
65
http.Error(w, "internal error", http.StatusInternalServerError)
67
66
return
68
67
}
+1
-2
internal/api/many_to_many_counts.go
+1
-2
internal/api/many_to_many_counts.go
···
3
3
import (
4
4
"encoding/base64"
5
5
"encoding/json"
6
-
"log"
7
6
"net/http"
8
7
"strconv"
9
8
···
57
56
58
57
counts, err := s.Store.ManyToManyCounts(r.Context(), subject, collection, path, pathToOther, linkDids, otherSubjects, after, limit)
59
58
if err != nil {
60
-
log.Println("many to many counts:", err)
59
+
s.Logger.Error("many to many counts", "err", err)
61
60
http.Error(w, "internal error", http.StatusInternalServerError)
62
61
return
63
62
}
+1
-2
internal/api/resolve_handle.go
+1
-2
internal/api/resolve_handle.go
···
3
3
import (
4
4
"context"
5
5
"encoding/json"
6
-
"log"
7
6
"net/http"
8
7
"time"
9
8
···
31
30
32
31
identity, err := s.Directory.LookupHandle(resolveCtx, handle)
33
32
if err != nil {
34
-
log.Println("resolve handle:", err)
33
+
s.Logger.Error("resolve handle", "err", err)
35
34
http.Error(w, "internal error", http.StatusInternalServerError)
36
35
return
37
36
}
+1
-2
internal/api/resolve_mini_doc.go
+1
-2
internal/api/resolve_mini_doc.go
···
3
3
import (
4
4
"context"
5
5
"encoding/json"
6
-
"log"
7
6
"net/http"
8
7
"time"
9
8
···
37
36
38
37
identity, err := s.Directory.Lookup(resolveCtx, atID)
39
38
if err != nil {
40
-
log.Println("resolve mini doc:", err)
39
+
s.Logger.Error("resolve mini doc", "err", err)
41
40
http.Error(w, "internal error", http.StatusInternalServerError)
42
41
return
43
42
}
+2
internal/api/server.go
+2
internal/api/server.go
···
1
1
package api
2
2
3
3
import (
4
+
"log/slog"
4
5
"net/http"
5
6
6
7
"github.com/alyraffauf/asterism/internal/store"
···
10
11
type Server struct {
11
12
Store *store.Store
12
13
Directory identity.Directory
14
+
Logger *slog.Logger
13
15
}
14
16
15
17
func (s *Server) Run(addr string) error {
+3
-1
internal/backfill/backfill.go
+3
-1
internal/backfill/backfill.go
···
4
4
"bytes"
5
5
"context"
6
6
"fmt"
7
+
"log/slog"
7
8
"net/http"
8
9
"strings"
9
10
"time"
···
25
26
Client *xrpc.Client
26
27
Directory identity.Directory
27
28
Store *store.Store
29
+
Logger *slog.Logger
28
30
29
31
lastRequest map[string]time.Time
30
32
}
···
60
62
61
63
for did, wantedCollections := range dids {
62
64
if err := b.Repo(ctx, did, wantedCollections); err != nil {
63
-
fmt.Println("could not backfill repo:", did, err)
65
+
b.Logger.Error("could not backfill repo", "did", did, "err", err)
64
66
continue
65
67
}
66
68
}
+1
-2
internal/firehose/account.go
+1
-2
internal/firehose/account.go
···
2
2
3
3
import (
4
4
"context"
5
-
"fmt"
6
5
7
6
"github.com/bluesky-social/indigo/api/atproto"
8
7
)
9
8
10
9
func (c *Consumer) HandleAccount(ctx context.Context, event *atproto.SyncSubscribeRepos_Account) error {
11
10
if err := c.Store.SaveCursor(ctx, event.Seq); err != nil {
12
-
fmt.Println("could not save cursor:", err)
11
+
c.Logger.Error("could not save cursor", "err", err)
13
12
}
14
13
15
14
if event.Status != nil && *event.Status == "deleted" {
+4
-6
internal/firehose/commit.go
+4
-6
internal/firehose/commit.go
···
15
15
)
16
16
17
17
func (c *Consumer) HandleCommit(ctx context.Context, event *atproto.SyncSubscribeRepos_Commit) error {
18
-
// fmt.Println("repo:", event.Repo, "commit:", event.Rev)
19
-
20
18
if err := c.Store.SaveCursor(ctx, event.Seq); err != nil {
21
-
fmt.Println("could not save cursor:", err)
19
+
c.Logger.Error("could not save cursor", "err", err)
22
20
}
23
21
24
22
if event.TooBig {
25
23
go func() {
26
24
if err := c.Backfill.Repo(ctx, event.Repo, c.WantedCollections); err != nil {
27
-
fmt.Println("could not backfill too-big repo:", event.Repo, err)
25
+
c.Logger.Error("could not backfill too-big repo", "repo", event.Repo, "err", err)
28
26
}
29
27
}()
30
28
return nil
···
40
38
}
41
39
42
40
if err := c.verifyCommit(ctx, repo); err != nil {
43
-
fmt.Println("commit verification failed:", event.Repo, err)
41
+
c.Logger.Error("commit verification failed", "repo", event.Repo, "err", err)
44
42
return nil
45
43
}
46
44
47
45
for _, operation := range event.Ops {
48
46
if err := c.handleOperation(ctx, event, repo, operation); err != nil {
49
-
fmt.Println("could not handle operation:", err)
47
+
c.Logger.Error("could not handle operation", "err", err)
50
48
continue
51
49
}
52
50
}
+3
internal/firehose/consumer.go
+3
internal/firehose/consumer.go
···
1
1
package firehose
2
2
3
3
import (
4
+
"log/slog"
5
+
4
6
"github.com/alyraffauf/asterism/internal/backfill"
5
7
"github.com/alyraffauf/asterism/internal/store"
6
8
"github.com/bluesky-social/indigo/atproto/identity"
···
11
13
Store *store.Store
12
14
Directory identity.Directory
13
15
Backfill *backfill.Backfill
16
+
Logger *slog.Logger
14
17
}
15
18
16
19
func (c *Consumer) wants(collection string) bool {
+7
-8
internal/firehose/stream.go
+7
-8
internal/firehose/stream.go
···
3
3
import (
4
4
"context"
5
5
"fmt"
6
-
"log/slog"
7
6
"net/http"
8
7
"time"
9
8
···
18
17
maxBackoff = 30 * time.Second
19
18
)
20
19
21
-
func (c *Consumer) Run(ctx context.Context, relayURL string, logger *slog.Logger) error {
20
+
func (c *Consumer) Run(ctx context.Context, relayURL string) error {
22
21
backoff := minBackoff
23
22
24
23
for {
25
24
dialURL := relayURL
26
25
27
26
if cursor, err := c.Store.GetCursor(ctx); err != nil {
28
-
logger.Warn("could not load cursor, starting from live tip", "err", err)
27
+
c.Logger.Warn("could not load cursor, starting from live tip", "err", err)
29
28
} else if cursor > 0 {
30
29
dialURL = fmt.Sprintf("%s?cursor=%d", relayURL, cursor)
31
30
}
32
31
33
32
conn, _, err := websocket.DefaultDialer.DialContext(ctx, dialURL, http.Header{})
34
33
if err != nil {
35
-
logger.Warn("dial failed", "err", err, "retry in", backoff)
34
+
c.Logger.Warn("dial failed", "err", err, "retry in", backoff)
36
35
} else {
37
36
backoff = minBackoff
38
-
c.stream(ctx, conn, logger)
37
+
c.stream(ctx, conn)
39
38
conn.Close()
40
39
}
41
40
···
50
49
51
50
}
52
51
53
-
func (c *Consumer) stream(ctx context.Context, conn *websocket.Conn, logger *slog.Logger) {
52
+
func (c *Consumer) stream(ctx context.Context, conn *websocket.Conn) {
54
53
callbacks := &events.RepoStreamCallbacks{
55
54
RepoCommit: func(event *atproto.SyncSubscribeRepos_Commit) error {
56
55
return c.HandleCommit(ctx, event)
···
62
61
63
62
scheduler := sequential.NewScheduler("asterism", callbacks.EventHandler)
64
63
65
-
if err := events.HandleRepoStream(ctx, conn, scheduler, logger); err != nil {
66
-
logger.Warn("stream ended", "err", err)
64
+
if err := events.HandleRepoStream(ctx, conn, scheduler, c.Logger); err != nil {
65
+
c.Logger.Warn("stream ended", "err", err)
67
66
}
68
67
}