···

       
3
       
3
       
 
       
import (

     

       
4
       
4
       
 
       
	"context"

     

       
5
       
5
       
 
       
	"log/slog"

     

       
6
       
6
       
-
       
	"sync"

     

       
7
       
6
       
 
       
	"time"

     

       
8
       
7
       
 
       


     

       
9
       
8
       
 
       
	"gorm.io/gorm"

     

       
10
       
9
       
 
       
)

     

       
11
       
10
       
 
       


     

       
12
       
11
       
 
       
type CursorService struct {

     

       
13
       
13
       
-
       
	store *gorm.DB

     

       
14
       
14
       
-
       


     

       
15
       
15
       
-
       
	firehoseLk  sync.Mutex

     

       
12
       
12
       
+
       
	store       *gorm.DB

     

       
16
       
13
       
 
       
	firehoseSeq int64

     

       
17
       
14
       
 
       
}

     

       
18
       
15
       
 
       


     

···

       
13
       
13
       
 
       
func NewFirehoseConnection(ctx context.Context, cctx *cli.Context, cursorSvc *CursorService) (*websocket.Conn, error) {

     

       
14
       
14
       
 
       
	url := fmt.Sprintf("wss://%s/xrpc/com.atproto.sync.subscribeRepos", cctx.String("relay-host"))

     

       
15
       
15
       
 
       
	curs, err := cursorSvc.GetFirehoseCursor()

     

       
16
       
16
       
-
       
	if err == nil { // reversed

     

       
16
       
16
       
+
       
	if err != nil {

     

       
17
       
17
       
+
       
		slog.Error("error getting firehose cursor", "err", err)

     

       
18
       
18
       
+
       
	} else if curs > 0 {

     

       
17
       
19
       
 
       
		url += fmt.Sprintf("?cursor=%d", curs)

     

       
18
       
20
       
 
       
	}

     

       
19
       
21
       
 
       


     

···

       
7
       
7
       
 
       
	"time"

     

       
8
       
8
       
 
       


     

       
9
       
9
       
 
       
	"github.com/bluesky-social/indigo/api/atproto"

     

       
10
       
10
       
-
       
	"github.com/bluesky-social/indigo/backfill"

     

       
10
       
10
       
+
       
	backfill "github.com/bluesky-social/indigo/backfill/next"

     

       
11
       
11
       
 
       
	"github.com/bluesky-social/indigo/xrpc"

     

       
12
       
12
       
 
       
	"github.com/urfave/cli/v2"

     

       
13
       
13
       
 
       
	"golang.org/x/sync/semaphore"

     
···

       
19
       
19
       
 
       
	seenHosts map[string]bool

     

       
20
       
20
       
 
       
}

     

       
21
       
21
       
 
       


     

       
22
       
22
       
-
       
type jobMaker interface {

     

       
23
       
23
       
-
       
	GetOrCreateJob(context.Context, string, string) (backfill.Job, error)

     

       
24
       
24
       
-
       
}

     

       
25
       
25
       
-
       


     

       
26
       
22
       
 
       
func NewCensusService(cursorSvc *CursorService, backfillSvc *backfill.Backfiller) *CensusService {

     

       
27
       
23
       
 
       
	return &CensusService{

     

       
28
       
24
       
 
       
		cursor:    cursorSvc,

     
···

       
102
       
98
       
 
       
		Host: "https://" + host,

     

       
103
       
99
       
 
       
	}

     

       
104
       
100
       
 
       


     

       
105
       
105
       
-
       
	jmstore, ok := cs.backfill.Store.(jobMaker)

     

       
106
       
106
       
-
       
	if !ok {

     

       
107
       
107
       
-
       
		slog.Error("configured job store doesn't support random job creation")

     

       
108
       
108
       
-
       
		return

     

       
109
       
109
       
-
       
	}

     

       
110
       
110
       
-
       


     

       
111
       
101
       
 
       
	hcur, err := cs.cursor.GetHostCursor(host)

     

       
112
       
102
       
 
       
	if err != nil {

     

       
113
       
103
       
 
       
		slog.Error("error fetching host cursor", "err", err)

     
···

       
131
       
121
       
 
       
		}

     

       
132
       
122
       
 
       


     

       
133
       
123
       
 
       
		for _, repo := range res.Repos {

     

       
134
       
134
       
-
       
			_, err := jmstore.GetOrCreateJob(ctx, repo.Did, backfill.StateEnqueued)

     

       
124
       
124
       
+
       
			err := cs.backfill.EnqueueJob(ctx, host, repo.Did)

     

       
135
       
125
       
 
       
			if err != nil {

     

       
136
       
126
       
 
       
				slog.Error("error adding repo to backfiller", "err", err)

     

       
137
       
127
       
 
       
			} else {

     

···

       
5
       
5
       
 
       
toolchain go1.24.5

     

       
6
       
6
       
 
       


     

       
7
       
7
       
 
       
require (

     

       
8
       
8
       
-
       
	github.com/bluesky-social/indigo v0.0.0-20250808182429-6f0837c2d12b

     

       
8
       
8
       
+
       
	github.com/bluesky-social/indigo v0.0.0-20250706015339-f9e61880249d

     

       
9
       
9
       
 
       
	github.com/gorilla/websocket v1.5.1

     

       
10
       
10
       
 
       
	github.com/ipfs/go-cid v0.4.1

     

       
11
       
11
       
 
       
	github.com/urfave/cli/v2 v2.25.7

     

       
12
       
12
       
-
       
	golang.org/x/sync v0.7.0

     

       
12
       
12
       
+
       
	golang.org/x/sync v0.15.0

     

       
13
       
13
       
 
       
	gorm.io/gorm v1.25.9

     

       
14
       
14
       
 
       
)

     

       
15
       
15
       
 
       


     
···

       
22
       
22
       
 
       
	github.com/cpuguy83/go-md2man/v2 v2.0.3 // indirect

     

       
23
       
23
       
 
       
	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 // indirect

     

       
24
       
24
       
 
       
	github.com/felixge/httpsnoop v1.0.4 // indirect

     

       
25
       
25
       
-
       
	github.com/go-logr/logr v1.4.1 // indirect

     

       
25
       
25
       
+
       
	github.com/go-logr/logr v1.4.2 // indirect

     

       
26
       
26
       
 
       
	github.com/go-logr/stdr v1.2.2 // indirect

     

       
27
       
27
       
 
       
	github.com/goccy/go-json v0.10.2 // indirect

     

       
28
       
28
       
 
       
	github.com/gocql/gocql v1.7.0 // indirect

     

       
29
       
29
       
 
       
	github.com/gogo/protobuf v1.3.2 // indirect

     

       
30
       
30
       
 
       
	github.com/golang/snappy v0.0.4 // indirect

     

       
31
       
31
       
-
       
	github.com/google/uuid v1.4.0 // indirect

     

       
31
       
31
       
+
       
	github.com/google/uuid v1.6.0 // indirect

     

       
32
       
32
       
 
       
	github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed // indirect

     

       
33
       
33
       
 
       
	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect

     

       
34
       
34
       
 
       
	github.com/hashicorp/go-retryablehttp v0.7.5 // indirect

     
···

       
58
       
58
       
 
       
	github.com/jackc/pgpassfile v1.0.0 // indirect

     

       
59
       
59
       
 
       
	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect

     

       
60
       
60
       
 
       
	github.com/jackc/pgx/v5 v5.5.0 // indirect

     

       
61
       
61
       
-
       
	github.com/jackc/puddle/v2 v2.2.1 // indirect

     

       
61
       
61
       
+
       
	github.com/jackc/puddle/v2 v2.2.2 // indirect

     

       
62
       
62
       
 
       
	github.com/jbenet/goprocess v0.1.4 // indirect

     

       
63
       
63
       
 
       
	github.com/jinzhu/inflection v1.0.0 // indirect

     

       
64
       
64
       
 
       
	github.com/jinzhu/now v1.1.5 // indirect

     
···

       
94
       
94
       
 
       
	github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect

     

       
95
       
95
       
 
       
	gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b // indirect

     

       
96
       
96
       
 
       
	gitlab.com/yawning/tuplehash v0.0.0-20230713102510-df83abbf9a02 // indirect

     

       
97
       
97
       
+
       
	go.opentelemetry.io/auto/sdk v1.1.0 // indirect

     

       
97
       
98
       
 
       
	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 // indirect

     

       
98
       
98
       
-
       
	go.opentelemetry.io/otel v1.21.0 // indirect

     

       
99
       
99
       
-
       
	go.opentelemetry.io/otel/metric v1.21.0 // indirect

     

       
100
       
100
       
-
       
	go.opentelemetry.io/otel/trace v1.21.0 // indirect

     

       
99
       
99
       
+
       
	go.opentelemetry.io/otel v1.36.0 // indirect

     

       
100
       
100
       
+
       
	go.opentelemetry.io/otel/metric v1.36.0 // indirect

     

       
101
       
101
       
+
       
	go.opentelemetry.io/otel/trace v1.36.0 // indirect

     

       
101
       
102
       
 
       
	go.uber.org/atomic v1.11.0 // indirect

     

       
102
       
103
       
 
       
	go.uber.org/multierr v1.11.0 // indirect

     

       
103
       
103
       
-
       
	go.uber.org/zap v1.26.0 // indirect

     

       
104
       
104
       
-
       
	golang.org/x/crypto v0.21.0 // indirect

     

       
105
       
105
       
-
       
	golang.org/x/net v0.23.0 // indirect

     

       
106
       
106
       
-
       
	golang.org/x/sys v0.22.0 // indirect

     

       
107
       
107
       
-
       
	golang.org/x/text v0.14.0 // indirect

     

       
104
       
104
       
+
       
	go.uber.org/zap v1.27.0 // indirect

     

       
105
       
105
       
+
       
	golang.org/x/crypto v0.33.0 // indirect

     

       
106
       
106
       
+
       
	golang.org/x/net v0.35.0 // indirect

     

       
107
       
107
       
+
       
	golang.org/x/sys v0.33.0 // indirect

     

       
108
       
108
       
+
       
	golang.org/x/text v0.22.0 // indirect

     

       
108
       
109
       
 
       
	golang.org/x/time v0.3.0 // indirect

     

       
109
       
110
       
 
       
	golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect

     

       
110
       
111
       
 
       
	google.golang.org/protobuf v1.33.0 // indirect