text search for the atmosphere
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172package main
import ( "context" "errors" "flag" "log/slog" "os" "os/signal" "strings" "syscall" "time"
"github.com/bluesky-social/jetstream"
"tangled.org/manoo.dev/stargazer/internal/classify" "tangled.org/manoo.dev/stargazer/internal/config" "tangled.org/manoo.dev/stargazer/internal/cursor" "tangled.org/manoo.dev/stargazer/internal/meili" "tangled.org/manoo.dev/stargazer/internal/stats")
func main() { configPath := flag.String("config", "config.yaml", "path to config yaml") flag.Parse()
log := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))
cfg, err := config.Load(*configPath) if err != nil { log.Error("config load failed", "err", err) os.Exit(1) }
ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer cancel()
mc := meili.New(cfg.Meilisearch.URL, cfg.Meilisearch.MasterKey, cfg.Meilisearch.Index) if err := mc.Ensure(ctx); err != nil { log.Error("meili ensure failed", "err", err) os.Exit(1) }
classifier := classify.New(cfg.Classify.Blocklist, cfg.Classify.ProseFields, cfg.Classify.MinRuneLen)
cur := cursor.New(cfg.Jetstream.CursorPath) start, err := cur.Load() if err != nil { log.Error("cursor load failed", "err", err) os.Exit(1) }
mode := "live-only" if cfg.Jetstream.APIKey != "" { mode = "archive-replay+live" } log.Info("starting", "host", cfg.Jetstream.Host, "cursor", start, "mode", mode, "blocklist", cfg.Classify.Blocklist)
// No WithCollections: we subscribe to the full firehose and drop blocked // NSIDs client-side. Bandwidth cost is real (bsky dominates volume) but // this is the model the user picked for its simpler mental model and // automatic discovery of new lexicons. opts := []jetstream.Option{ jetstream.WithLogger(log), } if cfg.Jetstream.APIKey != "" { // Archive replay requires an API key. WithAfterSeq is exclusive so // this cleanly resumes; cold start (start == 0) replays the full // sealed archive for the allowlisted collections and cuts over to // live. opts = append(opts, jetstream.WithAPIKey(cfg.Jetstream.APIKey), jetstream.WithAfterSeq(start)) } else { // Live tail is unauthenticated. WithLiveCursor resumes from a saved // position within the server's live buffer; 0 means "from now." opts = append(opts, jetstream.WithLiveCursor(start)) }
client, err := jetstream.Subscribe(cfg.Jetstream.Host, opts...) if err != nil { log.Error("subscribe failed", "err", err) os.Exit(1) } defer client.Close()
st := stats.New() go func() { ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: st.LogSummary(log, 10) } } }()
for batch, err := range client.Events(ctx) { if err != nil { if errors.Is(err, jetstream.ErrFatal) { log.Error("fatal stream error", "err", err) os.Exit(1) } log.Warn("batch error", "err", err) continue }
docs := make([]meili.Doc, 0, len(batch.Events())) var toDelete []string for _, ev := range batch.Events() { if ev.Commit == nil { continue } id := docID(ev.DID, ev.Commit.Collection, ev.Commit.Rkey) switch ev.Commit.Operation { case jetstream.OpDelete: toDelete = append(toDelete, id) case jetstream.OpCreate, jetstream.OpUpdate: st.Seen(ev.Commit.Collection) if classifier.IsBlocked(ev.Commit.Collection) { st.Blocked(ev.Commit.Collection) continue } ex := classifier.Extract(ev.Commit.Record) if ex.Text == "" { st.NoText(ev.Commit.Collection) continue } st.Indexed(ev.Commit.Collection) docs = append(docs, meili.Doc{ ID: id, URI: "at://" + ev.DID + "/" + ev.Commit.Collection + "/" + ev.Commit.Rkey, DID: ev.DID, Collection: ev.Commit.Collection, Rkey: ev.Commit.Rkey, Text: ex.Text, CreatedAt: ex.CreatedAt, Lang: ex.Lang, IndexedAt: time.Now().Unix(), }) } }
if err := mc.AddDocuments(ctx, docs); err != nil { log.Warn("add documents failed; will retry via replay", "err", err, "n", len(docs)) continue } for _, id := range toDelete { _ = mc.Delete(ctx, id) } if err := cur.Save(batch.LastCursor()); err != nil { log.Warn("cursor save failed", "err", err) } if len(docs) > 0 || len(toDelete) > 0 { log.Info("batch", "seq", batch.LastCursor(), "indexed", len(docs), "deleted", len(toDelete)) } }}
func docID(did, collection, rkey string) string { // Meili primary keys must match [A-Za-z0-9_-]. Replace '/' and '.' with // '_' and drop the 'did:' prefix's colon. s := did + "/" + collection + "/" + rkey r := strings.NewReplacer(":", "_", "/", "_", ".", "_") return r.Replace(s)}