Merge in indexing updates

cmd/palomar/Dockerfile copy.opensearch

···
+FROM opensearchproject/opensearch:2.13.0
+RUN /usr/share/opensearch/bin/opensearch-plugin install --batch analysis-icu
+RUN /usr/share/opensearch/bin/opensearch-plugin install --batch analysis-kuromoji

cmd/palomar/Dockerfile.opensearch-dashboards

···
+FROM opensearchproject/opensearch-dashboards:2.13.0
+RUN /usr/share/opensearch-dashboards/bin/opensearch-dashboards-plugin remove securityDashboards

cmd/palomar/docker-compose.yml

···
+version: "3.9"
+services:
+  opensearch:
+    container_name: opensearch
+    build:
+      context: ../../
+      dockerfile: cmd/palomar/Dockerfile.opensearch
+    ports:
+      - "9200:9200"
+      - "9600:9600"
+    environment:
+      - "discovery.type=single-node"
+      - "cluster.name=opensearch-palomar"
+      - "plugins.security.disabled=true"
+      - "bootstrap.memory_lock=true" # Disable JVM heap memory swapping
+      - "OPENSEARCH_JAVA_OPTS=-Xms4096m -Xmx4096m" # Set min and max JVM heap sizes to at least 50% of system RAM
+    ulimits:
+      memlock:
+        soft: -1
+        hard: -1
+      nofile:
+        soft: 65536
+        hard: 65536
+    volumes:
+      - type: bind
+        source: ../../data/opensearch
+        target: /usr/share/opensearch/data
+  indexer:
+    container_name: indexer
+    build:
+      context: ../../
+      dockerfile: cmd/palomar/Dockerfile
+    environment:
+      - "GOLOG_LOG_LEVEL=info"
+      - "ATP_PLC_HOST=https://plc.directory"
+      - "ATP_BGS_HOST=wss://bsky.network"
+      - "ELASTIC_HOSTS=http://opensearch:9200"
+      - "ES_INSECURE_SSL=true"
+      - "ENVIRONMENT=dev"
+      - "ES_POST_INDEX=palomar_post_dev"
+      - "ES_PROFILE_INDEX=palomar_profile_dev"
+      - "PALOMAR_DISCOVER_REPOS=false"
+      - "PALOMAR_BGS_SYNC_RATE_LIMIT=20"
+      - "PALOMAR_INDEX_MAX_CONCURRENCY=5"
+      - "DATABASE_URL=sqlite:///data/palomar/search.db"
+    depends_on:
+      - opensearch
+    volumes:
+      - type: bind
+        source: ../../data
+        target: /data
+  # pagerank:
+  #   container_name: pagerank
+  #   build:
+  #     context: ../../
+  #     dockerfile: cmd/palomar/Dockerfile
+  #   environment:
+  #     - "GOLOG_LOG_LEVEL=info"
+  #     - "ATP_PLC_HOST=https://plc.directory"
+  #     - "ATP_BGS_HOST=wss://bsky.network"
+  #     - "ELASTIC_HOSTS=http://opensearch:9200"
+  #     - "ES_INSECURE_SSL=true"
+  #     - "ENVIRONMENT=dev"
+  #     - "ES_POST_INDEX=palomar_post_dev"
+  #     - "ES_PROFILE_INDEX=palomar_profile_dev"
+  #     - "PALOMAR_DISCOVER_REPOS=false"
+  #     - "PALOMAR_BGS_SYNC_RATE_LIMIT=20"
+  #     - "PALOMAR_INDEX_MAX_CONCURRENCY=5"
+  #     - "DATABASE_URL=sqlite:///data/palomar/pagerank.db"
+  #     - "PAGERANK_FILE=/data/palomar/pageranks.csv"
+  #   depends_on:
+  #     - opensearch
+  #   volumes:
+  #     - type: bind
+  #       source: ../../data
+  #       target: /data
+  api:
+    container_name: api
+    build:
+      context: ../../
+      dockerfile: cmd/palomar/Dockerfile
+    ports:
+      - "3999:3999"
+      - "3998:3998"
+    environment:
+      - "GOLOG_LOG_LEVEL=info"
+      - "ATP_PLC_HOST=https://plc.directory"
+      - "ATP_BGS_HOST=wss://bsky.network"
+      - "ELASTIC_HOSTS=http://opensearch:9200"
+      - "ES_INSECURE_SSL=true"
+      - "ENVIRONMENT=dev"
+      - "ES_POST_INDEX=palomar_post_dev"
+      - "ES_PROFILE_INDEX=palomar_profile_dev"
+      - "DATABASE_URL=sqlite:///data/palomar/search.db"
+      - "PALOMAR_READONLY=true"
+    volumes:
+      - type: bind
+        source: ../../data
+        target: /data
+  opensearch-dashboards:
+    build:
+      context: ../../
+      dockerfile: cmd/palomar/Dockerfile.opensearch-dashboards
+    container_name: opensearch-dashboards
+    ports:
+      - 5601:5601
+    environment:
+      OPENSEARCH_HOSTS: '["http://opensearch:9200"]'
+networks:
+  default:

cmd/palomar/main.go

···
 			EnvVars: []string{"MAX_METADB_CONNECTIONS"},
 			Value:   40,
 		},
+		&cli.StringFlag{
+			Name:    "log-level",
+			Usage:   "log level (debug, info, warn, error)",
+			Value:   "info",
+			EnvVars: []string{"GOLOG_LOG_LEVEL", "LOG_LEVEL"},
+		},
 	}
 
 	app.Commands = []*cli.Command{
···
 			EnvVars: []string{"PALOMAR_INDEX_MAX_CONCURRENCY"},
 		},
 		&cli.IntFlag{
+			Name:    "indexing-rate-limit",
+			Usage:   "max number of documents per second to index",
+			Value:   50_000,
+			EnvVars: []string{"PALOMAR_INDEXING_RATE_LIMIT"},
+		},
+		&cli.IntFlag{
 			Name:    "plc-rate-limit",
 			Usage:   "max number of requests per second to PLC registry",
 			Value:   100,
···
 			EnvVars: []string{"PALOMAR_DISCOVER_REPOS"},
 			Value:   false,
 		},
+		&cli.StringFlag{
+			Name:    "pagerank-file",
+			EnvVars: []string{"PAGERANK_FILE"},
+		},
+		&cli.StringFlag{
+			Name:    "bulk-posts-file",
+			EnvVars: []string{"BULK_POSTS_FILE"},
+		},
+		&cli.StringFlag{
+			Name:    "bulk-profiles-file",
+			EnvVars: []string{"BULK_PROFILES_FILE"},
+		},
 	},
 	Action: func(cctx *cli.Context) error {
+		logLevel := slog.LevelInfo
+		switch cctx.String("log-level") {
+		case "debug":
+			logLevel = slog.LevelDebug
+		case "info":
+			logLevel = slog.LevelInfo
+		case "warn":
+			logLevel = slog.LevelWarn
+		case "error":
+			logLevel = slog.LevelError
+		}
+
 		logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
-			Level: slog.LevelInfo,
+			Level:     logLevel,
+			AddSource: true,
 		}))
 		slog.SetDefault(logger)
+
+		readonly := cctx.Bool("readonly")
 
 		// Enable OTLP HTTP exporter
 		// For relevant environment variables:
···
 			otel.SetTracerProvider(tp)
 		}
 
-		db, err := cliutil.SetupDatabase(cctx.String("database-url"), cctx.Int("max-metadb-connections"))
-		if err != nil {
-			return err
-		}
-
 		escli, err := createEsClient(cctx)
 		if err != nil {
 			return fmt.Errorf("failed to get elasticsearch: %w", err)
 		}
 
-		// TODO: replace this with "bingo" resolver
 		base := identity.BaseDirectory{
 			PLCURL: cctx.String("atp-plc-host"),
 			HTTPClient: http.Client{
···
 		}
 		dir := identity.NewCacheDirectory(&base, 1_500_000, time.Hour*24, time.Minute*2, time.Minute*5)
 
-		srv, err := search.NewServer(
-			db,
-			escli,
-			&dir,
-			search.Config{
+		apiConfig := search.ServerConfig{
+			Logger:       logger,
+			ProfileIndex: cctx.String("es-profile-index"),
+			PostIndex:    cctx.String("es-post-index"),
+		}
+
+		srv, err := search.NewServer(escli, &dir, apiConfig)
+		if err != nil {
+			return err
+		}
+
+		// Configure the indexer if we're not in readonly mode
+		if !readonly {
+			db, err := cliutil.SetupDatabase(cctx.String("database-url"), cctx.Int("max-metadb-connections"))
+			if err != nil {
+				return fmt.Errorf("failed to set up database: %w", err)
+			}
+
+			indexerConfig := search.IndexerConfig{
 				RelayHost:           cctx.String("atp-relay-host"),
 				ProfileIndex:        cctx.String("es-profile-index"),
 				PostIndex:           cctx.String("es-post-index"),
 				Logger:              logger,
 				RelaySyncRateLimit:  cctx.Int("relay-sync-rate-limit"),
 				IndexMaxConcurrency: cctx.Int("index-max-concurrency"),
-			},
-		)
-		if err != nil {
-			return err
+				DiscoverRepos:       cctx.Bool("discover-repos"),
+				IndexingRateLimit:   cctx.Int("indexing-rate-limit"),
+			}
+
+			idx, err := search.NewIndexer(db, escli, &dir, indexerConfig)
+			if err != nil {
+				return fmt.Errorf("failed to set up indexer: %w", err)
+			}
+
+			srv.Indexer = idx
 		}
 
 		go func() {
···
 			srv.RunAPI(cctx.String("bind"))
 		}()
 
-		if cctx.Bool("readonly") {
+		// If we're in readonly mode, just block forever
+		if readonly {
 			select {}
-		} else {
+		} else if cctx.String("pagerank-file") != "" && srv.Indexer != nil {
+			// If we're not in readonly mode, and we have a pagerank file, update pageranks
 			ctx := context.Background()
-			if err := srv.EnsureIndices(ctx); err != nil {
+			if err := srv.Indexer.BulkIndexPageranks(ctx, cctx.String("pagerank-file")); err != nil {
+				return fmt.Errorf("failed to update pageranks: %w", err)
+			}
+		} else if cctx.String("bulk-posts-file") != "" && srv.Indexer != nil {
+			// If we're not in readonly mode, and we have a bulk posts file, index posts
+			ctx := context.Background()
+			if err := srv.Indexer.BulkIndexPosts(ctx, cctx.String("bulk-posts-file")); err != nil {
+				return fmt.Errorf("failed to bulk index posts: %w", err)
+			}
+		} else if cctx.String("bulk-profiles-file") != "" && srv.Indexer != nil {
+			// If we're not in readonly mode, and we have a bulk profiles file, index profiles
+			ctx := context.Background()
+			if err := srv.Indexer.BulkIndexProfiles(ctx, cctx.String("bulk-profiles-file")); err != nil {
+				return fmt.Errorf("failed to bulk index profiles: %w", err)
+			}
+		} else if srv.Indexer != nil {
+			// Otherwise, just run the indexer
+			ctx := context.Background()
+			if err := srv.Indexer.EnsureIndices(ctx); err != nil {
 				return fmt.Errorf("failed to create opensearch indices: %w", err)
 			}
-			if err := srv.RunIndexer(ctx); err != nil {
+			if err := srv.Indexer.RunIndexer(ctx); err != nil {
 				return fmt.Errorf("failed to run indexer: %w", err)
 			}
 		}

cmd/palomar/opensearch_dashboards.yml

···
+---
+server.name: opensearch-dashboards
+server.host: "0.0.0.0"
+opensearch.hosts: http://opensearch:9200

cmd/palomar/pagerank.sh

···
+#!/bin/bash
+set -o errexit
+set -o nounset
+set -o pipefail
+
+export SCYLLA_KEYSPACE="${SCYLLA_KEYSPACE:-}"
+export SCYLLA_HOST="${SCYLLA_HOST:-}"
+
+# Used by pagerank.
+export FOLLOWS_FILE="/data/follows.csv"
+export ACTORS_FILE="/data/actors.csv"
+export OUTPUT_FILE="/data/pageranks.csv"
+export EXPECTED_ACTOR_COUNT="5000000"
+export RUST_LOG="info"
+
+# Used by palomar.
+export PAGERANK_FILE="${OUTPUT_FILE}"
+export PALOMAR_INDEXING_RATE_LIMIT="10000"
+
+function run_pagerank {
+    # Check that the required environment variables are set.
+    if [[ "${SCYLLA_KEYSPACE}" == "" ]]; then
+        echo "SCYLLA_KEYSPACE is not set"
+        exit 1
+    fi
+
+    if [[ "${SCYLLA_HOST}" == "" ]]; then
+        echo "SCYLLA_HOST is not set"
+        exit 1
+    fi
+
+    # Dump the tables to CSV files.
+    rm --force "${FOLLOWS_FILE}"
+    cqlsh \
+        "--keyspace=${SCYLLA_KEYSPACE}" \
+        "--request-timeout=1200" \
+        ---execute "COPY follows (actor_did, subject_did) TO '${FOLLOWS_FILE}' WITH HEADER = FALSE;" \
+        "${SCYLLA_HOST}"
+
+    rm --force "${ACTORS_FILE}"
+    cqlsh \
+        "--keyspace=${SCYLLA_KEYSPACE}" \
+        "--request-timeout=1200" \
+        ---execute "COPY actors (did) TO '${ACTORS_FILE}' WITH HEADER = FALSE;" \
+        "${SCYLLA_HOST}"
+
+    # Run the pagerank file which reads in the table CSV files and outputs a CSV.
+    /usr/local/bin/pagerank
+
+    # Run palomar with the pagerank CSV file.
+    /palomar run
+}
+
+while true; do
+    run_pagerank
+    sleep 24h
+done

go.mod

···
 
 require (
 	contrib.go.opencensus.io/exporter/prometheus v0.4.2
+	github.com/PuerkitoBio/purell v1.2.1
 	github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b
 	github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de
 	github.com/brianvoe/gofakeit/v6 v6.25.0
···
 )
 
 require (
-	github.com/PuerkitoBio/purell v1.2.1 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/go-redis/redis v6.15.9+incompatible // indirect
 	github.com/hashicorp/golang-lru v1.0.2 // indirect

search/bulk.go

···
+package search
+
+import (
+	"bufio"
+	"context"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+	"sync"
+
+	appbsky "github.com/bluesky-social/indigo/api/bsky"
+	"github.com/bluesky-social/indigo/atproto/identity"
+	"github.com/bluesky-social/indigo/atproto/syntax"
+
+	"github.com/ipfs/go-cid"
+)
+
+type pagerankJob struct {
+	did  syntax.DID
+	rank float64
+}
+
+// BulkIndexPageranks updates the pageranks for the DIDs in the Serch Index from a CSV file.
+func (idx *Indexer) BulkIndexPageranks(ctx context.Context, pagerankFile string) error {
+	f, err := os.Open(pagerankFile)
+	if err != nil {
+		return fmt.Errorf("failed to open csv file: %w", err)
+	}
+	defer f.Close()
+
+	// Run 5 pagerank indexers in parallel
+	for i := 0; i < 5; i++ {
+		go idx.runPagerankIndexer(ctx)
+	}
+
+	logger := idx.logger.With("source", "bulk_index_pageranks")
+
+	queue := make(chan string, 20_000)
+	wg := &sync.WaitGroup{}
+	workerCount := 20
+	for i := 0; i < workerCount; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for line := range queue {
+				if err := idx.processPagerankCSVLine(line); err != nil {
+					logger.Error("failed to process line", "err", err)
+				}
+			}
+		}()
+	}
+
+	// Create a scanner to read the file line by line
+	scanner := bufio.NewScanner(f)
+	buf := make([]byte, 0, 64*1024)
+	scanner.Buffer(buf, 1024*1024)
+
+	linesRead := 0
+
+	// Iterate over each line in the file
+	for scanner.Scan() {
+		line := scanner.Text()
+
+		queue <- line
+
+		linesRead++
+		if linesRead%100_000 == 0 {
+			idx.logger.Info("processed csv lines", "lines", linesRead)
+		}
+	}
+
+	close(queue)
+
+	// Check for any scanner errors
+	if err := scanner.Err(); err != nil {
+		return fmt.Errorf("error reading csv file: %w", err)
+	}
+
+	wg.Wait()
+
+	idx.logger.Info("finished processing csv file", "lines", linesRead)
+
+	return nil
+}
+
+// BulkIndexPosts indexes posts from a CSV file.
+func (idx *Indexer) BulkIndexPosts(ctx context.Context, postsFile string) error {
+	f, err := os.Open(postsFile)
+	if err != nil {
+		return fmt.Errorf("failed to open csv file: %w", err)
+	}
+	defer f.Close()
+
+	// Run 5 post indexers in parallel
+	for i := 0; i < 5; i++ {
+		go idx.runPostIndexer(ctx)
+	}
+
+	logger := idx.logger.With("source", "bulk_index_posts")
+
+	queue := make(chan string, 20_000)
+	wg := &sync.WaitGroup{}
+	workerCount := 20
+	for i := 0; i < workerCount; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for line := range queue {
+				if err := idx.processPostCSVLine(line); err != nil {
+					logger.Error("failed to process line", "err", err)
+				}
+			}
+		}()
+	}
+
+	// Create a scanner to read the file line by line
+	scanner := bufio.NewScanner(f)
+	buf := make([]byte, 0, 64*1024)
+	scanner.Buffer(buf, 1024*1024)
+
+	linesRead := 0
+
+	// Iterate over each line in the file
+	for scanner.Scan() {
+		line := scanner.Text()
+
+		queue <- line
+
+		linesRead++
+		if linesRead%100_000 == 0 {
+			idx.logger.Info("processed csv lines", "lines", linesRead)
+		}
+	}
+
+	close(queue)
+
+	// Check for any scanner errors
+	if err := scanner.Err(); err != nil {
+		return fmt.Errorf("error reading csv file: %w", err)
+	}
+
+	wg.Wait()
+
+	idx.logger.Info("finished processing csv file", "lines", linesRead)
+
+	return nil
+}
+
+// BulkIndexProfiles indexes profiles from a CSV file.
+func (idx *Indexer) BulkIndexProfiles(ctx context.Context, profilesFile string) error {
+	f, err := os.Open(profilesFile)
+	if err != nil {
+		return fmt.Errorf("failed to open csv file: %w", err)
+	}
+	defer f.Close()
+
+	for i := 0; i < 5; i++ {
+		go idx.runProfileIndexer(ctx)
+	}
+
+	logger := idx.logger.With("source", "bulk_index_profiles")
+
+	queue := make(chan string, 20_000)
+	wg := &sync.WaitGroup{}
+	workerCount := 20
+	for i := 0; i < workerCount; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for line := range queue {
+				if err := idx.processProfileCSVLine(line); err != nil {
+					logger.Error("failed to process line", "err", err)
+				}
+			}
+		}()
+	}
+
+	// Create a scanner to read the file line by line
+	scanner := bufio.NewScanner(f)
+	buf := make([]byte, 0, 64*1024)
+	scanner.Buffer(buf, 1024*1024)
+
+	linesRead := 0
+
+	// Iterate over each line in the file
+	for scanner.Scan() {
+		line := scanner.Text()
+
+		queue <- line
+
+		linesRead++
+		if linesRead%100_000 == 0 {
+			idx.logger.Info("processed csv lines", "lines", linesRead)
+		}
+	}
+
+	close(queue)
+
+	// Check for any scanner errors
+	if err := scanner.Err(); err != nil {
+		return fmt.Errorf("error reading csv file: %w", err)
+	}
+
+	wg.Wait()
+
+	idx.logger.Info("finished processing csv file", "lines", linesRead)
+
+	return nil
+}
+
+func (idx *Indexer) processPagerankCSVLine(line string) error {
+	// Split the line into DID and rank
+	parts := strings.Split(line, ",")
+	if len(parts) != 2 {
+		return fmt.Errorf("invalid pagerank line: %s", line)
+	}
+
+	did, err := syntax.ParseDID(parts[0])
+	if err != nil {
+		return fmt.Errorf("invalid DID: %s", parts[0])
+	}
+
+	rank, err := strconv.ParseFloat(parts[1], 64)
+	if err != nil {
+		return fmt.Errorf("invalid pagerank value: %s", parts[1])
+	}
+
+	job := PagerankIndexJob{
+		did:  did,
+		rank: rank,
+	}
+
+	// Send the job to the pagerank queue
+	idx.pagerankQueue <- &job
+
+	return nil
+}
+
+func (idx *Indexer) processPostCSVLine(line string) error {
+	// CSV is formatted as
+	// actor_did,rkey,taken_down(time or null),violates_threadgate(False or null),cid,raw(post JSON as hex)
+	parts := strings.Split(line, ",")
+	if len(parts) != 6 {
+		return fmt.Errorf("invalid csv line: %s", line)
+	}
+
+	did, err := syntax.ParseDID(parts[0])
+	if err != nil {
+		return fmt.Errorf("invalid DID: %s", parts[0])
+	}
+
+	rkey, err := syntax.ParseRecordKey(parts[1])
+	if err != nil {
+		return fmt.Errorf("invalid record key: %s", parts[1])
+	}
+
+	isTakenDown := false
+	if parts[2] != "" && parts[2] != "null" {
+		isTakenDown = true
+	}
+
+	violatesThreadgate := false
+	if parts[3] != "" && parts[3] != "False" {
+		violatesThreadgate = true
+	}
+
+	if isTakenDown || violatesThreadgate {
+		return nil
+	}
+
+	cid, err := cid.Parse(parts[4])
+	if err != nil {
+		return fmt.Errorf("invalid CID: %s", parts[4])
+	}
+
+	if len(parts[5]) <= 2 {
+		return nil
+	}
+
+	raw, err := hex.DecodeString(parts[5][2:])
+	if err != nil {
+		return fmt.Errorf("invalid raw record (%s/%s): %s", did, rkey, parts[5][2:])
+	}
+
+	post := appbsky.FeedPost{}
+	if err := json.Unmarshal(raw, &post); err != nil {
+		return fmt.Errorf("failed to unmarshal post: %w", err)
+	}
+
+	job := PostIndexJob{
+		did:    did,
+		rkey:   rkey.String(),
+		rcid:   cid,
+		record: &post,
+	}
+
+	// Send the job to the post queue
+	idx.postQueue <- &job
+
+	return nil
+}
+
+func (idx *Indexer) processProfileCSVLine(line string) error {
+	// CSV is formatted as
+	// actor_did,taken_down(time or null),cid,handle,raw(profile JSON as hex)
+	parts := strings.Split(line, ",")
+	if len(parts) != 5 {
+		return fmt.Errorf("invalid csv line: %s", line)
+	}
+
+	did, err := syntax.ParseDID(parts[0])
+	if err != nil {
+		return fmt.Errorf("invalid DID: %s", parts[0])
+	}
+
+	isTakenDown := false
+	if parts[1] != "" && parts[1] != "null" {
+		isTakenDown = true
+	}
+
+	if isTakenDown {
+		return nil
+	}
+
+	// Skip actors without profile records
+	if parts[2] == "" {
+		return nil
+	}
+
+	cid, err := cid.Parse(parts[2])
+	if err != nil {
+		return fmt.Errorf("invalid CID: %s", parts[2])
+	}
+
+	if len(parts[3]) <= 2 {
+		return nil
+	}
+
+	raw, err := hex.DecodeString(parts[4][2:])
+	if err != nil {
+		return fmt.Errorf("invalid raw record (%s): %s", did, parts[4][2:])
+	}
+
+	profile := appbsky.ActorProfile{}
+	if err := json.Unmarshal(raw, &profile); err != nil {
+		return fmt.Errorf("failed to unmarshal profile: %w", err)
+	}
+
+	ident := identity.Identity{DID: did}
+
+	handle, err := syntax.ParseHandle(parts[3])
+	if err != nil {
+		ident.Handle = syntax.HandleInvalid
+	} else {
+		ident.Handle = handle
+	}
+
+	job := ProfileIndexJob{
+		ident:  &ident,
+		rcid:   cid,
+		record: &profile,
+	}
+
+	// Send the job to the profile queue
+	idx.profileQueue <- &job
+
+	return nil
+}

search/firehose.go

···
 	"github.com/ipfs/go-cid"
 )
 
-func (s *Server) getLastCursor() (int64, error) {
+func (idx *Indexer) getLastCursor() (int64, error) {
 	var lastSeq LastSeq
-	if err := s.db.Find(&lastSeq).Error; err != nil {
+	if err := idx.db.Find(&lastSeq).Error; err != nil {
 		return 0, err
 	}
 
 	if lastSeq.ID == 0 {
-		return 0, s.db.Create(&lastSeq).Error
+		return 0, idx.db.Create(&lastSeq).Error
 	}
 
 	return lastSeq.Seq, nil
 }
 
-func (s *Server) updateLastCursor(curs int64) error {
-	return s.db.Model(LastSeq{}).Where("id = 1").Update("seq", curs).Error
+func (idx *Indexer) updateLastCursor(curs int64) error {
+	return idx.db.Model(LastSeq{}).Where("id = 1").Update("seq", curs).Error
 }
 
-func (s *Server) RunIndexer(ctx context.Context) error {
-	cur, err := s.getLastCursor()
+func (idx *Indexer) RunIndexer(ctx context.Context) error {
+	cur, err := idx.getLastCursor()
 	if err != nil {
 		return fmt.Errorf("get last cursor: %w", err)
 	}
 
-	err = s.bfs.LoadJobs(ctx)
+	// Start the indexer batch workers
+	go idx.runPostIndexer(ctx)
+	go idx.runProfileIndexer(ctx)
+
+	err = idx.bfs.LoadJobs(ctx)
 	if err != nil {
 		return fmt.Errorf("loading backfill jobs: %w", err)
 	}
-	go s.bf.Start()
+	go idx.bf.Start()
 
-	if s.enableRepoDiscovery {
-		go s.discoverRepos()
+	if idx.enableRepoDiscovery {
+		go idx.discoverRepos()
 	}
 
 	d := websocket.DefaultDialer
-	u, err := url.Parse(s.relayHost)
+	u, err := url.Parse(idx.relayhost)
 	if err != nil {
-		return fmt.Errorf("invalid relayHost URI: %w", err)
+		return fmt.Errorf("invalid bgshost URI: %w", err)
 	}
 	u.Path = "xrpc/com.atproto.sync.subscribeRepos"
 	if cur != 0 {
···
 
 			defer func() {
 				if evt.Seq%50 == 0 {
-					if err := s.updateLastCursor(evt.Seq); err != nil {
-						s.logger.Error("failed to persist cursor", "err", err)
+					if err := idx.updateLastCursor(evt.Seq); err != nil {
+						idx.logger.Error("failed to persist cursor", "err", err)
 					}
 				}
 			}()
-			logEvt := s.logger.With("repo", evt.Repo, "rev", evt.Rev, "seq", evt.Seq)
+			logEvt := idx.logger.With("repo", evt.Repo, "rev", evt.Rev, "seq", evt.Seq)
 			if evt.TooBig && evt.Since != nil {
 				// TODO: handle this case (instead of return nil)
 				logEvt.Error("skipping non-genesis tooBig events for now")
···
 			}
 
 			if evt.TooBig {
-				if err := s.processTooBigCommit(ctx, evt); err != nil {
+				if err := idx.processTooBigCommit(ctx, evt); err != nil {
 					// TODO: handle this case (instead of return nil)
 					logEvt.Error("failed to process tooBig event", "err", err)
 					return nil
···
 			}
 
 			// Pass events to the backfiller which will process or buffer as needed
-			if err := s.bf.HandleEvent(ctx, evt); err != nil {
+			if err := idx.bf.HandleEvent(ctx, evt); err != nil {
 				logEvt.Error("failed to handle event", "err", err)
 			}
 
···
 
 			did, err := syntax.ParseDID(evt.Did)
 			if err != nil {
-				s.logger.Error("bad DID in RepoHandle event", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err)
+				idx.logger.Error("bad DID in RepoHandle event", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err)
 				return nil
 			}
-			if err := s.updateUserHandle(ctx, did, evt.Handle); err != nil {
+			if err := idx.updateUserHandle(ctx, did, evt.Handle); err != nil {
 				// TODO: handle this case (instead of return nil)
-				s.logger.Error("failed to update user handle", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err)
+				idx.logger.Error("failed to update user handle", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err)
 			}
 			return nil
 		},
···
 	return events.HandleRepoStream(
 		ctx, con, autoscaling.NewScheduler(
 			autoscaling.DefaultAutoscaleSettings(),
-			s.relayHost,
+			idx.relayhost,
 			rsc.EventHandler,
 		),
 	)
 }
 
-func (s *Server) discoverRepos() {
+func (idx *Indexer) discoverRepos() {
 	ctx := context.Background()
-	log := s.logger.With("func", "discoverRepos")
+	log := idx.logger.With("func", "discoverRepos")
 	log.Info("starting repo discovery")
 
 	cursor := ""
···
 	totalErrored := 0
 
 	for {
-		resp, err := comatproto.SyncListRepos(ctx, s.relayClient, cursor, limit)
+		resp, err := comatproto.SyncListRepos(ctx, idx.relayXRPC, cursor, limit)
 		if err != nil {
 			log.Error("failed to list repos", "err", err)
 			time.Sleep(5 * time.Second)
···
 		log.Info("got repo page", "count", len(resp.Repos), "cursor", resp.Cursor)
 		errored := 0
 		for _, repo := range resp.Repos {
-			_, err := s.bfs.GetOrCreateJob(ctx, repo.Did, backfill.StateEnqueued)
+			_, err := idx.bfs.GetOrCreateJob(ctx, repo.Did, backfill.StateEnqueued)
 			if err != nil {
 				log.Error("failed to get or create job", "did", repo.Did, "err", err)
 				errored++
···
 	log.Info("finished repo discovery", "totalJobs", total, "totalErrored", totalErrored)
 }
 
-func (s *Server) handleCreateOrUpdate(ctx context.Context, rawDID string, rev string, path string, recB *[]byte, rcid *cid.Cid) error {
+func (idx *Indexer) handleCreateOrUpdate(ctx context.Context, rawDID string, rev string, path string, recB *[]byte, rcid *cid.Cid) error {
+	logger := idx.logger.With("func", "handleCreateOrUpdate", "did", rawDID, "rev", rev, "path", path)
 	// Since this gets called in a backfill job, we need to check if the path is a post or profile
 	if !strings.Contains(path, "app.bsky.feed.post") && !strings.Contains(path, "app.bsky.actor.profile") {
 		return nil
···
 		return fmt.Errorf("bad DID syntax in event: %w", err)
 	}
 
-	ident, err := s.dir.LookupDID(ctx, did)
-	if err != nil {
-		return fmt.Errorf("resolving identity: %w", err)
-	}
-	if ident == nil {
-		return fmt.Errorf("identity not found for did: %s", did.String())
-	}
-
 	// CBOR Unmarshal the record
 	recCBOR, err := lexutil.CborDecodeValue(*recB)
 	if err != nil {
···
 		return fmt.Errorf("failed to cast record to CBORMarshaler")
 	}
 
+	parts := strings.SplitN(path, "/", 3)
+	if len(parts) < 2 {
+		logger.Warn("skipping post record with malformed path")
+		return nil
+	}
+
 	switch rec := rec.(type) {
 	case *bsky.FeedPost:
-		if err := s.indexPost(ctx, ident, rec, path, *rcid); err != nil {
-			postsFailed.Inc()
-			return fmt.Errorf("indexing post for %s: %w", did.String(), err)
+		rkey, err := syntax.ParseTID(parts[1])
+		if err != nil {
+			logger.Warn("skipping post record with non-TID rkey")
+			return nil
+		}
+
+		job := PostIndexJob{
+			did:    did,
+			record: rec,
+			rcid:   *rcid,
+			rkey:   rkey.String(),
 		}
+
+		// Send the job to the bulk indexer
+		idx.postQueue <- &job
 		postsIndexed.Inc()
 	case *bsky.ActorProfile:
-		if err := s.indexProfile(ctx, ident, rec, path, *rcid); err != nil {
-			profilesFailed.Inc()
-			return fmt.Errorf("indexing profile for %s: %w", did.String(), err)
+		if parts[1] != "self" {
+			return nil
+		}
+
+		ident, err := idx.dir.LookupDID(ctx, did)
+		if err != nil {
+			return fmt.Errorf("resolving identity: %w", err)
 		}
+		if ident == nil {
+			return fmt.Errorf("identity not found for did: %s", did.String())
+		}
+
+		job := ProfileIndexJob{
+			ident:  ident,
+			record: rec,
+			rcid:   *rcid,
+		}
+
+		// Send the job to the bulk indexer
+		idx.profileQueue <- &job
 		profilesIndexed.Inc()
 	default:
 	}
 	return nil
 }
 
-func (s *Server) handleDelete(ctx context.Context, rawDID, rev, path string) error {
+func (idx *Indexer) handleDelete(ctx context.Context, rawDID, rev, path string) error {
 	// Since this gets called in a backfill job, we need to check if the path is a post or profile
 	if !strings.Contains(path, "app.bsky.feed.post") && !strings.Contains(path, "app.bsky.actor.profile") {
 		return nil
···
 		return fmt.Errorf("invalid DID in event: %w", err)
 	}
 
-	ident, err := s.dir.LookupDID(ctx, did)
-	if err != nil {
-		return err
-	}
-	if ident == nil {
-		return fmt.Errorf("identity not found for did: %s", did.String())
-	}
-
 	switch {
 	// TODO: handle profile deletes, its an edge case, but worth doing still
 	case strings.Contains(path, "app.bsky.feed.post"):
-		if err := s.deletePost(ctx, ident, path); err != nil {
+		if err := idx.deletePost(ctx, did, path); err != nil {
 			return err
 		}
 		postsDeleted.Inc()
···
 	return nil
 }
 
-func (s *Server) processTooBigCommit(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) error {
-	repodata, err := comatproto.SyncGetRepo(ctx, s.relayClient, evt.Repo, "")
+func (idx *Indexer) processTooBigCommit(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) error {
+	logger := idx.logger.With("func", "processTooBigCommit", "repo", evt.Repo, "rev", evt.Rev, "seq", evt.Seq)
+
+	repodata, err := comatproto.SyncGetRepo(ctx, idx.relayXRPC, evt.Repo, "")
 	if err != nil {
 		return err
 	}
···
 		return fmt.Errorf("bad DID in repo event: %w", err)
 	}
 
-	ident, err := s.dir.LookupDID(ctx, did)
+	ident, err := idx.dir.LookupDID(ctx, did)
 	if err != nil {
 		return err
 	}
···
 			rcid, rec, err := r.GetRecord(ctx, k)
 			if err != nil {
 				// TODO: handle this case (instead of return nil)
-				s.logger.Error("failed to get record from repo checkout", "path", k, "err", err)
+				idx.logger.Error("failed to get record from repo checkout", "path", k, "err", err)
+				return nil
+			}
+
+			parts := strings.SplitN(k, "/", 3)
+			if len(parts) < 2 {
+				logger.Warn("skipping post record with malformed path")
 				return nil
 			}
 
 			switch rec := rec.(type) {
 			case *bsky.FeedPost:
-				if err := s.indexPost(ctx, ident, rec, k, rcid); err != nil {
-					return fmt.Errorf("indexing post: %w", err)
+				rkey, err := syntax.ParseTID(parts[1])
+				if err != nil {
+					logger.Warn("skipping post record with non-TID rkey")
+					return nil
 				}
+
+				job := PostIndexJob{
+					did:    did,
+					record: rec,
+					rcid:   rcid,
+					rkey:   rkey.String(),
+				}
+
+				// Send the job to the bulk indexer
+				idx.postQueue <- &job
 			case *bsky.ActorProfile:
-				if err := s.indexProfile(ctx, ident, rec, k, rcid); err != nil {
-					return fmt.Errorf("indexing profile: %w", err)
+				if parts[1] != "self" {
+					return nil
+				}
+
+				job := ProfileIndexJob{
+					ident:  ident,
+					record: rec,
+					rcid:   rcid,
 				}
+
+				// Send the job to the bulk indexer
+				idx.profileQueue <- &job
 			default:
 			}
 

search/handlers.go

···
 	return e.JSON(200, out)
 }
 
-type IndexError struct {
-	DID string `json:"did"`
-	Err string `json:"err"`
-}
-
-func (s *Server) handleIndexRepos(e echo.Context) error {
-	ctx, span := tracer.Start(e.Request().Context(), "handleIndexRepos")
-	defer span.End()
-
-	dids, ok := e.QueryParams()["did"]
-	if !ok {
-		return e.JSON(400, map[string]any{
-			"error": "must pass at least one did to index",
-		})
-	}
-
-	for _, did := range dids {
-		_, err := syntax.ParseDID(did)
-		if err != nil {
-			return e.JSON(400, map[string]any{
-				"error": fmt.Sprintf("invalid DID (%s): %s", did, err),
-			})
-		}
-	}
-
-	errs := []IndexError{}
-	successes := 0
-	skipped := 0
-	for _, did := range dids {
-		job, err := s.bfs.GetJob(ctx, did)
-		if job == nil && err == nil {
-			err := s.bfs.EnqueueJob(ctx, did)
-			if err != nil {
-				errs = append(errs, IndexError{
-					DID: did,
-					Err: err.Error(),
-				})
-				continue
-			}
-			successes++
-			continue
-		}
-		skipped++
-	}
-
-	return e.JSON(200, map[string]any{
-		"numEnqueued": successes,
-		"numSkipped":  skipped,
-		"numErrored":  len(errs),
-		"errors":      errs,
-	})
-}
-
 func (s *Server) SearchPosts(ctx context.Context, params *PostSearchParams) (*appbsky.UnspeccedSearchPostsSkeleton_Output, error) {
 	ctx, span := tracer.Start(ctx, "SearchPosts")
 	defer span.End()

search/indexing.go

···
 import (
 	"bytes"
 	"context"
+	_ "embed"
 	"encoding/json"
 	"fmt"
 	"io"
+	"log/slog"
+	"os"
 	"strings"
+	"time"
 
 	appbsky "github.com/bluesky-social/indigo/api/bsky"
 	"github.com/bluesky-social/indigo/atproto/identity"
 	"github.com/bluesky-social/indigo/atproto/syntax"
+	"github.com/bluesky-social/indigo/backfill"
+	"github.com/bluesky-social/indigo/xrpc"
 	"github.com/ipfs/go-cid"
+	"github.com/labstack/echo/v4"
 	"go.opentelemetry.io/otel/attribute"
+	"golang.org/x/time/rate"
+	gorm "gorm.io/gorm"
 
+	es "github.com/opensearch-project/opensearch-go/v2"
 	esapi "github.com/opensearch-project/opensearch-go/v2/opensearchapi"
 )
 
-func (s *Server) deletePost(ctx context.Context, ident *identity.Identity, recordPath string) error {
+type Indexer struct {
+	escli        *es.Client
+	postIndex    string
+	profileIndex string
+	db           *gorm.DB
+	relayhost    string
+	relayXRPC    *xrpc.Client
+	dir          identity.Directory
+	echo         *echo.Echo
+	logger       *slog.Logger
+
+	bfs *backfill.Gormstore
+	bf  *backfill.Backfiller
+
+	enableRepoDiscovery bool
+
+	indexLimiter  *rate.Limiter
+	profileQueue  chan *ProfileIndexJob
+	postQueue     chan *PostIndexJob
+	pagerankQueue chan *PagerankIndexJob
+}
+
+type IndexerConfig struct {
+	RelayHost           string
+	ProfileIndex        string
+	PostIndex           string
+	Logger              *slog.Logger
+	RelaySyncRateLimit  int
+	IndexMaxConcurrency int
+	DiscoverRepos       bool
+	IndexingRateLimit   int
+}
+
+type ProfileIndexJob struct {
+	ident  *identity.Identity
+	record *appbsky.ActorProfile
+	rcid   cid.Cid
+}
+
+type PostIndexJob struct {
+	did    syntax.DID
+	record *appbsky.FeedPost
+	rcid   cid.Cid
+	rkey   string
+}
+
+type PagerankIndexJob struct {
+	did  syntax.DID
+	rank float64
+}
+
+func NewIndexer(db *gorm.DB, escli *es.Client, dir identity.Directory, config IndexerConfig) (*Indexer, error) {
+	logger := config.Logger
+	if logger == nil {
+		logger = slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
+			Level: slog.LevelInfo,
+		}))
+	}
+	logger = logger.With("component", "indexer")
+
+	logger.Info("running database migrations")
+	db.AutoMigrate(&LastSeq{})
+	db.AutoMigrate(&backfill.GormDBJob{})
+
+	relayWS := config.RelayHost
+	if !strings.HasPrefix(relayWS, "ws") {
+		return nil, fmt.Errorf("specified bgs host must include 'ws://' or 'wss://'")
+	}
+
+	relayHTTP := strings.Replace(relayWS, "ws", "http", 1)
+	relayXRPC := &xrpc.Client{
+		Host: relayHTTP,
+	}
+
+	limiter := rate.NewLimiter(rate.Limit(config.IndexingRateLimit), 10_000)
+
+	idx := &Indexer{
+		escli:               escli,
+		profileIndex:        config.ProfileIndex,
+		postIndex:           config.PostIndex,
+		db:                  db,
+		relayhost:           config.RelayHost,
+		relayXRPC:           relayXRPC,
+		dir:                 dir,
+		logger:              logger,
+		enableRepoDiscovery: config.DiscoverRepos,
+
+		indexLimiter:  limiter,
+		profileQueue:  make(chan *ProfileIndexJob, 1000),
+		postQueue:     make(chan *PostIndexJob, 1000),
+		pagerankQueue: make(chan *PagerankIndexJob, 1000),
+	}
+
+	bfstore := backfill.NewGormstore(db)
+	opts := backfill.DefaultBackfillOptions()
+
+	if config.RelaySyncRateLimit > 0 {
+		opts.SyncRequestsPerSecond = config.RelaySyncRateLimit
+		opts.ParallelBackfills = 2 * config.RelaySyncRateLimit
+	} else {
+		opts.SyncRequestsPerSecond = 8
+	}
+
+	opts.CheckoutPath = fmt.Sprintf("%s/xrpc/com.atproto.sync.getRepo", relayHTTP)
+	if config.IndexMaxConcurrency > 0 {
+		opts.ParallelRecordCreates = config.IndexMaxConcurrency
+	} else {
+		opts.ParallelRecordCreates = 20
+	}
+	opts.NSIDFilter = "app.bsky."
+	bf := backfill.NewBackfiller(
+		"search",
+		bfstore,
+		idx.handleCreateOrUpdate,
+		idx.handleCreateOrUpdate,
+		idx.handleDelete,
+		opts,
+	)
+
+	idx.bfs = bfstore
+	idx.bf = bf
+
+	return idx, nil
+}
+
+//go:embed post_schema.json
+var palomarPostSchemaJSON string
+
+//go:embed profile_schema.json
+var palomarProfileSchemaJSON string
+
+func (idx *Indexer) EnsureIndices(ctx context.Context) error {
+	indices := []struct {
+		Name       string
+		SchemaJSON string
+	}{
+		{Name: idx.postIndex, SchemaJSON: palomarPostSchemaJSON},
+		{Name: idx.profileIndex, SchemaJSON: palomarProfileSchemaJSON},
+	}
+	for _, index := range indices {
+		resp, err := idx.escli.Indices.Exists([]string{index.Name})
+		if err != nil {
+			return err
+		}
+		defer resp.Body.Close()
+		io.ReadAll(resp.Body)
+		if resp.IsError() && resp.StatusCode != 404 {
+			return fmt.Errorf("failed to check index existence")
+		}
+		if resp.StatusCode == 404 {
+			idx.logger.Warn("creating opensearch index", "index", index.Name)
+			if len(index.SchemaJSON) < 2 {
+				return fmt.Errorf("empty schema file (go:embed failed)")
+			}
+			buf := strings.NewReader(index.SchemaJSON)
+			resp, err := idx.escli.Indices.Create(
+				index.Name,
+				idx.escli.Indices.Create.WithBody(buf))
+			if err != nil {
+				return err
+			}
+			defer resp.Body.Close()
+			io.ReadAll(resp.Body)
+			if resp.IsError() {
+				return fmt.Errorf("failed to create index")
+			}
+		}
+	}
+	return nil
+}
+
+func (idx *Indexer) runPostIndexer(ctx context.Context) {
+	ctx, span := tracer.Start(ctx, "runPostIndexer")
+	defer span.End()
+
+	// Batch up to 1000 posts at a time, or every 5 seconds
+	tick := time.NewTicker(5 * time.Second)
+	defer tick.Stop()
+
+	var posts []*PostIndexJob
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-tick.C:
+			if len(posts) > 0 {
+				err := idx.indexLimiter.WaitN(ctx, len(posts))
+				if err != nil {
+					idx.logger.Error("failed to wait for rate limiter", "err", err)
+					continue
+				}
+				err = idx.indexPosts(ctx, posts)
+				if err != nil {
+					idx.logger.Error("failed to index posts", "err", err)
+				}
+				posts = posts[:0]
+			}
+		case job := <-idx.postQueue:
+			posts = append(posts, job)
+			if len(posts) >= 1000 {
+				err := idx.indexLimiter.WaitN(ctx, len(posts))
+				if err != nil {
+					idx.logger.Error("failed to wait for rate limiter", "err", err)
+					continue
+				}
+				err = idx.indexPosts(ctx, posts)
+				if err != nil {
+					idx.logger.Error("failed to index posts", "err", err)
+				}
+				posts = posts[:0]
+			}
+		}
+	}
+}
+
+func (idx *Indexer) runProfileIndexer(ctx context.Context) {
+	ctx, span := tracer.Start(ctx, "runProfileIndexer")
+	defer span.End()
+
+	// Batch up to 1000 profiles at a time, or every 5 seconds
+	tick := time.NewTicker(5 * time.Second)
+	defer tick.Stop()
+
+	var profiles []*ProfileIndexJob
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-tick.C:
+			if len(profiles) > 0 {
+				err := idx.indexLimiter.WaitN(ctx, len(profiles))
+				if err != nil {
+					idx.logger.Error("failed to wait for rate limiter", "err", err)
+					continue
+				}
+				err = idx.indexProfiles(ctx, profiles)
+				if err != nil {
+					idx.logger.Error("failed to index profiles", "err", err)
+				}
+				profiles = profiles[:0]
+			}
+		case job := <-idx.profileQueue:
+			profiles = append(profiles, job)
+			if len(profiles) >= 1000 {
+				err := idx.indexLimiter.WaitN(ctx, len(profiles))
+				if err != nil {
+					idx.logger.Error("failed to wait for rate limiter", "err", err)
+					continue
+				}
+				err = idx.indexProfiles(ctx, profiles)
+				if err != nil {
+					idx.logger.Error("failed to index profiles", "err", err)
+				}
+				profiles = profiles[:0]
+			}
+		}
+	}
+}
+
+func (idx *Indexer) runPagerankIndexer(ctx context.Context) {
+	ctx, span := tracer.Start(ctx, "runPagerankIndexer")
+	defer span.End()
+
+	// Batch up to 1000 pageranks at a time, or every 5 seconds
+	tick := time.NewTicker(5 * time.Second)
+	defer tick.Stop()
+
+	var pageranks []*PagerankIndexJob
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-tick.C:
+			if len(pageranks) > 0 {
+				err := idx.indexLimiter.WaitN(ctx, len(pageranks))
+				if err != nil {
+					idx.logger.Error("failed to wait for rate limiter", "err", err)
+					continue
+				}
+				err = idx.indexPageranks(ctx, pageranks)
+				if err != nil {
+					idx.logger.Error("failed to index pageranks", "err", err)
+				}
+				pageranks = pageranks[:0]
+			}
+		case job := <-idx.pagerankQueue:
+			pageranks = append(pageranks, job)
+			if len(pageranks) >= 1000 {
+				err := idx.indexLimiter.WaitN(ctx, len(pageranks))
+				if err != nil {
+					idx.logger.Error("failed to wait for rate limiter", "err", err)
+					continue
+				}
+				err = idx.indexPageranks(ctx, pageranks)
+				if err != nil {
+					idx.logger.Error("failed to index pageranks", "err", err)
+				}
+				pageranks = pageranks[:0]
+			}
+		}
+	}
+}
+
+func (idx *Indexer) deletePost(ctx context.Context, did syntax.DID, recordPath string) error {
 	ctx, span := tracer.Start(ctx, "deletePost")
 	defer span.End()
-	span.SetAttributes(attribute.String("repo", ident.DID.String()), attribute.String("path", recordPath))
+	span.SetAttributes(attribute.String("repo", did.String()), attribute.String("path", recordPath))
 
-	logger := s.logger.With("repo", ident.DID, "path", recordPath, "op", "deletePost")
+	logger := idx.logger.With("repo", did, "path", recordPath, "op", "deletePost")
 
 	parts := strings.SplitN(recordPath, "/", 3)
 	if len(parts) < 2 {
···
 		return nil
 	}
 
-	docID := fmt.Sprintf("%s_%s", ident.DID.String(), rkey)
+	docID := fmt.Sprintf("%s_%s", did.String(), rkey)
 	logger.Info("deleting post from index", "docID", docID)
 	req := esapi.DeleteRequest{
-		Index:      s.postIndex,
+		Index:      idx.postIndex,
 		DocumentID: docID,
 		Refresh:    "true",
 	}
 
-	res, err := req.Do(ctx, s.escli)
+	err = idx.indexLimiter.Wait(ctx)
+	if err != nil {
+		logger.Warn("failed to wait for rate limiter", "err", err)
+		return err
+	}
+	res, err := req.Do(ctx, idx.escli)
 	if err != nil {
 		return fmt.Errorf("failed to delete post: %w", err)
 	}
···
 	return nil
 }
 
-func (s *Server) indexPost(ctx context.Context, ident *identity.Identity, rec *appbsky.FeedPost, path string, rcid cid.Cid) error {
-	ctx, span := tracer.Start(ctx, "indexPost")
+func (idx *Indexer) indexPosts(ctx context.Context, jobs []*PostIndexJob) error {
+	ctx, span := tracer.Start(ctx, "indexPosts")
 	defer span.End()
-	span.SetAttributes(attribute.String("repo", ident.DID.String()), attribute.String("path", path))
+	span.SetAttributes(attribute.Int("num_posts", len(jobs)))
 
-	log := s.logger.With("repo", ident.DID, "path", path, "op", "indexPost")
-	parts := strings.SplitN(path, "/", 3)
-	if len(parts) < 2 {
-		log.Warn("skipping post record with malformed path")
-		return nil
-	}
-	rkey, err := syntax.ParseTID(parts[1])
-	if err != nil {
-		log.Warn("skipping post record with non-TID rkey")
-		return nil
-	}
+	log := idx.logger.With("op", "indexPosts")
+	start := time.Now()
 
-	log = log.With("rkey", rkey)
+	var buf bytes.Buffer
+	for i := range jobs {
+		job := jobs[i]
+		doc := TransformPost(job.record, job.did, job.rkey, job.rcid.String())
+		docBytes, err := json.Marshal(doc)
+		if err != nil {
+			log.Warn("failed to marshal post", "err", err)
+			return err
+		}
 
-	doc := TransformPost(rec, ident, rkey.String(), rcid.String())
-	b, err := json.Marshal(doc)
-	if err != nil {
-		return err
-	}
+		indexScript := []byte(fmt.Sprintf(`{"index":{"_id":"%s"}}%s`, doc.DocId(), "\n"))
+		docBytes = append(docBytes, "\n"...)
 
-	log.Debug("indexing post")
-	req := esapi.IndexRequest{
-		Index:      s.postIndex,
-		DocumentID: doc.DocId(),
-		Body:       bytes.NewReader(b),
+		buf.Grow(len(indexScript) + len(docBytes))
+		buf.Write(indexScript)
+		buf.Write(docBytes)
 	}
 
-	res, err := req.Do(ctx, s.escli)
+	log.Info("indexing posts", "num_posts", len(jobs))
+
+	res, err := idx.escli.Bulk(bytes.NewReader(buf.Bytes()), idx.escli.Bulk.WithIndex(idx.postIndex))
 	if err != nil {
-		log.Warn("failed to send indexing request", "err", err)
-		return fmt.Errorf("failed to send indexing request: %w", err)
+		log.Warn("failed to send bulk indexing request", "err", err)
+		return fmt.Errorf("failed to send bulk indexing request: %w", err)
 	}
 	defer res.Body.Close()
-	body, err := io.ReadAll(res.Body)
-	if err != nil {
-		log.Warn("failed to read indexing response", "err", err)
-		return fmt.Errorf("failed to read indexing response: %w", err)
-	}
+
 	if res.IsError() {
-		log.Warn("opensearch indexing error", "status_code", res.StatusCode, "response", res, "body", string(body))
-		return fmt.Errorf("indexing error, code=%d", res.StatusCode)
+		body, err := io.ReadAll(res.Body)
+		if err != nil {
+			log.Warn("failed to read bulk indexing response", "err", err)
+			return fmt.Errorf("failed to read bulk indexing response: %w", err)
+		}
+		log.Warn("opensearch bulk indexing error", "status_code", res.StatusCode, "response", res, "body", string(body))
+		return fmt.Errorf("bulk indexing error, code=%d", res.StatusCode)
 	}
+
+	log.Info("indexed posts", "num_posts", len(jobs), "duration", time.Since(start))
+
 	return nil
 }
 
-func (s *Server) indexProfile(ctx context.Context, ident *identity.Identity, rec *appbsky.ActorProfile, path string, rcid cid.Cid) error {
-	ctx, span := tracer.Start(ctx, "indexProfile")
+func (idx *Indexer) indexProfiles(ctx context.Context, jobs []*ProfileIndexJob) error {
+	ctx, span := tracer.Start(ctx, "indexProfiles")
 	defer span.End()
-	span.SetAttributes(attribute.String("repo", ident.DID.String()), attribute.String("path", path))
+	span.SetAttributes(attribute.Int("num_profiles", len(jobs)))
+
+	log := idx.logger.With("op", "indexProfiles")
+	start := time.Now()
 
-	log := s.logger.With("repo", ident.DID, "path", path, "op", "indexProfile")
-	parts := strings.SplitN(path, "/", 3)
-	if len(parts) != 2 || parts[1] != "self" {
-		log.Warn("skipping indexing non-canonical profile record", "did", ident.DID, "path", path)
-		return nil
+	var buf bytes.Buffer
+	for i := range jobs {
+		job := jobs[i]
+
+		doc := TransformProfile(job.record, job.ident, job.rcid.String())
+		docBytes, err := json.Marshal(doc)
+		if err != nil {
+			log.Warn("failed to marshal profile", "err", err)
+			return err
+		}
+
+		indexScript := []byte(fmt.Sprintf(`{"index":{"_id":"%s"}}%s`, job.ident.DID.String(), "\n"))
+		docBytes = append(docBytes, "\n"...)
+
+		buf.Grow(len(indexScript) + len(docBytes))
+		buf.Write(indexScript)
+		buf.Write(docBytes)
 	}
 
-	log.Info("indexing profile", "handle", ident.Handle)
+	log.Info("indexing profiles", "num_profiles", len(jobs))
 
-	doc := TransformProfile(rec, ident, rcid.String())
-	b, err := json.Marshal(doc)
+	res, err := idx.escli.Bulk(bytes.NewReader(buf.Bytes()), idx.escli.Bulk.WithIndex(idx.profileIndex))
 	if err != nil {
-		return err
+		log.Warn("failed to send bulk indexing request", "err", err)
+		return fmt.Errorf("failed to send bulk indexing request: %w", err)
+	}
+	defer res.Body.Close()
+
+	if res.IsError() {
+		body, err := io.ReadAll(res.Body)
+		if err != nil {
+			log.Warn("failed to read bulk indexing response", "err", err)
+			return fmt.Errorf("failed to read bulk indexing response: %w", err)
+		}
+		log.Warn("opensearch bulk indexing error", "status_code", res.StatusCode, "response", res, "body", string(body))
+		return fmt.Errorf("bulk indexing error, code=%d", res.StatusCode)
 	}
-	req := esapi.IndexRequest{
-		Index:      s.profileIndex,
-		DocumentID: ident.DID.String(),
-		Body:       bytes.NewReader(b),
+
+	log.Info("indexed profiles", "num_profiles", len(jobs), "duration", time.Since(start))
+
+	return nil
+}
+
+// updateProfilePagranks uses the OpenSearch bulk API to update the pageranks for the given DIDs
+func (idx *Indexer) indexPageranks(ctx context.Context, pageranks []*PagerankIndexJob) error {
+	ctx, span := tracer.Start(ctx, "indexPageranks")
+	defer span.End()
+	span.SetAttributes(attribute.Int("num_profiles", len(pageranks)))
+
+	log := idx.logger.With("op", "indexPageranks")
+
+	log.Info("updating profile pageranks")
+
+	var buf bytes.Buffer
+	for _, pr := range pageranks {
+		updateScript := map[string]any{
+			"script": map[string]any{
+				"source": "ctx._source.pagerank = params.pagerank",
+				"lang":   "painless",
+				"params": map[string]any{
+					"pagerank": pr.rank,
+				},
+			},
+		}
+		updateScriptJSON, err := json.Marshal(updateScript)
+		if err != nil {
+			log.Warn("failed to marshal update script", "err", err)
+			return err
+		}
+
+		updateMetaJSON := []byte(fmt.Sprintf(`{"update":{"_id":"%s"}}%s`, pr.did.String(), "\n"))
+		updateScriptJSON = append(updateScriptJSON, "\n"...)
+
+		buf.Grow(len(updateMetaJSON) + len(updateScriptJSON))
+		buf.Write(updateMetaJSON)
+		buf.Write(updateScriptJSON)
 	}
 
-	res, err := req.Do(ctx, s.escli)
+	res, err := idx.escli.Bulk(bytes.NewReader(buf.Bytes()), idx.escli.Bulk.WithIndex(idx.profileIndex))
 	if err != nil {
-		log.Warn("failed to send indexing request", "err", err)
-		return fmt.Errorf("failed to send indexing request: %w", err)
+		log.Warn("failed to send bulk indexing request", "err", err)
+		return fmt.Errorf("failed to send bulk indexing request: %w", err)
 	}
 	defer res.Body.Close()
-	body, err := io.ReadAll(res.Body)
-	if err != nil {
-		log.Warn("failed to read indexing response", "err", err)
-		return fmt.Errorf("failed to read indexing response: %w", err)
-	}
+
 	if res.IsError() {
-		log.Warn("opensearch indexing error", "status_code", res.StatusCode, "response", res, "body", string(body))
-		return fmt.Errorf("indexing error, code=%d", res.StatusCode)
+		body, err := io.ReadAll(res.Body)
+		if err != nil {
+			log.Warn("failed to read bulk indexing response", "err", err)
+			return fmt.Errorf("failed to read bulk indexing response: %w", err)
+		}
+		log.Warn("opensearch bulk indexing error", "status_code", res.StatusCode, "response", res, "body", string(body))
+		return fmt.Errorf("bulk indexing error, code=%d", res.StatusCode)
 	}
+
 	return nil
 }
 
-func (s *Server) updateUserHandle(ctx context.Context, did syntax.DID, handle string) error {
+func (idx *Indexer) updateUserHandle(ctx context.Context, did syntax.DID, handle string) error {
 	ctx, span := tracer.Start(ctx, "updateUserHandle")
 	defer span.End()
 	span.SetAttributes(attribute.String("repo", did.String()), attribute.String("event.handle", handle))
 
-	log := s.logger.With("repo", did.String(), "op", "updateUserHandle", "handle_from_event", handle)
+	log := idx.logger.With("repo", did.String(), "op", "updateUserHandle", "handle_from_event", handle)
 
-	err := s.dir.Purge(ctx, did.AtIdentifier())
+	err := idx.dir.Purge(ctx, did.AtIdentifier())
 	if err != nil {
 		log.Warn("failed to purge DID from directory", "err", err)
 		return err
 	}
 
-	ident, err := s.dir.LookupDID(ctx, did)
+	ident, err := idx.dir.LookupDID(ctx, did)
 	if err != nil {
 		log.Warn("failed to lookup DID in directory", "err", err)
 		return err
···
 	}
 
 	req := esapi.UpdateRequest{
-		Index:      s.profileIndex,
+		Index:      idx.profileIndex,
 		DocumentID: did.String(),
 		Body:       bytes.NewReader(b),
 	}
 
-	res, err := req.Do(ctx, s.escli)
+	err = idx.indexLimiter.Wait(ctx)
+	if err != nil {
+		log.Warn("failed to wait for rate limiter", "err", err)
+		return err
+	}
+	res, err := req.Do(ctx, idx.escli)
 	if err != nil {
 		log.Warn("failed to send indexing request", "err", err)
 		return fmt.Errorf("failed to send indexing request: %w", err)

search/server.go

···
 
 import (
 	"context"
-	_ "embed"
 	"fmt"
 	"io"
 	"log/slog"
···
 	"strings"
 
 	"github.com/bluesky-social/indigo/atproto/identity"
-	"github.com/bluesky-social/indigo/backfill"
-	"github.com/bluesky-social/indigo/xrpc"
 
 	"github.com/carlmjohnson/versioninfo"
 	"github.com/labstack/echo/v4"
···
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	slogecho "github.com/samber/slog-echo"
 	"go.opentelemetry.io/contrib/instrumentation/github.com/labstack/echo/otelecho"
-	gorm "gorm.io/gorm"
+
+	_ "net/http/pprof" // For pprof in the metrics server
 )
 
+type LastSeq struct {
+	ID  uint `gorm:"primarykey"`
+	Seq int64
+}
+
+type ServerConfig struct {
+	Logger            *slog.Logger
+	ProfileIndex      string
+	PostIndex         string
+	AtlantisAddresses []string
+}
+
 type Server struct {
 	escli        *es.Client
 	postIndex    string
 	profileIndex string
-	db           *gorm.DB
-	relayHost    string
-	relayClient  *xrpc.Client
 	dir          identity.Directory
 	echo         *echo.Echo
 	logger       *slog.Logger
 
-	bfs *backfill.Gormstore
-	bf  *backfill.Backfiller
-
-	enableRepoDiscovery bool
-}
-
-type LastSeq struct {
-	ID  uint `gorm:"primarykey"`
-	Seq int64
+	Indexer *Indexer
 }
 
-type Config struct {
-	RelayHost           string
-	ProfileIndex        string
-	PostIndex           string
-	Logger              *slog.Logger
-	RelaySyncRateLimit  int
-	IndexMaxConcurrency int
-	DiscoverRepos       bool
-}
-
-func NewServer(db *gorm.DB, escli *es.Client, dir identity.Directory, config Config) (*Server, error) {
+func NewServer(escli *es.Client, dir identity.Directory, config ServerConfig) (*Server, error) {
 	logger := config.Logger
 	if logger == nil {
 		logger = slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
···
 		}))
 	}
 
-	logger.Info("running database migrations")
-	db.AutoMigrate(&LastSeq{})
-	db.AutoMigrate(&backfill.GormDBJob{})
-
-	relayws := config.RelayHost
-	if !strings.HasPrefix(relayws, "ws") {
-		return nil, fmt.Errorf("specified relay host must include 'ws://' or 'wss://'")
+	serv := Server{
+		escli:        escli,
+		postIndex:    config.PostIndex,
+		profileIndex: config.ProfileIndex,
+		dir:          dir,
+		logger:       logger,
 	}
 
-	relayhttp := strings.Replace(relayws, "ws", "http", 1)
-	relayClient := &xrpc.Client{
-		Host: relayhttp,
-	}
-
-	s := &Server{
-		escli:               escli,
-		profileIndex:        config.ProfileIndex,
-		postIndex:           config.PostIndex,
-		db:                  db,
-		relayHost:           config.RelayHost, // NOTE: the original URL, not 'relayhttp'
-		relayClient:         relayClient,
-		dir:                 dir,
-		logger:              logger,
-		enableRepoDiscovery: config.DiscoverRepos,
-	}
-
-	bfstore := backfill.NewGormstore(db)
-	opts := backfill.DefaultBackfillOptions()
-	if config.RelaySyncRateLimit > 0 {
-		opts.SyncRequestsPerSecond = config.RelaySyncRateLimit
-		opts.ParallelBackfills = 2 * config.RelaySyncRateLimit
-	} else {
-		opts.SyncRequestsPerSecond = 8
-	}
-	opts.CheckoutPath = fmt.Sprintf("%s/xrpc/com.atproto.sync.getRepo", relayhttp)
-	if config.IndexMaxConcurrency > 0 {
-		opts.ParallelRecordCreates = config.IndexMaxConcurrency
-	} else {
-		opts.ParallelRecordCreates = 20
-	}
-	opts.NSIDFilter = "app.bsky."
-	bf := backfill.NewBackfiller(
-		"search",
-		bfstore,
-		s.handleCreateOrUpdate,
-		s.handleCreateOrUpdate,
-		s.handleDelete,
-		opts,
-	)
-
-	s.bfs = bfstore
-	s.bf = bf
-
-	return s, nil
+	return &serv, nil
 }
 
-//go:embed post_schema.json
-var palomarPostSchemaJSON string
-
-//go:embed profile_schema.json
-var palomarProfileSchemaJSON string
-
 func (s *Server) EnsureIndices(ctx context.Context) error {
 
 	indices := []struct {
···
 	Message string `json:"msg,omitempty"`
 }
 
-func (s *Server) handleHealthCheck(c echo.Context) error {
-	if err := s.db.Exec("SELECT 1").Error; err != nil {
-		s.logger.Error("healthcheck can't connect to database", "err", err)
-		return c.JSON(500, HealthStatus{Status: "error", Version: versioninfo.Short(), Message: "can't connect to database"})
+func (a *Server) handleHealthCheck(c echo.Context) error {
+	if a.Indexer != nil {
+		if err := a.Indexer.db.Exec("SELECT 1").Error; err != nil {
+			a.logger.Error("healthcheck can't connect to database", "err", err)
+			return c.JSON(500, HealthStatus{Status: "error", Version: versioninfo.Short(), Message: "can't connect to database"})
+		}
 	}
 	return c.JSON(200, HealthStatus{Status: "ok", Version: versioninfo.Short()})
 }
···
 	e.GET("/metrics", echo.WrapHandler(promhttp.Handler()))
 	e.GET("/xrpc/app.bsky.unspecced.searchPostsSkeleton", s.handleSearchPostsSkeleton)
 	e.GET("/xrpc/app.bsky.unspecced.searchActorsSkeleton", s.handleSearchActorsSkeleton)
-	e.GET("/xrpc/app.bsky.unspecced.indexRepos", s.handleIndexRepos)
 	s.echo = e
 
 	s.logger.Info("starting search API daemon", "bind", listen)

search/transform.go

···
 	}
 }
 
-func TransformPost(post *appbsky.FeedPost, ident *identity.Identity, rkey, cid string) PostDoc {
+func TransformPost(post *appbsky.FeedPost, did syntax.DID, rkey, cid string) PostDoc {
 	altText := []string{}
 	if post.Embed != nil && post.Embed.EmbedImages != nil {
 		for _, img := range post.Embed.EmbedImages.Images {
···
 	if post.Embed != nil && post.Embed.EmbedRecordWithMedia != nil {
 		embedATURI = &post.Embed.EmbedRecordWithMedia.Record.Record.Uri
 	}
-	var embedImgCount int = 0
+	var embedImgCount int
 	var embedImgAltText []string
 	var embedImgAltTextJA []string
 	if post.Embed != nil && post.Embed.EmbedImages != nil {
···
 
 	doc := PostDoc{
 		DocIndexTs:        syntax.DatetimeNow().String(),
-		DID:               ident.DID.String(),
+		DID:               did.String(),
 		RecordRkey:        rkey,
 		RecordCID:         cid,
 		Text:              post.Text,
···
 				s := dt.String()
 				doc.CreatedAt = &s
 			} else {
-				slog.Warn("rejecting future post CreatedAt", "datetime", dt.String(), "did", ident.DID.String(), "rkey", rkey)
+				slog.Warn("rejecting future post CreatedAt", "datetime", dt.String(), "did", did.String(), "rkey", rkey)
 				s := syntax.DatetimeNow().String()
 				doc.CreatedAt = &s
 			}

search/transform_test.go

···
 		Handle: syntax.Handle(row.Handle),
 		DID:    syntax.DID(row.DID),
 	}
-	doc := TransformPost(row.PostRecord, &repo, row.Rkey, row.Cid)
+	doc := TransformPost(row.PostRecord, repo.DID, row.Rkey, row.Cid)
 	doc.DocIndexTs = "2006-01-02T15:04:05.000Z"
 	assert.Equal(row.PostDoc, doc)
 	assert.Equal(row.DocId, doc.DocId())