···

       
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 {