···

       
10
       
 
       
		ParallelBackfills:     cctx.Int("backfill-workers"),

     

       
11
       
 
       
		ParallelRecordCreates: cctx.Int("backfill-consumers"),

     

       
12
       
 
       
		NSIDFilter:            "",

     

       
13
       
-
       
		SyncRequestsPerSecond: 10,

     

       
14
       
 
       
		RelayHost:             "https://" + cctx.String("relay-host"),

     

       
15
       
 
       
	}

     

       
16
       
 
       


     

···

       
10
       
 
       
		ParallelBackfills:     cctx.Int("backfill-workers"),

     

       
11
       
 
       
		ParallelRecordCreates: cctx.Int("backfill-consumers"),

     

       
12
       
 
       
		NSIDFilter:            "",

     

       
13
       
+
       
		SyncRequestsPerSecond: cctx.Int("sync-requests-limit"),

     

       
14
       
 
       
		RelayHost:             "https://" + cctx.String("relay-host"),

     

       
15
       
 
       
	}

     

       
16
       
 
       


     

···

       
14
       
 
       
)

     

       
15
       
 
       


     

       
16
       
 
       
type CensusService struct {

     

       
17
       
-
       
	cursor   *CursorService

     

       
18
       
-
       
	backfill *backfill.Backfiller

     

       
19
       
-
       


     

       
20
       
 
       
	seenHosts map[string]bool

     

       
21
       
-
       
	seenLk    sync.Mutex

     

       
22
       
-
       


     

       
23
       
-
       
	storeLk sync.Mutex

     

       
24
       
 
       
}

     

       
25
       
 
       


     

       
26
       
 
       
type jobMaker interface {

     
···

       
70
       
 
       


     

       
71
       
 
       
		for _, host := range res.Hosts {

     

       
72
       
 
       
			// don't reprocess hosts already handled

     

       
73
       
-
       
			cs.seenLk.Lock()

     

       
74
       
-
       
			_, ok := cs.seenHosts[host.Hostname]

     

       
75
       
-
       
			cs.seenLk.Unlock()

     

       
76
       
-
       
			if ok {

     

       
77
       
 
       
				slog.Info("already processed host, skipping", "host", host)

     

       
78
       
 
       
				continue

     

       
79
       
 
       
			}

     
···

       
114
       
 
       
		return

     

       
115
       
 
       
	}

     

       
116
       
 
       


     

       
117
       
-
       
	cs.storeLk.Lock()

     

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

     

       
119
       
 
       
	if err != nil {

     

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

     

       
121
       
 
       
	}

     

       
122
       
-
       
	cs.storeLk.Unlock()

     

       
123
       
 
       


     

       
124
       
 
       
	var added int

     

       
125
       
 
       
	curs := hcur.Cursor

     
···

       
138
       
 
       
			continue

     

       
139
       
 
       
		}

     

       
140
       
 
       


     

       
141
       
-
       
		cs.storeLk.Lock()

     

       
142
       
 
       
		for _, repo := range res.Repos {

     

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

     

       
144
       
 
       
			if err != nil {

     
···

       
147
       
 
       
				added += 1

     

       
148
       
 
       
			}

     

       
149
       
 
       
		}

     

       
150
       
-
       
		cs.storeLk.Unlock()

     

       
151
       
 
       


     

       
152
       
 
       
		if res.Cursor != nil && *res.Cursor != "" {

     

       
153
       
 
       
			curs = *res.Cursor

     

       
154
       
-
       
			cs.storeLk.Lock()

     

       
155
       
 
       
			if err := cs.cursor.SetHostCursor(host, curs); err != nil {

     

       
156
       
 
       
				slog.Error("error updating cursor for host", "err", err)

     

       
157
       
 
       
			}

     

       
158
       
-
       
			cs.storeLk.Unlock()

     

       
159
       
 
       
		} else {

     

       
160
       
 
       
			break

     

       
161
       
 
       
		}

     

       
162
       
 
       
	}

     

       
163
       
 
       


     

       
164
       
 
       
	slog.Info("finished listing repos", "host", host)

     

       
165
       
-
       


     

       
166
       
-
       
	cs.seenLk.Lock()

     

       
167
       
-
       
	defer cs.seenLk.Unlock()

     

       
168
       
-
       


     

       
169
       
 
       
	cs.seenHosts[host] = true

     

       
170
       
 
       
}

     

       
171
       
 
       


     

···

       
14
       
 
       
)

     

       
15
       
 
       


     

       
16
       
 
       
type CensusService struct {

     

       
17
       
+
       
	cursor    *CursorService

     

       
18
       
+
       
	backfill  *backfill.Backfiller

     

       
0
       
 
       

     

       
19
       
 
       
	seenHosts map[string]bool

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
20
       
 
       
}

     

       
21
       
 
       


     

       
22
       
 
       
type jobMaker interface {

     
···

       
66
       
 
       


     

       
67
       
 
       
		for _, host := range res.Hosts {

     

       
68
       
 
       
			// don't reprocess hosts already handled

     

       
69
       
+
       
			seen := cs.seenHosts[host.Hostname]

     

       
70
       
+
       
			if seen {

     

       
0
       
 
       

     

       
0
       
 
       

     

       
71
       
 
       
				slog.Info("already processed host, skipping", "host", host)

     

       
72
       
 
       
				continue

     

       
73
       
 
       
			}

     
···

       
108
       
 
       
		return

     

       
109
       
 
       
	}

     

       
110
       
 
       


     

       
0
       
 
       

     

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

     

       
112
       
 
       
	if err != nil {

     

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

     

       
114
       
 
       
	}

     

       
0
       
 
       

     

       
115
       
 
       


     

       
116
       
 
       
	var added int

     

       
117
       
 
       
	curs := hcur.Cursor

     
···

       
130
       
 
       
			continue

     

       
131
       
 
       
		}

     

       
132
       
 
       


     

       
0
       
 
       

     

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

     

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

     

       
135
       
 
       
			if err != nil {

     
···

       
138
       
 
       
				added += 1

     

       
139
       
 
       
			}

     

       
140
       
 
       
		}

     

       
0
       
 
       

     

       
141
       
 
       


     

       
142
       
 
       
		if res.Cursor != nil && *res.Cursor != "" {

     

       
143
       
 
       
			curs = *res.Cursor

     

       
0
       
 
       

     

       
144
       
 
       
			if err := cs.cursor.SetHostCursor(host, curs); err != nil {

     

       
145
       
 
       
				slog.Error("error updating cursor for host", "err", err)

     

       
146
       
 
       
			}

     

       
0
       
 
       

     

       
147
       
 
       
		} else {

     

       
148
       
 
       
			break

     

       
149
       
 
       
		}

     

       
150
       
 
       
	}

     

       
151
       
 
       


     

       
152
       
 
       
	slog.Info("finished listing repos", "host", host)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
153
       
 
       
	cs.seenHosts[host] = true

     

       
154
       
 
       
}

     

       
155
       
 
       


     

···

       
3
       
 
       
import (

     

       
4
       
 
       
	"context"

     

       
5
       
 
       
	"log/slog"

     

       
6
       
-
       
	"sync"

     

       
7
       
 
       
	"time"

     

       
8
       
 
       


     

       
9
       
 
       
	"gorm.io/gorm"

     

       
10
       
 
       
)

     

       
11
       
 
       


     

       
12
       
 
       
type CursorService struct {

     

       
13
       
-
       
	store *gorm.DB

     

       
14
       
-
       


     

       
15
       
-
       
	firehoseLk  sync.Mutex

     

       
16
       
 
       
	firehoseSeq int64

     

       
17
       
 
       
}

     

       
18
       
 
       


     

···

       
3
       
 
       
import (

     

       
4
       
 
       
	"context"

     

       
5
       
 
       
	"log/slog"

     

       
0
       
 
       

     

       
6
       
 
       
	"time"

     

       
7
       
 
       


     

       
8
       
 
       
	"gorm.io/gorm"

     

       
9
       
 
       
)

     

       
10
       
 
       


     

       
11
       
 
       
type CursorService struct {

     

       
12
       
+
       
	store       *gorm.DB

     

       
0
       
 
       

     

       
0
       
 
       

     

       
13
       
 
       
	firehoseSeq int64

     

       
14
       
 
       
}

     

       
15
       
 
       


     

···

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

     

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

     

       
15
       
 
       
	curs, err := cursorSvc.GetFirehoseCursor()

     

       
16
       
-
       
	if err == nil { // reversed

     

       
0
       
 
       

     

       
0
       
 
       

     

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

     

       
18
       
 
       
	}

     

       
19
       
 
       


     

···

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

     

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

     

       
15
       
 
       
	curs, err := cursorSvc.GetFirehoseCursor()

     

       
16
       
+
       
	if err != nil {

     

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

     

       
18
       
+
       
	} else if curs > 0 {

     

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

     

       
20
       
 
       
	}

     

       
21